[Tarantool-patches] [PATCH v4 10/12] txm: use new tx manager in memtx

[Aleksandr Lyapunov <alyapunov@tarantool.org>]

Use mvcc transaction engine in memtx if the engine is enabled.

Closes #4897
---
 src/box/memtx_engine.c | 38 +++++++++++++++++++++++++++++++++++---
 src/box/memtx_space.c  | 23 +++++++++++++++++++----
 src/box/txn.c          | 12 +++++++-----
 src/box/txn.h          | 14 +++++++++-----
 src/box/vinyl.c        |  8 ++++----
 5 files changed, 74 insertions(+), 21 deletions(-)

diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index 9f079a6..9a9ae26 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -40,6 +40,7 @@
 #include "coio_file.h"
 #include "tuple.h"
 #include "txn.h"
+#include "memtx_tx.h"
 #include "memtx_tree.h"
 #include "iproto_constants.h"
 #include "xrow.h"
@@ -335,11 +336,39 @@ static int
 memtx_engine_begin(struct engine *engine, struct txn *txn)
 {
 	(void)engine;
-	txn_can_yield(txn, false);
+	txn_can_yield(txn, memtx_tx_manager_use_mvcc_engine);
+	return 0;
+}
+
+static int
+memtx_engine_prepare(struct engine *engine, struct txn *txn)
+{
+	(void)engine;
+	struct txn_stmt *stmt;
+	stailq_foreach_entry(stmt, &txn->stmts, next) {
+		if (stmt->add_story != NULL || stmt->del_story != NULL)
+			memtx_tx_history_prepare_stmt(stmt);
+	}
 	return 0;
 }
 
 static void
