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:
Two new fields have been added to QueueBackendStatus:
+ QueuePosition advisoryPos; /* safe skip-ahead position */
+ bool advancingPos; /* backend is reading the queue */
These are used SignalBackends and asyncQueueReadAllNotifications to
handle the empheral state of the shared queue position, since we don't
take a lock while advancing it in asyncQueueReadAllNotifications.
In SignalBackends, we now don't signal laggers in other databases,
instead we will signal any listening backend that could possibly be
behind the old queue head, since we can't know if such backend is
interested in the notifications before the old queue head. Realistic
benchmarks will be needed to determine if this happens often enough to
warrant a more complex optimization, such as the ranges idea suggested
by Arseniy Mukhin.
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.
/Joel
