[Tarantool-patches] [PATCH 0/4] sync replication: add rollback processing.

[Tarantool-patches] [PATCH 0/4] sync replication: add rollback processing.

[Serge Petrenko]

Branch: gh-4842-sync-replication
Issues: https://github.com/tarantool/tarantool/issues/4848



Serge Petrenko (4):
  xrow: fix comment typo
  xrow: add ability to encode/decode ROLLBACK requests
  txn_limbo: add timeout when waiting for acks.
  txn_limbo: add ROLLBACK processing

 src/box/applier.cc         |  40 ++++++++++---
 src/box/iproto_constants.h |   9 +++
 src/box/relay.cc           |   2 +-
 src/box/txn.c              |   5 +-
 src/box/txn_limbo.c        | 117 ++++++++++++++++++++++++++++++++-----
 src/box/txn_limbo.h        |  12 +++-
 src/box/xrow.c             |  38 ++++++++++--
 src/box/xrow.h             |  29 ++++++++-
 8 files changed, 219 insertions(+), 33 deletions(-)

-- 
2.24.3 (Apple Git-128)

[Tarantool-patches] [PATCH 1/4] xrow: fix comment typo

[Serge Petrenko]

---
 src/box/xrow.h | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/box/xrow.h b/src/box/xrow.h
index 75af71b77..027b6b14f 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -214,7 +214,7 @@ xrow_encode_dml(const struct request *request, struct region *region,
  * @param replica_id master's instance id.
  * @param lsn last confirmed lsn.
  * @retval -1 on error.
- * @retval > 0 xrow bodycnt.
+ * @retval 0 success.
  */
 int
 xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn);
@@ -224,8 +224,8 @@ xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn);
  * @param row xrow header.
  * @param[out] replica_id master's instance id.
  * @param[out] lsn last confirmed lsn.
- * @retwal -1 on error.
- * @retwal 0 success.
+ * @retval -1 on error.
+ * @retval 0 success.
  */
 int
 xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn);
-- 
2.24.3 (Apple Git-128)

Re: [Tarantool-patches] [PATCH 1/4] xrow: fix comment typo

[Vladislav Shpilevoy]

This commit is fine. I suggest to merge it into the commit,
which introduced xrow_encode/decode_confirm().

Re: [Tarantool-patches] [PATCH 1/4] xrow: fix comment typo

[Vladislav Shpilevoy]

Thanks for the patch!

Consider these changes (you maybe should keep the old names,
mine are probably worse):

====================
diff --git a/src/box/xrow.c b/src/box/xrow.c
index 7a79a18dd..5055cba46 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -879,8 +879,8 @@ xrow_encode_dml(const struct request *request, struct region *region,
 }
 
 int
-xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
-			     int64_t lsn)
+xrow_encode_synchro_finish(struct xrow_header *row, uint32_t replica_id,
+			   int64_t lsn, int type)
 {
 	size_t len = mp_sizeof_map(2) + mp_sizeof_uint(IPROTO_REPLICA_ID) +
 		     mp_sizeof_uint(replica_id) + mp_sizeof_uint(IPROTO_LSN) +
@@ -903,6 +903,7 @@ xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
 	row->body[0].iov_base = buf;
 	row->body[0].iov_len = len;
 	row->bodycnt = 1;
+	row->type = type;
 
 	return 0;
 }
@@ -910,26 +911,19 @@ xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
 int
 xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
 {
-	int res = xrow_encode_confirm_rollback(row, replica_id, lsn);
-	if (res == 0) {
-		row->type = IPROTO_CONFIRM;
-	}
-	return res;
+	return xrow_encode_synchro_finish(row, replica_id, lsn, IPROTO_CONFIRM);
 }
 
 int
 xrow_encode_rollback(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
 {
-	int res = xrow_encode_confirm_rollback(row, replica_id, lsn);
-	if (res == 0) {
-		row->type = IPROTO_ROLLBACK;
-	}
-	return res;
+	return xrow_encode_synchro_finish(row, replica_id, lsn,
+					  IPROTO_ROLLBACK);
 }
 
 int
-xrow_decode_confirm_rollback(struct xrow_header *row, uint32_t *replica_id,
-			     int64_t *lsn)
+xrow_decode_synchro_finish(struct xrow_header *row, uint32_t *replica_id,
+			   int64_t *lsn)
 {
 	if (row->bodycnt == 0) {
 		diag_set(ClientError, ER_INVALID_MSGPACK, "request body");
@@ -976,14 +970,17 @@ xrow_decode_confirm_rollback(struct xrow_header *row, uint32_t *replica_id,
 	return 0;
 }
 
-int xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
+int
+xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
 {
-	return xrow_decode_confirm_rollback(row, replica_id, lsn);
+	return xrow_decode_synchro_finish(row, replica_id, lsn);
 }
 
-int xrow_decode_rollback(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
+int
+xrow_decode_rollback(struct xrow_header *row, uint32_t *replica_id,
+		     int64_t *lsn)
 {
-	return xrow_decode_confirm_rollback(row, replica_id, lsn);
+	return xrow_decode_synchro_finish(row, replica_id, lsn);
 }
 
 int

Re: [Tarantool-patches] [PATCH 1/4] xrow: fix comment typo

[Serge Petrenko]

19.06.2020 01:15, Vladislav Shpilevoy пишет:
> Thanks for the patch!

Hi! Thanks for the review!

>
> Consider these changes (you maybe should keep the old names,
> mine are probably worse):
>
> ====================
> diff --git a/src/box/xrow.c b/src/box/xrow.c
> index 7a79a18dd..5055cba46 100644
> --- a/src/box/xrow.c
> +++ b/src/box/xrow.c
> @@ -879,8 +879,8 @@ xrow_encode_dml(const struct request *request, struct region *region,
>   }
>   
>   int
> -xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
> -			     int64_t lsn)
> +xrow_encode_synchro_finish(struct xrow_header *row, uint32_t replica_id,
> +			   int64_t lsn, int type)
>   {
>   	size_t len = mp_sizeof_map(2) + mp_sizeof_uint(IPROTO_REPLICA_ID) +
>   		     mp_sizeof_uint(replica_id) + mp_sizeof_uint(IPROTO_LSN) +
> @@ -903,6 +903,7 @@ xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
>   	row->body[0].iov_base = buf;
>   	row->body[0].iov_len = len;
>   	row->bodycnt = 1;
> +	row->type = type;
>   
>   	return 0;
>   }
> @@ -910,26 +911,19 @@ xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
>   int
>   xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
>   {
> -	int res = xrow_encode_confirm_rollback(row, replica_id, lsn);
> -	if (res == 0) {
> -		row->type = IPROTO_CONFIRM;
> -	}
> -	return res;
> +	return xrow_encode_synchro_finish(row, replica_id, lsn, IPROTO_CONFIRM);
>   }
>   
>   int
>   xrow_encode_rollback(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
>   {
> -	int res = xrow_encode_confirm_rollback(row, replica_id, lsn);
> -	if (res == 0) {
> -		row->type = IPROTO_ROLLBACK;
> -	}
> -	return res;
> +	return xrow_encode_synchro_finish(row, replica_id, lsn,
> +					  IPROTO_ROLLBACK);
>   }
>   
>   int
> -xrow_decode_confirm_rollback(struct xrow_header *row, uint32_t *replica_id,
> -			     int64_t *lsn)
> +xrow_decode_synchro_finish(struct xrow_header *row, uint32_t *replica_id,
> +			   int64_t *lsn)
>   {
>   	if (row->bodycnt == 0) {
>   		diag_set(ClientError, ER_INVALID_MSGPACK, "request body");
> @@ -976,14 +970,17 @@ xrow_decode_confirm_rollback(struct xrow_header *row, uint32_t *replica_id,
>   	return 0;
>   }
>   
> -int xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
> +int
> +xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
>   {
> -	return xrow_decode_confirm_rollback(row, replica_id, lsn);
> +	return xrow_decode_synchro_finish(row, replica_id, lsn);
>   }
>   
> -int xrow_decode_rollback(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
> +int
> +xrow_decode_rollback(struct xrow_header *row, uint32_t *replica_id,
> +		     int64_t *lsn)
>   {
> -	return xrow_decode_confirm_rollback(row, replica_id, lsn);
> +	return xrow_decode_synchro_finish(row, replica_id, lsn);
>   }
>   
>   int


I applied your diff while keeping the old names and squashed the commit

into the one introducing CONFIRM entry.

(This answer is both for patches 1 and 2, since you answered here)

-- 
Serge Petrenko

[Tarantool-patches] [PATCH 2/4] xrow: add ability to encode/decode ROLLBACK requests

[Serge Petrenko]

ROLLBACK request contains the same data as CONFIRM request.
The only difference is the request semantics. While a CONFIRM request
releases all the limbo entries up to the given lsn, the ROLLBACK request
rolls back all the entries with lsn greater than given one.

Part-of #4848
---
 src/box/iproto_constants.h |  9 +++++++++
 src/box/xrow.c             | 38 ++++++++++++++++++++++++++++++++++----
 src/box/xrow.h             | 23 +++++++++++++++++++++++
 3 files changed, 66 insertions(+), 4 deletions(-)

diff --git a/src/box/iproto_constants.h b/src/box/iproto_constants.h
index 1466b456f..45c8af236 100644
--- a/src/box/iproto_constants.h
+++ b/src/box/iproto_constants.h
@@ -221,6 +221,8 @@ enum iproto_type {
 
 	/** A confirmation message for synchronous transactions. */
 	IPROTO_CONFIRM = 40,
+	/** A rollback message for synchronous transactions. */
+	IPROTO_ROLLBACK = 41,
 
 	/** PING request */
 	IPROTO_PING = 64,
@@ -337,6 +339,13 @@ iproto_type_is_request(uint32_t type)
 	return type > IPROTO_OK && type <= IPROTO_TYPE_STAT_MAX;
 }
 
+/** CONFIRM/ROLLBACK entries for synchronous replication. */
+static inline bool
+iproto_type_is_synchro_request(uint32_t type)
+{
+	return type == IPROTO_CONFIRM || type == IPROTO_ROLLBACK;
+}
+
 /**
  * The request is "synchronous": no other requests
  * on this connection should be taken before this one
diff --git a/src/box/xrow.c b/src/box/xrow.c
index 896e001b7..7a79a18dd 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -879,7 +879,8 @@ xrow_encode_dml(const struct request *request, struct region *region,
 }
 
 int
-xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
+xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
+			     int64_t lsn)
 {
 	size_t len = mp_sizeof_map(2) + mp_sizeof_uint(IPROTO_REPLICA_ID) +
 		     mp_sizeof_uint(replica_id) + mp_sizeof_uint(IPROTO_LSN) +
@@ -903,13 +904,32 @@ xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
 	row->body[0].iov_len = len;
 	row->bodycnt = 1;
 
-	row->type = IPROTO_CONFIRM;
-
 	return 0;
 }
 
 int
-xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
+xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
+{
+	int res = xrow_encode_confirm_rollback(row, replica_id, lsn);
+	if (res == 0) {
+		row->type = IPROTO_CONFIRM;
+	}
+	return res;
+}
+
+int
+xrow_encode_rollback(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
+{
+	int res = xrow_encode_confirm_rollback(row, replica_id, lsn);
+	if (res == 0) {
+		row->type = IPROTO_ROLLBACK;
+	}
+	return res;
+}
+
+int
+xrow_decode_confirm_rollback(struct xrow_header *row, uint32_t *replica_id,
+			     int64_t *lsn)
 {
 	if (row->bodycnt == 0) {
 		diag_set(ClientError, ER_INVALID_MSGPACK, "request body");
@@ -956,6 +976,16 @@ xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
 	return 0;
 }
 
+int xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
+{
+	return xrow_decode_confirm_rollback(row, replica_id, lsn);
+}
+
+int xrow_decode_rollback(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn)
+{
+	return xrow_decode_confirm_rollback(row, replica_id, lsn);
+}
+
 int
 xrow_to_iovec(const struct xrow_header *row, struct iovec *out)
 {
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 027b6b14f..1def394e7 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -230,6 +230,29 @@ xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn);
 int
 xrow_decode_confirm(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn);
 
+/**
+ * Encode the ROLLBACK row body and set row type to
+ * IPROTO_ROLLBACK.
+ * @param row xrow header.
+ * @param replica_id master's instance id.
+ * @param lsn lsn to rollback to.
+ * @retval -1  on error.
+ * @retval 0 success.
+ */
+int
+xrow_encode_rollback(struct xrow_header *row, uint32_t replica_id, int64_t lsn);
+
+/**
+ * Decode the ROLLBACK row body.
+ * @param row xrow header.
+ * @param[out] replica_id master's instance id.
+ * @param[out] lsn lsn to rollback to.
+ * @retval -1 on error.
+ * @retval 0 success.
+ */
+int
+xrow_decode_rollback(struct xrow_header *row, uint32_t *replica_id, int64_t *lsn);
+
 /**
  * CALL/EVAL request.
  */
-- 
2.24.3 (Apple Git-128)

Re: [Tarantool-patches] [PATCH 2/4] xrow: add ability to encode/decode ROLLBACK requests

[Cyrill Gorcunov]

> diff --git a/src/box/xrow.c b/src/box/xrow.c
> index 896e001b7..7a79a18dd 100644
> --- a/src/box/xrow.c
> +++ b/src/box/xrow.c
> @@ -879,7 +879,8 @@ xrow_encode_dml(const struct request *request, struct region *region,
>  }
>  
>  int
> -xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
> +xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
> +			     int64_t lsn)
>  {

Should not it be a static function?

Re: [Tarantool-patches] [PATCH 2/4] xrow: add ability to encode/decode ROLLBACK requests

[Serge Petrenko]

18.06.2020 17:46, Cyrill Gorcunov пишет:
>> diff --git a/src/box/xrow.c b/src/box/xrow.c
>> index 896e001b7..7a79a18dd 100644
>> --- a/src/box/xrow.c
>> +++ b/src/box/xrow.c
>> @@ -879,7 +879,8 @@ xrow_encode_dml(const struct request *request, struct region *region,
>>   }
>>   
>>   int
>> -xrow_encode_confirm(struct xrow_header *row, uint32_t replica_id, int64_t lsn)
>> +xrow_encode_confirm_rollback(struct xrow_header *row, uint32_t replica_id,
>> +			     int64_t lsn)
>>   {
> Should not it be a static function?


I agree, thanks for noticing!

Fixed.

-- 
Serge Petrenko

[Tarantool-patches] [PATCH 3/4] txn_limbo: add timeout when waiting for acks.

[Serge Petrenko]

Now txn_limbo_wait_complete() waits for acks only for txn_limbo_confirm_timeout
seconds. If a timeout is reached, the entry and all the ones following
it must be rolled back.

Part-of #4848
---
 src/box/txn_limbo.c | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index b45068fdd..a715a136e 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -128,12 +128,13 @@ txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 		return;
 	}
 	bool cancellable = fiber_set_cancellable(false);
-	while (!txn_limbo_entry_is_complete(entry))
-		fiber_yield();
+	bool timed_out = fiber_yield_timeout(txn_limbo_confirm_timeout(limbo));
 	fiber_set_cancellable(cancellable);
-	// TODO: implement rollback.
-	assert(!entry->is_rollback);
-	assert(entry->is_commit);
+	if (timed_out) {
+		// TODO: implement rollback.
+		entry->is_rollback = true;
+	}
+	assert(txn_limbo_entry_is_complete(entry));
 	txn_limbo_remove(limbo, entry);
 	txn_clear_flag(txn, TXN_WAIT_ACK);
 }
-- 
2.24.3 (Apple Git-128)

[Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Serge Petrenko]

Now txn_limbo writes a ROLLBACK entry to WAL when one of the limbo
entries fails to gather quorum during a txn_limbo_confirm_timeout.
All the limbo entries, starting with the failed one, are rolled back in
reverse order.

Closes #4848
---
 src/box/applier.cc  |  40 ++++++++++++----
 src/box/relay.cc    |   2 +-
 src/box/txn.c       |   5 +-
 src/box/txn_limbo.c | 110 +++++++++++++++++++++++++++++++++++++++-----
 src/box/txn_limbo.h |  12 ++++-
 5 files changed, 146 insertions(+), 23 deletions(-)

diff --git a/src/box/applier.cc b/src/box/applier.cc
index ad2ee18a5..872372f62 100644
--- a/src/box/applier.cc
+++ b/src/box/applier.cc
@@ -260,7 +260,7 @@ process_nop(struct request *request)
  * Confirms some of the txs waiting in txn_limbo.
  */
 static int
-applier_on_confirm(struct trigger *trig, void *data)
+applier_on_confirm_written(struct trigger *trig, void *data)
 {
 	(void) trig;
 	int64_t lsn = *(int64_t *)data;
@@ -268,10 +268,23 @@ applier_on_confirm(struct trigger *trig, void *data)
 	return 0;
 }
 
+/*
+ * An on_commit trigger set on a txn containing a ROLLBACK entry.
+ * Rolls back part of the txs waiting in limbo.
+ */
 static int
-process_confirm(struct request *request)
+applier_on_rollback_written(struct trigger *trig, void *data)
 {
-	assert(request->header->type == IPROTO_CONFIRM);
+	(void) trig;
+	int64_t lsn = *(int64_t *)data;
+	txn_limbo_read_rollback(&txn_limbo, lsn);
+	return 0;
+}
+
+static int
+process_confirm_rollback(struct request *request, bool is_confirm)
+{
+	assert(iproto_type_is_synchro_request(request->header->type));
 	uint32_t replica_id;
 	struct txn *txn = in_txn();
 	size_t size;
@@ -280,8 +293,14 @@ process_confirm(struct request *request)
 		diag_set(OutOfMemory, size, "region_alloc_object", "lsn");
 		return -1;
 	}
-	if (xrow_decode_confirm(request->header, &replica_id, lsn) != 0)
+	int res = 0;
+	if (is_confirm)
+		res = xrow_decode_confirm(request->header, &replica_id, lsn);
+	else
+		res = xrow_decode_rollback(request->header, &replica_id, lsn);
+	if (res == -1)
 		return -1;
+
 	/*
 	 * on_commit trigger failure is not allowed, so check for
 	 * instance id early.
@@ -294,7 +313,7 @@ process_confirm(struct request *request)
 
 	/*
 	 * Set an on_commit trigger which will perform the actual
-	 * confirmation processing.
+	 * confirmation/rollback processing.
 	 */
 	struct trigger *trig = region_alloc_object(&txn->region, typeof(*trig),
 						   &size);
@@ -302,7 +321,9 @@ process_confirm(struct request *request)
 		diag_set(OutOfMemory, size, "region_alloc_object", "trig");
 		return -1;
 	}
-	trigger_create(trig, applier_on_confirm, lsn, NULL);
+	trigger_create(trig, is_confirm ? applier_on_confirm_written :
+					  applier_on_rollback_written,
+		       lsn, NULL);
 
 	if (txn_begin_stmt(txn, NULL) != 0)
 		return -1;
@@ -319,9 +340,10 @@ static int
 apply_row(struct xrow_header *row)
 {
 	struct request request;
-	if (row->type == IPROTO_CONFIRM) {
+	if (iproto_type_is_synchro_request(row->type)) {
 		request.header = row;
-		return process_confirm(&request);
+		return process_confirm_rollback(&request,
+						row->type == IPROTO_CONFIRM);
 	}
 	if (xrow_decode_dml(row, &request, dml_request_key_map(row->type)) != 0)
 		return -1;
@@ -344,7 +366,7 @@ apply_final_join_row(struct xrow_header *row)
 	 * Confirms are ignored during join. All the data master
 	 * sends us is valid.
 	 */
-	if (row->type == IPROTO_CONFIRM)
+	if (iproto_type_is_synchro_request(row->type))
 		return 0;
 	struct txn *txn = txn_begin();
 	if (txn == NULL)
diff --git a/src/box/relay.cc b/src/box/relay.cc
index 0adc9fc98..29588b6ca 100644
--- a/src/box/relay.cc
+++ b/src/box/relay.cc
@@ -772,7 +772,7 @@ relay_send_row(struct xstream *stream, struct xrow_header *packet)
 {
 	struct relay *relay = container_of(stream, struct relay, stream);
 	assert(iproto_type_is_dml(packet->type) ||
-	       packet->type == IPROTO_CONFIRM);
+	       iproto_type_is_synchro_request(packet->type));
 	if (packet->group_id == GROUP_LOCAL) {
 		/*
 		 * We do not relay replica-local rows to other
diff --git a/src/box/txn.c b/src/box/txn.c
index 4f787db79..484b822db 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -751,7 +751,10 @@ txn_commit(struct txn *txn)
 	if (is_sync) {
 		txn_limbo_assign_lsn(&txn_limbo, limbo_entry,
 				     req->rows[req->n_rows - 1]->lsn);
-		txn_limbo_wait_complete(&txn_limbo, limbo_entry);
+		if (txn_limbo_wait_complete(&txn_limbo, limbo_entry) < 0) {
+			txn_free(txn);
+			return -1;
+		}
 	}
 	if (!txn_has_flag(txn, TXN_IS_DONE)) {
 		txn->signature = req->res;
diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index a715a136e..c55f5bda1 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -84,6 +84,16 @@ txn_limbo_remove(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 	rlist_del_entry(entry, in_queue);
 }
 
+static inline void
+txn_limbo_pop(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
+{
+	assert(!rlist_empty(&entry->in_queue));
+	assert(rlist_last_entry(&limbo->queue, struct txn_limbo_entry,
+				 in_queue) == entry);
+	(void) limbo;
+	rlist_del_entry(entry, in_queue);
+}
+
 void
 txn_limbo_abort(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 {
@@ -116,7 +126,11 @@ txn_limbo_check_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 	return entry->is_commit;
 }
 
-void
+static int
+txn_limbo_write_rollback(struct txn_limbo *limbo,
+			 struct txn_limbo_entry *entry);
+
+int
 txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 {
 	struct txn *txn = entry->txn;
@@ -125,33 +139,61 @@ txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
 	assert(txn_has_flag(txn, TXN_WAIT_ACK));
 	if (txn_limbo_check_complete(limbo, entry)) {
 		txn_limbo_remove(limbo, entry);
-		return;
+		return 0;
 	}
 	bool cancellable = fiber_set_cancellable(false);
 	bool timed_out = fiber_yield_timeout(txn_limbo_confirm_timeout(limbo));
 	fiber_set_cancellable(cancellable);
 	if (timed_out) {
-		// TODO: implement rollback.
-		entry->is_rollback = true;
+		txn_limbo_write_rollback(limbo, entry);
+		struct txn_limbo_entry *e, *tmp;
+		rlist_foreach_entry_safe_reverse(e, &limbo->queue,
+						 in_queue, tmp) {
+			e->is_rollback = true;
+			e->txn->signature = -1;
+			txn_limbo_pop(limbo, e);
+			txn_clear_flag(e->txn, TXN_WAIT_ACK);
+			txn_complete(e->txn);
+			if (e == entry)
+				break;
+			fiber_wakeup(e->txn->fiber);
+		}
+		return -1;
 	}
 	assert(txn_limbo_entry_is_complete(entry));
+	/*
+	 * The first tx to be rolled back already performed all
+	 * the necessary cleanups for us.
+	 */
+	if (entry->is_rollback)
+		return -1;
 	txn_limbo_remove(limbo, entry);
 	txn_clear_flag(txn, TXN_WAIT_ACK);
+	return 0;
 }
 
-/**
- * Write a confirmation entry to WAL. After it's written all the
- * transactions waiting for confirmation may be finished.
- */
 static int
-txn_limbo_write_confirm(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
+txn_limbo_write_confirm_rollback(struct txn_limbo *limbo,
+				 struct txn_limbo_entry *entry,
+				 bool is_confirm)
 {
 	struct xrow_header row;
 	struct request request = {
 		.header = &row,
 	};
 
-	if (xrow_encode_confirm(&row, limbo->instance_id, entry->lsn) < 0)
+	int res = 0;
+	if (is_confirm) {
+		res = xrow_encode_confirm(&row, limbo->instance_id, entry->lsn);
+	} else {
+		/*
+		 * This entry is the first to be rolled back, so
+		 * the last "safe" lsn is entry->lsn - 1.
+		 */
+		res = xrow_encode_rollback(&row, limbo->instance_id,
+					   entry->lsn - 1);
+	}
+	if (res == -1)
 		return -1;
 
 	struct txn *txn = txn_begin();
@@ -169,6 +211,17 @@ rollback:
 	return -1;
 }
 
+/**
+ * Write a confirmation entry to WAL. After it's written all the
+ * transactions waiting for confirmation may be finished.
+ */
+static int
+txn_limbo_write_confirm(struct txn_limbo *limbo,
+			struct txn_limbo_entry *entry)
+{
+	return txn_limbo_write_confirm_rollback(limbo, entry, true);
+}
+
 void
 txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn)
 {
@@ -191,6 +244,38 @@ txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn)
 	}
 }
 
+/**
+ * Write a rollback message to WAL. After it's written
+ * all the tarnsactions following the current one and waiting
+ * for confirmation must be rolled back.
+ */
+static int
+txn_limbo_write_rollback(struct txn_limbo *limbo,
+			 struct txn_limbo_entry *entry)
+{
+	return txn_limbo_write_confirm_rollback(limbo, entry, false);
+}
+
+void
+txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
+{
+	assert(limbo->instance_id != REPLICA_ID_NIL &&
+	       limbo->instance_id != instance_id);
+	struct txn_limbo_entry *e, *tmp;
+	rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
+		if (e->lsn < lsn)
+			break;
+		assert(e->txn->fiber == NULL);
+		e->is_rollback = true;
+		txn_limbo_pop(limbo, e);
+		txn_clear_flag(e->txn, TXN_WAIT_ACK);
+
+		/* Rollback the transaction. */
+		e->txn->signature = -1;
+		txn_complete(e->txn);
+	}
+}
+
 void
 txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn)
 {
@@ -214,7 +299,10 @@ txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn)
 	}
 	if (last_quorum != NULL) {
 		if (txn_limbo_write_confirm(limbo, last_quorum) != 0) {
-			// TODO: rollback.
+			// TODO: what to do here?.
+			// We already failed writing the CONFIRM
+			// message. What are the chances we'll be
+			// able to write ROLLBACK?
 			return;
 		}
 		/*
diff --git a/src/box/txn_limbo.h b/src/box/txn_limbo.h
index 23019e5d9..987cf9271 100644
--- a/src/box/txn_limbo.h
+++ b/src/box/txn_limbo.h
@@ -156,8 +156,12 @@ txn_limbo_ack(struct txn_limbo *limbo, uint32_t replica_id, int64_t lsn);
 /**
  * Block the current fiber until the transaction in the limbo
  * entry is either committed or rolled back.
+ * If timeout is reached before acks are collected, the tx is
+ * rolled back as well as all the txs in the limbo following it.
+ * Returns -1 when rollback was performed and tx has to be freed.
+ *          0 when tx processing can go on.
  */
-void
+int
 txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry);
 
 /**
@@ -166,6 +170,12 @@ txn_limbo_wait_complete(struct txn_limbo *limbo, struct txn_limbo_entry *entry);
 void
 txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn);
 
+/**
+ * Rollback all the entries  starting with given master's LSN.
+ */
+void
+txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn);
+
 /**
  * Return TRUE if limbo is empty.
  */
-- 
2.24.3 (Apple Git-128)

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Vladislav Shpilevoy]

