[PATCH 5/5] vinyl: lookup key in reader thread

[Vladimir Davydov <vdavydov.dev@gmail.com>]

If a key isn't found in the tuple cache, we fetch it from a run file. In
this case disk read and page decompression is done by a reader thread,
however key lookup in the fetched page is still performed by the tx
thread. Since pages are immutable, this could as well be done by the
reader thread, which would allow us to save some precious CPU cycles for
tx.

Close #4257
---
 src/box/vy_run.c | 79 ++++++++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 59 insertions(+), 20 deletions(-)

diff --git a/src/box/vy_run.c b/src/box/vy_run.c
index 1dc7271f..3485c500 100644
--- a/src/box/vy_run.c
+++ b/src/box/vy_run.c
@@ -95,6 +95,18 @@ struct vy_page_read_task {
 	struct vy_page_info *page_info;
 	/** vy_run with fd - ref. counted */
 	struct vy_run *run;
+	/** key to lookup within the page */
+	struct vy_entry key;
+	/** iterator type (needed for for key lookup) */
+	enum iterator_type iterator_type;
+	/** key definition (needed for key lookup) */
+	struct key_def *cmp_def;
+	/** disk format (needed for key lookup) */
+	struct tuple_format *format;
+	/** [out] position of the key in the page */
+	uint32_t pos_in_page;
+	/** [out] true if key was found in the page */
+	bool equal_found;
 	/** [out] resulting vinyl page */
 	struct vy_page *page;
 };
@@ -971,7 +983,15 @@ vy_page_read_cb(struct cbus_call_msg *base)
 	ZSTD_DStream *zdctx = vy_env_get_zdctx(task->run->env);
 	if (zdctx == NULL)
 		return -1;
-	return vy_page_read(task->page, task->page_info, task->run, zdctx);
+	if (vy_page_read(task->page, task->page_info, task->run, zdctx) != 0)
+		return -1;
+	if (task->key.stmt != NULL) {
+		task->pos_in_page = vy_page_find_key(task->page, task->key,
+						     task->cmp_def, task->format,
+						     task->iterator_type,
+						     &task->equal_found);
+	}
+	return 0;
 }
 
 /**
@@ -983,28 +1003,35 @@ vy_page_read_cb(struct cbus_call_msg *base)
  */
 static NODISCARD int
 vy_run_iterator_load_page(struct vy_run_iterator *itr, uint32_t page_no,
-			  struct vy_page **result)
+			  struct vy_entry key, enum iterator_type iterator_type,
+			  struct vy_page **result, uint32_t *pos_in_page,
+			  bool *equal_found)
 {
 	struct vy_slice *slice = itr->slice;
 	struct vy_run_env *env = slice->run->env;
 
 	/* Check cache */
-	if (itr->curr_page != NULL) {
-		if (itr->curr_page->page_no == page_no) {
-			*result = itr->curr_page;
-			return 0;
-		}
-		if (itr->prev_page != NULL &&
-		    itr->prev_page->page_no == page_no) {
-			SWAP(itr->prev_page, itr->curr_page);
-			*result = itr->curr_page;
-			return 0;
-		}
+	struct vy_page *page = NULL;
+	if (itr->curr_page != NULL &&
+	    itr->curr_page->page_no == page_no) {
+		page = itr->curr_page;
+	} else if (itr->prev_page != NULL &&
+		   itr->prev_page->page_no == page_no) {
+		SWAP(itr->prev_page, itr->curr_page);
+		page = itr->curr_page;
+	}
+	if (page != NULL) {
+		if (key.stmt != NULL)
+			*pos_in_page = vy_page_find_key(page, key, itr->cmp_def,
+							itr->format, iterator_type,
+							equal_found);
+		*result = page;
+		return 0;
 	}
 
 	/* Allocate buffers */
 	struct vy_page_info *page_info = vy_run_page_info(slice->run, page_no);
-	struct vy_page *page = vy_page_new(page_info);
+	page = vy_page_new(page_info);
 	if (page == NULL)
 		return -1;
 
@@ -1019,9 +1046,18 @@ vy_run_iterator_load_page(struct vy_run_iterator *itr, uint32_t page_no,
 	task->run = slice->run;
 	task->page_info = page_info;
 	task->page = page;
+	task->key = key;
+	task->iterator_type = iterator_type;
+	task->cmp_def = itr->cmp_def;
+	task->format = itr->format;
+	task->pos_in_page = 0;
+	task->equal_found = false;
 
 	int rc = vy_run_env_coio_call(env, &task->base, vy_page_read_cb);
 
+	*pos_in_page = task->pos_in_page;
+	*equal_found = task->equal_found;
+
 	mempool_free(&env->read_task_pool, task);
 	if (rc != 0) {
 		vy_page_delete(page);
@@ -1059,7 +1095,11 @@ vy_run_iterator_read(struct vy_run_iterator *itr,
 		     struct vy_entry *ret)
 {
 	struct vy_page *page;
-	int rc = vy_run_iterator_load_page(itr, pos.page_no, &page);
+	bool equal_found;
+	uint32_t pos_in_page;
+	int rc = vy_run_iterator_load_page(itr, pos.page_no, vy_entry_none(),
+					   ITER_GE, &page, &pos_in_page,
+					   &equal_found);
 	if (rc != 0)
 		return rc;
 	*ret = vy_page_stmt(page, pos.pos_in_page, itr->cmp_def, itr->format);
@@ -1089,14 +1129,13 @@ vy_run_iterator_search(struct vy_run_iterator *itr,
 					       equal_key);
 	if (pos->page_no == itr->slice->run->info.page_count)
 		return 1;
+	bool equal_in_page;
 	struct vy_page *page;
-	int rc = vy_run_iterator_load_page(itr, pos->page_no, &page);
+	int rc = vy_run_iterator_load_page(itr, pos->page_no, key,
+					   iterator_type, &page,
+					   &pos->pos_in_page, &equal_in_page);
 	if (rc != 0)
 		return rc;
-	bool equal_in_page = false;
-	pos->pos_in_page = vy_page_find_key(page, key, itr->cmp_def,
-					    itr->format, iterator_type,
-					    &equal_in_page);
 	if (pos->pos_in_page == page->row_count) {
 		pos->page_no++;
 		pos->pos_in_page = 0;
-- 
2.11.0