[Tarantool-patches] [PATCH 2/2] box: introduce a cfg handle to become syncro leader

[Vladislav Shpilevoy]

Hi! Thanks for the patch!

I didn't review it properly yet. Just one comment.

> diff --git a/src/box/box.cc b/src/box/box.cc
> index ca24b98ca..087710383 100644
> --- a/src/box/box.cc
> +++ b/src/box/box.cc
> @@ -945,6 +946,84 @@ box_set_replication_anon(void)
>  
>  }
>  
> +void
> +box_set_replication_synchro_leader(void)
> +{
> +	bool is_leader = cfg_geti("replication_synchro_leader");
> +	/*
> +	 * For now no actions required when an instance stops
> +	 * being a leader. We should probably wait until txn_limbo
> +	 * becomes empty.
> +	 */
> +	if (!is_leader)
> +		return;
> +	uint32_t former_leader_id = txn_limbo.instance_id;
> +	if (former_leader_id == REPLICA_ID_NIL ||
> +	    former_leader_id == instance_id) {
> +		return;
> +	}
> +
> +	/* Wait until pending confirmations/rollbacks reach us. */
> +	double timeout = 2 * txn_limbo_confirm_timeout(&txn_limbo);
> +	double start_tm = fiber_time();
> +	while (!txn_limbo_is_empty(&txn_limbo)) {
> +		if (fiber_time() - start_tm > timeout)
> +			break;
> +		fiber_sleep(0.001);
> +	}
> +
> +	if (!txn_limbo_is_empty(&txn_limbo)) {
> +		int64_t lsns[VCLOCK_MAX];
> +		int len = 0;
> +		const struct vclock  *vclock;
> +		replicaset_foreach(replica) {
> +			if (replica->relay != NULL &&
> +			    relay_get_state(replica->relay) != RELAY_OFF &&
> +			    !replica->anon) {
> +				assert(!tt_uuid_is_equal(&INSTANCE_UUID,
> +							 &replica->uuid));
> +				vclock = relay_vclock(replica->relay);
> +				int64_t lsn = vclock_get(vclock,
> +							 former_leader_id);
> +				lsns[len++] = lsn;
> +			}
> +		}
> +		lsns[len++] = vclock_get(box_vclock, former_leader_id);
> +		assert(len < VCLOCK_MAX);
> +
> +		int64_t confirm_lsn = 0;
> +		if (len >= replication_synchro_quorum) {
> +			qsort(lsns, len, sizeof(int64_t), cmp_i64);
> +			confirm_lsn = lsns[len - replication_synchro_quorum];
> +		}
> +

Can the code below be moved to txn_limbo.c somehow? Doesn't look
right to touch the queue here.

> +		struct txn_limbo_entry *e, *last_quorum = NULL;
> +		struct txn_limbo_entry *rollback = NULL;
> +		rlist_foreach_entry(e, &txn_limbo.queue, in_queue) {
> +			if (txn_has_flag(e->txn, TXN_WAIT_ACK)) {
> +				if (e->lsn <= confirm_lsn) {
> +					last_quorum = e;
> +				} else {
> +					rollback = e;
> +					break;
> +				}
> +			}
> +		}
> +
> +		if (last_quorum != NULL) {
> +			confirm_lsn = last_quorum->lsn;
> +			txn_limbo_write_confirm(&txn_limbo, confirm_lsn);
> +			txn_limbo_read_confirm(&txn_limbo, confirm_lsn);
> +		}
> +		if (rollback != NULL) {
> +			txn_limbo_write_rollback(&txn_limbo, rollback->lsn);
> +			txn_limbo_read_rollback(&txn_limbo, rollback->lsn - 1);
> +		}
> +
> +		assert(txn_limbo_is_empty(&txn_limbo));
> +	}
> +}