Thanks for the patch!

> diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
> index a715a136e..c55f5bda1 100644
> --- a/src/box/txn_limbo.c
> +++ b/src/box/txn_limbo.c
> @@ -84,6 +84,16 @@ txn_limbo_remove(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
>  	rlist_del_entry(entry, in_queue);
>  }
>  
> +static inline void
> +txn_limbo_pop(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
> +{
> +	assert(!rlist_empty(&entry->in_queue));
> +	assert(rlist_last_entry(&limbo->queue, struct txn_limbo_entry,
> +				 in_queue) == entry);
> +	(void) limbo;
> +	rlist_del_entry(entry, in_queue);
> +}

txn_limbo_remove is exactly the same as txn_limbo_pop. I suggest to keep
one of them.

Everything else looks nice.

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Serge Petrenko]

19.06.2020 01:15, Vladislav Shpilevoy пишет:
> Thanks for the patch!

Thanks for the review!

>
>> diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
>> index a715a136e..c55f5bda1 100644
>> --- a/src/box/txn_limbo.c
>> +++ b/src/box/txn_limbo.c
>> @@ -84,6 +84,16 @@ txn_limbo_remove(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
>>   	rlist_del_entry(entry, in_queue);
>>   }
>>   
>> +static inline void
>> +txn_limbo_pop(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
>> +{
>> +	assert(!rlist_empty(&entry->in_queue));
>> +	assert(rlist_last_entry(&limbo->queue, struct txn_limbo_entry,
>> +				 in_queue) == entry);
>> +	(void) limbo;
>> +	rlist_del_entry(entry, in_queue);
>> +}
> txn_limbo_remove is exactly the same as txn_limbo_pop. I suggest to keep
> one of them.
>
> Everything else looks nice.

Assertions are different. I wanted to stress  that `pop` removes entries 
starting

from the tail, and `remove`, on the contrary,  removes them starting 
from the

head.

Looks strange, though, I agree.

If  you merge the functions and put an assertion

`rlist_first_entry == ... || rlist_last_entry == ...`

you'll lose some strictness in their use.

Feel free to decide what to do.


-- 
Serge Petrenko

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Vladislav Shpilevoy]

