public inbox for [email protected]
help / color / mirror / Atom feedFrom: Zhijie Hou (Fujitsu) <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Alexander Lakhin <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
Date: Fri, 5 May 2023 03:44:47 +0000
Message-ID: <OS0PR01MB57161176B8DB0A662D6F053D94729@OS0PR01MB5716.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1+GGTX+qhBUEmjScVLFvAi2UsTBcaM_HUY6g5Pv5CgCew@mail.gmail.com>
References: <CAA4eK1LGmZsevrqJra0V4O8oBU_eKyzm2VMpSAYQaDgC6n4fkA@mail.gmail.com>
<CAFiTN-uhnJkrUsQByBHMK268T-GEx_D8DQ6b2T+aW6RiU75pbQ@mail.gmail.com>
<OS0PR01MB571621ED532C2D7C3E01625894FB9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAFiTN-t5+_zTf86EzgCwObs1ED-P_5Ab1KMw340d3oPY_+Cwpw@mail.gmail.com>
<OS3PR01MB5718AE486227CE3ACB844C5E94F89@OS3PR01MB5718.jpnprd01.prod.outlook.com>
<CAFiTN-sgQ-K9BnMWKLR7Hm36QNjS7ZOBF6hsXJmQz5ChC0Desw@mail.gmail.com>
<CAA4eK1+P9WOFBupaTZCzYtranSH=8TSUYV=kk5xQJ-VNKnvgmQ@mail.gmail.com>
<OS0PR01MB571685A645E509267AC5F30894F99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57168A5A1C31BC316CB63D3C94F99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KK7Xty7SaCReY_b5moPURowMGOsfvhOb6xQ1EAaJAh6A@mail.gmail.com>
<CAD21AoDDbM8_HJt-nMCvcjTK8K9hPzXWqJj7pyaUvR4mm_NrSg@mail.gmail.com>
<CAA4eK1K4zjeCmZUX_L09-VhjMxfqcCa+OMmpHre7YWTY2Tgxsw@mail.gmail.com>
<CAD21AoDD_P+6SCfQ2Nc79k=WwgDMyzMXzMVfhSEiGPyvYJ4H_Q@mail.gmail.com>
<[email protected]>
<OS0PR01MB57164DF9FC5366024A1952D594659@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1+Qg79oR0LknxC05DvWfh_eiyh7eO3sDOiuVsu_dCUDHA@mail.gmail.com>
<CAD21AoDo+yUwNq6nTrvE2h9bB2vZfcag=jxWc7QxuWCmkDAqcA@mail.gmail.com>
<OS0PR01MB5716E3C106DF49B3056EF72C946F9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1+JtTXVUxKAvicuvHMj3xXj=5v=Wxfka3Jt0PYe13kNQw@mail.gmail.com>
<CAA4eK1+GGTX+qhBUEmjScVLFvAi2UsTBcaM_HUY6g5Pv5CgCew@mail.gmail.com>
On Wednesday, May 3, 2023 3:17 PM Amit Kapila <[email protected]> wrote:
>
> On Tue, May 2, 2023 at 9:46 AM Amit Kapila <[email protected]>
> wrote:
> >
> > On Tue, May 2, 2023 at 9:06 AM Zhijie Hou (Fujitsu)
> > <[email protected]> wrote:
> > >
> > > On Friday, April 28, 2023 2:18 PM Masahiko Sawada
> <[email protected]> wrote:
> > > >
> > > > >
> > > > > Alexander, does the proposed patch fix the problem you are facing?
> > > > > Sawada-San, and others, do you see any better way to fix it than
> > > > > what has been proposed?
> > > >
> > > > I'm concerned that the idea of relying on IsNormalProcessingMode()
> > > > might not be robust since if we change the meaning of
> > > > IsNormalProcessingMode() some day it would silently break again.
> > > > So I prefer using something like InitializingApplyWorker, or
> > > > another idea would be to do cleanup work (e.g., fileset deletion
> > > > and lock release) in a separate callback that is registered after
> > > > connecting to the database.
> > >
> > > Thanks for the review. I agree that it’s better to use a new variable here.
> > > Attach the patch for the same.
> > >
> >
> > + *
> > + * However, if the worker is being initialized, there is no need to
> > + release
> > + * locks.
> > */
> > - LockReleaseAll(DEFAULT_LOCKMETHOD, true);
> > + if (!InitializingApplyWorker)
> > + LockReleaseAll(DEFAULT_LOCKMETHOD, true);
> >
> > Can we slightly reword this comment as: "The locks will be acquired
> > once the worker is initialized."?
> >
>
> After making this modification, I pushed your patch. Thanks!
Thanks for pushing.
Attach another patch to fix the problem that pa_shutdown will access invalid
MyLogicalRepWorker. I personally want to avoid introducing new static variable,
so I only reorder the callback registration in this version.
When testing this, I notice a rare case that the leader is possible to receive
the worker termination message after the leader stops the parallel worker. This
is unnecessary and have a risk that the leader would try to access the detached
memory queue. This is more likely to happen and sometimes cause the failure in
regression tests after the registration reorder patch because the dsm is
detached earlier after applying the patch.
So, put the patch that detach the error queue before stopping worker as 0001
and the registration reorder patch as 0002.
Best Regards,
Hou zj
Attachments:
[application/octet-stream] 0002-adjust-the-order-of-callback-registration-to-avoid-a.patch (2.8K, ../OS0PR01MB57161176B8DB0A662D6F053D94729@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-0002-adjust-the-order-of-callback-registration-to-avoid-a.patch)
download | inline diff:
From fc0824a5580b00ce685d91964a8193527f62d908 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 28 Apr 2023 15:50:04 +0800
Subject: [PATCH 2/2] adjust the order of callback registration to avoid
accessing invalid memory
The callback function pa_shutdown(), which accesses MyLogicalRepWorker
internally, was registered before initialization of MyLogicalRepWorker. As a
result, if an ERROR occurs before initialization completes, pa_shutdown() will
attempt to access invalid memory. Additionally, pa_shutdown() was registered
before an exit callback function that resets MyLogicalRepWorker. If an ERROR
occurs later, pa_shutdown() will be invoked after the reset, potentially
leading to incorrect information.
To fix this, reorder the registration process, so that pa_shutdown() is
registered after the initialization and other exit callback function.
While on it, add few comments atop pa_shutdown to enhance code readability.
---
.../replication/logical/applyparallelworker.c | 12 ++++++++++--
1 file changed, 10 insertions(+), 2 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index c79765ce20..1bfb7ee06d 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -836,6 +836,9 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
* Make sure the leader apply worker tries to read from our error queue one more
* time. This guards against the case where we exit uncleanly without sending
* an ErrorResponse, for example because some code calls proc_exit directly.
+ *
+ * Also explicitly detach from dsm segment to fire on_dsm_detach callbacks. See
+ * ParallelWorkerShutdown for details.
*/
static void
pa_shutdown(int code, Datum arg)
@@ -892,8 +895,6 @@ ParallelApplyWorkerMain(Datum main_arg)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("bad magic number in dynamic shared memory segment")));
- before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
-
/* Look up the shared information. */
shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
MyParallelShared = shared;
@@ -912,6 +913,13 @@ ParallelApplyWorkerMain(Datum main_arg)
*/
logicalrep_worker_attach(worker_slot);
+ /*
+ * Register the callback after attaching to the worker slot to ensure it is
+ * invoked after MyLogicalRepWorker is initialized but before detaching
+ * from the slot. This ensures that MyLogicalRepWorker remains valid.
+ */
+ before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
+
SpinLockAcquire(&MyParallelShared->mutex);
MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
MyParallelShared->logicalrep_worker_slot_no = worker_slot;
--
2.30.0.windows.2
[application/octet-stream] 0001-Detach-the-error-queue-before-stopping-parallel-appl.patch (3.7K, ../OS0PR01MB57161176B8DB0A662D6F053D94729@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-0001-Detach-the-error-queue-before-stopping-parallel-appl.patch)
download | inline diff:
From c188075f07e5dcfa53a77c92b25913af31622e54 Mon Sep 17 00:00:00 2001
From: sherlockcpp <[email protected]>
Date: Sun, 30 Apr 2023 20:58:42 +0800
Subject: [PATCH 1/2] Detach the error queue before stopping parallel apply
worker
Detach from the error_mq_handle for the parallel apply worker before
terminating it. This prevents the leader apply worker from receiving the
worker termination message and sending it to logs when the same is
already done by the parallel worker.
---
.../replication/logical/applyparallelworker.c | 11 +---------
src/backend/replication/logical/launcher.c | 21 ++++++++++++++++---
src/include/replication/worker_internal.h | 2 +-
3 files changed, 20 insertions(+), 14 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index ee7a18137f..c79765ce20 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -577,16 +577,7 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
list_length(ParallelApplyWorkerPool) >
(max_parallel_apply_workers_per_subscription / 2))
{
- int slot_no;
- uint16 generation;
-
- SpinLockAcquire(&winfo->shared->mutex);
- generation = winfo->shared->logicalrep_worker_generation;
- slot_no = winfo->shared->logicalrep_worker_slot_no;
- SpinLockRelease(&winfo->shared->mutex);
-
- logicalrep_pa_worker_stop(slot_no, generation);
-
+ logicalrep_pa_worker_stop(winfo);
pa_free_worker_info(winfo);
return;
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index ceea126231..ae107a2184 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -609,19 +609,34 @@ logicalrep_worker_stop(Oid subid, Oid relid)
}
/*
- * Stop the logical replication parallel apply worker corresponding to the
- * input slot number.
+ * Stop the given logical replication parallel apply worker.
*
* Node that the function sends SIGINT instead of SIGTERM to the parallel apply
* worker so that the worker exits cleanly.
*/
void
-logicalrep_pa_worker_stop(int slot_no, uint16 generation)
+logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
{
+ int slot_no;
+ uint16 generation;
LogicalRepWorker *worker;
+ SpinLockAcquire(&winfo->shared->mutex);
+ generation = winfo->shared->logicalrep_worker_generation;
+ slot_no = winfo->shared->logicalrep_worker_slot_no;
+ SpinLockRelease(&winfo->shared->mutex);
+
Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
+ /*
+ * Detach from the error_mq_handle for the parallel apply worker before
+ * terminating it. This prevents the leader apply worker from receiving the
+ * worker termination message and sending it to logs when the same is
+ * already done by the parallel worker.
+ */
+ shm_mq_detach(winfo->error_mq_handle);
+ winfo->error_mq_handle = NULL;
+
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
worker = &LogicalRepCtx->workers[slot_no];
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index b57eed052f..343e781896 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -235,7 +235,7 @@ extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
Oid userid, Oid relid,
dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
-extern void logicalrep_pa_worker_stop(int slot_no, uint16 generation);
+extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
--
2.30.0.windows.2
view thread (105+ 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]
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <OS0PR01MB57161176B8DB0A662D6F053D94729@OS0PR01MB5716.jpnprd01.prod.outlook.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