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

On Thu, Nov 10, 2022 at 8:41 PM houzj(dot)fnst(at)fujitsu(dot)com
<houzj(dot)fnst(at)fujitsu(dot)com> wrote:
>
> On Monday, November 7, 2022 6:18 PM Masahiko Sawada <sawada(dot)mshk(at)gmail(dot)com>
> >
> > Here are comments on v42-0001:
>
> Thanks for the comments.
>
> > We have the following three similar name functions regarding to
> > starting a new parallel apply worker:
> >
> > parallel_apply_start_worker()
> > parallel_apply_setup_worker()
> > parallel_apply_setup_dsm()
> >
> > It seems to me that we can somewhat merge them since
> > parallel_apply_setup_worker() and parallel_apply_setup_dsm() have only
> > one caller.
>
> Since these functions are doing different tasks(external function, Launch, DSM), so I
> personally feel it's OK to split them. But if others also feel it's unnecessary I will
> merge them.
>

I think it is fine either way but if you want to keep the
functionality of parallel_apply_setup_worker() separate then let's
name it to something like parallel_apply_init_and_launch_worker which
will make the function name bit long but it will be clear. I am
thinking that instead of using parallel_apply infront of each
function, shall we use PA? Then we can name this function as
PAInitializeAndLaunchWorker().

I feel you can even move the functionality to get the worker from pool
in parallel_apply_start_worker() to a separate function.

Another related comment:
+ /* Try to get a free parallel apply worker. */
+ foreach(lc, ParallelApplyWorkersList)
+ {
+ ParallelApplyWorkerInfo *tmp_winfo;
+
+ tmp_winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
+
+ /* Check if the transaction in the worker has finished. */
+ if (parallel_apply_free_worker(tmp_winfo, tmp_winfo->shared->xid, false))
+ {
+ /*
+ * Clean up the woker information if the parallel apply woker has
+ * been stopped.
+ */
+ ParallelApplyWorkersList =
foreach_delete_current(ParallelApplyWorkersList, lc);
+ parallel_apply_free_worker_info(tmp_winfo);
+ continue;
+ }

I find it bit odd that even parallel_apply_free_worker() has the
functionality to free the worker info, still, we are doing it outside.
Is there a specific reason for the same? I think we can add a comment
atop parallel_apply_free_worker() that on success, it will free the
passed winfo. In addition to that, we can write some comments before
trying to free worker suggesting that it would be possible for
rollback cases because after rollback we don't wait for workers to
finish so can't perform the cleanup.

> > ---
> > in parallel_apply_send_data():
> >
> > + result = shm_mq_send(winfo->mq_handle, nbytes, data,
> > true, true);
> > +
> > + if (result == SHM_MQ_SUCCESS)
> > + break;
> > + else if (result == SHM_MQ_DETACHED)
> > + ereport(ERROR,
> > +
> > (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> > + errmsg("could not send data
> > to shared-memory queue")))
> > +
> > + Assert(result == SHM_MQ_WOULD_BLOCK);
> > +
> > + if (++retry >= CHANGES_THRESHOLD)
> > + {
> > + MemoryContext oldcontext;
> > + StringInfoData msg;
> > + TimestampTz now = GetCurrentTimestamp();
> > +
> > + if (startTime == 0)
> > + startTime = now;
> > +
> > + if (!TimestampDifferenceExceeds(startTime,
> > now, SHM_SEND_TIMEOUT_MS))
> > + continue;
> >
> > IIUC since the parallel worker retries to send data without waits the
> > 'retry' will get larger than CHANGES_THRESHOLD in a very short time.
> > But the worker waits at least for SHM_SEND_TIMEOUT_MS to spool data
> > regardless of 'retry' count. Don't we need to nap somewhat and why do
> > we need CHANGES_THRESHOLD?
>
> Oh, I intended to only check for timeout after continuously retrying XX times to
> reduce the cost of getting the system time and calculating the time difference.
> I added some comments in the code.
>

Sure, but the patch assumes that immediate retry will help which I am
not sure is correct. IIUC, the patch has overall wait time 10s, if so,
I guess you can retry after 1s, that will amiliorate the cost of
getting the system time.

--
With Regards,
Amit Kapila.