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, 19 Dec 2022 18:17:25 +0530
Message-ID: <CAA4eK1KdVRffYEYiv9=4tVWhP=Jkfh1h6FM0A1R0NEe8cLT8+g@mail.gmail.com> (raw)
In-Reply-To: <OS0PR01MB5716C3561CF53027F22B47AF94E79@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>
<CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com>
<OS0PR01MB5716B802A1733548A99761AE94129@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1LGKYUDFZ_jFPrU497wQf2HNvt5a+tCTpqSeWSG6kfpSA@mail.gmail.com>
<CAA4eK1Jwh7j86Egk1cye=x2R_yrTjzXGj7Fx12wVybBAEq91kA@mail.gmail.com>
<OS0PR01MB571690D6FF24E9D000ECBDBD94199@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB571639E4141EBF8A2C501E6994189@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1+ELh2vPre3JHyoeKV0A9_V7aXQD0QBPv86WEn7P_rK-g@mail.gmail.com>
<OS0PR01MB57166A377638E57823219E32941A9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KpN=dZPcCj=TLwh3wk9uvPB-M1Jrjw7wt+-L62YrDHFQ@mail.gmail.com>
<OS0PR01MB571604D401DE7A0659BD6609941D9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1LJMpBYbgDVz=g4qig8C2bM10PK=DgW7o1zn8X426vJ+Q@mail.gmail.com>
<OS0PR01MB571663B4EC5D8BE749AACD98941E9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAD21AoD_3ta_29sopuf_KJCExverP5808HjwDeqaHDrbqbxYww@mail.gmail.com>
<OS0PR01MB5716D38323B07EB0A0D097F194E09@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1Kumnc+AR86AGJ666+4Ebfny+nMbs_JZSW3O__UEiGp6w@mail.gmail.com>
<OS0PR01MB57169DAA9A2A6E68EE5E05F094E19@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAD21AoA-i_Gr1Quo2JN56ORafL=phqD=5hUvbA40EHMRZjArQA@mail.gmail.com>
<OS0PR01MB5716E94C5E503CC53177A22994E69@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KkrkAzO-=_k_6s5V1+uNUFkvxMK-33MaFw=JCw5Hqs3w@mail.gmail.com>
<CAA4eK1LRqEXAmh2ZEG-8XK_xDLu2Zzdoj86v4iNXDy++B2SBnQ@mail.gmail.com>
<OS0PR01MB5716C3561CF53027F22B47AF94E79@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Sat, Dec 17, 2022 at 7:34 PM [email protected]
<[email protected]> wrote:
>
> Agreed. I have addressed all the comments and did some cosmetic changes.
> Attach the new version patch set.
>
Few comments:
============
1.
+ if (fileset_state == FS_SERIALIZE_IN_PROGRESS)
+ {
+ pa_lock_stream(MyParallelShared->xid, AccessShareLock);
+ pa_unlock_stream(MyParallelShared->xid, AccessShareLock);
+ }
+
+ /*
+ * We cannot read the file immediately after the leader has serialized all
+ * changes to the file because there may still be messages in the memory
+ * queue. We will apply all spooled messages the next time we call this
+ * function, which should ensure that there are no messages left in the
+ * memory queue.
+ */
+ else if (fileset_state == FS_SERIALIZE_DONE)
+ {
Once we have waited in the FS_SERIALIZE_IN_PROGRESS, the file state
can be FS_SERIALIZE_DONE immediately after that. So, won't it be
better to have a separate if block for FS_SERIALIZE_DONE state? If you
agree to do so then we can probably remove the comment: "* XXX It is
possible that immediately after we have waited for a lock in ...".
2.
+void
+pa_decr_and_wait_stream_block(void)
+{
+ Assert(am_parallel_apply_worker());
+
+ if (pg_atomic_sub_fetch_u32(&MyParallelShared->pending_stream_count, 1) == 0)
I think here the count can go negative when we are in serialize mode
because we don't increase it for serialize mode. I can't see any
problem due to that but OTOH, this doesn't seem to be intended because
in the future if we decide to implement the functionality of switching
back to non-serialize mode, this could be a problem. Also, I guess we
don't even need to try locking/unlocking the stream lock in that case.
One idea to avoid this is to check if the pending count is zero then
if file_set in not available raise an error (elog ERROR), otherwise,
simply return from here.
3. In apply_handle_stream_stop(), we are setting backendstate as idle
for cases TRANS_LEADER_SEND_TO_PARALLEL and TRANS_PARALLEL_APPLY. For
other cases, it is set by stream_stop_internal. I think it would be
better to set the state explicitly for all cases to make the code look
consistent and remove it from stream_stop_internal(). The other reason
to remove setting the state from stream_stop_internal() is that when
that function is invoked from other places like
apply_handle_stream_commit(), it seems to be setting the idle before
actually we reach the idle state.
4. Apart from the above, I have made a few changes in the comments,
see attached.
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] changes_amit_1.patch (4.7K, ../CAA4eK1KdVRffYEYiv9=4tVWhP=Jkfh1h6FM0A1R0NEe8cLT8+g@mail.gmail.com/2-changes_amit_1.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 7422521626..7a30828d44 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -138,10 +138,11 @@
* wait to send messages, and this wait doesn't appear in lmgr.
*
* To avoid this wait, we use a non-blocking write and wait with a timeout. If
- * the timeout is exceeded, the LA will serialize the message to a file and
- * indicate PA-2 that it needs to read that file for the remaining messages.
- * Then LA will start waiting for commit which will detect deadlock if any.
- * See pa_send_data() and enum TransApplyAction.
+ * the timeout is exceeded, the LA will serialize all the pending messages to
+ * a file and indicate PA-2 that it needs to read that file for the remaining
+ * messages. Then LA will start waiting for commit as in the previous case
+ * which will detect deadlock if any. See pa_send_data() and
+ * enum TransApplyAction.
*
* 4) Lock types
*
@@ -575,11 +576,13 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
/*
* Stop the worker if there are enough workers in the pool.
*
- * XXX The worker is also stopped if the leader apply worker needed to
+ * 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 but there 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. Directly stopping the worker avoids needing this complexity.
+ * 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) >
@@ -652,7 +655,7 @@ pa_detach_all_error_mq(void)
}
/*
- * Check if the parallel apply worker is pending due to spooled messages.
+ * Check if there are any pending spooled messages.
*/
static bool
pa_has_spooled_message_pending()
@@ -665,8 +668,8 @@ pa_has_spooled_message_pending()
}
/*
- * Replay the spooled messages in the parallel apply worker if the leader apply
- * worker has finished serializing changes to the file.
+ * Replay the spooled messages once the leader apply worker has finished
+ * serializing changes to the file.
*/
static void
pa_spooled_messages(void)
@@ -699,8 +702,8 @@ pa_spooled_messages(void)
* We cannot read the file immediately after the leader has serialized all
* changes to the file because there may still be messages in the memory
* queue. We will apply all spooled messages the next time we call this
- * function, which should ensure that there are no messages left in the
- * memory queue.
+ * function and that will ensure there are no messages left in the memory
+ * queue.
*/
else if (fileset_state == FS_SERIALIZE_DONE)
{
@@ -811,7 +814,7 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
}
else if (shmq_res == SHM_MQ_WOULD_BLOCK)
{
- /* Check if changes have been serialized to a file. */
+ /* Replay the changes from the file, if any. */
if (pa_has_spooled_message_pending())
{
pa_spooled_messages();
@@ -1207,8 +1210,7 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
* Switch to PARTIAL_SERIALIZE mode for the current transaction -- this means
* that the current data and any subsequent data for this transaction will be
* serialized to a file. This is done to prevent possible deadlocks with
- * another parallel apply worker (refer to the comments atop
- * applyparallelworker.c for details).
+ * another parallel apply worker (refer to the comments atop this file).
*/
void
pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
@@ -1217,7 +1219,7 @@ pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
/*
* The parallel apply worker could be stuck for some reason (say waiting on
* some lock by other backend), so stop trying to send data directly to it
- * and start to serialize data to file instead.
+ * and start serializing data to the file instead.
*/
winfo->serialize_changes = true;
@@ -1488,7 +1490,7 @@ pa_stream_abort(LogicalRepStreamAbortData *abort_data)
}
/*
- * Set the fileset state for the given parallel apply worker. The fileset
+ * Set the fileset state for a particular parallel apply worker. The fileset
* will be set once the leader worker serialized all changes to the file
* so that it can be used by parallel apply worker.
*/
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: <CAA4eK1KdVRffYEYiv9=4tVWhP=Jkfh1h6FM0A1R0NEe8cLT8+g@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