public inbox for [email protected]
help / color / mirror / Atom feedFrom: Amit Kapila <[email protected]>
To: Zhijie Hou (Fujitsu) <[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: Mon, 8 May 2023 16:39:07 +0530
Message-ID: <CAA4eK1LkPBTgcBxKrKahFu47oSK6rDq-AAs0oy4QXH6A8UzEig@mail.gmail.com> (raw)
In-Reply-To: <OS0PR01MB57161176B8DB0A662D6F053D94729@OS0PR01MB5716.jpnprd01.prod.outlook.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>
<OS0PR01MB57161176B8DB0A662D6F053D94729@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Fri, May 5, 2023 at 9:14 AM Zhijie Hou (Fujitsu)
<[email protected]> wrote:
>
> On Wednesday, May 3, 2023 3:17 PM Amit Kapila <[email protected]> wrote:
> >
>
> 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.
>
I think it is only possible for the leader apply can worker to try to
receive the error message from an error queue after your 0002 patch.
Because another place already detached from the queue before stopping
the parallel apply workers. So, I combined both the patches and
changed a few comments and a commit message. Let me know what you
think of the attached.
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] v2-0001-Fix-invalid-memory-access-during-the-shutdown-of-.patch (5.9K, ../CAA4eK1LkPBTgcBxKrKahFu47oSK6rDq-AAs0oy4QXH6A8UzEig@mail.gmail.com/2-v2-0001-Fix-invalid-memory-access-during-the-shutdown-of-.patch)
download | inline diff:
From 12bcb55f98cbd30de80d95378a7b0df4152ea31a Mon Sep 17 00:00:00 2001
From: Amit Kapila <[email protected]>
Date: Mon, 8 May 2023 16:21:21 +0530
Subject: [PATCH v2] Fix invalid memory access during the shutdown of the
parallel apply worker.
The callback function pa_shutdown() accesses MyLogicalRepWorker which may
not be initialized if there is an error during the initialization of the
parallel apply worker. The other problem is that by the time it is invoked
even after the initialization of the worker, the MyLogicalRepWorker will
be reset by another callback logicalrep_worker_onexit. So, it won't have
the required information.
To fix this, register the shutdown callback after we are attached to the
worker slot.
After this fix, we observed another issue which is that sometimes the
leader apply worker tries to receive the message from the error queue that
might already be detached by the parallel apply worker leading to an
error. To prevent such an error, we ensure that the leader apply worker
detaches from the parallel apply worker's error queue before stopping it.
Reported-by: Sawada Masahiko
Author: Hou Zhijie
Reviewed-by: Amit Kapila
Discussion: https://postgr.es/m/CAD21AoDo+yUwNq6nTrvE2h9bB2vZfcag=jxWc7QxuWCmkDAqcA@mail.gmail.com
---
.../replication/logical/applyparallelworker.c | 23 +++++++++----------
src/backend/replication/logical/launcher.c | 21 ++++++++++++++---
src/include/replication/worker_internal.h | 2 +-
3 files changed, 30 insertions(+), 16 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index ee7a18137f..096d9f4bee 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;
@@ -845,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 invoke on_dsm_detach callbacks,
+ * if any. See ParallelWorkerShutdown for details.
*/
static void
pa_shutdown(int code, Datum arg)
@@ -901,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;
@@ -921,6 +913,13 @@ ParallelApplyWorkerMain(Datum main_arg)
*/
logicalrep_worker_attach(worker_slot);
+ /*
+ * Register the shutdown callback after we are attached to the worker slot.
+ * This is to ensure that MyLogicalRepWorker remains valid when this
+ * callback is invoked.
+ */
+ before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
+
SpinLockAcquire(&MyParallelShared->mutex);
MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
MyParallelShared->logicalrep_worker_slot_no = worker_slot;
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index ceea126231..ec34a925f7 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
+ * stopping it. This prevents the leader apply worker from trying to
+ * receive the message from the error queue that might already be detached
+ * by the parallel apply 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.39.1
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: <CAA4eK1LkPBTgcBxKrKahFu47oSK6rDq-AAs0oy4QXH6A8UzEig@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