public inbox for [email protected]
help / color / mirror / Atom feedFrom: Andres Freund <[email protected]>
Subject: [PATCH v1 3/6] Use dlist for syncrep queue.
Date: Wed, 19 Feb 2020 12:33:09 -0800
---
src/include/replication/walsender_private.h | 3 +-
src/include/storage/proc.h | 2 +-
src/backend/replication/syncrep.c | 89 +++++++++------------
src/backend/replication/walsender.c | 2 +-
src/backend/storage/lmgr/proc.c | 2 +-
5 files changed, 41 insertions(+), 57 deletions(-)
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index 366828f0a47..c2f946c5c48 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -13,6 +13,7 @@
#define _WALSENDER_PRIVATE_H
#include "access/xlog.h"
+#include "lib/ilist.h"
#include "nodes/nodes.h"
#include "replication/syncrep.h"
#include "storage/latch.h"
@@ -96,7 +97,7 @@ typedef struct
* Synchronous replication queue with one queue per request type.
* Protected by SyncRepLock.
*/
- SHM_QUEUE SyncRepQueue[NUM_SYNC_REP_WAIT_MODE];
+ dlist_head SyncRepQueue[NUM_SYNC_REP_WAIT_MODE];
/*
* Current location of the head of the queue. All waiters should have a
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 2ba37f250de..de222f96681 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -150,7 +150,7 @@ struct PGPROC
*/
XLogRecPtr waitLSN; /* waiting for this LSN or higher */
int syncRepState; /* wait state for sync rep */
- SHM_QUEUE syncRepLinks; /* list link if process is in syncrep queue */
+ dlist_node syncRepLinks; /* list link if process is in syncrep queue */
/*
* All PROCLOCK objects for locks held or awaited by this backend are
diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c
index c284103b548..402d3e41fc6 100644
--- a/src/backend/replication/syncrep.c
+++ b/src/backend/replication/syncrep.c
@@ -168,7 +168,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
if (!SyncRepRequested())
return;
- Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
+ Assert(dlist_node_is_detached(&MyProc->syncRepLinks));
Assert(WalSndCtl != NULL);
LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
@@ -304,7 +304,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
* assertions, but better safe than sorry).
*/
pg_read_barrier();
- Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
+ Assert(dlist_node_is_detached(&MyProc->syncRepLinks));
MyProc->syncRepState = SYNC_REP_NOT_WAITING;
MyProc->waitLSN = 0;
@@ -325,31 +325,32 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
static void
SyncRepQueueInsert(int mode)
{
- PGPROC *proc;
+ dlist_head *queue;
+ dlist_iter iter;
Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE);
- proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue[mode]),
- &(WalSndCtl->SyncRepQueue[mode]),
- offsetof(PGPROC, syncRepLinks));
+ queue = &WalSndCtl->SyncRepQueue[mode];
- while (proc)
+ dlist_reverse_foreach(iter, queue)
{
+ PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
+
/*
- * Stop at the queue element that we should after to ensure the queue
- * is ordered by LSN.
+ * Stop at the queue element that we should insert after to ensure the
+ * queue is ordered by LSN.
*/
if (proc->waitLSN < MyProc->waitLSN)
- break;
-
- proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue[mode]),
- &(proc->syncRepLinks),
- offsetof(PGPROC, syncRepLinks));
+ {
+ dlist_insert_after(&proc->syncRepLinks, &MyProc->syncRepLinks);
+ return;
+ }
}
- if (proc)
- SHMQueueInsertAfter(&(proc->syncRepLinks), &(MyProc->syncRepLinks));
- else
- SHMQueueInsertAfter(&(WalSndCtl->SyncRepQueue[mode]), &(MyProc->syncRepLinks));
+ /*
+ * If we get here, the list was either empty, or this process needs to be
+ * at the head.
+ */
+ dlist_push_head(queue, &MyProc->syncRepLinks);
}
/*
@@ -359,8 +360,8 @@ static void
SyncRepCancelWait(void)
{
LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
- if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
- SHMQueueDelete(&(MyProc->syncRepLinks));
+ if (!dlist_node_is_detached(&MyProc->syncRepLinks))
+ dlist_delete_thoroughly(&MyProc->syncRepLinks);
MyProc->syncRepState = SYNC_REP_NOT_WAITING;
LWLockRelease(SyncRepLock);
}
@@ -372,13 +373,13 @@ SyncRepCleanupAtProcExit(void)
* First check if we are removed from the queue without the lock to not
* slow down backend exit.
*/
- if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
+ if (!dlist_node_is_detached(&MyProc->syncRepLinks))
{
LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
/* maybe we have just been removed, so recheck */
- if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
- SHMQueueDelete(&(MyProc->syncRepLinks));
+ if (!dlist_node_is_detached(&MyProc->syncRepLinks))
+ dlist_delete_thoroughly(&MyProc->syncRepLinks);
LWLockRelease(SyncRepLock);
}
@@ -1011,20 +1012,17 @@ static int
SyncRepWakeQueue(bool all, int mode)
{
volatile WalSndCtlData *walsndctl = WalSndCtl;
- PGPROC *proc = NULL;
- PGPROC *thisproc = NULL;
int numprocs = 0;
+ dlist_mutable_iter iter;
Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE);
Assert(LWLockHeldByMeInMode(SyncRepLock, LW_EXCLUSIVE));
Assert(SyncRepQueueIsOrderedByLSN(mode));
- proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
- &(WalSndCtl->SyncRepQueue[mode]),
- offsetof(PGPROC, syncRepLinks));
-
- while (proc)
+ dlist_foreach_modify(iter, &WalSndCtl->SyncRepQueue[mode])
{
+ PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
+
/*
* Assume the queue is ordered by LSN
*/
@@ -1032,18 +1030,9 @@ SyncRepWakeQueue(bool all, int mode)
return numprocs;
/*
- * Move to next proc, so we can delete thisproc from the queue.
- * thisproc is valid, proc may be NULL after this.
+ * Remove from queue.
*/
- thisproc = proc;
- proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
- &(proc->syncRepLinks),
- offsetof(PGPROC, syncRepLinks));
-
- /*
- * Remove thisproc from queue.
- */
- SHMQueueDelete(&(thisproc->syncRepLinks));
+ dlist_delete_thoroughly(&proc->syncRepLinks);
/*
* SyncRepWaitForLSN() reads syncRepState without holding the lock, so
@@ -1056,12 +1045,12 @@ SyncRepWakeQueue(bool all, int mode)
* Set state to complete; see SyncRepWaitForLSN() for discussion of
* the various states.
*/
- thisproc->syncRepState = SYNC_REP_WAIT_COMPLETE;
+ proc->syncRepState = SYNC_REP_WAIT_COMPLETE;
/*
* Wake only when we have set state and removed from queue.
*/
- SetLatch(&(thisproc->procLatch));
+ SetLatch(&(proc->procLatch));
numprocs++;
}
@@ -1115,19 +1104,17 @@ SyncRepUpdateSyncStandbysDefined(void)
static bool
SyncRepQueueIsOrderedByLSN(int mode)
{
- PGPROC *proc = NULL;
XLogRecPtr lastLSN;
+ dlist_iter iter;
Assert(mode >= 0 && mode < NUM_SYNC_REP_WAIT_MODE);
lastLSN = 0;
- proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
- &(WalSndCtl->SyncRepQueue[mode]),
- offsetof(PGPROC, syncRepLinks));
-
- while (proc)
+ dlist_foreach(iter, &WalSndCtl->SyncRepQueue[mode])
{
+ PGPROC *proc = dlist_container(PGPROC, syncRepLinks, iter.cur);
+
/*
* Check the queue is ordered by LSN and that multiple procs don't
* have matching LSNs
@@ -1136,10 +1123,6 @@ SyncRepQueueIsOrderedByLSN(int mode)
return false;
lastLSN = proc->waitLSN;
-
- proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
- &(proc->syncRepLinks),
- offsetof(PGPROC, syncRepLinks));
}
return true;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index abb533b9d03..2d2a0ceb503 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -3021,7 +3021,7 @@ WalSndShmemInit(void)
MemSet(WalSndCtl, 0, WalSndShmemSize());
for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
- SHMQueueInit(&(WalSndCtl->SyncRepQueue[i]));
+ dlist_init(&(WalSndCtl->SyncRepQueue[i]));
for (i = 0; i < max_wal_senders; i++)
{
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 5a157fc07d8..a4c338d7aff 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -415,7 +415,7 @@ InitProcess(void)
/* Initialize fields for sync rep */
MyProc->waitLSN = 0;
MyProc->syncRepState = SYNC_REP_NOT_WAITING;
- SHMQueueElemInit(&(MyProc->syncRepLinks));
+ dlist_node_init(&MyProc->syncRepLinks);
/* Initialize fields for group XID clearing. */
MyProc->procArrayGroupMember = false;
--
2.25.0.114.g5b0ca878e0
--jukqbc7dv5ymsafw
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="v1-0004-Use-dlists-for-predicate-locking.patch"
view thread (7+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH v1 3/6] Use dlist for syncrep queue.
In-Reply-To: <no-message-id-1880854@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox