[tarantool-patches] [PATCH v2 4/8] Remove fiber from a journal_entry structure

[Georgy Kirichenko <georgy@tarantool.org>]

Use a trigger to handle journal entry write done event. This relaxes
friction between fiber and transaction life cycles.

Prerequisites: #1254
---
 src/box/box.cc    |  4 +++-
 src/box/journal.c |  7 ++++---
 src/box/journal.h | 16 ++++++++++++++--
 src/box/wal.c     | 36 +++++++++++++++++++++++++-----------
 4 files changed, 46 insertions(+), 17 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index 4d6515dd4..e10b73277 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -312,9 +312,11 @@ struct recovery_journal {
  */
 static int64_t
 recovery_journal_write(struct journal *base,
-		       struct journal_entry * /* entry */)
+		       struct journal_entry *entry)
 {
 	struct recovery_journal *journal = (struct recovery_journal *) base;
+	entry->done = true;
+	trigger_run(&entry->done_trigger, NULL);
 	return vclock_sum(journal->vclock);
 }
 
diff --git a/src/box/journal.c b/src/box/journal.c
index fe13fb6ee..8d213d57e 100644
--- a/src/box/journal.c
+++ b/src/box/journal.c
@@ -30,7 +30,6 @@
  */
 #include "journal.h"
 #include <small/region.h>
-#include <fiber.h>
 #include <diag.h>
 
 /**
@@ -41,7 +40,8 @@ static int64_t
 dummy_journal_write(struct journal *journal, struct journal_entry *entry)
 {
 	(void) journal;
-	(void) entry;
+	entry->done = true;
+	trigger_run(&entry->done_trigger, NULL);
 	return 0;
 }
 
@@ -66,10 +66,11 @@ journal_entry_new(size_t n_rows, struct region *region)
 		diag_set(OutOfMemory, size, "region", "struct journal_entry");
 		return NULL;
 	}
+	rlist_create(&entry->done_trigger);
 	entry->approx_len = 0;
 	entry->n_rows = n_rows;
 	entry->res = -1;
-	entry->fiber = fiber();
+	entry->done = false;
 	return entry;
 }
 
diff --git a/src/box/journal.h b/src/box/journal.h
index 77d3073c8..5e1323464 100644
--- a/src/box/journal.h
+++ b/src/box/journal.h
@@ -34,6 +34,7 @@
 #include <stdbool.h>
 #include "trigger.h"
 #include "salad/stailq.h"
+#include "trigger.h"
 
 #if defined(__cplusplus)
 extern "C" {
@@ -56,9 +57,14 @@ struct journal_entry {
 	 */
 	int64_t res;
 	/**
-	 * The fiber issuing the request.
+	 * Turns to true when entry is processed by wal.
+	 */
+	bool done;
+	/**
+	 * Triggers fired when journal entry processing is done
+	 * despite of its success.
 	 */
-	struct fiber *fiber;
+	struct rlist done_trigger;
 	/**
 	 * Approximate size of this request when encoded.
 	 */
@@ -94,6 +100,12 @@ struct journal {
 	void (*destroy)(struct journal *journal);
 };
 
+static inline void
+journal_entry_on_done(struct journal_entry *entry, struct trigger *trigger)
+{
+	trigger_add(&entry->done_trigger, trigger);
+}
+
 /**
  * Depending on the step of recovery and instance configuration
  * points at a concrete implementation of the journal.
diff --git a/src/box/wal.c b/src/box/wal.c
index 628e7529a..4b0a7c802 100644
--- a/src/box/wal.c
+++ b/src/box/wal.c
@@ -260,8 +260,10 @@ tx_schedule_queue(struct stailq *queue)
 	 * are many ready fibers.
 	 */
 	struct journal_entry *req;
-	stailq_foreach_entry(req, queue, fifo)
-		fiber_wakeup(req->fiber);
+	stailq_foreach_entry(req, queue, fifo) {
+		req->done = true;
+		trigger_run(&req->done_trigger, NULL);
+	}
 }
 
 /**
@@ -1121,6 +1123,14 @@ wal_writer_f(va_list ap)
 	return 0;
 }
 
+static void
+on_wal_write_done(struct trigger *trigger, void *event)
+{
+	(void) event;
+	struct fiber_cond *cond = (struct fiber_cond *)trigger->data;
+	fiber_cond_signal(cond);
+}
+
 /**
  * WAL writer main entry point: queue a single request
  * to be written to disk and wait until this task is completed.
@@ -1171,15 +1181,19 @@ wal_write(struct journal *journal, struct journal_entry *entry)
 	batch->approx_len += entry->approx_len;
 	writer->wal_pipe.n_input += entry->n_rows * XROW_IOVMAX;
 	cpipe_flush_input(&writer->wal_pipe);
-	/**
-	 * It's not safe to spuriously wakeup this fiber
-	 * since in that case it will ignore a possible
-	 * error from WAL writer and not roll back the
-	 * transaction.
-	 */
-	bool cancellable = fiber_set_cancellable(false);
-	fiber_yield(); /* Request was inserted. */
-	fiber_set_cancellable(cancellable);
+
+	struct fiber_cond done_cond;
+	fiber_cond_create(&done_cond);
+
+	struct trigger done_trigger;
+	trigger_create(&done_trigger, on_wal_write_done, &done_cond, NULL);
+	journal_entry_on_done(entry, &done_trigger);
+	while (!entry->done)
+		fiber_cond_wait(&done_cond);
+
+	fiber_cond_destroy(&done_cond);
+	trigger_clear(&done_trigger);
+
 	return entry->res;
 }
 
-- 
2.21.0