From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from smtp17.mail.ru (smtp17.mail.ru [94.100.176.154]) (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 F35C64765E9 for ; Wed, 23 Dec 2020 15:03:59 +0300 (MSK) From: Serge Petrenko References: <449ae5e1cd4b4e22f55ff53a4404717c0229638a.1607633488.git.sergepetrenko@tarantool.org> <2f5e637e-5742-d9ea-dbf9-1092c1666b16@tarantool.org> <4b7f4fc1-6d48-4332-c432-1eeb0b28c016@tarantool.org> Message-ID: Date: Wed, 23 Dec 2020 15:03:59 +0300 MIME-Version: 1.0 In-Reply-To: <4b7f4fc1-6d48-4332-c432-1eeb0b28c016@tarantool.org> Content-Type: text/plain; charset="utf-8"; format="flowed" Content-Transfer-Encoding: 8bit Content-Language: en-GB Subject: Re: [Tarantool-patches] [PATCH 3/4] relay: introduce relay_lsn_watcher List-Id: Tarantool development patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: Vladislav Shpilevoy , Cyrill Gorcunov Cc: tarantool-patches@dev.tarantool.org 21.12.2020 18:31, Serge Petrenko пишет: > > > 18.12.2020 00:43, Vladislav Shpilevoy пишет: >> Thanks for the patch and for the design! Thanks for the review! >> >> See 3 comments below. >> >> On 10.12.2020 21:55, Serge Petrenko via Tarantool-patches wrote: >>> Add a list of lsn watchers to relay. >>> >>> Relay_lsn_watcher is a structure that lives in tx thread and monitors >>> replica's progress in applying rows coming from a given replica id. >>> >>> The watcher fires once relay reports that replica has acked the target lsn >>> for the given replica id. >>> >>> The watcher is owned by some tx fiber, which typically sleeps until the >>> watcher wakes it up. Besides waking the waiting fiber up, the watcher may >>> perform some work as dictated by the notify function. >>> >>> Prerequisite #5435 >>> --- >>> src/box/relay.cc | 73 ++++++++++++++++++++++++++++++++++++++++++++++-- >>> src/box/relay.h | 44 +++++++++++++++++++++++++++++ >>> 2 files changed, 114 insertions(+), 3 deletions(-) >>> >>> diff --git a/src/box/relay.cc b/src/box/relay.cc >>> index f342a79dd..4014792a6 100644 >>> --- a/src/box/relay.cc >>> +++ b/src/box/relay.cc >>> @@ -394,6 +397,54 @@ relay_final_join(int fd, uint64_t sync, struct vclock *start_vclock, >>> }); >>> } >>> >>> +void >>> +relay_set_lsn_watcher(struct relay *relay, struct relay_lsn_watcher *watcher) >>> +{ >>> + rlist_add_tail_entry(&relay->tx.lsn_watchers, watcher, in_list); >>> +} >>> + >>> +void >>> +relay_lsn_watcher_create(struct relay_lsn_watcher *watcher, uint32_t replica_id, >>> + int64_t target_lsn, relay_lsn_watcher_f notify, >>> + relay_lsn_watcher_f destroy, void *data) >>> +{ >>> + watcher->replica_id = replica_id; >>> + watcher->target_lsn = target_lsn; >>> + watcher->notify = notify; >>> + watcher->destroy = destroy; >>> + watcher->data = data; >>> + watcher->waiter = fiber(); >>> +} >> 1. The notify + destroy + invocation of the watchers looks very similar >> to what trigger.h API offers. Can you try to implement this as triggers? >> I mean as a list of struct trigger objects. > > Yes, it's possible  to implement this as a trigger. It'll be called on > every > tx_status_update() invocation then. P.S. I ended up implementing this watcher as a trigger. > >>> + >>> +/** >>> + * Destroy the watcher. >>> + * Wake the waiting fiber up, fire the on destroy callback and remove the >>> + * watcher from the relay's watcher list. >>> + */ >>> +static void >>> +relay_lsn_watcher_destroy(struct relay_lsn_watcher *watcher) >>> +{ >>> + watcher->destroy(watcher->data); >>> + fiber_wakeup(watcher->waiter); >>> + rlist_del_entry(watcher, in_list); >>> +} >>> + >>> +/** >>> + * Notify the watcher that the replica has advanced to the given vclock. >>> + * In case target_lsn is hit for watcher's replica_id, fire the notify >>> + * callback and destroy the watcher. >>> + */ >>> +static void >>> +relay_lsn_watcher_notify(struct relay_lsn_watcher *watcher, >>> + struct vclock *vclock) >>> +{ >>> + int64_t lsn = vclock_get(vclock, watcher->replica_id); >>> + if (lsn >= watcher->target_lsn) { >>> + watcher->notify(watcher->data); >>> + relay_lsn_watcher_destroy(watcher); >>> + } >>> +} >> 2. This all seems too intricate. Why does the watcher destroys itself? >> Wouldn't it be easier to let the watcher decide if he wants to die, or >> change LSN and wait more, or whatever else? >> >> I have a feeling that once you will try to switch to struct triggers, >> it will all come together automatically. Since triggers API is notably >> hard to misuse. > > Sounds good. > >>> + >>> /** >>> * The message which updated tx thread with a new vclock has returned back >>> * to the relay. >>> diff --git a/src/box/relay.h b/src/box/relay.h >>> index b32e2ea2a..da2e3498a 100644 >>> --- a/src/box/relay.h >>> +++ b/src/box/relay.h >>> @@ -134,4 +135,47 @@ relay_subscribe(struct replica *replica, int fd, uint64_t sync, >>> struct vclock *replica_vclock, uint32_t replica_version_id, >>> uint32_t replica_id_filter); >>> >>> +/** >>> + * A callback invoked once lsn watcher sees the replica has reached the target >>> + * lsn for the given replica id. >>> + */ >>> +typedef void (*relay_lsn_watcher_f)(void *data); >>> + >>> +/** >>> + * A relay watcher which notifies tx via calling a given function once the >>> + * replica confirms it has received the rows from replica_id up to target_lsn. >>> + */ >>> +struct relay_lsn_watcher { >>> + /** A replica_id to monitor rows from. */ >>> + uint32_t replica_id; >>> + /** The lsn to wait for. */ >>> + int64_t target_lsn; >>> + /** >>> + * A callback invoked in the tx thread once the watcher sees that >>> + * replica has reached the target lsn. >>> + */ >>> + relay_lsn_watcher_f notify; >>> + /** >>> + * A callback fired once the relay thread exits to notify the waiter >>> + * there's nothing to wait for anymore. >>> + */ >>> + relay_lsn_watcher_f destroy; >>> + /** Data passed to \a notify and \a destroy. */ >> 3. Lets better use @a. To be consistent with all the new code (except >> cases when a whole file uses tons of \a). > > Ok, thanks for pointing this out! >>> + void *data; >>> + /** A fiber waiting for this watcher to fire. */ >>> + struct fiber *waiter; >>> + /** A link in relay's watcher list. */ >>> + struct rlist in_list; >>> +}; > > -- > Serge Petrenko -- Serge Petrenko