[Tarantool-patches] [PATCH 1/1] [tosquash] txn_limbo: introduce rollback count

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

Limbo now has a counter - total number of performed
rollbacks. It is used as a guard to do some actions
assuming all limbo transactions will be confirmed, and
to check that there were no rollbacks in the end.

Should be squashed into "replication: only send confirmed data
during final join".

Follow up #5097
---
Branch: http://github.com/tarantool/tarantool/tree/gh-4842-sync-replication
Issue: https://github.com/tarantool/tarantool/issues/4842

 src/box/box.cc      | 18 ++++--------------
 src/box/txn_limbo.c |  6 ++----
 src/box/txn_limbo.h | 38 +++++---------------------------------
 3 files changed, 11 insertions(+), 51 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index 749c96ca1..5e28276f0 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -1762,10 +1762,7 @@ box_process_register(struct ev_io *io, struct xrow_header *header)
 		 tt_uuid_str(&instance_uuid), sio_socketname(io->fd));
 
 	/* See box_process_join() */
-	txn_limbo_start_recording(&txn_limbo);
-	auto limbo_guard = make_scoped_guard([&] {
-		txn_limbo_stop_recording(&txn_limbo);
-	});
+	int64_t limbo_rollback_count = txn_limbo.rollback_count;
 	struct vclock start_vclock;
 	vclock_copy(&start_vclock, &replicaset.vclock);
 
@@ -1781,10 +1778,8 @@ box_process_register(struct ev_io *io, struct xrow_header *header)
 	struct vclock stop_vclock;
 	vclock_copy(&stop_vclock, &replicaset.vclock);
 
-	if (txn_limbo_got_rollback(&txn_limbo))
+	if (txn_limbo.rollback_count != limbo_rollback_count)
 		tnt_raise(ClientError, ER_SYNC_ROLLBACK);
-	txn_limbo_stop_recording(&txn_limbo);
-	limbo_guard.is_active  = false;
 
 	if (txn_limbo_wait_confirm(&txn_limbo) != 0)
 		diag_raise();
@@ -1918,10 +1913,7 @@ box_process_join(struct ev_io *io, struct xrow_header *header)
 	 * before stop_vclock is confirmed, before we can proceed
 	 * to final join.
 	 */
-	txn_limbo_start_recording(&txn_limbo);
-	auto limbo_guard = make_scoped_guard([&] {
-		txn_limbo_stop_recording(&txn_limbo);
-	});
+	int64_t limbo_rollback_count = txn_limbo.rollback_count;
 	/*
 	 * Initial stream: feed replica with dirty data from engines.
 	 */
@@ -1943,10 +1935,8 @@ box_process_join(struct ev_io *io, struct xrow_header *header)
 	struct vclock stop_vclock;
 	vclock_copy(&stop_vclock, &replicaset.vclock);
 
-	if (txn_limbo_got_rollback(&txn_limbo))
+	if (txn_limbo.rollback_count != limbo_rollback_count)
 		tnt_raise(ClientError, ER_SYNC_ROLLBACK);
-	txn_limbo_stop_recording(&txn_limbo);
-	limbo_guard.is_active  = false;
 
 	if (txn_limbo_wait_confirm(&txn_limbo) != 0)
 		diag_raise();
diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index b6e353787..e28e2016f 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -40,8 +40,7 @@ txn_limbo_create(struct txn_limbo *limbo)
 	rlist_create(&limbo->queue);
 	limbo->instance_id = REPLICA_ID_NIL;
 	vclock_create(&limbo->vclock);
-	limbo->is_recording = false;
-	limbo->got_rollback = false;
+	limbo->rollback_count = 0;
 }
 
 static inline struct txn_limbo_entry *
@@ -102,8 +101,7 @@ txn_limbo_pop(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 	assert(entry->is_rollback);
 	(void) limbo;
 	rlist_del_entry(entry, in_queue);
-	if (limbo->is_recording)
-		limbo->got_rollback = true;
+	++limbo->rollback_count;
 }
 
 void
diff --git a/src/box/txn_limbo.h b/src/box/txn_limbo.h
index 1c945f21f..0d56d0d69 100644
--- a/src/box/txn_limbo.h
+++ b/src/box/txn_limbo.h
@@ -117,13 +117,13 @@ struct txn_limbo {
 	 * transactions, created on the limbo's owner node.
 	 */
 	struct vclock vclock;
-	/** Set to true when limbo records rollback occurrence. */
-	bool is_recording;
 	/**
-	 * Whether any rollbacks happened during the recording
-	 * period.
+	 * Total number of performed rollbacks. It used as a guard
+	 * to do some actions assuming all limbo transactions will
+	 * be confirmed, and to check that there were no rollbacks
+	 * in the end.
 	 */
-	bool got_rollback;
+	int64_t rollback_count;
 };
 
 /**
@@ -139,34 +139,6 @@ txn_limbo_is_empty(struct txn_limbo *limbo)
 	return rlist_empty(&limbo->queue);
 }
 
-/**
- * Make limbo remember the occurrence of rollbacks due to failed
- * quorum collection.
- */
-static inline void
-txn_limbo_start_recording(struct txn_limbo *limbo)
-{
-	limbo->is_recording = true;
-}
-
-/** Stop the recording of failed quorum collection events. */
-static inline void
-txn_limbo_stop_recording(struct txn_limbo *limbo)
-{
-	limbo->is_recording = false;
-	limbo->got_rollback = false;
-}
-
-/**
- * Returns true in case the limbo rolled back any tx since the
- * moment txn_limbo_start_recording() was called.
- */
-static inline bool
-txn_limbo_got_rollback(struct txn_limbo *limbo)
-{
-	return limbo->got_rollback;
-}
-
 /**
  * Allocate, create, and append a new transaction to the limbo.
  * The limbo entry is allocated on the transaction's region.
-- 
2.21.1 (Apple Git-122.3)