[tarantool-patches] [PATCH v4 6/9] wal: introduce a journal entry finalization callback

[Georgy Kirichenko <georgy@tarantool.org>]

Finalize a transaction thorough a journal entry callback. So transaction
processing doesn't rely on fiber schedule. This also enforce transaction
finalization order as triggers might fail.

Prerequisites: #1254
---
 src/box/alter.cc  |   5 +++
 src/box/box.cc    |   7 ++-
 src/box/journal.c |  10 ++++-
 src/box/journal.h |  12 ++++-
 src/box/txn.c     | 112 ++++++++++++++++++++++++++++++----------------
 src/box/vy_log.c  |   3 +-
 src/box/wal.c     |  20 ++++++++-
 7 files changed, 122 insertions(+), 47 deletions(-)

diff --git a/src/box/alter.cc b/src/box/alter.cc
index a37a68ce4..aa6a79264 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -3558,6 +3558,11 @@ unlock_after_dd(struct trigger *trigger, void *event)
 {
 	(void) trigger;
 	(void) event;
+	/*
+	 * A trigger could be processed by the wal scheduler fiber
+	 * so steal the latch first.
+	 */
+	latch_steal(&schema_lock);
 	latch_unlock(&schema_lock);
 }
 
diff --git a/src/box/box.cc b/src/box/box.cc
index 5e5cd2b08..2994363ab 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -309,10 +309,13 @@ 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;
-	return vclock_sum(journal->vclock);
+	entry->res = vclock_sum(journal->vclock);
+	if (entry->on_done_cb)
+		entry->on_done_cb(entry, entry->on_done_cb_data);
+	return entry->res;
 }
 
 static inline void
diff --git a/src/box/journal.c b/src/box/journal.c
index fe13fb6ee..eb0db9af2 100644
--- a/src/box/journal.c
+++ b/src/box/journal.c
@@ -41,7 +41,9 @@ static int64_t
 dummy_journal_write(struct journal *journal, struct journal_entry *entry)
 {
 	(void) journal;
-	(void) entry;
+	entry->res = 0;
+	if (entry->on_done_cb)
+		entry->on_done_cb(entry, entry->on_done_cb_data);
 	return 0;
 }
 
