From 519067cd12ad89abef3b1b2c8ed2efa31f7ffa1d Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Fri, 30 Aug 2024 22:01:45 +0300 Subject: [PATCH v2 08/12] Replace Latches with Interrupts The Latch was a flag, with an inter-process signalling mechanism that allowed a process to wait for the Latch to be set. My original vision for Latches was that you could have different latches for different things that you need to wait for, but but in practice, almost all code used one "process latch" that was shared for all wakeups. The only exception was the "recoveryWakeupLatch". I think the reason that we ended up just sharing the same latch for all wakeups is that it was cumbersome in practice to deal with multiple latches. For starters, there was no machinery for waiting for multiple latches at the same time. Secondly, changing the "ownership" of a latch needed extra locking or other coordination. Thirdly, each inter-process latch needed to be initialized at postmaster startup time. This patch embraces the reality of how Latches were used, and replaces the Latches with per-process interrupt mask. You can no longer allocate Latches in arbitrary shared memory structs, an interrupt is always directed at a particular process, addressed by its ProcNumber. Each process has a bitmask of pending interrupts in PGPROC. This commit introduces two interrupt bits. INTERRUPT_GENERAL_WAKEUP replaces the general purpose per-process latch. All code that previously set a process's process latch now sets its INTERRUPT_GENERAL_WAKEUP interrupt bit instead. The other interrupt bit, INTERRUPT_RECOVERY_WAKEUP, replaces recoveryWakeupLatch. With this new interface, the code can easily wait for the regular interrupts (INTERRUPT_GENERAL_WAKEUP) at the same time as INTERRUPT_RECOVERY_WAKEUP. Previously, the code that waited on recoveryWakeupLatch had to resort to timeouts to react to other signals, because it was not possible to wait for two latches at the same time. Previous attempt at unifying those wakeups was in commit ac22929a26, but that was reverted in commit 00f690a239 because it caused a lot of spurious wakeups when startup process was waiting for recovery conflicts. The new machinery avoids that problem, by making it easy to wait for two interrupts at the same time. More interrupt bits are planned for followup patches, to replace the various ProcSignal bits, as well as ConfigReloadPending and ShutdownRequestPending. This patch leaves behind a compatibility "latch.h" header, which defines macros to map existing Latch functions to the corresponding Interrupt functions. This reduces the code churn, and shows how we could keep limited backwards-compatibility for extensions if necessary. This also moves the WaitEventSet functions to a different source file, waiteventset.c. This separates the platform-dependent code waiting and signalling code from the platform-independent parts. --- contrib/postgres_fdw/postgres_fdw.c | 2 +- src/backend/access/transam/xlog.c | 14 +- src/backend/access/transam/xlogrecovery.c | 85 +-- src/backend/commands/waitlsn.c | 31 +- src/backend/executor/nodeAppend.c | 5 +- src/backend/libpq/be-secure.c | 14 +- src/backend/libpq/pqcomm.c | 29 +- src/backend/postmaster/auxprocess.c | 1 + src/backend/postmaster/checkpointer.c | 24 +- src/backend/postmaster/interrupt.c | 6 +- src/backend/postmaster/pgarch.c | 27 +- src/backend/postmaster/postmaster.c | 31 +- src/backend/postmaster/startup.c | 11 +- src/backend/postmaster/syslogger.c | 4 +- src/backend/postmaster/walsummarizer.c | 2 +- src/backend/postmaster/walwriter.c | 11 +- src/backend/replication/logical/launcher.c | 2 +- src/backend/replication/syncrep.c | 2 +- src/backend/replication/walreceiver.c | 2 +- src/backend/replication/walreceiverfuncs.c | 2 +- src/backend/replication/walsender.c | 21 +- src/backend/storage/buffer/freelist.c | 2 +- src/backend/storage/ipc/Makefile | 5 +- src/backend/storage/ipc/interrupt.c | 114 ++++ src/backend/storage/ipc/meson.build | 3 +- src/backend/storage/ipc/shm_mq.c | 89 +-- .../storage/ipc/{latch.c => waiteventset.c} | 601 +++++++----------- src/backend/storage/lmgr/condition_variable.c | 6 +- src/backend/storage/lmgr/proc.c | 105 +-- src/backend/storage/sync/sync.c | 4 +- src/backend/utils/init/globals.c | 9 - src/backend/utils/init/miscinit.c | 56 +- src/include/access/parallel.h | 2 + src/include/commands/waitlsn.h | 1 - src/include/libpq/libpq.h | 4 +- src/include/miscadmin.h | 4 - src/include/storage/interrupt.h | 159 +++++ src/include/storage/latch.h | 217 ++----- src/include/storage/proc.h | 29 +- src/include/storage/waiteventset.h | 119 ++++ src/test/modules/test_shm_mq/setup.c | 7 +- src/test/modules/test_shm_mq/test.c | 7 +- src/test/modules/test_shm_mq/worker.c | 3 +- src/tools/pgindent/typedefs.list | 2 +- 44 files changed, 980 insertions(+), 894 deletions(-) create mode 100644 src/backend/storage/ipc/interrupt.c rename src/backend/storage/ipc/{latch.c => waiteventset.c} (78%) create mode 100644 src/include/storage/interrupt.h create mode 100644 src/include/storage/waiteventset.h diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index adc62576d1..4652851b50 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -7357,7 +7357,7 @@ postgresForeignAsyncConfigureWait(AsyncRequest *areq) Assert(pendingAreq == areq); AddWaitEventToSet(set, WL_SOCKET_READABLE, PQsocket(fsstate->conn), - NULL, areq); + 0, areq); } /* diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index fdc0906314..fd19a8845f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -86,9 +86,9 @@ #include "replication/walsender.h" #include "storage/bufmgr.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/large_object.h" -#include "storage/latch.h" #include "storage/predicate.h" #include "storage/proc.h" #include "storage/procarray.h" @@ -2676,7 +2676,7 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN) walwriterProc = procglobal->walwriterProc; if (walwriterProc != INVALID_PROC_NUMBER) - SetLatch(&GetPGProcByNumber(walwriterProc)->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, ProcGlobal->walwriterProc); } } @@ -9324,11 +9324,11 @@ do_pg_backup_stop(BackupState *state, bool waitforarchive) reported_waiting = true; } - (void) WaitLatch(MyLatch, - WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, - 1000L, - WAIT_EVENT_BACKUP_WAIT_WAL_ARCHIVE); - ResetLatch(MyLatch); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + 1000L, + WAIT_EVENT_BACKUP_WAIT_WAL_ARCHIVE); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); if (++waits >= seconds_before_warning) { diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 35a5e31e3d..ab7570ecc3 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -54,8 +54,9 @@ #include "replication/walreceiver.h" #include "storage/fd.h" #include "storage/ipc.h" -#include "storage/latch.h" +#include "storage/interrupt.h" #include "storage/pmsignal.h" +#include "storage/proc.h" #include "storage/procarray.h" #include "storage/spin.h" #include "utils/datetime.h" @@ -316,23 +317,6 @@ typedef struct XLogRecoveryCtlData */ bool SharedPromoteIsTriggered; - /* - * recoveryWakeupLatch is used to wake up the startup process to continue - * WAL replay, if it is waiting for WAL to arrive or promotion to be - * requested. - * - * Note that the startup process also uses another latch, its procLatch, - * to wait for recovery conflict. If we get rid of recoveryWakeupLatch for - * signaling the startup process in favor of using its procLatch, which - * comports better with possible generic signal handlers using that latch. - * But we should not do that because the startup process doesn't assume - * that it's waken up by walreceiver process or SIGHUP signal handler - * while it's waiting for recovery conflict. The separate latches, - * recoveryWakeupLatch and procLatch, should be used for inter-process - * communication for WAL replay and recovery conflict, respectively. - */ - Latch recoveryWakeupLatch; - /* * Last record successfully replayed. */ @@ -467,7 +451,6 @@ XLogRecoveryShmemInit(void) memset(XLogRecoveryCtl, 0, sizeof(XLogRecoveryCtlData)); SpinLockInit(&XLogRecoveryCtl->info_lck); - InitSharedLatch(&XLogRecoveryCtl->recoveryWakeupLatch); ConditionVariableInit(&XLogRecoveryCtl->recoveryNotPausedCV); } @@ -541,13 +524,6 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr, readRecoverySignalFile(); validateRecoveryParameters(); - /* - * Take ownership of the wakeup latch if we're going to sleep during - * recovery, if required. - */ - if (ArchiveRecoveryRequested) - OwnLatch(&XLogRecoveryCtl->recoveryWakeupLatch); - /* * Set the WAL reading processor now, as it will be needed when reading * the checkpoint record required (backup_label or not). @@ -1635,13 +1611,6 @@ ShutdownWalRecovery(void) snprintf(recoveryPath, MAXPGPATH, XLOGDIR "/RECOVERYHISTORY"); unlink(recoveryPath); /* ignore any error */ } - - /* - * We don't need the latch anymore. It's not strictly necessary to disown - * it, but let's do it for the sake of tidiness. - */ - if (ArchiveRecoveryRequested) - DisownLatch(&XLogRecoveryCtl->recoveryWakeupLatch); } /* @@ -3042,11 +3011,19 @@ recoveryApplyDelay(XLogReaderState *record) while (true) { - ResetLatch(&XLogRecoveryCtl->recoveryWakeupLatch); + /* + * INTERRUPT_GENERAL_WAKUP is used for all the usual interrupts, like + * config reloads. The wakeups when more WAL arrive use a different + * interrupt number (INTERRUPT_RECOVERY_CONTINUE) so that more WAL + * arriving don't wake up the startup process excessively, when we're + * waiting in other places, like for recovery conflicts. + */ + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* This might change recovery_min_apply_delay. */ HandleStartupProcInterrupts(); + ClearInterrupt(INTERRUPT_RECOVERY_CONTINUE); if (CheckForStandbyTrigger()) break; @@ -3067,10 +3044,11 @@ recoveryApplyDelay(XLogReaderState *record) elog(DEBUG2, "recovery apply delay %ld milliseconds", msecs); - (void) WaitLatch(&XLogRecoveryCtl->recoveryWakeupLatch, - WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, - msecs, - WAIT_EVENT_RECOVERY_APPLY_DELAY); + (void) WaitInterrupt(1 << INTERRUPT_RECOVERY_CONTINUE | + 1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + msecs, + WAIT_EVENT_RECOVERY_APPLY_DELAY); } return true; } @@ -3713,15 +3691,17 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, /* Do background tasks that might benefit us later. */ KnownAssignedTransactionIdsIdleMaintenance(); - (void) WaitLatch(&XLogRecoveryCtl->recoveryWakeupLatch, - WL_LATCH_SET | WL_TIMEOUT | - WL_EXIT_ON_PM_DEATH, - wait_time, - WAIT_EVENT_RECOVERY_RETRIEVE_RETRY_INTERVAL); - ResetLatch(&XLogRecoveryCtl->recoveryWakeupLatch); + (void) WaitInterrupt(1 << INTERRUPT_RECOVERY_CONTINUE | + 1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_TIMEOUT | + WL_EXIT_ON_PM_DEATH, + wait_time, + WAIT_EVENT_RECOVERY_RETRIEVE_RETRY_INTERVAL); + ClearInterrupt(INTERRUPT_RECOVERY_CONTINUE); now = GetCurrentTimestamp(); /* Handle interrupt signals of startup process */ + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); HandleStartupProcInterrupts(); } last_fail_time = now; @@ -3989,11 +3969,12 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, * Wait for more WAL to arrive, when we will be woken * immediately by the WAL receiver. */ - (void) WaitLatch(&XLogRecoveryCtl->recoveryWakeupLatch, - WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, - -1L, - WAIT_EVENT_RECOVERY_WAL_STREAM); - ResetLatch(&XLogRecoveryCtl->recoveryWakeupLatch); + (void) WaitInterrupt(1 << INTERRUPT_RECOVERY_CONTINUE | + 1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, + -1L, + WAIT_EVENT_RECOVERY_WAL_STREAM); + ClearInterrupt(INTERRUPT_RECOVERY_CONTINUE); break; } @@ -4013,6 +3994,7 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, * This possibly-long loop needs to handle interrupts of startup * process. */ + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); HandleStartupProcInterrupts(); } @@ -4489,7 +4471,10 @@ CheckPromoteSignal(void) void WakeupRecovery(void) { - SetLatch(&XLogRecoveryCtl->recoveryWakeupLatch); + int procno = ((volatile PROC_HDR *) ProcGlobal)->startupProc; + + if (procno != INVALID_PROC_NUMBER) + SendInterrupt(INTERRUPT_RECOVERY_CONTINUE, procno); } /* diff --git a/src/backend/commands/waitlsn.c b/src/backend/commands/waitlsn.c index 1937bafafc..4fc34854b4 100644 --- a/src/backend/commands/waitlsn.c +++ b/src/backend/commands/waitlsn.c @@ -23,7 +23,7 @@ #include "commands/waitlsn.h" #include "funcapi.h" #include "miscadmin.h" -#include "storage/latch.h" +#include "storage/interrupt.h" #include "storage/proc.h" #include "storage/shmem.h" #include "utils/fmgrprotos.h" @@ -147,9 +147,9 @@ deleteLSNWaiter(void) } /* - * Remove waiters whose LSN has been replayed from the heap and set their - * latches. If InvalidXLogRecPtr is given, remove all waiters from the heap - * and set latches for all waiters. + * Remove waiters whose LSN has been replayed from the heap and send them + * interrupts. If InvalidXLogRecPtr is given, remove all waiters from the heap + * and send interrupts for all waiters. */ void WaitLSNWakeup(XLogRecPtr currentLSN) @@ -187,14 +187,13 @@ WaitLSNWakeup(XLogRecPtr currentLSN) LWLockRelease(WaitLSNLock); /* - * Set latches for processes, whose waited LSNs are already replayed. As - * the time consuming operations, we do it this outside of WaitLSNLock. - * This is actually fine because procLatch isn't ever freed, so we just - * can potentially set the wrong process' (or no process') latch. + * Send the interrupts. It's possible that the backends have exited since + * we released the lock, so we may interrupt wrong backend, but that's + * harmless. */ for (i = 0; i < numWakeUpProcs; i++) { - SetLatch(&GetPGProcByNumber(wakeUpProcs[i])->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, wakeUpProcs[i]); } pfree(wakeUpProcs); } @@ -216,15 +215,15 @@ WaitLSNCleanup(void) } /* - * Wait using MyLatch till the given LSN is replayed, the postmaster dies or - * timeout happens. + * Wait till the given LSN is replayed and we get interrupted, the postmaster + * dies or timeout happens. */ static void WaitForLSNReplay(XLogRecPtr targetLSN, int64 timeout) { XLogRecPtr currentLSN; TimestampTz endtime = 0; - int wake_events = WL_LATCH_SET | WL_EXIT_ON_PM_DEATH; + int wake_events = WL_INTERRUPT | WL_EXIT_ON_PM_DEATH; /* Shouldn't be called when shmem isn't initialized */ Assert(waitLSNState); @@ -312,11 +311,11 @@ WaitForLSNReplay(XLogRecPtr targetLSN, int64 timeout) CHECK_FOR_INTERRUPTS(); - rc = WaitLatch(MyLatch, wake_events, delay_ms, - WAIT_EVENT_WAIT_FOR_WAL_REPLAY); + rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, wake_events, delay_ms, + WAIT_EVENT_WAIT_FOR_WAL_REPLAY); - if (rc & WL_LATCH_SET) - ResetLatch(MyLatch); + if (rc & WL_INTERRUPT) + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* diff --git a/src/backend/executor/nodeAppend.c b/src/backend/executor/nodeAppend.c index ca0f54d676..314e3d8fbb 100644 --- a/src/backend/executor/nodeAppend.c +++ b/src/backend/executor/nodeAppend.c @@ -61,9 +61,8 @@ #include "executor/execPartition.h" #include "executor/executor.h" #include "executor/nodeAppend.h" -#include "miscadmin.h" #include "pgstat.h" -#include "storage/latch.h" +#include "storage/waiteventset.h" /* Shared state for parallel-aware Append. */ struct ParallelAppendState @@ -1028,7 +1027,7 @@ ExecAppendAsyncEventWait(AppendState *node) Assert(node->as_eventset == NULL); node->as_eventset = CreateWaitEventSet(CurrentResourceOwner, nevents); AddWaitEventToSet(node->as_eventset, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET, - NULL, NULL); + 0, NULL); /* Give each waiting subplan a chance to add an event. */ i = -1; diff --git a/src/backend/libpq/be-secure.c b/src/backend/libpq/be-secure.c index ef20ea755b..4fdbf2d873 100644 --- a/src/backend/libpq/be-secure.c +++ b/src/backend/libpq/be-secure.c @@ -28,7 +28,7 @@ #include #include "libpq/libpq.h" -#include "miscadmin.h" +#include "storage/interrupt.h" #include "tcop/tcopprot.h" #include "utils/injection_point.h" #include "utils/wait_event.h" @@ -212,7 +212,7 @@ retry: Assert(waitfor); - ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, waitfor, NULL); + ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, waitfor, 0); WaitEventSetWait(FeBeWaitSet, -1 /* no timeout */ , &event, 1, WAIT_EVENT_CLIENT_READ); @@ -240,9 +240,9 @@ retry: errmsg("terminating connection due to unexpected postmaster exit"))); /* Handle interrupt. */ - if (event.events & WL_LATCH_SET) + if (event.events & WL_INTERRUPT) { - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); ProcessClientReadInterrupt(true); /* @@ -337,7 +337,7 @@ retry: Assert(waitfor); - ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, waitfor, NULL); + ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, waitfor, 0); WaitEventSetWait(FeBeWaitSet, -1 /* no timeout */ , &event, 1, WAIT_EVENT_CLIENT_WRITE); @@ -349,9 +349,9 @@ retry: errmsg("terminating connection due to unexpected postmaster exit"))); /* Handle interrupt. */ - if (event.events & WL_LATCH_SET) + if (event.events & WL_INTERRUPT) { - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); ProcessClientWriteInterrupt(true); /* diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c index 896e1476b5..3b7bc99ca7 100644 --- a/src/backend/libpq/pqcomm.c +++ b/src/backend/libpq/pqcomm.c @@ -77,6 +77,7 @@ #include "miscadmin.h" #include "port/pg_bswap.h" #include "postmaster/postmaster.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "utils/guc_hooks.h" #include "utils/memutils.h" @@ -175,7 +176,7 @@ pq_init(ClientSocket *client_sock) { Port *port; int socket_pos PG_USED_FOR_ASSERTS_ONLY; - int latch_pos PG_USED_FOR_ASSERTS_ONLY; + int interrupt_pos PG_USED_FOR_ASSERTS_ONLY; /* allocate the Port struct and copy the ClientSocket contents to it */ port = palloc0(sizeof(Port)); @@ -287,8 +288,8 @@ pq_init(ClientSocket *client_sock) /* * In backends (as soon as forked) we operate the underlying socket in - * nonblocking mode and use latches to implement blocking semantics if - * needed. That allows us to provide safely interruptible reads and + * nonblocking mode and use WaitEventSet to implement blocking semantics + * if needed. That allows us to provide safely interruptible reads and * writes. */ #ifndef WIN32 @@ -306,18 +307,18 @@ pq_init(ClientSocket *client_sock) FeBeWaitSet = CreateWaitEventSet(NULL, FeBeWaitSetNEvents); socket_pos = AddWaitEventToSet(FeBeWaitSet, WL_SOCKET_WRITEABLE, - port->sock, NULL, NULL); - latch_pos = AddWaitEventToSet(FeBeWaitSet, WL_LATCH_SET, PGINVALID_SOCKET, - MyLatch, NULL); + port->sock, 0, NULL); + interrupt_pos = AddWaitEventToSet(FeBeWaitSet, WL_INTERRUPT, PGINVALID_SOCKET, + 1 << INTERRUPT_GENERAL_WAKEUP, NULL); AddWaitEventToSet(FeBeWaitSet, WL_POSTMASTER_DEATH, PGINVALID_SOCKET, - NULL, NULL); + 0, NULL); /* * The event positions match the order we added them, but let's sanity * check them to be sure. */ Assert(socket_pos == FeBeWaitSetSocketPos); - Assert(latch_pos == FeBeWaitSetLatchPos); + Assert(interrupt_pos == FeBeWaitSetInterruptPos); return port; } @@ -2060,7 +2061,7 @@ pq_check_connection(void) * It's OK to modify the socket event filter without restoring, because * all FeBeWaitSet socket wait sites do the same. */ - ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, WL_SOCKET_CLOSED, NULL); + ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, WL_SOCKET_CLOSED, 0); retry: rc = WaitEventSetWait(FeBeWaitSet, 0, events, lengthof(events), 0); @@ -2068,15 +2069,15 @@ retry: { if (events[i].events & WL_SOCKET_CLOSED) return false; - if (events[i].events & WL_LATCH_SET) + if (events[i].events & WL_INTERRUPT) { /* - * A latch event might be preventing other events from being + * An interrupt event might be preventing other events from being * reported. Reset it and poll again. No need to restore it - * because no code should expect latches to survive across - * CHECK_FOR_INTERRUPTS(). + * because no code should expect INTERRUPT_GENERAL_WAKEUP to + * survive across CHECK_FOR_INTERRUPTS(). */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); goto retry; } } diff --git a/src/backend/postmaster/auxprocess.c b/src/backend/postmaster/auxprocess.c index 74b8a00c94..736e230bd9 100644 --- a/src/backend/postmaster/auxprocess.c +++ b/src/backend/postmaster/auxprocess.c @@ -28,6 +28,7 @@ #include "storage/proc.h" #include "storage/procsignal.h" #include "utils/memutils.h" +#include "utils/resowner.h" #include "utils/ps_status.h" diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c index e556d7ecee..cbb5c7a7a4 100644 --- a/src/backend/postmaster/checkpointer.c +++ b/src/backend/postmaster/checkpointer.c @@ -49,6 +49,7 @@ #include "storage/bufmgr.h" #include "storage/condition_variable.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/lwlock.h" #include "storage/proc.h" @@ -343,7 +344,7 @@ CheckpointerMain(char *startup_data, size_t startup_data_len) bool chkpt_or_rstpt_timed = false; /* Clear any already-pending wakeups */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* * Process any requests or signals received recently. @@ -544,10 +545,10 @@ CheckpointerMain(char *startup_data, size_t startup_data_len) cur_timeout = Min(cur_timeout, XLogArchiveTimeout - elapsed_secs); } - (void) WaitLatch(MyLatch, - WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, - cur_timeout * 1000L /* convert to ms */ , - WAIT_EVENT_CHECKPOINTER_MAIN); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + cur_timeout * 1000L /* convert to ms */ , + WAIT_EVENT_CHECKPOINTER_MAIN); } } @@ -747,10 +748,11 @@ CheckpointWriteDelay(int flags, double progress) * Checkpointer and bgwriter are no longer related so take the Big * Sleep. */ - WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH | WL_TIMEOUT, - 100, - WAIT_EVENT_CHECKPOINT_WRITE_DELAY); - ResetLatch(MyLatch); + WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_EXIT_ON_PM_DEATH | WL_TIMEOUT, + 100, + WAIT_EVENT_CHECKPOINT_WRITE_DELAY); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); } else if (--absorb_counter <= 0) { @@ -862,7 +864,7 @@ ReqCheckpointHandler(SIGNAL_ARGS) * The signaling process should have set ckpt_flags nonzero, so all we * need do is ensure that our main loop gets kicked out of any wait. */ - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } @@ -1135,7 +1137,7 @@ ForwardSyncRequest(const FileTag *ftag, SyncRequestType type) checkpointerProc = procglobal->checkpointerProc; if (checkpointerProc != INVALID_PROC_NUMBER) - SetLatch(&GetPGProcByNumber(checkpointerProc)->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, ProcGlobal->checkpointerProc); } return true; diff --git a/src/backend/postmaster/interrupt.c b/src/backend/postmaster/interrupt.c index eedc0980cf..8cbde0698c 100644 --- a/src/backend/postmaster/interrupt.c +++ b/src/backend/postmaster/interrupt.c @@ -18,8 +18,8 @@ #include "miscadmin.h" #include "postmaster/interrupt.h" +#include "storage/interrupt.h" #include "storage/ipc.h" -#include "storage/latch.h" #include "storage/procsignal.h" #include "utils/guc.h" #include "utils/memutils.h" @@ -61,7 +61,7 @@ void SignalHandlerForConfigReload(SIGNAL_ARGS) { ConfigReloadPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* @@ -105,5 +105,5 @@ void SignalHandlerForShutdownRequest(SIGNAL_ARGS) { ShutdownRequestPending = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c index 02f91431f5..01405bf2a0 100644 --- a/src/backend/postmaster/pgarch.c +++ b/src/backend/postmaster/pgarch.c @@ -41,8 +41,8 @@ #include "postmaster/pgarch.h" #include "storage/condition_variable.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" -#include "storage/latch.h" #include "storage/pmsignal.h" #include "storage/proc.h" #include "storage/procsignal.h" @@ -247,8 +247,8 @@ PgArchiverMain(char *startup_data, size_t startup_data_len) on_shmem_exit(pgarch_die, 0); /* - * Advertise our proc number so that backends can use our latch to wake us - * up while we're sleeping. + * Advertise our proc number so that backends can wake us up while we're + * sleeping. */ PgArch->pgprocno = MyProcNumber; @@ -282,13 +282,12 @@ PgArchWakeup(void) int arch_pgprocno = PgArch->pgprocno; /* - * We don't acquire ProcArrayLock here. It's actually fine because - * procLatch isn't ever freed, so we just can potentially set the wrong - * process' (or no process') latch. Even in that case the archiver will - * be relaunched shortly and will start archiving. + * We don't acquire ProcArrayLock here, so we may send the interrupt to + * wrong process, but that's harmless. Even in that case the archiver + * will be relaunched shortly and will start archiving. */ if (arch_pgprocno != INVALID_PROC_NUMBER) - SetLatch(&ProcGlobal->allProcs[arch_pgprocno].procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, arch_pgprocno); } @@ -298,7 +297,7 @@ pgarch_waken_stop(SIGNAL_ARGS) { /* set flag to do a final cycle and shut down afterwards */ ready_to_stop = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* @@ -318,7 +317,7 @@ pgarch_MainLoop(void) */ do { - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* When we get SIGUSR2, we do one more archive cycle, then exit */ time_to_stop = ready_to_stop; @@ -355,10 +354,10 @@ pgarch_MainLoop(void) { int rc; - rc = WaitLatch(MyLatch, - WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, - PGARCH_AUTOWAKE_INTERVAL * 1000L, - WAIT_EVENT_ARCHIVER_MAIN); + rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_TIMEOUT | WL_POSTMASTER_DEATH, + PGARCH_AUTOWAKE_INTERVAL * 1000L, + WAIT_EVENT_ARCHIVER_MAIN); if (rc & WL_POSTMASTER_DEATH) time_to_stop = true; } diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c index a6fff93db3..9bb4678954 100644 --- a/src/backend/postmaster/postmaster.c +++ b/src/backend/postmaster/postmaster.c @@ -112,6 +112,7 @@ #include "replication/slotsync.h" #include "replication/walsender.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/pmsignal.h" #include "storage/proc.h" @@ -528,8 +529,7 @@ PostmasterMain(int argc, char *argv[]) pqsignal(SIGCHLD, handle_pm_child_exit_signal); /* This may configure SIGURG, depending on platform. */ - InitializeLatchSupport(); - InitProcessLocalLatch(); + InitializeWaitEventSupport(); /* * No other place in Postgres should touch SIGTTIN/SIGTTOU handling. We @@ -1580,14 +1580,14 @@ ConfigurePostmasterWaitSet(bool accept_connections) pm_wait_set = CreateWaitEventSet(NULL, accept_connections ? (1 + NumListenSockets) : 1); - AddWaitEventToSet(pm_wait_set, WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, + AddWaitEventToSet(pm_wait_set, WL_INTERRUPT, PGINVALID_SOCKET, 1 << INTERRUPT_GENERAL_WAKEUP, NULL); if (accept_connections) { for (int i = 0; i < NumListenSockets; i++) AddWaitEventToSet(pm_wait_set, WL_SOCKET_ACCEPT, ListenSockets[i], - NULL, NULL); + 0, NULL); } } @@ -1616,19 +1616,20 @@ ServerLoop(void) 0 /* postmaster posts no wait_events */ ); /* - * Latch set by signal handler, or new connection pending on any of - * our sockets? If the latter, fork a child process to deal with it. + * Interrupt raised by signal handler, or new connection pending on + * any of our sockets? If the latter, fork a child process to deal + * with it. */ for (int i = 0; i < nevents; i++) { - if (events[i].events & WL_LATCH_SET) - ResetLatch(MyLatch); + if (events[i].events & WL_INTERRUPT) + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* * The following requests are handled unconditionally, even if we - * didn't see WL_LATCH_SET. This gives high priority to shutdown - * and reload requests where the latch happens to appear later in - * events[] or will be reported by a later call to + * didn't see WL_INTERRUPT. This gives high priority to shutdown + * and reload requests where the interrupt event happens to appear + * later in events[] or will be reported by a later call to * WaitEventSetWait(). */ if (pending_pm_shutdown_request) @@ -1937,7 +1938,7 @@ static void handle_pm_pmsignal_signal(SIGNAL_ARGS) { pending_pm_pmsignal = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* @@ -1947,7 +1948,7 @@ static void handle_pm_reload_request_signal(SIGNAL_ARGS) { pending_pm_reload_request = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* @@ -2044,7 +2045,7 @@ handle_pm_shutdown_request_signal(SIGNAL_ARGS) pending_pm_shutdown_request = true; break; } - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* @@ -2205,7 +2206,7 @@ static void handle_pm_child_exit_signal(SIGNAL_ARGS) { pending_pm_child_exit = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* diff --git a/src/backend/postmaster/startup.c b/src/backend/postmaster/startup.c index ef6f98ebcd..ddbe29fbe0 100644 --- a/src/backend/postmaster/startup.c +++ b/src/backend/postmaster/startup.c @@ -28,6 +28,7 @@ #include "postmaster/startup.h" #include "storage/ipc.h" #include "storage/pmsignal.h" +#include "storage/proc.h" #include "storage/procsignal.h" #include "storage/standby.h" #include "utils/guc.h" @@ -40,7 +41,7 @@ * On systems that need to make a system call to find out if the postmaster has * gone away, we'll do so only every Nth call to HandleStartupProcInterrupts(). * This only affects how long it takes us to detect the condition while we're - * busy replaying WAL. Latch waits and similar which should react immediately + * busy replaying WAL. Interrupt waits and similar which should react immediately * through the usual techniques. */ #define POSTMASTER_POLL_RATE_LIMIT 1024 @@ -205,6 +206,8 @@ StartupProcExit(int code, Datum arg) /* Shutdown the recovery environment */ if (standbyState != STANDBY_DISABLED) ShutdownRecoveryTransactionEnvironment(); + + ProcGlobal->startupProc = INVALID_PROC_NUMBER; } @@ -220,6 +223,12 @@ StartupProcessMain(char *startup_data, size_t startup_data_len) MyBackendType = B_STARTUP; AuxiliaryProcessMainCommon(); + /* + * Advertise our proc number so that backends can wake us up, when the + * server is promoted or recovery is paused/resumed. + */ + ProcGlobal->startupProc = MyProcNumber; + /* Arrange to clean up at startup process exit */ on_shmem_exit(StartupProcExit, 0); diff --git a/src/backend/postmaster/syslogger.c b/src/backend/postmaster/syslogger.c index 7951599fa8..b33033b290 100644 --- a/src/backend/postmaster/syslogger.c +++ b/src/backend/postmaster/syslogger.c @@ -338,9 +338,9 @@ SysLoggerMain(char *startup_data, size_t startup_data_len) * (including the postmaster). */ wes = CreateWaitEventSet(NULL, 2); - AddWaitEventToSet(wes, WL_LATCH_SET, PGINVALID_SOCKET, MyLatch, NULL); + AddWaitEventToSet(wes, WL_LATCH_SET, PGINVALID_SOCKET, 1 << INTERRUPT_GENERAL_WAKEUP, NULL); #ifndef WIN32 - AddWaitEventToSet(wes, WL_SOCKET_READABLE, syslogPipe[0], NULL, NULL); + AddWaitEventToSet(wes, WL_SOCKET_READABLE, syslogPipe[0], 0, NULL); #endif /* main worker loop */ diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c index 48350bec52..096b1c2e03 100644 --- a/src/backend/postmaster/walsummarizer.c +++ b/src/backend/postmaster/walsummarizer.c @@ -646,7 +646,7 @@ WakeupWalSummarizer(void) LWLockRelease(WALSummarizerLock); if (pgprocno != INVALID_PROC_NUMBER) - SetLatch(&ProcGlobal->allProcs[pgprocno].procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, pgprocno); } /* diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c index 0c55d9fa59..23f5e1ea0f 100644 --- a/src/backend/postmaster/walwriter.c +++ b/src/backend/postmaster/walwriter.c @@ -54,6 +54,7 @@ #include "storage/bufmgr.h" #include "storage/condition_variable.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/lwlock.h" #include "storage/proc.h" @@ -238,7 +239,7 @@ WalWriterMain(char *startup_data, size_t startup_data_len) } /* Clear any already-pending wakeups */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* Process any signals received recently */ HandleMainLoopInterrupts(); @@ -265,9 +266,9 @@ WalWriterMain(char *startup_data, size_t startup_data_len) else cur_timeout = WalWriterDelay * HIBERNATE_FACTOR; - (void) WaitLatch(MyLatch, - WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, - cur_timeout, - WAIT_EVENT_WAL_WRITER_MAIN); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, + WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + cur_timeout, + WAIT_EVENT_WAL_WRITER_MAIN); } } diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index c566d50a07..84081dcf83 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -703,7 +703,7 @@ logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker) { Assert(LWLockHeldByMe(LogicalRepWorkerLock)); - SetLatch(&worker->proc->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(worker->proc)); } /* diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index fa5988c824..d8a52405b0 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -902,7 +902,7 @@ SyncRepWakeQueue(bool all, int mode) /* * Wake only when we have set state and removed from queue. */ - SetLatch(&(proc->procLatch)); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(proc)); numprocs++; } diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index d1d99de980..40d1fa4939 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -1367,7 +1367,7 @@ WalRcvForceReply(void) procno = WalRcv->procno; SpinLockRelease(&WalRcv->mutex); if (procno != INVALID_PROC_NUMBER) - SetLatch(&GetPGProcByNumber(procno)->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, procno); } /* diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c index b1780d0106..d015f03244 100644 --- a/src/backend/replication/walreceiverfuncs.c +++ b/src/backend/replication/walreceiverfuncs.c @@ -318,7 +318,7 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, if (launch) SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER); else if (walrcv_proc != INVALID_PROC_NUMBER) - SetLatch(&GetPGProcByNumber(walrcv_proc)->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, walrcv_proc); } /* diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 866b69ec85..7ce8f019f7 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -80,6 +80,7 @@ #include "replication/walsender_private.h" #include "storage/condition_variable.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/pmsignal.h" #include "storage/proc.h" @@ -1042,7 +1043,7 @@ StartReplication(StartReplicationCmd *cmd) * walsender process. * * Inside the walsender we can do better than read_local_xlog_page, - * which has to do a plain sleep/busy loop, because the walsender's latch gets + * which has to do a plain sleep/busy loop, because the walsender's interrupt gets * set every time WAL is flushed. */ static int @@ -1639,7 +1640,7 @@ ProcessPendingWrites(void) WAIT_EVENT_WAL_SENDER_WRITE_DATA); /* Clear any already-pending wakeups */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); CHECK_FOR_INTERRUPTS(); @@ -1657,7 +1658,7 @@ ProcessPendingWrites(void) } /* reactivate latch so WalSndLoop knows to continue */ - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* @@ -1845,7 +1846,7 @@ WalSndWaitForWal(XLogRecPtr loc) long sleeptime; /* Clear any already-pending wakeups */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); CHECK_FOR_INTERRUPTS(); @@ -1966,7 +1967,7 @@ WalSndWaitForWal(XLogRecPtr loc) } /* reactivate latch so WalSndLoop knows to continue */ - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); return RecentFlushPtr; } @@ -2783,7 +2784,7 @@ WalSndLoop(WalSndSendDataCallback send_data) for (;;) { /* Clear any already-pending wakeups */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); CHECK_FOR_INTERRUPTS(); @@ -3582,7 +3583,7 @@ static void WalSndLastCycleHandler(SIGNAL_ARGS) { got_SIGUSR2 = true; - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); } /* Set up signal handlers */ @@ -3688,7 +3689,7 @@ WalSndWait(uint32 socket_events, long timeout, uint32 wait_event) { WaitEvent event; - ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, socket_events, NULL); + ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, socket_events, 0); /* * We use a condition variable to efficiently wake up walsenders in @@ -3700,8 +3701,8 @@ WalSndWait(uint32 socket_events, long timeout, uint32 wait_event) * ConditionVariableSleep()). It still uses WaitEventSetWait() for * waiting, because we also need to wait for socket events. The processes * (startup process, walreceiver etc.) wanting to wake up walsenders use - * ConditionVariableBroadcast(), which in turn calls SetLatch(), helping - * walsenders come out of WaitEventSetWait(). + * ConditionVariableBroadcast(), which in turn calls SendInterrupt(), + * helping walsenders come out of WaitEventSetWait(). * * This approach is simple and efficient because, one doesn't have to loop * through all the walsenders slots, with a spinlock acquisition and diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c index dffdd57e9b..383c598c5d 100644 --- a/src/backend/storage/buffer/freelist.c +++ b/src/backend/storage/buffer/freelist.c @@ -239,7 +239,7 @@ StrategyGetBuffer(BufferAccessStrategy strategy, uint32 *buf_state, bool *from_r * actually fine because procLatch isn't ever freed, so we just can * potentially set the wrong process' (or no process') latch. */ - SetLatch(&ProcGlobal->allProcs[bgwprocno].procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, bgwprocno); } /* diff --git a/src/backend/storage/ipc/Makefile b/src/backend/storage/ipc/Makefile index d8a1653eb6..5c7c72f902 100644 --- a/src/backend/storage/ipc/Makefile +++ b/src/backend/storage/ipc/Makefile @@ -13,9 +13,9 @@ OBJS = \ dsm.o \ dsm_impl.o \ dsm_registry.o \ + interrupt.o \ ipc.o \ ipci.o \ - latch.o \ pmsignal.o \ procarray.o \ procsignal.o \ @@ -25,6 +25,7 @@ OBJS = \ signalfuncs.o \ sinval.o \ sinvaladt.o \ - standby.o + standby.o \ + waiteventset.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/storage/ipc/interrupt.c b/src/backend/storage/ipc/interrupt.c new file mode 100644 index 0000000000..147349842c --- /dev/null +++ b/src/backend/storage/ipc/interrupt.c @@ -0,0 +1,114 @@ +/*------------------------------------------------------------------------- + * + * interrupt.c + * Interrupt handling routines. + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/backend/storage/interrupt.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include + +#include "miscadmin.h" +#include "port/atomics.h" +#include "storage/interrupt.h" +#include "storage/ipc.h" +#include "storage/proc.h" +#include "storage/procsignal.h" +#include "storage/waiteventset.h" +#include "utils/guc.h" +#include "utils/memutils.h" + +static pg_atomic_uint32 LocalPendingInterrupts; +static pg_atomic_uint32 LocalMaybeSleepingOnInterrupts; + +pg_atomic_uint32 *MyPendingInterrupts = &LocalPendingInterrupts; +pg_atomic_uint32 *MyMaybeSleepingOnInterrupts = &LocalMaybeSleepingOnInterrupts; + +/* + * 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; + MyMaybeSleepingOnInterrupts = &LocalMaybeSleepingOnInterrupts; + + /* + * 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->pendingInterrupts, 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->pendingInterrupts) + return; + + MyPendingInterrupts = &MyProc->pendingInterrupts; + MyMaybeSleepingOnInterrupts = &MyProc->maybeSleepingOnInterrupts; + + /* + * 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); + WakeupMyProc(); +} + +/* + * 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->pendingInterrupts, 1 << reason); + WakeupOtherProc(proc); +} diff --git a/src/backend/storage/ipc/meson.build b/src/backend/storage/ipc/meson.build index 5a936171f7..4eba41b78a 100644 --- a/src/backend/storage/ipc/meson.build +++ b/src/backend/storage/ipc/meson.build @@ -5,9 +5,9 @@ backend_sources += files( 'dsm.c', 'dsm_impl.c', 'dsm_registry.c', + 'interrupt.c', 'ipc.c', 'ipci.c', - 'latch.c', 'pmsignal.c', 'procarray.c', 'procsignal.c', @@ -18,5 +18,6 @@ backend_sources += files( 'sinval.c', 'sinvaladt.c', 'standby.c', + 'waiteventset.c', ) diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c index 9235fcd08e..ebd6fa264a 100644 --- a/src/backend/storage/ipc/shm_mq.c +++ b/src/backend/storage/ipc/shm_mq.c @@ -4,7 +4,7 @@ * single-reader, single-writer shared memory message queue * * Both the sender and the receiver must have a PGPROC; their respective - * process latches are used for synchronization. Only the sender may send, + * process interrupts are used for synchronization. Only the sender may send, * and only the receiver may receive. This is intended to allow a user * backend to communicate with worker backends that it has registered. * @@ -22,6 +22,7 @@ #include "pgstat.h" #include "port/pg_bitutils.h" #include "postmaster/bgworker.h" +#include "storage/interrupt.h" #include "storage/shm_mq.h" #include "storage/spin.h" #include "utils/memutils.h" @@ -44,9 +45,9 @@ * * mq_detached needs no locking. It can be set by either the sender or the * receiver, but only ever from false to true, so redundant writes don't - * matter. It is important that if we set mq_detached and then set the - * counterparty's latch, the counterparty must be certain to see the change - * after waking up. Since SetLatch begins with a memory barrier and ResetLatch + * matter. It is important that if we set mq_detached and then send the + * interrup to the counterparty, the counterparty must be certain to see the change + * after waking up. Since SendInterrupt begins with a memory barrier and ClearInterrup * ends with one, this should be OK. * * mq_ring_size and mq_ring_offset never change after initialization, and @@ -112,7 +113,7 @@ struct shm_mq * yet updated in the shared memory. We will not update it until the written * data is 1/4th of the ring size or the tuple queue is full. This will * prevent frequent CPU cache misses, and it will also avoid frequent - * SetLatch() calls, which are quite expensive. + * SendInterrupt() calls, which are quite expensive. * * mqh_partial_bytes, mqh_expected_bytes, and mqh_length_word_complete * are used to track the state of non-blocking operations. When the caller @@ -214,7 +215,7 @@ shm_mq_set_receiver(shm_mq *mq, PGPROC *proc) SpinLockRelease(&mq->mq_mutex); if (sender != NULL) - SetLatch(&sender->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(sender)); } /* @@ -232,7 +233,7 @@ shm_mq_set_sender(shm_mq *mq, PGPROC *proc) SpinLockRelease(&mq->mq_mutex); if (receiver != NULL) - SetLatch(&receiver->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(receiver)); } /* @@ -341,14 +342,14 @@ shm_mq_send(shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, * Write a message into a shared message queue, gathered from multiple * addresses. * - * When nowait = false, we'll wait on our process latch when the ring buffer + * When nowait = false, we'll wait on our process interrupt when the ring buffer * fills up, and then continue writing once the receiver has drained some data. - * The process latch is reset after each wait. + * The process interrupt is cleared after each wait. * - * When nowait = true, we do not manipulate the state of the process latch; + * When nowait = true, we do not manipulate the state of the process interrupt; * instead, if the buffer becomes full, we return SHM_MQ_WOULD_BLOCK. In * this case, the caller should call this function again, with the same - * arguments, each time the process latch is set. (Once begun, the sending + * arguments, each time the process interrupt is set. (Once begun, the sending * of a message cannot be aborted except by detaching from the queue; changing * the length or payload will corrupt the queue.) * @@ -539,7 +540,7 @@ shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait, { shm_mq_inc_bytes_written(mq, mqh->mqh_send_pending); if (receiver != NULL) - SetLatch(&receiver->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(receiver)); mqh->mqh_send_pending = 0; } @@ -557,16 +558,16 @@ shm_mq_sendv(shm_mq_handle *mqh, shm_mq_iovec *iov, int iovcnt, bool nowait, * while still allowing longer messages. In either case, the return value * remains valid until the next receive operation is performed on the queue. * - * When nowait = false, we'll wait on our process latch when the ring buffer + * When nowait = false, we'll wait on our process interrupt when the ring buffer * is empty and we have not yet received a full message. The sender will - * set our process latch after more data has been written, and we'll resume + * set our process interrupt after more data has been written, and we'll resume * processing. Each call will therefore return a complete message * (unless the sender detaches the queue). * - * When nowait = true, we do not manipulate the state of the process latch; + * When nowait = true, we do not manipulate the state of the process interrupt; * instead, whenever the buffer is empty and we need to read from it, we * return SHM_MQ_WOULD_BLOCK. In this case, the caller should call this - * function again after the process latch has been set. + * function again after the process interrupt has been set. */ shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait) @@ -619,8 +620,8 @@ shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait) * If we've consumed an amount of data greater than 1/4th of the ring * size, mark it consumed in shared memory. We try to avoid doing this * unnecessarily when only a small amount of data has been consumed, - * because SetLatch() is fairly expensive and we don't want to do it too - * often. + * because SendInterrupt() is fairly expensive and we don't want to do it + * too often. */ if (mqh->mqh_consume_pending > mq->mq_ring_size / 4) { @@ -895,7 +896,7 @@ shm_mq_detach_internal(shm_mq *mq) SpinLockRelease(&mq->mq_mutex); if (victim != NULL) - SetLatch(&victim->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(victim)); } /* @@ -993,7 +994,7 @@ shm_mq_send_bytes(shm_mq_handle *mqh, Size nbytes, const void *data, * Therefore, we can read it without acquiring the spinlock. */ Assert(mqh->mqh_counterparty_attached); - SetLatch(&mq->mq_receiver->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(mq->mq_receiver)); /* * We have just updated the mqh_send_pending bytes in the shared @@ -1001,7 +1002,7 @@ shm_mq_send_bytes(shm_mq_handle *mqh, Size nbytes, const void *data, */ mqh->mqh_send_pending = 0; - /* Skip manipulation of our latch if nowait = true. */ + /* Skip waiting if nowait = true. */ if (nowait) { *bytes_written = sent; @@ -1009,17 +1010,17 @@ shm_mq_send_bytes(shm_mq_handle *mqh, Size nbytes, const void *data, } /* - * Wait for our latch to be set. It might already be set for some + * Wait for the interrupt. It might already be set for some * unrelated reason, but that'll just result in one extra trip - * through the loop. It's worth it to avoid resetting the latch - * at top of loop, because setting an already-set latch is much - * cheaper than setting one that has been reset. + * through the loop. It's worth it to avoid clearing the + * interrupt at top of loop, because setting an already-set + * interrupt is much cheaper than setting one that has been reset. */ - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - WAIT_EVENT_MESSAGE_QUEUE_SEND); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + WAIT_EVENT_MESSAGE_QUEUE_SEND); - /* Reset the latch so we don't spin. */ - ResetLatch(MyLatch); + /* Clear the interrupt so we don't spin. */ + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* An interrupt may have occurred while we were waiting. */ CHECK_FOR_INTERRUPTS(); @@ -1054,7 +1055,7 @@ shm_mq_send_bytes(shm_mq_handle *mqh, Size nbytes, const void *data, /* * For efficiency, we don't update the bytes written in the shared - * memory and also don't set the reader's latch here. Refer to + * memory and also don't send the reader interrupt here. Refer to * the comments atop the shm_mq_handle structure for more * information. */ @@ -1150,22 +1151,22 @@ shm_mq_receive_bytes(shm_mq_handle *mqh, Size bytes_needed, bool nowait, mqh->mqh_consume_pending = 0; } - /* Skip manipulation of our latch if nowait = true. */ + /* Skip waiting if nowait = true. */ if (nowait) return SHM_MQ_WOULD_BLOCK; /* - * Wait for our latch to be set. It might already be set for some + * Wait for the interrupt to be set. It might already be set for some * unrelated reason, but that'll just result in one extra trip through - * the loop. It's worth it to avoid resetting the latch at top of - * loop, because setting an already-set latch is much cheaper than - * setting one that has been reset. + * the loop. It's worth it to avoid clearing the interrupt at top of + * loop, because setting an already-set interrupt is much cheaper than + * setting one that has been cleared. */ - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - WAIT_EVENT_MESSAGE_QUEUE_RECEIVE); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + WAIT_EVENT_MESSAGE_QUEUE_RECEIVE); - /* Reset the latch so we don't spin. */ - ResetLatch(MyLatch); + /* Clear the interrupt so we don't spin. */ + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* An interrupt may have occurred while we were waiting. */ CHECK_FOR_INTERRUPTS(); @@ -1250,11 +1251,11 @@ shm_mq_wait_internal(shm_mq *mq, PGPROC **ptr, BackgroundWorkerHandle *handle) } /* Wait to be signaled. */ - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - WAIT_EVENT_MESSAGE_QUEUE_INTERNAL); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + WAIT_EVENT_MESSAGE_QUEUE_INTERNAL); - /* Reset the latch so we don't spin. */ - ResetLatch(MyLatch); + /* Clear the interrupt so we don't spin. */ + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* An interrupt may have occurred while we were waiting. */ CHECK_FOR_INTERRUPTS(); @@ -1293,7 +1294,7 @@ shm_mq_inc_bytes_read(shm_mq *mq, Size n) */ sender = mq->mq_sender; Assert(sender != NULL); - SetLatch(&sender->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(sender)); } /* diff --git a/src/backend/storage/ipc/latch.c b/src/backend/storage/ipc/waiteventset.c similarity index 78% rename from src/backend/storage/ipc/latch.c rename to src/backend/storage/ipc/waiteventset.c index 608eb66abe..9142917722 100644 --- a/src/backend/storage/ipc/latch.c +++ b/src/backend/storage/ipc/waiteventset.c @@ -1,12 +1,29 @@ /*------------------------------------------------------------------------- * - * latch.c - * Routines for inter-process latches + * waiteventset.c + * ppoll()/pselect() like abstraction + * + * WaitEvents are an abstraction for waiting for one or more events at a time. + * The waiting can be done in race free fashion, similar ppoll() or pselect() + * (as opposed to plain poll()/select()). + * + * You can wait for: + * - an interrupt from another process or from signal handler in the same + * process (WL_INTERRUPT) + * - data to become readable or writeable on a socket (WL_SOCKET_*) + * - postmaster death (WL_POSTMASTER_DEATH or WL_EXIT_ON_PM_DEATH) + * - timeout (WL_TIMEOUT) + * + * XXX The latch abstraction is built on top of these functions and the interrupts. + * The WL_LATCH_SET flag is built on that. + * + * Implementation + * -------------- * * The poll() implementation uses the so-called self-pipe trick to overcome the * race condition involved with poll() and setting a global flag in the signal * handler. When a latch is set and the current process is waiting for it, the - * signal handler wakes up the poll() in WaitLatch by writing a byte to a pipe. + * signal handler wakes up the poll() in WaitInterrupt by writing a byte to a pipe. * A signal by itself doesn't interrupt poll() on all platforms, and even on * platforms where it does, a signal that arrives just before the poll() call * does not prevent poll() from entering sleep. An incoming byte on a pipe @@ -27,7 +44,7 @@ * Portions Copyright (c) 1994, Regents of the University of California * * IDENTIFICATION - * src/backend/storage/ipc/latch.c + * src/backend/storage/ipc/waiteventset.c * *------------------------------------------------------------------------- */ @@ -57,9 +74,11 @@ #include "portability/instr_time.h" #include "postmaster/postmaster.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" -#include "storage/latch.h" #include "storage/pmsignal.h" +#include "storage/proc.h" +#include "storage/waiteventset.h" #include "utils/memutils.h" #include "utils/resowner.h" @@ -98,7 +117,7 @@ #endif #endif -/* typedef in latch.h */ +/* typedef in waitevents.h */ struct WaitEventSet { ResourceOwner owner; @@ -113,13 +132,13 @@ struct WaitEventSet WaitEvent *events; /* - * If WL_LATCH_SET is specified in any wait event, latch is a pointer to - * said latch, and latch_pos the offset in the ->events array. This is - * useful because we check the state of the latch before performing doing - * syscalls related to waiting. + * If WL_INTERRUPT is specified in any wait event, interruptMask is the + * interrupts to wait for, and interrupt_pos the offset in the ->events + * array. This is useful because we check the state of pending interrupts + * before performing doing syscalls related to waiting. */ - Latch *latch; - int latch_pos; + uint32 interrupt_mask; + int interrupt_pos; /* * WL_EXIT_ON_PM_DEATH is converted to WL_POSTMASTER_DEATH, but this flag @@ -151,14 +170,14 @@ struct WaitEventSet #endif }; -/* A common WaitEventSet used to implement WaitLatch() */ -static WaitEventSet *LatchWaitSet; +/* A common WaitEventSet used to implement WaitInterrupt() */ +static WaitEventSet *InterruptWaitSet; -/* The position of the latch in LatchWaitSet. */ -#define LatchWaitSetLatchPos 0 +/* The position of the interrupt in InterruptWaitSet. */ +#define InterruptWaitSetInterruptPos 0 #ifndef WIN32 -/* Are we currently in WaitLatch? The signal handler would like to know. */ +/* Are we currently in WaitInterrupt? The signal handler would like to know. */ static volatile sig_atomic_t waiting = false; #endif @@ -174,9 +193,16 @@ static int selfpipe_writefd = -1; /* Process owning the self-pipe --- needed for checking purposes */ static int selfpipe_owner_pid = 0; +#endif + +#ifdef WAIT_USE_WIN32 +static HANDLE LocalInterruptWakeupEvent; +#endif /* Private function prototypes */ -static void latch_sigurg_handler(SIGNAL_ARGS); + +#ifdef WAIT_USE_SELF_PIPE +static void interupt_sigurg_handler(SIGNAL_ARGS); static void sendSelfPipeByte(void); #endif @@ -223,13 +249,13 @@ ResourceOwnerForgetWaitEventSet(ResourceOwner owner, WaitEventSet *set) /* - * Initialize the process-local latch infrastructure. + * Initialize the process-local wait event infrastructure. * * This must be called once during startup of any process that can wait on - * latches, before it issues any InitLatch() or OwnLatch() calls. + * interrupts. */ void -InitializeLatchSupport(void) +InitializeWaitEventSupport(void) { #if defined(WAIT_USE_SELF_PIPE) int pipefd[2]; @@ -276,12 +302,12 @@ InitializeLatchSupport(void) /* * Set up the self-pipe that allows a signal handler to wake up the - * poll()/epoll_wait() in WaitLatch. Make the write-end non-blocking, so - * that SetLatch won't block if the event has already been set many times - * filling the kernel buffer. Make the read-end non-blocking too, so that - * we can easily clear the pipe by reading until EAGAIN or EWOULDBLOCK. - * Also, make both FDs close-on-exec, since we surely do not want any - * child processes messing with them. + * poll()/epoll_wait() in WaitInterrupt. Make the write-end non-blocking, + * so that SendInterrupt won't block if the event has already been set + * many times filling the kernel buffer. Make the read-end non-blocking + * too, so that we can easily clear the pipe by reading until EAGAIN or + * EWOULDBLOCK. Also, make both FDs close-on-exec, since we surely do not + * want any child processes messing with them. */ if (pipe(pipefd) < 0) elog(FATAL, "pipe() failed: %m"); @@ -302,7 +328,7 @@ InitializeLatchSupport(void) ReserveExternalFD(); ReserveExternalFD(); - pqsignal(SIGURG, latch_sigurg_handler); + pqsignal(SIGURG, interrupt_sigurg_handler); #endif #ifdef WAIT_USE_SIGNALFD @@ -340,37 +366,43 @@ InitializeLatchSupport(void) /* Ignore SIGURG, because we'll receive it via kqueue. */ pqsignal(SIGURG, SIG_IGN); #endif + +#ifdef WAIT_USE_WIN32 + LocalInterruptWakeupEvent = CreateEvent(NULL, TRUE, FALSE, NULL); + if (LocalInterruptWakeupEvent == NULL) + elog(ERROR, "CreateEvent failed: error code %lu", GetLastError()); +#endif } void -InitializeLatchWaitSet(void) +InitializeInterruptWaitSet(void) { - int latch_pos PG_USED_FOR_ASSERTS_ONLY; + int interrupt_pos PG_USED_FOR_ASSERTS_ONLY; - Assert(LatchWaitSet == NULL); + Assert(InterruptWaitSet == NULL); - /* Set up the WaitEventSet used by WaitLatch(). */ - LatchWaitSet = CreateWaitEventSet(NULL, 2); - latch_pos = AddWaitEventToSet(LatchWaitSet, WL_LATCH_SET, PGINVALID_SOCKET, - MyLatch, NULL); + /* Set up the WaitEventSet used by WaitInterrupt(). */ + InterruptWaitSet = CreateWaitEventSet(NULL, 2); + interrupt_pos = AddWaitEventToSet(InterruptWaitSet, WL_INTERRUPT, PGINVALID_SOCKET, + 0, NULL); if (IsUnderPostmaster) - AddWaitEventToSet(LatchWaitSet, WL_EXIT_ON_PM_DEATH, - PGINVALID_SOCKET, NULL, NULL); + AddWaitEventToSet(InterruptWaitSet, WL_EXIT_ON_PM_DEATH, + PGINVALID_SOCKET, 0, NULL); - Assert(latch_pos == LatchWaitSetLatchPos); + Assert(interrupt_pos == InterruptWaitSetInterruptPos); } void -ShutdownLatchSupport(void) +ShutdownWaitEventSupport(void) { #if defined(WAIT_USE_POLL) pqsignal(SIGURG, SIG_IGN); #endif - if (LatchWaitSet) + if (InterruptWaitSet) { - FreeWaitEventSet(LatchWaitSet); - LatchWaitSet = NULL; + FreeWaitEventSet(InterruptWaitSet); + InterruptWaitSet = NULL; } #if defined(WAIT_USE_SELF_PIPE) @@ -388,134 +420,24 @@ ShutdownLatchSupport(void) } /* - * Initialize a process-local latch. - */ -void -InitLatch(Latch *latch) -{ - latch->is_set = false; - latch->maybe_sleeping = false; - latch->owner_pid = MyProcPid; - latch->is_shared = false; - -#if defined(WAIT_USE_SELF_PIPE) - /* Assert InitializeLatchSupport has been called in this process */ - Assert(selfpipe_readfd >= 0 && selfpipe_owner_pid == MyProcPid); -#elif defined(WAIT_USE_SIGNALFD) - /* Assert InitializeLatchSupport has been called in this process */ - Assert(signal_fd >= 0); -#elif defined(WAIT_USE_WIN32) - latch->event = CreateEvent(NULL, TRUE, FALSE, NULL); - if (latch->event == NULL) - elog(ERROR, "CreateEvent failed: error code %lu", GetLastError()); -#endif /* WIN32 */ -} - -/* - * Initialize a shared latch that can be set from other processes. The latch - * is initially owned by no-one; use OwnLatch to associate it with the - * current process. - * - * InitSharedLatch needs to be called in postmaster before forking child - * processes, usually right after allocating the shared memory block - * containing the latch with ShmemInitStruct. (The Unix implementation - * doesn't actually require that, but the Windows one does.) Because of - * this restriction, we have no concurrency issues to worry about here. - * - * Note that other handles created in this module are never marked as - * inheritable. Thus we do not need to worry about cleaning up child - * process references to postmaster-private latches or WaitEventSets. - */ -void -InitSharedLatch(Latch *latch) -{ -#ifdef WIN32 - SECURITY_ATTRIBUTES sa; - - /* - * Set up security attributes to specify that the events are inherited. - */ - ZeroMemory(&sa, sizeof(sa)); - sa.nLength = sizeof(sa); - sa.bInheritHandle = TRUE; - - latch->event = CreateEvent(&sa, TRUE, FALSE, NULL); - if (latch->event == NULL) - elog(ERROR, "CreateEvent failed: error code %lu", GetLastError()); -#endif - - latch->is_set = false; - latch->maybe_sleeping = false; - latch->owner_pid = 0; - latch->is_shared = true; -} - -/* - * Associate a shared latch with the current process, allowing it to - * wait on the latch. - * - * Although there is a sanity check for latch-already-owned, we don't do - * any sort of locking here, meaning that we could fail to detect the error - * if two processes try to own the same latch at about the same time. If - * there is any risk of that, caller must provide an interlock to prevent it. - */ -void -OwnLatch(Latch *latch) -{ - int owner_pid; - - /* Sanity checks */ - Assert(latch->is_shared); - -#if defined(WAIT_USE_SELF_PIPE) - /* Assert InitializeLatchSupport has been called in this process */ - Assert(selfpipe_readfd >= 0 && selfpipe_owner_pid == MyProcPid); -#elif defined(WAIT_USE_SIGNALFD) - /* Assert InitializeLatchSupport has been called in this process */ - Assert(signal_fd >= 0); -#endif - - owner_pid = latch->owner_pid; - if (owner_pid != 0) - elog(PANIC, "latch already owned by PID %d", owner_pid); - - latch->owner_pid = MyProcPid; -} - -/* - * Disown a shared latch currently owned by the current process. - */ -void -DisownLatch(Latch *latch) -{ - Assert(latch->is_shared); - Assert(latch->owner_pid == MyProcPid); - - latch->owner_pid = 0; -} - -/* - * Wait for a given latch to be set, or for postmaster death, or until timeout - * is exceeded. 'wakeEvents' is a bitmask that specifies which of those events - * to wait for. If the latch is already set (and WL_LATCH_SET is given), the - * function returns immediately. + * Wait for any of the interrupts in interruptMask to be set, or for + * postmaster death, or until timeout is exceeded. 'wakeEvents' is a bitmask + * that specifies which of those events to wait for. If the interrupt is + * already pending (and WL_INTERRUPT is given), the function returns + * immediately. * * The "timeout" is given in milliseconds. It must be >= 0 if WL_TIMEOUT flag * is given. Although it is declared as "long", we don't actually support * timeouts longer than INT_MAX milliseconds. Note that some extra overhead * is incurred when WL_TIMEOUT is given, so avoid using a timeout if possible. * - * The latch must be owned by the current process, ie. it must be a - * process-local latch initialized with InitLatch, or a shared latch - * associated with the current process by calling OwnLatch. - * * Returns bit mask indicating which condition(s) caused the wake-up. Note * that if multiple wake-up conditions are true, there is no guarantee that * we return all of them in one call, but we will return at least one. */ int -WaitLatch(Latch *latch, int wakeEvents, long timeout, - uint32 wait_event_info) +WaitInterrupt(uint32 interruptMask, int wakeEvents, long timeout, + uint32 wait_event_info) { WaitEvent event; @@ -525,17 +447,17 @@ WaitLatch(Latch *latch, int wakeEvents, long timeout, (wakeEvents & WL_POSTMASTER_DEATH)); /* - * Some callers may have a latch other than MyLatch, or no latch at all, - * or want to handle postmaster death differently. It's cheap to assign - * those, so just do it every time. + * Some callers may have an interrupt mask different from last time, or no + * interrupt mask at all, or want to handle postmaster death differently. + * It's cheap to assign those, so just do it every time. */ - if (!(wakeEvents & WL_LATCH_SET)) - latch = NULL; - ModifyWaitEvent(LatchWaitSet, LatchWaitSetLatchPos, WL_LATCH_SET, latch); - LatchWaitSet->exit_on_postmaster_death = + if (!(wakeEvents & WL_INTERRUPT)) + interruptMask = 0; + ModifyWaitEvent(InterruptWaitSet, InterruptWaitSetInterruptPos, WL_INTERRUPT, interruptMask); + InterruptWaitSet->exit_on_postmaster_death = ((wakeEvents & WL_EXIT_ON_PM_DEATH) != 0); - if (WaitEventSetWait(LatchWaitSet, + if (WaitEventSetWait(InterruptWaitSet, (wakeEvents & WL_TIMEOUT) ? timeout : -1, &event, 1, wait_event_info) == 0) @@ -545,7 +467,7 @@ WaitLatch(Latch *latch, int wakeEvents, long timeout, } /* - * Like WaitLatch, but with an extra socket argument for WL_SOCKET_* + * Like WaitInterrupt, but with an extra socket argument for WL_SOCKET_* * conditions. * * When waiting on a socket, EOF and error conditions always cause the socket @@ -558,12 +480,12 @@ WaitLatch(Latch *latch, int wakeEvents, long timeout, * where some behavior other than immediate exit is needed. * * NB: These days this is just a wrapper around the WaitEventSet API. When - * using a latch very frequently, consider creating a longer living + * using an interrupt very frequently, consider creating a longer living * WaitEventSet instead; that's more efficient. */ int -WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, - long timeout, uint32 wait_event_info) +WaitInterruptOrSocket(uint32 interruptMask, int wakeEvents, pgsocket sock, + long timeout, uint32 wait_event_info) { int ret = 0; int rc; @@ -575,9 +497,9 @@ WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, else timeout = -1; - if (wakeEvents & WL_LATCH_SET) - AddWaitEventToSet(set, WL_LATCH_SET, PGINVALID_SOCKET, - latch, NULL); + if (wakeEvents & WL_INTERRUPT) + AddWaitEventToSet(set, WL_INTERRUPT, PGINVALID_SOCKET, + interruptMask, NULL); /* Postmaster-managed callers must handle postmaster death somehow. */ Assert(!IsUnderPostmaster || @@ -586,18 +508,18 @@ WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, if ((wakeEvents & WL_POSTMASTER_DEATH) && IsUnderPostmaster) AddWaitEventToSet(set, WL_POSTMASTER_DEATH, PGINVALID_SOCKET, - NULL, NULL); + 0, NULL); if ((wakeEvents & WL_EXIT_ON_PM_DEATH) && IsUnderPostmaster) AddWaitEventToSet(set, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET, - NULL, NULL); + 0, NULL); if (wakeEvents & WL_SOCKET_MASK) { int ev; ev = wakeEvents & WL_SOCKET_MASK; - AddWaitEventToSet(set, ev, sock, NULL, NULL); + AddWaitEventToSet(set, ev, sock, 0, NULL); } rc = WaitEventSetWait(set, timeout, &event, 1, wait_event_info); @@ -606,7 +528,7 @@ WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, ret |= WL_TIMEOUT; else { - ret |= event.events & (WL_LATCH_SET | + ret |= event.events & (WL_INTERRUPT | WL_POSTMASTER_DEATH | WL_SOCKET_MASK); } @@ -617,127 +539,50 @@ WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, } /* - * Sets a latch and wakes up anyone waiting on it. - * - * This is cheap if the latch is already set, otherwise not so much. + * Wakes up my process if it's currently waiting. * - * NB: when calling this in a signal handler, be sure to save and restore - * errno around it. (That's standard practice in most signal handlers, of - * course, but we used to omit it in handlers that only set a flag.) + * NB: be sure to save and restore errno around it. (That's standard practice + * in most signal handlers, of course, but we used to omit it in handlers that + * only set a flag.) * * NB: this function is called from critical sections and signal handlers so * throwing an error is not a good idea. */ void -SetLatch(Latch *latch) +WakeupMyProc(void) { #ifndef WIN32 - pid_t owner_pid; -#else - HANDLE handle; -#endif - - /* - * The memory barrier has to be placed here to ensure that any flag - * variables possibly changed by this process have been flushed to main - * memory, before we check/set is_set. - */ - pg_memory_barrier(); - - /* Quick exit if already set */ - if (latch->is_set) - return; - - latch->is_set = true; - - pg_memory_barrier(); - if (!latch->maybe_sleeping) - return; - -#ifndef WIN32 - - /* - * See if anyone's waiting for the latch. It can be the current process if - * we're in a signal handler. We use the self-pipe or SIGURG to ourselves - * to wake up WaitEventSetWaitBlock() without races in that case. If it's - * another process, send a signal. - * - * Fetch owner_pid only once, in case the latch is concurrently getting - * owned or disowned. XXX: This assumes that pid_t is atomic, which isn't - * guaranteed to be true! In practice, the effective range of pid_t fits - * in a 32 bit integer, and so should be atomic. In the worst case, we - * might end up signaling the wrong process. Even then, you're very - * unlucky if a process with that bogus pid exists and belongs to - * Postgres; and PG database processes should handle excess SIGUSR1 - * interrupts without a problem anyhow. - * - * Another sort of race condition that's possible here is for a new - * process to own the latch immediately after we look, so we don't signal - * it. This is okay so long as all callers of ResetLatch/WaitLatch follow - * the standard coding convention of waiting at the bottom of their loops, - * not the top, so that they'll correctly process latch-setting events - * that happen before they enter the loop. - */ - owner_pid = latch->owner_pid; - if (owner_pid == 0) - return; - else if (owner_pid == MyProcPid) - { #if defined(WAIT_USE_SELF_PIPE) - if (waiting) - sendSelfPipeByte(); + if (waiting) + sendSelfPipeByte(); #else - if (waiting) - kill(MyProcPid, SIGURG); + if (waiting) + kill(MyProcPid, SIGURG); #endif - } - else - kill(owner_pid, SIGURG); - #else - - /* - * See if anyone's waiting for the latch. It can be the current process if - * we're in a signal handler. - * - * Use a local variable here just in case somebody changes the event field - * concurrently (which really should not happen). - */ - handle = latch->event; - if (handle) - { - SetEvent(handle); - - /* - * Note that we silently ignore any errors. We might be in a signal - * handler or other critical path where it's not safe to call elog(). - */ - } + SetEvent(MyProc ? MyProc->interruptWakeupEvent : LocalInterruptWakeupEvent); #endif } /* - * Clear the latch. Calling WaitLatch after this will sleep, unless - * the latch is set again before the WaitLatch call. + * Wakes up another process if it's currently waiting. */ void -ResetLatch(Latch *latch) +WakeupOtherProc(PGPROC *proc) { - /* Only the owner should reset the latch */ - Assert(latch->owner_pid == MyProcPid); - Assert(latch->maybe_sleeping == false); - - latch->is_set = false; +#ifndef WIN32 + kill(proc->pid, SIGURG); +#else + SetEvent(proc->interruptWakeupEvent); /* - * Ensure that the write to is_set gets flushed to main memory before we - * examine any flag variables. Otherwise a concurrent SetLatch might - * falsely conclude that it needn't signal us, even though we have missed - * seeing some flag updates that SetLatch was supposed to inform us of. + * Note that we silently ignore any errors. We might be in a signal + * handler or other critical path where it's not safe to call elog(). */ - pg_memory_barrier(); +#endif } + /* * Create a WaitEventSet with space for nevents different events to wait for. * @@ -799,7 +644,8 @@ CreateWaitEventSet(ResourceOwner resowner, int nevents) data += MAXALIGN(sizeof(HANDLE) * nevents); #endif - set->latch = NULL; + set->interrupt_mask = 0; + set->interrupt_pos = -1; set->nevents_space = nevents; set->exit_on_postmaster_death = false; @@ -890,9 +736,9 @@ FreeWaitEventSet(WaitEventSet *set) cur_event < (set->events + set->nevents); cur_event++) { - if (cur_event->events & WL_LATCH_SET) + if (cur_event->events & WL_INTERRUPT) { - /* uses the latch's HANDLE */ + /* uses the process's wakeup HANDLE */ } else if (cur_event->events & WL_POSTMASTER_DEATH) { @@ -929,7 +775,7 @@ FreeWaitEventSetAfterFork(WaitEventSet *set) /* --- * Add an event to the set. Possible events are: - * - WL_LATCH_SET: Wait for the latch to be set + * - WL_INTERRUPT: Wait for the interrupt to be set * - WL_POSTMASTER_DEATH: Wait for postmaster to die * - WL_SOCKET_READABLE: Wait for socket to become readable, * can be combined in one event with other WL_SOCKET_* events @@ -947,10 +793,6 @@ FreeWaitEventSetAfterFork(WaitEventSet *set) * Returns the offset in WaitEventSet->events (starting from 0), which can be * used to modify previously added wait events using ModifyWaitEvent(). * - * In the WL_LATCH_SET case the latch must be owned by the current process, - * i.e. it must be a process-local latch initialized with InitLatch, or a - * shared latch associated with the current process by calling OwnLatch. - * * In the WL_SOCKET_READABLE/WRITEABLE/CONNECTED/ACCEPT cases, EOF and error * conditions cause the socket to be reported as readable/writable/connected, * so that the caller can deal with the condition. @@ -960,7 +802,7 @@ FreeWaitEventSetAfterFork(WaitEventSet *set) * events. */ int -AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch, +AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, uint32 interruptMask, void *user_data) { WaitEvent *event; @@ -974,19 +816,16 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch, set->exit_on_postmaster_death = true; } - if (latch) - { - if (latch->owner_pid != MyProcPid) - elog(ERROR, "cannot wait on a latch owned by another process"); - if (set->latch) - elog(ERROR, "cannot wait on more than one latch"); - if ((events & WL_LATCH_SET) != WL_LATCH_SET) - elog(ERROR, "latch events only support being set"); - } - else + /* + * It doesn't make much sense to wait for WL_INTERRUPT with empty + * interruptMask, but we allow it so that you can use ModifyEvent to set + * the interruptMask later. Non-zero interruptMask doesn't make sense + * without WL_INTERRUPT, however. + */ + if (interruptMask != 0) { - if (events & WL_LATCH_SET) - elog(ERROR, "cannot wait on latch without a specified latch"); + if ((events & WL_INTERRUPT) != WL_INTERRUPT) + elog(ERROR, "interrupted events only support being set"); } /* waiting for socket readiness without a socket indicates a bug */ @@ -1002,10 +841,10 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch, event->reset = false; #endif - if (events == WL_LATCH_SET) + if (events == WL_INTERRUPT) { - set->latch = latch; - set->latch_pos = event->pos; + set->interrupt_mask = interruptMask; + set->interrupt_pos = event->pos; #if defined(WAIT_USE_SELF_PIPE) event->fd = selfpipe_readfd; #elif defined(WAIT_USE_SIGNALFD) @@ -1039,14 +878,14 @@ AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, Latch *latch, } /* - * Change the event mask and, in the WL_LATCH_SET case, the latch associated - * with the WaitEvent. The latch may be changed to NULL to disable the latch - * temporarily, and then set back to a latch later. + * Change the event mask and, in the WL_INTERRUPT case, the interrupt mask + * associated with the WaitEvent. The interrupt mask may be changed to 0 to + * disable the wakeups on interrupts temporarily, and then set back later. * * 'pos' is the id returned by AddWaitEventToSet. */ void -ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch) +ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, uint32 interruptMask) { WaitEvent *event; #if defined(WAIT_USE_KQUEUE) @@ -1061,19 +900,19 @@ ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch) #endif /* - * If neither the event mask nor the associated latch changes, return - * early. That's an important optimization for some sockets, where + * If neither the event mask nor the associated interrupt mask changes, + * return early. That's an important optimization for some sockets, where * ModifyWaitEvent is frequently used to switch from waiting for reads to * waiting on writes. */ if (events == event->events && - (!(event->events & WL_LATCH_SET) || set->latch == latch)) + (!(event->events & WL_INTERRUPT) || set->interrupt_mask == interruptMask)) return; - if (event->events & WL_LATCH_SET && + if (event->events & WL_INTERRUPT && events != event->events) { - elog(ERROR, "cannot modify latch event"); + elog(ERROR, "cannot modify interrupts event"); } if (event->events & WL_POSTMASTER_DEATH) @@ -1084,25 +923,19 @@ ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch) /* FIXME: validate event mask */ event->events = events; - if (events == WL_LATCH_SET) + if (events == WL_INTERRUPT) { - if (latch && latch->owner_pid != MyProcPid) - elog(ERROR, "cannot wait on a latch owned by another process"); - set->latch = latch; + set->interrupt_mask = interruptMask; /* - * On Unix, we don't need to modify the kernel object because the - * underlying pipe (if there is one) is the same for all latches so we - * can return immediately. On Windows, we need to update our array of - * handles, but we leave the old one in place and tolerate spurious - * wakeups if the latch is disabled. + * We don't bother to adjust the event when the interrupt mask + * changes. It means that when interrupt mask is set to 0, we will + * listen on the kernel object unnecessarily, and might get some + * spurious wakeups. The interrupt sending code should not wakes us up + * if the maybeSleepingOnInterrupts is zero, though, so it should + * be rare. */ -#if defined(WAIT_USE_WIN32) - if (!latch) - return; -#else return; -#endif } #if defined(WAIT_USE_EPOLL) @@ -1132,9 +965,8 @@ WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action) epoll_ev.events = EPOLLERR | EPOLLHUP; /* prepare pollfd entry once */ - if (event->events == WL_LATCH_SET) + if (event->events == WL_INTERRUPT) { - Assert(set->latch != NULL); epoll_ev.events |= EPOLLIN; } else if (event->events == WL_POSTMASTER_DEATH) @@ -1181,9 +1013,8 @@ WaitEventAdjustPoll(WaitEventSet *set, WaitEvent *event) pollfd->fd = event->fd; /* prepare pollfd entry once */ - if (event->events == WL_LATCH_SET) + if (event->events == WL_INTERRUPT) { - Assert(set->latch != NULL); pollfd->events = POLLIN; } else if (event->events == WL_POSTMASTER_DEATH) @@ -1245,9 +1076,9 @@ WaitEventAdjustKqueueAddPostmaster(struct kevent *k_ev, WaitEvent *event) } static inline void -WaitEventAdjustKqueueAddLatch(struct kevent *k_ev, WaitEvent *event) +WaitEventAdjustKqueueAddInterruptWakeup(struct kevent *k_ev, WaitEvent *event) { - /* For now latch can only be added, not removed. */ + /* For now interrupt wakeup can only be added, not removed. */ k_ev->ident = SIGURG; k_ev->filter = EVFILT_SIGNAL; k_ev->flags = EV_ADD; @@ -1273,8 +1104,7 @@ WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events) if (old_events == event->events) return; - Assert(event->events != WL_LATCH_SET || set->latch != NULL); - Assert(event->events == WL_LATCH_SET || + Assert(event->events == WL_INTERRUPT || event->events == WL_POSTMASTER_DEATH || (event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE | @@ -1289,10 +1119,10 @@ WaitEventAdjustKqueue(WaitEventSet *set, WaitEvent *event, int old_events) */ WaitEventAdjustKqueueAddPostmaster(&k_ev[count++], event); } - else if (event->events == WL_LATCH_SET) + else if (event->events == WL_INTERRUPT) { - /* We detect latch wakeup using a signal event. */ - WaitEventAdjustKqueueAddLatch(&k_ev[count++], event); + /* We detect interrupt wakeup using a signal event. */ + WaitEventAdjustKqueueAddInterruptWakeup(&k_ev[count++], event); } else { @@ -1370,10 +1200,13 @@ WaitEventAdjustWin32(WaitEventSet *set, WaitEvent *event) { HANDLE *handle = &set->handles[event->pos + 1]; - if (event->events == WL_LATCH_SET) + if (event->events == WL_INTERRUPT) { - Assert(set->latch != NULL); - *handle = set->latch->event; + /* + * We register the event even if interrupt_mask is zero. We might get + * some spurious wakeups, but that's OK + */ + *handle = MyProc ? MyProc->interruptWakeupEvent : LocalInterruptWakeupEvent; } else if (event->events == WL_POSTMASTER_DEATH) { @@ -1450,7 +1283,7 @@ WaitEventSetWait(WaitEventSet *set, long timeout, #ifndef WIN32 waiting = true; #else - /* Ensure that signals are serviced even if latch is already set */ + /* Ensure that signals are serviced even if interrupt is already pending */ pgwin32_dispatch_queued_signals(); #endif while (returned_events == 0) @@ -1458,52 +1291,52 @@ WaitEventSetWait(WaitEventSet *set, long timeout, int rc; /* - * Check if the latch is set already. If so, leave the loop + * Check if the interrupt is already pending. If so, leave the loop * immediately, avoid blocking again. We don't attempt to report any * other events that might also be satisfied. * - * If someone sets the latch between this and the + * If someone sets the interrupt flag between this and the * WaitEventSetWaitBlock() below, the setter will write a byte to the * pipe (or signal us and the signal handler will do that), and the * readiness routine will return immediately. * * On unix, If there's a pending byte in the self pipe, we'll notice * whenever blocking. Only clearing the pipe in that case avoids - * having to drain it every time WaitLatchOrSocket() is used. Should - * the pipe-buffer fill up we're still ok, because the pipe is in - * nonblocking mode. It's unlikely for that to happen, because the + * having to drain it every time WaitInterruptOrSocket() is used. + * Should the pipe-buffer fill up we're still ok, because the pipe is + * in nonblocking mode. It's unlikely for that to happen, because the * self pipe isn't filled unless we're blocking (waiting = true), or - * from inside a signal handler in latch_sigurg_handler(). + * from inside a signal handler in interrupt_sigurg_handler(). * * On windows, we'll also notice if there's a pending event for the - * latch when blocking, but there's no danger of anything filling up, - * as "Setting an event that is already set has no effect.". + * interrupt when blocking, but there's no danger of anything filling + * up, as "Setting an event that is already set has no effect.". * - * Note: we assume that the kernel calls involved in latch management - * will provide adequate synchronization on machines with weak memory - * ordering, so that we cannot miss seeing is_set if a notification - * has already been queued. + * Note: we assume that the kernel calls involved in interrupt + * management will provide adequate synchronization on machines with + * weak memory ordering, so that we cannot miss seeing is_set if a + * notification has already been queued. */ - if (set->latch && !set->latch->is_set) + if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) == 0) { - /* about to sleep on a latch */ - set->latch->maybe_sleeping = true; + /* about to sleep wait_ing for interrupts */ + pg_atomic_write_u32(MyMaybeSleepingOnInterrupts, set->interrupt_mask); pg_memory_barrier(); /* and recheck */ } - if (set->latch && set->latch->is_set) + if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) { occurred_events->fd = PGINVALID_SOCKET; - occurred_events->pos = set->latch_pos; + occurred_events->pos = set->interrupt_pos; occurred_events->user_data = - set->events[set->latch_pos].user_data; - occurred_events->events = WL_LATCH_SET; + set->events[set->interrupt_pos].user_data; + occurred_events->events = WL_INTERRUPT; occurred_events++; returned_events++; /* could have been set above */ - set->latch->maybe_sleeping = false; + pg_atomic_write_u32(MyMaybeSleepingOnInterrupts, 0); break; } @@ -1516,10 +1349,10 @@ WaitEventSetWait(WaitEventSet *set, long timeout, rc = WaitEventSetWaitBlock(set, cur_timeout, occurred_events, nevents); - if (set->latch) + if (set->interrupt_mask != 0) { - Assert(set->latch->maybe_sleeping); - set->latch->maybe_sleeping = false; + Assert(pg_atomic_read_u32(MyMaybeSleepingOnInterrupts) == set->interrupt_mask); + pg_atomic_write_u32(MyMaybeSleepingOnInterrupts, 0); } if (rc == -1) @@ -1607,16 +1440,16 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, occurred_events->user_data = cur_event->user_data; occurred_events->events = 0; - if (cur_event->events == WL_LATCH_SET && + if (cur_event->events == WL_INTERRUPT && cur_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP)) { /* Drain the signalfd. */ drain(); - if (set->latch && set->latch->is_set) + if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) { occurred_events->fd = PGINVALID_SOCKET; - occurred_events->events = WL_LATCH_SET; + occurred_events->events = WL_INTERRUPT; occurred_events++; returned_events++; } @@ -1769,13 +1602,13 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, occurred_events->user_data = cur_event->user_data; occurred_events->events = 0; - if (cur_event->events == WL_LATCH_SET && + if (cur_event->events == WL_INTERRUPT && cur_kqueue_event->filter == EVFILT_SIGNAL) { - if (set->latch && set->latch->is_set) + if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) { occurred_events->fd = PGINVALID_SOCKET; - occurred_events->events = WL_LATCH_SET; + occurred_events->events = WL_INTERRUPT; occurred_events++; returned_events++; } @@ -1891,16 +1724,16 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, occurred_events->user_data = cur_event->user_data; occurred_events->events = 0; - if (cur_event->events == WL_LATCH_SET && + if (cur_event->events == WL_INTERRUPT && (cur_pollfd->revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL))) { /* There's data in the self-pipe, clear it. */ drain(); - if (set->latch && set->latch->is_set) + if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) { occurred_events->fd = PGINVALID_SOCKET; - occurred_events->events = WL_LATCH_SET; + occurred_events->events = WL_INTERRUPT; occurred_events++; returned_events++; } @@ -1999,6 +1832,15 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, cur_event->reset = false; } + /* + * We need to use different event object depending on whether "local" + * or "shared memory" interrupts are in use. There's no easy way to + * adjust all existing WaitEventSet when you switch from local to + * shared or back, so we refresh it on every call. + */ + if (cur_event->events & WL_INTERRUPT) + WaitEventAdjustWin32(set, cur_event); + /* * We associate the socket with a new event handle for each * WaitEventSet. FD_CLOSE is only generated once if the other end @@ -2104,19 +1946,15 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, occurred_events->user_data = cur_event->user_data; occurred_events->events = 0; - if (cur_event->events == WL_LATCH_SET) + if (cur_event->events == WL_INTERRUPT) { - /* - * We cannot use set->latch->event to reset the fired event if we - * aren't waiting on this latch now. - */ if (!ResetEvent(set->handles[cur_event->pos + 1])) elog(ERROR, "ResetEvent failed: error code %lu", GetLastError()); - if (set->latch && set->latch->is_set) + if (set->interrupt_mask != 0 && (pg_atomic_read_u32(MyPendingInterrupts) & set->interrupt_mask) != 0) { occurred_events->fd = PGINVALID_SOCKET; - occurred_events->events = WL_LATCH_SET; + occurred_events->events = WL_INTERRUPT; occurred_events++; returned_events++; } @@ -2161,7 +1999,7 @@ WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout, /*------ * WaitForMultipleObjects doesn't guarantee that a read event - * will be returned if the latch is set at the same time. Even + * will be returned if the interrupt is set at the same time. Even * if it did, the caller might drop that event expecting it to * reoccur on next call. So, we must force the event to be * reset if this WaitEventSet is used again in order to avoid @@ -2267,9 +2105,8 @@ GetNumRegisteredWaitEvents(WaitEventSet *set) #if defined(WAIT_USE_SELF_PIPE) /* - * SetLatch uses SIGURG to wake up the process waiting on the latch. - * - * Wake up WaitLatch, if we're waiting. + * WakeupOtherProc and WakupMyProc use SIGURG to wake up the process waiting + * on the latch. */ static void latch_sigurg_handler(SIGNAL_ARGS) @@ -2278,7 +2115,7 @@ latch_sigurg_handler(SIGNAL_ARGS) sendSelfPipeByte(); } -/* Send one byte to the self-pipe, to wake up WaitLatch */ +/* Send one byte to the self-pipe, to wake up WaitInterrupt */ static void sendSelfPipeByte(void) { @@ -2295,7 +2132,7 @@ retry: /* * If the pipe is full, we don't need to retry, the data that's there - * already is enough to wake up WaitLatch. + * already is enough to wake up WaitInterrupt. */ if (errno == EAGAIN || errno == EWOULDBLOCK) return; diff --git a/src/backend/storage/lmgr/condition_variable.c b/src/backend/storage/lmgr/condition_variable.c index 112a518bae..d5fabef171 100644 --- a/src/backend/storage/lmgr/condition_variable.c +++ b/src/backend/storage/lmgr/condition_variable.c @@ -268,7 +268,7 @@ ConditionVariableSignal(ConditionVariable *cv) /* If we found someone sleeping, set their latch to wake them up. */ if (proc != NULL) - SetLatch(&proc->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(proc)); } /* @@ -331,7 +331,7 @@ ConditionVariableBroadcast(ConditionVariable *cv) /* Awaken first waiter, if there was one. */ if (proc != NULL) - SetLatch(&proc->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(proc)); while (have_sentinel) { @@ -355,6 +355,6 @@ ConditionVariableBroadcast(ConditionVariable *cv) SpinLockRelease(&cv->mutex); if (proc != NULL && proc != MyProc) - SetLatch(&proc->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(proc)); } } diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 56c60704f5..0c1b238e0e 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -43,6 +43,7 @@ #include "replication/slotsync.h" #include "replication/syncrep.h" #include "storage/condition_variable.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/lmgr.h" #include "storage/pmsignal.h" @@ -224,8 +225,22 @@ InitProcGlobal(void) */ if (i < MaxBackends + NUM_AUXILIARY_PROCS) { +#ifdef WIN32 + SECURITY_ATTRIBUTES sa; + + /* + * Set up security attributes to specify that the events are + * inherited. + */ + ZeroMemory(&sa, sizeof(sa)); + sa.nLength = sizeof(sa); + sa.bInheritHandle = TRUE; + + proc->interruptWakeupEvent = CreateEvent(&sa, TRUE, FALSE, NULL); + if (proc->interruptWakeupEvent == NULL) + elog(ERROR, "CreateEvent failed: error code %lu", GetLastError()); +#endif proc->sem = PGSemaphoreCreate(); - InitSharedLatch(&(proc->procLatch)); LWLockInitialize(&(proc->fpInfoLock), LWTRANCHE_LOCK_FASTPATH); } @@ -437,13 +452,8 @@ InitProcess(void) MyProc->clogGroupMemberLsn = InvalidXLogRecPtr; Assert(pg_atomic_read_u32(&MyProc->clogGroupNext) == INVALID_PROC_NUMBER); - /* - * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch - * on it. That allows us to repoint the process latch, which so far - * points to process local one, to the shared one. - */ - OwnLatch(&MyProc->procLatch); - SwitchToSharedLatch(); + /* Start accepting interrupts from other processes */ + SwitchToSharedInterrupts(); /* now that we have a proc, report wait events to shared memory */ pgstat_set_wait_event_storage(&MyProc->wait_event_info); @@ -604,13 +614,8 @@ InitAuxiliaryProcess(void) } #endif - /* - * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch - * on it. That allows us to repoint the process latch, which so far - * points to process local one, to the shared one. - */ - OwnLatch(&MyProc->procLatch); - SwitchToSharedLatch(); + /* Start accepting interrupts from other processes */ + SwitchToSharedInterrupts(); /* now that we have a proc, report wait events to shared memory */ pgstat_set_wait_event_storage(&MyProc->wait_event_info); @@ -904,21 +909,20 @@ ProcKill(int code, Datum arg) } /* - * Reset MyLatch to the process local one. This is so that signal - * handlers et al can continue using the latch after the shared latch - * isn't ours anymore. + * Reset interrupt vector to the process local one. This is so that + * signal handlers et al can continue using interrupts after the PGPROC + * entry isn't ours anymore. * * Similarly, stop reporting wait events to MyProc->wait_event_info. * - * After that clear MyProc and disown the shared latch. + * After that clear MyProc. */ - SwitchBackToLocalLatch(); + SwitchToLocalInterrupts(); pgstat_reset_wait_event_storage(); proc = MyProc; MyProc = NULL; MyProcNumber = INVALID_PROC_NUMBER; - DisownLatch(&proc->procLatch); /* Mark the proc no longer in use */ proc->pid = 0; @@ -993,13 +997,12 @@ AuxiliaryProcKill(int code, Datum arg) ConditionVariableCancelSleep(); /* look at the equivalent ProcKill() code for comments */ - SwitchBackToLocalLatch(); + SwitchToLocalInterrupts(); pgstat_reset_wait_event_storage(); proc = MyProc; MyProc = NULL; MyProcNumber = INVALID_PROC_NUMBER; - DisownLatch(&proc->procLatch); SpinLockAcquire(ProcStructLock); @@ -1301,18 +1304,18 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait) } /* - * If somebody wakes us between LWLockRelease and WaitLatch, the latch - * will not wait. But a set latch does not necessarily mean that the lock - * is free now, as there are many other sources for latch sets than - * somebody releasing the lock. + * If somebody wakes us between LWLockRelease and WaitInterrupt, + * WaitInterrupt will not wait. But an interrupt does not necessarily mean + * that the lock is free now, as there are many other sources for the + * interrupt than somebody releasing the lock. * - * We process interrupts whenever the latch has been set, so cancel/die - * interrupts are processed quickly. This means we must not mind losing - * control to a cancel/die interrupt here. We don't, because we have no - * shared-state-change work to do after being granted the lock (the - * grantor did it all). We do have to worry about canceling the deadlock - * timeout and updating the locallock table, but if we lose control to an - * error, LockErrorCleanup will fix that up. + * We process interrupts whenever the interrupt has been set, so + * cancel/die interrupts are processed quickly. This means we must not + * mind losing control to a cancel/die interrupt here. We don't, because + * we have no shared-state-change work to do after being granted the lock + * (the grantor did it all). We do have to worry about canceling the + * deadlock timeout and updating the locallock table, but if we lose + * control to an error, LockErrorCleanup will fix that up. */ do { @@ -1358,9 +1361,9 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait) } else { - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - PG_WAIT_LOCK | locallock->tag.lock.locktag_type); - ResetLatch(MyLatch); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + PG_WAIT_LOCK | locallock->tag.lock.locktag_type); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* check for deadlocks first, as that's probably log-worthy */ if (got_deadlock_timeout) { @@ -1669,7 +1672,7 @@ ProcSleep(LOCALLOCK *locallock, LockMethod lockMethodTable, bool dontWait) /* - * ProcWakeup -- wake up a process by setting its latch. + * ProcWakeup -- wake up a process by sending it an interrupt. * * Also remove the process from the wait queue and set its links invalid. * @@ -1698,7 +1701,7 @@ ProcWakeup(PGPROC *proc, ProcWaitStatus waitStatus) pg_atomic_write_u64(&MyProc->waitStart, 0); /* And awaken it */ - SetLatch(&proc->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(proc)); } /* @@ -1850,35 +1853,37 @@ CheckDeadLockAlert(void) got_deadlock_timeout = true; /* - * Have to set the latch again, even if handle_sig_alarm already did. Back - * then got_deadlock_timeout wasn't yet set... It's unlikely that this - * ever would be a problem, but setting a set latch again is cheap. + * Have to raise the interrupt again, even if handle_sig_alarm already + * did. Back then got_deadlock_timeout wasn't yet set... It's unlikely + * that this ever would be a problem, but raising an interrupt again is + * cheap. * * Note that, when this function runs inside procsignal_sigusr1_handler(), - * the handler function sets the latch again after the latch is set here. + * the handler function raises the interrupt again after the interrupt is + * raised here. */ - SetLatch(MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); errno = save_errno; } /* * ProcWaitForSignal - wait for a signal from another backend. * - * As this uses the generic process latch the caller has to be robust against + * As this uses the generic process interurpt the caller has to be robust against * unrelated wakeups: Always check that the desired state has occurred, and * wait again if not. */ void ProcWaitForSignal(uint32 wait_event_info) { - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - wait_event_info); - ResetLatch(MyLatch); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + wait_event_info); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); CHECK_FOR_INTERRUPTS(); } /* - * ProcSendSignal - set the latch of a backend identified by ProcNumber + * ProcSendSignal - send the interrupt of a backend identified by ProcNumber */ void ProcSendSignal(ProcNumber procNumber) @@ -1886,7 +1891,7 @@ ProcSendSignal(ProcNumber procNumber) if (procNumber < 0 || procNumber >= ProcGlobal->allProcCount) elog(ERROR, "procNumber out of range"); - SetLatch(&ProcGlobal->allProcs[procNumber].procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, procNumber); } /* diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.c index ab7137d0ff..cd5d1436ab 100644 --- a/src/backend/storage/sync/sync.c +++ b/src/backend/storage/sync/sync.c @@ -611,8 +611,8 @@ RegisterSyncRequest(const FileTag *ftag, SyncRequestType type, if (ret || (!ret && !retryOnError)) break; - WaitLatch(NULL, WL_EXIT_ON_PM_DEATH | WL_TIMEOUT, 10, - WAIT_EVENT_REGISTER_SYNC_REQUEST); + WaitInterrupt(0, WL_EXIT_ON_PM_DEATH | WL_TIMEOUT, 10, + WAIT_EVENT_REGISTER_SYNC_REQUEST); } return ret; diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c index 03a54451ac..cb5343d28c 100644 --- a/src/backend/utils/init/globals.c +++ b/src/backend/utils/init/globals.c @@ -52,15 +52,6 @@ bool MyCancelKeyValid = false; int32 MyCancelKey = 0; int MyPMChildSlot; -/* - * MyLatch points to the latch that should be used for signal handling by the - * current process. It will either point to a process local latch if the - * current process does not have a PGPROC entry in that moment, or to - * PGPROC->procLatch if it has. Thus it can always be used in signal handlers, - * without checking for its existence. - */ -struct Latch *MyLatch; - /* * DataDir is the absolute path to the top level of the PGDATA directory tree. * Except during early startup, this is also the server's working directory; diff --git a/src/backend/utils/init/miscinit.c b/src/backend/utils/init/miscinit.c index 537d92c0cf..7e0f0a87f0 100644 --- a/src/backend/utils/init/miscinit.c +++ b/src/backend/utils/init/miscinit.c @@ -41,8 +41,8 @@ #include "postmaster/postmaster.h" #include "replication/slotsync.h" #include "storage/fd.h" +#include "storage/interrupt.h" #include "storage/ipc.h" -#include "storage/latch.h" #include "storage/pg_shmem.h" #include "storage/pmsignal.h" #include "storage/proc.h" @@ -65,8 +65,6 @@ BackendType MyBackendType; /* List of lock files to be removed at proc exit */ static List *lock_files = NIL; -static Latch LocalLatchData; - /* ---------------------------------------------------------------- * ignoring system indexes support stuff * @@ -133,9 +131,8 @@ InitPostmasterChild(void) #endif /* Initialize process-local latch support */ - InitializeLatchSupport(); - InitProcessLocalLatch(); - InitializeLatchWaitSet(); + InitializeWaitEventSupport(); + InitializeInterruptWaitSet(); /* * If possible, make this process a group leader, so that the postmaster @@ -194,9 +191,8 @@ InitStandaloneProcess(const char *argv0) InitProcessGlobals(); /* Initialize process-local latch support */ - InitializeLatchSupport(); - InitProcessLocalLatch(); - InitializeLatchWaitSet(); + InitializeWaitEventSupport(); + InitializeInterruptWaitSet(); /* * For consistency with InitPostmasterChild, initialize signal mask here. @@ -217,48 +213,6 @@ InitStandaloneProcess(const char *argv0) get_pkglib_path(my_exec_path, pkglib_path); } -void -SwitchToSharedLatch(void) -{ - Assert(MyLatch == &LocalLatchData); - Assert(MyProc != NULL); - - MyLatch = &MyProc->procLatch; - - if (FeBeWaitSet) - ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetLatchPos, WL_LATCH_SET, - MyLatch); - - /* - * Set the shared latch as the local one might have been set. This - * shouldn't normally be necessary as code is supposed to check the - * condition before waiting for the latch, but a bit care can't hurt. - */ - SetLatch(MyLatch); -} - -void -InitProcessLocalLatch(void) -{ - MyLatch = &LocalLatchData; - InitLatch(MyLatch); -} - -void -SwitchBackToLocalLatch(void) -{ - Assert(MyLatch != &LocalLatchData); - Assert(MyProc != NULL && MyLatch == &MyProc->procLatch); - - MyLatch = &LocalLatchData; - - if (FeBeWaitSet) - ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetLatchPos, WL_LATCH_SET, - MyLatch); - - SetLatch(MyLatch); -} - const char * GetBackendTypeDesc(BackendType backendType) { diff --git a/src/include/access/parallel.h b/src/include/access/parallel.h index 69ffe5498f..7daec7ef83 100644 --- a/src/include/access/parallel.h +++ b/src/include/access/parallel.h @@ -14,6 +14,8 @@ #ifndef PARALLEL_H #define PARALLEL_H +#include + #include "access/xlogdefs.h" #include "lib/ilist.h" #include "postmaster/bgworker.h" diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.h index 81fcb9c232..94ef787618 100644 --- a/src/include/commands/waitlsn.h +++ b/src/include/commands/waitlsn.h @@ -15,7 +15,6 @@ #include "lib/pairingheap.h" #include "postgres.h" #include "port/atomics.h" -#include "storage/latch.h" #include "storage/procnumber.h" #include "storage/spin.h" #include "tcop/dest.h" diff --git a/src/include/libpq/libpq.h b/src/include/libpq/libpq.h index 142c98462e..fe89101be4 100644 --- a/src/include/libpq/libpq.h +++ b/src/include/libpq/libpq.h @@ -18,7 +18,7 @@ #include "lib/stringinfo.h" #include "libpq/libpq-be.h" -#include "storage/latch.h" +#include "storage/waiteventset.h" /* @@ -61,7 +61,7 @@ extern const PGDLLIMPORT PQcommMethods *PqCommMethods; extern PGDLLIMPORT WaitEventSet *FeBeWaitSet; #define FeBeWaitSetSocketPos 0 -#define FeBeWaitSetLatchPos 1 +#define FeBeWaitSetInterruptPos 1 #define FeBeWaitSetNEvents 3 extern int ListenServerPort(int family, const char *hostName, diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 25348e71eb..9797d60039 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -190,7 +190,6 @@ extern PGDLLIMPORT int MyProcPid; extern PGDLLIMPORT pg_time_t MyStartTime; extern PGDLLIMPORT TimestampTz MyStartTimestamp; extern PGDLLIMPORT struct Port *MyProcPort; -extern PGDLLIMPORT struct Latch *MyLatch; extern PGDLLIMPORT bool MyCancelKeyValid; extern PGDLLIMPORT int32 MyCancelKey; extern PGDLLIMPORT int MyPMChildSlot; @@ -317,9 +316,6 @@ extern PGDLLIMPORT char *DatabasePath; /* now in utils/init/miscinit.c */ extern void InitPostmasterChild(void); extern void InitStandaloneProcess(const char *argv0); -extern void InitProcessLocalLatch(void); -extern void SwitchToSharedLatch(void); -extern void SwitchBackToLocalLatch(void); /* * MyBackendType indicates what kind of a backend this is. diff --git a/src/include/storage/interrupt.h b/src/include/storage/interrupt.h new file mode 100644 index 0000000000..1dc85339a9 --- /dev/null +++ b/src/include/storage/interrupt.h @@ -0,0 +1,159 @@ +/*------------------------------------------------------------------------- + * + * interrupt.h + * Interrupt handling routines. + * + * "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, or asynchronously by timer signal handlers, other signal + * handlers or "sent" by other backends setting them directly. + * + * Most code currently deals with the INTERRUPT_GENERAL_WAKEUP interrupt. It + * is raised by any of the events checked by CHECK_FOR_INTERRUPTS), like query + * cancellation or idle session timeout. Well behaved backend code performs + * CHECK_FOR_INTERRUPTS() periodically in long computations, and should never + * sleep using mechanisms other than the WaitEventSet mechanism or the more + * convenient WaitInterrupt/WaitSockerOrInterrupt functions (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(). + * + * + * The correct pattern to wait for event(s) using INTERRUPT_GENERAL_WAKEUP is: + * + * for (;;) + * { + * ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); + * if (work to do) + * Do Stuff(); + * WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, ...); + * } + * + * It's important to reset the latch *before* checking if there's work to + * do. Otherwise, if someone sets the latch between the check and the + * ResetLatch call, you will miss it and Wait will incorrectly block. + * + * Another valid coding pattern looks like: + * + * for (;;) + * { + * if (work to do) + * Do Stuff(); // in particular, exit loop if some condition satisfied + * WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, ...); + * ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); + * } + * + * This is useful to reduce interrupt traffic if it's expected that the loop's + * termination condition will often be satisfied in the first iteration; + * the cost is an extra loop iteration before blocking when it is not. + * What must be avoided is placing any checks for asynchronous events after + * WaitInterrupt and before ClearInterrupt, as that creates a race condition. + * + * To wake up the waiter, you must first set a global flag or something else + * that the wait loop tests in the "if (work to do)" part, and call + * SendInterrupt(INTERRUPT_GENERAL_WAKEP) *after* that. SendInterrupt is + * designed to return quickly if the latch is already set. In more complex + * scenarios with nested loops that can consume different events, you can + * define your own INTERRUPT_* flag instead of relying on + * INTERRUPT_GENERAL_WAKEUP. + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/include/storage/interrupt.h + * + *------------------------------------------------------------------------- + */ + +#ifndef STORAGE_INTERRUPT_H +#define STORAGE_INTERRUPT_H + +#include "port/atomics.h" +#include "storage/procnumber.h" +#include "storage/waiteventset.h" + +#include + +extern PGDLLIMPORT pg_atomic_uint32 *MyPendingInterrupts; +extern PGDLLIMPORT pg_atomic_uint32 *MyMaybeSleepingOnInterrupts; + +typedef enum +{ + /* + * INTERRUPT_GENERAL_WAKEUP is multiplexed for many reasons, like query + * cancellation termination requests, recovery conflicts, and config + * reload requests. Upon receiving INTERRUPT_GENERAL_WAKEUP, you should + * call CHECK_FOR_INTERRUPTS() to process those requests. It is also used + * for various other context-dependent purposes, but note that if it's + * used to wake up for other reasons, you must still call + * CHECK_FOR_INTERRUPTS() once per iteration. + */ + INTERRUPT_GENERAL_WAKEUP, + + /* + * INTERRUPT_RECOVERY_WAKEUP is used to wake up startup process, to tell + * it that it should continue WAL replay. It's sent by WAL receiver when + * more WAL arrives, or when promotion is requested. + */ + INTERRUPT_RECOVERY_CONTINUE, +} InterruptType; + +/* + * Test an interrupt flag. + */ +static inline bool +InterruptIsPending(InterruptType reason) +{ + return (pg_atomic_read_u32(MyPendingInterrupts) & (1 << reason)) != 0; +} + +/* + * Test an interrupt flag. + */ +static inline bool +InterruptsPending(uint32 mask) +{ + return (pg_atomic_read_u32(MyPendingInterrupts) & (mask)) != 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 int WaitInterrupt(uint32 interruptMask, int wakeEvents, long timeout, + uint32 wait_event_info); +extern int WaitInterruptOrSocket(uint32 interruptMask, int wakeEvents, pgsocket sock, + long timeout, uint32 wait_event_info); +extern void SwitchToLocalInterrupts(void); +extern void SwitchToSharedInterrupts(void); +extern void InitializeInterruptWaitSet(void); + +#endif diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h index 7e194d536f..084d64a19b 100644 --- a/src/include/storage/latch.h +++ b/src/include/storage/latch.h @@ -1,94 +1,17 @@ /*------------------------------------------------------------------------- * * latch.h - * Routines for interprocess latches + * Backwards-compatibility macros for the old Latch interface * - * A latch is a boolean variable, with operations that let processes sleep - * until it is set. A latch can be set from another process, or a signal - * handler within the same process. - * - * The latch interface is a reliable replacement for the common pattern of - * using pg_usleep() or select() to wait until a signal arrives, where the - * signal handler sets a flag variable. Because on some platforms an - * incoming signal doesn't interrupt sleep, and even on platforms where it - * does there is a race condition if the signal arrives just before - * entering the sleep, the common pattern must periodically wake up and - * poll the flag variable. The pselect() system call was invented to solve - * this problem, but it is not portable enough. Latches are designed to - * overcome these limitations, allowing you to sleep without polling and - * ensuring quick response to signals from other processes. - * - * There are two kinds of latches: local and shared. A local latch is - * initialized by InitLatch, and can only be set from the same process. - * A local latch can be used to wait for a signal to arrive, by calling - * SetLatch in the signal handler. A shared latch resides in shared memory, - * and must be initialized at postmaster startup by InitSharedLatch. Before - * a shared latch can be waited on, it must be associated with a process - * with OwnLatch. Only the process owning the latch can wait on it, but any - * process can set it. - * - * There are three basic operations on a latch: - * - * SetLatch - Sets the latch - * ResetLatch - Clears the latch, allowing it to be set again - * WaitLatch - Waits for the latch to become set - * - * WaitLatch includes a provision for timeouts (which should be avoided - * when possible, as they incur extra overhead) and a provision for - * postmaster child processes to wake up immediately on postmaster death. - * See latch.c for detailed specifications for the exported functions. - * - * The correct pattern to wait for event(s) is: - * - * for (;;) - * { - * ResetLatch(); - * if (work to do) - * Do Stuff(); - * WaitLatch(); - * } - * - * It's important to reset the latch *before* checking if there's work to - * do. Otherwise, if someone sets the latch between the check and the - * ResetLatch call, you will miss it and Wait will incorrectly block. - * - * Another valid coding pattern looks like: - * - * for (;;) - * { - * if (work to do) - * Do Stuff(); // in particular, exit loop if some condition satisfied - * WaitLatch(); - * ResetLatch(); - * } - * - * This is useful to reduce latch traffic if it's expected that the loop's - * termination condition will often be satisfied in the first iteration; - * the cost is an extra loop iteration before blocking when it is not. - * What must be avoided is placing any checks for asynchronous events after - * WaitLatch and before ResetLatch, as that creates a race condition. - * - * To wake up the waiter, you must first set a global flag or something - * else that the wait loop tests in the "if (work to do)" part, and call - * SetLatch *after* that. SetLatch is designed to return quickly if the - * latch is already set. - * - * On some platforms, signals will not interrupt the latch wait primitive - * by themselves. Therefore, it is critical that any signal handler that - * is meant to terminate a WaitLatch wait calls SetLatch. - * - * Note that use of the process latch (PGPROC.procLatch) is generally better - * than an ad-hoc shared latch for signaling auxiliary processes. This is - * because generic signal handlers will call SetLatch on the process latch - * only, so using any latch other than the process latch effectively precludes - * use of any generic handler. - * - * - * WaitEventSets allow to wait for latches being set and additional events - - * postmaster dying and socket readiness of several sockets currently - at the - * same time. On many platforms using a long lived event set is more - * efficient than using WaitLatch or WaitLatchOrSocket. + * Latches were an inter-process signalling mechanism that was replaced + * in PostgreSQL verson 18 with Interrupts, see "storage/interrupt.h". + * This file contains macros that map old Latch calls to the new interface. + * The mapping is not perfect, it only covers the basic usage of setting + * and waiting for the current process's own latch (MyLatch), which is + * mapped to raising or waiting for INTERRUPT_GENERAL_WAKUP. * + * The WaitEventSet functions that used to be here are now in + * "storage/waitevents.h". * * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California @@ -97,100 +20,60 @@ * *------------------------------------------------------------------------- */ + #ifndef LATCH_H #define LATCH_H -#include +#include "storage/interrupt.h" -#include "utils/resowner.h" +#define WL_LATCH_SET WL_INTERRUPT /* - * Latch structure should be treated as opaque and only accessed through - * the public functions. It is defined here to allow embedding Latches as - * part of bigger structs. + * These compatibility macros only support operating on MyLatch. It used to + * be a pointer to a Latch struct, but now it's just a dummy int variable. */ -typedef struct Latch + +#define MyLatch ((void *) 0xAAAA) + +static inline void +SetLatch(void *dummy) { - sig_atomic_t is_set; - sig_atomic_t maybe_sleeping; - bool is_shared; - int owner_pid; -#ifdef WIN32 - HANDLE event; -#endif -} Latch; + Assert(dummy == MyLatch); + RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP); +} -/* - * Bitmasks for events that may wake-up WaitLatch(), WaitLatchOrSocket(), or - * WaitEventSetWait(). - */ -#define WL_LATCH_SET (1 << 0) -#define WL_SOCKET_READABLE (1 << 1) -#define WL_SOCKET_WRITEABLE (1 << 2) -#define WL_TIMEOUT (1 << 3) /* not for WaitEventSetWait() */ -#define WL_POSTMASTER_DEATH (1 << 4) -#define WL_EXIT_ON_PM_DEATH (1 << 5) -#ifdef WIN32 -#define WL_SOCKET_CONNECTED (1 << 6) -#else -/* avoid having to deal with case on platforms not requiring it */ -#define WL_SOCKET_CONNECTED WL_SOCKET_WRITEABLE -#endif -#define WL_SOCKET_CLOSED (1 << 7) -#ifdef WIN32 -#define WL_SOCKET_ACCEPT (1 << 8) -#else -/* avoid having to deal with case on platforms not requiring it */ -#define WL_SOCKET_ACCEPT WL_SOCKET_READABLE -#endif -#define WL_SOCKET_MASK (WL_SOCKET_READABLE | \ - WL_SOCKET_WRITEABLE | \ - WL_SOCKET_CONNECTED | \ - WL_SOCKET_ACCEPT | \ - WL_SOCKET_CLOSED) +static inline void +ResetLatch(void *dummy) +{ + Assert(dummy == MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); +} -typedef struct WaitEvent +static inline int +WaitLatch(void *dummy, int wakeEvents, long timeout, uint32 wait_event_info) { - int pos; /* position in the event data structure */ - uint32 events; /* triggered events */ - pgsocket fd; /* socket fd associated with event */ - void *user_data; /* pointer provided in AddWaitEventToSet */ -#ifdef WIN32 - bool reset; /* Is reset of the event required? */ -#endif -} WaitEvent; + uint32 interrupt_mask = 0; -/* forward declaration to avoid exposing latch.c implementation details */ -typedef struct WaitEventSet WaitEventSet; + Assert(dummy == MyLatch || dummy == NULL); -/* - * prototypes for functions in latch.c - */ -extern void InitializeLatchSupport(void); -extern void InitLatch(Latch *latch); -extern void InitSharedLatch(Latch *latch); -extern void OwnLatch(Latch *latch); -extern void DisownLatch(Latch *latch); -extern void SetLatch(Latch *latch); -extern void ResetLatch(Latch *latch); -extern void ShutdownLatchSupport(void); + if ((wakeEvents & WL_LATCH_SET) && dummy == MyLatch) + interrupt_mask = 1 << INTERRUPT_GENERAL_WAKEUP; + return WaitInterrupt(interrupt_mask, wakeEvents, + timeout, wait_event_info); +} -extern WaitEventSet *CreateWaitEventSet(ResourceOwner resowner, int nevents); -extern void FreeWaitEventSet(WaitEventSet *set); -extern void FreeWaitEventSetAfterFork(WaitEventSet *set); -extern int AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, - Latch *latch, void *user_data); -extern void ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, Latch *latch); +static inline int +WaitLatchOrSocket(void *dummy, int wakeEvents, pgsocket sock, long timeout, + uint32 wait_event_info) +{ + uint32 interrupt_mask = 0; + + Assert(dummy == MyLatch || dummy == NULL); -extern int WaitEventSetWait(WaitEventSet *set, long timeout, - WaitEvent *occurred_events, int nevents, - uint32 wait_event_info); -extern int WaitLatch(Latch *latch, int wakeEvents, long timeout, - uint32 wait_event_info); -extern int WaitLatchOrSocket(Latch *latch, int wakeEvents, - pgsocket sock, long timeout, uint32 wait_event_info); -extern void InitializeLatchWaitSet(void); -extern int GetNumRegisteredWaitEvents(WaitEventSet *set); -extern bool WaitEventSetCanReportClosed(void); + if ((wakeEvents & WL_LATCH_SET) && dummy == MyLatch) + interrupt_mask = 1 << INTERRUPT_GENERAL_WAKEUP; + return WaitInterruptOrSocket(interrupt_mask, wakeEvents, + sock, timeout, wait_event_info); +} -#endif /* LATCH_H */ +#endif diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index 95710e416f..15b3c7d0e2 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -167,9 +167,6 @@ struct PGPROC PGSemaphore sem; /* ONE semaphore to sleep on */ ProcWaitStatus waitStatus; - Latch procLatch; /* generic latch for process */ - - TransactionId xid; /* id of top-level transaction currently being * executed by this proc, if running and XID * is assigned; else InvalidTransactionId. @@ -305,6 +302,14 @@ 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 pendingInterrupts; + pg_atomic_uint32 maybeSleepingOnInterrupts; + +#ifdef WIN32 + /* Event handle to wake up the process when sending an interrupt */ + HANDLE interruptWakeupEvent; +#endif }; /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */ @@ -420,6 +425,24 @@ typedef struct PROC_HDR */ ProcNumber walwriterProc; ProcNumber checkpointerProc; + ProcNumber walreceiverProc; + + /* + * recoveryWakeupLatch is used to wake up the startup process to continue + * WAL replay, if it is waiting for WAL to arrive or promotion to be + * requested. + * + * Note that the startup process also uses another latch, its procLatch, + * to wait for recovery conflict. If we get rid of recoveryWakeupLatch for + * signaling the startup process in favor of using its procLatch, which + * comports better with possible generic signal handlers using that latch. + * But we should not do that because the startup process doesn't assume + * that it's waken up by walreceiver process or SIGHUP signal handler + * while it's waiting for recovery conflict. The separate latches, + * recoveryWakeupLatch and procLatch, should be used for inter-process + * communication for WAL replay and recovery conflict, respectively. XXX + */ + ProcNumber startupProc; /* Current shared estimate of appropriate spins_per_delay value */ int spins_per_delay; diff --git a/src/include/storage/waiteventset.h b/src/include/storage/waiteventset.h new file mode 100644 index 0000000000..c7780b7beb --- /dev/null +++ b/src/include/storage/waiteventset.h @@ -0,0 +1,119 @@ +/*------------------------------------------------------------------------- + * + * waiteventset.h + * ppoll() / pselect() like interface for waiting for events + * + * This is a reliable replacement for the common pattern of using pg_usleep() + * or select() to wait until a signal arrives, where the signal handler raises + * an interrupt (see storage/interrupt.h). Because on some platforms an + * incoming signal doesn't interrupt sleep, and even on platforms where it + * does there is a race condition if the signal arrives just before entering + * the sleep, the common pattern must periodically wake up and poll the flag + * variable. The pselect() system call was invented to solve this problem, but + * it is not portable enough. WaitEventSets and Interrupts are designed to + * overcome these limitations, allowing you to sleep without polling and + * ensuring quick response to signals from other processes. + * + * WaitInterrupt includes a provision for timeouts (which should be avoided + * when possible, as they incur extra overhead) and a provision for postmaster + * child processes to wake up immediately on postmaster death. See + * interrupt.c for detailed specifications for the exported functions. + * + * On some platforms, signals will not interrupt the wait primitive by + * themselves. Therefore, it is critical that any signal handler that is + * meant to terminate a WaitInterrupt wait calls RaiseInterrupt. + * + * WaitEventSets allow to wait for interrupts being set and additional events - + * postmaster dying and socket readiness of several sockets currently - at the + * same time. On many platforms using a long lived event set is more + * efficient than using WaitInterrupt or WaitInterruptOrSocket. + * + * + * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/storage/waiteventset.h + * + *------------------------------------------------------------------------- + */ +#ifndef WAITEVENTSET_H +#define WAITEVENTSET_H + +#include + +#include "storage/procnumber.h" +#include "utils/resowner.h" + +/* + * Bitmasks for events that may wake-up WaitLatch(), WaitLatchOrSocket(), or + * WaitEventSetWait(). + */ +#define WL_INTERRUPT (1 << 0) +#define WL_SOCKET_READABLE (1 << 1) +#define WL_SOCKET_WRITEABLE (1 << 2) +#define WL_TIMEOUT (1 << 3) /* not for WaitEventSetWait() */ +#define WL_POSTMASTER_DEATH (1 << 4) +#define WL_EXIT_ON_PM_DEATH (1 << 5) +#ifdef WIN32 +#define WL_SOCKET_CONNECTED (1 << 6) +#else +/* avoid having to deal with case on platforms not requiring it */ +#define WL_SOCKET_CONNECTED WL_SOCKET_WRITEABLE +#endif +#define WL_SOCKET_CLOSED (1 << 7) +#ifdef WIN32 +#define WL_SOCKET_ACCEPT (1 << 8) +#else +/* avoid having to deal with case on platforms not requiring it */ +#define WL_SOCKET_ACCEPT WL_SOCKET_READABLE +#endif +#define WL_SOCKET_MASK (WL_SOCKET_READABLE | \ + WL_SOCKET_WRITEABLE | \ + WL_SOCKET_CONNECTED | \ + WL_SOCKET_ACCEPT | \ + WL_SOCKET_CLOSED) + +typedef struct WaitEvent +{ + int pos; /* position in the event data structure */ + uint32 events; /* triggered events */ + pgsocket fd; /* socket fd associated with event */ + void *user_data; /* pointer provided in AddWaitEventToSet */ +#ifdef WIN32 + bool reset; /* Is reset of the event required? */ +#endif +} WaitEvent; + +/* forward declaration to avoid exposing latch.c implementation details */ +typedef struct WaitEventSet WaitEventSet; + +struct PGPROC; + +/* + * prototypes for functions in waiteventset.c + */ +extern void InitializeWaitEventSupport(void); +extern void ShutdownWaitEventSupport(void); +#ifdef WIN32 +extern HANDLE CreateSharedWakeupHandle(void); +#endif + +extern WaitEventSet *CreateWaitEventSet(ResourceOwner resowner, int nevents); +extern void FreeWaitEventSet(WaitEventSet *set); +extern void FreeWaitEventSetAfterFork(WaitEventSet *set); +extern int AddWaitEventToSet(WaitEventSet *set, uint32 events, pgsocket fd, + uint32 interruptMask, void *user_data); +extern void ModifyWaitEvent(WaitEventSet *set, int pos, uint32 events, uint32 interruptMask); + +extern int WaitEventSetWait(WaitEventSet *set, long timeout, + WaitEvent *occurred_events, int nevents, + uint32 wait_event_info); +extern void InitializeInterruptWaitSet(void); +extern int GetNumRegisteredWaitEvents(WaitEventSet *set); +extern bool WaitEventSetCanReportClosed(void); + +/* low level functions used to implement SendInterrupt */ +extern void WakeupMyProc(void); +extern void WakeupOtherProc(struct PGPROC *proc); + +#endif /* WAITEVENTSET_H */ diff --git a/src/test/modules/test_shm_mq/setup.c b/src/test/modules/test_shm_mq/setup.c index b3dac44d97..fe62a4a428 100644 --- a/src/test/modules/test_shm_mq/setup.c +++ b/src/test/modules/test_shm_mq/setup.c @@ -18,6 +18,7 @@ #include "miscadmin.h" #include "pgstat.h" #include "postmaster/bgworker.h" +#include "storage/interrupt.h" #include "storage/procsignal.h" #include "storage/shm_toc.h" #include "test_shm_mq.h" @@ -286,11 +287,11 @@ wait_for_workers_to_become_ready(worker_state *wstate, we_bgworker_startup = WaitEventExtensionNew("TestShmMqBgWorkerStartup"); /* Wait to be signaled. */ - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - we_bgworker_startup); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + we_bgworker_startup); /* Reset the latch so we don't spin. */ - ResetLatch(MyLatch); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); /* An interrupt may have occurred while we were waiting. */ CHECK_FOR_INTERRUPTS(); diff --git a/src/test/modules/test_shm_mq/test.c b/src/test/modules/test_shm_mq/test.c index 3d235568b8..212e562d2d 100644 --- a/src/test/modules/test_shm_mq/test.c +++ b/src/test/modules/test_shm_mq/test.c @@ -16,6 +16,7 @@ #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" +#include "storage/interrupt.h" #include "varatt.h" #include "test_shm_mq.h" @@ -238,9 +239,9 @@ test_shm_mq_pipelined(PG_FUNCTION_ARGS) * have read or written data and therefore there may now be work * for us to do. */ - (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0, - we_message_queue); - ResetLatch(MyLatch); + (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0, + we_message_queue); + ClearInterrupt(INTERRUPT_GENERAL_WAKEUP); CHECK_FOR_INTERRUPTS(); } } diff --git a/src/test/modules/test_shm_mq/worker.c b/src/test/modules/test_shm_mq/worker.c index 6c4fbc7827..6fe9f9e4eb 100644 --- a/src/test/modules/test_shm_mq/worker.c +++ b/src/test/modules/test_shm_mq/worker.c @@ -20,6 +20,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/interrupt.h" #include "storage/ipc.h" #include "storage/procarray.h" #include "storage/shm_mq.h" @@ -128,7 +129,7 @@ test_shm_mq_main(Datum main_arg) elog(DEBUG1, "registrant backend has exited prematurely"); proc_exit(1); } - SetLatch(®istrant->procLatch); + SendInterrupt(INTERRUPT_GENERAL_WAKEUP, GetNumberFromPGProc(registrant)); /* Do the work. */ copy_messages(inqh, outqh); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 9e951a9e6f..2ae4be606f 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1252,6 +1252,7 @@ Integer IntegerSet InternalDefaultACL InternalGrant +InterruptType Interval IntervalAggState IntoClause @@ -1489,7 +1490,6 @@ LZ4State LabelProvider LagTracker LargeObjectDesc -Latch LauncherLastStartTimesEntry LerpFunc LexDescr -- 2.39.2