>From 5e456f71e5e83b6ccb8ad471b6327592242f0cd4 Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Fri, 31 Jan 2014 10:53:19 +0100 Subject: [PATCH 4/5] Convert the PGPROC->lwWaitLink list into a dlist instead of open coding it. Besides being shorter and much easier to read it: * changes the logic in LWLockRelease() to release all shared lockers when waking up any. * adds a memory pg_write_barrier() in the wakeup paths between dequeuing and unsetting ->lwWaiting. On a weakly ordered machine the current coding could lead to problems. --- src/backend/access/transam/twophase.c | 1 - src/backend/storage/lmgr/lwlock.c | 151 +++++++++++++--------------------- src/backend/storage/lmgr/proc.c | 2 - src/include/storage/lwlock.h | 5 +- src/include/storage/proc.h | 3 +- 5 files changed, 60 insertions(+), 102 deletions(-) diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index d5409a6..6401943 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -389,7 +389,6 @@ MarkAsPreparing(TransactionId xid, const char *gid, proc->roleId = owner; proc->lwWaiting = false; proc->lwWaitMode = 0; - proc->lwWaitLink = NULL; proc->waitLock = NULL; proc->waitProcLock = NULL; for (i = 0; i < NUM_LOCK_PARTITIONS; i++) diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c index 7c96da5..055174d 100644 --- a/src/backend/storage/lmgr/lwlock.c +++ b/src/backend/storage/lmgr/lwlock.c @@ -35,6 +35,7 @@ #include "miscadmin.h" #include "pg_trace.h" #include "replication/slot.h" +#include "storage/barrier.h" #include "storage/ipc.h" #include "storage/predicate.h" #include "storage/proc.h" @@ -115,9 +116,9 @@ inline static void PRINT_LWDEBUG(const char *where, const volatile LWLock *lock) { if (Trace_lwlocks) - elog(LOG, "%s(%s %d): excl %d shared %d head %p rOK %d", + elog(LOG, "%s(%s %d): excl %d shared %d rOK %d", where, T_NAME(lock), T_ID(lock), - (int) lock->exclusive, lock->shared, lock->head, + (int) lock->exclusive, lock->shared, (int) lock->releaseOK); } @@ -479,8 +480,7 @@ LWLockInitialize(LWLock *lock, int tranche_id) lock->exclusive = 0; lock->shared = 0; lock->tranche = tranche_id; - lock->head = NULL; - lock->tail = NULL; + dlist_init(&lock->waiters); } @@ -620,12 +620,7 @@ LWLockAcquireCommon(LWLock *l, LWLockMode mode, uint64 *valptr, uint64 val) proc->lwWaiting = true; proc->lwWaitMode = mode; - proc->lwWaitLink = NULL; - if (lock->head == NULL) - lock->head = proc; - else - lock->tail->lwWaitLink = proc; - lock->tail = proc; + dlist_push_head((dlist_head *) &lock->waiters, &proc->lwWaitLink); /* Can release the mutex now */ SpinLockRelease(&lock->mutex); @@ -841,12 +836,7 @@ LWLockAcquireOrWait(LWLock *l, LWLockMode mode) proc->lwWaiting = true; proc->lwWaitMode = LW_WAIT_UNTIL_FREE; - proc->lwWaitLink = NULL; - if (lock->head == NULL) - lock->head = proc; - else - lock->tail->lwWaitLink = proc; - lock->tail = proc; + dlist_push_head((dlist_head *) &lock->waiters, &proc->lwWaitLink); /* Can release the mutex now */ SpinLockRelease(&lock->mutex); @@ -1000,13 +990,8 @@ LWLockWaitForVar(LWLock *l, uint64 *valptr, uint64 oldval, uint64 *newval) */ proc->lwWaiting = true; proc->lwWaitMode = LW_WAIT_UNTIL_FREE; - proc->lwWaitLink = NULL; - /* waiters are added to the front of the queue */ - proc->lwWaitLink = lock->head; - if (lock->head == NULL) - lock->tail = proc; - lock->head = proc; + dlist_push_head((dlist_head *) &lock->waiters, &proc->lwWaitLink); /* Can release the mutex now */ SpinLockRelease(&lock->mutex); @@ -1082,9 +1067,10 @@ LWLockUpdateVar(LWLock *l, uint64 *valptr, uint64 val) { volatile LWLock *lock = l; volatile uint64 *valp = valptr; - PGPROC *head; - PGPROC *proc; - PGPROC *next; + dlist_head wakeup; + dlist_mutable_iter iter; + + dlist_init(&wakeup); /* Acquire mutex. Time spent holding mutex should be short! */ SpinLockAcquire(&lock->mutex); @@ -1099,24 +1085,16 @@ LWLockUpdateVar(LWLock *l, uint64 *valptr, uint64 val) * See if there are any LW_WAIT_UNTIL_FREE waiters that need to be woken * up. They are always in the front of the queue. */ - head = lock->head; - - if (head != NULL && head->lwWaitMode == LW_WAIT_UNTIL_FREE) + dlist_foreach_modify(iter, (dlist_head *) &lock->waiters) { - proc = head; - next = proc->lwWaitLink; - while (next && next->lwWaitMode == LW_WAIT_UNTIL_FREE) - { - proc = next; - next = next->lwWaitLink; - } + PGPROC *waiter = dlist_container(PGPROC, lwWaitLink, iter.cur); - /* proc is now the last PGPROC to be released */ - lock->head = next; - proc->lwWaitLink = NULL; + if (waiter->lwWaitMode != LW_WAIT_UNTIL_FREE) + break; + + dlist_delete(&waiter->lwWaitLink); + dlist_push_tail(&wakeup, &waiter->lwWaitLink); } - else - head = NULL; /* We are done updating shared state of the lock itself. */ SpinLockRelease(&lock->mutex); @@ -1124,13 +1102,13 @@ LWLockUpdateVar(LWLock *l, uint64 *valptr, uint64 val) /* * Awaken any waiters I removed from the queue. */ - while (head != NULL) + dlist_foreach_modify(iter, (dlist_head *) &wakeup) { - proc = head; - head = proc->lwWaitLink; - proc->lwWaitLink = NULL; - proc->lwWaiting = false; - PGSemaphoreUnlock(&proc->sem); + PGPROC *waiter = dlist_container(PGPROC, lwWaitLink, iter.cur); + dlist_delete(&waiter->lwWaitLink); + pg_write_barrier(); + waiter->lwWaiting = false; + PGSemaphoreUnlock(&waiter->sem); } } @@ -1142,10 +1120,12 @@ void LWLockRelease(LWLock *l) { volatile LWLock *lock = l; - PGPROC *head; - PGPROC *proc; + dlist_head wakeup; + dlist_mutable_iter iter; int i; + dlist_init(&wakeup); + PRINT_LWDEBUG("LWLockRelease", lock); /* @@ -1181,58 +1161,39 @@ LWLockRelease(LWLock *l) * if someone has already awakened waiters that haven't yet acquired the * lock. */ - head = lock->head; - if (head != NULL) + if (lock->exclusive == 0 && lock->shared == 0 && lock->releaseOK) { - if (lock->exclusive == 0 && lock->shared == 0 && lock->releaseOK) + /* + * Remove the to-be-awakened PGPROCs from the queue. + */ + bool releaseOK = true; + bool wokeup_somebody = false; + + dlist_foreach_modify(iter, (dlist_head *) &lock->waiters) { - /* - * Remove the to-be-awakened PGPROCs from the queue. - */ - bool releaseOK = true; + PGPROC *waiter = dlist_container(PGPROC, lwWaitLink, iter.cur); - proc = head; + if (wokeup_somebody && waiter->lwWaitMode == LW_EXCLUSIVE) + continue; - /* - * First wake up any backends that want to be woken up without - * acquiring the lock. - */ - while (proc->lwWaitMode == LW_WAIT_UNTIL_FREE && proc->lwWaitLink) - proc = proc->lwWaitLink; + dlist_delete(&waiter->lwWaitLink); + dlist_push_tail(&wakeup, &waiter->lwWaitLink); /* - * If the front waiter wants exclusive lock, awaken him only. - * Otherwise awaken as many waiters as want shared access. + * Prevent additional wakeups until retryer gets to + * run. Backends that are just waiting for the lock to become + * free don't retry automatically. */ - if (proc->lwWaitMode != LW_EXCLUSIVE) + if (waiter->lwWaitMode != LW_WAIT_UNTIL_FREE) { - while (proc->lwWaitLink != NULL && - proc->lwWaitLink->lwWaitMode != LW_EXCLUSIVE) - { - if (proc->lwWaitMode != LW_WAIT_UNTIL_FREE) - releaseOK = false; - proc = proc->lwWaitLink; - } - } - /* proc is now the last PGPROC to be released */ - lock->head = proc->lwWaitLink; - proc->lwWaitLink = NULL; - - /* - * Prevent additional wakeups until retryer gets to run. Backends - * that are just waiting for the lock to become free don't retry - * automatically. - */ - if (proc->lwWaitMode != LW_WAIT_UNTIL_FREE) releaseOK = false; + wokeup_somebody = true; + } - lock->releaseOK = releaseOK; - } - else - { - /* lock is still held, can't awaken anything */ - head = NULL; + if(waiter->lwWaitMode == LW_EXCLUSIVE) + break; } + lock->releaseOK = releaseOK; } /* We are done updating shared state of the lock itself. */ @@ -1243,14 +1204,14 @@ LWLockRelease(LWLock *l) /* * Awaken any waiters I removed from the queue. */ - while (head != NULL) + dlist_foreach_modify(iter, (dlist_head *) &wakeup) { + PGPROC *waiter = dlist_container(PGPROC, lwWaitLink, iter.cur); LOG_LWDEBUG("LWLockRelease", T_NAME(l), T_ID(l), "release waiter"); - proc = head; - head = proc->lwWaitLink; - proc->lwWaitLink = NULL; - proc->lwWaiting = false; - PGSemaphoreUnlock(&proc->sem); + dlist_delete(&waiter->lwWaitLink); + pg_write_barrier(); + waiter->lwWaiting = false; + PGSemaphoreUnlock(&waiter->sem); } /* diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index ea88a24..a4789fc 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -372,7 +372,6 @@ InitProcess(void) MyPgXact->vacuumFlags |= PROC_IS_AUTOVACUUM; MyProc->lwWaiting = false; MyProc->lwWaitMode = 0; - MyProc->lwWaitLink = NULL; MyProc->waitLock = NULL; MyProc->waitProcLock = NULL; #ifdef USE_ASSERT_CHECKING @@ -535,7 +534,6 @@ InitAuxiliaryProcess(void) MyPgXact->vacuumFlags = 0; MyProc->lwWaiting = false; MyProc->lwWaitMode = 0; - MyProc->lwWaitLink = NULL; MyProc->waitLock = NULL; MyProc->waitProcLock = NULL; #ifdef USE_ASSERT_CHECKING diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h index 1d90b9f..1bca57b 100644 --- a/src/include/storage/lwlock.h +++ b/src/include/storage/lwlock.h @@ -14,6 +14,7 @@ #ifndef LWLOCK_H #define LWLOCK_H +#include "lib/ilist.h" #include "storage/s_lock.h" struct PGPROC; @@ -50,9 +51,7 @@ typedef struct LWLock char exclusive; /* # of exclusive holders (0 or 1) */ int shared; /* # of shared holders (0..MaxBackends) */ int tranche; /* tranche ID */ - struct PGPROC *head; /* head of list of waiting PGPROCs */ - struct PGPROC *tail; /* tail of list of waiting PGPROCs */ - /* tail is undefined when head is NULL */ + dlist_head waiters; /* list of waiting PGPROCs */ } LWLock; /* diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index c23f4da..31f1b04 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -15,6 +15,7 @@ #define _PROC_H_ #include "access/xlogdefs.h" +#include "lib/ilist.h" #include "storage/latch.h" #include "storage/lock.h" #include "storage/pg_sema.h" @@ -104,7 +105,7 @@ struct PGPROC /* Info about LWLock the process is currently waiting for, if any. */ bool lwWaiting; /* true if waiting for an LW lock */ uint8 lwWaitMode; /* lwlock mode being waited for */ - struct PGPROC *lwWaitLink; /* next waiter for same LW lock */ + dlist_node lwWaitLink; /* next waiter for same LW lock */ /* Info about lock the process is currently waiting for, if any. */ /* waitLock and waitProcLock are NULL if not currently waiting. */ -- 2.0.0.rc2.4.g1dc51c6.dirty