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

On Tue, Nov 22, 2022 at 7:30 AM houzj(dot)fnst(at)fujitsu(dot)com
<houzj(dot)fnst(at)fujitsu(dot)com> wrote:
>

Few minor comments and questions:
============================
1.
+static void
+LogicalParallelApplyLoop(shm_mq_handle *mqh)
{
+ for (;;)
+ {
+ void *data;
+ Size len;
+
+ ProcessParallelApplyInterrupts();
...
...
+ if (rc & WL_LATCH_SET)
+ {
+ ResetLatch(MyLatch);
+ ProcessParallelApplyInterrupts();
+ }
...
}

Why ProcessParallelApplyInterrupts() is called twice in
LogicalParallelApplyLoop()?

2.
+ * This scenario is similar to the first case but TX-1 and TX-2 are executed by
+ * two parallel apply workers (PA-1 and PA-2 respectively). In this scenario,
+ * PA-2 is waiting for PA-1 to complete its transaction while PA-1 is waiting
+ * for subsequent input from LA. Also, LA is waiting for PA-2 to complete its
+ * transaction in order to preserve the commit order. There is a deadlock among
+ * three processes.
+ *
...
...
+ *
+ * LA (waiting to acquire the local transaction lock) -> PA-1 (waiting to
+ * acquire the lock on the unique index) -> PA-2 (waiting to acquire the lock
+ * on the remote transaction) -> LA
+ *

Isn't the order of PA-1 and PA-2 different in the second paragraph as
compared to the first one.

3.
+ * Deadlock-detection
+ * ------------------

It may be better to keep the title of this section as Locking Considerations.

4. In the section mentioned in Point 3, it would be better to
separately explain why we need session-level locks instead of
transaction level.

5. Add the below comments in the code:
diff --git a/src/backend/replication/logical/applyparallelworker.c
b/src/backend/replication/logical/applyparallelworker.c
index 9385afb6d2..56f00defcf 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -431,6 +431,9 @@ pa_free_worker_info(ParallelApplyWorkerInfo *winfo)
if (winfo->dsm_seg != NULL)
dsm_detach(winfo->dsm_seg);

+ /*
+ * Ensure this worker information won't be reused during
worker allocation.
+ */
ParallelApplyWorkersList = list_delete_ptr(ParallelApplyWorkersList,

winfo);

@@ -762,6 +765,10 @@
HandleParallelApplyMessage(ParallelApplyWorkerInfo *winfo, StringInfo
msg)
*/
error_context_stack = apply_error_context_stack;

+ /*
+ * The actual error must be already
reported by parallel apply
+ * worker.
+ */
ereport(ERROR,

(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("parallel
apply worker exited abnormally"),

--
With Regards,
Amit Kapila.