MIME-Version: 1.0
References: 
 <OS0PR01MB5716730C06159452335D870D947F9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAA4eK1Lu-6oXMk7ZaGYLwm3CRLBuzueGbasyHnNpJxu6Mq3mmg@mail.gmail.com>
 <OS3PR01MB6275F145878B4A44586C46CE9E499@OS3PR01MB6275.jpnprd01.prod.outlook.com>
 <CAA4eK1LMVdS6uM7Tw7ANL0BetAd76TKkmAXNNQa0haTe2tax6g@mail.gmail.com>
 <OS3PR01MB6275EFC4B707650DAB9392859E4D9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
 <CAHut+PuVjRgGr4saN7qwq0oB8DANHVR7UfDiciB1Q3cYN54F6A@mail.gmail.com>
 <OS3PR01MB627567CF4B96A23DAFAF77269E4E9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
 <CAA4eK1KEf6w1azJytYnLhd2Lo-=rPkRh1Jv2ePcvs=SKqNQnAQ@mail.gmail.com>
 <CAA4eK1KjGNA8T8O77rRhkv6bRT6OsdQaEy--2hNrJFCc80bN0A@mail.gmail.com>
 <OS0PR01MB571620741B90C769D2727D7F94559@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAD21AoBLPDPCE4jj16ZRE9J8AoxLtvo3yasToaP5kRp6++=yyg@mail.gmail.com>
 <OS0PR01MB57161EFFC1D03B9080A9B812945C9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAD21AoCRDWMebeDNeMNgf4LpTA23si_1xU0T_F2Rh+BD9HU6=Q@mail.gmail.com>
 <CAA4eK1LJoA868HCMmrzPqfcFL=bcFtAi6WWSJ0NKKQp8gspYVQ@mail.gmail.com>
 <CAD21AoDm3224e=se7=ZYt=R+v0_ZJ4E9dd5y2816_rTTCV+G+Q@mail.gmail.com>
 <CAA4eK1JYFXEoFhJAvg1qU=nZrZLw_87X=2YWQGFBbcBGirAUwA@mail.gmail.com>
 <CAD21AoBDLiFHThzfzvrnViTKnsm-pM5YvfTys_96-jBSXpWYqw@mail.gmail.com>
 <CAD21AoCWovvhGBD2uKcQqbk6px6apswuBrs6dR9+WhP1j2LdsQ@mail.gmail.com>
 <OS0PR01MB5716410E7551394B730D433294389@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAD21AoDGLd9h8_DQm8Of5fvAM+MDYzLtFYmEaTSibkg=iwWX7A@mail.gmail.com>
 <OS0PR01MB5716965E416547A42DE1E35094009@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAA4eK1LZY_LM_10JDBNx63RCcVGOcvTjf_c66axFimJNhDZa9g@mail.gmail.com>
 <OS0PR01MB571628A8975A7A984D45134A94009@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <CAA4eK1Jom_hmw19YrtEmZ4MswSsu8imi_JSsRcqgz7G76jvs6Q@mail.gmail.com>
 <OS0PR01MB5716A380FDDE576CCD84F5BE94049@OS0PR01MB5716.jpnprd01.prod.outlook.com>
 <OS0PR01MB5716CF7D3BDE465AFAF17DC094079@OS0PR01MB5716.jpnprd01.prod.outlook.com>
In-Reply-To: 
 <OS0PR01MB5716CF7D3BDE465AFAF17DC094079@OS0PR01MB5716.jpnprd01.prod.outlook.com>
From: Amit Kapila <amit.kapila16@gmail.com>
Date: Fri, 18 Nov 2022 07:56:45 +0530
Message-ID: 
 <CAA4eK1KWgReYbpwEMh1H1ohHoYirv4Aa=6v13MutCF9NvHTc5A@mail.gmail.com>
Subject: Re: Perform streaming logical transactions by background workers and
 parallel apply
