Re: [Tarantool-patches] [PATCH] replication: add support of qsync to the snapshot machinery

[Serge Petrenko <sergepetrenko@tarantool.org>]

10.06.2020 17:34, Leonid Vasiliev пишет:
> To support qsync replication, the waiting for confirmation of
> current "sync" transactions during a timeout has been added to
> the snapshot machinery. In the case of rollback or the timeout
> expiration, the snapshot will be cancelled.
>
> Closes #4850
> ---
> Dirty version without tests.
>
> https://github.com/tarantool/tarantool/issues/4850
> a part of https://github.com/tarantool/tarantool/tree/gh-4842-sync-replication
>
>   src/box/gc.c        | 92 +++++++++++++++++++++++++++++++++++++++++++++++++++++
>   src/box/txn_limbo.h | 19 +++++++++++
>   2 files changed, 111 insertions(+)
>
> diff --git a/src/box/gc.c b/src/box/gc.c
> index 8e8ffea..fb7b37a 100644
> --- a/src/box/gc.c
> +++ b/src/box/gc.c
> @@ -57,6 +57,9 @@
>   #include "engine.h"		/* engine_collect_garbage() */
>   #include "wal.h"		/* wal_collect_garbage() */
>   #include "checkpoint_schedule.h"
> +#include "trigger.h"
> +#include "txn.h"
> +#include "txn_limbo.h"
>   
>   struct gc_state gc;
>   
> @@ -65,6 +68,26 @@ gc_cleanup_fiber_f(va_list);
>   static int
>   gc_checkpoint_fiber_f(va_list);
>   
> +//TODO: quorum timeout should be used instead
> +double snap_confirm_timeout = 5.0; /* seconds */
> +
> +/**
> + * Waitpoint stores information about the progress of confirmation.
> + * In the case of multimaster support, it will store a bitset
> + * or array instead of the boolean.
> + */
> +struct confirm_waitpoint {
> +	/**
> +	 * Variable for wake up the fiber that is waiting for
> +	 * the end of confirmation.
> +	 */
> +	struct fiber_cond confirm_cond;
> +	/**
> +	 * Result flag.
> +	 */
> +	bool is_confirm;
> +};
> +
>   /**
>    * Comparator used for ordering gc_consumer objects
>    * lexicographically by their vclock in a binary tree.
> @@ -377,6 +400,64 @@ gc_add_checkpoint(const struct vclock *vclock)
>   }
>   
>   static int
> +gc_txn_commit_cb(struct trigger *trigger, void *event)
> +{
> +	(void)event;
> +	struct confirm_waitpoint *cwp =
> +		(struct confirm_waitpoint *)trigger->data;
> +	cwp->is_confirm = true;
> +	fiber_cond_signal(&cwp->confirm_cond);
> +	return 0;
> +}
> +
> +static int
> +gc_txn_rollback_cb(struct trigger *trigger, void *event)
> +{
> +	(void)event;
> +	struct confirm_waitpoint *cwp =
> +		(struct confirm_waitpoint *)trigger->data;
> +	fiber_cond_signal(&cwp->confirm_cond);
> +	return 0;
> +}
> +
> +/**
> + * Waiting for confirmation of all "sync" transactions
> + * during snap_confirm_timeout or fail.
> + */
> +static int
> +gc_wait_confirm(void)
> +{
> +	/* initialization of a waitpoint. */
> +	struct confirm_waitpoint cwp;
> +	fiber_cond_create(&cwp.confirm_cond);
> +	cwp.is_confirm = false;
> +
> +	/* Set triggers for the last limbo transaction. */
> +	struct trigger on_complete;
> +	trigger_create(&on_complete, gc_txn_commit_cb, &cwp, NULL);
> +	struct trigger on_rollback;
> +	trigger_create(&on_rollback, gc_txn_rollback_cb, &cwp, NULL);
> +	struct txn_limbo_entry *tle = txn_limbo_last_entry(&txn_limbo);
> +	txn_on_commit(tle->txn, &on_complete);
> +	txn_on_rollback(tle->txn, &on_rollback);
> +
> +	int rc = fiber_cond_wait_timeout(&cwp.confirm_cond,
> +					 snap_confirm_timeout);
> +	fiber_cond_destroy(&cwp.confirm_cond);
> +	if (rc != 0) {
> +		/* Clear the triggers if the timeout has been reached. */
> +		trigger_clear(&on_complete);
> +		trigger_clear(&on_rollback);
> +		return -1;
> +	}
> +	if (!cwp.is_confirm) {
> +		/* The transaction has been rollbacked. */
> +		return -1;
> +	}
> +	return 0;
> +}
> +
> +static int
>   gc_do_checkpoint(bool is_scheduled)
>   {
>   	int rc;
> @@ -395,6 +476,17 @@ gc_do_checkpoint(bool is_scheduled)
>   	rc = wal_begin_checkpoint(&checkpoint);
>   	if (rc != 0)
>   		goto out;
> +
> +	/*
> +	 * Wait the confirms on all "sync" transactions before
> +	 * create a snapshot.
> +	 */
> +	if (!txn_limbo_is_empty(&txn_limbo)) {
> +		rc = gc_wait_confirm();
> +		if (rc != 0)
> +			goto out;
> +	}
> +
>   	rc = engine_commit_checkpoint(&checkpoint.vclock);
>   	if (rc != 0)
>   		goto out;
> diff --git a/src/box/txn_limbo.h b/src/box/txn_limbo.h
> index de415cd..c0b821a 100644
> --- a/src/box/txn_limbo.h
> +++ b/src/box/txn_limbo.h
> @@ -166,6 +166,25 @@ 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);
>   
> +/**
> + * Return TRUE if limbo is empty.
> + */
> +static inline bool
> +txn_limbo_is_empty(struct txn_limbo *limbo)
> +{
> +	return rlist_empty(&limbo->queue);
> +}
> +
> +/**
> + * Return a pointer to the last txn_limbo_entry of limbo.
> + */
> +static inline struct txn_limbo_entry *
> +txn_limbo_last_entry(struct txn_limbo *limbo)
> +{
> +	return rlist_last_entry(&limbo->queue, struct txn_limbo_entry,
> +				in_queue);
> +}
> +
>   void
>   txn_limbo_init();
>   


Thanks for the  patch! LGTM.

-- 
Serge Petrenko