From: Chao Li <li.evan.chao@gmail.com>
Message-Id: <0BCA1C2D-B92C-459E-B1A6-6D06BA4C62CF@gmail.com>
Content-Type: multipart/mixed;
	boundary="Apple-Mail=_7B646548-F559-49EC-8DB9-78AF6DB09FAE"
Mime-Version: 1.0 (Mac OS X Mail 16.0 \(3826.700.81\))
Subject: Re: Optimize LISTEN/NOTIFY
Date: Sun, 26 Oct 2025 12:11:25 +0800
In-Reply-To: 
 <CAE7r3MK-3AOdh1mpZ8hw9h6F_i0D5RMoAy7CttnfCJRpB8GJDA@mail.gmail.com>
Cc: Joel Jacobson <joel@compiler.org>,
 Tom Lane <tgl@sss.pgh.pa.us>,
 pgsql-hackers <pgsql-hackers@postgresql.org>
To: Arseniy Mukhin <arseniy.mukhin.dev@gmail.com>
References: <6899c044-4a82-49be-8117-e6f669765f7e@app.fastmail.com>
 <165530.1752362320@sss.pgh.pa.us>
 <02a7cd37-e2fc-4212-8b19-f8c239c95fb8@app.fastmail.com>
 <e396ecf3-4227-4918-b9ff-e9568dcebcf0@app.fastmail.com>
 <b6a427fe-2ed6-4568-85d9-207a68172617@app.fastmail.com>
 <CAK80=jhmE40KVqQ3ho37MArS7cAED1p9m7uikDxcnDmqdW7t8A@mail.gmail.com>
 <96f00bf1-cc9d-4520-9d02-9e14e7767c88@app.fastmail.com>
 <30c2aa7d-dd6c-4b68-a2e4-f217a1a34acf@app.fastmail.com>
 <CA+hUKGLrMGkWDB0cwTa0RqD+AF7O-Ywgck8aVYKwOQnZgYRRug@mail.gmail.com>
 <0b4d402a-9ac2-4aa8-acf8-8231dbe579ea@app.fastmail.com>
 <c8e84bb2-d2e5-4a77-b547-3a93c5778e7e@app.fastmail.com>
 <3095599.1758644879@sss.pgh.pa.us>
 <0dc6a2cc-5216-4dc1-9dd2-430cafc6095b@app.fastmail.com>
 <E7D8A5CD-C445-4713-B6B7-6CC1D3A71161@gmail.com>
 <baf71275-9083-41be-ae13-f80821541a1a@app.fastmail.com>
 <52CC167F-763B-4ECA-B0B4-DAB381816828@gmail.com>
 <d35a114d-204a-4b40-9284-06749532b724@app.fastmail.com>
 <9186C6D0-F7A9-482A-9183-89E530B57E36@gmail.com>
 <da0996!!!!!!d8-ecf9-490d-bbd0-2a5bdb5b115a@app.fastmail.com>
 <D1CF7CA5-213A-4FC1-86DB-052D58904A0A@gmail.com>
 <cad78ff4-aae9-4c1c-8e16-ff0ccd4957f1@app.fastmail.com>
 <1073593.1759423179@sss.pgh.pa.us>
 <4bd5e6c4-6fa7-44bb-869d-59a32a331fa8@app.fastmail.com>
 <CAFY6G8dap-bCnAnMG-2Gzew8yv2Vbi9gsx9+yszKMmd57ygfvA@mail.gmail.com>
 <85828f29-e72e-4400-94f3-9a69bc8dc239@app.fastmail.com>
 <2495353.1759860890@sss.pgh.pa.us>
 <8aeae418-92a6-4bbd-9c06-9574c79e59f7@app.fastmail.com>
 <2531672.1759868124@sss.pgh.pa.us>
 <474efa78-337c-41cd-a73a-f845a0115109@app.fastmail.com>
 <2749343.1759949176@sss.pgh.pa.us>
 <8bfca2be-1ec0-4e15-aafb-0b7b661fe936@app.fastmail.com>
 <9eba307f-f2fb-48f0-9507-2e197f39ef9e@app.fastmail.com>
 <cde2a91d-d47a-43b5-ab27-f98aa739d38e@app.fastmail.com>
 <8c71183a-0d28-4bcf-a806-78446ff95404@app.fastmail.com>
 <1009807.1760476747@sss.pgh.pa.us>
 <1F7227F5-C33D-4E2C-8511-33F1468590D0@gmail.com>
 <0a5a20d3-4621-46b3-b2ab-903f63a20dea@app.fastmail.com>
 <6F913129-ABEF-4004-AAF3-F22FC34!29AE8@gmail.com>
 <cbabbb45-ec27-47f1-a8f3-eb29073e9aef@app.fastmail.com>
 <ceab2c54-3463-4029-a7d8-895a617bb457@app.fastmail.com>
 <1547585.1760645808@sss.pgh.pa.us>
 <b8df8040-1b15-4a1a-b5f6-74acebbd2d0c@app.fastmail.com>
 <CAE7r3MLivh1sHWF06hrVXkiQbw-KChPcQsh+9CheXprm5vRVMQ@mail.gmail.com>
 <14865EB6-0BF4-462B-9072-10BDAC10C052@gmail.com>
 <CAE7r3MK-3AOdh1mpZ8hw9h6F_i0D5RMoAy7CttnfCJRpB8GJDA@mail.gmail.com>
