From 53f0dc9b79aa072e2b49d41be1dc63f3e838aa2b Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Wed, 19 Feb 2020 12:23:28 -0800 Subject: [PATCH v1 2/6] Use dlists instead of SHM_QUEUE for heavyweight locks. Todo: - Consider using singly linked list? - Consider removing PROC_QUEUE - the size doesn't seem important. --- src/include/storage/lock.h | 9 +- src/include/storage/proc.h | 16 +-- src/backend/access/transam/twophase.c | 4 +- src/backend/storage/lmgr/deadlock.c | 51 +++---- src/backend/storage/lmgr/lock.c | 121 ++++++---------- src/backend/storage/lmgr/proc.c | 192 +++++++++++--------------- 6 files changed, 161 insertions(+), 232 deletions(-) diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h index bb8e4e6e5b7..3569f145092 100644 --- a/src/include/storage/lock.h +++ b/src/include/storage/lock.h @@ -18,6 +18,7 @@ #error "lock.h may not be included from frontend code" #endif +#include "lib/ilist.h" #include "storage/backendid.h" #include "storage/lockdefs.h" #include "storage/lwlock.h" @@ -28,7 +29,7 @@ typedef struct PGPROC PGPROC; typedef struct PROC_QUEUE { - SHM_QUEUE links; /* head of list of PGPROC objects */ + dlist_head links; /* list of PGPROC objects */ int size; /* number of entries in list */ } PROC_QUEUE; @@ -292,7 +293,7 @@ typedef struct LOCK /* data */ LOCKMASK grantMask; /* bitmask for lock types already granted */ LOCKMASK waitMask; /* bitmask for lock types awaited */ - SHM_QUEUE procLocks; /* list of PROCLOCK objects assoc. with lock */ + dlist_head procLocks; /* list of PROCLOCK objects assoc. with lock */ PROC_QUEUE waitProcs; /* list of PGPROC objects waiting on lock */ int requested[MAX_LOCKMODES]; /* counts of requested locks */ int nRequested; /* total of requested[] array */ @@ -353,8 +354,8 @@ typedef struct PROCLOCK PGPROC *groupLeader; /* proc's lock group leader, or proc itself */ LOCKMASK holdMask; /* bitmask for lock types currently held */ LOCKMASK releaseMask; /* bitmask for lock types to be released */ - SHM_QUEUE lockLink; /* list link in LOCK's list of proclocks */ - SHM_QUEUE procLink; /* list link in PGPROC's list of proclocks */ + dlist_node lockLink; /* list link in LOCK's list of proclocks */ + dlist_node procLink; /* list link in PGPROC's list of proclocks */ } PROCLOCK; #define PROCLOCK_LOCKMETHOD(proclock) \ diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index d21780108bb..2ba37f250de 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -95,8 +95,8 @@ struct XidCache struct PGPROC { /* proc->links MUST BE FIRST IN STRUCT (see ProcSleep,ProcWakeup,etc) */ - SHM_QUEUE links; /* list link if process is in a list */ - PGPROC **procgloballist; /* procglobal list that owns this PGPROC */ + dlist_node links; /* list link if process is in a list */ + dlist_head *procgloballist; /* procglobal list that owns this PGPROC */ PGSemaphore sem; /* ONE semaphore to sleep on */ int waitStatus; /* STATUS_WAITING, STATUS_OK or STATUS_ERROR */ @@ -157,7 +157,7 @@ struct PGPROC * linked into one of these lists, according to the partition number of * their lock. */ - SHM_QUEUE myProcLocks[NUM_LOCK_PARTITIONS]; + dlist_head myProcLocks[NUM_LOCK_PARTITIONS]; struct XidCache subxids; /* cache for subtransaction XIDs */ @@ -250,13 +250,13 @@ typedef struct PROC_HDR /* Length of allProcs array */ uint32 allProcCount; /* Head of list of free PGPROC structures */ - PGPROC *freeProcs; + dlist_head freeProcs; /* Head of list of autovacuum's free PGPROC structures */ - PGPROC *autovacFreeProcs; + dlist_head autovacFreeProcs; /* Head of list of bgworker free PGPROC structures */ - PGPROC *bgworkerFreeProcs; + dlist_head bgworkerFreeProcs; /* Head of list of walsender free PGPROC structures */ - PGPROC *walsenderFreeProcs; + dlist_head walsenderFreeProcs; /* First pgproc waiting for group XID clear */ pg_atomic_uint32 procArrayGroupFirst; /* First pgproc waiting for group transaction status update */ @@ -318,7 +318,7 @@ extern void ProcReleaseLocks(bool isCommit); extern void ProcQueueInit(PROC_QUEUE *queue); extern int ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable); -extern PGPROC *ProcWakeup(PGPROC *proc, int waitStatus); +extern void ProcWakeup(PGPROC *proc, int waitStatus); extern void ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock); extern void CheckDeadLockAlert(void); extern bool IsWaitingForLock(void); diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 5adf956f413..39d7e5463c1 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -459,7 +459,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid, /* Initialize the PGPROC entry */ MemSet(proc, 0, sizeof(PGPROC)); proc->pgprocno = gxact->pgprocno; - SHMQueueElemInit(&(proc->links)); + dlist_node_init(&proc->links); proc->waitStatus = STATUS_OK; /* We set up the gxact's VXID as InvalidBackendId/XID */ proc->lxid = (LocalTransactionId) xid; @@ -478,7 +478,7 @@ MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid, const char *gid, proc->waitLock = NULL; proc->waitProcLock = NULL; for (i = 0; i < NUM_LOCK_PARTITIONS; i++) - SHMQueueInit(&(proc->myProcLocks[i])); + dlist_init(&proc->myProcLocks[i]); /* subxid data must be filled later by GXactLoadSubxactData */ pgxact->overflowed = false; pgxact->nxids = 0; diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c index f8c5df08e69..ca2abea07f1 100644 --- a/src/backend/storage/lmgr/deadlock.c +++ b/src/backend/storage/lmgr/deadlock.c @@ -216,9 +216,6 @@ InitDeadLockChecking(void) DeadLockState DeadLockCheck(PGPROC *proc) { - int i, - j; - /* Initialize to "no constraints" */ nCurConstraints = 0; nPossibleConstraints = 0; @@ -246,7 +243,7 @@ DeadLockCheck(PGPROC *proc) } /* Apply any needed rearrangements of wait queues */ - for (i = 0; i < nWaitOrders; i++) + for (int i = 0; i < nWaitOrders; i++) { LOCK *lock = waitOrders[i].lock; PGPROC **procs = waitOrders[i].procs; @@ -261,9 +258,9 @@ DeadLockCheck(PGPROC *proc) /* Reset the queue and re-add procs in the desired order */ ProcQueueInit(waitQueue); - for (j = 0; j < nProcs; j++) + for (int j = 0; j < nProcs; j++) { - SHMQueueInsertBefore(&(waitQueue->links), &(procs[j]->links)); + dlist_push_tail(&waitQueue->links, &(procs[j]->links)); waitQueue->size++; } @@ -545,8 +542,7 @@ FindLockCycleRecurseMember(PGPROC *checkProc, PGPROC *proc; LOCK *lock = checkProc->waitLock; PGXACT *pgxact; - PROCLOCK *proclock; - SHM_QUEUE *procLocks; + dlist_iter proclock_iter; LockMethod lockMethodTable; PROC_QUEUE *waitQueue; int queue_size; @@ -563,13 +559,9 @@ FindLockCycleRecurseMember(PGPROC *checkProc, * Scan for procs that already hold conflicting locks. These are "hard" * edges in the waits-for graph. */ - procLocks = &(lock->procLocks); - - proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, lockLink)); - - while (proclock) + dlist_foreach(proclock_iter, &lock->procLocks) { + PROCLOCK *proclock = dlist_container(PROCLOCK, lockLink, proclock_iter.cur); PGPROC *leader; proc = proclock->tag.myProc; @@ -629,9 +621,6 @@ FindLockCycleRecurseMember(PGPROC *checkProc, } } } - - proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink, - offsetof(PROCLOCK, lockLink)); } /* @@ -704,6 +693,7 @@ FindLockCycleRecurseMember(PGPROC *checkProc, else { PGPROC *lastGroupMember = NULL; + dlist_iter proc_iter; /* Use the true lock wait queue order */ waitQueue = &(lock->waitProcs); @@ -719,13 +709,14 @@ FindLockCycleRecurseMember(PGPROC *checkProc, lastGroupMember = checkProc; else { - proc = (PGPROC *) waitQueue->links.next; - queue_size = waitQueue->size; - while (queue_size-- > 0) + dlist_iter iter; + + dlist_foreach(iter, &waitQueue->links) { + proc = dlist_container(PGPROC, links, iter.cur); + if (proc->lockGroupLeader == checkProcLeader) lastGroupMember = proc; - proc = (PGPROC *) proc->links.next; } Assert(lastGroupMember != NULL); } @@ -733,12 +724,12 @@ FindLockCycleRecurseMember(PGPROC *checkProc, /* * OK, now rescan (or scan) the queue to identify the soft conflicts. */ - queue_size = waitQueue->size; - proc = (PGPROC *) waitQueue->links.next; - while (queue_size-- > 0) + dlist_foreach(proc_iter, &waitQueue->links) { PGPROC *leader; + proc = dlist_container(PGPROC, links, proc_iter.cur); + leader = proc->lockGroupLeader == NULL ? proc : proc->lockGroupLeader; @@ -772,8 +763,6 @@ FindLockCycleRecurseMember(PGPROC *checkProc, return true; } } - - proc = (PGPROC *) proc->links.next; } } @@ -882,14 +871,16 @@ TopoSort(LOCK *lock, k, kk, last; + dlist_iter proc_iter; /* First, fill topoProcs[] array with the procs in their current order */ - proc = (PGPROC *) waitQueue->links.next; - for (i = 0; i < queue_size; i++) + i = 0; + dlist_foreach(proc_iter, &waitQueue->links) { - topoProcs[i] = proc; - proc = (PGPROC *) proc->links.next; + proc = dlist_container(PGPROC, links, proc_iter.cur); + topoProcs[i++] = proc; } + Assert(i == queue_size); /* * Scan the constraints, and for each proc in the array, generate a count diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c index 56dba09299d..01ac3c06c5e 100644 --- a/src/backend/storage/lmgr/lock.c +++ b/src/backend/storage/lmgr/lock.c @@ -1005,8 +1005,8 @@ LockAcquireExtended(const LOCKTAG *locktag, uint32 proclock_hashcode; proclock_hashcode = ProcLockHashCode(&proclock->tag, hashcode); - SHMQueueDelete(&proclock->lockLink); - SHMQueueDelete(&proclock->procLink); + dlist_delete(&proclock->lockLink); + dlist_delete(&proclock->procLink); if (!hash_search_with_hash_value(LockMethodProcLockHash, (void *) &(proclock->tag), proclock_hashcode, @@ -1141,7 +1141,7 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc, { lock->grantMask = 0; lock->waitMask = 0; - SHMQueueInit(&(lock->procLocks)); + dlist_init(&lock->procLocks); ProcQueueInit(&(lock->waitProcs)); lock->nRequested = 0; lock->nGranted = 0; @@ -1184,7 +1184,7 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc, * of shared memory, because there won't be anything to cause * anyone to release the lock object later. */ - Assert(SHMQueueEmpty(&(lock->procLocks))); + Assert(dlist_is_empty(&(lock->procLocks))); if (!hash_search_with_hash_value(LockMethodLockHash, (void *) &(lock->tag), hashcode, @@ -1217,9 +1217,8 @@ SetupLockInTable(LockMethod lockMethodTable, PGPROC *proc, proclock->holdMask = 0; proclock->releaseMask = 0; /* Add proclock to appropriate lists */ - SHMQueueInsertBefore(&lock->procLocks, &proclock->lockLink); - SHMQueueInsertBefore(&(proc->myProcLocks[partition]), - &proclock->procLink); + dlist_push_tail(&lock->procLocks, &proclock->lockLink); + dlist_push_tail(&proc->myProcLocks[partition], &proclock->procLink); PROCLOCK_PRINT("LockAcquire: new", proclock); } else @@ -1349,9 +1348,8 @@ LockCheckConflicts(LockMethod lockMethodTable, int conflictMask = lockMethodTable->conflictTab[lockmode]; int conflictsRemaining[MAX_LOCKMODES]; int totalConflictsRemaining = 0; + dlist_iter proclock_iter; int i; - SHM_QUEUE *procLocks; - PROCLOCK *otherproclock; /* * first check for global conflicts: If no locks conflict with my request, @@ -1411,11 +1409,11 @@ LockCheckConflicts(LockMethod lockMethodTable, * shared memory state more complex (and larger) but it doesn't seem worth * it. */ - procLocks = &(lock->procLocks); - otherproclock = (PROCLOCK *) - SHMQueueNext(procLocks, procLocks, offsetof(PROCLOCK, lockLink)); - while (otherproclock != NULL) + dlist_foreach(proclock_iter, &lock->procLocks) { + PROCLOCK *otherproclock = + dlist_container(PROCLOCK, lockLink, proclock_iter.cur); + if (proclock != otherproclock && proclock->groupLeader == otherproclock->groupLeader && (otherproclock->holdMask & conflictMask) != 0) @@ -1440,9 +1438,6 @@ LockCheckConflicts(LockMethod lockMethodTable, return false; } } - otherproclock = (PROCLOCK *) - SHMQueueNext(procLocks, &otherproclock->lockLink, - offsetof(PROCLOCK, lockLink)); } /* Nope, it's a real conflict. */ @@ -1555,8 +1550,8 @@ CleanUpLock(LOCK *lock, PROCLOCK *proclock, uint32 proclock_hashcode; PROCLOCK_PRINT("CleanUpLock: deleting", proclock); - SHMQueueDelete(&proclock->lockLink); - SHMQueueDelete(&proclock->procLink); + dlist_delete(&proclock->lockLink); + dlist_delete(&proclock->procLink); proclock_hashcode = ProcLockHashCode(&proclock->tag, hashcode); if (!hash_search_with_hash_value(LockMethodProcLockHash, (void *) &(proclock->tag), @@ -1573,7 +1568,7 @@ CleanUpLock(LOCK *lock, PROCLOCK *proclock, * object. */ LOCK_PRINT("CleanUpLock: deleting", lock, 0); - Assert(SHMQueueEmpty(&(lock->procLocks))); + Assert(dlist_is_empty(&lock->procLocks)); if (!hash_search_with_hash_value(LockMethodLockHash, (void *) &(lock->tag), hashcode, @@ -1837,7 +1832,7 @@ RemoveFromWaitQueue(PGPROC *proc, uint32 hashcode) Assert(0 < lockmethodid && lockmethodid < lengthof(LockMethods)); /* Remove proc from lock's wait queue */ - SHMQueueDelete(&(proc->links)); + dlist_delete(&proc->links); waitLock->waitProcs.size--; /* Undo increments of request counts by waiting process */ @@ -2092,7 +2087,6 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks) numLockModes; LOCALLOCK *locallock; LOCK *lock; - PROCLOCK *proclock; int partition; bool have_fast_path_lwlock = false; @@ -2249,8 +2243,8 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks) for (partition = 0; partition < NUM_LOCK_PARTITIONS; partition++) { LWLock *partitionLock; - SHM_QUEUE *procLocks = &(MyProc->myProcLocks[partition]); - PROCLOCK *nextplock; + dlist_head *procLocks = &MyProc->myProcLocks[partition]; + dlist_mutable_iter proclock_iter; partitionLock = LockHashPartitionLockByIndex(partition); @@ -2273,24 +2267,16 @@ LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks) * locallock situation, we lose that guarantee for fast-path locks. * This is not ideal. */ - if (SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, procLink)) == NULL) + if (dlist_is_empty(procLocks)) continue; /* needn't examine this partition */ LWLockAcquire(partitionLock, LW_EXCLUSIVE); - for (proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, procLink)); - proclock; - proclock = nextplock) + dlist_foreach_modify(proclock_iter, procLocks) { + PROCLOCK *proclock = dlist_container(PROCLOCK, procLink, proclock_iter.cur); bool wakeupNeeded = false; - /* Get link first, since we may unlink/delete this proclock */ - nextplock = (PROCLOCK *) - SHMQueueNext(procLocks, &proclock->procLink, - offsetof(PROCLOCK, procLink)); - Assert(proclock->tag.myProc == MyProc); lock = proclock->tag.myLock; @@ -2823,7 +2809,7 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp) LockMethod lockMethodTable; LOCK *lock; LOCKMASK conflictMask; - SHM_QUEUE *procLocks; + dlist_iter proclock_iter; PROCLOCK *proclock; uint32 hashcode; LWLock *partitionLock; @@ -2971,14 +2957,10 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp) /* * Examine each existing holder (or awaiter) of the lock. */ - - procLocks = &(lock->procLocks); - - proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, lockLink)); - - while (proclock) + dlist_foreach(proclock_iter, &lock->procLocks) { + proclock = dlist_container(PROCLOCK, lockLink, proclock_iter.cur); + if (conflictMask & proclock->holdMask) { PGPROC *proc = proclock->tag.myProc; @@ -3008,9 +2990,6 @@ GetLockConflicts(const LOCKTAG *locktag, LOCKMODE lockmode, int *countp) } } } - - proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink, - offsetof(PROCLOCK, lockLink)); } LWLockRelease(partitionLock); @@ -3328,8 +3307,8 @@ PostPrepare_Locks(TransactionId xid) for (partition = 0; partition < NUM_LOCK_PARTITIONS; partition++) { LWLock *partitionLock; - SHM_QUEUE *procLocks = &(MyProc->myProcLocks[partition]); - PROCLOCK *nextplock; + dlist_head *procLocks = &(MyProc->myProcLocks[partition]); + dlist_mutable_iter proclock_iter; partitionLock = LockHashPartitionLockByIndex(partition); @@ -3341,21 +3320,14 @@ PostPrepare_Locks(TransactionId xid) * another backend is adding something to our lists now. For safety, * though, we code this the same way as in LockReleaseAll. */ - if (SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, procLink)) == NULL) + if (dlist_is_empty(procLocks)) continue; /* needn't examine this partition */ LWLockAcquire(partitionLock, LW_EXCLUSIVE); - for (proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, procLink)); - proclock; - proclock = nextplock) + dlist_foreach_modify(proclock_iter, procLocks) { - /* Get link first, since we may unlink/relink this proclock */ - nextplock = (PROCLOCK *) - SHMQueueNext(procLocks, &proclock->procLink, - offsetof(PROCLOCK, procLink)); + proclock = dlist_container(PROCLOCK, procLink, proclock_iter.cur); Assert(proclock->tag.myProc == MyProc); @@ -3393,7 +3365,7 @@ PostPrepare_Locks(TransactionId xid) * same hash partition, cf proclock_hash(). So the partition lock * we already hold is sufficient for this. */ - SHMQueueDelete(&proclock->procLink); + dlist_delete(&proclock->procLink); /* * Create the new hash key for the proclock. @@ -3419,8 +3391,7 @@ PostPrepare_Locks(TransactionId xid) elog(PANIC, "duplicate entry found while reassigning a prepared transaction's locks"); /* Re-link into the new proc's proclock list */ - SHMQueueInsertBefore(&(newproc->myProcLocks[partition]), - &proclock->procLink); + dlist_push_tail(&newproc->myProcLocks[partition], &proclock->procLink); PROCLOCK_PRINT("PostPrepare_Locks: updated", proclock); } /* loop over PROCLOCKs within this partition */ @@ -3741,12 +3712,10 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data) { LOCK *theLock = blocked_proc->waitLock; BlockedProcData *bproc; - SHM_QUEUE *procLocks; - PROCLOCK *proclock; + dlist_iter proclock_iter; + dlist_iter proc_iter; PROC_QUEUE *waitQueue; - PGPROC *proc; int queue_size; - int i; /* Nothing to do if this proc is not blocked */ if (theLock == NULL) @@ -3764,11 +3733,10 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data) */ /* Collect all PROCLOCKs associated with theLock */ - procLocks = &(theLock->procLocks); - proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, lockLink)); - while (proclock) + dlist_foreach(proclock_iter, &theLock->procLocks) { + PROCLOCK *proclock = + dlist_container(PROCLOCK, lockLink, proclock_iter.cur); PGPROC *proc = proclock->tag.myProc; LOCK *lock = proclock->tag.myLock; LockInstanceData *instance; @@ -3793,9 +3761,6 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data) instance->leaderPid = proclock->groupLeader->pid; instance->fastpath = false; data->nlocks++; - - proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink, - offsetof(PROCLOCK, lockLink)); } /* Enlarge waiter_pids[] if it's too small to hold all wait queue PIDs */ @@ -3811,9 +3776,9 @@ GetSingleProcBlockerStatusData(PGPROC *blocked_proc, BlockedProcsData *data) } /* Collect PIDs from the lock's wait queue, stopping at blocked_proc */ - proc = (PGPROC *) waitQueue->links.next; - for (i = 0; i < queue_size; i++) + dlist_foreach(proc_iter, &waitQueue->links) { + PGPROC *proc = dlist_container(PGPROC, links, proc_iter.cur); if (proc == blocked_proc) break; data->waiter_pids[data->npids++] = proc->pid; @@ -4090,7 +4055,7 @@ lock_twophase_recover(TransactionId xid, uint16 info, { lock->grantMask = 0; lock->waitMask = 0; - SHMQueueInit(&(lock->procLocks)); + dlist_init(&lock->procLocks); ProcQueueInit(&(lock->waitProcs)); lock->nRequested = 0; lock->nGranted = 0; @@ -4133,7 +4098,7 @@ lock_twophase_recover(TransactionId xid, uint16 info, * of shared memory, because there won't be anything to cause * anyone to release the lock object later. */ - Assert(SHMQueueEmpty(&(lock->procLocks))); + Assert(dlist_is_empty(&lock->procLocks)); if (!hash_search_with_hash_value(LockMethodLockHash, (void *) &(lock->tag), hashcode, @@ -4158,9 +4123,9 @@ lock_twophase_recover(TransactionId xid, uint16 info, proclock->holdMask = 0; proclock->releaseMask = 0; /* Add proclock to appropriate lists */ - SHMQueueInsertBefore(&lock->procLocks, &proclock->lockLink); - SHMQueueInsertBefore(&(proc->myProcLocks[partition]), - &proclock->procLink); + dlist_push_tail(&lock->procLocks, &proclock->lockLink); + dlist_push_tail(&proc->myProcLocks[partition], + &proclock->procLink); PROCLOCK_PRINT("lock_twophase_recover: new", proclock); } else diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index eb321f72ea4..5a157fc07d8 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -178,10 +178,10 @@ InitProcGlobal(void) * Initialize the data structures. */ ProcGlobal->spins_per_delay = DEFAULT_SPINS_PER_DELAY; - ProcGlobal->freeProcs = NULL; - ProcGlobal->autovacFreeProcs = NULL; - ProcGlobal->bgworkerFreeProcs = NULL; - ProcGlobal->walsenderFreeProcs = NULL; + dlist_init(&ProcGlobal->freeProcs); + dlist_init(&ProcGlobal->autovacFreeProcs); + dlist_init(&ProcGlobal->bgworkerFreeProcs); + dlist_init(&ProcGlobal->walsenderFreeProcs); ProcGlobal->startupProc = NULL; ProcGlobal->startupProcPid = 0; ProcGlobal->startupBufferPinWaitBufId = -1; @@ -218,6 +218,8 @@ InitProcGlobal(void) for (i = 0; i < TotalProcs; i++) { + PGPROC *proc = &procs[i]; + /* Common initialization for all PGPROCs, regardless of type. */ /* @@ -227,11 +229,11 @@ InitProcGlobal(void) */ if (i < MaxBackends + NUM_AUXILIARY_PROCS) { - procs[i].sem = PGSemaphoreCreate(); - InitSharedLatch(&(procs[i].procLatch)); - LWLockInitialize(&(procs[i].backendLock), LWTRANCHE_PROC); + proc->sem = PGSemaphoreCreate(); + InitSharedLatch(&(proc->procLatch)); + LWLockInitialize(&(proc->backendLock), LWTRANCHE_PROC); } - procs[i].pgprocno = i; + proc->pgprocno = i; /* * Newly created PGPROCs for normal backends, autovacuum and bgworkers @@ -244,45 +246,41 @@ InitProcGlobal(void) if (i < MaxConnections) { /* PGPROC for normal backend, add to freeProcs list */ - procs[i].links.next = (SHM_QUEUE *) ProcGlobal->freeProcs; - ProcGlobal->freeProcs = &procs[i]; - procs[i].procgloballist = &ProcGlobal->freeProcs; + dlist_push_head(&ProcGlobal->freeProcs, &proc->links); + proc->procgloballist = &ProcGlobal->freeProcs; } else if (i < MaxConnections + autovacuum_max_workers + 1) { /* PGPROC for AV launcher/worker, add to autovacFreeProcs list */ - procs[i].links.next = (SHM_QUEUE *) ProcGlobal->autovacFreeProcs; - ProcGlobal->autovacFreeProcs = &procs[i]; - procs[i].procgloballist = &ProcGlobal->autovacFreeProcs; + dlist_push_head(&ProcGlobal->autovacFreeProcs, &proc->links); + proc->procgloballist = &ProcGlobal->autovacFreeProcs; } else if (i < MaxConnections + autovacuum_max_workers + 1 + max_worker_processes) { /* PGPROC for bgworker, add to bgworkerFreeProcs list */ - procs[i].links.next = (SHM_QUEUE *) ProcGlobal->bgworkerFreeProcs; - ProcGlobal->bgworkerFreeProcs = &procs[i]; - procs[i].procgloballist = &ProcGlobal->bgworkerFreeProcs; + dlist_push_head(&ProcGlobal->bgworkerFreeProcs, &proc->links); + proc->procgloballist = &ProcGlobal->bgworkerFreeProcs; } else if (i < MaxBackends) { /* PGPROC for walsender, add to walsenderFreeProcs list */ - procs[i].links.next = (SHM_QUEUE *) ProcGlobal->walsenderFreeProcs; - ProcGlobal->walsenderFreeProcs = &procs[i]; - procs[i].procgloballist = &ProcGlobal->walsenderFreeProcs; + dlist_push_head(&ProcGlobal->walsenderFreeProcs, &proc->links); + proc->procgloballist = &ProcGlobal->walsenderFreeProcs; } /* Initialize myProcLocks[] shared memory queues. */ for (j = 0; j < NUM_LOCK_PARTITIONS; j++) - SHMQueueInit(&(procs[i].myProcLocks[j])); + dlist_init(&(proc->myProcLocks[j])); /* Initialize lockGroupMembers list. */ - dlist_init(&procs[i].lockGroupMembers); + dlist_init(&proc->lockGroupMembers); /* * Initialize the atomic variables, otherwise, it won't be safe to * access them for backends that aren't currently in use. */ - pg_atomic_init_u32(&(procs[i].procArrayGroupNext), INVALID_PGPROCNO); - pg_atomic_init_u32(&(procs[i].clogGroupNext), INVALID_PGPROCNO); + pg_atomic_init_u32(&(proc->procArrayGroupNext), INVALID_PGPROCNO); + pg_atomic_init_u32(&(proc->clogGroupNext), INVALID_PGPROCNO); } /* @@ -303,7 +301,7 @@ InitProcGlobal(void) void InitProcess(void) { - PGPROC *volatile *procgloballist; + dlist_head *procgloballist; /* * ProcGlobal should be set up already (if we are a backend, we inherit @@ -336,11 +334,9 @@ InitProcess(void) set_spins_per_delay(ProcGlobal->spins_per_delay); - MyProc = *procgloballist; - - if (MyProc != NULL) + if (!dlist_is_empty(procgloballist)) { - *procgloballist = (PGPROC *) MyProc->links.next; + MyProc = (PGPROC*) dlist_pop_head_node(procgloballist); SpinLockRelease(ProcStructLock); } else @@ -382,7 +378,7 @@ InitProcess(void) * Initialize all fields of MyProc, except for those previously * initialized by InitProcGlobal. */ - SHMQueueElemInit(&(MyProc->links)); + dlist_node_init(&MyProc->links); MyProc->waitStatus = STATUS_OK; MyProc->lxid = InvalidLocalTransactionId; MyProc->fpVXIDLock = false; @@ -411,7 +407,7 @@ InitProcess(void) /* Last process should have released all locks. */ for (i = 0; i < NUM_LOCK_PARTITIONS; i++) - Assert(SHMQueueEmpty(&(MyProc->myProcLocks[i]))); + Assert(dlist_is_empty(&(MyProc->myProcLocks[i]))); } #endif MyProc->recoveryConflictPending = false; @@ -566,7 +562,7 @@ InitAuxiliaryProcess(void) * Initialize all fields of MyProc, except for those previously * initialized by InitProcGlobal. */ - SHMQueueElemInit(&(MyProc->links)); + dlist_node_init(&MyProc->links); MyProc->waitStatus = STATUS_OK; MyProc->lxid = InvalidLocalTransactionId; MyProc->fpVXIDLock = false; @@ -590,7 +586,7 @@ InitAuxiliaryProcess(void) /* Last process should have released all locks. */ for (i = 0; i < NUM_LOCK_PARTITIONS; i++) - Assert(SHMQueueEmpty(&(MyProc->myProcLocks[i]))); + Assert(dlist_is_empty(&(MyProc->myProcLocks[i]))); } #endif @@ -670,16 +666,15 @@ GetStartupBufferPinWaitBufId(void) bool HaveNFreeProcs(int n) { - PGPROC *proc; + dlist_iter iter; SpinLockAcquire(ProcStructLock); - proc = ProcGlobal->freeProcs; - - while (n > 0 && proc != NULL) + dlist_foreach(iter, &ProcGlobal->freeProcs) { - proc = (PGPROC *) proc->links.next; n--; + if (n == 0) + break; } SpinLockRelease(ProcStructLock); @@ -742,7 +737,7 @@ LockErrorCleanup(void) partitionLock = LockHashPartitionLock(lockAwaited->hashcode); LWLockAcquire(partitionLock, LW_EXCLUSIVE); - if (MyProc->links.next != NULL) + if (!dlist_node_is_detached(&MyProc->links)) { /* We could not have been granted the lock yet */ RemoveFromWaitQueue(MyProc, lockAwaited->hashcode); @@ -815,7 +810,7 @@ static void ProcKill(int code, Datum arg) { PGPROC *proc; - PGPROC *volatile *procgloballist; + dlist_head *procgloballist; Assert(MyProc != NULL); @@ -828,7 +823,7 @@ ProcKill(int code, Datum arg) /* Last process should have released all locks. */ for (i = 0; i < NUM_LOCK_PARTITIONS; i++) - Assert(SHMQueueEmpty(&(MyProc->myProcLocks[i]))); + Assert(dlist_is_empty(&(MyProc->myProcLocks[i]))); } #endif @@ -851,7 +846,7 @@ ProcKill(int code, Datum arg) /* * Detach from any lock group of which we are a member. If the leader - * exist before all other group members, its PGPROC will remain allocated + * exits before all other group members, its PGPROC will remain allocated * until the last group process exits; that process must return the * leader's PGPROC to the appropriate list. */ @@ -872,8 +867,7 @@ ProcKill(int code, Datum arg) /* Leader exited first; return its PGPROC. */ SpinLockAcquire(ProcStructLock); - leader->links.next = (SHM_QUEUE *) *procgloballist; - *procgloballist = leader; + dlist_push_head(procgloballist, &leader->links); SpinLockRelease(ProcStructLock); } } @@ -907,8 +901,7 @@ ProcKill(int code, Datum arg) Assert(dlist_is_empty(&proc->lockGroupMembers)); /* Return PGPROC structure (and semaphore) to appropriate freelist */ - proc->links.next = (SHM_QUEUE *) *procgloballist; - *procgloballist = proc; + dlist_push_tail(procgloballist, &proc->links); } /* Update shared estimate of spins_per_delay */ @@ -1037,7 +1030,7 @@ ProcQueueAlloc(const char *name) void ProcQueueInit(PROC_QUEUE *queue) { - SHMQueueInit(&(queue->links)); + dlist_init(&queue->links); queue->size = 0; } @@ -1068,12 +1061,11 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) LWLock *partitionLock = LockHashPartitionLock(hashcode); PROC_QUEUE *waitQueue = &(lock->waitProcs); LOCKMASK myHeldLocks = MyProc->heldLocks; + PGPROC *insert_before = NULL; bool early_deadlock = false; bool allow_autovacuum_cancel = true; int myWaitStatus; - PGPROC *proc; PGPROC *leader = MyProc->lockGroupLeader; - int i; /* * If group locking is in use, locks held by members of my locking group @@ -1081,18 +1073,16 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) */ if (leader != NULL) { - SHM_QUEUE *procLocks = &(lock->procLocks); - PROCLOCK *otherproclock; + dlist_iter iter; - otherproclock = (PROCLOCK *) - SHMQueueNext(procLocks, procLocks, offsetof(PROCLOCK, lockLink)); - while (otherproclock != NULL) + dlist_foreach(iter, &lock->procLocks) { + PROCLOCK *otherproclock; + + otherproclock = dlist_container(PROCLOCK, lockLink, iter.cur); + if (otherproclock->groupLeader == leader) myHeldLocks |= otherproclock->holdMask; - otherproclock = (PROCLOCK *) - SHMQueueNext(procLocks, &otherproclock->lockLink, - offsetof(PROCLOCK, lockLink)); } } @@ -1116,20 +1106,23 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) if (myHeldLocks != 0) { LOCKMASK aheadRequests = 0; + dlist_iter iter; - proc = (PGPROC *) waitQueue->links.next; - for (i = 0; i < waitQueue->size; i++) + // FIXME: Shouldn't we just use the correct offset math? + StaticAssertStmt(offsetof(PGPROC, links) == 0, "odd casting"); + + dlist_foreach(iter, &waitQueue->links) { + PGPROC *proc = dlist_container(PGPROC, links, iter.cur); + /* * If we're part of the same locking group as this waiter, its * locks neither conflict with ours nor contribute to * aheadRequests. */ if (leader != NULL && leader == proc->lockGroupLeader) - { - proc = (PGPROC *) proc->links.next; continue; - } + /* Must he wait for me? */ if (lockMethodTable->conflictTab[proc->waitLockMode] & myHeldLocks) { @@ -1157,29 +1150,25 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) GrantAwaitedLock(); return STATUS_OK; } - /* Break out of loop to put myself before him */ + + /* Put myself into wait queue before conflicting process */ + insert_before = proc; break; } /* Nope, so advance to next waiter */ aheadRequests |= LOCKBIT_ON(proc->waitLockMode); - proc = (PGPROC *) proc->links.next; } - - /* - * If we fall out of loop normally, proc points to waitQueue head, so - * we will insert at tail of queue as desired. - */ - } - else - { - /* I hold no locks, so I can't push in front of anyone. */ - proc = (PGPROC *) &(waitQueue->links); } /* - * Insert self into queue, ahead of the given proc (or at tail of queue). + * Insert self into queue, ahead of the determined proc or at the tail. */ - SHMQueueInsertBefore(&(proc->links), &(MyProc->links)); + if (insert_before) + dlist_insert_before(&insert_before->links, &MyProc->links); + else + dlist_push_tail(&waitQueue->links, &MyProc->links); + + waitQueue->size++; lock->waitMask |= LOCKBIT_ON(lockmode); @@ -1384,7 +1373,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) long secs; int usecs; long msecs; - SHM_QUEUE *procLocks; + dlist_iter proc_iter; PROCLOCK *proclock; bool first_holder = true, first_waiter = true; @@ -1414,12 +1403,11 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) LWLockAcquire(partitionLock, LW_SHARED); - procLocks = &(lock->procLocks); - proclock = (PROCLOCK *) SHMQueueNext(procLocks, procLocks, - offsetof(PROCLOCK, lockLink)); - - while (proclock) + dlist_foreach(proc_iter, &lock->procLocks) { + proclock = + dlist_container(PROCLOCK, lockLink, proc_iter.cur); + /* * we are a waiter if myProc->waitProcLock == proclock; we are * a holder if it is NULL or something different @@ -1450,9 +1438,6 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) lockHoldersNum++; } - - proclock = (PROCLOCK *) SHMQueueNext(procLocks, &proclock->lockLink, - offsetof(PROCLOCK, lockLink)); } LWLockRelease(partitionLock); @@ -1577,7 +1562,6 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) * ProcWakeup -- wake up a process by setting its latch. * * Also remove the process from the wait queue and set its links invalid. - * RETURN: the next process in the wait queue. * * The appropriate lock partition lock must be held by caller. * @@ -1586,22 +1570,17 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable) * to twiddle the lock's request counts too --- see RemoveFromWaitQueue. * Hence, in practice the waitStatus parameter must be STATUS_OK. */ -PGPROC * +void ProcWakeup(PGPROC *proc, int waitStatus) { - PGPROC *retProc; - /* Proc should be sleeping ... */ - if (proc->links.prev == NULL || - proc->links.next == NULL) - return NULL; + if (dlist_node_is_detached(&proc->links)) + return; + Assert(proc->waitStatus == STATUS_WAITING); - /* Save next process before we zap the list link */ - retProc = (PGPROC *) proc->links.next; - /* Remove process from wait queue */ - SHMQueueDelete(&(proc->links)); + dlist_delete_thoroughly(&proc->links); (proc->waitLock->waitProcs.size)--; /* Clean up process' state and pass it the ok/fail signal */ @@ -1611,8 +1590,6 @@ ProcWakeup(PGPROC *proc, int waitStatus) /* And awaken it */ SetLatch(&proc->procLatch); - - return retProc; } /* @@ -1626,19 +1603,17 @@ void ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock) { PROC_QUEUE *waitQueue = &(lock->waitProcs); - int queue_size = waitQueue->size; - PGPROC *proc; LOCKMASK aheadRequests = 0; + dlist_mutable_iter miter; - Assert(queue_size >= 0); + Assert(waitQueue->size >= 0); - if (queue_size == 0) + if (waitQueue->size == 0) return; - proc = (PGPROC *) waitQueue->links.next; - - while (queue_size-- > 0) + dlist_foreach_modify(miter, &waitQueue->links) { + PGPROC *proc = dlist_container(PGPROC, links, miter.cur); LOCKMODE lockmode = proc->waitLockMode; /* @@ -1651,7 +1626,7 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock) { /* OK to waken */ GrantLock(lock, proc->waitProcLock, lockmode); - proc = ProcWakeup(proc, STATUS_OK); + ProcWakeup(proc, STATUS_OK); /* * ProcWakeup removes proc from the lock's waiting process queue @@ -1661,11 +1636,8 @@ ProcLockWakeup(LockMethod lockMethodTable, LOCK *lock) } else { - /* - * Cannot wake this guy. Remember his request for later checks. - */ + /* Lock conflicts: don't wake, but remember for later checks. */ aheadRequests |= LOCKBIT_ON(lockmode); - proc = (PGPROC *) proc->links.next; } } -- 2.25.0.114.g5b0ca878e0