RE: Perform streaming logical transactions by background workers and parallel apply

On Tues, November 22, 2022 13:20 PM Peter Smith <smithpb2250@gmail.com> wrote:
> Thanks for addressing my review comments on v47-0001.
> 
> Here are my review comments for v49-0001.

Thanks for your comments.

> ======
> 
> src/backend/replication/logical/applyparallelworker.c
> 
> 1. GENERAL - NULL checks
> 
> There is inconsistent NULL checking in the patch.
> 
> Sometimes it is like (!winfo)
> Sometimes explicit NULL checks like  (winfo->mq_handle != NULL)
> 
> (That is just one example -- there are differences in many places)
> 
> It would be better to use a consistent style everywhere.

Changed.

> ~
> 
> 2. GENERAL - Error message worker name
> 
> 2a.
> In worker.c all the messages are now "logical replication apply 
> worker" or "logical replication parallel apply worker" etc, but in the 
> applyparallel.c sometimes the "logical replication" part is missing.
> IMO all the messages in this patch should be consistently worded.
> 
> I've reported some of them in the following comment below, but please 
> search the whole patch for any I might have missed.

Rename LA and PA to the following styles:
```
LA -> logical replication apply worker
PA -> logical replication parallel apply worker ```

> 2b.
> Consider if maybe all of these ought to be calling get_worker_name() 
> which is currently static in worker.c. Doing this means any future 
> changes to get_worker_name won't cause more inconsistencies.

The most error message in applyparallelxx.c can only use "xx parallel worker",
so I think it's fine not to call get_worker_name

> ~~~
> 
> 3. File header comment
> 
> + * IDENTIFICATION 
> + src/backend/replication/logical/applyparallelworker.c
> 
> The word "IDENTIFICATION" should be on a separate line (for 
> consistency with every other PG source file)

Fixed.

> ~
> 
> 4.
> 
> + * In order for lmgr to detect this, we have LA acquire a session 
> + lock on the
> + * remote transaction (by pa_lock_stream()) and have PA wait on the 
> + lock
> before
> + * trying to receive messages. In other words, LA acquires the lock 
> + before
> + * sending STREAM_STOP and releases it if already acquired before 
> + sending
> + * STREAM_START, STREAM_ABORT(for toplevel transaction),
> STREAM_PREPARE and
> + * STREAM_COMMIT. For PA, it always needs to acquire the lock after
> processing
> + * STREAM_STOP and STREAM_ABORT(for subtransaction) and then release
> + * immediately after acquiring it. That way, when PA is waiting for 
> + LA, we can
> + * have a wait-edge from PA to LA in lmgr, which will make a deadlock 
> + in lmgr
> + * like:
> 
> Missing spaces before '(' deliberate?

Added.

> ~~~
> 
> 5. globals
> 
> +/*
> + * Is there a message sent by parallel apply worker which the leader 
> +apply
> + * worker need to receive?
> + */
> +volatile sig_atomic_t ParallelApplyMessagePending = false;
> 
> SUGGESTION
> Is there a message sent by a parallel apply worker that the leader 
> apply worker needs to receive?

Changed.

> ~~~
> 
> 6. pa_get_available_worker
> 
> +/*
> + * get an available parallel apply worker from the worker pool.
> + */
> +static ParallelApplyWorkerInfo *
> +pa_get_available_worker(void)
> 
> Uppercase comment

Changed.

> ~
> 
> 7.
> 
> + /*
> + * We first try to free the worker to improve our chances of getting
> + * the worker. Normally, we free the worker after ensuring that the
> + * transaction is committed by parallel worker but for rollbacks, we
> + * don't wait for the transaction to finish so can't free the worker
> + * information immediately.
> + */
> 
> 7a.
> "We first try to free the worker to improve our chances of getting the worker."
> 
> SUGGESTION
> We first try to free the worker to improve our chances of finding one 
> that is not in use.
> 
> ~
> 
> 7b.
> "parallel worker" -> "the parallel worker"

Changed.

> ~~~
> 
> 8. pa_allocate_worker
> 
> + /* Try to get a free parallel apply worker. */ winfo = 
> + pa_get_available_worker();
> +
> 
> SUGGESTION
> First, try to get a parallel apply worker from the pool.

Changed.

> ~~~
> 
> 9. pa_free_worker
> 
> + * This removes the parallel apply worker entry from the hash table 
> + so that it
> + * can't be used. This either stops the worker and free the 
> + corresponding info,
> + * if there are enough workers in the pool or just marks it available 
> + for
> + * reuse.
> 
> BEFORE
> This either stops the worker and free the corresponding info, if there 
> are enough workers in the pool or just marks it available for reuse.
> 
> SUGGESTION
> If there are enough workers in the pool it stops the worker and frees 
> the corresponding info, otherwise it just marks the worker as 
> available for reuse.

