[tarantool-patches] [PATCH v4 5/9] wal: a dedicated wal scheduling fiber

[Georgy Kirichenko]

As we intend to implement asynchronous transaction we should be able to
process a transaction finalization despite the fact whether a
transaction has a waiting fiber or not. So introduce a wal->tx scheduler
fiber to process the transaction finalization. As we don't have a
asynchronous transaction right now the fiber only wakes a fiber.

Note: this induces one context switch for each batch and doesn't look too
expensive to be unaffordable luxury.

Prerequisites: #1254
---
 src/box/wal.c | 53 ++++++++++++++++++++++++++++++++++++++-------------
 1 file changed, 40 insertions(+), 13 deletions(-)

diff --git a/src/box/wal.c b/src/box/wal.c
index 0ea15a432..71f6dbb5c 100644
--- a/src/box/wal.c
+++ b/src/box/wal.c
@@ -91,6 +91,12 @@ struct wal_writer
 	struct cpipe wal_pipe;
 	/** A memory pool for messages. */
 	struct mempool msg_pool;
+	/** A queue to schedule journal entry completions. */
+	struct stailq schedule_queue;
+	/** True if writer is in rollback state. */
+	bool is_in_rollback;
+	/** A condition to signal about new schedule queue entries. */
+	struct fiber_cond schedule_cond;
 	/* ----------------- wal ------------------- */
 	/** A setting from instance configuration - rows_per_wal */
 	int64_t wal_max_rows;
@@ -245,23 +251,35 @@ xlog_write_entry(struct xlog *l, struct journal_entry *entry)
 	return xlog_tx_commit(l);
 }
 
+/*
+ * Tx schedule fiber function.
+ */
+static int
+tx_schedule_f(va_list ap)
+{
+	struct wal_writer *writer = va_arg(ap, struct wal_writer *);
+	while (!fiber_is_cancelled()) {
+		while (!stailq_empty(&writer->schedule_queue)) {
+			struct journal_entry *req =
+				stailq_shift_entry(&writer->schedule_queue,
+						   struct journal_entry, fifo);
+			fiber_wakeup(req->fiber);
+		}
+		writer->is_in_rollback = false;
+		fiber_cond_wait(&writer->schedule_cond);
+	}
+	return 0;
+}
+
 /**
- * Invoke fibers waiting for their journal_entry's to be
- * completed. The fibers are invoked in strict fifo order:
- * this ensures that, in case of rollback, requests are
- * rolled back in strict reverse order, producing
- * a consistent database state.
+ * Attach requests to a scheduling queue.
  */
 static void
 tx_schedule_queue(struct stailq *queue)
 {
-	/*
-	 * fiber_wakeup() is faster than fiber_call() when there
-	 * are many ready fibers.
-	 */
-	struct journal_entry *req;
-	stailq_foreach_entry(req, queue, fifo)
-		fiber_wakeup(req->fiber);
+	struct wal_writer *writer = &wal_writer_singleton;
+	stailq_concat(&writer->schedule_queue, queue);
+	fiber_cond_signal(&writer->schedule_cond);
 }
 
 /**
@@ -306,6 +324,8 @@ tx_schedule_rollback(struct cmsg *msg)
 	/* Must not yield. */
 	tx_schedule_queue(&writer->rollback);
 	stailq_create(&writer->rollback);
+	writer->is_in_rollback = true;
+
 	if (msg != &writer->in_rollback)
 		mempool_free(&writer->msg_pool,
 			     container_of(msg, struct wal_msg, base));
@@ -356,6 +376,7 @@ wal_writer_create(struct wal_writer *writer, enum wal_mode wal_mode,
 	writer->wal_mode = wal_mode;
 	writer->wal_max_rows = wal_max_rows;
 	writer->wal_max_size = wal_max_size;
+	writer->is_in_rollback = false;
 	journal_create(&writer->base, wal_mode == WAL_NONE ?
 		       wal_write_in_wal_mode_none : wal_write, NULL);
 
@@ -368,6 +389,12 @@ wal_writer_create(struct wal_writer *writer, enum wal_mode wal_mode,
 
 	stailq_create(&writer->rollback);
 	cmsg_init(&writer->in_rollback, NULL);
+	stailq_create(&writer->schedule_queue);
+	fiber_cond_create(&writer->schedule_cond);
+	struct fiber *schedule_fiber = fiber_new("tx_schedule", tx_schedule_f);
+	if (schedule_fiber == NULL)
+		panic("Could not create schedule fiber");
+	fiber_start(schedule_fiber, writer);
 
 	writer->checkpoint_wal_size = 0;
 	writer->checkpoint_threshold = INT64_MAX;
@@ -1133,7 +1160,7 @@ wal_write(struct journal *journal, struct journal_entry *entry)
 
 	ERROR_INJECT_RETURN(ERRINJ_WAL_IO);
 
-	if (! stailq_empty(&writer->rollback)) {
+	if (writer->is_in_rollback) {
 		/*
 		 * The writer rollback queue is not empty,
 		 * roll back this transaction immediately.
-- 
2.22.0