From 5d105c937338c848e60783ae3a9f0ee5b81070e5 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Fri, 20 Feb 2026 16:14:09 +0200 Subject: [PATCH v12a 4/4] Introduce PendingInterrupts with separate 'flags' field and "attention" This does a couple of things: - Split the interrupts mask into two 32-bit fields on PG_HAVE_ATOMIC_U64_SIMULATION systems. This fixes the self-deadlock risk when used from signal handlers. - Introduce an "attention mask" field, where the backend can advertise the interrupts it is currently interested in, and a separate "flags" field. We no longer need to steal bits from the interrupt mask for the flags. Whenever a backend sends an interrupt, it checks the attention mask and only wakes up the backend if the interrupt is in the attention mask. When not sleeping, the same mechanism is used to set a flag for CHECK_FOR_INTERRUPTS(), telling the next CHECK_FOR_INTERRUPTS() that it has some work to do. By moving that work of checking the attention mask to the sender, CHECK_FOR_INTERRUPTS() needs fewer instructions. --- src/include/ipc/interrupt.h | 387 +++++++----------- .../{interrupt.h => standard_interrupts.h} | 192 +-------- src/include/storage/proc.h | 2 +- src/backend/access/spgist/spgdoinsert.c | 9 +- src/backend/access/transam/parallel.c | 2 +- src/backend/ipc/interrupt.c | 335 +++++++++++---- src/backend/storage/ipc/ipc.c | 9 +- src/backend/storage/ipc/waiteventset.c | 58 ++- src/backend/tcop/postgres.c | 5 +- src/backend/utils/error/elog.c | 4 +- src/tools/pgindent/typedefs.list | 1 + 11 files changed, 459 insertions(+), 545 deletions(-) copy src/include/ipc/{interrupt.h => standard_interrupts.h} (53%) diff --git a/src/include/ipc/interrupt.h b/src/include/ipc/interrupt.h index 239d58f7597..10234d4f987 100644 --- a/src/include/ipc/interrupt.h +++ b/src/include/ipc/interrupt.h @@ -15,6 +15,7 @@ #ifndef IPC_INTERRUPT_H #define IPC_INTERRUPT_H +#include "ipc/standard_interrupts.h" #include "port/atomics.h" #include "storage/procnumber.h" @@ -23,212 +24,128 @@ * needed which are needed by all callers of WaitInterrupt, so include it * here. * - * Note: InterruptMask is defind in waiteventset.h to avoid circular dependency + * Note: InterruptMask is defined in waiteventset.h to avoid circular dependency */ #include "storage/waiteventset.h" - /* - * Flags in the pending interrupts bitmask. Each value is a different bit, so that - * these can be conveniently OR'd together. - */ -#define UINT64_BIT(shift) (UINT64_C(1) << (shift)) - -/*********************************************************************** - * Begin definitions of built-in interrupt bits - ***********************************************************************/ - -/* - * INTERRUPT_WAIT_WAKEUP is shared by many use cases that need to wake up - * a process, which don't need a dedicated interrupt bit. - */ -#define INTERRUPT_WAIT_WAKEUP UINT64_BIT(0) - - -/*********************************************************************** - * Standard interrupts handled the same by most processes + * PendingInterrupts is used to receive and wait for interrupts. The + * 'interrupts' field is a bitmask representing interrupts that are currently + * pending for the process. * - * Most of these are normally processed by CHECK_FOR_INTERRUPTS() once - * process startup has reached SetStandardInterrupts(). - ***********************************************************************/ - -/* - * Backend has been requested to terminate gracefully. + * We support up to 64 different interrupts. That way, an interrupt mask can + * be conveniently stored as one 64-bit atomic integer, on systems with 64-bit + * atomics. On other systems, it's split into two 32-bit atomic fields, which + * is good enough because we don't rely on atomicity between different + * interrupt bits. (Note that the 64-bit atomics simulation relies on + * spinlocks, which creates a deadlock risk when used from signal handlers, so + * we cannot rely on the simulated 64-bit atomics.) * - * This is raised by the SIGTERM signal handler, or can be sent directly by - * another backend e.g. with pg_terminate_backend(). - */ -#define INTERRUPT_TERMINATE UINT64_BIT(1) - -/* - * Cancel current query, if any. + * Attention mechanism + * ------------------- * - * Sent to regular backends by pg_cancel_backend(), SIGINT, or in response to - * a query cancellation packet. Some other processes like autovacuum workers - * and logical decoding processes also react to this. - */ -#define INTERRUPT_QUERY_CANCEL UINT64_BIT(2) - -/* - * Recovery conflict. This is sent by the startup process in hot standby mode - * when a backend holds back the WAL replay for too long. The reason for the - * conflict indicated by the PGPROC->pendingRecoveryConflicts - * bitmask. Conflicts are generally resolved by terminating the current query - * or session. The exact reaction depends on the reason and what state the - * backend is in. - */ -#define INTERRUPT_RECOVERY_CONFLICT UINT64_BIT(3) - -/* - * Config file reload is requested. + * The 'attention_mask' field lets a backend advertise which interrupts it is + * currently interested in. When a backend is sleeping, waiting for an + * interrupt to arrive, it sets the bits for the waited-for interrupts in + * 'attention_mask'. At other times, the 'attention_mask' equals + * EnabledInterruptsMask, i.e. the interrupts that can be processed by a + * CHECK_FOR_INTERRUPTS(). * - * This is normally disabled and therefore not handled at - * CHECK_FOR_INTERRUPTS(). The "main loop" in each process is expected to - * check for it explicitly. - */ -#define INTERRUPT_CONFIG_RELOAD UINT64_BIT(4) - -/* - * Log current memory contexts, sent by pg_log_backend_memory_contexts() - */ -#define INTERRUPT_LOG_MEMORY_CONTEXT UINT64_BIT(5) - -/* - * procsignal global barrier interrupt - */ -#define INTERRUPT_BARRIER UINT64_BIT(6) - - -/*********************************************************************** - * Interrupts used by client backends and most other processes that - * connect to a particular database. + * When a backend sets the interrupt bit of another backend (or the same + * backend), it also checks if that interrupt is in the target's + * 'attention_mask'. If so, it sets the ATTENTION flag. Furthermore, if the + * target backend is currently sleeping, i.e. if the SLEEPING is set, it also + * wakes it up. * - * Most of these are also processed by CHECK_FOR_INTERRUPTS() once process - * startup has reached SetStandardInterrupts(). - ***********************************************************************/ - -/* Raised by timers */ -#define INTERRUPT_TRANSACTION_TIMEOUT UINT64_BIT(7) -#define INTERRUPT_IDLE_SESSION_TIMEOUT UINT64_BIT(8) -#define INTERRUPT_IDLE_IN_TRANSACTION_SESSION_TIMEOUT UINT64_BIT(9) -#define INTERRUPT_CLIENT_CHECK_TIMEOUT UINT64_BIT(10) - -/* Raised by timer while idle, to send a stats update */ -#define INTERRUPT_IDLE_STATS_TIMEOUT UINT64_BIT(11) - -/* Raised synchronously when the client connection is lost */ -#define INTERRUPT_CLIENT_CONNECTION_LOST UINT64_BIT(12) - -/* - * INTERRUPT_ASYNC_NOTIFY is sent to notify backends that have registered to - * LISTEN on any channels that they might have messages they need to deliver - * to the frontend. It is also processed whenever starting to read from the - * client or while doing so, but only when there is no transaction in - * progress. - */ -#define INTERRUPT_ASYNC_NOTIFY UINT64_BIT(13) - -/* - * Because backends sitting idle will not be reading sinval events, we need a - * way to give an idle backend a swift kick in the rear and make it catch up - * before the sinval queue overflows and forces it to go through a cache reset - * exercise. This is done by sending INTERRUPT_SINVAL_CATCHUP to any backend - * that gets too far behind. + * When not sleeping, the ATTENTION flag is used as a quick check in + * CHECK_FOR_INTERRUPTS() for whether any interrupts need to be processed. + * Checking a single flag requires fewer instructions than checking the + * interrupt bits against EnabledInterruptsMask; the attention mechanism + * shifts that work to the sending backend. * - * The interrupt is processed whenever starting to read from the client, or - * when interrupted while doing so. + * There are race conditions in how the ATTENTION flag is set. If a backend + * clears a bit from its 'attention_mask', and another backend is concurrently + * sending that interrupt, it's possible that the ATTENTION flag gets set or + * the process is woken up after the 'attention_mask' has already been + * cleared. The system tolerates spuriously set ATTENTION flag and wakeups, + * so that's OK. The operations are ordered so that the opposite is not + * possible: if you set a bit in the 'attention_mask' and then check that the + * bit is not set in the 'interrupts' mask, you are guaranteed to receive the + * attention flag or a wakeup if the interrupt is set later. */ -#define INTERRUPT_SINVAL_CATCHUP UINT64_BIT(14) - -/* Message from a cooperating parallel backend or apply worker */ -#define INTERRUPT_PARALLEL_MESSAGE UINT64_BIT(15) - - -/*********************************************************************** - * Process-specific interrupts - * - * Some processes need dedicated interrupts for various purposes. Ignored - * by other processes. - ***********************************************************************/ - -/* ask walsenders to prepare for shutdown */ -#define INTERRUPT_WALSND_INIT_STOPPING UINT64_BIT(16) - -/* TODO: document the difference with INTERRUPT_WALSND_INIT_STOPPING */ -#define INTERRUPT_WALSND_STOP UINT64_BIT(17) +typedef struct +{ + pg_atomic_uint32 flags; /* PI_FLAG_* */ + +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + pg_atomic_uint64 interrupts; /* pending interrupts */ + pg_atomic_uint64 attention_mask; /* interrupts that set the ATTENTION + * flag */ +#else + pg_atomic_uint32 interrupts_lo; + pg_atomic_uint32 interrupts_hi; + pg_atomic_uint32 attention_mask_lo; + pg_atomic_uint32 attention_mask_hi; +#endif +} PendingInterrupts; + +#define PI_FLAG_ATTENTION 0x01 +#define PI_FLAG_SLEEPING 0x02 /* - * INTERRUPT_WAL_ARRIVED is used to wake up the startup process, to tell - * it that it should continue WAL replay. It's sent by WAL receiver when - * more WAL arrives, or when promotion is requested. + * Interrupt vector currently in use for this process. Most of the time this + * points to MyProc->pendingInterrupts, but in processes that have no PGPROC + * entry (yet), it points to a process-private variable, so that interrupts + * can nevertheless be used from signal handlers in the same process. */ -#define INTERRUPT_WAL_ARRIVED UINT64_BIT(18) - -/* Wake up startup process to check for the promotion signal file */ -#define INTERRUPT_CHECK_PROMOTE UINT64_BIT(19) - -/* sent to logical replication launcher, when a subscription changes */ -#define INTERRUPT_SUBSCRIPTION_CHANGE UINT64_BIT(20) - -/* Graceful shutdown request for a parallel apply worker */ -#define INTERRUPT_SHUTDOWN_PARALLEL_APPLY_WORKER UINT64_BIT(21) - -/* Request checkpointer to perform one last checkpoint, then shut down. */ -#define INTERRUPT_SHUTDOWN_XLOG UINT64_BIT(22) - -#define INTERRUPT_SHUTDOWN_PGARCH UINT64_BIT(23) +extern PGDLLIMPORT PendingInterrupts *MyPendingInterrupts; /* - * This is sent to the autovacuum launcher when an autovacuum worker exits - */ -#define INTERRUPT_AUTOVACUUM_WORKER_FINISHED UINT64_BIT(24) - - -/*********************************************************************** - * End of built-in interrupt bits + * Test if an interrupt is pending * - * The remaining bits are handed out by RequestAddinInterrupt, for - * extensions - ***********************************************************************/ -#define BEGIN_ADDIN_INTERRUPTS 25 -#define END_ADDIN_INTERRUPTS 63 - -/* - * SLEEPING_ON_INTERRUPTS indicates that the backend is currently blocked - * waiting for an interrupt. If set, the backend needs to be woken up when a - * bit in the pending interrupts mask is set. It's used internally by the - * interrupt machinery, and cannot be used directly in the public functions. - * It's named differently to distinguish it from the actual interrupt flags. - */ -#define SLEEPING_ON_INTERRUPTS UINT64_BIT(63) - -extern PGDLLIMPORT pg_atomic_uint64 *MyPendingInterrupts; - -/* - * Test an interrupt flag (or flags). + * If 'interruptMask' has multiple bits set, returns true if any of them are + * pending. */ static inline bool InterruptPending(InterruptMask interruptMask) { /* - * Note that there is no memory barrier here. This is used in - * CHECK_FOR_INTERRUPTS(), so we want this to be as cheap as possible. - * - * That means that if the interrupt is concurrently set by another - * process, we might miss it. That should be OK, because the next - * WaitInterrupt() or equivalent call acts as a synchronization barrier. - * We will see the updated value before sleeping. + * Note that there is no memory barrier here, because we want this to be + * as cheap as possible. That means that if the interrupt is concurrently + * set by another process, we might miss it. That should be OK, because + * the next WaitInterrupt() or equivalent call acts as a synchronization + * barrier; we will see the updated value before sleeping. */ - return (pg_atomic_read_u64(MyPendingInterrupts) & interruptMask) != 0; + uint64 pending; + +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + pending = pg_atomic_read_u64(&MyPendingInterrupts->interrupts); +#else + pending = (uint64) pg_atomic_read_u32(&MyPendingInterrupts->interrupts_lo); + pending |= (uint64) pg_atomic_read_u32(&MyPendingInterrupts->interrupts_hi) << 32; +#endif + + return (pending & interruptMask) != 0; } /* * Clear an interrupt flag (or flags). + * + * Note that this does not clear the ATTENTION flag, so if it was already set, + * the next CHECK_FOR_INTERRUPTS() will make an unnecessary but harmless + * ProcessInterrupts() call. */ static inline void ClearInterrupt(InterruptMask interruptMask) { - pg_atomic_fetch_and_u64(MyPendingInterrupts, ~interruptMask); + uint64 mask = ~interruptMask; + +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + (void) pg_atomic_fetch_and_u64(&MyPendingInterrupts->interrupts, mask); +#else + (void) pg_atomic_fetch_and_u32(&MyPendingInterrupts->interrupts_lo, (uint32) mask); + (void) pg_atomic_fetch_and_u32(&MyPendingInterrupts->interrupts_hi, (uint32) (mask >> 32)); +#endif } /* @@ -254,106 +171,110 @@ extern void SwitchToLocalInterrupts(void); extern void SwitchToSharedInterrupts(void); extern void InitializeInterruptWaitSet(void); -typedef void (*pg_interrupt_handler_t) (void); -extern void SetInterruptHandler(InterruptMask interruptMask, pg_interrupt_handler_t handler); - -extern void EnableInterrupt(InterruptMask interruptMask); -extern void DisableInterrupt(InterruptMask interruptMask); - -/* for extensions */ -extern InterruptMask RequestAddinInterrupt(void); - -/* Standard interrupt handlers. Defined in tcop/postgres.c */ -extern void SetStandardInterruptHandlers(void); - -extern void ProcessQueryCancelInterrupt(void); -extern void ProcessTerminateInterrupt(void); -extern void ProcessConfigReloadInterrupt(void); -extern void ProcessAsyncNotifyInterrupt(void); -extern void ProcessIdleStatsTimeoutInterrupt(void); -extern void ProcessRecoveryConflictInterrupts(void); -extern void ProcessTransactionTimeoutInterrupt(void); -extern void ProcessIdleSessionTimeoutInterrupt(void); -extern void ProcessIdleInTransactionSessionTimeoutInterrupt(void); -extern void ProcessClientCheckTimeoutInterrupt(void); -extern void ProcessClientConnectionLost(void); - -extern void ProcessAuxProcessShutdownInterrupt(void); - /***************************************************************************** * CHECK_FOR_INTERRUPTS() and friends *****************************************************************************/ +/* Interrupts currently enabled for CHECK_FOR_INTERRUPTS() processing */ +extern PGDLLIMPORT InterruptMask EnabledInterruptsMask; -extern PGDLLIMPORT volatile InterruptMask EnabledInterruptsMask; -extern PGDLLIMPORT volatile InterruptMask CheckForInterruptsMask; +/* + * Pointer to MyPendingInterrupts->flags, except when interrupt holdoff or a + * critical section prevents interrupts processing, in which case this points + * to a dummy all-zeros variable (ZeroPendingInterruptsFlags) instead. This + * allows CHECK_FOR_INTERRUPTS() to follow just this one pointer, and not have + * to check the holdoff counts separately. + */ +extern PGDLLIMPORT pg_atomic_uint32 *MyPendingInterruptsFlags; -extern void ProcessInterrupts(void); - -/* Test whether an interrupt is pending */ +/* + * Check whether any enabled interrupt is pending, without trying to service + * it immediately. This is can be used in a HOLD_INTERRUPTS() block to check + * if the HOLD_INTERRUPTS() is delaying the interrupt processing. + */ #ifndef WIN32 -#define INTERRUPTS_PENDING_CONDITION(mask) \ - (unlikely(InterruptPending(mask))) +#define INTERRUPTS_PENDING_CONDITION() \ + (unlikely(InterruptPending(EnabledInterruptsMask))) #else -#define INTERRUPTS_PENDING_CONDITION(mask) \ +#define INTERRUPTS_PENDING_CONDITION() \ (unlikely(UNBLOCKED_SIGNAL_QUEUE()) ? \ pgwin32_dispatch_queued_signals() : (void) 0, \ - unlikely(InterruptPending(mask))) + unlikely(InterruptPending(EnabledInterruptsMask))) #endif /* - * Is ProcessInterrupts() guaranteed to clear all the bits in 'mask'? + * Can interrupts be processed in the current state, i.e. are the interrupts + * not prevented by the HOLD_INTERRUPTS() or a critical section critical + * section? + */ +#define INTERRUPTS_CAN_BE_PROCESSED() \ + (InterruptHoldoffCount == 0 && CritSectionCount == 0) + +/* + * Interrupts that would be processed by CHECK_FOR_INTERRUPTS(). This is + * equal to EnabledInterruptsMask, except when interrupts are held off by + * HOLD/RESUME_INTERRUPTS() or a critical section. + */ +#define CheckForInterruptsMask \ + (INTERRUPTS_CAN_BE_PROCESSED() ? EnabledInterruptsMask : 0) + +/* + * Service an interrupt, if one is pending and it's safe to service it now. * - * (The interrupt handler may re-raise the interrupt, though) + * NB: This is called from all over the codebase, and in fairly tight loops, + * so this needs to be very short and fast when there is no work to do! */ -#define INTERRUPTS_CAN_BE_PROCESSED(mask) \ - (((mask) & CheckForInterruptsMask) == (mask)) - -/* Service interrupt, if one is pending and it's safe to service it now */ #define CHECK_FOR_INTERRUPTS() \ do { \ - if (INTERRUPTS_PENDING_CONDITION(CheckForInterruptsMask)) \ + if (unlikely(pg_atomic_read_u32(MyPendingInterruptsFlags) != 0)) \ ProcessInterrupts(); \ } while(0) +typedef void (*pg_interrupt_handler_t) (void); +extern void SetInterruptHandler(InterruptMask interruptMask, pg_interrupt_handler_t handler); + +extern void EnableInterrupt(InterruptMask interruptMask); +extern void DisableInterrupt(InterruptMask interruptMask); + +extern void ProcessInterrupts(void); +extern void SetInterruptAttentionMask(InterruptMask mask); /***************************************************************************** * Critical section and interrupt holdoff mechanism *****************************************************************************/ -/* these are marked volatile because they are examined by signal handlers: */ -/* - * XXX: is that still true? Should we use local vars to avoid repeated access - * e.g. inside RESUME_INTERRUPTS() ? - */ -extern PGDLLIMPORT volatile uint32 InterruptHoldoffCount; -extern PGDLLIMPORT volatile uint32 CritSectionCount; +extern PGDLLIMPORT uint32 InterruptHoldoffCount; +extern PGDLLIMPORT uint32 CritSectionCount; + +extern const pg_atomic_uint32 ZeroPendingInterruptsFlags; static inline void HOLD_INTERRUPTS(void) { InterruptHoldoffCount++; - CheckForInterruptsMask = (InterruptMask) 0; + + /* + * Disable CHECK_FOR_INTERRUPTS() by pointing MyPendingInterruptsFlags to + * an all-zeros constant. + */ + MyPendingInterruptsFlags = unconstify(pg_atomic_uint32 *, &ZeroPendingInterruptsFlags); } static inline void RESUME_INTERRUPTS(void) { - Assert(CheckForInterruptsMask == 0); Assert(InterruptHoldoffCount > 0); InterruptHoldoffCount--; if (InterruptHoldoffCount == 0 && CritSectionCount == 0) - CheckForInterruptsMask = EnabledInterruptsMask; - else - Assert(CheckForInterruptsMask == 0); + MyPendingInterruptsFlags = &MyPendingInterrupts->flags; } static inline void START_CRIT_SECTION(void) { CritSectionCount++; - CheckForInterruptsMask = 0; + MyPendingInterruptsFlags = unconstify(pg_atomic_uint32 *, &ZeroPendingInterruptsFlags); } static inline void @@ -362,9 +283,9 @@ END_CRIT_SECTION(void) Assert(CritSectionCount > 0); CritSectionCount--; if (InterruptHoldoffCount == 0 && CritSectionCount == 0) - CheckForInterruptsMask = EnabledInterruptsMask; - else - Assert(CheckForInterruptsMask == 0); + MyPendingInterruptsFlags = &MyPendingInterrupts->flags; } +extern void ResetInterruptHoldoffCounts(uint32 new_holdoff_count, uint32 new_crit_section_count); + #endif /* IPC_INTERRUPT_H */ diff --git a/src/include/ipc/interrupt.h b/src/include/ipc/standard_interrupts.h similarity index 53% copy from src/include/ipc/interrupt.h copy to src/include/ipc/standard_interrupts.h index 239d58f7597..8907b378f4f 100644 --- a/src/include/ipc/interrupt.h +++ b/src/include/ipc/standard_interrupts.h @@ -1,36 +1,12 @@ -/*------------------------------------------------------------------------- - * - * interrupt.h - * Inter-process interrupts - * - * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group - * Portions Copyright (c) 1994, Regents of the University of California - * - * IDENTIFICATION - * src/include/ipc/interrupt.h - * - *------------------------------------------------------------------------- - */ - -#ifndef IPC_INTERRUPT_H -#define IPC_INTERRUPT_H - -#include "port/atomics.h" -#include "storage/procnumber.h" - -/* - * Include waiteventset.h for the WL_* flags. They're not needed her, but are - * needed which are needed by all callers of WaitInterrupt, so include it - * here. - * - * Note: InterruptMask is defind in waiteventset.h to avoid circular dependency - */ -#include "storage/waiteventset.h" +#ifndef IPC_STANDARD_INTERRUPTS_H +#define IPC_STANDARD_INTERRUPTS_H +/* defined here to avoid circular dependency to interrupt.h */ +typedef uint64 InterruptMask; /* - * Flags in the pending interrupts bitmask. Each value is a different bit, so that - * these can be conveniently OR'd together. + * Interrupt bits that used in PendingIntrrupts->interrupts bitmask. Each + * value is a different bit, so that these can be conveniently OR'd together. */ #define UINT64_BIT(shift) (UINT64_C(1) << (shift)) @@ -191,74 +167,8 @@ * extensions ***********************************************************************/ #define BEGIN_ADDIN_INTERRUPTS 25 -#define END_ADDIN_INTERRUPTS 63 +#define END_ADDIN_INTERRUPTS 64 -/* - * SLEEPING_ON_INTERRUPTS indicates that the backend is currently blocked - * waiting for an interrupt. If set, the backend needs to be woken up when a - * bit in the pending interrupts mask is set. It's used internally by the - * interrupt machinery, and cannot be used directly in the public functions. - * It's named differently to distinguish it from the actual interrupt flags. - */ -#define SLEEPING_ON_INTERRUPTS UINT64_BIT(63) - -extern PGDLLIMPORT pg_atomic_uint64 *MyPendingInterrupts; - -/* - * Test an interrupt flag (or flags). - */ -static inline bool -InterruptPending(InterruptMask interruptMask) -{ - /* - * Note that there is no memory barrier here. This is used in - * CHECK_FOR_INTERRUPTS(), so we want this to be as cheap as possible. - * - * That means that if the interrupt is concurrently set by another - * process, we might miss it. That should be OK, because the next - * WaitInterrupt() or equivalent call acts as a synchronization barrier. - * We will see the updated value before sleeping. - */ - return (pg_atomic_read_u64(MyPendingInterrupts) & interruptMask) != 0; -} - -/* - * Clear an interrupt flag (or flags). - */ -static inline void -ClearInterrupt(InterruptMask interruptMask) -{ - pg_atomic_fetch_and_u64(MyPendingInterrupts, ~interruptMask); -} - -/* - * Test and clear an interrupt flag (or flags). - */ -static inline bool -ConsumeInterrupt(InterruptMask interruptMask) -{ - if (likely(!InterruptPending(interruptMask))) - return false; - - ClearInterrupt(interruptMask); - return true; -} - -extern void RaiseInterrupt(InterruptMask interruptMask); -extern void SendInterrupt(InterruptMask interruptMask, ProcNumber pgprocno); -extern int WaitInterrupt(InterruptMask interruptMask, int wakeEvents, long timeout, - uint32 wait_event_info); -extern int WaitInterruptOrSocket(InterruptMask interruptMask, int wakeEvents, pgsocket sock, - long timeout, uint32 wait_event_info); -extern void SwitchToLocalInterrupts(void); -extern void SwitchToSharedInterrupts(void); -extern void InitializeInterruptWaitSet(void); - -typedef void (*pg_interrupt_handler_t) (void); -extern void SetInterruptHandler(InterruptMask interruptMask, pg_interrupt_handler_t handler); - -extern void EnableInterrupt(InterruptMask interruptMask); -extern void DisableInterrupt(InterruptMask interruptMask); /* for extensions */ extern InterruptMask RequestAddinInterrupt(void); @@ -281,90 +191,4 @@ extern void ProcessClientConnectionLost(void); extern void ProcessAuxProcessShutdownInterrupt(void); -/***************************************************************************** - * CHECK_FOR_INTERRUPTS() and friends - *****************************************************************************/ - - -extern PGDLLIMPORT volatile InterruptMask EnabledInterruptsMask; -extern PGDLLIMPORT volatile InterruptMask CheckForInterruptsMask; - -extern void ProcessInterrupts(void); - -/* Test whether an interrupt is pending */ -#ifndef WIN32 -#define INTERRUPTS_PENDING_CONDITION(mask) \ - (unlikely(InterruptPending(mask))) -#else -#define INTERRUPTS_PENDING_CONDITION(mask) \ - (unlikely(UNBLOCKED_SIGNAL_QUEUE()) ? \ - pgwin32_dispatch_queued_signals() : (void) 0, \ - unlikely(InterruptPending(mask))) -#endif - -/* - * Is ProcessInterrupts() guaranteed to clear all the bits in 'mask'? - * - * (The interrupt handler may re-raise the interrupt, though) - */ -#define INTERRUPTS_CAN_BE_PROCESSED(mask) \ - (((mask) & CheckForInterruptsMask) == (mask)) - -/* Service interrupt, if one is pending and it's safe to service it now */ -#define CHECK_FOR_INTERRUPTS() \ -do { \ - if (INTERRUPTS_PENDING_CONDITION(CheckForInterruptsMask)) \ - ProcessInterrupts(); \ -} while(0) - - -/***************************************************************************** - * Critical section and interrupt holdoff mechanism - *****************************************************************************/ - -/* these are marked volatile because they are examined by signal handlers: */ -/* - * XXX: is that still true? Should we use local vars to avoid repeated access - * e.g. inside RESUME_INTERRUPTS() ? - */ -extern PGDLLIMPORT volatile uint32 InterruptHoldoffCount; -extern PGDLLIMPORT volatile uint32 CritSectionCount; - -static inline void -HOLD_INTERRUPTS(void) -{ - InterruptHoldoffCount++; - CheckForInterruptsMask = (InterruptMask) 0; -} - -static inline void -RESUME_INTERRUPTS(void) -{ - Assert(CheckForInterruptsMask == 0); - Assert(InterruptHoldoffCount > 0); - InterruptHoldoffCount--; - if (InterruptHoldoffCount == 0 && CritSectionCount == 0) - CheckForInterruptsMask = EnabledInterruptsMask; - else - Assert(CheckForInterruptsMask == 0); -} - -static inline void -START_CRIT_SECTION(void) -{ - CritSectionCount++; - CheckForInterruptsMask = 0; -} - -static inline void -END_CRIT_SECTION(void) -{ - Assert(CritSectionCount > 0); - CritSectionCount--; - if (InterruptHoldoffCount == 0 && CritSectionCount == 0) - CheckForInterruptsMask = EnabledInterruptsMask; - else - Assert(CheckForInterruptsMask == 0); -} - -#endif /* IPC_INTERRUPT_H */ +#endif /* IPC_STANDARD_INTERRUPTS_H */ diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 1ca4c70715c..2590006270e 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -255,7 +255,7 @@ typedef struct PGPROC int delayChkptFlags; /* for DELAY_CHKPT_* flags */ /* Bit mask of pending interrupts, waiting to be processed */ - pg_atomic_uint64 pendingInterrupts; + PendingInterrupts pendingInterrupts; #ifdef WIN32 /* Event handle to wake up the process when sending an interrupt */ diff --git a/src/backend/access/spgist/spgdoinsert.c b/src/backend/access/spgist/spgdoinsert.c index b8b5eaac57a..7c7371c69e8 100644 --- a/src/backend/access/spgist/spgdoinsert.c +++ b/src/backend/access/spgist/spgdoinsert.c @@ -2035,11 +2035,8 @@ spgdoinsert(Relation index, SpGistState *state, * pending, break out of the loop and deal with the situation below. * Set result = false because we must restart the insertion if the * interrupt isn't a query-cancel-or-die case. - * - * FIXME: CheckForInterruptsMask covers more than just query cancel - * and die. Could we be more precise here? */ - if (INTERRUPTS_PENDING_CONDITION(CheckForInterruptsMask)) + if (INTERRUPTS_PENDING_CONDITION()) { result = false; break; @@ -2165,7 +2162,7 @@ spgdoinsert(Relation index, SpGistState *state, * repeatedly, check for query cancel (see comments above). */ process_inner_tuple: - if (INTERRUPTS_PENDING_CONDITION(CheckForInterruptsMask)) + if (INTERRUPTS_PENDING_CONDITION()) { result = false; break; @@ -2341,7 +2338,7 @@ spgdoinsert(Relation index, SpGistState *state, * were the case, telling the caller to retry would create an infinite * loop. */ - Assert(INTERRUPTS_CAN_BE_PROCESSED(CheckForInterruptsMask)); + Assert(INTERRUPTS_CAN_BE_PROCESSED()); /* * Finally, check for interrupts again. If there was a query cancel, diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 52f23f027f0..18e09bb8b28 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -239,7 +239,7 @@ InitializeParallelDSM(ParallelContext *pcxt) * We can deal with that edge case by pretending no workers were * requested. */ - if (!INTERRUPTS_CAN_BE_PROCESSED(CheckForInterruptsMask)) + if (!INTERRUPTS_CAN_BE_PROCESSED()) pcxt->nworkers = 0; /* diff --git a/src/backend/ipc/interrupt.c b/src/backend/ipc/interrupt.c index 61d148409bc..cf901eb1d9d 100644 --- a/src/backend/ipc/interrupt.c +++ b/src/backend/ipc/interrupt.c @@ -22,29 +22,18 @@ #include "storage/proc.h" #include "utils/resowner.h" + +/* Variables for the holdoff mechanism */ +uint32 InterruptHoldoffCount = 0; +uint32 CritSectionCount = 0; + /* * Currently installed interrupt handlers */ static pg_interrupt_handler_t interrupt_handlers[64]; -/* - * XXX: is 'volatile' still needed on all the variables below? Which ones are - * accessed from signal handlers? - */ - /* Bitmask of currently enabled interrupts */ -volatile InterruptMask EnabledInterruptsMask; - -/* - * Interrupts that would be processed by CHECK_FOR_INTERRUPTS(). This is - * equal to EnabledInterruptsMask, except when interrupts are held off by - * HOLD/RESUME_INTERRUPTS() or a critical section. - */ -volatile InterruptMask CheckForInterruptsMask; - -/* Variables for holdoff mechanism */ -volatile uint32 InterruptHoldoffCount = 0; -volatile uint32 CritSectionCount = 0; +InterruptMask EnabledInterruptsMask; /* A common WaitEventSet used to implement WaitInterrupt() */ static WaitEventSet *InterruptWaitSet; @@ -53,9 +42,10 @@ static WaitEventSet *InterruptWaitSet; #define InterruptWaitSetInterruptPos 0 #define InterruptWaitSetPostmasterDeathPos 1 -static pg_atomic_uint64 LocalPendingInterrupts; - -pg_atomic_uint64 *MyPendingInterrupts = &LocalPendingInterrupts; +static PendingInterrupts LocalPendingInterrupts; +PendingInterrupts *MyPendingInterrupts = &LocalPendingInterrupts; +pg_atomic_uint32 *MyPendingInterruptsFlags = &LocalPendingInterrupts.flags; +const pg_atomic_uint32 ZeroPendingInterruptsFlags; static int nextAddinInterruptBit = BEGIN_ADDIN_INTERRUPTS; @@ -94,10 +84,8 @@ EnableInterrupt(InterruptMask interruptMask) Assert(interrupt_handlers[i] != NULL); } #endif - EnabledInterruptsMask |= interruptMask; - if (InterruptHoldoffCount == 0 && CritSectionCount == 0) - CheckForInterruptsMask = EnabledInterruptsMask; + SetInterruptAttentionMask(EnabledInterruptsMask); } /* @@ -112,54 +100,207 @@ void DisableInterrupt(InterruptMask interruptMask) { EnabledInterruptsMask &= ~interruptMask; +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + pg_atomic_write_u64(&MyPendingInterrupts->attention_mask, EnabledInterruptsMask); +#else + pg_atomic_write_u32(&MyPendingInterrupts->attention_mask_lo, (uint32) EnabledInterruptsMask); + pg_atomic_write_u32(&MyPendingInterrupts->attention_mask_hi, (uint32) (EnabledInterruptsMask >> 32)); +#endif + + /* + * Note: the ATTENTION flag might now be unnecessarily set. We don't try + * to clear it here, the next CHECK_FOR_INTERRUPTS() will take care of it. + */ +} + +/* + * Reset InterruptHoldoffCount and CritSectionCount to given values. Used + * when recovering from an error. + */ +void +ResetInterruptHoldoffCounts(uint32 new_holdoff_count, uint32 new_crit_section_count) +{ + InterruptHoldoffCount = new_holdoff_count; + CritSectionCount = new_crit_section_count; if (InterruptHoldoffCount == 0 && CritSectionCount == 0) - CheckForInterruptsMask = EnabledInterruptsMask; + MyPendingInterruptsFlags = &MyPendingInterrupts->flags; + else + MyPendingInterruptsFlags = unconstify(pg_atomic_uint32 *, &ZeroPendingInterruptsFlags); } /* * ProcessInterrupts: out-of-line portion of CHECK_FOR_INTERRUPTS() macro * - * If an interrupt condition is pending, and it's safe to service it, - * then clear the flag and call the interrupt handler. + * If an interrupt condition is pending, and it's safe to service it, then + * clear the flag and call the interrupt handler. * - * Note: if INTERRUPTS_CAN_BE_PROCESSED(interrupt) is true, then - * ProcessInterrupts is guaranteed to clear the given interrupt before - * returning, if it was set when entering. (This is not the same as - * guaranteeing that it's still clear when we return; another interrupt could - * have arrived. But we promise that any pre-existing one will have been - * serviced.) + * Note: if INTERRUPTS_CAN_BE_PROCESSED() is true, ProcessInterrupts() is + * guaranteed to clear all the enabled interrupts before returning. (This is + * not the same as guaranteeing that it's still clear when we return; another + * interrupt could have arrived. But we promise that any pre-existing one + * will have been serviced.) */ void ProcessInterrupts(void) { + uint64 pending; InterruptMask interruptsToProcess; - Assert(InterruptHoldoffCount == 0 && CritSectionCount == 0); + Assert(INTERRUPTS_CAN_BE_PROCESSED()); + Assert((pg_atomic_read_u32(&MyPendingInterrupts->flags) & PI_FLAG_SLEEPING) == 0); + + /* + * Clear the ATTENTION flag first. This ensures that if any interrupts are + * received while we're processing, the flag is set again. + */ + (void) pg_atomic_write_u32(&MyPendingInterrupts->flags, 0); + + /* + * Make sure others see the clearing of the flags, before we read the + * pending interrupts. + */ + pg_memory_barrier(); /* Check once what interrupts are pending */ - interruptsToProcess = - pg_atomic_read_u64(MyPendingInterrupts) & CheckForInterruptsMask; +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + pending = pg_atomic_read_u64(&MyPendingInterrupts->interrupts); +#else + pending = (uint64) pg_atomic_read_u32(&MyPendingInterrupts->interrupts_lo); + pending |= (uint64) pg_atomic_read_u32(&MyPendingInterrupts->interrupts_hi) << 32; +#endif + interruptsToProcess = pending & EnabledInterruptsMask; - for (int i = 0; i < lengthof(interrupt_handlers); i++) + if (interruptsToProcess != 0) { - if ((interruptsToProcess & UINT64_BIT(i)) != 0) + Assert(InterruptHoldoffCount == 0 && CritSectionCount == 0); + + /* Interrupt handlers are not expected to be re-entrant. */ + HOLD_INTERRUPTS(); + + for (int i = 0; i < lengthof(interrupt_handlers); i++) { - /* - * Clear the interrupt *before* calling the handler function, so - * that if the interrupt is received again while the handler - * function is being executed, we won't miss it. - * - * For similar reasons, we also clear the flags one by one even if - * multiple interrupts are pending. Otherwise if one of the - * interrupt handlers bail out with an ERROR, we would have - * already cleared the other bits, and would miss processing them. - */ - ClearInterrupt(UINT64_BIT(i)); + if ((interruptsToProcess & UINT64_BIT(i)) != 0) + { + /* + * Clear the interrupt *before* calling the handler function, + * so that if the interrupt is received again while the + * handler function is being executed, we won't miss it. + * + * For similar reasons, we also clear the flags one by one + * even if multiple interrupts are pending. Otherwise if one + * of the interrupt handlers bail out with an ERROR, we would + * have already cleared the other bits, and would miss + * processing them. + */ + ClearInterrupt(UINT64_BIT(i)); - /* Call the handler function */ - (*interrupt_handlers[i]) (); + /* Call the handler function */ + (*interrupt_handlers[i]) (); + } } + + RESUME_INTERRUPTS(); } + + /* + * If we get here, we processed all the interrupts that were pending when + * we started. If any new interrupts arrived while we were processing, + * they must've set the ATTENTION flag again so we'll get back here on the + * next CHECK_FOR_INTERRUPTS(). + */ +} + +/* + * Update MyPendingInterrupts->attention_mask, setting PI_FLAG_ATTENTION if + * any of the interrupts in the new mask are already pending. This should be + * called every time after enabling new bits in EnabledInterruptsMask, so that + * the next CHECK_FOR_INTERRUPTS() will react correctly to the newly enabled + * interrupts. + */ +void +SetInterruptAttentionMask(InterruptMask mask) +{ + /* + * This should not be called while sleeping. No other process sets the + * flag, so when we clear/set 'flags' below, we don't need to worry about + * overwriting it. + */ + Assert((pg_atomic_read_u32(&MyPendingInterrupts->flags) & PI_FLAG_SLEEPING) == 0); + +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + pg_atomic_write_u64(&MyPendingInterrupts->attention_mask, mask); +#else + pg_atomic_write_u32(&MyPendingInterrupts->attention_mask_lo, (uint32) mask); + pg_atomic_write_u32(&MyPendingInterrupts->attention_mask_hi, (uint32) (mask >> 32)); +#endif + + /* + * Make sure other processes see the updated attention_mask before we read + * the interrupts that are currently pending. + * + * XXX: It might be cheaper to use pg_atomic_write_membarrier_u64 variant + * above, paired with pg_atomic_read_membarrier_u64() here to read the + * pending interrupts instead of the barrier-less InterruptPending(). + */ + pg_memory_barrier(); + + if (InterruptPending(mask)) + pg_atomic_write_u32(&MyPendingInterrupts->flags, PI_FLAG_ATTENTION); +} + + +/* + * Make 'new_ptr' the active interrupt vector, transfering all the pending + * interrupt bits from the old MyPendingInterrupts vector to the new one. + */ +static void +SwitchMyPendingInterruptsPtr(PendingInterrupts *new_ptr) +{ + PendingInterrupts *old_ptr = MyPendingInterrupts; + + /* should not be called while sleeping */ + Assert((pg_atomic_read_u32(&MyPendingInterrupts->flags) & PI_FLAG_SLEEPING) == 0); + + if (new_ptr == old_ptr) + return; + + MyPendingInterrupts = new_ptr; + if (MyPendingInterruptsFlags == &old_ptr->flags) + MyPendingInterruptsFlags = &new_ptr->flags; + + /* + * Make sure that SIGALRM handlers that call RaiseInterrupt() are now + * seeing the new MyPendingInterrupts destination. + */ + pg_memory_barrier(); + + /* + * Mix in the interrupts that we have received already in 'new_ptr', while + * atomically clearing them from 'old_ptr'. Other backends may continue + * to set bits in 'old_ptr' after this point, but we've atomically + * transferred the existing bits to our local vector so we won't get + * duplicated interrupts later if we switch back. + */ +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + { + uint64 old_interrupts; + + old_interrupts = pg_atomic_exchange_u64(&old_ptr->interrupts, 0); + pg_atomic_fetch_or_u64(&new_ptr->interrupts, old_interrupts); + } +#else + { + uint32 old_interrupts_lo; + uint32 old_interrupts_hi; + + old_interrupts_lo = pg_atomic_exchange_u32(&old_ptr->interrupts_lo, 0); + old_interrupts_hi = pg_atomic_exchange_u32(&old_ptr->interrupts_hi, 0); + pg_atomic_fetch_or_u32(&new_ptr->interrupts_lo, old_interrupts_lo); + pg_atomic_fetch_or_u32(&new_ptr->interrupts_hi, old_interrupts_hi); + } +#endif + + SetInterruptAttentionMask(EnabledInterruptsMask); } /* @@ -169,26 +310,7 @@ ProcessInterrupts(void) void SwitchToLocalInterrupts(void) { - if (MyPendingInterrupts == &LocalPendingInterrupts) - return; - - MyPendingInterrupts = &LocalPendingInterrupts; - - /* - * Make sure that SIGALRM handlers that call RaiseInterrupt() are now - * seeing the new MyPendingInterrupts destination. - */ - pg_memory_barrier(); - - /* - * Mix in the interrupts that we have received already in our shared - * interrupt vector, while atomically clearing it. Other backends may - * continue to set bits in it after this point, but we've atomically - * transferred the existing bits to our local vector so we won't get - * duplicated interrupts later if we switch back. - */ - pg_atomic_fetch_or_u64(MyPendingInterrupts, - pg_atomic_exchange_u64(&MyProc->pendingInterrupts, 0)); + SwitchMyPendingInterruptsPtr(&LocalPendingInterrupts); } /* @@ -198,20 +320,61 @@ SwitchToLocalInterrupts(void) void SwitchToSharedInterrupts(void) { - if (MyPendingInterrupts == &MyProc->pendingInterrupts) - return; + SwitchMyPendingInterruptsPtr(&MyProc->pendingInterrupts); +} - MyPendingInterrupts = &MyProc->pendingInterrupts; +static bool +SendOrRaiseInterrupt(PendingInterrupts *ptr, InterruptMask interruptMask) +{ + uint64 old_pending; + uint64 attention_mask; + uint32 old_flags; + bool wakeup = false; /* - * Make sure that SIGALRM handlers that call RaiseInterrupt() are now - * seeing the new MyPendingInterrupts destination. + * Do an "unlocked" read first, for a quick exit if all the bits are + * already set. */ - pg_memory_barrier(); +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + old_pending = pg_atomic_read_u64(&ptr->interrupts); +#else + old_pending = (uint64) pg_atomic_read_u32(&ptr->interrupts_lo); + old_pending |= (uint64) pg_atomic_read_u32(&ptr->interrupts_hi) << 32; +#endif - /* Mix in any unhandled bits from LocalPendingInterrupts. */ - pg_atomic_fetch_or_u64(MyPendingInterrupts, - pg_atomic_exchange_u64(&LocalPendingInterrupts, 0)); + if ((interruptMask & ~old_pending) == 0) + return false; /* no new bits were set */ + + /* OR our bits to the target */ +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + (void) pg_atomic_fetch_or_u64(&ptr->interrupts, interruptMask); +#else + (void) pg_atomic_fetch_or_u32(&ptr->interrupts_lo, (uint32) interruptMask); + (void) pg_atomic_fetch_or_u32(&ptr->interrupts_hi, (uint32) (interruptMask >> 32)); +#endif + + /* + * Did we set any bits that the requires the target process's ATTENTION? + */ +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + attention_mask = pg_atomic_read_u64(&ptr->attention_mask); +#else + attention_mask = (uint64) pg_atomic_read_u32(&ptr->attention_mask_lo); + attention_mask |= (uint64) pg_atomic_read_u32(&ptr->attention_mask_hi) << 32; +#endif + if ((attention_mask & interruptMask) != 0) + { + old_flags = pg_atomic_fetch_or_u32(&ptr->flags, PI_FLAG_ATTENTION); + + /* + * Furthermore, if the process is currently sleeping on these + * interrupts, wake it up. + */ + if ((old_flags & PI_FLAG_SLEEPING) != 0) + wakeup = true; + } + + return wakeup; } /* @@ -223,15 +386,7 @@ SwitchToSharedInterrupts(void) void RaiseInterrupt(InterruptMask interruptMask) { - uint64 old_pending; - - old_pending = pg_atomic_fetch_or_u64(MyPendingInterrupts, interruptMask); - - /* - * If the process is currently blocked waiting for an interrupt to arrive, - * and the interrupt wasn't already pending, wake it up. - */ - if ((old_pending & (interruptMask | SLEEPING_ON_INTERRUPTS)) == SLEEPING_ON_INTERRUPTS) + if (SendOrRaiseInterrupt(MyPendingInterrupts, interruptMask)) WakeupMyProc(); } @@ -248,14 +403,12 @@ void SendInterrupt(InterruptMask interruptMask, ProcNumber pgprocno) { PGPROC *proc; - uint64 old_pending; Assert(pgprocno != INVALID_PROC_NUMBER); Assert(pgprocno >= 0); Assert(pgprocno < ProcGlobal->allProcCount); proc = &ProcGlobal->allProcs[pgprocno]; - old_pending = pg_atomic_fetch_or_u64(&proc->pendingInterrupts, interruptMask); elog(DEBUG1, "sending interrupt 0x016%" PRIx64 " to pid %d", interruptMask, proc->pid); @@ -263,7 +416,7 @@ SendInterrupt(InterruptMask interruptMask, ProcNumber pgprocno) * If the process is currently blocked waiting for an interrupt to arrive, * and the interrupt wasn't already pending, wake it up. */ - if ((old_pending & (interruptMask | SLEEPING_ON_INTERRUPTS)) == SLEEPING_ON_INTERRUPTS) + if (SendOrRaiseInterrupt(&proc->pendingInterrupts, interruptMask)) WakeupOtherProc(proc); } diff --git a/src/backend/storage/ipc/ipc.c b/src/backend/storage/ipc/ipc.c index 806947571ae..0e86491e1b6 100644 --- a/src/backend/storage/ipc/ipc.c +++ b/src/backend/storage/ipc/ipc.c @@ -173,14 +173,13 @@ proc_exit_prepare(int code) proc_exit_inprogress = true; /* - * Forget any pending cancel or die requests and set InterruptHoldoffCount - * to prevent any more interrupts from being processed; we're doing our - * best to close up shop already. + * Forget any pending cancel or die requests and hold interrupts to + * prevent any more interrupts from being processed; we're doing our best + * to close up shop already. */ ClearInterrupt(INTERRUPT_TERMINATE); ClearInterrupt(INTERRUPT_QUERY_CANCEL); - InterruptHoldoffCount = 1; - CritSectionCount = 0; + ResetInterruptHoldoffCounts(1, 0); /* * Also clear the error context stack, to prevent error callbacks from diff --git a/src/backend/storage/ipc/waiteventset.c b/src/backend/storage/ipc/waiteventset.c index 81999613313..b7a139a5b2d 100644 --- a/src/backend/storage/ipc/waiteventset.c +++ b/src/backend/storage/ipc/waiteventset.c @@ -715,8 +715,7 @@ ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, InterruptMask interru * changes. It means that when interrupt mask is set to 0, we will * listen on the kernel object unnecessarily, and might get some * spurious wakeups. The interrupt sending code should not wake us up - * if the SLEEPING_ON_INTERRUPTS bit is not armed, though, so it - * should be rare. + * if the SLEEPING bit is not armed, though, so it should be rare. */ return; } @@ -1073,6 +1072,14 @@ WaitEventSetWait(WaitEventSet *set, long timeout, pgwin32_dispatch_queued_signals(); #endif + /* + * We will change the 'attention_mask' in MyPendingInterrupts for the + * sleep, which means that CHECK_FOR_INTERRUPTS() won't work correctly. + * There are no CHECK_FOR_INTERRUPTS() calls below, but hold interrupts + * until we've restored 'attention_mask' just to be sure. + */ + HOLD_INTERRUPTS(); + /* * Atomically check if the interrupt is already pending and advertise that * we are about to start sleeping. If it was already pending, avoid @@ -1085,28 +1092,35 @@ WaitEventSetWait(WaitEventSet *set, long timeout, */ if (set->interrupt_mask != 0) { - InterruptMask old_mask; bool already_pending = false; /* * Perform a plain atomic read first as a fast path for the case that * an interrupt is already pending. */ - old_mask = pg_atomic_read_u64(MyPendingInterrupts); - already_pending = ((old_mask & set->interrupt_mask) != 0); + already_pending = InterruptPending(set->interrupt_mask); if (!already_pending) { /* - * Atomically set the SLEEPING_ON_INTERRUPTS bit and re-check if - * an interrupt is already pending. The atomic op provides - * synchronization so that if an interrupt bit is set after this, - * the setter will wake us up. + * Set the attention mask and SLEEPING bit and re-check if an + * interrupt is already pending. The memory barrier synchronizes + * with the atomic fetch-or in SendOrRaiseInterrupt() so that if + * an interrupt bit is set after setting the flag, the setter will + * see the SLEEPING flag and will wake us up. */ - old_mask = pg_atomic_fetch_or_u64(MyPendingInterrupts, SLEEPING_ON_INTERRUPTS); - already_pending = ((old_mask & set->interrupt_mask) != 0); +#ifndef PG_HAVE_ATOMIC_U64_SIMULATION + pg_atomic_write_u64(&MyPendingInterrupts->attention_mask, set->interrupt_mask); +#else + pg_atomic_write_u32(&MyPendingInterrupts->attention_mask_lo, (uint32) set->interrupt_mask); + pg_atomic_write_u32(&MyPendingInterrupts->attention_mask_hi, (uint32) (set->interrupt_mask >> 32)); +#endif + pg_atomic_write_u32(&MyPendingInterrupts->flags, PI_FLAG_SLEEPING); - /* remember to clear the SLEEPING_ON_INTERRUPTS flag later */ + pg_memory_barrier(); + already_pending = InterruptPending(set->interrupt_mask); + + /* Remember to clear the SLEEPING flag afterwards. */ sleeping_flag_armed = true; } @@ -1176,9 +1190,17 @@ WaitEventSetWait(WaitEventSet *set, long timeout, } } - /* If we set the SLEEPING_ON_INTERRUPTS flag, reset it again */ + /* + * If we slept, clear the SLEEPING flag again and reset the attention mask + * for CHECK_FOR_INTERRUPTS() + */ if (sleeping_flag_armed) - pg_atomic_fetch_and_u64(MyPendingInterrupts, ~((uint32) SLEEPING_ON_INTERRUPTS)); + { + pg_atomic_write_u32(&MyPendingInterrupts->flags, 0); + SetInterruptAttentionMask(EnabledInterruptsMask); + } + + RESUME_INTERRUPTS(); #ifndef WIN32 waiting = false; @@ -1256,7 +1278,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, /* Drain the signalfd. */ drain(); - if (set->interrupt_mask != 0 && (pg_atomic_read_u64(MyPendingInterrupts) & set->interrupt_mask) != 0) + if (set->interrupt_mask != 0 && InterruptPending(set->interrupt_mask)) { occurred_events->fd = PGINVALID_SOCKET; occurred_events->events = WL_INTERRUPT; @@ -1415,7 +1437,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, if (cur_event->events == WL_INTERRUPT && cur_kqueue_event->filter == EVFILT_SIGNAL) { - if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) + if (set->interrupt_mask != 0 && InterruptPending(set->interrupt_mask)) { occurred_events->fd = PGINVALID_SOCKET; occurred_events->events = WL_INTERRUPT; @@ -1540,7 +1562,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, /* There's data in the self-pipe, clear it. */ drain(); - if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) + if (set->interrupt_mask != 0 && InterruptPending(set->interrupt_mask)) { occurred_events->fd = PGINVALID_SOCKET; occurred_events->events = WL_INTERRUPT; @@ -1761,7 +1783,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, if (!ResetEvent(set->handles[cur_event->pos + 1])) elog(ERROR, "ResetEvent failed: error code %lu", GetLastError()); - if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) + if (set->interrupt_mask != 0 && InterruptPending(set->interrupt_mask)) { occurred_events->fd = PGINVALID_SOCKET; occurred_events->events = WL_INTERRUPT; diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 5e595e51987..94e59aff4ea 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -3202,11 +3202,11 @@ report_recovery_conflict(RecoveryConflictReason reason) if (!DoingCommandRead) { /* Avoid losing sync in the FE/BE protocol. */ - if (!INTERRUPTS_CAN_BE_PROCESSED(INTERRUPT_QUERY_CANCEL)) + if ((EnabledInterruptsMask & INTERRUPT_QUERY_CANCEL) == 0) { /* * Re-arm and defer this interrupt until later. See similar - * code in ProcessInterrupts(). + * code in ProcessInterrupts(). FIXME: what similar code */ (void) pg_atomic_fetch_or_u32(&MyProc->pendingRecoveryConflicts, (1 << reason)); RaiseInterrupt(INTERRUPT_RECOVERY_CONFLICT); @@ -3261,7 +3261,6 @@ void ProcessTerminateInterrupt(void) { /* OK to accept any interrupts now? */ - Assert(InterruptHoldoffCount == 0); Assert(CritSectionCount == 0); { diff --git a/src/backend/utils/error/elog.c b/src/backend/utils/error/elog.c index a36c7c33249..404e4247831 100644 --- a/src/backend/utils/error/elog.c +++ b/src/backend/utils/error/elog.c @@ -536,9 +536,7 @@ errfinish(const char *filename, int lineno, const char *funcname) * could save and restore InterruptHoldoffCount for itself, but this * should make life easier for most.) */ - InterruptHoldoffCount = 0; - - CritSectionCount = 0; /* should be unnecessary, but... */ + ResetInterruptHoldoffCounts(0, 0); /* * Note that we leave CurrentMemoryContext set to ErrorContext. The diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 742fc43e5da..369023154a1 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2186,6 +2186,7 @@ PatternInfoArray Pattern_Prefix_Status Pattern_Type PendingFsyncEntry +PendingInterrupts PendingListenAction PendingListenEntry PendingRelDelete -- 2.53.0.1.gb2826b52eb