Archived-At: 
 <https://www.postgresql.org/message-id/0BCA1C2D-B92C-459E-B1A6-6D06BA4C62CF%40gmail.com>
Precedence: bulk


--Apple-Mail=_7B646548-F559-49EC-8DB9-78AF6DB09FAE
Content-Transfer-Encoding: quoted-printable
Content-Type: text/plain;
	charset=utf-8


> On Oct 23, 2025, at 18:02, Arseniy Mukhin =
<arseniy.mukhin.dev@gmail.com> wrote:
>=20
> Hi,
>=20
> On Thu, Oct 23, 2025 at 11:17=E2=80=AFAM Chao Li =
<li.evan.chao@gmail.com> wrote:
>>=20
>>=20
>>=20
>>> On Oct 21, 2025, at 00:43, Arseniy Mukhin =
<arseniy.mukhin.dev@gmail.com> wrote:
>>>=20
>>>=20
>>> I managed to reproduce the race with v20-alt3. I tried to write a =
TAP
>>> test reproducing the issue, so it was easier to validate changes.
>>> Please find the attached TAP test. I added it to some random package
>>> for simplicity.
>>>=20
>>=20
>> With alt3, as we have acquired the notification lock after reading =
every message to update the POS, I think we can do a little bit more =
optimization:
>>=20
>> The notifier: in SignalBackend()
>>    * Now we check if a listener=E2=80=99s pos equals to =
beforeWritePos, then we do =E2=80=9Cdirectly advancement=E2=80=9D
>>    * We can change to if a listener=E2=80=99s pos is between =
beforeWritePos and afterWritePos, then we can do the advancement.
>>=20
>> The listener: in asyncQueueReadAllNotifications():
>>    * With alt3, we only lock and update pos
>>    * We can do more. If current pos in shared memory is after that =
local pos, then meaning some notifier has done an advancement, so it can =
stop reading.
>>=20
>=20
> I think this would be a reasonable optimization if it weren't for the
> race condition mentioned above. The problem is that if the local pos
> lags behind the shared memory pos, it could point to a truncated queue
> segment, so we shouldn't allow that.
>=20

I figured out a way to resolve the race condition for alt3:

* add an awakening flag for every listener, this flag is only set by =
listeners
* add an advisory pos for every listener, similar to alt1
* if a listener is awaken, notify only updates the listener=E2=80=99s =
advisory pos; otherwise directly advance its position.
* If a running listener see current pos is behind advisory pos, then =
stop reading

See more details in attach patch file, I added code comments for my =
changes. Now the TAP test won=E2=80=99t hit the race condition.=20
```
# +++ tap check in src/test/authentication +++
t/008_listen-pos-race.pl .. skipped: Injection points not supported by =
this build
Files=3D1, Tests=3D0,  0 wallclock secs ( 0.00 usr  0.00 sys +  0.03 =
cusr  0.01 csys =3D  0.04 CPU)
Result: NOTESTS
```

And with my solution, listeners longer will still use local pos, so that =
no longer need to acquire notification lock in every loop.

The patch stack is: v20 patch -> alt3 patch -> tap patch -> my patch. =
Please see if my solution works.

I also made a tiny change in the TAP script to allow it to terminate =
gracefully.

Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/


--Apple-Mail=_7B646548-F559-49EC-8DB9-78AF6DB09FAE
Content-Disposition: attachment;
	filename=fix-race.patch
Content-Type: application/octet-stream;
	x-unix-mode=0644;
	name="fix-race.patch"
Content-Transfer-Encoding: 7bit

commit c7daefa51118d2041623b14b7f26c9177ac0b6cd
Author: Chao Li (Evan) <lic@highgo.com>
Date:   Sat Oct 25 15:42:26 2025 +0800

    fix race condition

diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 6e8d728e9ce..3c8a640ebed 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -250,6 +250,11 @@ typedef struct QueuePosition
 #define QUEUE_POS_EQUAL(x,y) \
 	((x).page == (y).page && (x).offset == (y).offset)
 
+/* returns true if x comes before y in queue order */
+#define QUEUE_POS_PRECEDES(x,y) \
+	(asyncQueuePagePrecedes((x).page, (y).page) || \
+	 ((x).page == (y).page && (x).offset < (y).offset))
+
 #define QUEUE_POS_IS_ZERO(x) \
 	((x).page == 0 && (x).offset == 0)
 