Changed.

> ~
> 
> 10.
> 
> + /* Free the corresponding info if the worker exited cleanly. */ if 
> + (winfo->error_mq_handle == NULL) { pa_free_worker_info(winfo); 
> + return true; }
> 
> Is it correct that this bypasses the removal from the hash table?

I rethink about this, it seems unnecessary to free the information here as
we don't expect the worker to stop unless the leader as them to stop.
So, I temporarily remove this and will think about this in next version.

> ~
> 
> 14.
> 
> + case 'X': /* Terminate, indicating clean exit. */ { 
> + shm_mq_detach(winfo->error_mq_handle);
> + winfo->error_mq_handle = NULL;
> + break;
> + }
> + default:
> 
> 
> No need for the { } here.

Changed.

> ~~~
> 
> 16. pa_init_and_launch_worker
> 
> + /* Setup shared memory. */
> + if (!pa_setup_dsm(winfo))
> + {
> + MemoryContextSwitchTo(oldcontext);
> + pfree(winfo);
> + return NULL;
> + }
> 
> 
> Wouldn't it be better to do the pfree before switching back to the oldcontext?

I think either style seems fine.

> ~~~
> 
> 17. pa_send_data
> 
> + /* Wait before retrying. */
> + rc = WaitLatch(MyLatch,
> +    WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
> +    SHM_SEND_RETRY_INTERVAL_MS,
> +    WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
> +
> + if (rc & WL_LATCH_SET)
> + {
> + ResetLatch(MyLatch);
> + CHECK_FOR_INTERRUPTS();
> + }
> 
> 
> Instead of CHECK_FOR_INTERRUPTS, should this be calling your new 
> function ProcessParallelApplyInterrupts?

I thought the ProcessParallelApplyInterrupts is intended to be invoked only in main
loop(LogicalParallelApplyLoop) to make the parallel apply worker exit cleanly.

> ~
> 
> 18.
> 
> + if (startTime == 0)
> + startTime = GetCurrentTimestamp();
> + else if (TimestampDifferenceExceeds(startTime, 
> + GetCurrentTimestamp(),
> + SHM_SEND_TIMEOUT_MS))
> + ereport(ERROR,
> + (errcode(ERRCODE_CONNECTION_FAILURE),
> + errmsg("terminating logical replication parallel apply worker due to
> timeout")));
> 
> 
> I'd previously commented that the timeout calculation seemed wrong.
> Hou-san replied [1,#9] "start counting from the first failure looks 
> fine to me." but I am not so sure - e.g. If the timeout is 10s then I 
> expect it to fail ~10s after the function is called, not 11s after. I 
> know it's pedantic, but where's the harm in making the calculation 
> right instead of just nearly right?
> 
> IMO probably an easy fix for this is like:
> 
> #define SHM_SEND_RETRY_INTERVAL_MS 1000 #define SHM_SEND_TIMEOUT_MS 
> (10000 - SHM_SEND_RETRY_INTERVAL_MS)

OK, I moved the place of setting startTime before the WaitLatch.

> ~~~
> 
> 20. pa_savepoint_name
> 
> +static void
> +pa_savepoint_name(Oid suboid, TransactionId xid, char *spname,
> +   Size szsp)
> 
> Unnecessary wrapping?

Changed.

> ======
> 
> src/backend/replication/logical/origin.c
> 
> 21. replorigin_session_setup
> 
> + * However, we do allow multiple processes to point to the same 
> + origin slot
> + * if requested by the caller by passing PID of the process that has 
> + already
> + * acquired it. This is to allow using the same origin by multiple 
> + parallel
> + * apply processes the provided they maintain commit order, for 
> + example, by
> + * allowing only one process to commit at a time.
> 
> 21a.
> I thought the comment should mention this is optional and the special 
> value acquired_by=0 means don't do this.

Added.

> ~
> 
> 21b.
> "the provided they" ?? typo?

Changed.

> ======
> 
> src/backend/replication/logical/tablesync.c
> 
> 22. process_syncing_tables
> 
>  process_syncing_tables(XLogRecPtr current_lsn)  {
> + /*
> + * Skip for parallel apply workers as they don't operate on tables 
> + that
> + * are not in ready state. See pa_can_start() and
> + * should_apply_changes_for_rel().
> + */
> + if (am_parallel_apply_worker())
> + return;
> 
> SUGGESTION (remove the double negative) Skip for parallel apply 
> workers because they only operate on tables that are in a READY state. 
> See pa_can_start() and should_apply_changes_for_rel().

Changed.

> ======
> 
> src/backend/replication/logical/worker.c
> 
> 23. apply_handle_stream_stop
> 
> 
> Previously I suggested that this lock/unlock seems too tricky and 
> needed a comment. The reply [1,#12] was that this is already described 
> atop parallelapplyworker.c. OK, but in that case maybe here the 
> comment can just refer to that explanation:
> 
> SUGGESTION
> Refer to the comments atop applyparallelworker.c for what this lock 
> and immediate unlock is doing.
> 
> ~~~
> 
> 24. apply_handle_stream_abort
> 
> + if 
> + (pg_atomic_sub_fetch_u32(&(MyParallelShared->pending_stream_count),
> 1) == 0)
> + {
> + pa_lock_stream(MyParallelShared->xid, AccessShareLock); 
> + pa_unlock_stream(MyParallelShared->xid, AccessShareLock); }
> 
> ditto comment #23

I feel the place atop the definition of pa_lock_xxx function is a better place to
put the comments, so added there. User can check it when reading the lock
functions.

> ~~~
> 
> 25. apply_worker_clean_exit
> 
> +void
> +apply_worker_clean_exit(void)
> +{
> + /* Notify the leader apply worker that we have exited cleanly. */  
> +if (am_parallel_apply_worker())  pq_putmessage('X', NULL, 0);
> +
> + proc_exit(0);
> +}
> 
> Somehow it doesn't seem right that the PA worker sending 'X' is here 
> in worker.c, while the LA worker receipt of this 'X' is in the other 
> applyparallelworker.c module. Maybe that other function 
> HandleParallelApplyMessage should also be here in worker.c?

I thought the function apply_worker_clean_exit is widely used in worker.c and
is a common function for both leader/parallel apply workers, so I put it in
worker.c. But HandleParallelApplyMessage is a function only for parallel
worker, so it would be better to put it in applyparallelworker.c.

> ======
> 
> src/backend/utils/misc/guc_tables.c
> 
> 26.
> 
> @@ -2957,6 +2957,18 @@ struct config_int ConfigureNamesInt[] =
>   NULL,
>   },
>   &max_sync_workers_per_subscription,
> + 2, 0, MAX_PARALLEL_WORKER_LIMIT,
> + NULL, NULL, NULL
> + },
> +
> + {
> + {"max_parallel_apply_workers_per_subscription",
> + PGC_SIGHUP,
> + REPLICATION_SUBSCRIBERS,
> + gettext_noop("Maximum number of parallel apply workers per 
> + subscription."), NULL, }, 
> + &max_parallel_apply_workers_per_subscription,
>   2, 0, MAX_BACKENDS,
>   NULL, NULL, NULL
> 
> Is this correct? Did you mean to change 
> max_sync_workers_per_subscription, My 1st impression is that there has 
> been some mixup with the MAX_PARALLEL_WORKER_LIMIT and MAX_BACKENDS or 
> that this change was accidentally made to the wrong GUC.

Fixed.

> ======
> 
> src/include/replication/worker_internal.h
> 
> 27. ParallelApplyWorkerShared
> 
> + /*
> + * Indicates whether there are pending streaming blocks in the queue. 
> + The
> + * parallel apply worker will check it before starting to wait.
> + */
> + pg_atomic_uint32 pending_stream_count;
> 
> A better name might be 'n_pending_stream_blocks'.

I am not sure if the name looks better, so didn’t change this.

> ~
> 
> 28. function names
> 
>  extern void logicalrep_worker_stop(Oid subid, Oid relid);
> +extern void logicalrep_parallel_apply_worker_stop(int slot_no, uint16
> generation);
>  extern void logicalrep_worker_wakeup(Oid subid, Oid relid);  extern 
> void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
> 
>  extern int logicalrep_sync_worker_count(Oid subid);
> +extern int logicalrep_parallel_apply_worker_count(Oid subid);
> 
> Would it be better to call those new functions using similar shorter 
> names as done elsewhere?
> 
> logicalrep_parallel_apply_worker_stop -> logicalrep_pa_worker_stop 
> logicalrep_parallel_apply_worker_count -> logicalrep_pa_worker_count

Changed.

Attach new version patch which also fixed an invalid shared memory access bug
in 0002 patch reported by Kuroda-San offlist. 

Best regards,
Hou zj