From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from smtp20.mail.ru (smtp20.mail.ru [94.100.179.251]) (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 D0A5940F3AE for ; Wed, 10 Jun 2020 17:34:19 +0300 (MSK) From: Leonid Vasiliev Date: Wed, 10 Jun 2020 17:34:16 +0300 Message-Id: Subject: [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: v.shpilevoy@tarantool.org, gorcunov@gmail.com, sergepetrenko@tarantool.org Cc: tarantool-patches@dev.tarantool.org 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(); -- 2.7.4