Re: [PATCH] Reuse Workers and Replication Slots during Logical Replication

On Fri, Jul 21, 2023 at 7:30 AM Peter Smith <smithpb2250@gmail.com> wrote:
>
> ~~~
>
> 2. StartLogRepWorker
>
> /* Common function to start the leader apply or tablesync worker. */
> void
> StartLogRepWorker(int worker_slot)
> {
> /* Attach to slot */
> logicalrep_worker_attach(worker_slot);
>
> /* Setup signal handling */
> pqsignal(SIGHUP, SignalHandlerForConfigReload);
> pqsignal(SIGTERM, die);
> BackgroundWorkerUnblockSignals();
>
> /*
> * We don't currently need any ResourceOwner in a walreceiver process, but
> * if we did, we could call CreateAuxProcessResourceOwner here.
> */
>
> /* Initialise stats to a sanish value */
> MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
> MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
>
> /* Load the libpq-specific functions */
> load_file("libpqwalreceiver", false);
>
> InitializeLogRepWorker();
>
> /* Connect to the origin and start the replication. */
> elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
> MySubscription->conninfo);
>
> /*
> * Setup callback for syscache so that we know when something changes in
> * the subscription relation state.
> */
> CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
>   invalidate_syncing_table_states,
>   (Datum) 0);
> }
>
> ~
>
> 2a.
> The function name seems a bit misleading because it is not really
> "starting" anything here - it is just more "initialization" code,
> right? Nor is it common to all kinds of LogRepWorker. Maybe the
> function could be named something else like 'InitApplyOrSyncWorker()'.
> -- see also #2c
>

How about SetupLogRepWorker? The other thing I noticed is that we
don't seem to be consistent in naming functions in these files. For
example, shall we make all exposed functions follow camel case (like
InitializeLogRepWorker) and static functions follow _ style (like
run_apply_worker) or the other possibility is to use _ style for all
functions except may be the entry functions like ApplyWorkerMain()? I
don't know if there is already a pattern but if not then let's form it
now, so that code looks consistent.

> ~
>
> 2b.
> Should this have Assert to ensure this is only called from leader
> apply or tablesync? -- see also #2c
>
> ~
>
> 2c.
> IMO maybe the best/tidiest way to do this is not to introduce a new
> function at all. Instead, just put all this "common init" code into
> the existing "common init" function ('InitializeLogRepWorker') and
> execute it only if (am_tablesync_worker() || am_leader_apply_worker())
> { }.
>

I don't like 2c much because it will make InitializeLogRepWorker()
have two kinds of initializations.

> ======
> src/include/replication/worker_internal.h
>
> 3.
>  extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
>      XLogRecPtr remote_lsn);
> +extern void set_stream_options(WalRcvStreamOptions *options,
> +    char *slotname,
> +    XLogRecPtr *origin_startpos);
> +
> +extern void start_apply(XLogRecPtr origin_startpos);
> +extern void DisableSubscriptionAndExit(void);
> +extern void StartLogRepWorker(int worker_slot);
>
> This placement (esp. with the missing whitespace) seems to be grouping
> the set_stream_options with the other 'pa' externs, which are all
> under the comment "/* Parallel apply worker setup and interactions
> */".
>
> Putting all these up near the other "extern void
> InitializeLogRepWorker(void)" might be less ambiguous.
>

+1. Also, note that they should be in the same order as they are in .c files.

--
With Regards,
Amit Kapila.