+memtx_engine_commit(struct engine *engine, struct txn *txn)
+{
+	(void)engine;
+	struct txn_stmt *stmt;
+	stailq_foreach_entry(stmt, &txn->stmts, next) {
+		if (stmt->add_story != NULL || stmt->del_story != NULL) {
+			ssize_t bsize = memtx_tx_history_commit_stmt(stmt);
+			assert(stmt->space->engine == engine);
+			struct memtx_space *mspace =
+				(struct memtx_space *)stmt->space;
+			mspace->bsize += bsize;
+		}
+	}
+}
+
+static void
 memtx_engine_rollback_statement(struct engine *engine, struct txn *txn,
 				struct txn_stmt *stmt)
 {
@@ -355,6 +384,9 @@ memtx_engine_rollback_statement(struct engine *engine, struct txn *txn,
 	if (stmt->engine_savepoint == NULL)
 		return;
 
+	if (stmt->add_story != NULL || stmt->del_story != NULL)
+		return memtx_tx_history_rollback_stmt(stmt);
+
 	if (memtx_space->replace == memtx_space_replace_all_keys)
 		index_count = space->index_count;
 	else if (memtx_space->replace == memtx_space_replace_primary_key)
@@ -914,8 +946,8 @@ static const struct engine_vtab memtx_engine_vtab = {
 	/* .complete_join = */ memtx_engine_complete_join,
 	/* .begin = */ memtx_engine_begin,
 	/* .begin_statement = */ generic_engine_begin_statement,
-	/* .prepare = */ generic_engine_prepare,
-	/* .commit = */ generic_engine_commit,
+	/* .prepare = */ memtx_engine_prepare,
+	/* .commit = */ memtx_engine_commit,
 	/* .rollback_statement = */ memtx_engine_rollback_statement,
 	/* .rollback = */ generic_engine_rollback,
 	/* .switch_to_ro = */ generic_engine_switch_to_ro,
diff --git a/src/box/memtx_space.c b/src/box/memtx_space.c
index 7624130..d4b18d9 100644
--- a/src/box/memtx_space.c
+++ b/src/box/memtx_space.c
@@ -32,6 +32,7 @@
 #include "space.h"
 #include "iproto_constants.h"
 #include "txn.h"
+#include "memtx_tx.h"
 #include "tuple.h"
 #include "xrow_update.h"
 #include "xrow.h"
@@ -260,6 +261,20 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
 	if (pk == NULL)
 		return -1;
 	assert(pk->def->opts.is_unique);
+
+	if (memtx_tx_manager_use_mvcc_engine) {
+		struct txn *txn = in_txn();
+		struct txn_stmt *stmt =
+			txn == NULL ? NULL : txn_current_stmt(txn);
+		if (stmt != NULL) {
+			return memtx_tx_history_add_stmt(stmt, old_tuple, new_tuple,
+						    mode, result);
+		} else {
+			/** Ephemeral space */
+			assert(space->def->id == 0);
+		}
+	}
+
 	/*
 	 * If old_tuple is not NULL, the index has to
 	 * find and delete it, or return an error.
@@ -888,7 +903,7 @@ memtx_space_check_format(struct space *space, struct tuple_format *format)
 	if (txn_check_singlestatement(txn, "space format check") != 0)
 		return -1;
 
-	txn_can_yield(txn, true);
+	bool could_yield = txn_can_yield(txn, true);
 
 	struct memtx_engine *memtx = (struct memtx_engine *)space->engine;
 	struct memtx_ddl_state state;
@@ -932,7 +947,7 @@ memtx_space_check_format(struct space *space, struct tuple_format *format)
 	iterator_delete(it);
 	diag_destroy(&state.diag);
 	trigger_clear(&on_replace);
-	txn_can_yield(txn, false);
+	txn_can_yield(txn, could_yield);
 	return rc;
 }
 
@@ -1046,7 +1061,7 @@ memtx_space_build_index(struct space *src_space, struct index *new_index,
 	if (txn_check_singlestatement(txn, "index build") != 0)
 		return -1;
 
-	txn_can_yield(txn, true);
+	bool could_yield = txn_can_yield(txn, true);
 
 	struct memtx_engine *memtx = (struct memtx_engine *)src_space->engine;
 	struct memtx_ddl_state state;
@@ -1124,7 +1139,7 @@ memtx_space_build_index(struct space *src_space, struct index *new_index,
 	iterator_delete(it);
 	diag_destroy(&state.diag);
 	trigger_clear(&on_replace);
-	txn_can_yield(txn, false);
+	txn_can_yield(txn, could_yield);
 	return rc;
 }
 
diff --git a/src/box/txn.c b/src/box/txn.c
index 023da91..4f5484e 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -123,6 +123,8 @@ txn_stmt_new(struct region *region)
 static inline void
 txn_stmt_destroy(struct txn_stmt *stmt)
 {
+	assert(stmt->add_story == NULL && stmt->del_story == NULL);
+
 	if (stmt->old_tuple != NULL)
 		tuple_unref(stmt->old_tuple);
 	if (stmt->new_tuple != NULL)
@@ -993,20 +995,20 @@ txn_check_singlestatement(struct txn *txn, const char *where)
 	return 0;
 }
 
-void
+bool
 txn_can_yield(struct txn *txn, bool set)
 {
 	assert(txn == in_txn());
-	if (set) {
-		assert(!txn_has_flag(txn, TXN_CAN_YIELD));
+	bool could = txn_has_flag(txn, TXN_CAN_YIELD);
+	if (set && !could) {
 		txn_set_flag(txn, TXN_CAN_YIELD);
 		trigger_clear(&txn->fiber_on_yield);
-	} else {
-		assert(txn_has_flag(txn, TXN_CAN_YIELD));
+	} else if (!set && could) {
 		txn_clear_flag(txn, TXN_CAN_YIELD);
 		trigger_create(&txn->fiber_on_yield, txn_on_yield, NULL, NULL);
 		trigger_add(&fiber()->on_yield, &txn->fiber_on_yield);
 	}
+	return could;
 }
 
 int64_t
diff --git a/src/box/txn.h b/src/box/txn.h
index ba818d0..ed2ce82 100644
--- a/src/box/txn.h
+++ b/src/box/txn.h
@@ -620,12 +620,16 @@ txn_check_singlestatement(struct txn *txn, const char *where);
  * as aborted, which results in rolling back the transaction on
  * commit.
  *
- * This function is used by the memtx engine, because it doesn't
- * support yields inside transactions. It is also used to temporarily
- * enable yields for long DDL operations such as building an index
- * or checking a space format.
+ *
+ * This function is used by the memtx engine, because there are cases
+ * when if doesn't support yields inside transactions. It is also
+ * used to temporarily enable yields for long DDL operations such as
+ * building an index or checking a space format.
+ *
+ * Return previous state of the flag: true - yields were enabled,
+ * false - yields were disabled before the function call.
  */
-void
+bool
 txn_can_yield(struct txn *txn, bool set);
 
 /**
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index aa6e50f..cee39c5 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1076,7 +1076,7 @@ vinyl_space_check_format(struct space *space, struct tuple_format *format)
 		return -1;
 
 	/* See the comment in vinyl_space_build_index(). */
-	txn_can_yield(txn, true);
+	bool could_yield = txn_can_yield(txn, true);
 
 	struct trigger on_replace;
 	struct vy_check_format_ctx ctx;
@@ -1128,7 +1128,7 @@ vinyl_space_check_format(struct space *space, struct tuple_format *format)
 out:
 	diag_destroy(&ctx.diag);
 	trigger_clear(&on_replace);
-	txn_can_yield(txn, false);
+	txn_can_yield(txn, could_yield);
 	return rc;
 }
 
@@ -4175,7 +4175,7 @@ vinyl_space_build_index(struct space *src_space, struct index *new_index,
 	 * change the data dictionary, so there is no dirty state
 	 * that can be observed.
 	 */
-	txn_can_yield(txn, true);
+	bool could_yield = txn_can_yield(txn, true);
 
 	/*
 	 * Iterate over all tuples stored in the space and insert
@@ -4276,7 +4276,7 @@ vinyl_space_build_index(struct space *src_space, struct index *new_index,
 out:
 	diag_destroy(&ctx.diag);
 	trigger_clear(&on_replace);
-	txn_can_yield(txn, false);
+	txn_can_yield(txn, could_yield);
 	return rc;
 }
 
-- 
2.7.4