>>> diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
>>> index a715a136e..c55f5bda1 100644
>>> --- a/src/box/txn_limbo.c
>>> +++ b/src/box/txn_limbo.c
>>> @@ -84,6 +84,16 @@ txn_limbo_remove(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
>>>       rlist_del_entry(entry, in_queue);
>>>   }
>>>   +static inline void
>>> +txn_limbo_pop(struct txn_limbo *limbo, struct txn_limbo_entry *entry)
>>> +{
>>> +    assert(!rlist_empty(&entry->in_queue));
>>> +    assert(rlist_last_entry(&limbo->queue, struct txn_limbo_entry,
>>> +                 in_queue) == entry);
>>> +    (void) limbo;
>>> +    rlist_del_entry(entry, in_queue);
>>> +}
>> txn_limbo_remove is exactly the same as txn_limbo_pop. I suggest to keep
>> one of them.
>>
>> Everything else looks nice.
> 
> Assertions are different. I wanted to stress  that `pop` removes entries starting
> 
> from the tail, and `remove`, on the contrary,  removes them starting from the
> 
> head.

I didn't notice the assertions are different. In that case it is fine.

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Serge Petrenko]

18.06.2020 15:14, Serge Petrenko пишет:
> Now txn_limbo writes a ROLLBACK entry to WAL when one of the limbo
> entries fails to gather quorum during a txn_limbo_confirm_timeout.
> All the limbo entries, starting with the failed one, are rolled back in
> reverse order.
>
> Closes #4848

