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

On Mon, November 28, 2022 20:26 PM Amit Kapila <amit(dot)kapila16(at)gmail(dot)com> wrote:
> On Sun, Nov 27, 2022 at 9:43 AM houzj(dot)fnst(at)fujitsu(dot)com
> <houzj(dot)fnst(at)fujitsu(dot)com> wrote:
> >
> > Attach the new version patch which addressed all comments so far.
> >
>
> Few comments on v52-0001*
> ========================
> 1.
> pa_free_worker()
> {
> ...
> + /* Free the worker information if the worker exited cleanly. */ if
> + (!winfo->error_mq_handle) { pa_free_worker_info(winfo);
> +
> + if (winfo->in_use &&
> + !hash_search(ParallelApplyWorkersHash, &xid, HASH_REMOVE, NULL))
> + elog(ERROR, "hash table corrupted");
>
> pa_free_worker_info() pfrees the winfo, so how is it legal to
> winfo->in_use in the above check?
>
> Also, why is this check (!winfo->error_mq_handle) required in the
> first place in the patch? The worker exits cleanly only when the
> leader apply worker sends a SIGINT signal and in that case, we already
> detach from the error queue and clean up other worker information.

It was intended for the case when a user send a signal, but it seems not standard way to do that.
So, I removed this check (!winfo->error_mq_handle).

> 2.
> +HandleParallelApplyMessages(void)
> +{
> ...
> ...
> + foreach(lc, ParallelApplyWorkersList) { shm_mq_result res; Size
> + nbytes;
> + void *data;
> + ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *)
> + lfirst(lc);
> +
> + if (!winfo->error_mq_handle)
> + continue;
>
> Similar to the previous comment, it is not clear whether we need this
> check. If required, can we add a comment to indicate the case where it
> happens to be true?
> Note, there is a similar check for winfo->error_mq_handle in
> pa_wait_for_xact_state(). Please add some comments if that is
> required.

Removed this check in these two functions.

> 3. Why is there apply_worker_clean_exit() at the end of
> ParallelApplyWorkerMain()? Normally either the leader worker stops
> parallel apply, or parallel apply gets stopped because of a parameter
> change, or exits because of error, and in none of those cases it can
> hit this code path unless I am missing something.
>
> Additionally, I think in LogicalParallelApplyLoop, we will never
> receive zero-length messages so that is also wrong and should be
> converted to elog(ERROR,..).

Agreed and changed.

> 4. I think in logicalrep_worker_detach(), we should detach from the
> shm error queue so that the parallel apply worker won't try to send a
> termination message back to the leader worker.

Agreed and changed.

> 5.
> pa_send_data()
> {
> ...
> + if (startTime == 0)
> + startTime = GetCurrentTimestamp();
> ...
>
> What is the use of getting the current timestamp before waitlatch
> logic, if it is not used before that? It seems that is for the time
> logic to look correct. We can probably reduce the 10s interval to 9s
> for that.

Changed.

> In this function, we need to add some comments to indicate why the
> current logic is used, and also probably we can refer to the comments
> atop this file.

Added some comments.

> 6. I think it will be better if we keep stream_apply_worker local to
> applyparallelworker.c by exposing functions to cache/resetting the
> required info.

Agree. Added a new function to set the stream_apply_worker.

> 7. Apart from the above, I have made a few changes in the comments and
> some miscellaneous cosmetic changes in the attached. Kindly include
> these in the next version unless you see a problem with any change.

Thanks, I have checked and merge them.

Attach the new version patch which addressed all comments.

Best regards,
Hou zj