[Tarantool-patches] [PATCH 06/14] box/journal: supersede journal_write with journal_write_async

[Cyrill Gorcunov]

1) name it journal_write_async, since it operates in async
   mode where completion handlers are deferred upon journal
   engine does a real write;

2) require a caller to ship a fiber with active transaction
   bound so we could restore it back if write operation failed
   thus rollback will obtain consistent active transaction;

3) move fiber_set_txn to the header file for reuse in
   journal_write_async.

Suggested-by: Konstantin Osipov <kostja.osipov at gmail.com>
Signed-off-by: Cyrill Gorcunov <gorcunov at gmail.com>
---
 src/box/journal.c | 22 ++++++++++++++++++++++
 src/box/journal.h |  9 +++------
 src/box/txn.c     |  9 +--------
 src/box/txn.h     |  9 +++++++++
 4 files changed, 35 insertions(+), 14 deletions(-)

diff --git a/src/box/journal.c b/src/box/journal.c
index 11e78990d..238860165 100644
--- a/src/box/journal.c
+++ b/src/box/journal.c
@@ -29,6 +29,8 @@
  * SUCH DAMAGE.
  */
 #include "journal.h"
+#include "fiber.h"
+#include "txn.h"
 #include <small/region.h>
 #include <diag.h>
 
@@ -76,3 +78,23 @@ journal_entry_new(size_t n_rows, struct region *region,
 	return entry;
 }
 
+int
+journal_write_async(struct journal_entry *entry)
+{
+	struct txn *txn = in_txn();
+	assert(txn != NULL);
+
+	/*
+	 * The transaction should be unbound
+	 * from current fiber, once operation
+	 * is complete the journal engine will
+	 * bound it back. For rollback sake we
+	 * bind it by self on error path.
+	 */
+	fiber_set_txn(fiber(), NULL);
+	int ret = current_journal->write(current_journal, entry);
+	if (ret != 0)
+		fiber_set_txn(fiber(), txn);
+
+	return ret;
+}
diff --git a/src/box/journal.h b/src/box/journal.h
index 64f167c6f..efb6ac8e1 100644
--- a/src/box/journal.h
+++ b/src/box/journal.h
@@ -124,15 +124,12 @@ struct journal {
 extern struct journal *current_journal;
 
 /**
- * Send a single entry to write.
+ * Queue a single entry to write.
  *
  * @return 0 if write was scheduled or -1 in case of an error.
  */
-static inline int
-journal_write(struct journal_entry *entry)
-{
-	return current_journal->write(current_journal, entry);
-}
+extern int
+journal_write_async(struct journal_entry *entry);
 
 /**
  * Change the current implementation of the journaling API.
diff --git a/src/box/txn.c b/src/box/txn.c
index 682480c16..05321e97d 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -49,12 +49,6 @@ txn_on_yield(struct trigger *trigger, void *event);
 static void
 txn_run_rollback_triggers(struct txn *txn, struct rlist *triggers);
 
-static inline void
-fiber_set_txn(struct fiber *fiber, struct txn *txn)
-{
-	fiber->storage.txn = txn;
-}
-
 static int
 txn_add_redo(struct txn *txn, struct txn_stmt *stmt, struct request *request)
 {
@@ -520,7 +514,7 @@ txn_write_to_wal_async(struct txn *txn)
 	assert(local_row == remote_row + txn->n_new_rows);
 
 	/* Send the entry to the journal. */
-	if (journal_write(req) < 0) {
+	if (journal_write_async(req) < 0) {
 		diag_set(ClientError, ER_WAL_IO);
 		diag_log();
 		return -1;
@@ -583,7 +577,6 @@ txn_write(struct txn *txn)
 		fiber_set_txn(fiber(), NULL);
 		return 0;
 	}
-	fiber_set_txn(fiber(), NULL);
 	return txn_write_to_wal_async(txn);
 }
 
diff --git a/src/box/txn.h b/src/box/txn.h
index ae2c3a58f..fe13ef5f8 100644
--- a/src/box/txn.h
+++ b/src/box/txn.h
@@ -256,6 +256,15 @@ in_txn(void)
 	return fiber()->storage.txn;
 }
 
+/**
+ * Set pointer to the current transaction (if any).
+ */
+static inline void
+fiber_set_txn(struct fiber *fiber, struct txn *txn)
+{
+	fiber->storage.txn = txn;
+}
+
 /**
  * Start a transaction explicitly.
  * @pre no transaction is active
-- 
2.20.1