Added a new commit:

commit c3c3d6739add2d26b45e6ba0fd571b502b125c57
Author: Serge Petrenko <sergepetrenko@tarantool.org>
Date:   Fri Jun 19 08:30:43 2020 +0300

     Fix ROLLBACK handling during recovery and in row encoding

     [TO BE SQUASHED INTO THE PREVIOUS COMMIT]

diff --git a/src/box/box.cc b/src/box/box.cc
index 23c5aed95..8ba7ffafb 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -343,7 +343,7 @@ apply_wal_row(struct xstream *stream, struct 
xrow_header *row)
  {
         struct request request;
         // TODO: process confirmation during recovery.
-       if (row->type == IPROTO_CONFIRM)
+       if (iproto_type_is_synchro_request(row->type))
                 return;
         xrow_decode_dml_xc(row, &request, dml_request_key_map(row->type));
         if (request.type != IPROTO_NOP) {
diff --git a/src/box/txn.c b/src/box/txn.c
index 52e1c36dd..2360ecae3 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -83,10 +83,10 @@ txn_add_redo(struct txn *txn, struct txn_stmt *stmt, 
struct request *request)
         struct space *space = stmt->space;
         row->group_id = space != NULL ? space_group_id(space) : 0;
         /*
-        * IPROTO_CONFIRM entries are supplementary and aren't
-        * valid dml requests. They're encoded manually.
+        * Sychronous replication entries are supplementary and
+        * aren't valid dml requests. They're encoded manually.
          */
-       if (likely(row->type != IPROTO_CONFIRM))
+       if (likely(!iproto_type_is_synchro_request(row->type)))
                 row->bodycnt = xrow_encode_dml(request, &txn->region, 
row->body);
         if (row->bodycnt < 0)
                 return -1;


>    * Return TRUE if limbo is empty.
>    */

-- 
Serge Petrenko

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Serge Petrenko]

18.06.2020 15:14, Serge Petrenko пишет:
> Now txn_limbo writes a ROLLBACK entry to WAL when one of the limbo
> entries fails to gather quorum during a txn_limbo_confirm_timeout.
> All the limbo entries, starting with the failed one, are rolled back in
> reverse order.
>
> Closes #4848


New commit:

commit 2ee0c4380f17658b0fd1a507d565cd79b6910e3d
Author: Serge Petrenko <sergepetrenko@tarantool.org>
Date:   Mon Jun 22 21:08:49 2020 +0300

     fix for 'txn_limbo: add ROLLBACK processing'

     Fix parameter use inside applier_on_rollback_written().

     Make applier_txn_rollback_cb() respect rollback reason in
     txn->signature.

     Make limbo set rollback reason after timeout or read ROLLBACK message.

     [TO BE SQUASHED INTO THE PREVIOUS COMMIT]

diff --git a/src/box/applier.cc b/src/box/applier.cc
index 98a140a57..774af0149 100644
--- a/src/box/applier.cc
+++ b/src/box/applier.cc
@@ -273,10 +273,10 @@ applier_on_confirm_written(struct trigger *trig, 
void *event)
   * Rolls back part of the txs waiting in limbo.
   */
  static int
-applier_on_rollback_written(struct trigger *trig, void *data)
+applier_on_rollback_written(struct trigger *trig, void *event)
  {
-    (void) trig;
-    int64_t lsn = *(int64_t *)data;
+    (void) event;
+    int64_t lsn = *(int64_t *)trig->data;
      txn_limbo_read_rollback(&txn_limbo, lsn);
      return 0;
  }
@@ -801,6 +801,14 @@ static int
  applier_txn_rollback_cb(struct trigger *trigger, void *event)
  {
      (void) trigger;
+    struct txn *txn = (struct txn *) event;
+    /*
+     * Synchronous transaction rollback due to receiving a
+     * ROLLBACK entry is a normal event and requires no
+     * special handling.
+     */
+    if (txn->signature == TXN_SIGNATURE_SYNC_ROLLBACK)
+        return 0;

      /*
       * Setup shared applier diagnostic area.
diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
index 03badf9fc..931f5c3d4 100644
--- a/src/box/txn_limbo.c
+++ b/src/box/txn_limbo.c
@@ -150,7 +150,7 @@ txn_limbo_wait_complete(struct txn_limbo *limbo, 
struct txn_limbo_entry *entry)
          rlist_foreach_entry_safe_reverse(e, &limbo->queue,
                           in_queue, tmp) {
              e->is_rollback = true;
-            e->txn->signature = -1;
+            e->txn->signature = TXN_SIGNATURE_QUORUM_TIMEOUT;
              txn_limbo_pop(limbo, e);
              txn_clear_flag(e->txn, TXN_WAIT_ACK);
              txn_complete(e->txn);
@@ -274,7 +274,7 @@ txn_limbo_read_rollback(struct txn_limbo *limbo, 
int64_t lsn)
          txn_clear_flag(e->txn, TXN_WAIT_ACK);

          /* Rollback the transaction. */
-        e->txn->signature = -1;
+        e->txn->signature = TXN_SIGNATURE_SYNC_ROLLBACK;
          txn_complete(e->txn);
      }
  }

