From db89ef9fcbe172203826e1b9fb1b068a827c8165 Mon Sep 17 00:00:00 2001 From: Thomas Munro Date: Sat, 5 Jun 2021 17:13:11 +1200 Subject: [PATCH v3 2/2] Redesign interrupts, remove ProcSignals. "ProcSignals" were a two-step way for one backend to ask another backend to do something, by multiplexing SIGUSR1. Historically, real work was done in the SIGUSR1 handler, but commit 0da096d7 removed the last case of that. Now the handler just set "interrupt pending" flags, called SetLatch(MyLatch) to close races, and then did all the interesting work at the next CHECK_FOR_INTERRUPTS() call. This commit removes the ProcSignal step. We expose each backend's interrupt vector as a single word in shared memory, so other backends can set interrupt flags directly before setting the target's latch. CHECK_FOR_INTERRUPTS() is still about as cheap: it just does a relaxed load from a signal uint32_t to see if any flags are set. Instead of "how soon do we expect another process to handle the signal, and then reach a CHECK_FOR_INTERRUPTS()?", we have "how soon do we expect another process to reach a CHECK_FOR_INTERRUPTS() and see the change in memory?". Benefits of this refactoring include removing unnecessary concepts and terminology, avoiding the risk of people putting buggy code back into signal handlers, removing a whole lot of global variables, removing dependencies on PIDs and the process model, removing risks of sending a signal that has default action of termination to the wrong process, and taking a step towards being able remove the fake signal system from the Windows port. Notable changes: * all calls to SendProcSignal(pid, PROCSIG_XXX) become SendInterrupt(INTERRUPT_XXX, procno) * all XxxPending global variables are gone; they are represented as bits in MyProc->pending_interrupts * the SIGUSR1 handler is gone; SendInterrupts() sets latches immediately * places that know the PIDs of other backends must now use procnos instead * in the remaining cases of actual signal handlers, eg SIGALRM for timers, RaiseInterrupt() can be used to set your own interrupt flags The only code left in src/backend/storage/ipc/procsignal.c relates to ProcSignalBarriers. XXX Those should perhaps be renamed. SIGUSR1 can now be set to SIG_IGN in most backends. Some special cases remain, that were not using the general sigusr1_handler (now removed). Reviewed-by: Heikki Linnakangas Reviewed-by: Discussion: https://postgr.es/m/CA%2BhUKG%2B3MkS21yK4jL4cgZywdnnGKiBg0jatoV6kzaniBmcqbQ%40mail.gmail.com --- contrib/pg_prewarm/autoprewarm.c | 2 +- src/backend/access/transam/parallel.c | 28 +-- src/backend/commands/async.c | 100 ++------ src/backend/commands/vacuum.c | 2 +- src/backend/libpq/pqcomm.c | 4 +- src/backend/libpq/pqmq.c | 23 +- src/backend/optimizer/util/pathnode.c | 1 + src/backend/postmaster/autovacuum.c | 15 +- src/backend/postmaster/bgworker.c | 3 +- src/backend/postmaster/bgwriter.c | 2 +- src/backend/postmaster/checkpointer.c | 8 +- src/backend/postmaster/interrupt.c | 89 ++++++- src/backend/postmaster/pgarch.c | 6 +- src/backend/postmaster/startup.c | 6 +- src/backend/postmaster/walsummarizer.c | 6 +- src/backend/postmaster/walwriter.c | 2 +- .../replication/logical/applyparallelworker.c | 26 +-- src/backend/replication/logical/launcher.c | 29 ++- src/backend/replication/logical/slotsync.c | 34 +-- src/backend/replication/logical/tablesync.c | 1 + src/backend/replication/slot.c | 20 +- src/backend/replication/syncrep.c | 7 +- src/backend/replication/walreceiver.c | 2 +- src/backend/replication/walsender.c | 7 +- src/backend/storage/buffer/bufmgr.c | 8 +- src/backend/storage/ipc/ipc.c | 6 +- src/backend/storage/ipc/procarray.c | 26 +-- src/backend/storage/ipc/procsignal.c | 217 ++---------------- src/backend/storage/ipc/signalfuncs.c | 1 + src/backend/storage/ipc/sinval.c | 60 ++--- src/backend/storage/ipc/sinvaladt.c | 16 +- src/backend/storage/ipc/standby.c | 48 ++-- src/backend/storage/lmgr/proc.c | 15 +- src/backend/tcop/postgres.c | 204 ++++++---------- src/backend/utils/activity/pgstat_database.c | 14 +- src/backend/utils/adt/mcxtfuncs.c | 10 +- src/backend/utils/init/globals.c | 17 +- src/backend/utils/init/postinit.c | 20 +- src/backend/utils/mmgr/mcxt.c | 26 +-- src/include/access/parallel.h | 1 - src/include/commands/async.h | 4 - src/include/libpq/pqmq.h | 2 +- src/include/miscadmin.h | 68 +----- src/include/postmaster/interrupt.h | 173 +++++++++++++- src/include/regex/regcustom.h | 1 + src/include/replication/slot.h | 1 + src/include/replication/walsender_private.h | 1 + src/include/replication/worker_internal.h | 5 +- src/include/storage/proc.h | 2 + src/include/storage/procarray.h | 6 +- src/include/storage/procsignal.h | 38 --- src/include/storage/sinval.h | 5 - src/include/storage/standby.h | 5 +- src/include/tcop/tcopprot.h | 2 +- 54 files changed, 586 insertions(+), 839 deletions(-) diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c index bb590f77c54..d26809ac3aa 100644 --- a/contrib/pg_prewarm/autoprewarm.c +++ b/contrib/pg_prewarm/autoprewarm.c @@ -157,7 +157,7 @@ autoprewarm_main(Datum main_arg) /* Establish signal handlers; once that's done, unblock signals. */ pqsignal(SIGTERM, SignalHandlerForShutdownRequest); pqsignal(SIGHUP, SignalHandlerForConfigReload); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); BackgroundWorkerUnblockSignals(); /* Create (if necessary) and attach to our shared memory area. */ diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 6a14c0057b7..a10e565c21a 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -123,9 +123,6 @@ static FixedParallelState *MyFixedParallelState; /* List of active parallel contexts. */ static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list); -/* Backend-local copy of data from FixedParallelState. */ -static pid_t ParallelLeaderPid; - /* * List of internal parallel worker entry points. We need this for * reasons explained in LookupParallelWorkerFunction(), below. @@ -1004,21 +1001,6 @@ ParallelContextActive(void) return !dlist_is_empty(&pcxt_list); } -/* - * Handle receipt of an interrupt indicating a parallel worker message. - * - * Note: this is called within a signal handler! All we can do is set - * a flag that will cause the next CHECK_FOR_INTERRUPTS() to invoke - * HandleParallelMessages(). - */ -void -HandleParallelMessageInterrupt(void) -{ - InterruptPending = true; - ParallelMessagePending = true; - SetLatch(MyLatch); -} - /* * Handle any queued protocol messages received from parallel workers. */ @@ -1334,8 +1316,7 @@ ParallelWorkerMain(Datum main_arg) fps = shm_toc_lookup(toc, PARALLEL_KEY_FIXED, false); MyFixedParallelState = fps; - /* Arrange to signal the leader if we exit. */ - ParallelLeaderPid = fps->parallel_leader_pid; + /* Arrange to interrupt the leader if we exit. */ ParallelLeaderProcNumber = fps->parallel_leader_proc_number; before_shmem_exit(ParallelWorkerShutdown, PointerGetDatum(seg)); @@ -1351,8 +1332,7 @@ ParallelWorkerMain(Datum main_arg) shm_mq_set_sender(mq, MyProc); mqh = shm_mq_attach(mq, seg, NULL); pq_redirect_to_shm_mq(seg, mqh); - pq_set_parallel_leader(fps->parallel_leader_pid, - fps->parallel_leader_proc_number); + pq_set_parallel_leader(fps->parallel_leader_proc_number); /* * Hooray! Primary initialization is complete. Now, we need to set up our @@ -1565,9 +1545,7 @@ ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end) static void ParallelWorkerShutdown(int code, Datum arg) { - SendProcSignal(ParallelLeaderPid, - PROCSIG_PARALLEL_MESSAGE, - ParallelLeaderProcNumber); + SendInterrupt(INTERRUPT_PARALLEL_MESSAGE, ParallelLeaderProcNumber); dsm_detach((dsm_segment *) DatumGetPointer(arg)); } diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index ab4c72762d8..db058cb1252 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -242,7 +242,7 @@ typedef struct QueuePosition */ typedef struct QueueBackendStatus { - int32 pid; /* either a PID or InvalidPid */ + ProcNumber pgprocno; /* either a pgprocno or INVALID_PGPROCNO */ Oid dboid; /* backend's database OID, or InvalidOid */ ProcNumber nextListener; /* id of next listener, or INVALID_PROC_NUMBER */ QueuePosition pos; /* backend has read queue up to here */ @@ -297,7 +297,7 @@ static AsyncQueueControl *asyncQueueControl; #define QUEUE_TAIL (asyncQueueControl->tail) #define QUEUE_STOP_PAGE (asyncQueueControl->stopPage) #define QUEUE_FIRST_LISTENER (asyncQueueControl->firstListener) -#define QUEUE_BACKEND_PID(i) (asyncQueueControl->backend[i].pid) +#define QUEUE_BACKEND_PGPROCNO(i) (asyncQueueControl->backend[i].pgprocno) #define QUEUE_BACKEND_DBOID(i) (asyncQueueControl->backend[i].dboid) #define QUEUE_NEXT_LISTENER(i) (asyncQueueControl->backend[i].nextListener) #define QUEUE_BACKEND_POS(i) (asyncQueueControl->backend[i].pos) @@ -403,15 +403,6 @@ struct NotificationHash static NotificationList *pendingNotifies = NULL; -/* - * Inbound notifications are initially processed by HandleNotifyInterrupt(), - * called from inside a signal handler. That just sets the - * notifyInterruptPending flag and sets the process - * latch. ProcessNotifyInterrupt() will then be called whenever it's safe to - * actually deal with the interrupt. - */ -volatile sig_atomic_t notifyInterruptPending = false; - /* True if we've registered an on_shmem_exit cleanup */ static bool unlistenExitRegistered = false; @@ -523,7 +514,7 @@ AsyncShmemInit(void) asyncQueueControl->lastQueueFillWarn = 0; for (int i = 0; i < MaxBackends; i++) { - QUEUE_BACKEND_PID(i) = InvalidPid; + QUEUE_BACKEND_PGPROCNO(i) = INVALID_PROC_NUMBER; QUEUE_BACKEND_DBOID(i) = InvalidOid; QUEUE_NEXT_LISTENER(i) = INVALID_PROC_NUMBER; SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0); @@ -1097,7 +1088,7 @@ Exec_ListenPreCommit(void) prevListener = i; } QUEUE_BACKEND_POS(MyProcNumber) = max; - QUEUE_BACKEND_PID(MyProcNumber) = MyProcPid; + QUEUE_BACKEND_PGPROCNO(MyProcNumber) = MyProcNumber; QUEUE_BACKEND_DBOID(MyProcNumber) = MyDatabaseId; /* Insert backend into list of listeners at correct position */ if (prevListener != INVALID_PROC_NUMBER) @@ -1240,7 +1231,7 @@ asyncQueueUnregister(void) */ LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE); /* Mark our entry as invalid */ - QUEUE_BACKEND_PID(MyProcNumber) = InvalidPid; + QUEUE_BACKEND_PGPROCNO(MyProcNumber) = INVALID_PROC_NUMBER; QUEUE_BACKEND_DBOID(MyProcNumber) = InvalidOid; /* and remove it from the list */ if (QUEUE_FIRST_LISTENER == MyProcNumber) @@ -1539,22 +1530,22 @@ asyncQueueFillWarning(void) t, QUEUE_FULL_WARN_INTERVAL)) { QueuePosition min = QUEUE_HEAD; - int32 minPid = InvalidPid; + int32 minPgprocno = INVALID_PROC_NUMBER; for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i = QUEUE_NEXT_LISTENER(i)) { - Assert(QUEUE_BACKEND_PID(i) != InvalidPid); + Assert(QUEUE_BACKEND_PGPROCNO(i) != INVALID_PROC_NUMBER); min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i)); if (QUEUE_POS_EQUAL(min, QUEUE_BACKEND_POS(i))) - minPid = QUEUE_BACKEND_PID(i); + minPgprocno = QUEUE_BACKEND_PGPROCNO(i); } ereport(WARNING, (errmsg("NOTIFY queue is %.0f%% full", fillDegree * 100), - (minPid != InvalidPid ? - errdetail("The server process with PID %d is among those with the oldest transactions.", minPid) + (minPgprocno != INVALID_PROC_NUMBER ? + errdetail("The server process with pgprocno %d is among those with the oldest transactions.", minPgprocno) : 0), - (minPid != InvalidPid ? + (minPgprocno != INVALID_PROC_NUMBER ? errhint("The NOTIFY queue cannot be emptied until that process ends its current transaction.") : 0))); @@ -1580,7 +1571,6 @@ asyncQueueFillWarning(void) static void SignalBackends(void) { - int32 *pids; ProcNumber *procnos; int count; @@ -1592,17 +1582,14 @@ SignalBackends(void) * XXX in principle these pallocs could fail, which would be bad. Maybe * preallocate the arrays? They're not that large, though. */ - pids = (int32 *) palloc(MaxBackends * sizeof(int32)); procnos = (ProcNumber *) palloc(MaxBackends * sizeof(ProcNumber)); count = 0; LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE); for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i = QUEUE_NEXT_LISTENER(i)) { - int32 pid = QUEUE_BACKEND_PID(i); QueuePosition pos; - Assert(pid != InvalidPid); pos = QUEUE_BACKEND_POS(i); if (QUEUE_BACKEND_DBOID(i) == MyDatabaseId) { @@ -1623,39 +1610,16 @@ SignalBackends(void) QUEUE_POS_PAGE(pos)) < QUEUE_CLEANUP_DELAY) continue; } - /* OK, need to signal this one */ - pids[count] = pid; + /* OK, need to wake this one */ procnos[count] = i; count++; } LWLockRelease(NotifyQueueLock); - /* Now send signals */ + /* Now send interrupts */ for (int i = 0; i < count; i++) - { - int32 pid = pids[i]; - - /* - * If we are signaling our own process, no need to involve the kernel; - * just set the flag directly. - */ - if (pid == MyProcPid) - { - notifyInterruptPending = true; - continue; - } - - /* - * Note: assuming things aren't broken, a signal failure here could - * only occur if the target backend exited since we released - * NotifyQueueLock; which is unlikely but certainly possible. So we - * just log a low-level debug message if it happens. - */ - if (SendProcSignal(pid, PROCSIG_NOTIFY_INTERRUPT, procnos[i]) < 0) - elog(DEBUG3, "could not signal backend with PID %d: %m", pid); - } + SendInterrupt(INTERRUPT_NOTIFY, procnos[i]); - pfree(pids); pfree(procnos); } @@ -1792,37 +1756,12 @@ AtSubAbort_Notify(void) } } -/* - * HandleNotifyInterrupt - * - * Signal handler portion of interrupt handling. Let the backend know - * that there's a pending notify interrupt. If we're currently reading - * from the client, this will interrupt the read and - * ProcessClientReadInterrupt() will call ProcessNotifyInterrupt(). - */ -void -HandleNotifyInterrupt(void) -{ - /* - * Note: this is called by a SIGNAL HANDLER. You must be very wary what - * you do here. - */ - - /* signal that work needs to be done */ - notifyInterruptPending = true; - - /* make sure the event is processed in due course */ - SetLatch(MyLatch); -} - /* * ProcessNotifyInterrupt * - * This is called if we see notifyInterruptPending set, just before + * This is called if we see INTERRUPT_NOTIFY_INTERRUPT set, just before * transmitting ReadyForQuery at the end of a frontend command, and * also if a notify signal occurs while reading from the frontend. - * HandleNotifyInterrupt() will cause the read to be interrupted - * via the process's latch, and this routine will get called. * If we are truly idle (ie, *not* inside a transaction block), * process the incoming notifies. * @@ -1837,7 +1776,7 @@ ProcessNotifyInterrupt(bool flush) return; /* not really idle */ /* Loop in case another signal arrives while sending messages */ - while (notifyInterruptPending) + while (ConsumeInterrupt(INTERRUPT_NOTIFY)) ProcessIncomingNotify(flush); } @@ -1864,7 +1803,7 @@ asyncQueueReadAllNotifications(void) /* Fetch current state */ LWLockAcquire(NotifyQueueLock, LW_SHARED); /* Assert checks that we have a valid state entry */ - Assert(MyProcPid == QUEUE_BACKEND_PID(MyProcNumber)); + Assert(MyProcNumber == QUEUE_BACKEND_PGPROCNO(MyProcNumber)); pos = QUEUE_BACKEND_POS(MyProcNumber); head = QUEUE_HEAD; LWLockRelease(NotifyQueueLock); @@ -2136,7 +2075,7 @@ asyncQueueAdvanceTail(void) min = QUEUE_HEAD; for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i = QUEUE_NEXT_LISTENER(i)) { - Assert(QUEUE_BACKEND_PID(i) != InvalidPid); + Assert(QUEUE_BACKEND_PGPROCNO(i) != INVALID_PROC_NUMBER); min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i)); } QUEUE_TAIL = min; @@ -2182,9 +2121,6 @@ asyncQueueAdvanceTail(void) static void ProcessIncomingNotify(bool flush) { - /* We *must* reset the flag */ - notifyInterruptPending = false; - /* Do nothing else if we aren't actively listening */ if (listenChannels == NIL) return; diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index 48f8eab2022..9aa8ff729ea 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -2344,7 +2344,7 @@ vacuum_delay_point(void) /* Always check for interrupts */ CHECK_FOR_INTERRUPTS(); - if (InterruptPending || + if (INTERRUPTS_PENDING_CONDITION() || (!VacuumCostActive && !ConfigReloadPending)) return; diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c index 896e1476b50..98fff413ed3 100644 --- a/src/backend/libpq/pqcomm.c +++ b/src/backend/libpq/pqcomm.c @@ -76,6 +76,7 @@ #include "libpq/libpq.h" #include "miscadmin.h" #include "port/pg_bswap.h" +#include "postmaster/interrupt.h" #include "postmaster/postmaster.h" #include "storage/ipc.h" #include "utils/guc_hooks.h" @@ -1409,8 +1410,7 @@ internal_flush_buffer(const char *buf, size_t *start, size_t *end) * the connection. */ *start = *end = 0; - ClientConnectionLost = 1; - InterruptPending = 1; + RaiseInterrupt(INTERRUPT_CONNECTION_LOST); return EOF; } diff --git a/src/backend/libpq/pqmq.c b/src/backend/libpq/pqmq.c index 00a44ca803f..b1d1f292963 100644 --- a/src/backend/libpq/pqmq.c +++ b/src/backend/libpq/pqmq.c @@ -25,8 +25,7 @@ static shm_mq_handle *pq_mq_handle; static bool pq_mq_busy = false; -static pid_t pq_mq_parallel_leader_pid = 0; -static pid_t pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER; +static ProcNumber pq_mq_parallel_leader_proc_number = INVALID_PROC_NUMBER; static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg); static void mq_comm_reset(void); @@ -71,14 +70,13 @@ pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg) } /* - * Arrange to SendProcSignal() to the parallel leader each time we transmit + * Arrange to interrupt the parallel leader each time we transmit * message data via the shm_mq. */ void -pq_set_parallel_leader(pid_t pid, ProcNumber procNumber) +pq_set_parallel_leader(ProcNumber procNumber) { Assert(PqCommMethods == &PqCommMqMethods); - pq_mq_parallel_leader_pid = pid; pq_mq_parallel_leader_proc_number = procNumber; } @@ -163,19 +161,14 @@ mq_putmessage(char msgtype, const char *s, size_t len) */ result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true); - if (pq_mq_parallel_leader_pid != 0) + if (pq_mq_parallel_leader_proc_number != INVALID_PROC_NUMBER) { if (IsLogicalParallelApplyWorker()) - SendProcSignal(pq_mq_parallel_leader_pid, - PROCSIG_PARALLEL_APPLY_MESSAGE, - pq_mq_parallel_leader_proc_number); + SendInterrupt(INTERRUPT_PARALLEL_APPLY_MESSAGE, + pq_mq_parallel_leader_proc_number); else - { - Assert(IsParallelWorker()); - SendProcSignal(pq_mq_parallel_leader_pid, - PROCSIG_PARALLEL_MESSAGE, - pq_mq_parallel_leader_proc_number); - } + SendInterrupt(INTERRUPT_PARALLEL_MESSAGE, + pq_mq_parallel_leader_proc_number); } if (result != SHM_MQ_WOULD_BLOCK) diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c index c42742d2c7b..bfb89049020 100644 --- a/src/backend/optimizer/util/pathnode.c +++ b/src/backend/optimizer/util/pathnode.c @@ -18,6 +18,7 @@ #include "foreign/fdwapi.h" #include "miscadmin.h" +#include "postmaster/interrupt.h" #include "nodes/extensible.h" #include "optimizer/appendinfo.h" #include "optimizer/clauses.h" diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c index 928754b51c4..c42b96f5086 100644 --- a/src/backend/postmaster/autovacuum.c +++ b/src/backend/postmaster/autovacuum.c @@ -395,7 +395,7 @@ AutoVacLauncherMain(char *startup_data, size_t startup_data_len) InitializeTimeouts(); /* establishes SIGALRM handler */ pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, avl_sigusr2_handler); pqsignal(SIGFPE, FloatExceptionHandler); pqsignal(SIGCHLD, SIG_DFL); @@ -445,7 +445,8 @@ AutoVacLauncherMain(char *startup_data, size_t startup_data_len) /* Forget any pending QueryCancel or timeout request */ disable_all_timeouts(false); - QueryCancelPending = false; /* second to avoid race condition */ + ClearInterrupt(INTERRUPT_QUERY_CANCEL); /* second to avoid race + * condition */ /* Report the error to the server log */ EmitErrorReport(); @@ -757,11 +758,11 @@ HandleAutoVacLauncherInterrupts(void) } /* Process barrier events */ - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); /* Perform logging of memory contexts of this process */ - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); /* Process sinval catchup interrupts that happened while sleeping */ @@ -1393,7 +1394,7 @@ AutoVacWorkerMain(char *startup_data, size_t startup_data_len) InitializeTimeouts(); /* establishes SIGALRM handler */ pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); pqsignal(SIGFPE, FloatExceptionHandler); pqsignal(SIGCHLD, SIG_DFL); @@ -2422,7 +2423,7 @@ do_autovacuum(void) * current table (we're done with it, so it would make no sense to * cancel at this point.) */ - QueryCancelPending = false; + ClearInterrupt(INTERRUPT_QUERY_CANCEL); } PG_CATCH(); { @@ -2620,7 +2621,7 @@ perform_work_item(AutoVacuumWorkItem *workitem) * (we're done with it, so it would make no sense to cancel at this * point.) */ - QueryCancelPending = false; + ClearInterrupt(INTERRUPT_QUERY_CANCEL); } PG_CATCH(); { diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c index 2e3aeba8b4f..149b02b5d75 100644 --- a/src/backend/postmaster/bgworker.c +++ b/src/backend/postmaster/bgworker.c @@ -757,7 +757,6 @@ BackgroundWorkerMain(char *startup_data, size_t startup_data_len) * SIGINT is used to signal canceling the current action */ pqsignal(SIGINT, StatementCancelHandler); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); pqsignal(SIGFPE, FloatExceptionHandler); /* XXX Any other handlers needed here? */ @@ -765,9 +764,9 @@ BackgroundWorkerMain(char *startup_data, size_t startup_data_len) else { pqsignal(SIGINT, SIG_IGN); - pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGFPE, SIG_IGN); } + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGTERM, bgworker_die); /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGHUP, SIG_IGN); diff --git a/src/backend/postmaster/bgwriter.c b/src/backend/postmaster/bgwriter.c index 0f75548759a..2b100da24ca 100644 --- a/src/backend/postmaster/bgwriter.c +++ b/src/backend/postmaster/bgwriter.c @@ -105,7 +105,7 @@ BackgroundWriterMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); /* diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c index 199f008bcda..2aa69e43b30 100644 --- a/src/backend/postmaster/checkpointer.c +++ b/src/backend/postmaster/checkpointer.c @@ -196,7 +196,7 @@ CheckpointerMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SignalHandlerForShutdownRequest); /* @@ -557,7 +557,7 @@ CheckpointerMain(char *startup_data, size_t startup_data_len) static void HandleCheckpointerInterrupts(void) { - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); if (ConfigReloadPending) @@ -603,7 +603,7 @@ HandleCheckpointerInterrupts(void) } /* Perform logging of memory contexts of this process */ - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); } @@ -764,7 +764,7 @@ CheckpointWriteDelay(int flags, double progress) } /* Check for barrier events. */ - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); } diff --git a/src/backend/postmaster/interrupt.c b/src/backend/postmaster/interrupt.c index eedc0980cf1..59f5e9b90ed 100644 --- a/src/backend/postmaster/interrupt.c +++ b/src/backend/postmaster/interrupt.c @@ -17,9 +17,11 @@ #include #include "miscadmin.h" +#include "port/atomics.h" #include "postmaster/interrupt.h" #include "storage/ipc.h" #include "storage/latch.h" +#include "storage/proc.h" #include "storage/procsignal.h" #include "utils/guc.h" #include "utils/memutils.h" @@ -27,13 +29,96 @@ volatile sig_atomic_t ConfigReloadPending = false; volatile sig_atomic_t ShutdownRequestPending = false; +static pg_atomic_uint32 LocalPendingInterrupts; + +pg_atomic_uint32 *MyPendingInterrupts = &LocalPendingInterrupts; + +/* + * Switch to local interrupts. Other backends can't send interrupts to this + * one. Only RaiseInterrupt() can set them, from inside this process. + */ +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 backx. + */ + pg_atomic_fetch_or_u32(MyPendingInterrupts, + pg_atomic_exchange_u32(&MyProc->pending_interrupts, 0)); +} + +/* + * Switch to shared memory interrupts. Other backends can send interrupts to + * this one if they know its ProcNumber, and we'll now see any that we missed. + */ +void +SwitchToSharedInterrupts(void) +{ + if (MyPendingInterrupts == &MyProc->pending_interrupts) + return; + + MyPendingInterrupts = &MyProc->pending_interrupts; + + /* + * Make sure that SIGALRM handlers that call RaiseInterrupt() are now + * seeing the new MyPendingInterrupts destination. + */ + pg_memory_barrier(); + + /* Mix in any unhandled bits from LocalPendingInterrupts. */ + pg_atomic_fetch_or_u32(MyPendingInterrupts, + pg_atomic_exchange_u32(&LocalPendingInterrupts, 0)); +} + +/* + * Set an interrupt flag in this backend. + */ +void +RaiseInterrupt(InterruptType reason) +{ + pg_atomic_fetch_or_u32(MyPendingInterrupts, 1 << reason); + SetLatch(MyLatch); +} + +/* + * Set an interrupt flag in another backend. + */ +void +SendInterrupt(InterruptType reason, ProcNumber pgprocno) +{ + PGPROC *proc; + + Assert(pgprocno != INVALID_PROC_NUMBER); + Assert(pgprocno >= 0); + Assert(pgprocno < ProcGlobal->allProcCount); + + proc = &ProcGlobal->allProcs[pgprocno]; + pg_atomic_fetch_or_u32(&proc->pending_interrupts, 1 << reason); + SetLatch(&proc->procLatch); +} + /* * Simple interrupt handler for main loops of background processes. */ void HandleMainLoopInterrupts(void) { - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); if (ConfigReloadPending) @@ -46,7 +131,7 @@ HandleMainLoopInterrupts(void) proc_exit(0); /* Perform logging of memory contexts of this process */ - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); } diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c index 02f91431f5f..be6426b95df 100644 --- a/src/backend/postmaster/pgarch.c +++ b/src/backend/postmaster/pgarch.c @@ -231,7 +231,7 @@ PgArchiverMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, pgarch_waken_stop); /* Reset some signals that are accepted by postmaster but not here */ @@ -858,11 +858,11 @@ pgarch_die(int code, Datum arg) static void HandlePgArchInterrupts(void) { - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); /* Perform logging of memory contexts of this process */ - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); if (ConfigReloadPending) diff --git a/src/backend/postmaster/startup.c b/src/backend/postmaster/startup.c index ef6f98ebcd7..0887e6ed9ca 100644 --- a/src/backend/postmaster/startup.c +++ b/src/backend/postmaster/startup.c @@ -186,11 +186,11 @@ HandleStartupProcInterrupts(void) exit(1); /* Process barrier events */ - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); /* Perform logging of memory contexts of this process */ - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); } @@ -232,7 +232,7 @@ StartupProcessMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ InitializeTimeouts(); /* establishes SIGALRM handler */ pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, StartupProcTriggerHandler); /* diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c index 55dc2ea8f53..4c63184c565 100644 --- a/src/backend/postmaster/walsummarizer.c +++ b/src/backend/postmaster/walsummarizer.c @@ -249,7 +249,7 @@ WalSummarizerMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); /* not used */ /* Advertise ourselves. */ @@ -745,7 +745,7 @@ GetLatestLSN(TimeLineID *tli) static void HandleWalSummarizerInterrupts(void) { - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); if (ConfigReloadPending) @@ -762,7 +762,7 @@ HandleWalSummarizerInterrupts(void) } /* Perform logging of memory contexts of this process */ - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); } diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c index 6e7918a78d4..8d814603500 100644 --- a/src/backend/postmaster/walwriter.c +++ b/src/backend/postmaster/walwriter.c @@ -110,7 +110,7 @@ WalWriterMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); /* not used */ /* diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index e7f7d4c5e4b..40d9486adf2 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -843,9 +843,9 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh) static void pa_shutdown(int code, Datum arg) { - SendProcSignal(MyLogicalRepWorker->leader_pid, - PROCSIG_PARALLEL_APPLY_MESSAGE, - INVALID_PROC_NUMBER); + if (MyLogicalRepWorker->leader_pgprocno != INVALID_PROC_NUMBER) + SendInterrupt(INTERRUPT_PARALLEL_APPLY_MESSAGE, + MyLogicalRepWorker->leader_pgprocno); dsm_detach((dsm_segment *) DatumGetPointer(arg)); } @@ -933,8 +933,7 @@ ParallelApplyWorkerMain(Datum main_arg) error_mqh = shm_mq_attach(mq, seg, NULL); pq_redirect_to_shm_mq(seg, error_mqh); - pq_set_parallel_leader(MyLogicalRepWorker->leader_pid, - INVALID_PROC_NUMBER); + pq_set_parallel_leader(MyLogicalRepWorker->leader_pgprocno); MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time = MyLogicalRepWorker->reply_time = 0; @@ -978,21 +977,6 @@ ParallelApplyWorkerMain(Datum main_arg) Assert(false); } -/* - * Handle receipt of an interrupt indicating a parallel apply worker message. - * - * Note: this is called within a signal handler! All we can do is set a flag - * that will cause the next CHECK_FOR_INTERRUPTS() to invoke - * HandleParallelApplyMessages(). - */ -void -HandleParallelApplyMessageInterrupt(void) -{ - InterruptPending = true; - ParallelApplyMessagePending = true; - SetLatch(MyLatch); -} - /* * Handle a single protocol message received from a single parallel apply * worker. @@ -1090,8 +1074,6 @@ HandleParallelApplyMessages(void) oldcontext = MemoryContextSwitchTo(hpam_context); - ParallelApplyMessagePending = false; - foreach(lc, ParallelApplyWorkerPool) { shm_mq_result res; diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 9894415cd9b..4f141e5e93b 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -56,7 +56,7 @@ LogicalRepWorker *MyLogicalRepWorker = NULL; typedef struct LogicalRepCtxStruct { /* Supervisor process. */ - pid_t launcher_pid; + int launcher_pgprocno; /* Hash table holding last start times of subscriptions' apply workers. */ dsa_handle last_start_dsa; @@ -443,6 +443,7 @@ retry: worker->relstate_lsn = InvalidXLogRecPtr; worker->stream_fileset = NULL; worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid; + worker->leader_pgprocno = is_parallel_apply_worker ? MyProcNumber : INVALID_PROC_NUMBER; worker->parallel_apply = is_parallel_apply_worker; worker->last_lsn = InvalidXLogRecPtr; TIMESTAMP_NOBEGIN(worker->last_send_time); @@ -523,7 +524,7 @@ retry: * slot. */ static void -logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo) +logicalrep_worker_stop_internal(LogicalRepWorker *worker, int interrupt) { uint16 generation; @@ -573,7 +574,7 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo) } /* Now terminate the worker ... */ - kill(worker->proc->pid, signo); + SendInterrupt(interrupt, GetNumberFromPGProc(worker->proc)); /* ... and wait for it to die. */ for (;;) @@ -616,7 +617,7 @@ logicalrep_worker_stop(Oid subid, Oid relid) if (worker) { Assert(!isParallelApplyWorker(worker)); - logicalrep_worker_stop_internal(worker, SIGTERM); + logicalrep_worker_stop_internal(worker, INTERRUPT_DIE); } LWLockRelease(LogicalRepWorkerLock); @@ -663,7 +664,7 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo) * Only stop the worker if the generation matches and the worker is alive. */ if (worker->generation == generation && worker->proc) - logicalrep_worker_stop_internal(worker, SIGINT); + logicalrep_worker_stop_internal(worker, INTERRUPT_QUERY_CANCEL); LWLockRelease(LogicalRepWorkerLock); } @@ -764,7 +765,7 @@ logicalrep_worker_detach(void) LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc); if (isParallelApplyWorker(w)) - logicalrep_worker_stop_internal(w, SIGTERM); + logicalrep_worker_stop_internal(w, INTERRUPT_DIE); } LWLockRelease(LogicalRepWorkerLock); @@ -794,6 +795,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker) worker->subid = InvalidOid; worker->relid = InvalidOid; worker->leader_pid = InvalidPid; + worker->leader_pgprocno = INVALID_PROC_NUMBER; worker->parallel_apply = false; } @@ -805,7 +807,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker) static void logicalrep_launcher_onexit(int code, Datum arg) { - LogicalRepCtx->launcher_pid = 0; + LogicalRepCtx->launcher_pgprocno = INVALID_PROC_NUMBER; } /* @@ -964,6 +966,7 @@ ApplyLauncherShmemInit(void) memset(LogicalRepCtx, 0, ApplyLauncherShmemSize()); + LogicalRepCtx->launcher_pgprocno = INVALID_PROC_NUMBER; LogicalRepCtx->last_start_dsa = DSA_HANDLE_INVALID; LogicalRepCtx->last_start_dsh = DSHASH_HANDLE_INVALID; @@ -1109,8 +1112,10 @@ ApplyLauncherWakeupAtCommit(void) static void ApplyLauncherWakeup(void) { - if (LogicalRepCtx->launcher_pid != 0) - kill(LogicalRepCtx->launcher_pid, SIGUSR1); + int pgprocno; + + if ((pgprocno = LogicalRepCtx->launcher_pgprocno) != INVALID_PROC_NUMBER) + SetLatch(&ProcGlobal->allProcs[pgprocno].procLatch); } /* @@ -1124,8 +1129,8 @@ ApplyLauncherMain(Datum main_arg) before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0); - Assert(LogicalRepCtx->launcher_pid == 0); - LogicalRepCtx->launcher_pid = MyProcPid; + Assert(LogicalRepCtx->launcher_pgprocno == INVALID_PROC_NUMBER); + LogicalRepCtx->launcher_pgprocno = MyProcNumber; /* Establish signal handlers. */ pqsignal(SIGHUP, SignalHandlerForConfigReload); @@ -1240,7 +1245,7 @@ ApplyLauncherMain(Datum main_arg) bool IsLogicalLauncher(void) { - return LogicalRepCtx->launcher_pid == MyProcPid; + return LogicalRepCtx->launcher_pgprocno == MyProcNumber; } /* diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index fe2f07cf449..436b2a16832 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -96,7 +96,7 @@ */ typedef struct SlotSyncCtxStruct { - pid_t pid; + ProcNumber procno; bool stopSignaled; bool syncing; time_t last_start_time; @@ -1208,7 +1208,7 @@ slotsync_worker_onexit(int code, Datum arg) SpinLockAcquire(&SlotSyncCtx->mutex); - SlotSyncCtx->pid = InvalidPid; + SlotSyncCtx->procno = INVALID_PROC_NUMBER; /* * If syncing_slots is true, it indicates that the process errored out @@ -1268,12 +1268,12 @@ wait_for_slot_activity(bool some_slot_updated) * Otherwise, advertise that a sync is in progress. */ static void -check_and_set_sync_info(pid_t worker_pid) +check_and_set_sync_info(ProcNumber procno) { SpinLockAcquire(&SlotSyncCtx->mutex); /* The worker pid must not be already assigned in SlotSyncCtx */ - Assert(worker_pid == InvalidPid || SlotSyncCtx->pid == InvalidPid); + Assert(procno == INVALID_PROC_NUMBER || SlotSyncCtx->procno == INVALID_PROC_NUMBER); /* * Emit an error if startup process signaled the slot sync machinery to @@ -1298,10 +1298,10 @@ check_and_set_sync_info(pid_t worker_pid) SlotSyncCtx->syncing = true; /* - * Advertise the required PID so that the startup process can kill the - * slot sync worker on promotion. + * Advertise the required procno so that the startup process can notify + * the slot sync worker on promotion. */ - SlotSyncCtx->pid = worker_pid; + SlotSyncCtx->procno = procno; SpinLockRelease(&SlotSyncCtx->mutex); @@ -1393,16 +1393,16 @@ ReplSlotSyncWorkerMain(char *startup_data, size_t startup_data_len) pqsignal(SIGINT, SignalHandlerForShutdownRequest); pqsignal(SIGTERM, die); pqsignal(SIGFPE, FloatExceptionHandler); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); pqsignal(SIGCHLD, SIG_DFL); - check_and_set_sync_info(MyProcPid); + check_and_set_sync_info(MyProcNumber); ereport(LOG, errmsg("slot sync worker started")); - /* Register it as soon as SlotSyncCtx->pid is initialized. */ + /* Register it as soon as SlotSyncCtx->procno is initialized. */ before_shmem_exit(slotsync_worker_onexit, (Datum) 0); /* @@ -1522,7 +1522,7 @@ update_synced_slots_inactive_since(void) return; /* The slot sync worker or SQL function mustn't be running by now */ - Assert((SlotSyncCtx->pid == InvalidPid) && !SlotSyncCtx->syncing); + Assert((SlotSyncCtx->procno == INVALID_PROC_NUMBER) && !SlotSyncCtx->syncing); LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); @@ -1561,7 +1561,7 @@ update_synced_slots_inactive_since(void) void ShutDownSlotSync(void) { - pid_t worker_pid; + ProcNumber worker_procno; SpinLockAcquire(&SlotSyncCtx->mutex); @@ -1578,12 +1578,12 @@ ShutDownSlotSync(void) return; } - worker_pid = SlotSyncCtx->pid; + worker_procno = SlotSyncCtx->procno; SpinLockRelease(&SlotSyncCtx->mutex); - if (worker_pid != InvalidPid) - kill(worker_pid, SIGINT); + if (worker_procno != INVALID_PROC_NUMBER) + SendInterrupt(INTERRUPT_QUERY_CANCEL, worker_procno); /* Wait for slot sync to end */ for (;;) @@ -1676,7 +1676,7 @@ SlotSyncShmemInit(void) if (!found) { memset(SlotSyncCtx, 0, size); - SlotSyncCtx->pid = InvalidPid; + SlotSyncCtx->procno = INVALID_PROC_NUMBER; SpinLockInit(&SlotSyncCtx->mutex); } } @@ -1726,7 +1726,7 @@ SyncReplicationSlots(WalReceiverConn *wrconn) { PG_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn)); { - check_and_set_sync_info(InvalidPid); + check_and_set_sync_info(INVALID_PROC_NUMBER); validate_remote_info(wrconn); diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index b00267f0427..45c9fff0a5a 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -104,6 +104,7 @@ #include "nodes/makefuncs.h" #include "parser/parse_relation.h" #include "pgstat.h" +#include "postmaster/interrupt.h" #include "replication/logicallauncher.h" #include "replication/logicalrelation.h" #include "replication/logicalworker.h" diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index baf9b89dc42..55bf0497e05 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -213,6 +213,7 @@ ReplicationSlotsShmemInit(void) LWLockInitialize(&slot->io_in_progress_lock, LWTRANCHE_REPLICATION_SLOT_IO); ConditionVariableInit(&slot->active_cv); + slot->active_pgprocno = INVALID_PROC_NUMBER; } } } @@ -388,6 +389,7 @@ ReplicationSlotCreate(const char *name, bool db_specific, */ Assert(!slot->in_use); Assert(slot->active_pid == 0); + Assert(slot->active_pgprocno == INVALID_PROC_NUMBER); /* first initialize persistent data */ memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData)); @@ -430,7 +432,9 @@ ReplicationSlotCreate(const char *name, bool db_specific, /* We can now mark the slot active, and that makes it our slot. */ SpinLockAcquire(&slot->mutex); Assert(slot->active_pid == 0); + Assert(slot->active_pgprocno == INVALID_PROC_NUMBER); slot->active_pid = MyProcPid; + slot->active_pgprocno = MyProcNumber; SpinLockRelease(&slot->mutex); MyReplicationSlot = slot; @@ -577,12 +581,17 @@ retry: SpinLockAcquire(&s->mutex); if (s->active_pid == 0) + { s->active_pid = MyProcPid; + s->active_pgprocno = MyProcNumber; + } active_pid = s->active_pid; SpinLockRelease(&s->mutex); } else + { active_pid = MyProcPid; + } LWLockRelease(ReplicationSlotControlLock); /* @@ -703,6 +712,7 @@ ReplicationSlotRelease(void) */ SpinLockAcquire(&slot->mutex); slot->active_pid = 0; + slot->active_pgprocno = INVALID_PROC_NUMBER; slot->inactive_since = now; SpinLockRelease(&slot->mutex); ConditionVariableBroadcast(&slot->active_cv); @@ -926,6 +936,7 @@ ReplicationSlotDropPtr(ReplicationSlot *slot) SpinLockAcquire(&slot->mutex); slot->active_pid = 0; + slot->active_pgprocno = INVALID_PROC_NUMBER; SpinLockRelease(&slot->mutex); /* wake up anyone waiting on this slot */ @@ -948,6 +959,7 @@ ReplicationSlotDropPtr(ReplicationSlot *slot) */ LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE); slot->active_pid = 0; + slot->active_pgprocno = INVALID_PROC_NUMBER; slot->in_use = false; LWLockRelease(ReplicationSlotControlLock); ConditionVariableBroadcast(&slot->active_cv); @@ -1307,6 +1319,7 @@ restart: { MyReplicationSlot = s; s->active_pid = MyProcPid; + s->active_pgprocno = MyProcNumber; } SpinLockRelease(&s->mutex); @@ -1559,6 +1572,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, XLogRecPtr restart_lsn; NameData slotname; int active_pid = 0; + int active_pgprocno = INVALID_PROC_NUMBER; ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE; Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED)); @@ -1646,6 +1660,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, slotname = s->data.name; active_pid = s->active_pid; + active_pgprocno = s->active_pgprocno; /* * If the slot can be acquired, do so and mark it invalidated @@ -1710,9 +1725,8 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, oldestLSN, snapshotConflictHorizon); if (MyBackendType == B_STARTUP) - (void) SendProcSignal(active_pid, - PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT, - INVALID_PROC_NUMBER); + SendInterrupt(INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT, + active_pgprocno); else (void) kill(active_pid, SIGTERM); diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index fa5988c824e..724f67758e7 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -254,7 +254,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) * We do NOT reset ProcDiePending, so that the process will die after * the commit is cleaned up. */ - if (ProcDiePending) + if (InterruptIsPending(INTERRUPT_DIE)) { ereport(WARNING, (errcode(ERRCODE_ADMIN_SHUTDOWN), @@ -271,9 +271,8 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) * altogether is not helpful, so we just terminate the wait with a * suitable warning. */ - if (QueryCancelPending) + if (ConsumeInterrupt(INTERRUPT_QUERY_CANCEL)) { - QueryCancelPending = false; ereport(WARNING, (errmsg("canceling wait for synchronous replication due to user request"), errdetail("The transaction has already committed locally, but might not have been replicated to the standby."))); @@ -294,7 +293,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) */ if (rc & WL_POSTMASTER_DEATH) { - ProcDiePending = true; + RaiseInterrupt(INTERRUPT_DIE); whereToSendOutput = DestNone; SyncRepCancelWait(); break; diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index acda5f68d9a..6d103c440ff 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -283,7 +283,7 @@ WalReceiverMain(char *startup_data, size_t startup_data_len) /* SIGQUIT handler was already set up by InitPostmasterChild */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); /* Reset some signals that are accepted by postmaster but not here */ diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 754f505c139..d8c1f83c35b 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2920,6 +2920,7 @@ InitWalSenderSlot(void) * Found a free slot. Reserve it for us. */ walsnd->pid = MyProcPid; + walsnd->proc_number = MyProcNumber; walsnd->state = WALSNDSTATE_STARTUP; walsnd->sentPtr = InvalidXLogRecPtr; walsnd->needreload = false; @@ -3594,7 +3595,7 @@ WalSndSignals(void) /* SIGQUIT handler was already set up by InitPostmasterChild */ InitializeTimeouts(); /* establishes SIGALRM handler */ pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, WalSndLastCycleHandler); /* request a last cycle and * shutdown */ @@ -3748,15 +3749,17 @@ WalSndInitStopping(void) { WalSnd *walsnd = &WalSndCtl->walsnds[i]; pid_t pid; + int proc_number; SpinLockAcquire(&walsnd->mutex); pid = walsnd->pid; + proc_number = walsnd->proc_number; SpinLockRelease(&walsnd->mutex); if (pid == 0) continue; - SendProcSignal(pid, PROCSIG_WALSND_INIT_STOPPING, INVALID_PROC_NUMBER); + SendInterrupt(INTERRUPT_WALSND_INIT_STOPPING, proc_number); } } diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 61816730955..95e66d7bf8d 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -2966,7 +2966,7 @@ BufferSync(int flags) UnlockBufHdr(bufHdr, buf_state); /* Check for barrier events in case NBuffers is large. */ - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); } @@ -3047,7 +3047,7 @@ BufferSync(int flags) s->num_to_scan++; /* Check for barrier events. */ - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); } @@ -5246,7 +5246,7 @@ LockBufferForCleanup(Buffer buffer) * deadlock_timeout for it. */ if (logged_recovery_conflict) - LogRecoveryConflict(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN, + LogRecoveryConflict(INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN, waitStart, GetCurrentTimestamp(), NULL, false); @@ -5296,7 +5296,7 @@ LockBufferForCleanup(Buffer buffer) if (TimestampDifferenceExceeds(waitStart, now, DeadlockTimeout)) { - LogRecoveryConflict(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN, + LogRecoveryConflict(INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN, waitStart, now, NULL, true); logged_recovery_conflict = true; } diff --git a/src/backend/storage/ipc/ipc.c b/src/backend/storage/ipc/ipc.c index b06e4b84528..69beb12b70c 100644 --- a/src/backend/storage/ipc/ipc.c +++ b/src/backend/storage/ipc/ipc.c @@ -175,9 +175,9 @@ proc_exit_prepare(int code) * close up shop already. Note that the signal handlers will not set * these flags again, now that proc_exit_inprogress is set. */ - InterruptPending = false; - ProcDiePending = false; - QueryCancelPending = false; + SwitchToLocalInterrupts(); + ClearInterrupt(INTERRUPT_DIE); + ClearInterrupt(INTERRUPT_QUERY_CANCEL); InterruptHoldoffCount = 1; CritSectionCount = 0; diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 16b5803d388..7fb4cad0a64 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -3492,13 +3492,13 @@ GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid) * Returns pid of the process signaled, or 0 if not found. */ pid_t -CancelVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode) +CancelVirtualTransaction(VirtualTransactionId vxid, InterruptType sigmode) { return SignalVirtualTransaction(vxid, sigmode, true); } pid_t -SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode, +SignalVirtualTransaction(VirtualTransactionId vxid, InterruptType sigmode, bool conflictPending) { ProcArrayStruct *arrayP = procArray; @@ -3519,15 +3519,7 @@ SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode, procvxid.localTransactionId == vxid.localTransactionId) { proc->recoveryConflictPending = conflictPending; - pid = proc->pid; - if (pid != 0) - { - /* - * Kill the pid if it's still here. If not, that's what we - * wanted so ignore any errors. - */ - (void) SendProcSignal(pid, sigmode, vxid.procNumber); - } + SendInterrupt(sigmode, pgprocno); break; } } @@ -3661,7 +3653,7 @@ CountDBConnections(Oid databaseid) * CancelDBBackends --- cancel backends that are using specified database */ void -CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending) +CancelDBBackends(Oid databaseid, InterruptType sigmode, bool conflictPending) { ProcArrayStruct *arrayP = procArray; int index; @@ -3676,20 +3668,14 @@ CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending) if (databaseid == InvalidOid || proc->databaseId == databaseid) { - VirtualTransactionId procvxid; pid_t pid; - GET_VXID_FROM_PGPROC(procvxid, *proc); - proc->recoveryConflictPending = conflictPending; pid = proc->pid; if (pid != 0) { - /* - * Kill the pid if it's still here. If not, that's what we - * wanted so ignore any errors. - */ - (void) SendProcSignal(pid, sigmode, procvxid.procNumber); + /* Cancel the backend if it's still here. */ + SendInterrupt(sigmode, pgprocno); } } } diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index 4ed9cedcdd4..89638e15d16 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -34,22 +34,9 @@ #include "utils/memutils.h" /* - * The SIGUSR1 signal is multiplexed to support signaling multiple event - * types. The specific reason is communicated via flags in shared memory. - * We keep a boolean flag for each possible "reason", so that different - * reasons can be signaled to a process concurrently. (However, if the same - * reason is signaled more than once nearly simultaneously, the process may - * observe it only once.) + * State for the ProcSignalBarrier mechanism. * - * Each process that wants to receive signals registers its process ID - * in the ProcSignalSlots array. The array is indexed by ProcNumber to make - * slot allocation simple, and to avoid having to search the array when you - * know the ProcNumber of the process you're signaling. (We do support - * signaling without ProcNumber, but it's a bit less efficient.) - * - * The flags are actually declared as "volatile sig_atomic_t" for maximum - * portability. This should ensure that loads and stores of the flag - * values are atomic, allowing us to dispense with any explicit locking. + * XXX Rename all of this stuff? * * pss_signalFlags are intended to be set in cases where we don't need to * keep track of whether or not the target process has handled the signal, @@ -63,7 +50,7 @@ typedef struct { volatile pid_t pss_pid; - volatile sig_atomic_t pss_signalFlags[NUM_PROCSIGNALS]; + volatile int pss_pgprocno; pg_atomic_uint64 pss_barrierGeneration; pg_atomic_uint32 pss_barrierCheckMask; ConditionVariable pss_barrierCV; @@ -99,7 +86,6 @@ typedef struct static ProcSignalHeader *ProcSignal = NULL; static ProcSignalSlot *MyProcSignalSlot = NULL; -static bool CheckProcSignal(ProcSignalReason reason); static void CleanupProcSignalState(int status, Datum arg); static void ResetProcSignalBarrierBits(uint32 flags); @@ -142,7 +128,7 @@ ProcSignalShmemInit(void) ProcSignalSlot *slot = &ProcSignal->psh_slot[i]; slot->pss_pid = 0; - MemSet(slot->pss_signalFlags, 0, sizeof(slot->pss_signalFlags)); + slot->pss_pgprocno = INVALID_PROC_NUMBER; pg_atomic_init_u64(&slot->pss_barrierGeneration, PG_UINT64_MAX); pg_atomic_init_u32(&slot->pss_barrierCheckMask, 0); ConditionVariableInit(&slot->pss_barrierCV); @@ -171,9 +157,6 @@ ProcSignalInit(void) elog(LOG, "process %d taking over ProcSignal slot %d, but it's not empty", MyProcPid, MyProcNumber); - /* Clear out any leftover signal reasons */ - MemSet(slot->pss_signalFlags, 0, NUM_PROCSIGNALS * sizeof(sig_atomic_t)); - /* * Initialize barrier state. Since we're a brand-new process, there * shouldn't be any leftover backend-private state that needs to be @@ -191,8 +174,9 @@ ProcSignalInit(void) pg_atomic_write_u64(&slot->pss_barrierGeneration, barrier_generation); pg_memory_barrier(); - /* Mark slot with my PID */ + /* Mark slot with my procno, so my latch will be set for proc signals */ slot->pss_pid = MyProcPid; + slot->pss_pgprocno = MyProcNumber; /* Remember slot location for CheckProcSignal */ MyProcSignalSlot = slot; @@ -221,14 +205,14 @@ CleanupProcSignalState(int status, Datum arg) MyProcSignalSlot = NULL; /* sanity check */ - if (slot->pss_pid != MyProcPid) + if (slot->pss_pgprocno != MyProcNumber) { /* * don't ERROR here. We're exiting anyway, and don't want to get into * infinite loop trying to exit */ elog(LOG, "process %d releasing ProcSignal slot %d, but it contains %d", - MyProcPid, (int) (slot - ProcSignal->psh_slot), (int) slot->pss_pid); + MyProcPid, (int) (slot - ProcSignal->psh_slot), slot->pss_pgprocno); return; /* XXX better to zero the slot anyway? */ } @@ -240,73 +224,7 @@ CleanupProcSignalState(int status, Datum arg) ConditionVariableBroadcast(&slot->pss_barrierCV); slot->pss_pid = 0; -} - -/* - * SendProcSignal - * Send a signal to a Postgres process - * - * Providing procNumber is optional, but it will speed up the operation. - * - * On success (a signal was sent), zero is returned. - * On error, -1 is returned, and errno is set (typically to ESRCH or EPERM). - * - * Not to be confused with ProcSendSignal - */ -int -SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber) -{ - volatile ProcSignalSlot *slot; - - if (procNumber != INVALID_PROC_NUMBER) - { - slot = &ProcSignal->psh_slot[procNumber]; - - /* - * Note: Since there's no locking, it's possible that the target - * process detaches from shared memory and exits right after this - * test, before we set the flag and send signal. And the signal slot - * might even be recycled by a new process, so it's remotely possible - * that we set a flag for a wrong process. That's OK, all the signals - * are such that no harm is done if they're mistakenly fired. - */ - if (slot->pss_pid == pid) - { - /* Atomically set the proper flag */ - slot->pss_signalFlags[reason] = true; - /* Send signal */ - return kill(pid, SIGUSR1); - } - } - else - { - /* - * Pronumber not provided, so search the array using pid. We search - * the array back to front so as to reduce search overhead. Passing - * INVALID_PROC_NUMBER means that the target is most likely an - * auxiliary process, which will have a slot near the end of the - * array. - */ - int i; - - for (i = NumProcSignalSlots - 1; i >= 0; i--) - { - slot = &ProcSignal->psh_slot[i]; - - if (slot->pss_pid == pid) - { - /* the above note about race conditions applies here too */ - - /* Atomically set the proper flag */ - slot->pss_signalFlags[reason] = true; - /* Send signal */ - return kill(pid, SIGUSR1); - } - } - } - - errno = ESRCH; - return -1; + slot->pss_pgprocno = 0; } /* @@ -353,8 +271,8 @@ EmitProcSignalBarrier(ProcSignalBarrierType type) pg_atomic_add_fetch_u64(&ProcSignal->psh_barrierGeneration, 1); /* - * Signal all the processes, so that they update their advertised barrier - * generation. + * Interrupt all the processes, so that they update their advertised + * barrier generation. * * Concurrency is not a problem here. Backends that have exited don't * matter, and new backends that have joined since we entered this @@ -365,18 +283,8 @@ EmitProcSignalBarrier(ProcSignalBarrierType type) * backends that need to update state - but they won't actually need to * change any state. */ - for (int i = NumProcSignalSlots - 1; i >= 0; i--) - { - volatile ProcSignalSlot *slot = &ProcSignal->psh_slot[i]; - pid_t pid = slot->pss_pid; - - if (pid != 0) - { - /* see SendProcSignal for details */ - slot->pss_signalFlags[PROCSIG_BARRIER] = true; - kill(pid, SIGUSR1); - } - } + for (int i = 0; i < ProcGlobal->allProcCount; ++i) + SendInterrupt(INTERRUPT_BARRIER, i); return generation; } @@ -435,23 +343,6 @@ WaitForProcSignalBarrier(uint64 generation) pg_memory_barrier(); } -/* - * Handle receipt of an interrupt indicating a global barrier event. - * - * All the actual work is deferred to ProcessProcSignalBarrier(), because we - * cannot safely access the barrier generation inside the signal handler as - * 64bit atomics might use spinlock based emulation, even for reads. As this - * routine only gets called when PROCSIG_BARRIER is sent that won't cause a - * lot of unnecessary work. - */ -static void -HandleProcSignalBarrierInterrupt(void) -{ - InterruptPending = true; - ProcSignalBarrierPending = true; - /* latch will be set by procsignal_sigusr1_handler */ -} - /* * Perform global barrier related interrupt checking. * @@ -469,11 +360,6 @@ ProcessProcSignalBarrier(void) Assert(MyProcSignalSlot); - /* Exit quickly if there's no work to do. */ - if (!ProcSignalBarrierPending) - return; - ProcSignalBarrierPending = false; - /* * It's not unlikely to process multiple barriers at once, before the * signals for all the barriers have arrived. To avoid unnecessary work in @@ -601,80 +487,5 @@ static void ResetProcSignalBarrierBits(uint32 flags) { pg_atomic_fetch_or_u32(&MyProcSignalSlot->pss_barrierCheckMask, flags); - ProcSignalBarrierPending = true; - InterruptPending = true; -} - -/* - * CheckProcSignal - check to see if a particular reason has been - * signaled, and clear the signal flag. Should be called after receiving - * SIGUSR1. - */ -static bool -CheckProcSignal(ProcSignalReason reason) -{ - volatile ProcSignalSlot *slot = MyProcSignalSlot; - - if (slot != NULL) - { - /* Careful here --- don't clear flag if we haven't seen it set */ - if (slot->pss_signalFlags[reason]) - { - slot->pss_signalFlags[reason] = false; - return true; - } - } - - return false; -} - -/* - * procsignal_sigusr1_handler - handle SIGUSR1 signal. - */ -void -procsignal_sigusr1_handler(SIGNAL_ARGS) -{ - if (CheckProcSignal(PROCSIG_CATCHUP_INTERRUPT)) - HandleCatchupInterrupt(); - - if (CheckProcSignal(PROCSIG_NOTIFY_INTERRUPT)) - HandleNotifyInterrupt(); - - if (CheckProcSignal(PROCSIG_PARALLEL_MESSAGE)) - HandleParallelMessageInterrupt(); - - if (CheckProcSignal(PROCSIG_WALSND_INIT_STOPPING)) - HandleWalSndInitStopping(); - - if (CheckProcSignal(PROCSIG_BARRIER)) - HandleProcSignalBarrierInterrupt(); - - if (CheckProcSignal(PROCSIG_LOG_MEMORY_CONTEXT)) - HandleLogMemoryContextInterrupt(); - - if (CheckProcSignal(PROCSIG_PARALLEL_APPLY_MESSAGE)) - HandleParallelApplyMessageInterrupt(); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_DATABASE)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_DATABASE); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_TABLESPACE)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_TABLESPACE); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOCK)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOCK); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK); - - if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN)) - HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN); - - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_BARRIER); } diff --git a/src/backend/storage/ipc/signalfuncs.c b/src/backend/storage/ipc/signalfuncs.c index 88e9bf8125d..5fdc44e4345 100644 --- a/src/backend/storage/ipc/signalfuncs.c +++ b/src/backend/storage/ipc/signalfuncs.c @@ -109,6 +109,7 @@ pg_signal_backend(int pid, int sig) (errmsg("could not send signal to process %d: %m", pid))); return SIGNAL_BACKEND_ERROR; } + return SIGNAL_BACKEND_SUCCESS; } diff --git a/src/backend/storage/ipc/sinval.c b/src/backend/storage/ipc/sinval.c index d9b16f84d19..70732782b95 100644 --- a/src/backend/storage/ipc/sinval.c +++ b/src/backend/storage/ipc/sinval.c @@ -16,7 +16,9 @@ #include "access/xact.h" #include "miscadmin.h" -#include "storage/latch.h" +#include "postmaster/interrupt.h" +#include "storage/ipc.h" +#include "storage/proc.h" #include "storage/sinvaladt.h" #include "utils/inval.h" @@ -24,21 +26,6 @@ uint64 SharedInvalidMessageCounter; -/* - * 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 - * PROCSIG_CATCHUP_INTERRUPT to any backend that gets too far behind. - * - * The signal handler will set an interrupt pending flag and will set the - * processes latch. Whenever starting to read from the client, or when - * interrupted while doing so, ProcessClientReadInterrupt() will call - * ProcessCatchupEvent(). - */ -volatile sig_atomic_t catchupInterruptPending = false; - - /* * SendSharedInvalidMessages * Add shared-cache-invalidation message(s) to the global SI message queue. @@ -132,48 +119,29 @@ ReceiveSharedInvalidMessages(void (*invalFunction) (SharedInvalidationMessage *m * catchup signal this way avoids creating spikes in system load for what * should be just a background maintenance activity. */ - if (catchupInterruptPending) + if (ConsumeInterrupt(INTERRUPT_SINVAL_CATCHUP)) { - catchupInterruptPending = false; elog(DEBUG4, "sinval catchup complete, cleaning queue"); SICleanupQueue(false, 0); } } - -/* - * HandleCatchupInterrupt - * - * This is called when PROCSIG_CATCHUP_INTERRUPT is received. - * - * We used to directly call ProcessCatchupEvent directly when idle. These days - * we just set a flag to do it later and notify the process of that fact by - * setting the process's latch. - */ -void -HandleCatchupInterrupt(void) -{ - /* - * Note: this is called by a SIGNAL HANDLER. You must be very wary what - * you do here. - */ - - catchupInterruptPending = true; - - /* make sure the event is processed in due course */ - SetLatch(MyLatch); -} - /* * ProcessCatchupInterrupt * - * The portion of catchup interrupt handling that runs outside of the signal - * handler, which allows it to actually process pending invalidations. + * Called by ProcessClientReadInterrupt() if another backend asks us to catch + * up. + * + * 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. */ void ProcessCatchupInterrupt(void) { - while (catchupInterruptPending) + do { /* * What we need to do here is cause ReceiveSharedInvalidMessages() to @@ -199,5 +167,5 @@ ProcessCatchupInterrupt(void) StartTransactionCommand(); CommitTransactionCommand(); } - } + } while (ConsumeInterrupt(INTERRUPT_SINVAL_CATCHUP)); } diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c index b486d8ddd1d..a84d2891e99 100644 --- a/src/backend/storage/ipc/sinvaladt.c +++ b/src/backend/storage/ipc/sinvaladt.c @@ -19,6 +19,7 @@ #include "access/transam.h" #include "miscadmin.h" +#include "postmaster/interrupt.h" #include "storage/ipc.h" #include "storage/proc.h" #include "storage/procnumber.h" @@ -138,7 +139,8 @@ typedef struct ProcState { /* procPid is zero in an inactive ProcState array entry. */ - pid_t procPid; /* PID of backend, for signaling */ + pid_t procPid; /* pid of backend */ + ProcNumber pgprocno; /* for sending interrupts */ /* nextMsgNum is meaningless if procPid == 0 or resetState is true. */ int nextMsgNum; /* next message number to read */ bool resetState; /* backend needs to reset its state */ @@ -254,6 +256,7 @@ CreateSharedInvalidationState(void) for (i = 0; i < NumProcStateSlots; i++) { shmInvalBuffer->procState[i].procPid = 0; /* inactive */ + shmInvalBuffer->procState[i].pgprocno = INVALID_PROC_NUMBER; shmInvalBuffer->procState[i].nextMsgNum = 0; /* meaningless */ shmInvalBuffer->procState[i].resetState = false; shmInvalBuffer->procState[i].signaled = false; @@ -304,6 +307,7 @@ SharedInvalBackendInit(bool sendOnly) /* mark myself active, with all extant messages already read */ stateP->procPid = MyProcPid; + stateP->pgprocno = MyProcNumber; stateP->nextMsgNum = segP->maxMsgNum; stateP->resetState = false; stateP->signaled = false; @@ -342,6 +346,7 @@ CleanupInvalidationState(int status, Datum arg) /* Mark myself inactive */ stateP->procPid = 0; + stateP->pgprocno = INVALID_PROC_NUMBER; stateP->nextMsgNum = 0; stateP->resetState = false; stateP->signaled = false; @@ -659,19 +664,16 @@ SICleanupQueue(bool callerHasWriteLock, int minFree) /* * Lastly, signal anyone who needs a catchup interrupt. Since - * SendProcSignal() might not be fast, we don't want to hold locks while + * SendInterrupt() might not be fast, we don't want to hold locks while * executing it. */ if (needSig) { - pid_t his_pid = needSig->procPid; - ProcNumber his_procNumber = (needSig - &segP->procState[0]); - needSig->signaled = true; LWLockRelease(SInvalReadLock); LWLockRelease(SInvalWriteLock); - elog(DEBUG4, "sending sinval catchup signal to PID %d", (int) his_pid); - SendProcSignal(his_pid, PROCSIG_CATCHUP_INTERRUPT, his_procNumber); + elog(DEBUG4, "sending sinval catchup interrupt to PID %d", (int) needSig->procPid); + SendInterrupt(INTERRUPT_SINVAL_CATCHUP, needSig->pgprocno); if (callerHasWriteLock) LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE); } diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 872679ca447..dbe76a37ea0 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -70,13 +70,13 @@ static volatile sig_atomic_t got_standby_delay_timeout = false; static volatile sig_atomic_t got_standby_lock_timeout = false; static void ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist, - ProcSignalReason reason, + InterruptType reason, uint32 wait_event_info, bool report_waiting); -static void SendRecoveryConflictWithBufferPin(ProcSignalReason reason); +static void SendRecoveryConflictWithBufferPin(InterruptType reason); static XLogRecPtr LogCurrentRunningXacts(RunningTransactions CurrRunningXacts); static void LogAccessExclusiveLocks(int nlocks, xl_standby_lock *locks); -static const char *get_recovery_conflict_desc(ProcSignalReason reason); +static const char *get_recovery_conflict_desc(InterruptType reason); /* * InitRecoveryTransactionEnvironment @@ -270,7 +270,7 @@ WaitExceedsMaxStandbyDelay(uint32 wait_event_info) * to be resolved or not. */ void -LogRecoveryConflict(ProcSignalReason reason, TimestampTz wait_start, +LogRecoveryConflict(InterruptType reason, TimestampTz wait_start, TimestampTz now, VirtualTransactionId *wait_list, bool still_waiting) { @@ -357,7 +357,7 @@ LogRecoveryConflict(ProcSignalReason reason, TimestampTz wait_start, */ static void ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist, - ProcSignalReason reason, uint32 wait_event_info, + InterruptType reason, uint32 wait_event_info, bool report_waiting) { TimestampTz waitStart = 0; @@ -488,7 +488,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId snapshotConflictHorizon, backends = GetConflictingVirtualXIDs(snapshotConflictHorizon, locator.dbOid); ResolveRecoveryConflictWithVirtualXIDs(backends, - PROCSIG_RECOVERY_CONFLICT_SNAPSHOT, + INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT, WAIT_EVENT_RECOVERY_CONFLICT_SNAPSHOT, true); @@ -559,7 +559,7 @@ ResolveRecoveryConflictWithTablespace(Oid tsid) temp_file_users = GetConflictingVirtualXIDs(InvalidTransactionId, InvalidOid); ResolveRecoveryConflictWithVirtualXIDs(temp_file_users, - PROCSIG_RECOVERY_CONFLICT_TABLESPACE, + INTERRUPT_RECOVERY_CONFLICT_TABLESPACE, WAIT_EVENT_RECOVERY_CONFLICT_TABLESPACE, true); } @@ -580,7 +580,7 @@ ResolveRecoveryConflictWithDatabase(Oid dbid) */ while (CountDBBackends(dbid) > 0) { - CancelDBBackends(dbid, PROCSIG_RECOVERY_CONFLICT_DATABASE, true); + CancelDBBackends(dbid, INTERRUPT_RECOVERY_CONFLICT_DATABASE, true); /* * Wait awhile for them to die so that we avoid flooding an @@ -664,7 +664,7 @@ ResolveRecoveryConflictWithLock(LOCKTAG locktag, bool logging_conflict) * because the caller, WaitOnLock(), has already reported that. */ ResolveRecoveryConflictWithVirtualXIDs(backends, - PROCSIG_RECOVERY_CONFLICT_LOCK, + INTERRUPT_RECOVERY_CONFLICT_LOCK, PG_WAIT_LOCK | locktag.locktag_type, false); } @@ -723,7 +723,7 @@ ResolveRecoveryConflictWithLock(LOCKTAG locktag, bool logging_conflict) while (VirtualTransactionIdIsValid(*backends)) { SignalVirtualTransaction(*backends, - PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK, + INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK, false); backends++; } @@ -802,7 +802,7 @@ ResolveRecoveryConflictWithBufferPin(void) /* * We're already behind, so clear a path as quickly as possible. */ - SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN); + SendRecoveryConflictWithBufferPin(INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN); } else { @@ -842,7 +842,7 @@ ResolveRecoveryConflictWithBufferPin(void) ProcWaitForSignal(WAIT_EVENT_BUFFER_PIN); if (got_standby_delay_timeout) - SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN); + SendRecoveryConflictWithBufferPin(INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN); else if (got_standby_deadlock_timeout) { /* @@ -858,7 +858,7 @@ ResolveRecoveryConflictWithBufferPin(void) * not be so harmful because the period that the buffer is kept pinned * is basically no so long. But we should fix this? */ - SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK); + SendRecoveryConflictWithBufferPin(INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK); } /* @@ -873,10 +873,10 @@ ResolveRecoveryConflictWithBufferPin(void) } static void -SendRecoveryConflictWithBufferPin(ProcSignalReason reason) +SendRecoveryConflictWithBufferPin(InterruptType reason) { - Assert(reason == PROCSIG_RECOVERY_CONFLICT_BUFFERPIN || - reason == PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK); + Assert(reason == INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN || + reason == INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK); /* * We send signal to all backends to ask them if they are holding the @@ -1481,31 +1481,31 @@ LogStandbyInvalidations(int nmsgs, SharedInvalidationMessage *msgs, /* Return the description of recovery conflict */ static const char * -get_recovery_conflict_desc(ProcSignalReason reason) +get_recovery_conflict_desc(InterruptType reason) { const char *reasonDesc = _("unknown reason"); switch (reason) { - case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN: + case INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN: reasonDesc = _("recovery conflict on buffer pin"); break; - case PROCSIG_RECOVERY_CONFLICT_LOCK: + case INTERRUPT_RECOVERY_CONFLICT_LOCK: reasonDesc = _("recovery conflict on lock"); break; - case PROCSIG_RECOVERY_CONFLICT_TABLESPACE: + case INTERRUPT_RECOVERY_CONFLICT_TABLESPACE: reasonDesc = _("recovery conflict on tablespace"); break; - case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT: + case INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT: reasonDesc = _("recovery conflict on snapshot"); break; - case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT: + case INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT: reasonDesc = _("recovery conflict on replication slot"); break; - case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK: + case INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK: reasonDesc = _("recovery conflict on buffer deadlock"); break; - case PROCSIG_RECOVERY_CONFLICT_DATABASE: + case INTERRUPT_RECOVERY_CONFLICT_DATABASE: reasonDesc = _("recovery conflict on database"); break; default: diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 08bfa485b50..0291e111bb4 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -444,6 +444,10 @@ InitProcess(void) OwnLatch(&MyProc->procLatch); SwitchToSharedLatch(); + /* We're now ready to accept interrupts from other processes. */ + pg_atomic_init_u32(&MyProc->pending_interrupts, 0); + SwitchToSharedInterrupts(); + /* now that we have a proc, report wait events to shared memory */ pgstat_set_wait_event_storage(&MyProc->wait_event_info); @@ -611,6 +615,10 @@ InitAuxiliaryProcess(void) OwnLatch(&MyProc->procLatch); SwitchToSharedLatch(); + /* We're now ready to accept interrupts from other processes. */ + pg_atomic_init_u32(&MyProc->pending_interrupts, 0); + SwitchToSharedInterrupts(); + /* now that we have a proc, report wait events to shared memory */ pgstat_set_wait_event_storage(&MyProc->wait_event_info); @@ -986,6 +994,9 @@ AuxiliaryProcKill(int code, Datum arg) /* Cancel any pending condition variable sleep, too */ ConditionVariableCancelSleep(); + /* Revert to local interrupt vector. */ + SwitchToLocalInterrupts(); + /* look at the equivalent ProcKill() code for comments */ SwitchBackToLocalLatch(); pgstat_reset_wait_event_storage(); @@ -1343,7 +1354,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait) * because the startup process here has already waited * longer than deadlock_timeout. */ - LogRecoveryConflict(PROCSIG_RECOVERY_CONFLICT_LOCK, + LogRecoveryConflict(INTERRUPT_RECOVERY_CONFLICT_LOCK, standbyWaitStart, now, cnt > 0 ? vxids : NULL, true); logged_recovery_conflict = true; @@ -1632,7 +1643,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait) * startup process waited longer than deadlock_timeout for it. */ if (InHotStandby && logged_recovery_conflict) - LogRecoveryConflict(PROCSIG_RECOVERY_CONFLICT_LOCK, + LogRecoveryConflict(INTERRUPT_RECOVERY_CONFLICT_LOCK, standbyWaitStart, GetCurrentTimestamp(), NULL, false); diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index e39c6804a73..3c462dd9c81 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -162,10 +162,6 @@ static const char *userDoption = NULL; /* -D switch */ static bool EchoQuery = false; /* -E switch */ static bool UseSemiNewlineNewline = false; /* -j switch */ -/* whether or not, and why, we were canceled by conflict with recovery */ -static volatile sig_atomic_t RecoveryConflictPending = false; -static volatile sig_atomic_t RecoveryConflictPendingReasons[NUM_PROCSIGNALS]; - /* reused buffer to pass to SendRowDescriptionMessage() */ static MemoryContext row_description_context = NULL; static StringInfoData row_description_buf; @@ -516,14 +512,14 @@ ProcessClientReadInterrupt(bool blocked) CHECK_FOR_INTERRUPTS(); /* Process sinval catchup interrupts, if any */ - if (catchupInterruptPending) + if (InterruptIsPending(INTERRUPT_SINVAL_CATCHUP)) ProcessCatchupInterrupt(); /* Process notify interrupts, if any */ - if (notifyInterruptPending) + if (InterruptIsPending(INTERRUPT_NOTIFY)) ProcessNotifyInterrupt(true); } - else if (ProcDiePending) + else if (InterruptIsPending(INTERRUPT_DIE)) { /* * We're dying. If there is no data available to read, then it's safe @@ -556,7 +552,7 @@ ProcessClientWriteInterrupt(bool blocked) { int save_errno = errno; - if (ProcDiePending) + if (InterruptIsPending(INTERRUPT_DIE)) { /* * We're dying. If it's not possible to write, then we should handle @@ -2506,29 +2502,29 @@ errdetail_abort(void) * Add an errdetail() line showing conflict source. */ static int -errdetail_recovery_conflict(ProcSignalReason reason) +errdetail_recovery_conflict(InterruptType reason) { switch (reason) { - case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN: + case INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN: errdetail("User was holding shared buffer pin for too long."); break; - case PROCSIG_RECOVERY_CONFLICT_LOCK: + case INTERRUPT_RECOVERY_CONFLICT_LOCK: errdetail("User was holding a relation lock for too long."); break; - case PROCSIG_RECOVERY_CONFLICT_TABLESPACE: + case INTERRUPT_RECOVERY_CONFLICT_TABLESPACE: errdetail("User was or might have been using tablespace that must be dropped."); break; - case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT: + case INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT: errdetail("User query might have needed to see row versions that must be removed."); break; - case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT: + case INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT: errdetail("User was using a logical replication slot that must be invalidated."); break; - case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK: + case INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK: errdetail("User transaction caused buffer deadlock with recovery."); break; - case PROCSIG_RECOVERY_CONFLICT_DATABASE: + case INTERRUPT_RECOVERY_CONFLICT_DATABASE: errdetail("User was connected to a database that must be dropped."); break; default: @@ -2973,10 +2969,7 @@ die(SIGNAL_ARGS) { /* Don't joggle the elbow of proc_exit */ if (!proc_exit_inprogress) - { - InterruptPending = true; - ProcDiePending = true; - } + RaiseInterrupt(INTERRUPT_DIE); /* for the cumulative stats system */ pgStatSessionEndCause = DISCONNECT_KILLED; @@ -3005,10 +2998,7 @@ StatementCancelHandler(SIGNAL_ARGS) * Don't joggle the elbow of proc_exit */ if (!proc_exit_inprogress) - { - InterruptPending = true; - QueryCancelPending = true; - } + RaiseInterrupt(INTERRUPT_QUERY_CANCEL); /* If we're still here, waken anything waiting on the process latch */ SetLatch(MyLatch); @@ -3028,27 +3018,14 @@ FloatExceptionHandler(SIGNAL_ARGS) } /* - * Tell the next CHECK_FOR_INTERRUPTS() to check for a particular type of - * recovery conflict. Runs in a SIGUSR1 handler. - */ -void -HandleRecoveryConflictInterrupt(ProcSignalReason reason) -{ - RecoveryConflictPendingReasons[reason] = true; - RecoveryConflictPending = true; - InterruptPending = true; - /* latch will be set by procsignal_sigusr1_handler */ -} - -/* - * Check one individual conflict reason. + * Process one individual conflict reason. */ static void -ProcessRecoveryConflictInterrupt(ProcSignalReason reason) +ProcessRecoveryConflictInterrupt(InterruptType reason) { switch (reason) { - case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK: + case INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK: /* * If we aren't waiting for a lock we can never deadlock. @@ -3059,21 +3036,21 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) /* Intentional fall through to check wait for pin */ /* FALLTHROUGH */ - case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN: + case INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN: /* - * If PROCSIG_RECOVERY_CONFLICT_BUFFERPIN is requested but we + * If INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN is requested but we * aren't blocking the Startup process there is nothing more to * do. * - * When PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK is requested, + * When INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK is requested, * if we're waiting for locks and the startup process is not * waiting for buffer pin (i.e., also waiting for locks), we set * the flag so that ProcSleep() will check for deadlocks. */ if (!HoldingBufferPinThatDelaysRecovery()) { - if (reason == PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK && + if (reason == INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK && GetStartupBufferPinWaitBufId() < 0) CheckDeadLockAlert(); return; @@ -3084,9 +3061,9 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) /* Intentional fall through to error handling */ /* FALLTHROUGH */ - case PROCSIG_RECOVERY_CONFLICT_LOCK: - case PROCSIG_RECOVERY_CONFLICT_TABLESPACE: - case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT: + case INTERRUPT_RECOVERY_CONFLICT_LOCK: + case INTERRUPT_RECOVERY_CONFLICT_TABLESPACE: + case INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT: /* * If we aren't in a transaction any longer then ignore. @@ -3096,14 +3073,14 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) /* FALLTHROUGH */ - case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT: + case INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT: /* * If we're not in a subtransaction then we are OK to throw an * ERROR to resolve the conflict. Otherwise drop through to the * FATAL case. * - * PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT is a special case that + * INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT is a special case that * always throws an ERROR (ie never promotes to FATAL), though it * still has to respect QueryCancelHoldoffCount, so it shares this * code path. Logical decoding slots are only acquired while @@ -3113,17 +3090,17 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) * intercept an error before the replication slot is released. * * XXX other times that we can throw just an ERROR *may* be - * PROCSIG_RECOVERY_CONFLICT_LOCK if no locks are held in parent + * INTERRUPT_RECOVERY_CONFLICT_LOCK if no locks are held in parent * transactions * - * PROCSIG_RECOVERY_CONFLICT_SNAPSHOT if no snapshots are held by - * parent transactions and the transaction is not + * INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT if no snapshots are held + * by parent transactions and the transaction is not * transaction-snapshot mode * - * PROCSIG_RECOVERY_CONFLICT_TABLESPACE if no temp files or + * INTERRUPT_RECOVERY_CONFLICT_TABLESPACE if no temp files or * cursors open in parent transactions */ - if (reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT || + if (reason == INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT || !IsSubTransaction()) { /* @@ -3150,9 +3127,7 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) * Re-arm and defer this interrupt until later. See * similar code in ProcessInterrupts(). */ - RecoveryConflictPendingReasons[reason] = true; - RecoveryConflictPending = true; - InterruptPending = true; + RaiseInterrupt(reason); return; } @@ -3175,7 +3150,7 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) /* Intentional fall through to session cancel */ /* FALLTHROUGH */ - case PROCSIG_RECOVERY_CONFLICT_DATABASE: + case INTERRUPT_RECOVERY_CONFLICT_DATABASE: /* * Retrying is not possible because the database is dropped, or we @@ -3184,7 +3159,7 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) */ pgstat_report_recovery_conflict(reason); ereport(FATAL, - (errcode(reason == PROCSIG_RECOVERY_CONFLICT_DATABASE ? + (errcode(reason == INTERRUPT_RECOVERY_CONFLICT_DATABASE ? ERRCODE_DATABASE_DROPPED : ERRCODE_T_R_SERIALIZATION_FAILURE), errmsg("terminating connection due to conflict with recovery"), @@ -3198,47 +3173,18 @@ ProcessRecoveryConflictInterrupt(ProcSignalReason reason) } } -/* - * Check each possible recovery conflict reason. - */ -static void -ProcessRecoveryConflictInterrupts(void) -{ - /* - * We don't need to worry about joggling the elbow of proc_exit, because - * proc_exit_prepare() holds interrupts, so ProcessInterrupts() won't call - * us. - */ - Assert(!proc_exit_inprogress); - Assert(InterruptHoldoffCount == 0); - Assert(RecoveryConflictPending); - - RecoveryConflictPending = false; - - for (ProcSignalReason reason = PROCSIG_RECOVERY_CONFLICT_FIRST; - reason <= PROCSIG_RECOVERY_CONFLICT_LAST; - reason++) - { - if (RecoveryConflictPendingReasons[reason]) - { - RecoveryConflictPendingReasons[reason] = false; - ProcessRecoveryConflictInterrupt(reason); - } - } -} - /* * 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 accept the interrupt. Called only when - * InterruptPending is true. + * InterruptIsPending() is true for one of the "standard" interrupts. * * Note: if INTERRUPTS_CAN_BE_PROCESSED() is true, then ProcessInterrupts - * is guaranteed to clear the InterruptPending flag 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.) + * is guaranteed to clear all pending flag 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) @@ -3246,12 +3192,12 @@ ProcessInterrupts(void) /* OK to accept any interrupts now? */ if (InterruptHoldoffCount != 0 || CritSectionCount != 0) return; - InterruptPending = false; - if (ProcDiePending) + pg_read_barrier(); + + if (ConsumeInterrupt(INTERRUPT_DIE)) { - ProcDiePending = false; - QueryCancelPending = false; /* ProcDie trumps QueryCancel */ + ClearInterrupt(INTERRUPT_QUERY_CANCEL); /* ProcDie trumps QueryCancel */ LockErrorCleanup(); /* As in quickdie, don't risk sending to client during auth */ if (ClientAuthInProgress && whereToSendOutput == DestRemote) @@ -3290,10 +3236,8 @@ ProcessInterrupts(void) errmsg("terminating connection due to administrator command"))); } - if (CheckClientConnectionPending) + if (ConsumeInterrupt(INTERRUPT_CHECK_CONNECTION_TIMEOUT)) { - CheckClientConnectionPending = false; - /* * Check for lost connection and re-arm, if still configured, but not * if we've arrived back at DoingCommandRead state. We don't want to @@ -3303,16 +3247,16 @@ ProcessInterrupts(void) if (!DoingCommandRead && client_connection_check_interval > 0) { if (!pq_check_connection()) - ClientConnectionLost = true; + RaiseInterrupt(INTERRUPT_CONNECTION_LOST); else enable_timeout_after(CLIENT_CONNECTION_CHECK_TIMEOUT, client_connection_check_interval); } } - if (ClientConnectionLost) + if (ConsumeInterrupt(INTERRUPT_CONNECTION_LOST)) { - QueryCancelPending = false; /* lost connection trumps QueryCancel */ + ClearInterrupt(INTERRUPT_QUERY_CANCEL); /* lost connection trumps */ LockErrorCleanup(); /* don't send to client, we already know the connection to be dead. */ whereToSendOutput = DestNone; @@ -3329,25 +3273,24 @@ ProcessInterrupts(void) * * See similar logic in ProcessRecoveryConflictInterrupts(). */ - if (QueryCancelPending && QueryCancelHoldoffCount != 0) + if (QueryCancelHoldoffCount != 0 && + InterruptIsPending(INTERRUPT_QUERY_CANCEL)) { /* - * Re-arm InterruptPending so that we process the cancel request as + * Leave interrupt pending so that we process the cancel request as * soon as we're done reading the message. (XXX this is seriously * ugly: it complicates INTERRUPTS_CAN_BE_PROCESSED(), and it means we * can't use that macro directly as the initial test in this function, * meaning that this code also creates opportunities for other bugs to * appear.) */ - InterruptPending = true; + RaiseInterrupt(INTERRUPT_QUERY_CANCEL); } - else if (QueryCancelPending) + else if (ConsumeInterrupt(INTERRUPT_QUERY_CANCEL)) { bool lock_timeout_occurred; bool stmt_timeout_occurred; - QueryCancelPending = false; - /* * If LOCK_TIMEOUT and STATEMENT_TIMEOUT indicators are both set, we * need to clear both, so always fetch both. @@ -3401,10 +3344,16 @@ ProcessInterrupts(void) } } - if (RecoveryConflictPending) - ProcessRecoveryConflictInterrupts(); + /* Check all of the individual recovery conflict reasons. */ + for (InterruptType reason = INTERRUPT_RECOVERY_CONFLICT_FIRST; + reason <= INTERRUPT_RECOVERY_CONFLICT_LAST; + reason++) + { + if (ConsumeInterrupt(reason)) + ProcessRecoveryConflictInterrupt(reason); + } - if (IdleInTransactionSessionTimeoutPending) + if (ConsumeInterrupt(INTERRUPT_IDLE_TRANSACTION_TIMEOUT)) { /* * If the GUC has been reset to zero, ignore the signal. This is @@ -3412,7 +3361,6 @@ ProcessInterrupts(void) * interrupt. We need to unset the flag before the injection point, * otherwise we could loop in interrupts checking. */ - IdleInTransactionSessionTimeoutPending = false; if (IdleInTransactionSessionTimeout > 0) { INJECTION_POINT("idle-in-transaction-session-timeout"); @@ -3422,10 +3370,9 @@ ProcessInterrupts(void) } } - if (TransactionTimeoutPending) + if (ConsumeInterrupt(INTERRUPT_TRANSACTION_TIMEOUT)) { /* As above, ignore the signal if the GUC has been reset to zero. */ - TransactionTimeoutPending = false; if (TransactionTimeout > 0) { INJECTION_POINT("transaction-timeout"); @@ -3435,10 +3382,9 @@ ProcessInterrupts(void) } } - if (IdleSessionTimeoutPending) + if (ConsumeInterrupt(INTERRUPT_IDLE_SESSION_TIMEOUT)) { /* As above, ignore the signal if the GUC has been reset to zero. */ - IdleSessionTimeoutPending = false; if (IdleSessionTimeout > 0) { INJECTION_POINT("idle-session-timeout"); @@ -3452,23 +3398,23 @@ ProcessInterrupts(void) * If there are pending stats updates and we currently are truly idle * (matching the conditions in PostgresMain(), report stats now. */ - if (IdleStatsUpdateTimeoutPending && - DoingCommandRead && !IsTransactionOrTransactionBlock()) - { - IdleStatsUpdateTimeoutPending = false; + if (DoingCommandRead && !IsTransactionOrTransactionBlock() && + ConsumeInterrupt(INTERRUPT_IDLE_STATS_UPDATE_TIMEOUT)) pgstat_report_stat(true); - } - if (ProcSignalBarrierPending) + if (ConsumeInterrupt(INTERRUPT_BARRIER)) ProcessProcSignalBarrier(); - if (ParallelMessagePending) + if (ConsumeInterrupt(INTERRUPT_PARALLEL_MESSAGE)) HandleParallelMessages(); - if (LogMemoryContextPending) + if (ConsumeInterrupt(INTERRUPT_WALSND_INIT_STOPPING)) + HandleWalSndInitStopping(); + + if (ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT)) ProcessLogMemoryContextInterrupt(); - if (ParallelApplyMessagePending) + if (ConsumeInterrupt(INTERRUPT_PARALLEL_APPLY_MESSAGE)) HandleParallelApplyMessages(); } @@ -4207,7 +4153,7 @@ PostgresMain(const char *dbname, const char *username) * midst of output during who-knows-what operation... */ pqsignal(SIGPIPE, SIG_IGN); - pqsignal(SIGUSR1, procsignal_sigusr1_handler); + pqsignal(SIGUSR1, SIG_IGN); pqsignal(SIGUSR2, SIG_IGN); pqsignal(SIGFPE, FloatExceptionHandler); @@ -4364,7 +4310,7 @@ PostgresMain(const char *dbname, const char *username) * forgetting a timeout cancel. */ disable_all_timeouts(false); /* do first to avoid race condition */ - QueryCancelPending = false; + ClearInterrupt(INTERRUPT_QUERY_CANCEL); idle_in_transaction_timeout_enabled = false; idle_session_timeout_enabled = false; @@ -4549,7 +4495,7 @@ PostgresMain(const char *dbname, const char *username) * were received during the just-finished transaction, they'll * be seen by the client before ReadyForQuery is. */ - if (notifyInterruptPending) + if (InterruptIsPending(INTERRUPT_NOTIFY)) ProcessNotifyInterrupt(false); /* diff --git a/src/backend/utils/activity/pgstat_database.c b/src/backend/utils/activity/pgstat_database.c index 29bc0909748..f494c5afb76 100644 --- a/src/backend/utils/activity/pgstat_database.c +++ b/src/backend/utils/activity/pgstat_database.c @@ -90,29 +90,29 @@ pgstat_report_recovery_conflict(int reason) switch (reason) { - case PROCSIG_RECOVERY_CONFLICT_DATABASE: + case INTERRUPT_RECOVERY_CONFLICT_DATABASE: /* * Since we drop the information about the database as soon as it * replicates, there is no point in counting these conflicts. */ break; - case PROCSIG_RECOVERY_CONFLICT_TABLESPACE: + case INTERRUPT_RECOVERY_CONFLICT_TABLESPACE: dbentry->conflict_tablespace++; break; - case PROCSIG_RECOVERY_CONFLICT_LOCK: + case INTERRUPT_RECOVERY_CONFLICT_LOCK: dbentry->conflict_lock++; break; - case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT: + case INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT: dbentry->conflict_snapshot++; break; - case PROCSIG_RECOVERY_CONFLICT_BUFFERPIN: + case INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN: dbentry->conflict_bufferpin++; break; - case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT: + case INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT: dbentry->conflict_logicalslot++; break; - case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK: + case INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK: dbentry->conflict_startup_deadlock++; break; } diff --git a/src/backend/utils/adt/mcxtfuncs.c b/src/backend/utils/adt/mcxtfuncs.c index 10859414848..6432c32e7fc 100644 --- a/src/backend/utils/adt/mcxtfuncs.c +++ b/src/backend/utils/adt/mcxtfuncs.c @@ -166,7 +166,6 @@ pg_log_backend_memory_contexts(PG_FUNCTION_ARGS) { int pid = PG_GETARG_INT32(0); PGPROC *proc; - ProcNumber procNumber = INVALID_PROC_NUMBER; /* * See if the process with given pid is a backend or an auxiliary process. @@ -195,14 +194,7 @@ pg_log_backend_memory_contexts(PG_FUNCTION_ARGS) PG_RETURN_BOOL(false); } - procNumber = GetNumberFromPGProc(proc); - if (SendProcSignal(pid, PROCSIG_LOG_MEMORY_CONTEXT, procNumber) < 0) - { - /* Again, just a warning to allow loops */ - ereport(WARNING, - (errmsg("could not send signal to process %d: %m", pid))); - PG_RETURN_BOOL(false); - } + SendInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT, GetNumberFromPGProc(proc)); PG_RETURN_BOOL(true); } diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c index 927bccfbea8..35bbeee09eb 100644 --- a/src/backend/utils/init/globals.c +++ b/src/backend/utils/init/globals.c @@ -28,20 +28,9 @@ ProtocolVersion FrontendProtocol; -volatile sig_atomic_t InterruptPending = false; -volatile sig_atomic_t QueryCancelPending = false; -volatile sig_atomic_t ProcDiePending = false; -volatile sig_atomic_t CheckClientConnectionPending = false; -volatile sig_atomic_t ClientConnectionLost = false; -volatile sig_atomic_t IdleInTransactionSessionTimeoutPending = false; -volatile sig_atomic_t TransactionTimeoutPending = false; -volatile sig_atomic_t IdleSessionTimeoutPending = false; -volatile sig_atomic_t ProcSignalBarrierPending = false; -volatile sig_atomic_t LogMemoryContextPending = false; -volatile sig_atomic_t IdleStatsUpdateTimeoutPending = false; -volatile uint32 InterruptHoldoffCount = 0; -volatile uint32 QueryCancelHoldoffCount = 0; -volatile uint32 CritSectionCount = 0; +uint32 InterruptHoldoffCount = 0; +uint32 QueryCancelHoldoffCount = 0; +uint32 CritSectionCount = 0; int MyProcPid; pg_time_t MyStartTime; diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c index 25867c8bd5b..2fdf33aa162 100644 --- a/src/backend/utils/init/postinit.c +++ b/src/backend/utils/init/postinit.c @@ -1365,41 +1365,31 @@ LockTimeoutHandler(void) static void TransactionTimeoutHandler(void) { - TransactionTimeoutPending = true; - InterruptPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_TRANSACTION_TIMEOUT); } static void IdleInTransactionSessionTimeoutHandler(void) { - IdleInTransactionSessionTimeoutPending = true; - InterruptPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_IDLE_TRANSACTION_TIMEOUT); } static void IdleSessionTimeoutHandler(void) { - IdleSessionTimeoutPending = true; - InterruptPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_IDLE_SESSION_TIMEOUT); } static void IdleStatsUpdateTimeoutHandler(void) { - IdleStatsUpdateTimeoutPending = true; - InterruptPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_IDLE_STATS_UPDATE_TIMEOUT); } static void ClientCheckTimeoutHandler(void) { - CheckClientConnectionPending = true; - InterruptPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_CHECK_CONNECTION_TIMEOUT); } /* diff --git a/src/backend/utils/mmgr/mcxt.c b/src/backend/utils/mmgr/mcxt.c index bde54326c66..995d203df88 100644 --- a/src/backend/utils/mmgr/mcxt.c +++ b/src/backend/utils/mmgr/mcxt.c @@ -1260,36 +1260,18 @@ MemoryContextAllocExtended(MemoryContext context, Size size, int flags) return ret; } -/* - * HandleLogMemoryContextInterrupt - * Handle receipt of an interrupt indicating logging of memory - * contexts. - * - * All the actual work is deferred to ProcessLogMemoryContextInterrupt(), - * because we cannot safely emit a log message inside the signal handler. - */ -void -HandleLogMemoryContextInterrupt(void) -{ - InterruptPending = true; - LogMemoryContextPending = true; - /* latch will be set by procsignal_sigusr1_handler */ -} - /* * ProcessLogMemoryContextInterrupt * Perform logging of memory contexts of this backend process. * - * Any backend that participates in ProcSignal signaling must arrange - * to call this function if we see LogMemoryContextPending set. - * It is called from CHECK_FOR_INTERRUPTS(), which is enough because - * the target process for logging of memory contexts is a backend. + * Any backend connected to shared memory must arrange to call this function + * when ConsumeInterrupt(INTERRUPT_LOG_MEMORY_CONTEXT) returns true. It is + * called from CHECK_FOR_INTERRUPTS(), which is enough because the target + * process for logging of memory contexts is a backend. */ void ProcessLogMemoryContextInterrupt(void) { - LogMemoryContextPending = false; - /* * Use LOG_SERVER_ONLY to prevent this message from being sent to the * connected client. diff --git a/src/include/access/parallel.h b/src/include/access/parallel.h index 69ffe5498f9..21000a525d6 100644 --- a/src/include/access/parallel.h +++ b/src/include/access/parallel.h @@ -70,7 +70,6 @@ extern void WaitForParallelWorkersToFinish(ParallelContext *pcxt); extern void DestroyParallelContext(ParallelContext *pcxt); extern bool ParallelContextActive(void); -extern void HandleParallelMessageInterrupt(void); extern void HandleParallelMessages(void); extern void AtEOXact_Parallel(bool isCommit); extern void AtEOSubXact_Parallel(bool isCommit, SubTransactionId mySubId); diff --git a/src/include/commands/async.h b/src/include/commands/async.h index 78daa25fa08..b30d0a5b59b 100644 --- a/src/include/commands/async.h +++ b/src/include/commands/async.h @@ -17,7 +17,6 @@ extern PGDLLIMPORT bool Trace_notify; extern PGDLLIMPORT int max_notify_queue_pages; -extern PGDLLIMPORT volatile sig_atomic_t notifyInterruptPending; extern Size AsyncShmemSize(void); extern void AsyncShmemInit(void); @@ -40,9 +39,6 @@ extern void AtSubCommit_Notify(void); extern void AtSubAbort_Notify(void); extern void AtPrepare_Notify(void); -/* signal handler for inbound notifies (PROCSIG_NOTIFY_INTERRUPT) */ -extern void HandleNotifyInterrupt(void); - /* process interrupts */ extern void ProcessNotifyInterrupt(bool flush); diff --git a/src/include/libpq/pqmq.h b/src/include/libpq/pqmq.h index 227df8976f4..6729e948d9a 100644 --- a/src/include/libpq/pqmq.h +++ b/src/include/libpq/pqmq.h @@ -17,7 +17,7 @@ #include "storage/shm_mq.h" extern void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh); -extern void pq_set_parallel_leader(pid_t pid, ProcNumber procNumber); +extern void pq_set_parallel_leader(ProcNumber pgprocno); extern void pq_parse_errornotice(StringInfo msg, ErrorData *edata); diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 90f9b21b258..b93d024c4a9 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -23,11 +23,15 @@ #ifndef MISCADMIN_H #define MISCADMIN_H -#include - #include "datatype/timestamp.h" /* for TimestampTz */ #include "pgtime.h" /* for pg_time_t */ +#ifndef FRONTEND +#include "postmaster/interrupt.h" +#endif + +#include "storage/procsignal.h" + #define InvalidPid (-1) @@ -85,66 +89,8 @@ * *****************************************************************************/ -/* in globals.c */ -/* these are marked volatile because they are set by signal handlers: */ -extern PGDLLIMPORT volatile sig_atomic_t InterruptPending; -extern PGDLLIMPORT volatile sig_atomic_t QueryCancelPending; -extern PGDLLIMPORT volatile sig_atomic_t ProcDiePending; -extern PGDLLIMPORT volatile sig_atomic_t IdleInTransactionSessionTimeoutPending; -extern PGDLLIMPORT volatile sig_atomic_t TransactionTimeoutPending; -extern PGDLLIMPORT volatile sig_atomic_t IdleSessionTimeoutPending; -extern PGDLLIMPORT volatile sig_atomic_t ProcSignalBarrierPending; -extern PGDLLIMPORT volatile sig_atomic_t LogMemoryContextPending; -extern PGDLLIMPORT volatile sig_atomic_t IdleStatsUpdateTimeoutPending; - -extern PGDLLIMPORT volatile sig_atomic_t CheckClientConnectionPending; -extern PGDLLIMPORT volatile sig_atomic_t ClientConnectionLost; - -/* these are marked volatile because they are examined by signal handlers: */ -extern PGDLLIMPORT volatile uint32 InterruptHoldoffCount; -extern PGDLLIMPORT volatile uint32 QueryCancelHoldoffCount; -extern PGDLLIMPORT volatile uint32 CritSectionCount; - -/* in tcop/postgres.c */ -extern void ProcessInterrupts(void); - -/* Test whether an interrupt is pending */ -#ifndef WIN32 -#define INTERRUPTS_PENDING_CONDITION() \ - (unlikely(InterruptPending)) -#else -#define INTERRUPTS_PENDING_CONDITION() \ - (unlikely(UNBLOCKED_SIGNAL_QUEUE()) ? pgwin32_dispatch_queued_signals() : 0, \ - unlikely(InterruptPending)) -#endif - -/* Service interrupt, if one is pending and it's safe to service it now */ -#define CHECK_FOR_INTERRUPTS() \ -do { \ - if (INTERRUPTS_PENDING_CONDITION()) \ - ProcessInterrupts(); \ -} while(0) - -/* Is ProcessInterrupts() guaranteed to clear InterruptPending? */ -#define INTERRUPTS_CAN_BE_PROCESSED() \ - (InterruptHoldoffCount == 0 && CritSectionCount == 0 && \ - QueryCancelHoldoffCount == 0) - -#define HOLD_INTERRUPTS() (InterruptHoldoffCount++) +extern PGDLLIMPORT uint32 CritSectionCount; -#define RESUME_INTERRUPTS() \ -do { \ - Assert(InterruptHoldoffCount > 0); \ - InterruptHoldoffCount--; \ -} while(0) - -#define HOLD_CANCEL_INTERRUPTS() (QueryCancelHoldoffCount++) - -#define RESUME_CANCEL_INTERRUPTS() \ -do { \ - Assert(QueryCancelHoldoffCount > 0); \ - QueryCancelHoldoffCount--; \ -} while(0) #define START_CRIT_SECTION() (CritSectionCount++) diff --git a/src/include/postmaster/interrupt.h b/src/include/postmaster/interrupt.h index 341a09d83fe..827ef592ac1 100644 --- a/src/include/postmaster/interrupt.h +++ b/src/include/postmaster/interrupt.h @@ -3,9 +3,34 @@ * interrupt.h * Interrupt handling routines. * - * Responses to interrupts are fairly varied and many types of backends - * have their own implementations, but we provide a few generic things - * here to facilitate code reuse. + * "Interrupts" are a set of flags that represent conditions that should be + * handled at a later time. They are roughly analogous to Unix signals, + * except that they are handled cooperatively by checking for them at many + * points in the code. + * + * Interrupt flags can be "raised" synchronously by code that wants to defer + * an action (for example: INTERRUPT_CONNECTION_LOST), or asynchronously by + * timer signal handlers (for example: INTERRUPT_IDLE_SESSION_TIMEOUT), other + * signal handlers (for example: INTERRUPT_QUERY_CANCEL) or "sent" by other + * backends setting them directly. + * + * In the case of asynchronous interrupts, the target backend's latch is also + * set, to make sure that the backend wakes from latch sleeps. Well behaved + * backend code performs CHECK_FOR_INTERRUPTS() periodically in long + * computations, and should never sleep using mechanisms other than the latch + * wait mechanism (except for bounded short periods, eg LWLock waits), so they + * should react in good time. + * + * The "standard" set of interrupts is handled by CHECK_FOR_INTERRUPTS(), and + * consists of tasks that are safe to perform at most times. They can be + * suppressed by HOLD_INTERRUPTS()/RESUME_INTERRUPTS(). + * + * Other special interrupts are checked for explicitly. + + * Responses to signals that are translated to interrupts are fairly varied + * and many types of backends have their own implementations, but we provide a + * few generic signal handlers and interrupt checks here to facilitate code + * reuse. * * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California @@ -19,12 +44,154 @@ #ifndef INTERRUPT_H #define INTERRUPT_H +#include "port/atomics.h" +#include "storage/procnumber.h" + #include extern PGDLLIMPORT volatile sig_atomic_t ConfigReloadPending; extern PGDLLIMPORT volatile sig_atomic_t ShutdownRequestPending; +extern PGDLLIMPORT uint32 InterruptHoldoffCount; +extern PGDLLIMPORT uint32 QueryCancelHoldoffCount; + +extern PGDLLIMPORT pg_atomic_uint32 *MyPendingInterrupts; + +typedef enum +{ + /* Sent by other backends. */ + INTERRUPT_SINVAL_CATCHUP, + INTERRUPT_NOTIFY, /* listen/notify interrupt */ + INTERRUPT_PARALLEL_APPLY_MESSAGE, /* message from cooperating parallel + * backend */ + INTERRUPT_PARALLEL_MESSAGE, /* message from cooperating parallel backend */ + INTERRUPT_WALSND_INIT_STOPPING, /* ask walsenders to prepare for shutdown */ + INTERRUPT_BARRIER, /* global barrier interrupt */ + INTERRUPT_LOG_MEMORY_CONTEXT, /* ask backend to log the memory contexts */ + + /* Raised by timers. */ + INTERRUPT_TRANSACTION_TIMEOUT, + INTERRUPT_IDLE_SESSION_TIMEOUT, + INTERRUPT_IDLE_TRANSACTION_TIMEOUT, + INTERRUPT_IDLE_STATS_UPDATE_TIMEOUT, + INTERRUPT_CHECK_CONNECTION_TIMEOUT, + + /* Raised by signal handlers (usually sent from the postmaster). */ + INTERRUPT_QUERY_CANCEL, + INTERRUPT_DIE, + + /* Raised synchronously. */ + INTERRUPT_CONNECTION_LOST, + + /* Sent by startup process. */ + INTERRUPT_RECOVERY_CONFLICT_FIRST, + INTERRUPT_RECOVERY_CONFLICT_DATABASE = INTERRUPT_RECOVERY_CONFLICT_FIRST, + INTERRUPT_RECOVERY_CONFLICT_TABLESPACE, + INTERRUPT_RECOVERY_CONFLICT_LOCK, + INTERRUPT_RECOVERY_CONFLICT_SNAPSHOT, + INTERRUPT_RECOVERY_CONFLICT_BUFFERPIN, + INTERRUPT_RECOVERY_CONFLICT_STARTUP_DEADLOCK, + INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT, + INTERRUPT_RECOVERY_CONFLICT_LAST = INTERRUPT_RECOVERY_CONFLICT_LOGICALSLOT +} InterruptType; + +/* + * CHECK_FOR_INTERRUPTS() ignores special interrupts that are only handled when + * the backend is idle. + */ +#define INTERRUPT_IDLE_MASK \ + ((1 << INTERRUPT_SINVAL_CATCHUP) | \ + (1 << INTERRUPT_NOTIFY)) +#define INTERRUPT_REGULAR_MASK ~INTERRUPT_IDLE_MASK + +/* Test whether a regular interrupt is pending */ +#ifndef WIN32 +#define INTERRUPTS_PENDING_CONDITION() \ + unlikely((pg_atomic_read_u32(MyPendingInterrupts) & INTERRUPT_REGULAR_MASK) != 0) +#else +#define INTERRUPTS_PENDING_CONDITION() \ + (unlikely(UNBLOCKED_SIGNAL_QUEUE()) ? pgwin32_dispatch_queued_signals() : 0, \ + unlikely((pg_atomic_read_u32(MyPendingInterrupts) & INTERRUPT_REGULAR_MASK) != 0)) +#endif + +/* + * Service regular interrupt, if one is pending and it's safe to service it + * now. + */ +#define CHECK_FOR_INTERRUPTS() \ +do { \ + if (INTERRUPTS_PENDING_CONDITION()) \ + ProcessInterrupts(); \ +} while(0) + +/* Is ProcessInterrupts() guaranteed to clear regular interrupts? */ +#define INTERRUPTS_CAN_BE_PROCESSED() \ + (InterruptHoldoffCount == 0 && CritSectionCount == 0 && \ + QueryCancelHoldoffCount == 0) + +#define HOLD_INTERRUPTS() (InterruptHoldoffCount++) + +#define RESUME_INTERRUPTS() \ +do { \ + Assert(InterruptHoldoffCount > 0); \ + InterruptHoldoffCount--; \ +} while(0) + +#define HOLD_CANCEL_INTERRUPTS() (QueryCancelHoldoffCount++) + +#define RESUME_CANCEL_INTERRUPTS() \ +do { \ + Assert(QueryCancelHoldoffCount > 0); \ + QueryCancelHoldoffCount--; \ +} while(0) + +/* + * Test an interrupt flag. + */ +static inline bool +InterruptIsPending(InterruptType reason) +{ + return (pg_atomic_read_u32(MyPendingInterrupts) & (1 << reason)) != 0; +} + +/* + * Clear an interrupt flag. + */ +static inline void +ClearInterrupt(InterruptType reason) +{ + pg_atomic_fetch_and_u32(MyPendingInterrupts, ~(1 << reason)); +} + +/* + * Test and clear an interrupt flag. + */ +static inline bool +ConsumeInterrupt(InterruptType reason) +{ + if (likely(!InterruptIsPending(reason))) + return false; + + ClearInterrupt(reason); + + return true; +} + +extern void RaiseInterrupt(InterruptType reason); +extern void SendInterrupt(InterruptType reason, ProcNumber pgprocno); +extern void SwitchToLocalInterrupts(void); +extern void SwitchToSharedInterrupts(void); + +/* in tcop/postgres.c */ +extern void ProcessInterrupts(void); + +/* + * A handler for INTERRUPT_BARRIER, and the reload/exit/shutdown flags set by + * the signal handlers below. + */ extern void HandleMainLoopInterrupts(void); + +/* Common signal handlers. */ extern void SignalHandlerForConfigReload(SIGNAL_ARGS); extern void SignalHandlerForCrashExit(SIGNAL_ARGS); extern void SignalHandlerForShutdownRequest(SIGNAL_ARGS); diff --git a/src/include/regex/regcustom.h b/src/include/regex/regcustom.h index af0fe97c796..5d9986c70e8 100644 --- a/src/include/regex/regcustom.h +++ b/src/include/regex/regcustom.h @@ -45,6 +45,7 @@ #include "mb/pg_wchar.h" #include "miscadmin.h" /* needed by rstacktoodeep */ +#include "postmaster/interrupt.h" /* overrides for regguts.h definitions, if any */ diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index c9675ee87cc..c92d7a8ec4b 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -155,6 +155,7 @@ typedef struct ReplicationSlot /* Who is streaming out changes for this slot? 0 in unused slots. */ pid_t active_pid; + ProcNumber active_pgprocno; /* any outstanding modifications? */ bool just_dirtied; diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index cf32ac2488a..b52ace32a43 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -42,6 +42,7 @@ typedef enum WalSndState typedef struct WalSnd { pid_t pid; /* this walsender's PID, or 0 if not active */ + ProcNumber proc_number; /* this walsender's proc number */ WalSndState state; /* this walsender's state */ XLogRecPtr sentPtr; /* WAL has been sent up to this point */ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 515aefd5191..dc8e577562b 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -78,10 +78,11 @@ typedef struct LogicalRepWorker FileSet *stream_fileset; /* - * PID of leader apply worker if this slot is used for a parallel apply - * worker, InvalidPid otherwise. + * Leader apply worker if this slot is used for a parallel apply worker, + * InvalidPid and INVALID_PROC_NUMBER otherwise. */ pid_t leader_pid; + ProcNumber leader_pgprocno; /* Indicates whether apply can be performed in parallel. */ bool parallel_apply; diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index ae483928ba7..188fba82e94 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -305,6 +305,8 @@ struct PGPROC PGPROC *lockGroupLeader; /* lock group leader, if I'm a member */ dlist_head lockGroupMembers; /* list of members, if I'm a leader */ dlist_node lockGroupLink; /* my member link, if I'm a member */ + + pg_atomic_uint32 pending_interrupts; }; /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */ diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index 8ca60504622..6123adad2fc 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -77,14 +77,14 @@ extern VirtualTransactionId *GetCurrentVirtualXIDs(TransactionId limitXmin, bool excludeXmin0, bool allDbs, int excludeVacuum, int *nvxids); extern VirtualTransactionId *GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid); -extern pid_t CancelVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode); -extern pid_t SignalVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode, +extern pid_t CancelVirtualTransaction(VirtualTransactionId vxid, InterruptType sigmode); +extern pid_t SignalVirtualTransaction(VirtualTransactionId vxid, InterruptType sigmode, bool conflictPending); extern bool MinimumActiveBackends(int min); extern int CountDBBackends(Oid databaseid); extern int CountDBConnections(Oid databaseid); -extern void CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending); +extern void CancelDBBackends(Oid databaseid, InterruptType sigmode, bool conflictPending); extern int CountUserBackends(Oid roleid); extern bool CountOtherDBBackends(Oid databaseId, int *nbackends, int *nprepared); diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index 7d290ea7d05..169e6c5de91 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -17,40 +17,6 @@ #include "storage/procnumber.h" -/* - * Reasons for signaling a Postgres child process (a backend or an auxiliary - * process, like checkpointer). We can cope with concurrent signals for different - * reasons. However, if the same reason is signaled multiple times in quick - * succession, the process is likely to observe only one notification of it. - * This is okay for the present uses. - * - * Also, because of race conditions, it's important that all the signals be - * defined so that no harm is done if a process mistakenly receives one. - */ -typedef enum -{ - PROCSIG_CATCHUP_INTERRUPT, /* sinval catchup interrupt */ - PROCSIG_NOTIFY_INTERRUPT, /* listen/notify interrupt */ - PROCSIG_PARALLEL_MESSAGE, /* message from cooperating parallel backend */ - PROCSIG_WALSND_INIT_STOPPING, /* ask walsenders to prepare for shutdown */ - PROCSIG_BARRIER, /* global barrier interrupt */ - PROCSIG_LOG_MEMORY_CONTEXT, /* ask backend to log the memory contexts */ - PROCSIG_PARALLEL_APPLY_MESSAGE, /* Message from parallel apply workers */ - - /* Recovery conflict reasons */ - PROCSIG_RECOVERY_CONFLICT_FIRST, - PROCSIG_RECOVERY_CONFLICT_DATABASE = PROCSIG_RECOVERY_CONFLICT_FIRST, - PROCSIG_RECOVERY_CONFLICT_TABLESPACE, - PROCSIG_RECOVERY_CONFLICT_LOCK, - PROCSIG_RECOVERY_CONFLICT_SNAPSHOT, - PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT, - PROCSIG_RECOVERY_CONFLICT_BUFFERPIN, - PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK, - PROCSIG_RECOVERY_CONFLICT_LAST = PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK, - - NUM_PROCSIGNALS /* Must be last! */ -} ProcSignalReason; - typedef enum { PROCSIGNAL_BARRIER_SMGRRELEASE, /* ask smgr to close files */ @@ -63,13 +29,9 @@ extern Size ProcSignalShmemSize(void); extern void ProcSignalShmemInit(void); extern void ProcSignalInit(void); -extern int SendProcSignal(pid_t pid, ProcSignalReason reason, - ProcNumber procNumber); extern uint64 EmitProcSignalBarrier(ProcSignalBarrierType type); extern void WaitForProcSignalBarrier(uint64 generation); extern void ProcessProcSignalBarrier(void); -extern void procsignal_sigusr1_handler(SIGNAL_ARGS); - #endif /* PROCSIGNAL_H */ diff --git a/src/include/storage/sinval.h b/src/include/storage/sinval.h index 8f5744b21bc..16370b14158 100644 --- a/src/include/storage/sinval.h +++ b/src/include/storage/sinval.h @@ -125,16 +125,11 @@ typedef union /* Counter of messages processed; don't worry about overflow. */ extern PGDLLIMPORT uint64 SharedInvalidMessageCounter; -extern PGDLLIMPORT volatile sig_atomic_t catchupInterruptPending; - extern void SendSharedInvalidMessages(const SharedInvalidationMessage *msgs, int n); extern void ReceiveSharedInvalidMessages(void (*invalFunction) (SharedInvalidationMessage *msg), void (*resetFunction) (void)); -/* signal handler for catchup events (PROCSIG_CATCHUP_INTERRUPT) */ -extern void HandleCatchupInterrupt(void); - /* * enable/disable processing of catchup events directly from signal handler. * The enable routine first performs processing of any catchup events that diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h index cce0bc521e7..fa3ddd416dc 100644 --- a/src/include/storage/standby.h +++ b/src/include/storage/standby.h @@ -15,6 +15,7 @@ #define STANDBY_H #include "datatype/timestamp.h" +#include "postmaster/interrupt.h" #include "storage/lock.h" #include "storage/procsignal.h" #include "storage/relfilelocator.h" @@ -43,8 +44,8 @@ extern void CheckRecoveryConflictDeadlock(void); extern void StandbyDeadLockHandler(void); extern void StandbyTimeoutHandler(void); extern void StandbyLockTimeoutHandler(void); -extern void LogRecoveryConflict(ProcSignalReason reason, TimestampTz wait_start, - TimestampTz now, VirtualTransactionId *wait_list, +extern void LogRecoveryConflict(InterruptType reason, TimestampTz wait_start, + TimestampTz cur_ts, VirtualTransactionId *wait_list, bool still_waiting); /* diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h index 643ce9cffab..f4bd1295ccc 100644 --- a/src/include/tcop/tcopprot.h +++ b/src/include/tcop/tcopprot.h @@ -16,6 +16,7 @@ #include "nodes/params.h" #include "nodes/plannodes.h" +#include "postmaster/interrupt.h" #include "storage/procsignal.h" #include "utils/guc.h" #include "utils/queryenvironment.h" @@ -69,7 +70,6 @@ extern void die(SIGNAL_ARGS); extern void quickdie(SIGNAL_ARGS) pg_attribute_noreturn(); extern void StatementCancelHandler(SIGNAL_ARGS); extern void FloatExceptionHandler(SIGNAL_ARGS) pg_attribute_noreturn(); -extern void HandleRecoveryConflictInterrupt(ProcSignalReason reason); extern void ProcessClientReadInterrupt(bool blocked); extern void ProcessClientWriteInterrupt(bool blocked); -- 2.45.2