[PATCH 4/5] vinyl: do not allow to cancel a fiber reading a page

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

To handle fiber cancellation during page read we need to pin all objects
referenced by vy_page_read_task. Currently, there's the only such
object, vy_run. It has reference counting so pinning it is trivial.
However, to move page lookup to a reader thread, we need to also
reference key def, tuple format, and key. Format and key have reference
counting, but key def doesn't - we typically copy it. Copying it in this
case is too heavy. Let's simply drop this functionality - nothing bad
happens if a cancelled fiber won't exit until disk read is complete.
---
 src/box/vy_run.c | 51 +++++++++++++--------------------------------------
 1 file changed, 13 insertions(+), 38 deletions(-)

diff --git a/src/box/vy_run.c b/src/box/vy_run.c
index a8ca3afe..1dc7271f 100644
--- a/src/box/vy_run.c
+++ b/src/box/vy_run.c
@@ -198,18 +198,14 @@ vy_run_env_enable_coio(struct vy_run_env *env)
 
 /**
  * Execute a task on behalf of a reader thread.
- * Calls free_cb on failure.
  */
 static int
 vy_run_env_coio_call(struct vy_run_env *env, struct cbus_call_msg *msg,
-		     cbus_call_f func, cbus_call_f free_cb, double timeout)
+		     cbus_call_f func)
 {
 	/* Optimization: use blocking I/O during WAL recovery. */
-	if (env->reader_pool == NULL) {
-		if (func(msg) != 0)
-			goto fail;
-		return 0;
-	}
+	if (env->reader_pool == NULL)
+		return func(msg);
 
 	/* Pick a reader thread. */
 	struct vy_run_reader *reader;
@@ -217,21 +213,14 @@ vy_run_env_coio_call(struct vy_run_env *env, struct cbus_call_msg *msg,
 	env->next_reader %= env->reader_pool_size;
 
 	/* Post the task to the reader thread. */
+	bool cancellable = fiber_set_cancellable(false);
 	int rc = cbus_call(&reader->reader_pipe, &reader->tx_pipe,
-			   msg, func, free_cb, timeout);
-	if (!msg->complete) {
-		/*
-		 * Operation timed out or the fiber was cancelled.
-		 * free_cb will be called on task completion.
-		 */
-		return -1;
-	}
+			   msg, func, NULL, TIMEOUT_INFINITY);
+	fiber_set_cancellable(cancellable);
 	if (rc != 0)
-		goto fail;
+		return -1;
+
 	return 0;
-fail:
-	free_cb(msg);
-	return -1;
 }
 
 /**
@@ -986,20 +975,6 @@ vy_page_read_cb(struct cbus_call_msg *base)
 }
 
 /**
- * vinyl read task cleanup callback
- */
-static int
-vy_page_read_cb_free(struct cbus_call_msg *base)
-{
-	struct vy_page_read_task *task = (struct vy_page_read_task *)base;
-	struct vy_run_env *env = task->run->env;
-	vy_page_delete(task->page);
-	vy_run_unref(task->run);
-	mempool_free(&env->read_task_pool, task);
-	return 0;
-}
-
-/**
  * Read a page from disk given its number.
  * The function caches two most recently read pages.
  *
@@ -1044,14 +1019,14 @@ 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;
-	vy_run_ref(task->run);
 
-	if (vy_run_env_coio_call(env, &task->base, vy_page_read_cb,
-				 vy_page_read_cb_free, TIMEOUT_INFINITY) != 0)
-		return -1;
+	int rc = vy_run_env_coio_call(env, &task->base, vy_page_read_cb);
 
-	vy_run_unref(task->run);
 	mempool_free(&env->read_task_pool, task);
+	if (rc != 0) {
+		vy_page_delete(page);
+		return -1;
+	}
 
 	/* Update cache */
 	if (itr->prev_page != NULL)
-- 
2.11.0