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
