Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit8c8a834

Browse files
committed
Improve LISTEN startup time when there are many unread notifications.
If some existing listener is far behind, incoming new listener sessionswould start from that session's read pointer and then need to advance overmany already-committed notification messages, which they have no interestin. This was expensive in itself and also thrashed the pg_notify SLRUbuffers a lot more than necessary. We can improve matters considerablyin typical scenarios, without much added cost, by starting from thefurthest-ahead read pointer, not the furthest-behind one. We do have toconsider only sessions in our own database when doing this, which requiresan extra field in the data structure, but that's a pretty small cost.Back-patch to 9.0 where the current LISTEN/NOTIFY logic was introduced.Matt Newell, slightly adjusted by me
1 parent91d97f0 commit8c8a834

File tree

1 file changed

+45
-5
lines changed

1 file changed

+45
-5
lines changed

‎src/backend/commands/async.c‎

Lines changed: 45 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -202,12 +202,19 @@ typedef struct QueuePosition
202202
(x).page != (y).page ? (y) : \
203203
(x).offset < (y).offset ? (x) : (y))
204204

205+
/* choose logically larger QueuePosition */
206+
#defineQUEUE_POS_MAX(x,y) \
207+
(asyncQueuePagePrecedes((x).page, (y).page) ? (y) : \
208+
(x).page != (y).page ? (x) : \
209+
(x).offset > (y).offset ? (x) : (y))
210+
205211
/*
206212
* Struct describing a listening backend's status
207213
*/
208214
typedefstructQueueBackendStatus
209215
{
210216
int32pid;/* either a PID or InvalidPid */
217+
Oiddboid;/* backend's database OID, or InvalidOid */
211218
QueuePositionpos;/* backend has read queue up to here */
212219
}QueueBackendStatus;
213220

@@ -224,6 +231,7 @@ typedef struct QueueBackendStatus
224231
* When holding the lock in EXCLUSIVE mode, backends can inspect the entries
225232
* of other backends and also change the head and tail pointers.
226233
*
234+
* AsyncCtlLock is used as the control lock for the pg_notify SLRU buffers.
227235
* In order to avoid deadlocks, whenever we need both locks, we always first
228236
* get AsyncQueueLock and then AsyncCtlLock.
229237
*
@@ -234,8 +242,8 @@ typedef struct QueueBackendStatus
234242
typedefstructAsyncQueueControl
235243
{
236244
QueuePositionhead;/* head points to the next free location */
237-
QueuePositiontail;/* the global tail is equivalent to thetail
238-
*ofthe "slowest" backend */
245+
QueuePositiontail;/* the global tail is equivalent to thepos of
246+
* the "slowest" backend */
239247
TimestampTzlastQueueFillWarn;/* time of last queue-full msg */
240248
QueueBackendStatusbackend[FLEXIBLE_ARRAY_MEMBER];
241249
/* backend[0] is not used; used entries are from [1] to [MaxBackends] */
@@ -246,6 +254,7 @@ static AsyncQueueControl *asyncQueueControl;
246254
#defineQUEUE_HEAD(asyncQueueControl->head)
247255
#defineQUEUE_TAIL(asyncQueueControl->tail)
248256
#defineQUEUE_BACKEND_PID(i)(asyncQueueControl->backend[i].pid)
257+
#defineQUEUE_BACKEND_DBOID(i)(asyncQueueControl->backend[i].dboid)
249258
#defineQUEUE_BACKEND_POS(i)(asyncQueueControl->backend[i].pos)
250259

251260
/*
@@ -459,6 +468,7 @@ AsyncShmemInit(void)
459468
for (i=0;i <=MaxBackends;i++)
460469
{
461470
QUEUE_BACKEND_PID(i)=InvalidPid;
471+
QUEUE_BACKEND_DBOID(i)=InvalidOid;
462472
SET_QUEUE_POS(QUEUE_BACKEND_POS(i),0,0);
463473
}
464474
}
@@ -905,6 +915,10 @@ AtCommit_Notify(void)
905915
staticvoid
906916
Exec_ListenPreCommit(void)
907917
{
918+
QueuePositionhead;
919+
QueuePositionmax;
920+
inti;
921+
908922
/*
909923
* Nothing to do if we are already listening to something, nor if we
910924
* already ran this routine in this transaction.
@@ -932,10 +946,34 @@ Exec_ListenPreCommit(void)
932946
* over already-committed notifications. This ensures we cannot miss any
933947
* not-yet-committed notifications. We might get a few more but that
934948
* doesn't hurt.
949+
*
950+
* In some scenarios there might be a lot of committed notifications that
951+
* have not yet been pruned away (because some backend is being lazy about
952+
* reading them). To reduce our startup time, we can look at other
953+
* backends and adopt the maximum "pos" pointer of any backend that's in
954+
* our database; any notifications it's already advanced over are surely
955+
* committed and need not be re-examined by us. (We must consider only
956+
* backends connected to our DB, because others will not have bothered to
957+
* check committed-ness of notifications in our DB.) But we only bother
958+
* with that if there's more than a page worth of notifications
959+
* outstanding, otherwise scanning all the other backends isn't worth it.
960+
*
961+
* We need exclusive lock here so we can look at other backends' entries.
935962
*/
936-
LWLockAcquire(AsyncQueueLock,LW_SHARED);
937-
QUEUE_BACKEND_POS(MyBackendId)=QUEUE_TAIL;
963+
LWLockAcquire(AsyncQueueLock,LW_EXCLUSIVE);
964+
head=QUEUE_HEAD;
965+
max=QUEUE_TAIL;
966+
if (QUEUE_POS_PAGE(max)!=QUEUE_POS_PAGE(head))
967+
{
968+
for (i=1;i <=MaxBackends;i++)
969+
{
970+
if (QUEUE_BACKEND_DBOID(i)==MyDatabaseId)
971+
max=QUEUE_POS_MAX(max,QUEUE_BACKEND_POS(i));
972+
}
973+
}
974+
QUEUE_BACKEND_POS(MyBackendId)=max;
938975
QUEUE_BACKEND_PID(MyBackendId)=MyProcPid;
976+
QUEUE_BACKEND_DBOID(MyBackendId)=MyDatabaseId;
939977
LWLockRelease(AsyncQueueLock);
940978

941979
/* Now we are listed in the global array, so remember we're listening */
@@ -951,7 +989,8 @@ Exec_ListenPreCommit(void)
951989
*
952990
* This will also advance the global tail pointer if possible.
953991
*/
954-
asyncQueueReadAllNotifications();
992+
if (!QUEUE_POS_EQUAL(max,head))
993+
asyncQueueReadAllNotifications();
955994
}
956995

957996
/*
@@ -1154,6 +1193,7 @@ asyncQueueUnregister(void)
11541193
QUEUE_POS_EQUAL(QUEUE_BACKEND_POS(MyBackendId),QUEUE_TAIL);
11551194
/* ... then mark it invalid */
11561195
QUEUE_BACKEND_PID(MyBackendId)=InvalidPid;
1196+
QUEUE_BACKEND_DBOID(MyBackendId)=InvalidOid;
11571197
LWLockRelease(AsyncQueueLock);
11581198

11591199
/* mark ourselves as no longer listed in the global array */

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp