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: Wed, 14 Dec 2022 12:18:41 +0530
Message-ID: <CAA4eK1Kumnc+AR86AGJ666+4Ebfny+nMbs_JZSW3O__UEiGp6w@mail.gmail.com> (raw)
In-Reply-To: <OS0PR01MB5716D38323B07EB0A0D097F194E09@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>
On Wed, Dec 14, 2022 at 9:50 AM [email protected]
<[email protected]> wrote:
>
> On Tuesday, December 13, 2022 11:25 PM Masahiko Sawada <[email protected]> wrote:
> >
> > Here are comments on v59 0001, 0002 patches:
>
> Thanks for the comments!
>
> > +void
> > +pa_increment_stream_block(ParallelApplyWorkerShared *wshared) {
> > + while (1)
> > + {
> > + SpinLockAcquire(&wshared->mutex);
> > +
> > + /*
> > + * Don't try to increment the count if the parallel
> > apply worker is
> > + * taking the stream lock. Otherwise, there would be
> > a race condition
> > + * that the parallel apply worker checks there is no
> > pending streaming
> > + * block and before it actually starts waiting on a
> > lock, the leader
> > + * sends another streaming block and take the stream
> > lock again. In
> > + * this case, the parallel apply worker will start
> > waiting for the next
> > + * streaming block whereas there is actually a
> > pending streaming block
> > + * available.
> > + */
> > + if (!wshared->pa_wait_for_stream)
> > + {
> > + wshared->pending_stream_count++;
> > + SpinLockRelease(&wshared->mutex);
> > + break;
> > + }
> > +
> > + SpinLockRelease(&wshared->mutex);
> > + }
> > +}
> >
> > I think we should add an assertion to check if we don't hold the stream lock.
> >
> > I think that waiting for pa_wait_for_stream to be false in a busy loop is not a
> > good idea. It's not interruptible and there is not guarantee that we can break
> > from this loop in a short time. For instance, if PA executes
> > pa_decr_and_wait_stream_block() a bit earlier than LA executes
> > pa_increment_stream_block(), LA has to wait for PA to acquire and release the
> > stream lock in a busy loop. It should not be long in normal cases but the
> > duration LA needs to wait for PA depends on PA, which could be long. Also
> > what if PA raises an error in
> > pa_lock_stream() due to some reasons? I think LA won't be able to detect the
> > failure.
> >
> > I think we should at least make it interruptible and maybe need to add some
> > sleep. Or perhaps we can use the condition variable for this case.
>
Or we can leave this while (true) logic altogether for the first
version and have a comment to explain this race. Anyway, after
restarting, it will probably be solved. We can always change this part
of the code later if this really turns out to be problematic.
> Thanks for the analysis, I will research this part.
>
> > ---
> > In worker.c, we have the following common pattern:
> >
> > case TRANS_LEADER_PARTIAL_SERIALIZE:
> > write change to the file;
> > do some work;
> > break;
> >
> > case TRANS_LEADER_SEND_TO_PARALLEL:
> > pa_send_data();
> >
> > if (winfo->serialize_changes)
> > {
> > do some worker required after writing changes to the file.
> > }
> > :
> > break;
> >
> > IIUC there are two different paths for partial serialization: (a) where
> > apply_action is TRANS_LEADER_PARTIAL_SERIALIZE, and (b) where
> > apply_action is TRANS_LEADER_PARTIAL_SERIALIZE and
> > winfo->serialize_changes became true. And we need to match what we do
> > in (a) and (b). Rather than having two different paths for the same case, how
> > about falling through TRANS_LEADER_PARTIAL_SERIALIZE when we could not
> > send the changes? That is, pa_send_data() just returns false when the timeout
> > exceeds and we need to switch to serialize changes, otherwise returns true. If it
> > returns false, we prepare for switching to serialize changes such as initializing
> > fileset, and fall through TRANS_LEADER_PARTIAL_SERIALIZE case. The code
> > would be like:
> >
> > case TRANS_LEADER_SEND_TO_PARALLEL:
> > ret = pa_send_data();
> >
> > if (ret)
> > {
> > do work for sending changes to PA.
> > break;
> > }
> >
> > /* prepare for switching to serialize changes */
> > winfo->serialize_changes = true;
> > initialize fileset;
> > acquire stream lock if necessary;
> >
> > /* FALLTHROUGH */
> > case TRANS_LEADER_PARTIAL_SERIALIZE:
> > do work for serializing changes;
> > break;
>
> I think that the suggestion is to extract the code that switch to serialize
> mode out of the pa_send_data(), and then we need to add that logic in all the
> functions which call pa_send_data(), I am not sure if it looks better as it
> might introduce some more codes in each handling function.
>
How about extracting the common code from apply_handle_stream_commit
and apply_handle_stream_prepare to a separate function say
pa_xact_finish_common()? I see there is a lot of common code (unlock
the stream, wait for the finish, store flush location, free worker
info) in both the functions for TRANS_LEADER_PARTIAL_SERIALIZE and
TRANS_LEADER_SEND_TO_PARALLEL cases.
>
> > ---
> > void
> > pa_lock_stream(TransactionId xid, LOCKMODE lockmode) {
> > LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
> > PARALLEL_APPLY_LOCK_STREAM,
> > lockmode); }
> >
> > I think since we don't need to let the caller to specify the lock mode but need
> > only shared and exclusive modes, we can make it simple by having a boolean
> > argument say shared instead of lockmode.
>
> I personally think passing the lockmode would make the code more clear
> than passing a Boolean value.
>
+1.
I have made a few changes in the newly added comments and function
name in the attached patch. Kindly include this if you find the
changes okay.
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] changes_amit_v60.patch (2.1K, ../CAA4eK1Kumnc+AR86AGJ666+4Ebfny+nMbs_JZSW3O__UEiGp6w@mail.gmail.com/2-changes_amit_v60.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 2e2cd0d535..158dbd93c9 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -707,10 +707,10 @@ pa_stop_idle_workers(void)
}
/*
- * Check if the parallel apply worker is pending due to spooled messages.
+ * Returns true, if there is any pending spooled message, false otherwise.
*/
static bool
-pa_has_spooled_message_pending()
+pa_is_spooled_message_pending()
{
PartialFileSetState fileset_state;
@@ -870,7 +870,7 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
else if (shmq_res == SHM_MQ_WOULD_BLOCK)
{
/* Check if changes have been serialized to a file. */
- if (pa_has_spooled_message_pending())
+ if (pa_is_spooled_message_pending())
{
pa_spooled_messages();
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index faebceb4c7..59a4366fec 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1558,15 +1558,17 @@ apply_handle_stream_start(StringInfo s)
pa_send_data(winfo, s->len, s->data, !first_segment);
+ /*
+ * Once we start serializing the changes, the parallel apply worker
+ * will wait for the leader to release the stream lock until the
+ * end of the transaction. So, we don't need to release the lock or
+ * increment the stream count in that case.
+ */
if (!winfo->serialize_changes)
{
/*
- * Unlock the shared object lock when no change was serialized
- * so that the parallel apply worker can continue to receive
- * changes. No need to take the lock if switched to serialize
- * mode because the parallel apply worker should wait for the
- * leader to release the stream lock until the end of the
- * transaction.
+ * Unlock the shared object lock so that the parallel apply
+ * worker can continue to receive changes.
*/
if (!first_segment)
pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
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: <CAA4eK1Kumnc+AR86AGJ666+4Ebfny+nMbs_JZSW3O__UEiGp6w@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