public inbox for [email protected]
help / color / mirror / Atom feedFrom: Amit Kapila <[email protected]>
To: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Masahiko Sawada <[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: Tue, 30 Aug 2022 12:12:57 +0530
Message-ID: <CAA4eK1+e8JsiC8uMZPU25xQRyxNvVS24M4=Zy-xD18jzX+vrmA@mail.gmail.com> (raw)
In-Reply-To: <OS0PR01MB5716DB4C67394D1676429D3594769@OS0PR01MB5716.jpnprd01.prod.outlook.com>
References: <OS3PR01MB62758DBE8FA12BA72A43AC819EB89@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+Pu5ah+NhYMfb7KTbkYTxnuGY2j-7vT0G1MZZsS4SKiHew@mail.gmail.com>
<OS3PR01MB62755C6C9A75EB09F7218B589E839@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB627594D75E870BBB45E2E80A9E839@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+PvN7fwtUE=bidzrsOUXSt+JpnkJztZ-Jn5t86moofaZ6g@mail.gmail.com>
<OS3PR01MB62758A6AAED27B3A848CEB7A9E8F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB62752B29CEC2592E18236A9A9E909@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAFiTN-uq5Mp+o+_u0iDc8OQV-q7CWf5W=dPgw2ORwxsi6yjVfw@mail.gmail.com>
<CAA4eK1LB1s=B+krpn9mGbZq46pVei0FD7WjhJORq4bD2GU4fng@mail.gmail.com>
<CAFiTN-sK3bVppVAioPfxGjZqY9sHKVM8NooK9Wuex2OCTK25wg@mail.gmail.com>
<OS0PR01MB571651D89C744F84E20AAFE394979@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1K0zkBJQtF1ZNhSvcwh_+pdCaUTf8+bBZR95yP92c0CuQ@mail.gmail.com>
<OS3PR01MB6275D64BE7726B0221B15F389E9F9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+PutJBXuoy3DB5S7X9sZitfZp4qsXfzgkt5C5=CTNtX-Ug@mail.gmail.com>
<OS3PR01MB6275739E73E8BEC5D13FB6739E6B9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275BA4CBB62CBE8C094E53D9E6A9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1+_oHZHoDooAR7QcYD2CeTUWNSwkqVcLWC2iQijAJC4Cg@mail.gmail.com>
<CAA4eK1++B2PFQiin2ZGMs_tszsQ=NJeKuj18-M4PEgckfEYqNA@mail.gmail.com>
<OS0PR01MB57166F15F4542FAE69A74EC094739@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1JERguGiYoid9CMEOamZoD4jF_R8+mkNVS=mV-X=CXe=A@mail.gmail.com>
<OS0PR01MB5716DB4C67394D1676429D3594769@OS0PR01MB5716.jpnprd01.prod.outlook.com>
On Mon, Aug 29, 2022 at 5:01 PM [email protected]
<[email protected]> wrote:
>
> On Thursday, August 25, 2022 7:33 PM Amit Kapila <[email protected]> wrote:
> >
>
> > 11.
> > + /*
> > + * Attach to the message queue.
> > + */
> > + mq = shm_toc_lookup(toc, APPLY_BGWORKER_KEY_ERROR_QUEUE, false);
> > + shm_mq_set_sender(mq, MyProc);
> > + error_mqh = shm_mq_attach(mq, seg, NULL);
> > + pq_redirect_to_shm_mq(seg, error_mqh);
> > +
> > + /*
> > + * Now, we have initialized DSM. Attach to slot.
> > + */
> > + logicalrep_worker_attach(worker_slot);
> > + MyParallelShared->logicalrep_worker_generation =
> > MyLogicalRepWorker->generation;
> > + MyParallelShared->logicalrep_worker_slot_no = worker_slot;
> > +
> > + pq_set_parallel_leader(MyLogicalRepWorker->apply_leader_pid,
> > + InvalidBackendId);
> >
> > Is there a reason to set parallel_leader immediately after
> > pq_redirect_to_shm_mq() as we are doing parallel.c?
>
> Moved the code.
>
Sorry, if I was not clear but what I wanted was something like the below:
diff --git a/src/backend/replication/logical/applyparallelworker.c
b/src/backend/replication/logical/applyparallelworker.c
index 832e99cd48..6646e00658 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -480,6 +480,9 @@ ApplyParallelWorkerMain(Datum main_arg)
mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, false);
shm_mq_set_sender(mq, MyProc);
error_mqh = shm_mq_attach(mq, seg, NULL);
+ pq_redirect_to_shm_mq(seg, error_mqh);
+ pq_set_parallel_leader(MyLogicalRepWorker->apply_leader_pid,
+ InvalidBackendId);
/*
* Primary initialization is complete. Now, we can attach to
our slot. This
@@ -490,10 +493,6 @@ ApplyParallelWorkerMain(Datum main_arg)
MyParallelShared->logicalrep_worker_generation =
MyLogicalRepWorker->generation;
MyParallelShared->logicalrep_worker_slot_no = worker_slot;
- pq_redirect_to_shm_mq(seg, error_mqh);
- pq_set_parallel_leader(MyLogicalRepWorker->apply_leader_pid,
- InvalidBackendId);
-
MyLogicalRepWorker->last_send_time =
MyLogicalRepWorker->last_recv_time =
MyLogicalRepWorker->reply_time = 0;
Few other comments on v25-0001*
============================
1.
+ {
+ {"max_apply_parallel_workers_per_subscription",
+ PGC_SIGHUP,
+ REPLICATION_SUBSCRIBERS,
+ gettext_noop("Maximum number of apply parallel workers per subscription."),
+ NULL,
+ },
+ &max_apply_parallel_workers_per_subscription,
Let's model this to max_parallel_workers_per_gather and name this
max_parallel_apply_workers_per_subscription.
+typedef struct ApplyParallelWorkerEntry
+{
+ TransactionId xid; /* Hash key -- must be first */
+ ApplyParallelWorkerInfo *winfo;
+} ApplyParallelWorkerEntry;
+
+/* Apply parallel workers hash table (initialized on first use). */
+static HTAB *ApplyParallelWorkersHash = NULL;
+static List *ApplyParallelWorkersFreeList = NIL;
+static List *ApplyParallelWorkersList = NIL;
Similarly, for above let's name them as ParallelApply*. I think in
comments/doc changes it is better to refer as parallel apply worker.
we can keep filename as it is.
2.
+ * If there are enough apply parallel workers(reache half of the
+ * max_apply_parallel_workers_per_subscription)
/reache/reached. There should be a space before (.
3.
+ * The dynamic shared memory segment will contain (1) a shm_mq that can be used
+ * to transport errors (and other messages reported via elog/ereport) from the
+ * apply parallel worker to leader apply worker (2) another shm_mq that can
+ * be used to transport changes in the transaction from leader apply worker to
+ * apply parallel worker (3) necessary information to be shared among apply
+ * parallel workers to leader apply worker
I think it is better to use send instead of transport in above
paragraph. In (3), /apply parallel workers to leader apply
worker/apply parallel workers and leader apply worker
4.
handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
{
...
...
+ else if (apply_action == TA_SEND_TO_PARALLEL_WORKER)
+ {
+ parallel_apply_send_data(winfo, s->len, s->data);
It is better to have an Assert for winfo being non-null here and other
similar usages.
--
With Regards,
Amit Kapila.
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: <CAA4eK1+e8JsiC8uMZPU25xQRyxNvVS24M4=Zy-xD18jzX+vrmA@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