-- 
Serge Petrenko

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Vladislav Shpilevoy]

> diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
> index a715a136e..c55f5bda1 100644
> --- a/src/box/txn_limbo.c
> +++ b/src/box/txn_limbo.c
> @@ -191,6 +244,38 @@ txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn)
> +void
> +txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
> +{
> +	assert(limbo->instance_id != REPLICA_ID_NIL &&
> +	       limbo->instance_id != instance_id);
> +	struct txn_limbo_entry *e, *tmp;
> +	rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
> +		if (e->lsn < lsn)
> +			break;

Shouldn't this be 'continue' instead of 'break'? As I understand rollback,
we need to find entry, *from* which all the entries will be rolled back.
Here it seems that if the oldest entry in the limbo (with the smallest LSN)
is smaller than rollback lsn, we just won't rollback anything.

> +		assert(e->txn->fiber == NULL);
> +		e->is_rollback = true;
> +		txn_limbo_pop(limbo, e);
> +		txn_clear_flag(e->txn, TXN_WAIT_ACK);
> +
> +		/* Rollback the transaction. */
> +		e->txn->signature = -1;
> +		txn_complete(e->txn);
> +	}
> +}

Re: [Tarantool-patches] [PATCH 4/4] txn_limbo: add ROLLBACK processing

