From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from smtp63.i.mail.ru (smtp63.i.mail.ru [217.69.128.43]) (using TLSv1.2 with cipher ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)) (No client certificate requested) by dev.tarantool.org (Postfix) with ESMTPS id 070BE42F4AD for ; Thu, 11 Jun 2020 12:02:55 +0300 (MSK) References: From: Serge Petrenko Message-ID: Date: Thu, 11 Jun 2020 12:02:54 +0300 MIME-Version: 1.0 In-Reply-To: Content-Type: text/plain; charset="utf-8"; format="flowed" Content-Transfer-Encoding: 8bit Content-Language: ru Subject: Re: [Tarantool-patches] [PATCH] replication: add support of qsync to the snapshot machinery List-Id: Tarantool development patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: Leonid Vasiliev , v.shpilevoy@tarantool.org, gorcunov@gmail.com Cc: tarantool-patches@dev.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