MIME-Version: 1.0
References: 
 <CA+UBfa=vDV8wbmAV0pgrx-FuJh+x8YOW23vJ90Jzr=14rV+9jA@mail.gmail.com>
 <OS9PR01MB12149A4E7927072A215AEED69F565A@OS9PR01MB12149.jpnprd01.prod.outlook.com>
 <CA+UBfakmkdtauuRsOVXFqhFVJt0nTdEadx94tJn+qG0Pe8Wjfw@mail.gmail.com>
 <CAN4CZFM7FV0VTNkujD=Mb7tNa+jkmEfnX7carvj95fY6Tp11FQ@mail.gmail.com>
In-Reply-To: 
 <CAN4CZFM7FV0VTNkujD=Mb7tNa+jkmEfnX7carvj95fY6Tp11FQ@mail.gmail.com>
From: Imran Zaheer <imran.zhir@gmail.com>
Date: Wed, 18 Mar 2026 12:43:37 +0500
Message-ID: 
 <CA+UBfamW6NuuMMQTDRPDQ0a9fWN_u2OvjEF98u3CfYKTBcOZMw@mail.gmail.com>
Subject: Re: [WIP] Pipelined Recovery
To: Zsolt Parragi <zsolt.parragi@percona.com>
Cc: "Hayato Kuroda (Fujitsu)" <kuroda.hayato@fujitsu.com>,
 pgsql-hackers <pgsql-hackers@postgresql.org>
Content-Type: multipart/alternative; boundary="000000000000c3f42f064d479cee"
Archived-At: 
 <https://www.postgresql.org/message-id/CA%2BUBfamW6NuuMMQTDRPDQ0a9fWN_u2OvjEF98u3CfYKTBcOZMw%40mail.gmail.com>
Precedence: bulk

--000000000000c3f42f064d479cee
Content-Type: text/plain; charset="UTF-8"

Hi Zsolt.

Thanks alot for the review and pointing out the bugs. I have fixed the bugs
you mentioned in my new patch set. But
patchset mail is held for moderation for some reason.

>
>   if (reachedRecoveryTarget)
>   {
> + if (wal_pipeline_enabled)
> + WalPipeline_Stop();
>
> What if we didn't reach the recovery target, shouldn't we stop the
> pipelines then?
>

I have fixed the bugs shutdown logic.

As we already know we will exist the recovery redo loop in
`PerformWalRecovery()` only in two cases

1: Recovery target reached:
In this case consumers will call to stop the pipeline.

@@ -1807,6 +1931,9 @@ PerformWalRecovery(void)

  if (reachedRecoveryTarget)
  {
+ if (wal_pipeline_enabled)
+ WalPipeline_Stop();
+

2: Available pg_wal is consumed and now more wal to read.
In this case pipeline producers will send the shutdown msg to the consumer.
Consumer will
detect this message and then call ` WalPipeline_Stop`. This is the case
where we cannot read
more records and the while loop will break here.

+ if (decoded_record)
+ {
+ record = &decoded_record->header;
+ return record;
+ }
+ else
+ {
+ /*
+ * We will end up here only when pipeline couldn't read more
+ * records and have sent a shutdown msg. We will acknowldge this
+ * and will trigger request to stop the pipeline workers.
+ */
+ WalPipeline_Stop();
+ return NULL;
+ }


Hope this makes sense.

Once again thanks for reporting the bugs. You will receive the new patchset
mail soon once it is cleared from
the moderation.

Looking forward to your reviews, comments, etc.

Regards,
Imran Zaheer

--000000000000c3f42f064d479cee
Content-Type: text/html; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr"><div><div><div><div><div>Hi Zsolt.<br><br>Thanks alot for =
the review and pointing out the bugs. I have fixed the bugs you mentioned i=
n my new patch set. But<br>patchset mail is held for moderation for some re=
ason.<br><br>&gt;<br>&gt; =C2=A0 if (reachedRecoveryTarget)<br>&gt; =C2=A0 =
{<br>&gt; + if (wal_pipeline_enabled)<br>&gt; + WalPipeline_Stop();<br>&gt;=
<br>&gt; What if we didn&#39;t reach the recovery target, shouldn&#39;t we =
stop the<br>&gt; pipelines then?<br>&gt;<br><br></div>I have fixed the bugs=
 shutdown logic.<br><br></div>As we already know we will exist the recovery=
 redo loop in `PerformWalRecovery()` only in two cases<br><br></div>1: Reco=
very target reached:<br></div><div>In this case consumers will call to stop=
 the pipeline.<br><br>@@ -1807,6 +1931,9 @@ PerformWalRecovery(void)<br>=C2=
=A0<br>=C2=A0		if (reachedRecoveryTarget)<br>=C2=A0		{<br>+			if (wal_pipel=
ine_enabled)<br>+				WalPipeline_Stop();<br>+</div><div><br></div>2: Availa=
ble pg_wal is consumed and now more wal to read.</div><div>In this case pip=
eline producers will send the shutdown msg to the consumer. Consumer will</=
div><div>detect this message=C2=A0and then call `			WalPipeline_Stop`. This=
 is the case where we cannot read</div><div>more records and the while=C2=
=A0loop will break here.<br><br>+		if (decoded_record)<br>+		{<br>+			recor=
d =3D &amp;decoded_record-&gt;header;<br>+			return record;<br>+		}<br>+		e=
lse<br>+		{<br>+			/*<br>+			 * We will end up here only when pipeline coul=
dn&#39;t read more<br>+			 * records and have sent a shutdown msg. We will =
acknowldge this<br>+			 * and will trigger request to stop the pipeline wor=
kers.<br>+			 */<br>+			WalPipeline_Stop();<br>+			return NULL;<br>+		}<br>=
<br><br></div><div>Hope this makes sense.=C2=A0<br><br></div><div>Once agai=
n thanks for reporting the bugs. You will receive the new patchset mail soo=
n once it is cleared=C2=A0from</div><div>the moderation.</div><div><br>Look=
ing forward to your reviews, comments, etc.<br><br>Regards,<br>
Imran Zaheer</div></div>

--000000000000c3f42f064d479cee--