@@ -287,7 +292,9 @@ typedef struct QueueBackendStatus
 	Oid			dboid;			/* backend's database OID, or InvalidOid */
 	ProcNumber	nextListener;	/* id of next listener, or INVALID_PROC_NUMBER */
 	QueuePosition pos;			/* backend has read queue up to here */
+	QueuePosition advisoryPos;	/* advisory position for this backend */
 	bool		wakeupPending;	/* signal sent but not yet processed */
+	bool        awakening;	/* backend is awakening */
 } QueueBackendStatus;
 
 /*
@@ -348,7 +355,9 @@ static dshash_table *channelHash = NULL;
 #define QUEUE_BACKEND_DBOID(i)		(asyncQueueControl->backend[i].dboid)
 #define QUEUE_NEXT_LISTENER(i)		(asyncQueueControl->backend[i].nextListener)
 #define QUEUE_BACKEND_POS(i)		(asyncQueueControl->backend[i].pos)
+#define QUEUE_BACKEND_ADVISORY_POS(i)	(asyncQueueControl->backend[i].advisoryPos)
 #define QUEUE_BACKEND_WAKEUP_PENDING(i)	(asyncQueueControl->backend[i].wakeupPending)
+#define QUEUE_BACKEND_AWAKENING(i)	(asyncQueueControl->backend[i].awakening)
 
 /*
  * The SLRU buffer area through which we access the notification queue
@@ -675,7 +684,9 @@ AsyncShmemInit(void)
 			QUEUE_BACKEND_DBOID(i) = InvalidOid;
 			QUEUE_NEXT_LISTENER(i) = INVALID_PROC_NUMBER;
 			SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0);
+			SET_QUEUE_POS(QUEUE_BACKEND_ADVISORY_POS(i), 0, 0);
 			QUEUE_BACKEND_WAKEUP_PENDING(i) = false;
+			QUEUE_BACKEND_AWAKENING(i) = false;
 		}
 	}
 
@@ -1954,6 +1965,7 @@ SignalBackends(void)
 	ProcNumber *procnos;
 	int			count;
 	ListCell   *lc;
+	List *interestedProcs = NIL;
 
 	INJECTION_POINT("listen-notify-signal-backends", NULL);
 
@@ -2002,6 +2014,12 @@ SignalBackends(void)
 			int32		pid;
 			QueuePosition pos;
 
+			// XXX: Use a list to record listeners interested in any of the pending channels.
+			// List is not the best choice, so it we decide to take this apprach, we
+			// can optimize it later by using a hash or bitmap.
+			if (!list_member_int(interestedProcs, i))
+				interestedProcs = lappend_int(interestedProcs, i);
+
 			if (QUEUE_BACKEND_WAKEUP_PENDING(i))
 				continue;
 
@@ -2054,19 +2072,41 @@ SignalBackends(void)
 		int64		lag;
 		int32		pid;
 
-		if (QUEUE_BACKEND_WAKEUP_PENDING(i))
-			continue;
+		/* XXX we cannot rely on wakeupPending here, because the flag might be set by another notifier. */
+		//if (QUEUE_BACKEND_WAKEUP_PENDING(i))
+		//	continue;
 
 		pos = QUEUE_BACKEND_POS(i);
 
-		/* Direct advancement for idle backends at the old head */
-		if (pendingNotifies != NULL &&
-			QUEUE_POS_EQUAL(pos, queueHeadBeforeWrite))
+		if (!list_member_int(interestedProcs, i))
 		{
-			QUEUE_BACKEND_POS(i) = queueHeadAfterWrite;
-			pos = queueHeadAfterWrite;
+			/* Direct advancement for idle backends at the old head */
+			if (pendingNotifies != NULL)
+			{
+				if (QUEUE_BACKEND_AWAKENING(i))
+				{
+					// For awakening backend, advice a new position.
+					if ((QUEUE_POS_EQUAL(pos, queueHeadBeforeWrite) ||
+					     (QUEUE_POS_PRECEDES(queueHeadBeforeWrite, pos) && QUEUE_POS_PRECEDES(pos, queueHeadAfterWrite))) &&
+						 QUEUE_POS_EQUAL(QUEUE_BACKEND_ADVISORY_POS(i), queueHeadBeforeWrite))
+						QUEUE_BACKEND_ADVISORY_POS(i) = queueHeadAfterWrite;
+				}
+				else
+				{
+					// For non-awakening backend, directly advance its position.
+					if (QUEUE_POS_EQUAL(pos, queueHeadBeforeWrite))
+					{
+						QUEUE_BACKEND_POS(i) = queueHeadAfterWrite;
+						SET_QUEUE_POS(QUEUE_BACKEND_ADVISORY_POS(i), 0, 0);
+					}
+				}
+				pos = queueHeadAfterWrite;
+			}
 		}
 
+		if (QUEUE_BACKEND_WAKEUP_PENDING(i))
+			continue;
+
 		/* Signal backends that have fallen too far behind */
 		lag = asyncQueuePageDiff(QUEUE_POS_PAGE(QUEUE_HEAD),
 								 QUEUE_POS_PAGE(pos));
@@ -2321,6 +2361,7 @@ asyncQueueReadAllNotifications(void)
 	/* Assert checks that we have a valid state entry */
 	Assert(MyProcPid == QUEUE_BACKEND_PID(MyProcNumber));
 	QUEUE_BACKEND_WAKEUP_PENDING(MyProcNumber) = false;
+	QUEUE_BACKEND_AWAKENING(MyProcNumber) = true;
 	pos = QUEUE_BACKEND_POS(MyProcNumber);
 	head = QUEUE_HEAD;
 	LWLockRelease(NotifyQueueLock);
@@ -2330,6 +2371,9 @@ asyncQueueReadAllNotifications(void)
 	if (QUEUE_POS_EQUAL(pos, head))
 	{
 		/* Nothing to do, we have read all notifications already. */
+		LWLockAcquire(NotifyQueueLock, LW_SHARED);
+		QUEUE_BACKEND_AWAKENING(MyProcNumber) = false;
+		LWLockRelease(NotifyQueueLock);
 		return;
 	}
 
@@ -2441,21 +2485,29 @@ asyncQueueReadAllNotifications(void)
 												   page_buffer.buf,
 												   snapshot);
 
-		/*
-		 * Update our position in shared memory.  The 'pos' variable now
-		 * holds our new position (advanced past all messages we just
-		 * processed).  This ensures that if we fail while processing
-		 * messages from the next page, we won't reprocess the ones we
-		 * just handled.  It also prevents us from overwriting any direct
-		 * advancement that another backend might have done while we were
-		 * processing messages.
-		 */
-		LWLockAcquire(NotifyQueueLock, LW_SHARED);
-		QUEUE_BACKEND_POS(MyProcNumber) = pos;
-		LWLockRelease(NotifyQueueLock);
-
+		// If there is a direct advancement, let's stop reading.
+		// We don't need to lock here because even if the position
+		// changes right after we read it, we just do one more loop.
+		//LWLockAcquire(NotifyQueueLock, LW_SHARED);
+		if (QUEUE_POS_PRECEDES(pos, QUEUE_BACKEND_ADVISORY_POS(MyProcNumber)))
+		{
+			reachedStop = true;
+		}
+		//QUEUE_BACKEND_POS(MyProcNumber) = pos;
+		//LWLockRelease(NotifyQueueLock);
 	} while (!reachedStop);
 
+	LWLockAcquire(NotifyQueueLock, LW_SHARED);
+	if (QUEUE_POS_PRECEDES(pos, QUEUE_BACKEND_ADVISORY_POS(MyProcNumber)))
+	{
+		/* respect direct advancement */
+		pos = QUEUE_BACKEND_ADVISORY_POS(MyProcNumber);
+	}
+	SET_QUEUE_POS(QUEUE_BACKEND_ADVISORY_POS(MyProcNumber), 0, 0);
+	QUEUE_BACKEND_POS(MyProcNumber) = pos;
+	QUEUE_BACKEND_AWAKENING(MyProcNumber) = false;
+	LWLockRelease(NotifyQueueLock);
+
 	/* Done with snapshot */
 	UnregisterSnapshot(snapshot);
 }
diff --git a/src/test/authentication/t/008_listen-pos-race.pl b/src/test/authentication/t/008_listen-pos-race.pl
index 060e33ed391..c858e8da524 100644
--- a/src/test/authentication/t/008_listen-pos-race.pl
+++ b/src/test/authentication/t/008_listen-pos-race.pl
@@ -8,7 +8,7 @@ use PostgreSQL::Test::Utils;
 use Time::HiRes qw(usleep);
 use Test::More;
 
-if ($ENV{enable_injection_points} ne 'yes') {
+if ($ENV{enable_injection_points} // '' ne 'yes') {
     plan skip_all => 'Injection points not supported by this build';
 }
 
@@ -18,6 +18,7 @@ $node->start;
 
 
 if (!$node->check_extension('injection_points')) {
+	$node->stop;
     plan skip_all => 'Extension injection_points not installed';
 }
 

--Apple-Mail=_7B646548-F559-49EC-8DB9-78AF6DB09FAE
Content-Transfer-Encoding: 7bit
Content-Type: text/plain;
	charset=us-ascii


--Apple-Mail=_7B646548-F559-49EC-8DB9-78AF6DB09FAE--