/*------------------------------------------------------------------------- * * lwlock.c * Lightweight lock manager * * Lightweight locks are intended primarily to provide mutual exclusion of * access to shared-memory data structures. Therefore, they offer both * exclusive and shared lock modes (to support read/write and read-only * access to a shared object). There are few other frammishes. User-level * locking should be done with the full lock manager --- which depends on * LWLocks to protect its shared state. * * In addition to exclusive and shared modes, lightweight locks can be used to * wait until a variable changes value. The variable is initially not set * when the lock is acquired with LWLockAcquire, i.e. it remains set to the * value it was set to when the lock was released last, and can be updated * without releasing the lock by calling LWLockUpdateVar. LWLockWaitForVar * waits for the variable to be updated, or until the lock is free. When * releasing the lock with LWLockReleaseClearVar() the value can be set to an * appropriate value for a free lock. The meaning of the variable is up to * the caller, the lightweight lock code just assigns and compares it. * * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * IDENTIFICATION * src/backend/storage/lmgr/lwlock.c * * NOTES: * * This used to be a pretty straight forward reader-writer lock * implementation, in which the internal state was protected by a * spinlock. Unfortunately the overhead of taking the spinlock proved to be * too high for workloads/locks that were taken in shared mode very * frequently. Often we were spinning in the (obviously exclusive) spinlock, * while trying to acquire a shared lock that was actually free. * * Thus a new implementation was devised that provides wait-free shared lock * acquisition for locks that aren't exclusively locked. * * The basic idea is to have a single atomic variable 'lockcount' instead of * the formerly separate shared and exclusive counters and to use atomic * operations to acquire the lock. That's fairly easy to do for plain * rw-spinlocks, but a lot harder for something like LWLocks that want to wait * in the OS. * * For lock acquisition we use an atomic compare-and-exchange on the lockcount * variable. For exclusive lock we swap in a sentinel value * (LW_VAL_EXCLUSIVE), for shared locks we count the number of holders. * * To release the lock we use an atomic decrement to release the lock. If the * new value is zero (we get that atomically), we know we can/have to release * waiters. * * Obviously it is important that the sentinel value for exclusive locks * doesn't conflict with the maximum number of possible share lockers - * luckily MAX_BACKENDS makes that easily possible. * * * The attentive reader might have noticed that naively doing the above has a * glaring race condition: We try to lock using the atomic operations and * notice that we have to wait. Unfortunately by the time we have finished * queuing, the former locker very well might have already finished its * work. That's problematic because we're now stuck waiting inside the OS. * To mitigate those races we use a two phased attempt at locking: * Phase 1: Try to do it atomically, if we succeed, nice * Phase 2: Add ourselves to the waitqueue of the lock * Phase 3: Try to grab the lock again, if we succeed, remove ourselves from * the queue * Phase 4: Sleep till wake-up, goto Phase 1 * * This protects us against the problem from above as nobody can release too * quick, before we're queued, since after Phase 2 we're already queued. * ------------------------------------------------------------------------- */ #include "postgres.h" #include "miscadmin.h" #include "pg_trace.h" #include "pgstat.h" #include "port/pg_bitutils.h" #include "storage/proc.h" #include "storage/proclist.h" #include "storage/procnumber.h" #include "storage/spin.h" #include "utils/memutils.h" #ifdef LWLOCK_STATS #include "utils/hsearch.h" #endif #define LW_FLAG_HAS_WAITERS ((uint32) 1 << 31) #define LW_FLAG_RELEASE_OK ((uint32) 1 << 30) #define LW_FLAG_LOCKED ((uint32) 1 << 29) #define LW_FLAG_BITS 3 #define LW_FLAG_MASK (((1<tranche) #ifdef LWLOCK_STATS typedef struct lwlock_stats_key { int tranche; void *instance; } lwlock_stats_key; typedef struct lwlock_stats { lwlock_stats_key key; int sh_acquire_count; int ex_acquire_count; int block_count; int dequeue_self_count; int spin_delay_count; } lwlock_stats; static HTAB *lwlock_stats_htab; static lwlock_stats lwlock_stats_dummy; #endif #ifdef LOCK_DEBUG bool Trace_lwlocks = false; inline static void PRINT_LWDEBUG(const char *where, LWLock *lock, LWLockMode mode) { /* hide statement & context here, otherwise the log is just too verbose */ if (Trace_lwlocks) { uint32 state = pg_atomic_read_u32(&lock->state); ereport(LOG, (errhidestmt(true), errhidecontext(true), errmsg_internal("%d: %s(%s %p): excl %u shared %u haswaiters %u waiters %u rOK %d", MyProcPid, where, T_NAME(lock), lock, (state & LW_VAL_EXCLUSIVE) != 0, state & LW_SHARED_MASK, (state & LW_FLAG_HAS_WAITERS) != 0, pg_atomic_read_u32(&lock->nwaiters), (state & LW_FLAG_RELEASE_OK) != 0))); } } inline static void LOG_LWDEBUG(const char *where, LWLock *lock, const char *msg) { /* hide statement & context here, otherwise the log is just too verbose */ if (Trace_lwlocks) { ereport(LOG, (errhidestmt(true), errhidecontext(true), errmsg_internal("%s(%s %p): %s", where, T_NAME(lock), lock, msg))); } } #else /* not LOCK_DEBUG */ #define PRINT_LWDEBUG(a,b,c) ((void)0) #define LOG_LWDEBUG(a,b,c) ((void)0) #endif /* LOCK_DEBUG */ #ifdef LWLOCK_STATS static void init_lwlock_stats(void); static void print_lwlock_stats(int code, Datum arg); static lwlock_stats * get_lwlock_stats_entry(LWLock *lock); static void init_lwlock_stats(void) { HASHCTL ctl; static MemoryContext lwlock_stats_cxt = NULL; static bool exit_registered = false; if (lwlock_stats_cxt != NULL) MemoryContextDelete(lwlock_stats_cxt); /* * The LWLock stats will be updated within a critical section, which * requires allocating new hash entries. Allocations within a critical * section are normally not allowed because running out of memory would * lead to a PANIC, but LWLOCK_STATS is debugging code that's not normally * turned on in production, so that's an acceptable risk. The hash entries * are small, so the risk of running out of memory is minimal in practice. */ lwlock_stats_cxt = AllocSetContextCreate(TopMemoryContext, "LWLock stats", ALLOCSET_DEFAULT_SIZES); MemoryContextAllowInCriticalSection(lwlock_stats_cxt, true); ctl.keysize = sizeof(lwlock_stats_key); ctl.entrysize = sizeof(lwlock_stats); ctl.hcxt = lwlock_stats_cxt; lwlock_stats_htab = hash_create("lwlock stats", 16384, &ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); if (!exit_registered) { on_shmem_exit(print_lwlock_stats, 0); exit_registered = true; } } static void print_lwlock_stats(int code, Datum arg) { HASH_SEQ_STATUS scan; lwlock_stats *lwstats; hash_seq_init(&scan, lwlock_stats_htab); /* Grab an LWLock to keep different backends from mixing reports */ LWLockAcquire(&MainLWLockArray[0].lock, LW_EXCLUSIVE); while ((lwstats = (lwlock_stats *) hash_seq_search(&scan)) != NULL) { fprintf(stderr, "PID %d lwlock %s %p: shacq %u exacq %u blk %u spindelay %u dequeue self %u\n", MyProcPid, GetLWTrancheName(lwstats->key.tranche), lwstats->key.instance, lwstats->sh_acquire_count, lwstats->ex_acquire_count, lwstats->block_count, lwstats->spin_delay_count, lwstats->dequeue_self_count); } LWLockRelease(&MainLWLockArray[0].lock); } static lwlock_stats * get_lwlock_stats_entry(LWLock *lock) { lwlock_stats_key key; lwlock_stats *lwstats; bool found; /* * During shared memory initialization, the hash table doesn't exist yet. * Stats of that phase aren't very interesting, so just collect operations * on all locks in a single dummy entry. */ if (lwlock_stats_htab == NULL) return &lwlock_stats_dummy; /* Fetch or create the entry. */ MemSet(&key, 0, sizeof(key)); key.tranche = lock->tranche; key.instance = lock; lwstats = hash_search(lwlock_stats_htab, &key, HASH_ENTER, &found); if (!found) { lwstats->sh_acquire_count = 0; lwstats->ex_acquire_count = 0; lwstats->block_count = 0; lwstats->dequeue_self_count = 0; lwstats->spin_delay_count = 0; } return lwstats; } #endif /* LWLOCK_STATS */ /* * Compute number of LWLocks required by named tranches. These will be * allocated in the main array. */ static int NumLWLocksForNamedTranches(void) { int numLocks = 0; int i; for (i = 0; i < NamedLWLockTrancheRequests; i++) numLocks += NamedLWLockTrancheRequestArray[i].num_lwlocks; return numLocks; } /* * Compute shmem space needed for LWLocks and named tranches. */ Size LWLockShmemSize(void) { Size size; int i; int numLocks = NUM_FIXED_LWLOCKS; /* Calculate total number of locks needed in the main array. */ numLocks += NumLWLocksForNamedTranches(); /* Space for the LWLock array. */ size = mul_size(numLocks, sizeof(LWLockPadded)); /* Space for dynamic allocation counter, plus room for alignment. */ size = add_size(size, sizeof(int) + LWLOCK_PADDED_SIZE); /* space for named tranches. */ size = add_size(size, mul_size(NamedLWLockTrancheRequests, sizeof(NamedLWLockTranche))); /* space for name of each tranche. */ for (i = 0; i < NamedLWLockTrancheRequests; i++) size = add_size(size, strlen(NamedLWLockTrancheRequestArray[i].tranche_name) + 1); return size; } /* * Allocate shmem space for the main LWLock array and all tranches and * initialize it. We also register extension LWLock tranches here. */ void CreateLWLocks(void) { if (!IsUnderPostmaster) { Size spaceLocks = LWLockShmemSize(); int *LWLockCounter; char *ptr; /* Allocate space */ ptr = (char *) ShmemAlloc(spaceLocks); /* Leave room for dynamic allocation of tranches */ ptr += sizeof(int); /* Ensure desired alignment of LWLock array */ ptr += LWLOCK_PADDED_SIZE - ((uintptr_t) ptr) % LWLOCK_PADDED_SIZE; MainLWLockArray = (LWLockPadded *) ptr; /* * Initialize the dynamic-allocation counter for tranches, which is * stored just before the first LWLock. */ LWLockCounter = (int *) ((char *) MainLWLockArray - sizeof(int)); *LWLockCounter = LWTRANCHE_FIRST_USER_DEFINED; /* Initialize all LWLocks */ InitializeLWLocks(); } /* Register named extension LWLock tranches in the current process. */ for (int i = 0; i < NamedLWLockTrancheRequests; i++) LWLockRegisterTranche(NamedLWLockTrancheArray[i].trancheId, NamedLWLockTrancheArray[i].trancheName); } /* * Initialize LWLocks that are fixed and those belonging to named tranches. */ static void InitializeLWLocks(void) { int numNamedLocks = NumLWLocksForNamedTranches(); int id; int i; int j; LWLockPadded *lock; /* Initialize all individual LWLocks in main array */ for (id = 0, lock = MainLWLockArray; id < NUM_INDIVIDUAL_LWLOCKS; id++, lock++) LWLockInitialize(&lock->lock, id); /* Initialize buffer mapping LWLocks in main array */ lock = MainLWLockArray + BUFFER_MAPPING_LWLOCK_OFFSET; for (id = 0; id < NUM_BUFFER_PARTITIONS; id++, lock++) LWLockInitialize(&lock->lock, LWTRANCHE_BUFFER_MAPPING); /* Initialize lmgrs' LWLocks in main array */ lock = MainLWLockArray + LOCK_MANAGER_LWLOCK_OFFSET; for (id = 0; id < NUM_LOCK_PARTITIONS; id++, lock++) LWLockInitialize(&lock->lock, LWTRANCHE_LOCK_MANAGER); /* Initialize predicate lmgrs' LWLocks in main array */ lock = MainLWLockArray + PREDICATELOCK_MANAGER_LWLOCK_OFFSET; for (id = 0; id < NUM_PREDICATELOCK_PARTITIONS; id++, lock++) LWLockInitialize(&lock->lock, LWTRANCHE_PREDICATE_LOCK_MANAGER); /* * Copy the info about any named tranches into shared memory (so that * other processes can see it), and initialize the requested LWLocks. */ if (NamedLWLockTrancheRequests > 0) { char *trancheNames; NamedLWLockTrancheArray = (NamedLWLockTranche *) &MainLWLockArray[NUM_FIXED_LWLOCKS + numNamedLocks]; trancheNames = (char *) NamedLWLockTrancheArray + (NamedLWLockTrancheRequests * sizeof(NamedLWLockTranche)); lock = &MainLWLockArray[NUM_FIXED_LWLOCKS]; for (i = 0; i < NamedLWLockTrancheRequests; i++) { NamedLWLockTrancheRequest *request; NamedLWLockTranche *tranche; char *name; request = &NamedLWLockTrancheRequestArray[i]; tranche = &NamedLWLockTrancheArray[i]; name = trancheNames; trancheNames += strlen(request->tranche_name) + 1; strcpy(name, request->tranche_name); tranche->trancheId = LWLockNewTrancheId(); tranche->trancheName = name; for (j = 0; j < request->num_lwlocks; j++, lock++) LWLockInitialize(&lock->lock, tranche->trancheId); } } } /* * InitLWLockAccess - initialize backend-local state needed to hold LWLocks */ void InitLWLockAccess(void) { #ifdef LWLOCK_STATS init_lwlock_stats(); #endif } /* * GetNamedLWLockTranche - returns the base address of LWLock from the * specified tranche. * * Caller needs to retrieve the requested number of LWLocks starting from * the base lock address returned by this API. This can be used for * tranches that are requested by using RequestNamedLWLockTranche() API. */ LWLockPadded * GetNamedLWLockTranche(const char *tranche_name) { int lock_pos; int i; /* * Obtain the position of base address of LWLock belonging to requested * tranche_name in MainLWLockArray. LWLocks for named tranches are placed * in MainLWLockArray after fixed locks. */ lock_pos = NUM_FIXED_LWLOCKS; for (i = 0; i < NamedLWLockTrancheRequests; i++) { if (strcmp(NamedLWLockTrancheRequestArray[i].tranche_name, tranche_name) == 0) return &MainLWLockArray[lock_pos]; lock_pos += NamedLWLockTrancheRequestArray[i].num_lwlocks; } elog(ERROR, "requested tranche is not registered"); /* just to keep compiler quiet */ return NULL; } /* * Allocate a new tranche ID. */ int LWLockNewTrancheId(void) { int result; int *LWLockCounter; LWLockCounter = (int *) ((char *) MainLWLockArray - sizeof(int)); /* We use the ShmemLock spinlock to protect LWLockCounter */ SpinLockAcquire(ShmemLock); result = (*LWLockCounter)++; SpinLockRelease(ShmemLock); return result; } /* * Register a dynamic tranche name in the lookup table of the current process. * * This routine will save a pointer to the tranche name passed as an argument, * so the name should be allocated in a backend-lifetime context * (shared memory, TopMemoryContext, static constant, or similar). * * The tranche name will be user-visible as a wait event name, so try to * use a name that fits the style for those. */ void LWLockRegisterTranche(int tranche_id, const char *tranche_name) { /* This should only be called for user-defined tranches. */ if (tranche_id < LWTRANCHE_FIRST_USER_DEFINED) return; /* Convert to array index. */ tranche_id -= LWTRANCHE_FIRST_USER_DEFINED; /* If necessary, create or enlarge array. */ if (tranche_id >= LWLockTrancheNamesAllocated) { int newalloc; newalloc = pg_nextpower2_32(Max(8, tranche_id + 1)); if (LWLockTrancheNames == NULL) LWLockTrancheNames = (const char **) MemoryContextAllocZero(TopMemoryContext, newalloc * sizeof(char *)); else LWLockTrancheNames = repalloc0_array(LWLockTrancheNames, const char *, LWLockTrancheNamesAllocated, newalloc); LWLockTrancheNamesAllocated = newalloc; } LWLockTrancheNames[tranche_id] = tranche_name; } /* * RequestNamedLWLockTranche * Request that extra LWLocks be allocated during postmaster * startup. * * This may only be called via the shmem_request_hook of a library that is * loaded into the postmaster via shared_preload_libraries. Calls from * elsewhere will fail. * * The tranche name will be user-visible as a wait event name, so try to * use a name that fits the style for those. */ void RequestNamedLWLockTranche(const char *tranche_name, int num_lwlocks) { NamedLWLockTrancheRequest *request; if (!process_shmem_requests_in_progress) elog(FATAL, "cannot request additional LWLocks outside shmem_request_hook"); if (NamedLWLockTrancheRequestArray == NULL) { NamedLWLockTrancheRequestsAllocated = 16; NamedLWLockTrancheRequestArray = (NamedLWLockTrancheRequest *) MemoryContextAlloc(TopMemoryContext, NamedLWLockTrancheRequestsAllocated * sizeof(NamedLWLockTrancheRequest)); } if (NamedLWLockTrancheRequests >= NamedLWLockTrancheRequestsAllocated) { int i = pg_nextpower2_32(NamedLWLockTrancheRequests + 1); NamedLWLockTrancheRequestArray = (NamedLWLockTrancheRequest *) repalloc(NamedLWLockTrancheRequestArray, i * sizeof(NamedLWLockTrancheRequest)); NamedLWLockTrancheRequestsAllocated = i; } request = &NamedLWLockTrancheRequestArray[NamedLWLockTrancheRequests]; Assert(strlen(tranche_name) + 1 <= NAMEDATALEN); strlcpy(request->tranche_name, tranche_name, NAMEDATALEN); request->num_lwlocks = num_lwlocks; NamedLWLockTrancheRequests++; } /* * LWLockInitialize - initialize a new lwlock; it's initially unlocked */ void LWLockInitialize(LWLock *lock, int tranche_id) { pg_atomic_init_u32(&lock->state, LW_FLAG_RELEASE_OK); #ifdef LOCK_DEBUG pg_atomic_init_u32(&lock->nwaiters, 0); #endif lock->tranche = tranche_id; proclist_init(&lock->waiters); } /* * Report start of wait event for light-weight locks. * * This function will be used by all the light-weight lock calls which * needs to wait to acquire the lock. This function distinguishes wait * event based on tranche and lock id. */ static inline void LWLockReportWaitStart(LWLock *lock) { pgstat_report_wait_start(PG_WAIT_LWLOCK | lock->tranche); } /* * Report end of wait event for light-weight locks. */ static inline void LWLockReportWaitEnd(void) { pgstat_report_wait_end(); } /* * Return the name of an LWLock tranche. */ static const char * GetLWTrancheName(uint16 trancheId) { /* Built-in tranche or individual LWLock? */ if (trancheId < LWTRANCHE_FIRST_USER_DEFINED) return BuiltinTrancheNames[trancheId]; /* * It's an extension tranche, so look in LWLockTrancheNames[]. However, * it's possible that the tranche has never been registered in the current * process, in which case give up and return "extension". */ trancheId -= LWTRANCHE_FIRST_USER_DEFINED; if (trancheId >= LWLockTrancheNamesAllocated || LWLockTrancheNames[trancheId] == NULL) return "extension"; return LWLockTrancheNames[trancheId]; } /* * Return an identifier for an LWLock based on the wait class and event. */ const char * GetLWLockIdentifier(uint32 classId, uint16 eventId) { Assert(classId == PG_WAIT_LWLOCK); /* The event IDs are just tranche numbers. */ return GetLWTrancheName(eventId); } /* * Internal function that tries to atomically acquire the lwlock in the passed * in mode. * * This function will not block waiting for a lock to become free - that's the * caller's job. * * Returns true if the lock isn't free and we need to wait. */ static bool LWLockAttemptLock(LWLock *lock, LWLockMode mode) { uint32 old_state; Assert(mode == LW_EXCLUSIVE || mode == LW_SHARED); /* * Read once outside the loop, later iterations will get the newer value * via compare & exchange. */ old_state = pg_atomic_read_u32(&lock->state); /* loop until we've determined whether we could acquire the lock or not */ while (true) { uint32 desired_state; bool lock_free; desired_state = old_state; if (mode == LW_EXCLUSIVE) { lock_free = (old_state & LW_LOCK_MASK) == 0; if (lock_free) desired_state += LW_VAL_EXCLUSIVE; } else { lock_free = (old_state & LW_VAL_EXCLUSIVE) == 0; if (lock_free) desired_state += LW_VAL_SHARED; } /* * Attempt to swap in the state we are expecting. If we didn't see * lock to be free, that's just the old value. If we saw it as free, * we'll attempt to mark it acquired. The reason that we always swap * in the value is that this doubles as a memory barrier. We could try * to be smarter and only swap in values if we saw the lock as free, * but benchmark haven't shown it as beneficial so far. * * Retry if the value changed since we last looked at it. */ if (pg_atomic_compare_exchange_u32(&lock->state, &old_state, desired_state)) { if (lock_free) { /* Great! Got the lock. */ #ifdef LOCK_DEBUG if (mode == LW_EXCLUSIVE) lock->owner = MyProc; #endif return false; } else return true; /* somebody else has the lock */ } } pg_unreachable(); } /* * Lock the LWLock's wait list against concurrent activity. * * NB: even though the wait list is locked, non-conflicting lock operations * may still happen concurrently. * * Time spent holding mutex should be short! */ static void LWLockWaitListLock(LWLock *lock) { uint32 old_state; #ifdef LWLOCK_STATS lwlock_stats *lwstats; uint32 delays = 0; lwstats = get_lwlock_stats_entry(lock); #endif while (true) { /* always try once to acquire lock directly */ old_state = pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_LOCKED); if (!(old_state & LW_FLAG_LOCKED)) break; /* got lock */ /* and then spin without atomic operations until lock is released */ { SpinDelayStatus delayStatus; init_local_spin_delay(&delayStatus); while (old_state & LW_FLAG_LOCKED) { perform_spin_delay(&delayStatus); old_state = pg_atomic_read_u32(&lock->state); } #ifdef LWLOCK_STATS delays += delayStatus.delays; #endif finish_spin_delay(&delayStatus); } /* * Retry. The lock might obviously already be re-acquired by the time * we're attempting to get it again. */ } #ifdef LWLOCK_STATS lwstats->spin_delay_count += delays; #endif } /* * Unlock the LWLock's wait list. * * Note that it can be more efficient to manipulate flags and release the * locks in a single atomic operation. */ static void LWLockWaitListUnlock(LWLock *lock) { uint32 old_state PG_USED_FOR_ASSERTS_ONLY; old_state = pg_atomic_fetch_and_u32(&lock->state, ~LW_FLAG_LOCKED); Assert(old_state & LW_FLAG_LOCKED); } /* * Wakeup all the lockers that currently have a chance to acquire the lock. */ static void LWLockWakeup(LWLock *lock) { bool new_release_ok; bool wokeup_somebody = false; proclist_head wakeup; proclist_mutable_iter iter; proclist_init(&wakeup); new_release_ok = true; /* lock wait list while collecting backends to wake up */ LWLockWaitListLock(lock); proclist_foreach_modify(iter, &lock->waiters, lwWaitLink) { PGPROC *waiter = GetPGProcByNumber(iter.cur); if (wokeup_somebody && waiter->lwWaitMode == LW_EXCLUSIVE) continue; proclist_delete(&lock->waiters, iter.cur, lwWaitLink); proclist_push_tail(&wakeup, iter.cur, lwWaitLink); if (waiter->lwWaitMode != LW_WAIT_UNTIL_FREE) { /* * Prevent additional wakeups until retryer gets to run. Backends * that are just waiting for the lock to become free don't retry * automatically. */ new_release_ok = false; /* * Don't wakeup (further) exclusive locks. */ wokeup_somebody = true; } /* * Signal that the process isn't on the wait list anymore. This allows * LWLockDequeueSelf() to remove itself of the waitlist with a * proclist_delete(), rather than having to check if it has been * removed from the list. */ Assert(waiter->lwWaiting == LW_WS_WAITING); waiter->lwWaiting = LW_WS_PENDING_WAKEUP; /* * Once we've woken up an exclusive lock, there's no point in waking * up anybody else. */ if (waiter->lwWaitMode == LW_EXCLUSIVE) break; } Assert(proclist_is_empty(&wakeup) || pg_atomic_read_u32(&lock->state) & LW_FLAG_HAS_WAITERS); /* unset required flags, and release lock, in one fell swoop */ { uint32 old_state; uint32 desired_state; old_state = pg_atomic_read_u32(&lock->state); while (true) { desired_state = old_state; /* compute desired flags */ if (new_release_ok) desired_state |= LW_FLAG_RELEASE_OK; else desired_state &= ~LW_FLAG_RELEASE_OK; if (proclist_is_empty(&wakeup)) desired_state &= ~LW_FLAG_HAS_WAITERS; desired_state &= ~LW_FLAG_LOCKED; /* release lock */ if (pg_atomic_compare_exchange_u32(&lock->state, &old_state, desired_state)) break; } } /* Awaken any waiters I removed from the queue. */ proclist_foreach_modify(iter, &wakeup, lwWaitLink) { PGPROC *waiter = GetPGProcByNumber(iter.cur); LOG_LWDEBUG("LWLockRelease", lock, "release waiter"); proclist_delete(&wakeup, iter.cur, lwWaitLink); /* * Guarantee that lwWaiting being unset only becomes visible once the * unlink from the link has completed. Otherwise the target backend * could be woken up for other reason and enqueue for a new lock - if * that happens before the list unlink happens, the list would end up * being corrupted. * * The barrier pairs with the LWLockWaitListLock() when enqueuing for * another lock. */ pg_write_barrier(); waiter->lwWaiting = LW_WS_NOT_WAITING; PGSemaphoreUnlock(waiter->sem); } } /* * Add ourselves to the end of the queue. * * NB: Mode can be LW_WAIT_UNTIL_FREE here! */ static void LWLockQueueSelf(LWLock *lock, LWLockMode mode) { /* * If we don't have a PGPROC structure, there's no way to wait. This * should never occur, since MyProc should only be null during shared * memory initialization. */ if (MyProc == NULL) elog(PANIC, "cannot wait without a PGPROC structure"); if (MyProc->lwWaiting != LW_WS_NOT_WAITING) elog(PANIC, "queueing for lock while waiting on another one"); LWLockWaitListLock(lock); /* setting the flag is protected by the spinlock */ pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_HAS_WAITERS); MyProc->lwWaiting = LW_WS_WAITING; MyProc->lwWaitMode = mode; /* LW_WAIT_UNTIL_FREE waiters are always at the front of the queue */ if (mode == LW_WAIT_UNTIL_FREE) proclist_push_head(&lock->waiters, MyProcNumber, lwWaitLink); else proclist_push_tail(&lock->waiters, MyProcNumber, lwWaitLink); /* Can release the mutex now */ LWLockWaitListUnlock(lock); #ifdef LOCK_DEBUG pg_atomic_fetch_add_u32(&lock->nwaiters, 1); #endif } /* * Remove ourselves from the waitlist. * * This is used if we queued ourselves because we thought we needed to sleep * but, after further checking, we discovered that we don't actually need to * do so. */ static void LWLockDequeueSelf(LWLock *lock) { bool on_waitlist; #ifdef LWLOCK_STATS lwlock_stats *lwstats; lwstats = get_lwlock_stats_entry(lock); lwstats->dequeue_self_count++; #endif LWLockWaitListLock(lock); /* * Remove ourselves from the waitlist, unless we've already been removed. * The removal happens with the wait list lock held, so there's no race in * this check. */ on_waitlist = MyProc->lwWaiting == LW_WS_WAITING; if (on_waitlist) proclist_delete(&lock->waiters, MyProcNumber, lwWaitLink); if (proclist_is_empty(&lock->waiters) && (pg_atomic_read_u32(&lock->state) & LW_FLAG_HAS_WAITERS) != 0) { pg_atomic_fetch_and_u32(&lock->state, ~LW_FLAG_HAS_WAITERS); } /* XXX: combine with fetch_and above? */ LWLockWaitListUnlock(lock); /* clear waiting state again, nice for debugging */ if (on_waitlist) MyProc->lwWaiting = LW_WS_NOT_WAITING; else { int extraWaits = 0; /* * Somebody else dequeued us and has or will wake us up. Deal with the * superfluous absorption of a wakeup. */ /* * Reset RELEASE_OK flag if somebody woke us before we removed * ourselves - they'll have set it to false. */ pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_RELEASE_OK); /* * Now wait for the scheduled wakeup, otherwise our ->lwWaiting would * get reset at some inconvenient point later. Most of the time this * will immediately return. */ for (;;) { PGSemaphoreLock(MyProc->sem); if (MyProc->lwWaiting == LW_WS_NOT_WAITING) break; extraWaits++; } /* * Fix the process wait semaphore's count for any absorbed wakeups. */ while (extraWaits-- > 0) PGSemaphoreUnlock(MyProc->sem); } #ifdef LOCK_DEBUG { /* not waiting anymore */ uint32 nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&lock->nwaiters, 1); Assert(nwaiters < MAX_BACKENDS); } #endif } /* * LWLockAcquire - acquire a lightweight lock in the specified mode * * If the lock is not available, sleep until it is. Returns true if the lock * was available immediately, false if we had to sleep. * * Side effect: cancel/die interrupts are held off until lock release. */ bool LWLockAcquire(LWLock *lock, LWLockMode mode) { PGPROC *proc = MyProc; bool result = true; int extraWaits = 0; #ifdef LWLOCK_STATS lwlock_stats *lwstats; lwstats = get_lwlock_stats_entry(lock); #endif Assert(mode == LW_SHARED || mode == LW_EXCLUSIVE); PRINT_LWDEBUG("LWLockAcquire", lock, mode); #ifdef LWLOCK_STATS /* Count lock acquisition attempts */ if (mode == LW_EXCLUSIVE) lwstats->ex_acquire_count++; else lwstats->sh_acquire_count++; #endif /* LWLOCK_STATS */ /* * We can't wait if we haven't got a PGPROC. This should only occur * during bootstrap or shared memory initialization. Put an Assert here * to catch unsafe coding practices. */ Assert(!(proc == NULL && IsUnderPostmaster)); /* Ensure we will have room to remember the lock */ if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS) elog(ERROR, "too many LWLocks taken"); /* * Lock out cancel/die interrupts until we exit the code section protected * by the LWLock. This ensures that interrupts will not interfere with * manipulations of data structures in shared memory. */ HOLD_INTERRUPTS(); /* * Loop here to try to acquire lock after each time we are signaled by * LWLockRelease. * * NOTE: it might seem better to have LWLockRelease actually grant us the * lock, rather than retrying and possibly having to go back to sleep. But * in practice that is no good because it means a process swap for every * lock acquisition when two or more processes are contending for the same * lock. Since LWLocks are normally used to protect not-very-long * sections of computation, a process needs to be able to acquire and * release the same lock many times during a single CPU time slice, even * in the presence of contention. The efficiency of being able to do that * outweighs the inefficiency of sometimes wasting a process dispatch * cycle because the lock is not free when a released waiter finally gets * to run. See pgsql-hackers archives for 29-Dec-01. */ for (;;) { bool mustwait; /* * Try to grab the lock the first time, we're not in the waitqueue * yet/anymore. */ mustwait = LWLockAttemptLock(lock, mode); if (!mustwait) { LOG_LWDEBUG("LWLockAcquire", lock, "immediately acquired lock"); break; /* got the lock */ } /* * Ok, at this point we couldn't grab the lock on the first try. We * cannot simply queue ourselves to the end of the list and wait to be * woken up because by now the lock could long have been released. * Instead add us to the queue and try to grab the lock again. If we * succeed we need to revert the queuing and be happy, otherwise we * recheck the lock. If we still couldn't grab it, we know that the * other locker will see our queue entries when releasing since they * existed before we checked for the lock. */ /* add to the queue */ LWLockQueueSelf(lock, mode); /* we're now guaranteed to be woken up if necessary */ mustwait = LWLockAttemptLock(lock, mode); /* ok, grabbed the lock the second time round, need to undo queueing */ if (!mustwait) { LOG_LWDEBUG("LWLockAcquire", lock, "acquired, undoing queue"); LWLockDequeueSelf(lock); break; } /* * Wait until awakened. * * It is possible that we get awakened for a reason other than being * signaled by LWLockRelease. If so, loop back and wait again. Once * we've gotten the LWLock, re-increment the sema by the number of * additional signals received. */ LOG_LWDEBUG("LWLockAcquire", lock, "waiting"); #ifdef LWLOCK_STATS lwstats->block_count++; #endif LWLockReportWaitStart(lock); if (TRACE_POSTGRESQL_LWLOCK_WAIT_START_ENABLED()) TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(lock), mode); for (;;) { PGSemaphoreLock(proc->sem); if (proc->lwWaiting == LW_WS_NOT_WAITING) break; extraWaits++; } /* Retrying, allow LWLockRelease to release waiters again. */ pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_RELEASE_OK); #ifdef LOCK_DEBUG { /* not waiting anymore */ uint32 nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&lock->nwaiters, 1); Assert(nwaiters < MAX_BACKENDS); } #endif if (TRACE_POSTGRESQL_LWLOCK_WAIT_DONE_ENABLED()) TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(lock), mode); LWLockReportWaitEnd(); LOG_LWDEBUG("LWLockAcquire", lock, "awakened"); /* Now loop back and try to acquire lock again. */ result = false; } if (TRACE_POSTGRESQL_LWLOCK_ACQUIRE_ENABLED()) TRACE_POSTGRESQL_LWLOCK_ACQUIRE(T_NAME(lock), mode); /* Add lock to list of locks held by this backend */ held_lwlocks[num_held_lwlocks].lock = lock; held_lwlocks[num_held_lwlocks++].mode = mode; /* * Fix the process wait semaphore's count for any absorbed wakeups. */ while (extraWaits-- > 0) PGSemaphoreUnlock(proc->sem); return result; } /* * LWLockConditionalAcquire - acquire a lightweight lock in the specified mode * * If the lock is not available, return false with no side-effects. * * If successful, cancel/die interrupts are held off until lock release. */ bool LWLockConditionalAcquire(LWLock *lock, LWLockMode mode) { bool mustwait; Assert(mode == LW_SHARED || mode == LW_EXCLUSIVE); PRINT_LWDEBUG("LWLockConditionalAcquire", lock, mode); /* Ensure we will have room to remember the lock */ if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS) elog(ERROR, "too many LWLocks taken"); /* * Lock out cancel/die interrupts until we exit the code section protected * by the LWLock. This ensures that interrupts will not interfere with * manipulations of data structures in shared memory. */ HOLD_INTERRUPTS(); /* Check for the lock */ mustwait = LWLockAttemptLock(lock, mode); if (mustwait) { /* Failed to get lock, so release interrupt holdoff */ RESUME_INTERRUPTS(); LOG_LWDEBUG("LWLockConditionalAcquire", lock, "failed"); if (TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE_FAIL_ENABLED()) TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE_FAIL(T_NAME(lock), mode); } else { /* Add lock to list of locks held by this backend */ held_lwlocks[num_held_lwlocks].lock = lock; held_lwlocks[num_held_lwlocks++].mode = mode; if (TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE_ENABLED()) TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE(T_NAME(lock), mode); } return !mustwait; } /* * LWLockAcquireOrWait - Acquire lock, or wait until it's free * * The semantics of this function are a bit funky. If the lock is currently * free, it is acquired in the given mode, and the function returns true. If * the lock isn't immediately free, the function waits until it is released * and returns false, but does not acquire the lock. * * This is currently used for WALWriteLock: when a backend flushes the WAL, * holding WALWriteLock, it can flush the commit records of many other * backends as a side-effect. Those other backends need to wait until the * flush finishes, but don't need to acquire the lock anymore. They can just * wake up, observe that their records have already been flushed, and return. */ bool LWLockAcquireOrWait(LWLock *lock, LWLockMode mode) { PGPROC *proc = MyProc; bool mustwait; int extraWaits = 0; #ifdef LWLOCK_STATS lwlock_stats *lwstats; lwstats = get_lwlock_stats_entry(lock); #endif Assert(mode == LW_SHARED || mode == LW_EXCLUSIVE); PRINT_LWDEBUG("LWLockAcquireOrWait", lock, mode); /* Ensure we will have room to remember the lock */ if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS) elog(ERROR, "too many LWLocks taken"); /* * Lock out cancel/die interrupts until we exit the code section protected * by the LWLock. This ensures that interrupts will not interfere with * manipulations of data structures in shared memory. */ HOLD_INTERRUPTS(); /* * NB: We're using nearly the same twice-in-a-row lock acquisition * protocol as LWLockAcquire(). Check its comments for details. */ mustwait = LWLockAttemptLock(lock, mode); if (mustwait) { LWLockQueueSelf(lock, LW_WAIT_UNTIL_FREE); mustwait = LWLockAttemptLock(lock, mode); if (mustwait) { /* * Wait until awakened. Like in LWLockAcquire, be prepared for * bogus wakeups. */ LOG_LWDEBUG("LWLockAcquireOrWait", lock, "waiting"); #ifdef LWLOCK_STATS lwstats->block_count++; #endif LWLockReportWaitStart(lock); if (TRACE_POSTGRESQL_LWLOCK_WAIT_START_ENABLED()) TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(lock), mode); for (;;) { PGSemaphoreLock(proc->sem); if (proc->lwWaiting == LW_WS_NOT_WAITING) break; extraWaits++; } #ifdef LOCK_DEBUG { /* not waiting anymore */ uint32 nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&lock->nwaiters, 1); Assert(nwaiters < MAX_BACKENDS); } #endif if (TRACE_POSTGRESQL_LWLOCK_WAIT_DONE_ENABLED()) TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(lock), mode); LWLockReportWaitEnd(); LOG_LWDEBUG("LWLockAcquireOrWait", lock, "awakened"); } else { LOG_LWDEBUG("LWLockAcquireOrWait", lock, "acquired, undoing queue"); /* * Got lock in the second attempt, undo queueing. We need to treat * this as having successfully acquired the lock, otherwise we'd * not necessarily wake up people we've prevented from acquiring * the lock. */ LWLockDequeueSelf(lock); } } /* * Fix the process wait semaphore's count for any absorbed wakeups. */ while (extraWaits-- > 0) PGSemaphoreUnlock(proc->sem); if (mustwait) { /* Failed to get lock, so release interrupt holdoff */ RESUME_INTERRUPTS(); LOG_LWDEBUG("LWLockAcquireOrWait", lock, "failed"); if (TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT_FAIL_ENABLED()) TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT_FAIL(T_NAME(lock), mode); } else { LOG_LWDEBUG("LWLockAcquireOrWait", lock, "succeeded"); /* Add lock to list of locks held by this backend */ held_lwlocks[num_held_lwlocks].lock = lock; held_lwlocks[num_held_lwlocks++].mode = mode; if (TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT_ENABLED()) TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT(T_NAME(lock), mode); } return !mustwait; } /* * Does the lwlock in its current state need to wait for the variable value to * change? * * If we don't need to wait, and it's because the value of the variable has * changed, store the current value in newval. * * *result is set to true if the lock was free, and false otherwise. */ static bool LWLockConflictsWithVar(LWLock *lock, pg_atomic_uint64 *valptr, uint64 oldval, uint64 *newval, bool *result) { bool mustwait; uint64 value; /* * Test first to see if it the slot is free right now. * * XXX: the unique caller of this routine, WaitXLogInsertionsToFinish() * via LWLockWaitForVar(), uses an implied barrier with a spinlock before * this, so we don't need a memory barrier here as far as the current * usage is concerned. But that might not be safe in general. */ mustwait = (pg_atomic_read_u32(&lock->state) & LW_VAL_EXCLUSIVE) != 0; if (!mustwait) { *result = true; return false; } *result = false; /* * Reading this value atomically is safe even on platforms where uint64 * cannot be read without observing a torn value. */ value = pg_atomic_read_u64(valptr); if (value != oldval) { mustwait = false; *newval = value; } else { mustwait = true; } return mustwait; } /* * LWLockWaitForVar - Wait until lock is free, or a variable is updated. * * If the lock is held and *valptr equals oldval, waits until the lock is * either freed, or the lock holder updates *valptr by calling * LWLockUpdateVar. If the lock is free on exit (immediately or after * waiting), returns true. If the lock is still held, but *valptr no longer * matches oldval, returns false and sets *newval to the current value in * *valptr. * * Note: this function ignores shared lock holders; if the lock is held * in shared mode, returns 'true'. * * Be aware that LWLockConflictsWithVar() does not include a memory barrier, * hence the caller of this function may want to rely on an explicit barrier or * an implied barrier via spinlock or LWLock to avoid memory ordering issues. */ bool LWLockWaitForVar(LWLock *lock, pg_atomic_uint64 *valptr, uint64 oldval, uint64 *newval) { PGPROC *proc = MyProc; int extraWaits = 0; bool result = false; #ifdef LWLOCK_STATS lwlock_stats *lwstats; lwstats = get_lwlock_stats_entry(lock); #endif PRINT_LWDEBUG("LWLockWaitForVar", lock, LW_WAIT_UNTIL_FREE); /* * Lock out cancel/die interrupts while we sleep on the lock. There is no * cleanup mechanism to remove us from the wait queue if we got * interrupted. */ HOLD_INTERRUPTS(); /* * Loop here to check the lock's status after each time we are signaled. */ for (;;) { bool mustwait; mustwait = LWLockConflictsWithVar(lock, valptr, oldval, newval, &result); if (!mustwait) break; /* the lock was free or value didn't match */ /* * Add myself to wait queue. Note that this is racy, somebody else * could wakeup before we're finished queuing. NB: We're using nearly * the same twice-in-a-row lock acquisition protocol as * LWLockAcquire(). Check its comments for details. The only * difference is that we also have to check the variable's values when * checking the state of the lock. */ LWLockQueueSelf(lock, LW_WAIT_UNTIL_FREE); /* * Set RELEASE_OK flag, to make sure we get woken up as soon as the * lock is released. */ pg_atomic_fetch_or_u32(&lock->state, LW_FLAG_RELEASE_OK); /* * We're now guaranteed to be woken up if necessary. Recheck the lock * and variables state. */ mustwait = LWLockConflictsWithVar(lock, valptr, oldval, newval, &result); /* Ok, no conflict after we queued ourselves. Undo queueing. */ if (!mustwait) { LOG_LWDEBUG("LWLockWaitForVar", lock, "free, undoing queue"); LWLockDequeueSelf(lock); break; } /* * Wait until awakened. * * It is possible that we get awakened for a reason other than being * signaled by LWLockRelease. If so, loop back and wait again. Once * we've gotten the LWLock, re-increment the sema by the number of * additional signals received. */ LOG_LWDEBUG("LWLockWaitForVar", lock, "waiting"); #ifdef LWLOCK_STATS lwstats->block_count++; #endif LWLockReportWaitStart(lock); if (TRACE_POSTGRESQL_LWLOCK_WAIT_START_ENABLED()) TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(lock), LW_EXCLUSIVE); for (;;) { PGSemaphoreLock(proc->sem); if (proc->lwWaiting == LW_WS_NOT_WAITING) break; extraWaits++; } #ifdef LOCK_DEBUG { /* not waiting anymore */ uint32 nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&lock->nwaiters, 1); Assert(nwaiters < MAX_BACKENDS); } #endif if (TRACE_POSTGRESQL_LWLOCK_WAIT_DONE_ENABLED()) TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(lock), LW_EXCLUSIVE); LWLockReportWaitEnd(); LOG_LWDEBUG("LWLockWaitForVar", lock, "awakened"); /* Now loop back and check the status of the lock again. */ } /* * Fix the process wait semaphore's count for any absorbed wakeups. */ while (extraWaits-- > 0) PGSemaphoreUnlock(proc->sem); /* * Now okay to allow cancel/die interrupts. */ RESUME_INTERRUPTS(); return result; } /* * LWLockUpdateVar - Update a variable and wake up waiters atomically * * Sets *valptr to 'val', and wakes up all processes waiting for us with * LWLockWaitForVar(). It first sets the value atomically and then wakes up * waiting processes so that any process calling LWLockWaitForVar() on the same * lock is guaranteed to see the new value, and act accordingly. * * The caller must be holding the lock in exclusive mode. */ void LWLockUpdateVar(LWLock *lock, pg_atomic_uint64 *valptr, uint64 val) { proclist_head wakeup; proclist_mutable_iter iter; PRINT_LWDEBUG("LWLockUpdateVar", lock, LW_EXCLUSIVE); /* * Note that pg_atomic_exchange_u64 is a full barrier, so we're guaranteed * that the variable is updated before waking up waiters. */ pg_atomic_exchange_u64(valptr, val); proclist_init(&wakeup); LWLockWaitListLock(lock); Assert(pg_atomic_read_u32(&lock->state) & LW_VAL_EXCLUSIVE); /* * 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. */ proclist_foreach_modify(iter, &lock->waiters, lwWaitLink) { PGPROC *waiter = GetPGProcByNumber(iter.cur); if (waiter->lwWaitMode != LW_WAIT_UNTIL_FREE) break; proclist_delete(&lock->waiters, iter.cur, lwWaitLink); proclist_push_tail(&wakeup, iter.cur, lwWaitLink); /* see LWLockWakeup() */ Assert(waiter->lwWaiting == LW_WS_WAITING); waiter->lwWaiting = LW_WS_PENDING_WAKEUP; } /* We are done updating shared state of the lock itself. */ LWLockWaitListUnlock(lock); /* * Awaken any waiters I removed from the queue. */ proclist_foreach_modify(iter, &wakeup, lwWaitLink) { PGPROC *waiter = GetPGProcByNumber(iter.cur); proclist_delete(&wakeup, iter.cur, lwWaitLink); /* check comment in LWLockWakeup() about this barrier */ pg_write_barrier(); waiter->lwWaiting = LW_WS_NOT_WAITING; PGSemaphoreUnlock(waiter->sem); } } /* * Stop treating lock as held by current backend. * * This is the code that can be shared between actually releasing a lock * (LWLockRelease()) and just not tracking ownership of the lock anymore * without releasing the lock (LWLockDisown()). * * Returns the mode in which the lock was held by the current backend. * * NB: This does not call RESUME_INTERRUPTS(), but leaves that responsibility * of the caller. * * NB: This will leave lock->owner pointing to the current backend (if * LOCK_DEBUG is set). This is somewhat intentional, as it makes it easier to * debug cases of missing wakeups during lock release. */ static inline LWLockMode LWLockDisownInternal(LWLock *lock) { LWLockMode mode; int i; /* * Remove lock from list of locks held. Usually, but not always, it will * be the latest-acquired lock; so search array backwards. */ for (i = num_held_lwlocks; --i >= 0;) if (lock == held_lwlocks[i].lock) break; if (i < 0) elog(ERROR, "lock %s is not held", T_NAME(lock)); mode = held_lwlocks[i].mode; num_held_lwlocks--; for (; i < num_held_lwlocks; i++) held_lwlocks[i] = held_lwlocks[i + 1]; return mode; } /* * Helper function to release lock, shared between LWLockRelease() and * LWLockReleaseDisowned(). */ static void LWLockReleaseInternal(LWLock *lock, LWLockMode mode) { uint32 oldstate; bool check_waiters; /* * Release my hold on lock, after that it can immediately be acquired by * others, even if we still have to wakeup other waiters. */ if (mode == LW_EXCLUSIVE) oldstate = pg_atomic_sub_fetch_u32(&lock->state, LW_VAL_EXCLUSIVE); else oldstate = pg_atomic_sub_fetch_u32(&lock->state, LW_VAL_SHARED); /* nobody else can have that kind of lock */ Assert(!(oldstate & LW_VAL_EXCLUSIVE)); if (TRACE_POSTGRESQL_LWLOCK_RELEASE_ENABLED()) TRACE_POSTGRESQL_LWLOCK_RELEASE(T_NAME(lock)); /* * We're still waiting for backends to get scheduled, don't wake them up * again. */ if ((oldstate & (LW_FLAG_HAS_WAITERS | LW_FLAG_RELEASE_OK)) == (LW_FLAG_HAS_WAITERS | LW_FLAG_RELEASE_OK) && (oldstate & LW_LOCK_MASK) == 0) check_waiters = true; else check_waiters = false; /* * As waking up waiters requires the spinlock to be acquired, only do so * if necessary. */ if (check_waiters) { /* XXX: remove before commit? */ LOG_LWDEBUG("LWLockRelease", lock, "releasing waiters"); LWLockWakeup(lock); } } /* * Stop treating lock as held by current backend. * * After calling this function it's the callers responsibility to ensure that * the lock gets released (via LWLockReleaseDisowned()), even in case of an * error. This only is desirable if the lock is going to be released in a * different process than the process that acquired it. */ void LWLockDisown(LWLock *lock) { LWLockDisownInternal(lock); RESUME_INTERRUPTS(); } /* * LWLockRelease - release a previously acquired lock */ void LWLockRelease(LWLock *lock) { LWLockMode mode; mode = LWLockDisownInternal(lock); PRINT_LWDEBUG("LWLockRelease", lock, mode); LWLockReleaseInternal(lock, mode); /* * Now okay to allow cancel/die interrupts. */ RESUME_INTERRUPTS(); } /* * Release lock previously disowned with LWLockDisown(). */ void LWLockReleaseDisowned(LWLock *lock, LWLockMode mode) { LWLockReleaseInternal(lock, mode); } /* * LWLockReleaseClearVar - release a previously acquired lock, reset variable */ void LWLockReleaseClearVar(LWLock *lock, pg_atomic_uint64 *valptr, uint64 val) { /* * Note that pg_atomic_exchange_u64 is a full barrier, so we're guaranteed * that the variable is updated before releasing the lock. */ pg_atomic_exchange_u64(valptr, val); LWLockRelease(lock); } /* * LWLockReleaseAll - release all currently-held locks * * Used to clean up after ereport(ERROR). An important difference between this * function and retail LWLockRelease calls is that InterruptHoldoffCount is * unchanged by this operation. This is necessary since InterruptHoldoffCount * has been set to an appropriate level earlier in error recovery. We could * decrement it below zero if we allow it to drop for each released lock! */ void LWLockReleaseAll(void) { while (num_held_lwlocks > 0) { HOLD_INTERRUPTS(); /* match the upcoming RESUME_INTERRUPTS */ LWLockRelease(held_lwlocks[num_held_lwlocks - 1].lock); } } /* * ForEachLWLockHeldByMe - run a callback for each held lock * * This is meant as debug support only. */ void ForEachLWLockHeldByMe(void (*callback) (LWLock *, LWLockMode, void *), void *context) { int i; for (i = 0; i < num_held_lwlocks; i++) callback(held_lwlocks[i].lock, held_lwlocks[i].mode, context); } /* * LWLockHeldByMe - test whether my process holds a lock in any mode * * This is meant as debug support only. */ bool LWLockHeldByMe(LWLock *lock) { int i; for (i = 0; i < num_held_lwlocks; i++) { if (held_lwlocks[i].lock == lock) return true; } return false; } /* * LWLockAnyHeldByMe - test whether my process holds any of an array of locks * * This is meant as debug support only. */ bool LWLockAnyHeldByMe(LWLock *lock, int nlocks, size_t stride) { char *held_lock_addr; char *begin; char *end; int i; begin = (char *) lock; end = begin + nlocks * stride; for (i = 0; i < num_held_lwlocks; i++) { held_lock_addr = (char *) held_lwlocks[i].lock; if (held_lock_addr >= begin && held_lock_addr < end && (held_lock_addr - begin) % stride == 0) return true; } return false; } /* * LWLockHeldByMeInMode - test whether my process holds a lock in given mode * * This is meant as debug support only. */ bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode) { int i; for (i = 0; i < num_held_lwlocks; i++) { if (held_lwlocks[i].lock == lock && held_lwlocks[i].mode == mode) return true; } return false; }