public inbox for [email protected]
help / color / mirror / Atom feedFrom: Amit Kapila <[email protected]>
To: [email protected] <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
Date: Mon, 28 Nov 2022 17:55:59 +0530
Message-ID: <CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com> (raw)
In-Reply-To: <OS0PR01MB571663F65904D00895AD159994109@OS0PR01MB5716.jpnprd01.prod.outlook.com>
References: <CAA4eK1JQTDXTfvJ5d+L0ggG4+doyd0Xji=e0OJsb=qcn_jWALA@mail.gmail.com>
<OS0PR01MB5716D6D2765E54DC739F288E940A9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716997A7115715F9E4EE520940D9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1JEFVZcymteMYXhYghiX7Lb=MKuEsfxgD1VfaRrLCHyzg@mail.gmail.com>
<OS0PR01MB571663F65904D00895AD159994109@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Sun, Nov 27, 2022 at 9:43 AM [email protected]
<[email protected]> 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.
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.
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,..).
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.
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.
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.
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.
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.
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] changes_amit_v52_1.patch (2.4K, ../CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com/2-changes_amit_v52_1.patch)
download | inline diff:
diff --git a/src/backend/postmaster/interrupt.c b/src/backend/postmaster/interrupt.c
index e2e5a1373b..e64029fac4 100644
--- a/src/backend/postmaster/interrupt.c
+++ b/src/backend/postmaster/interrupt.c
@@ -99,7 +99,7 @@ SignalHandlerForCrashExit(SIGNAL_ARGS)
*
* Typically, this handler would be used for SIGTERM, but some processes use
* other signals. In particular, the checkpointer exits on SIGUSR2, and the WAL
- * writer and the logical replication parallel apply worker exit on either
+ * writer and the logical replication parallel apply worker exits on either
* SIGINT or SIGTERM.
*
* ShutdownRequestPending should be checked at a convenient place within the
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 85bb909104..384ee32d44 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -155,8 +155,9 @@ static HTAB *ParallelApplyWorkersHash = NULL;
/*
* A list to maintain the active parallel apply workers. The information for
* the new worker is added to the list after successfully launching it. The
- * list entry is removed at the end of the transaction if there are already
- * enough workers in the worker pool. For more information about the worker
+ * list entry is removed if there are already enough workers in the worker
+ * pool either at the end of the transaction or while trying to find a free
+ * worker for applying the transaction. For more information about the worker
* pool, see comments atop worker.c.
*/
static List *ParallelApplyWorkersList = NIL;
@@ -297,14 +298,14 @@ pa_allocate_worker(TransactionId xid)
HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
}
- /* First, try to get a parallel apply worker from the pool. */
+ /*
+ * First, try to get a parallel apply worker from the pool, if available.
+ * Otherwise, try to start a new parallel apply worker.
+ */
winfo = pa_get_available_worker();
-
if (!winfo)
{
- /* Try to start a new parallel apply worker. */
winfo = pa_init_and_launch_worker();
-
if (!winfo)
return;
}
@@ -454,7 +455,7 @@ pa_free_worker_info(ParallelApplyWorkerInfo *winfo)
}
/*
- * Interrupt handler for main loops of parallel apply worker.
+ * Interrupt handler for main loop of parallel apply worker.
*/
static void
ProcessParallelApplyInterrupts(void)
[application/octet-stream] changes_amit_v52_1.patch (2.4K, ../CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com/3-changes_amit_v52_1.patch)
download | inline diff:
diff --git a/src/backend/postmaster/interrupt.c b/src/backend/postmaster/interrupt.c
index e2e5a1373b..e64029fac4 100644
--- a/src/backend/postmaster/interrupt.c
+++ b/src/backend/postmaster/interrupt.c
@@ -99,7 +99,7 @@ SignalHandlerForCrashExit(SIGNAL_ARGS)
*
* Typically, this handler would be used for SIGTERM, but some processes use
* other signals. In particular, the checkpointer exits on SIGUSR2, and the WAL
- * writer and the logical replication parallel apply worker exit on either
+ * writer and the logical replication parallel apply worker exits on either
* SIGINT or SIGTERM.
*
* ShutdownRequestPending should be checked at a convenient place within the
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 85bb909104..384ee32d44 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -155,8 +155,9 @@ static HTAB *ParallelApplyWorkersHash = NULL;
/*
* A list to maintain the active parallel apply workers. The information for
* the new worker is added to the list after successfully launching it. The
- * list entry is removed at the end of the transaction if there are already
- * enough workers in the worker pool. For more information about the worker
+ * list entry is removed if there are already enough workers in the worker
+ * pool either at the end of the transaction or while trying to find a free
+ * worker for applying the transaction. For more information about the worker
* pool, see comments atop worker.c.
*/
static List *ParallelApplyWorkersList = NIL;
@@ -297,14 +298,14 @@ pa_allocate_worker(TransactionId xid)
HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
}
- /* First, try to get a parallel apply worker from the pool. */
+ /*
+ * First, try to get a parallel apply worker from the pool, if available.
+ * Otherwise, try to start a new parallel apply worker.
+ */
winfo = pa_get_available_worker();
-
if (!winfo)
{
- /* Try to start a new parallel apply worker. */
winfo = pa_init_and_launch_worker();
-
if (!winfo)
return;
}
@@ -454,7 +455,7 @@ pa_free_worker_info(ParallelApplyWorkerInfo *winfo)
}
/*
- * Interrupt handler for main loops of parallel apply worker.
+ * Interrupt handler for main loop of parallel apply worker.
*/
static void
ProcessParallelApplyInterrupts(void)
view thread (625+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox