Re: Optimize LISTEN/NOTIFY - Mailing list pgsql-hackers

Hi,

On Mon, Oct 27, 2025 at 2:25 AM Joel Jacobson <joel@compiler.org> wrote:
>
> On Sun, Oct 26, 2025, at 07:33, Joel Jacobson wrote:
> > On Sun, Oct 26, 2025, at 05:11, Chao Li wrote:
> >> 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’s advisory
> >> pos; otherwise directly advance its position.
> >> * If a running listener see current pos is behind advisory pos, then
> >> stop reading
> ...
> > This sounds promising, similar to what I had in mind. I was thinking
> > about the idea of using the advisoryPos only when the listening backend
> > is known to be running (which felt like it would need another shared
> > boolean field), and to move its pos field directly only when it's not
> > running, since if it's running we don't need to optimize for context
> > switching, since it's by definition already running.
>
> Write-up of changes since v20:
>

Thank you for working on this! There are few points about 'direct
advancement' part:

> Two new fields have been added to QueueBackendStatus:
> + QueuePosition advisoryPos;    /* safe skip-ahead position */
> + bool          advancingPos;   /* backend is reading the queue */
>
> ...
>
> In SignalBackends, if a backend that is uninterested in our
> notifications, has a shared pos that is at the old queue head, then we
> will check if it's not currently advancing its pos, in which case we can
> set its shared pos to the new queue head, i.e. "direct advance" it,
> otherwise, if it's currently advancing its pos, and if its advisory pos
> is behind our new queue head, we will update its advisory pos to our new
> queue head.
>
> In asyncQueueReadAllNotifications, we start by setting wakupPending to
> false and advisoryPos to true, to indicate that we've woken up, and that
> we will now start advancing the pos.  We also check if the pos is behind
> the advisory pos, and if so use the advisory pos to update the pos.
>
> In asyncQueueReadAllNotifications's PG_FINALLY block, we reset
> advancingPos to false, and detect if the advisoryPos was set by
> SignalBackends while we were processing messages on the queue, and if
> so, and if the advisoryPos is ahead of our pos, we update our shared pos
> with the advisoryPos, and otherwise update the shared pos with the new
> pos.

Looks like the bug with truncating of the queue is gone, advancingPos
does the trick, great.

Maybe I missed something, but I failed to find an example where we can
take advantage of advisoryPos:

    SignalBackends(void)
    ...

        if (QUEUE_POS_EQUAL(pos, queueHeadBeforeWrite))
        {
           ...
           if (!QUEUE_BACKEND_ADVANCING_POS(i))
              QUEUE_BACKEND_POS(i) = queueHeadAfterWrite;
           else if (QUEUE_POS_PRECEDES(advisoryPos, queueHeadAfterWrite))
              QUEUE_BACKEND_ADVISORY_POS(i) = queueHeadAfterWrite;
        }

We update advisoryPos if:
1) listener's advancingPos is true
2) listener's pos equals queueHeadBeforeWrite

(1) means the listener is currently reading. (2) means notifications
that the listener is currently reading belong to us (or it's even
possible that the listener is reading notifications that were added in
the queue after ours). And since the listener is reading, it will only
see updated advancingPos in the PG_FINALLY, where listener's pos will
already be >= queueHeadAfterWrite (as result of reading).


This condition seems to be redundant. I would say it should always be
true, otherwise it would mean that somebody allowed the listener to
skip our notification.

    else if (QUEUE_POS_PRECEDES(advisoryPos, queueHeadAfterWrite))


Best regards,
Arseniy Mukhin