@@ -53,7 +55,9 @@ static struct journal dummy_journal = {
 struct journal *current_journal = &dummy_journal;
 
 struct journal_entry *
-journal_entry_new(size_t n_rows, struct region *region)
+journal_entry_new(size_t n_rows, struct region *region,
+		  void (*on_done_cb)(struct journal_entry *entry, void *data),
+		  void *on_done_cb_data)
 {
 	struct journal_entry *entry;
 
@@ -70,6 +74,8 @@ journal_entry_new(size_t n_rows, struct region *region)
 	entry->n_rows = n_rows;
 	entry->res = -1;
 	entry->fiber = fiber();
+	entry->on_done_cb = on_done_cb;
+	entry->on_done_cb_data = on_done_cb_data;
 	return entry;
 }
 
diff --git a/src/box/journal.h b/src/box/journal.h
index 8ac32ee5e..b704b5c67 100644
--- a/src/box/journal.h
+++ b/src/box/journal.h
@@ -58,6 +58,14 @@ struct journal_entry {
 	 * The fiber issuing the request.
 	 */
 	struct fiber *fiber;
+	/**
+	 * A journal entry completion callback.
+	 */
+	void (*on_done_cb)(struct journal_entry *entry, void *data);
+	/**
+	 * A journal entry completion callback argument.
+	 */
+	void *on_done_cb_data;
 	/**
 	 * Approximate size of this request when encoded.
 	 */
@@ -80,7 +88,9 @@ struct region;
  * @return NULL if out of memory, fiber diagnostics area is set
  */
 struct journal_entry *
-journal_entry_new(size_t n_rows, struct region *region);
+journal_entry_new(size_t n_rows, struct region *region,
+		  void (*on_done_cb)(struct journal_entry *entry, void *data),
+		  void *on_done_cb_data);
 
 /**
  * An API for an abstract journal for all transactions of this
diff --git a/src/box/txn.c b/src/box/txn.c
index 21f7e98b4..52e16f3e6 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -337,6 +337,66 @@ fail:
 	return -1;
 }
 
+/**
+ * Complete transaction processing.
+ */
+static void
+txn_complete(struct txn *txn)
+{
+	if (txn->signature < 0) {
+		if (txn->engine)
+			engine_rollback(txn->engine, txn);
+		/*
+		 * Some of triggers require for in_txn variable is set so
+		 * restore it for time a trigger is in progress.
+		 */
+		fiber_set_txn(fiber(), txn);
+		/* Rollback triggers must not throw. */
+		if (txn->has_triggers &&
+		    trigger_run(&txn->on_rollback, txn) != 0) {
+			diag_log();
+			unreachable();
+			panic("rollback trigger failed");
+		}
+		fiber_set_txn(fiber(), NULL);
+
+		return;
+	}
+	/*
+	 * Engine can be NULL if transaction contains IPROTO_NOP
+	 * statements only.
+	 */
+	if (txn->engine != NULL)
+		engine_commit(txn->engine, txn);
+	/*
+	 * Some of triggers require for in_txn variable is set so
+	 * restore it for time a trigger is in progress.
+	 */
+	fiber_set_txn(fiber(), txn);
+	/*
+	 * The transaction is in the binary log. No action below
+	 * may throw. In case an error has happened, there is
+	 * no other option but terminate.
+	 */
+	if (txn->has_triggers &&
+	    trigger_run(&txn->on_commit, txn) != 0) {
+		diag_log();
+		unreachable();
+		panic("commit trigger failed");
+	}
+
+	fiber_set_txn(fiber(), NULL);
+}
+
+static void
+txn_entry_done_cb(struct journal_entry *entry, void *data)
+{
+	struct txn *txn = (struct txn *)data;
+	txn->signature = entry->res;
+	txn_complete(txn);
+}
+
+
 static int64_t
 txn_write_to_wal(struct txn *txn)
 {
@@ -344,7 +404,9 @@ txn_write_to_wal(struct txn *txn)
 
 	struct journal_entry *req = journal_entry_new(txn->n_new_rows +
 						      txn->n_applier_rows,
-						      &txn->region);
+						      &txn->region,
+						      txn_entry_done_cb,
+						      txn);
 	if (req == NULL) {
 		txn_rollback(txn);
 		return -1;
@@ -370,14 +432,6 @@ txn_write_to_wal(struct txn *txn)
 	ev_tstamp stop = ev_monotonic_now(loop());
 
 	if (res < 0) {
-		/* Cascading rollback. */
-		txn_rollback(txn); /* Perform our part of cascading rollback. */
-		/*
-		 * Move fiber to end of event loop to avoid
-		 * execution of any new requests before all
-		 * pending rollbacks are processed.
-		 */
-		fiber_reschedule();
 		diag_set(ClientError, ER_WAL_IO);
 		diag_log();
 	} else if (stop - start > too_long_threshold) {
@@ -418,31 +472,20 @@ txn_commit(struct txn *txn)
 	}
 	trigger_clear(&txn->fiber_on_stop);
 
+	fiber_set_txn(fiber(), NULL);
 	if (txn->n_new_rows + txn->n_applier_rows > 0) {
 		txn->signature = txn_write_to_wal(txn);
 		if (txn->signature < 0)
 			return -1;
+	} else {
+		/*
+		 * However there is noting to write to wal a completion
+		 * should be fired.
+		 */
+		txn->signature = 0;
+		txn_complete(txn);
 	}
-	/*
-	 * Engine can be NULL if transaction contains IPROTO_NOP
-	 * statements only.
-	 */
-	if (txn->engine != NULL)
-		engine_commit(txn->engine, txn);
-	/*
-	 * The transaction is in the binary log. No action below
-	 * may throw. In case an error has happened, there is
-	 * no other option but terminate.
-	 */
-	if (txn->has_triggers &&
-	    trigger_run(&txn->on_commit, txn) != 0) {
-		diag_log();
-		unreachable();
-		panic("commit trigger failed");
-	}
-
 
-	fiber_set_txn(fiber(), NULL);
 	txn_free(txn);
 	return 0;
 fail:
@@ -464,17 +507,8 @@ txn_rollback(struct txn *txn)
 {
 	assert(txn == in_txn());
 	trigger_clear(&txn->fiber_on_stop);
-	if (txn->engine)
-		engine_rollback(txn->engine, txn);
-	/* Rollback triggers must not throw. */
-	if (txn->has_triggers &&
-	    trigger_run(&txn->on_rollback, txn) != 0) {
-		diag_log();
-		unreachable();
-		panic("rollback trigger failed");
-	}
-
-	fiber_set_txn(fiber(), NULL);
+	txn->signature = -1;
+	txn_complete(txn);
 	txn_free(txn);
 }
 
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index bdc1cfa31..7cf5ff3e9 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -772,7 +772,8 @@ vy_log_flush(void)
 		tx_size++;
 
 	size_t used = region_used(&fiber()->gc);
-	struct journal_entry *entry = journal_entry_new(tx_size, &fiber()->gc);
+	struct journal_entry *entry = journal_entry_new(tx_size, &fiber()->gc,
+							NULL, NULL);
 	if (entry == NULL)
 		goto err;
 
diff --git a/src/box/wal.c b/src/box/wal.c
index 71f6dbb5c..62b6391fd 100644
--- a/src/box/wal.c
+++ b/src/box/wal.c
@@ -263,6 +263,8 @@ tx_schedule_f(va_list ap)
 			struct journal_entry *req =
 				stailq_shift_entry(&writer->schedule_queue,
 						   struct journal_entry, fifo);
+			if (req->on_done_cb != NULL)
+				req->on_done_cb(req, req->on_done_cb_data);
 			fiber_wakeup(req->fiber);
 		}
 		writer->is_in_rollback = false;
@@ -1158,7 +1160,12 @@ wal_write(struct journal *journal, struct journal_entry *entry)
 {
 	struct wal_writer *writer = (struct wal_writer *) journal;
 
-	ERROR_INJECT_RETURN(ERRINJ_WAL_IO);
+	ERROR_INJECT(ERRINJ_WAL_IO, {
+		entry->res = -1;
+		if (entry->on_done_cb != NULL)
+			entry->on_done_cb(entry, entry->on_done_cb_data);
+		return -1;
+	});
 
 	if (writer->is_in_rollback) {
 		/*
@@ -1171,6 +1178,9 @@ wal_write(struct journal *journal, struct journal_entry *entry)
 		say_error("Aborting transaction %llu during "
 			  "cascading rollback",
 			  vclock_sum(&writer->vclock));
+		entry->res = -1;
+		if (entry->on_done_cb != NULL)
+			entry->on_done_cb(entry, entry->on_done_cb_data);
 		return -1;
 	}
 
@@ -1185,6 +1195,9 @@ wal_write(struct journal *journal, struct journal_entry *entry)
 		if (batch == NULL) {
 			diag_set(OutOfMemory, sizeof(struct wal_msg),
 				 "region", "struct wal_msg");
+			entry->res = -1;
+			if (entry->on_done_cb != NULL)
+				entry->on_done_cb(entry, entry->on_done_cb_data);
 			return -1;
 		}
 		wal_msg_create(batch);
@@ -1222,7 +1235,10 @@ wal_write_in_wal_mode_none(struct journal *journal,
 		       entry->rows + entry->n_rows);
 	vclock_merge(&writer->vclock, &vclock_diff);
 	vclock_copy(&replicaset.vclock, &writer->vclock);
-	return vclock_sum(&writer->vclock);
+	entry->res = vclock_sum(&writer->vclock);
+	if (entry->on_done_cb)
+		entry->on_done_cb(entry, entry->on_done_cb_data);
+	return entry->res;
 }
 
 void
-- 
2.22.0