[Vladislav Shpilevoy]

On 26/06/2020 00:14, Vladislav Shpilevoy wrote:
>> diff --git a/src/box/txn_limbo.c b/src/box/txn_limbo.c
>> index a715a136e..c55f5bda1 100644
>> --- a/src/box/txn_limbo.c
>> +++ b/src/box/txn_limbo.c
>> @@ -191,6 +244,38 @@ txn_limbo_read_confirm(struct txn_limbo *limbo, int64_t lsn)
>> +void
>> +txn_limbo_read_rollback(struct txn_limbo *limbo, int64_t lsn)
>> +{
>> +	assert(limbo->instance_id != REPLICA_ID_NIL &&
>> +	       limbo->instance_id != instance_id);
>> +	struct txn_limbo_entry *e, *tmp;
>> +	rlist_foreach_entry_safe_reverse(e, &limbo->queue, in_queue, tmp) {
>> +		if (e->lsn < lsn)
>> +			break;
> 
> Shouldn't this be 'continue' instead of 'break'? As I understand rollback,
> we need to find entry, *from* which all the entries will be rolled back.
> Here it seems that if the oldest entry in the limbo (with the smallest LSN)
> is smaller than rollback lsn, we just won't rollback anything.

Sorry, didn't notice it is 'foreach reverse'. Not just 'foreach'.