To: "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>
Cc: Masahiko Sawada <sawada.mshk@gmail.com>,
	"wangw.fnst@fujitsu.com" <wangw.fnst@fujitsu.com>,
 Peter Smith <smithpb2250@gmail.com>,
	Dilip Kumar <dilipbalaut@gmail.com>,
 "shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>,
	PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Content-Type: text/plain; charset="UTF-8"
Archived-At: 
 <https://www.postgresql.org/message-id/CAA4eK1KWgReYbpwEMh1H1ohHoYirv4Aa%3D6v13MutCF9NvHTc5A%40mail.gmail.com>
Precedence: bulk

On Wed, Nov 16, 2022 at 1:50 PM houzj.fnst@fujitsu.com
<houzj.fnst@fujitsu.com> wrote:
>
> On Tuesday, November 15, 2022 7:58 PM houzj.fnst@fujitsu.com <houzj.fnst@fujitsu.com> wrote:
>
> I noticed that I didn't add CHECK_FOR_INTERRUPTS while retrying send message.
> So, attach the new version which adds that. Also attach the 0004 patch that
> restarts logical replication with temporarily disabling the parallel apply if
> failed to apply a transaction in parallel apply worker.
>

Few comments on v48-0001
======================
1. The variable name pending_message_count seems to indicate a number
of pending messages but normally it is pending start/stop streams
except for probably rollback to savepoint case. Shall we name it
pending_stream_count and change the comments accordingly?

2. The variable name abort_toplevel_transaction seems unnecessarily
long. Shall we rename it to toplevel_xact or something like that?

3.
+ /*
+ * Increment the number of messages waiting to be processed by
+ * parallel apply worker.
+ */
+ if (!abort_toplevel_transaction)
+ pg_atomic_add_fetch_u32(&(winfo->shared->pending_message_count), 1);
+ else
+ pa_unlock_stream(xid, AccessExclusiveLock);

It is better to explain here why different actions are required for
subtransaction and transaction rather than the current comment.

4.
+
+ if (abort_toplevel_transaction)
+ {
+ (void) pa_free_worker(winfo, xid);
+ }

{} is not required here.

5.
/*
+ * Although the lock can be automatically released during transaction
+ * rollback, but we still release the lock here as we may not in a
+ * transaction.
+ */
+ pa_unlock_transaction(xid, AccessShareLock);
+

It is better to explain for which case (I think it is for empty xacts)
it will be useful to release it explicitly.

6.
+ *
+ * XXX We can avoid sending pairs of the START/STOP messages to the parallel
+ * worker because unlike apply worker it will process only one transaction at a
+ * time. However, it is not clear whether any optimization is worthwhile
+ * because these messages are sent only when the logical_decoding_work_mem
+ * threshold is exceeded.
  */
 static void
 apply_handle_stream_start(StringInfo s)

I think this comment is no longer valid as now we need to wait for the
next stream at stream_stop message and also need to acquire the lock
in stream_start message. So, I think it is better to remove it unless
I am missing something.

7. I am able to compile applyparallelworker.c by commenting few of the
header includes. Please check if those are really required.
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
//#include "mb/pg_wchar.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logicallauncher.h"
//#include "replication/logicalworker.h"
#include "replication/origin.h"
//#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
//#include "storage/procarray.h"
#include "tcop/tcopprot.h"
#include "utils/inval.h"
#include "utils/memutils.h"
//#include "utils/resowner.h"
#include "utils/syscache.h"

8.
+/*
+ * Is there a message sent by parallel apply worker which we need to receive?
+ */
+volatile sig_atomic_t ParallelApplyMessagePending = false;

This comment and variable are placed in applyparallelworker.c, so 'we'
in the above sentence is not clear. I think you need to use leader
apply worker instead.

9.
+static ParallelApplyWorkerInfo *pa_get_free_worker(void);

Will it be better if we name this function pa_get_available_worker()?

-- 
With Regards,
Amit Kapila.