[Tarantool-patches] [PATCH 15/15] tx: use new tx managet in memtx

[Aleksandr Lyapunov <alyapunov@tarantool.org>]

---
 src/box/memtx_engine.c | 48 +++++++++++++++++++++++++-----------------------
 src/box/memtx_space.c  | 45 +++++++++++++++++++++++++++++----------------
 src/box/txn.c          |  5 +++++
 3 files changed, 59 insertions(+), 39 deletions(-)

diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index af80d9e..7fc495c 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -339,6 +339,29 @@ memtx_engine_begin(struct engine *engine, struct txn *txn)
 	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->new_tuple != NULL)
+			txm_prepare_add_story(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->new_tuple != NULL)
+			txm_let_it_go(stmt);
+	}
+}
+
 static void
 memtx_engine_rollback_statement(struct engine *engine, struct txn *txn,
 				struct txn_stmt *stmt)
@@ -348,32 +371,11 @@ memtx_engine_rollback_statement(struct engine *engine, struct txn *txn,
 	if (stmt->old_tuple == NULL && stmt->new_tuple == NULL)
 		return;
 	struct space *space = stmt->space;
-	struct memtx_space *memtx_space = (struct memtx_space *)space;
-	uint32_t index_count;
 
 	/* Only roll back the changes if they were made. */
 	if (stmt->engine_savepoint == NULL)
 		return;
 
-	if (memtx_space->replace == memtx_space_replace_all_keys)
-		index_count = space->index_count;
-	else if (memtx_space->replace == memtx_space_replace_primary_key)
-		index_count = 1;
-	else
-		panic("transaction rolled back during snapshot recovery");
-
-	for (uint32_t i = 0; i < index_count; i++) {
-		struct tuple *unused;
-		struct index *index = space->index[i];
-		/* Rollback must not fail. */
-		if (index_replace(index, stmt->new_tuple, stmt->old_tuple,
-				  DUP_INSERT, &unused) != 0) {
-			diag_log();
-			unreachable();
-			panic("failed to rollback change");
-		}
-	}
-
 	memtx_space_update_bsize(space, stmt->new_tuple, stmt->old_tuple);
 	if (stmt->old_tuple != NULL)
 		tuple_ref(stmt->old_tuple);
@@ -914,8 +916,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 5820c40..6cfa634 100644
--- a/src/box/memtx_space.c
+++ b/src/box/memtx_space.c
@@ -244,6 +244,10 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
 			     struct tuple **result)
 {
 	struct memtx_engine *memtx = (struct memtx_engine *)space->engine;
+	struct txn *txn = in_txn();
+	struct txn_stmt *stmt = txn_current_stmt(txn);
+	uint32_t index_count = space->index_count;
+
 	/*
 	 * Ensure we have enough slack memory to guarantee
 	 * successful statement-level rollback.
@@ -253,27 +257,31 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
 				       RESERVE_EXTENTS_BEFORE_DELETE) != 0)
 		return -1;
 
-	uint32_t i = 0;
-
 	/* Update the primary key */
 	struct index *pk = index_find(space, 0);
 	if (pk == NULL)
 		return -1;
 	assert(pk->def->opts.is_unique);
-	/*
-	 * If old_tuple is not NULL, the index has to
-	 * find and delete it, or return an error.
-	 */
-	if (index_replace(pk, old_tuple, new_tuple, mode, &old_tuple) != 0)
-		return -1;
-	assert(old_tuple || new_tuple);
 
-	/* Update secondary keys. */
-	for (i++; i < space->index_count; i++) {
-		struct tuple *unused;
-		struct index *index = space->index[i];
-		if (index_replace(index, old_tuple, new_tuple,
-				  DUP_INSERT, &unused) != 0)
+	uint32_t i = 0;
+	struct txm_story* story = NULL;
+	if (new_tuple != NULL) {
+		story = txm_story_new(new_tuple, stmt, index_count);
+		if (story == NULL)
+			return -1;
+
+		for (; i < index_count; i++) {
+			struct index *index = space->index[i];
+			struct tuple **replaced = &story->link[i].old_tuple;
+			if (index_replace(index, NULL, new_tuple,
+					  DUP_REPLACE_OR_INSERT, replaced) != 0)
+				goto rollback;
+		}
+		if (txm_check_and_link_add_story(story, stmt, mode) != 0)
+			goto rollback;
+	} else {
+		assert(old_tuple != NULL);
+		if (txm_link_del_story(old_tuple, stmt, index_count) != 0)
 			goto rollback;
 	}
 
@@ -287,13 +295,18 @@ rollback:
 	for (; i > 0; i--) {
 		struct tuple *unused;
 		struct index *index = space->index[i - 1];
+		struct tuple *replaced = story->link[i - 1].old_tuple;
 		/* Rollback must not fail. */
-		if (index_replace(index, new_tuple, old_tuple,
+		if (index_replace(index, new_tuple, replaced,
 				  DUP_INSERT, &unused) != 0) {
 			diag_log();
 			unreachable();
 			panic("failed to rollback change");
 		}
+		if (replaced != NULL) {
+			tuple_unref(replaced);
+			story->link[i - 1].old_tuple = NULL;
+		}
 	}
 	return -1;
 }
diff --git a/src/box/txn.c b/src/box/txn.c
index f1a7785..e0ff0e7 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -170,6 +170,11 @@ txn_stmt_new(struct region *region)
 static inline void
 txn_stmt_destroy(struct txn_stmt *stmt)
 {
+	if (stmt->add_story != NULL)
+		txm_unlink_add_story(stmt);
+	if (stmt->del_story != NULL)
+		txm_unlink_del_story(stmt);
+
 	if (stmt->old_tuple != NULL)
 		tuple_unref(stmt->old_tuple);
 	if (stmt->new_tuple != NULL)
-- 
2.7.4