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

From: "[email-masked]" <[email-masked]>
Date: Fri, 6 Jan 2023 04:07:49 +0000

On Thursday, January 5, 2023 7:54 PM Dilip Kumar <[email-masked]> wrote:
> 
> On Thu, Jan 5, 2023 at 5:03 PM [email-masked]
> <[email-masked]> wrote:
> >
> > On Thursday, January 5, 2023 4:22 PM Dilip Kumar <[email-masked]>
> wrote:
> > >
> 
> > Thanks for reporting the problem.
> >
> > After analyzing the behavior, I think it's a bug on publisher side
> > which is not directly related to parallel apply.
> >
> > I think the root reason is that we didn't try to send a stream
> > end(stream
> > abort) message to subscriber for the crashed transaction which was
> > streamed before.
> > The behavior is that, after restarting, the publisher will start to
> > decode the transaction that aborted due to crash, and when try to
> > stream the first change of that transaction, it will send a stream
> > start message but then it realizes that the transaction was aborted,
> > so it will enter the PG_CATCH block of
> > ReorderBufferProcessTXN() and call ReorderBufferResetTXN() which send
> > the stream stop message. And in this case, there would be a parallel
> > apply worker started on subscriber waiting for stream end message which
> will never come.
> 
> I suspected it but didn't analyze this.
> 
> > I think the same behavior happens for the non-parallel mode which will
> > cause a stream file left on subscriber and will not be cleaned until
> > the apply worker is restarted.
> > To fix it, I think we need to send a stream abort message when we are
> > cleaning up crashed transaction on publisher(e.g., in
> > ReorderBufferAbortOld()). And here is a tiny patch which change the
> > same. I have confirmed that the bug is fixed and all regression tests pass.
> >
> > What do you think ?
> > I will start a new thread and try to write a testcase if possible
> > after reaching a consensus.
> 
> I think your analysis looks correct and we can raise this in a new thread.

Thanks, I have started another thread[1]

Attach the parallel apply patch set here again. I didn't change the patch set,
attach it here just to let the CFbot keep testing it.

[1] https://www.postgresql.org/message-id/OS0PR01MB5716A773F46768A1B75BE24394FB9%40OS0PR01MB5716.jpnprd01.prod.outlook.com

Best regards,
Hou zj

---

From: Dilip Kumar <[email-masked]>
Date: Fri, 6 Jan 2023 11:24:25 +0530

> attach it here just to let the CFbot keep testing it.

I have completed the review and some basic testing and it mostly looks
fine to me.  Here is my last set of comments/suggestions.

1.
    /*
     * Don't start a new parallel worker if user has set skiplsn as it's
     * possible that they want to skip the streaming transaction. For
     * streaming transactions, we need to serialize the transaction to a file
     * so that we can get the last LSN of the transaction to judge whether to
     * skip before starting to apply the change.
     */
    if (!XLogRecPtrIsInvalid(MySubscription->skiplsn))
        return false;

I think this is fine to block parallelism in this case, but it is also
possible to make it less restrictive, basically, only if the first lsn
of the transaction is <= skiplsn, then only it is possible that the
final_lsn might match with skiplsn otherwise that is not possible. And
if we want then we can allow parallelism in that case.

I understand that currently we do not have first_lsn of the
transaction in stream start message but I think that should be easy to
do?  Although I am not sure if it is worth it, it's good to make a
note at least.

2.

+     * XXX Additionally, we also stop the worker if the leader apply worker
+     * serialize part of the transaction data due to a send timeout. This is
+     * because the message could be partially written to the queue and there
+     * is no way to clean the queue other than resending the message until it
+     * succeeds. Instead of trying to send the data which anyway would have
+     * been serialized and then letting the parallel apply worker deal with
+     * the spurious message, we stop the worker.
+     */
+    if (winfo->serialize_changes ||
+        list_length(ParallelApplyWorkerPool) >
+        (max_parallel_apply_workers_per_subscription / 2))

IMHO this reason (XXX Additionally, we also stop the worker if the
leader apply worker serialize part of the transaction data due to a
send timeout) for stopping the worker looks a bit hackish to me.  It
may be a rare case so I am not talking about the performance but the
reasoning behind stopping is not good. Ideally we should be able to
clean up the message queue and reuse the worker.

3.
+        else if (shmq_res == SHM_MQ_WOULD_BLOCK)
+        {
+            /* Replay the changes from the file, if any. */
+            if (pa_has_spooled_message_pending())
+            {
+                pa_spooled_messages();
+            }

I think we do not need this pa_has_spooled_message_pending() function.
Because this function is just calling pa_get_fileset_state() which is
acquiring mutex and getting filestate then if the filestate is not
FS_EMPTY then we call pa_spooled_messages() that will again call
pa_get_fileset_state() which will again acquire mutex.  I think when
the state is FS_SERIALIZE_IN_PROGRESS it will frequently call
pa_get_fileset_state() consecutively 2 times, and I think we can
easily achieve the same behavior with just one call.

4.

+     * leader, or when there there is an error. None of these cases will allow
+     * the code to reach here.

/when there there is an error/when there is an error

-- 
Regards,
Dilip Kumar
EnterpriseDB: http://www.enterprisedb.com

---

From: Amit Kapila <[email-masked]>
Date: Fri, 6 Jan 2023 12:05:42 +0530

> do?  Although I am not sure if it is worth it, it's good to make a

Yeah, I also don't think sending extra eight bytes with stream_start
message is worth it. But it is fine to mention the same in the
comments.

> clean up the message queue and reuse the worker.

TBH, I don't know what is the better way to deal with this with the
current infrastructure. I thought we can do this as a separate
enhancement in the future.

> easily achieve the same behavior with just one call.

This is just to keep the code easy to follow. As this would be a rare
case, so thought of giving preference to code clarity.

-- 
With Regards,
Amit Kapila.

---

From: Dilip Kumar <[email-masked]>
Date: Fri, 6 Jan 2023 12:59:22 +0530

On Fri, Jan 6, 2023 at 12:05 PM Amit Kapila <[email-masked]> wrote:

> message is worth it. But it is fine to mention the same in the

Right.

> enhancement in the future.

Okay.

> case, so thought of giving preference to code clarity.

I think the code will be simpler with just one function no? I mean
instead of calling pa_has_spooled_message_pending() in if condition
what if we directly call pa_spooled_messages();, this is anyway
fetching the file_state and if the filestate is EMPTY then it can
return false, and if it returns false we can execute the code which is
there in else condition.  We might need to change the name of the
function though.

---

From: Dilip Kumar <[email-masked]>
Date: Fri, 6 Jan 2023 13:01:02 +0530

> there in else condition.  We might need to change the name of the
But anyway it is not a performance-critical path so if you think the
current way looks cleaner then I am fine with that too.