public inbox for [email protected]
help / color / mirror / Atom feedFrom: Heikki Linnakangas <[email protected]>
To: Thomas Munro <[email protected]>
To: Fujii Masao <[email protected]>
Cc: pgsql-hackers <[email protected]>
Cc: Andres Freund <[email protected]>
Cc: Robert Haas <[email protected]>
Subject: Re: Interrupts vs signals
Date: Sat, 31 Aug 2024 01:17:40 +0300
Message-ID: <[email protected]> (raw)
In-Reply-To: <CA+hUKG+6rn1FNxvO7Wuz_aAQ4MnGW5G6+UW4nBnzwnGnuBVWyw@mail.gmail.com>
References: <CA+hUKG+3MkS21yK4jL4cgZywdnnGKiBg0jatoV6kzaniBmcqbQ@mail.gmail.com>
<[email protected]>
<CA+hUKGLseL7e7EFMZZECzCijjov76PcXEGYy2swXa8jfnuV=2A@mail.gmail.com>
<CA+TgmobVnFHGww-7z6vC1Ht63Pj-Kus4dFYPOp2zgMY2M6_iDw@mail.gmail.com>
<[email protected]>
<CA+TgmobfJHdTjh8sDGct_bP0wfB0J6Csz7OHOEbiEu62ZD5jGA@mail.gmail.com>
<CA+hUKGJACzP++3PifHTMgG2xgTD4irap6+CXBMogKKiD43VqoA@mail.gmail.com>
<CA+hUKGLAZCxzyhyoYz0BmGkqkmj-38ZfToF1GqNYhU+Br+XCLQ@mail.gmail.com>
<[email protected]>
<CA+hUKGJvt8wg_KfjFLvr27LT6sYK_szngkchig6Dg_CnNGPskw@mail.gmail.com>
<[email protected]>
<[email protected]>
<CA+hUKG+6rn1FNxvO7Wuz_aAQ4MnGW5G6+UW4nBnzwnGnuBVWyw@mail.gmail.com>
On 26/08/2024 02:05, Thomas Munro wrote:
> On Sun, Aug 25, 2024 at 5:17 AM Heikki Linnakangas <[email protected]> wrote:
>> On 07/08/2024 17:59, Heikki Linnakangas wrote:
>>> I'm also wondering about the relationship between interrupts and
>>> latches. Currently, SendInterrupt sets a latch to wake up the target
>>> process. I wonder if it should be the other way 'round? Move all the
>>> wakeup code, with the signalfd, the self-pipe etc to interrupt.c, and in
>>> SetLatch, call SendInterrupt to wake up the target process? Somehow that
>>> feels more natural to me, I think.
>>
>> I explored that a little, see attached patch set. It's going towards the
>> same end state as your patches, I think, but it starts from different
>> angle. In a nutshell:
>>
>> Remove Latch as an abstraction, and replace all use of Latches with
>> Interrupts. When I originally created the Latch abstraction, I imagined
>> that we would have different latches for different purposes, but in
>> reality, almost all code just used the general-purpose "process latch".
>> this patch accepts that reality and replaces the Latch struct directly
>> with the interrupt mask in PGPROC.
>
> Some very initial reactions:
>
> * I like it!
Ok, here's a new version, with a bunch of bugs and FIXMEs fixed,
comments and other polish. A few things remain that I'm not sure about:
- Terminology. We now have storage/interrupt.h and
postmaster/interrupt.h. They're not the same thing, but there's also
some overlap, especially after your patch set is applied.
Aside from filenames, "interrupt" now means at least two things: a
wakeup that you can send to a backend with SendInterrupt, and the thing
that you check with CHECK_FOR_INTERRUPTS(). They're related, but
different. To show case that confusion, the patch now contains this gem,
which was a result of mechanically replacing "latch" with "interrupt":
* We process interrupts whenever the interrupt has been set, so
* cancel/die interrupts are processed quickly.
- Fujii: this replaces the "recoveryWakeupLatch" with a separate
interrupt type. There was an earlier attempt at replacing
recoveryWakeupLatch with the process's regular latch which was reverted,
commits ac22929a26, and 00f690a239. I think this solution doesn't suffer
from the same problems as that earlier attempt, but if you have a chance
to review this, I would appreciate that. In a nutshell,
INTERRUPT_RECOVERY_CONTINUE interrupt is now used instead of
recoveryWakeupLatch, but the places that previously waited just on
recoveryWakeupLatch, now wait on both INTERRUPT_GENERAL_WAKEUP, which is
equivalent to waiting on the process latch, and
INTERRUPT_RECOVERY_CONTINUE. So those loops should now react more
quickly to signals like SIGHUP.
- Backwards-compatibility and extensions. This will break any extensions
that use Latches. Extensions that just used WaitLatch(MyLatch) or
similar are easy to convert to use latches instead. Or we could keep
around some backwards-compatibility macros like 0008 here does, to avoid
the code churn. However, if an extension is creating its own latches or
doing more complicated stuff with them, it gets harder to maintain
source-code compatibility for them.
Aside from the backwards-compatibility aspect, should we reserve a few
INTERRUPT_* values for extensions?
> * This direction seems to fit quite nicely with future ideas about
> asynchronous network I/O. That may sound unrelated, but imagine that
> a future version of WaitEventSet is built on Linux io_uring (or
> Windows iorings, or Windows IOCP, or kqueue), and waits for the kernel
> to tell you that network data has been transferred directly into a
> user space buffer. You could wait for the interrupt word to change at
> the same time by treating it as a futex[1]. Then all that other stuff
> -- signalfd, is_set, maybe_sleeping -- just goes away, and all we have
> left is one single word in memory. (That it is possible to do that is
> not really a coincidence, as our own Mr Freund asked Mr Axboe to add
> it[2]. The existing latch implementation techniques could be used as
> fallbacks, but when looked at from the right angle, once you squish
> all the wakeup reasons into a single word, it's all just an
> implementation of a multiplexable futex with extra steps.)
Cool
> * Speaking of other problems in other threads that might be solved by
> this redesign, I think I see the outline of some solutions to the
> problem of different classes of wakeup which you can handle at
> different times, using masks. There is a tension in a few places
> where we want to handle some kind of interrupts but not others in
> localised wait points, which we sort of try to address by holding
> interrupts or holding cancel interrupts, but it's not satisfying and
> there are some places where it doesn't work well. Needs a lot more
> thought, but a basic step would be: after old_interrupt_vector =
> pg_atomic_fetch_or_u32(interrupt_vector, new_bits), if
> (old_interrupt_vector & new_bits) == new_bits, then you didn't
> actually change any bits, so you probably don't really need to wake
> the other backend. If someone is currently unable to handle that type
> of interrupt (has ignored, ie not cleared, those bits) or is already
> in the process of handling it (is currently being rescheduled but
> hasn't cleared those bits yet), then you don't bother to wake it up.
> Concretely, it could mean that we avoid some of the useless wakeup
> storm problems we see in vacuum delays or while executing a query and
> not in a good place to handle sinval wakeups, etc. These are just
> some raw thoughts, I am not sure about the bigger picture of that
> topic yet.
Yeah, I expect this work to help with those issues, but also not sure of
the details yet.
--
Heikki Linnakangas
Neon (https://neon.tech)
Attachments:
[text/x-patch] v2-0001-Remove-unused-latch.patch (2.0K, ../[email protected]/2-v2-0001-Remove-unused-latch.patch)
download | inline diff:
From a18421815183f66495a91e7344bf606f66dfbb0d Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Sat, 24 Aug 2024 01:04:21 +0300
Subject: [PATCH v2 01/12] Remove unused latch
It was left unused by commit bc971f4025, which replaced the latch
usage with a condition variable
---
src/backend/replication/walsender.c | 3 ---
src/include/replication/walsender_private.h | 7 -------
2 files changed, 10 deletions(-)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c5f1009f37..866b69ec85 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -2935,7 +2935,6 @@ InitWalSenderSlot(void)
walsnd->flushLag = -1;
walsnd->applyLag = -1;
walsnd->sync_standby_priority = 0;
- walsnd->latch = &MyProc->procLatch;
walsnd->replyTime = 0;
/*
@@ -2979,8 +2978,6 @@ WalSndKill(int code, Datum arg)
MyWalSnd = NULL;
SpinLockAcquire(&walsnd->mutex);
- /* clear latch while holding the spinlock, so it can safely be read */
- walsnd->latch = NULL;
/* Mark WalSnd struct as no longer being in use. */
walsnd->pid = 0;
SpinLockRelease(&walsnd->mutex);
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index cf32ac2488..41ac736b95 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -18,7 +18,6 @@
#include "nodes/replnodes.h"
#include "replication/syncrep.h"
#include "storage/condition_variable.h"
-#include "storage/latch.h"
#include "storage/shmem.h"
#include "storage/spin.h"
@@ -71,12 +70,6 @@ typedef struct WalSnd
/* Protects shared variables in this structure. */
slock_t mutex;
- /*
- * Pointer to the walsender's latch. Used by backends to wake up this
- * walsender when it has work to do. NULL if the walsender isn't active.
- */
- Latch *latch;
-
/*
* Timestamp of the last message received from standby.
*/
--
2.39.2
[text/x-patch] v2-0002-Remove-unneeded-include.patch (772B, ../[email protected]/3-v2-0002-Remove-unneeded-include.patch)
download | inline diff:
From 2a028fe4759163cdbc561cc83c95d8214bfd0522 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Sat, 24 Aug 2024 00:26:46 +0300
Subject: [PATCH v2 02/12] Remove unneeded #include
Unneeded since commit d72731a704.
---
src/include/storage/buf_internals.h | 1 -
1 file changed, 1 deletion(-)
diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h
index f190e6e5e4..eda6c69921 100644
--- a/src/include/storage/buf_internals.h
+++ b/src/include/storage/buf_internals.h
@@ -20,7 +20,6 @@
#include "storage/buf.h"
#include "storage/bufmgr.h"
#include "storage/condition_variable.h"
-#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/shmem.h"
#include "storage/smgr.h"
--
2.39.2
[text/x-patch] v2-0003-Rename-SetWalSummarizerLatch-function.patch (2.3K, ../[email protected]/4-v2-0003-Rename-SetWalSummarizerLatch-function.patch)
download | inline diff:
From 8eae424a9a76e14ddbc22ac6abb268ea2e3b6ce7 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Sat, 24 Aug 2024 00:21:15 +0300
Subject: [PATCH v2 03/12] Rename SetWalSummarizerLatch function
The fact that it uses a latch for the wakeup is an implementation detail
---
src/backend/access/transam/xlog.c | 2 +-
src/backend/postmaster/walsummarizer.c | 4 ++--
src/include/postmaster/walsummarizer.h | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index ee0fb0e28f..45a4a40eca 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -7268,7 +7268,7 @@ CreateCheckPoint(int flags)
* until after the above call that flushes the XLOG_CHECKPOINT_ONLINE
* record.
*/
- SetWalSummarizerLatch();
+ WakeupWalSummarizer();
/*
* Let smgr do post-checkpoint cleanup (eg, deleting old files).
diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c
index c1bf4a70dd..3a407f8daa 100644
--- a/src/backend/postmaster/walsummarizer.c
+++ b/src/backend/postmaster/walsummarizer.c
@@ -626,7 +626,7 @@ GetOldestUnsummarizedLSN(TimeLineID *tli, bool *lsn_is_exact)
}
/*
- * Attempt to set the WAL summarizer's latch.
+ * Wake up the WAL summarizer process.
*
* This might not work, because there's no guarantee that the WAL summarizer
* process was successfully started, and it also might have started but
@@ -634,7 +634,7 @@ GetOldestUnsummarizedLSN(TimeLineID *tli, bool *lsn_is_exact)
* latch set, but there's no guarantee.
*/
void
-SetWalSummarizerLatch(void)
+WakeupWalSummarizer(void)
{
ProcNumber pgprocno;
diff --git a/src/include/postmaster/walsummarizer.h b/src/include/postmaster/walsummarizer.h
index aedca55676..2642aa701d 100644
--- a/src/include/postmaster/walsummarizer.h
+++ b/src/include/postmaster/walsummarizer.h
@@ -29,7 +29,7 @@ extern void GetWalSummarizerState(TimeLineID *summarized_tli,
int *summarizer_pid);
extern XLogRecPtr GetOldestUnsummarizedLSN(TimeLineID *tli,
bool *lsn_is_exact);
-extern void SetWalSummarizerLatch(void);
+extern void WakeupWalSummarizer(void);
extern void WaitForWalSummarization(XLogRecPtr lsn);
#endif
--
2.39.2
[text/x-patch] v2-0004-Address-walwriter-and-checkpointer-by-proc-number.patch (4.5K, ../[email protected]/5-v2-0004-Address-walwriter-and-checkpointer-by-proc-number.patch)
download | inline diff:
From 22854917df3ce62026626aaeca43c8146c09f3b9 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Fri, 23 Aug 2024 17:25:15 +0300
Subject: [PATCH v2 04/12] Address walwriter and checkpointer by proc number
instead of pointing directly to their latch
---
src/backend/access/transam/xlog.c | 11 +++++++++--
src/backend/postmaster/checkpointer.c | 17 ++++++++++++-----
src/backend/postmaster/walwriter.c | 6 +++---
src/backend/storage/lmgr/proc.c | 4 ++--
src/include/storage/proc.h | 12 ++++++++----
5 files changed, 34 insertions(+), 16 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 45a4a40eca..63a8ab5a29 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -2669,8 +2669,15 @@ XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN)
wakeup = true;
}
- if (wakeup && ProcGlobal->walwriterLatch)
- SetLatch(ProcGlobal->walwriterLatch);
+ if (wakeup)
+ {
+ volatile PROC_HDR *procglobal = ProcGlobal;
+ ProcNumber walwriterProc;
+
+ walwriterProc = procglobal->walwriterProc;
+ if (walwriterProc != INVALID_PROC_NUMBER)
+ SetLatch(&GetPGProcByNumber(walwriterProc)->procLatch);
+ }
}
/*
diff --git a/src/backend/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c
index 199f008bcd..e556d7ecee 100644
--- a/src/backend/postmaster/checkpointer.c
+++ b/src/backend/postmaster/checkpointer.c
@@ -324,10 +324,10 @@ CheckpointerMain(char *startup_data, size_t startup_data_len)
UpdateSharedMemoryConfig();
/*
- * Advertise our latch that backends can use to wake us up while we're
- * sleeping.
+ * Advertise our proc number that backends can use to wake us up while
+ * we're sleeping.
*/
- ProcGlobal->checkpointerLatch = &MyProc->procLatch;
+ ProcGlobal->checkpointerProc = MyProcNumber;
/*
* Loop forever
@@ -1128,8 +1128,15 @@ ForwardSyncRequest(const FileTag *ftag, SyncRequestType type)
LWLockRelease(CheckpointerCommLock);
/* ... but not till after we release the lock */
- if (too_full && ProcGlobal->checkpointerLatch)
- SetLatch(ProcGlobal->checkpointerLatch);
+ if (too_full)
+ {
+ volatile PROC_HDR *procglobal = ProcGlobal;
+ ProcNumber checkpointerProc;
+
+ checkpointerProc = procglobal->checkpointerProc;
+ if (checkpointerProc != INVALID_PROC_NUMBER)
+ SetLatch(&GetPGProcByNumber(checkpointerProc)->procLatch);
+ }
return true;
}
diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c
index 6e7918a78d..0c55d9fa59 100644
--- a/src/backend/postmaster/walwriter.c
+++ b/src/backend/postmaster/walwriter.c
@@ -210,10 +210,10 @@ WalWriterMain(char *startup_data, size_t startup_data_len)
SetWalWriterSleeping(false);
/*
- * Advertise our latch that backends can use to wake us up while we're
- * sleeping.
+ * Advertise our proc number that backends can use to wake us up while
+ * we're sleeping.
*/
- ProcGlobal->walwriterLatch = &MyProc->procLatch;
+ ProcGlobal->walwriterProc = MyProcNumber;
/*
* Loop forever
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index ac66da8638..56c60704f5 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -177,8 +177,8 @@ InitProcGlobal(void)
dlist_init(&ProcGlobal->bgworkerFreeProcs);
dlist_init(&ProcGlobal->walsenderFreeProcs);
ProcGlobal->startupBufferPinWaitBufId = -1;
- ProcGlobal->walwriterLatch = NULL;
- ProcGlobal->checkpointerLatch = NULL;
+ ProcGlobal->walwriterProc = INVALID_PROC_NUMBER;
+ ProcGlobal->checkpointerProc = INVALID_PROC_NUMBER;
pg_atomic_init_u32(&ProcGlobal->procArrayGroupFirst, INVALID_PROC_NUMBER);
pg_atomic_init_u32(&ProcGlobal->clogGroupFirst, INVALID_PROC_NUMBER);
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index deeb06c9e0..95710e416f 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -413,10 +413,14 @@ typedef struct PROC_HDR
pg_atomic_uint32 procArrayGroupFirst;
/* First pgproc waiting for group transaction status update */
pg_atomic_uint32 clogGroupFirst;
- /* WALWriter process's latch */
- Latch *walwriterLatch;
- /* Checkpointer process's latch */
- Latch *checkpointerLatch;
+
+ /*
+ * Current slot numbers of some auxiliary processes. There can be only one
+ * of each of these running at a time.
+ */
+ ProcNumber walwriterProc;
+ ProcNumber checkpointerProc;
+
/* Current shared estimate of appropriate spins_per_delay value */
int spins_per_delay;
/* Buffer id of the buffer that Startup process waits for pin on, or -1 */
--
2.39.2
[text/x-patch] v2-0005-Address-walreceiver-by-procno-not-direct-pointer-.patch (6.3K, ../[email protected]/6-v2-0005-Address-walreceiver-by-procno-not-direct-pointer-.patch)
download | inline diff:
From afff68be45932c0340fda79ff1f7a154130beaf5 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Fri, 23 Aug 2024 23:54:53 +0300
Subject: [PATCH v2 05/12] Address walreceiver by procno, not direct pointer to
its latch
---
src/backend/access/transam/xlogfuncs.c | 1 +
.../libpqwalreceiver/libpqwalreceiver.c | 1 +
src/backend/replication/walreceiver.c | 17 ++++++-------
src/backend/replication/walreceiverfuncs.c | 12 ++++++----
src/include/replication/walreceiver.h | 24 +++++++++----------
5 files changed, 29 insertions(+), 26 deletions(-)
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index 4e46baaebd..b0c6d7c687 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -28,6 +28,7 @@
#include "pgstat.h"
#include "replication/walreceiver.h"
#include "storage/fd.h"
+#include "storage/latch.h"
#include "storage/standby.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 97f957cd87..c74369953f 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -30,6 +30,7 @@
#include "pgstat.h"
#include "pqexpbuffer.h"
#include "replication/walreceiver.h"
+#include "storage/latch.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index a27aee63de..d1d99de980 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -266,8 +266,8 @@ WalReceiverMain(char *startup_data, size_t startup_data_len)
walrcv->lastMsgSendTime =
walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
- /* Report the latch to use to awaken this process */
- walrcv->latch = &MyProc->procLatch;
+ /* Report our PGPROC entry to use to awaken this process */
+ walrcv->procno = MyProcNumber;
SpinLockRelease(&walrcv->mutex);
@@ -819,8 +819,8 @@ WalRcvDie(int code, Datum arg)
Assert(walrcv->pid == MyProcPid);
walrcv->walRcvState = WALRCV_STOPPED;
walrcv->pid = 0;
+ walrcv->procno = INVALID_PROC_NUMBER;
walrcv->ready_to_display = false;
- walrcv->latch = NULL;
SpinLockRelease(&walrcv->mutex);
ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
@@ -1358,15 +1358,16 @@ WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now)
void
WalRcvForceReply(void)
{
- Latch *latch;
+ ProcNumber procno;
WalRcv->force_reply = true;
- /* fetching the latch pointer might not be atomic, so use spinlock */
+
+ /* fetching the proc number is probably atomic, but don't rely on it */
SpinLockAcquire(&WalRcv->mutex);
- latch = WalRcv->latch;
+ procno = WalRcv->procno;
SpinLockRelease(&WalRcv->mutex);
- if (latch)
- SetLatch(latch);
+ if (procno != INVALID_PROC_NUMBER)
+ SetLatch(&GetPGProcByNumber(procno)->procLatch);
}
/*
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index 85a19cdfa5..b1780d0106 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -26,7 +26,9 @@
#include "access/xlogrecovery.h"
#include "pgstat.h"
#include "replication/walreceiver.h"
+#include "storage/latch.h"
#include "storage/pmsignal.h"
+#include "storage/proc.h"
#include "storage/shmem.h"
#include "utils/timestamp.h"
@@ -66,7 +68,7 @@ WalRcvShmemInit(void)
ConditionVariableInit(&WalRcv->walRcvStoppedCV);
SpinLockInit(&WalRcv->mutex);
pg_atomic_init_u64(&WalRcv->writtenUpto, 0);
- WalRcv->latch = NULL;
+ WalRcv->procno = INVALID_PROC_NUMBER;
}
}
@@ -248,7 +250,7 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo,
WalRcvData *walrcv = WalRcv;
bool launch = false;
pg_time_t now = (pg_time_t) time(NULL);
- Latch *latch;
+ ProcNumber walrcv_proc;
/*
* We always start at the beginning of the segment. That prevents a broken
@@ -309,14 +311,14 @@ RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo,
walrcv->receiveStart = recptr;
walrcv->receiveStartTLI = tli;
- latch = walrcv->latch;
+ walrcv_proc = walrcv->procno;
SpinLockRelease(&walrcv->mutex);
if (launch)
SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
- else if (latch)
- SetLatch(latch);
+ else if (walrcv_proc != INVALID_PROC_NUMBER)
+ SetLatch(&GetPGProcByNumber(walrcv_proc)->procLatch);
}
/*
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 132e789948..f86e4cda4c 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -21,7 +21,6 @@
#include "replication/logicalproto.h"
#include "replication/walsender.h"
#include "storage/condition_variable.h"
-#include "storage/latch.h"
#include "storage/spin.h"
#include "utils/tuplestore.h"
@@ -58,11 +57,19 @@ typedef enum
typedef struct
{
/*
- * PID of currently active walreceiver process, its current state and
- * start time (actually, the time at which it was requested to be
- * started).
+ * Proc number of currently active walreceiver process, so that the
+ * startup process can wake it up after telling it where to start
+ * streaming (after setting receiveStart and receiveStartTLI), and also to
+ * tell it to send apply feedback to the primary whenever specially marked
+ * commit records are applied.
*/
+ ProcNumber procno;
pid_t pid;
+
+ /*
+ * Its current state and start time (actually, the time at which it was
+ * requested to be started).
+ */
WalRcvState walRcvState;
ConditionVariable walRcvStoppedCV;
pg_time_t startTime;
@@ -134,15 +141,6 @@ typedef struct
/* set true once conninfo is ready to display (obfuscated pwds etc) */
bool ready_to_display;
- /*
- * Latch used by startup process to wake up walreceiver after telling it
- * where to start streaming (after setting receiveStart and
- * receiveStartTLI), and also to tell it to send apply feedback to the
- * primary whenever specially marked commit records are applied. This is
- * normally mapped to procLatch when walreceiver is running.
- */
- Latch *latch;
-
slock_t mutex; /* locks shared variables shown above */
/*
--
2.39.2
[text/x-patch] v2-0006-Use-proc-number-for-wakeups-in-waitlsn.c.patch (4.6K, ../[email protected]/7-v2-0006-Use-proc-number-for-wakeups-in-waitlsn.c.patch)
download | inline diff:
From ed7fdab78b847d903c7f4f4c29f725971e3113f7 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Fri, 30 Aug 2024 22:49:25 +0300
Subject: [PATCH v2 06/12] Use proc number for wakeups in waitlsn.c
Also rename WaitLSNSetLatches() to WaitLSNWakeup(), to emphasize what
it does, rather than how it does it.
---
src/backend/access/transam/xlog.c | 2 +-
src/backend/access/transam/xlogrecovery.c | 2 +-
src/backend/commands/waitlsn.c | 18 ++++++++++--------
src/include/commands/waitlsn.h | 10 ++++------
4 files changed, 16 insertions(+), 16 deletions(-)
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 63a8ab5a29..fdc0906314 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6155,7 +6155,7 @@ StartupXLOG(void)
* Wake up all waiters for replay LSN. They need to report an error that
* recovery was ended before achieving the target LSN.
*/
- WaitLSNSetLatches(InvalidXLogRecPtr);
+ WaitLSNWakeup(InvalidXLogRecPtr);
/*
* Shutdown the recovery environment. This must occur after
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index ad817fbca6..35a5e31e3d 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -1837,7 +1837,7 @@ PerformWalRecovery(void)
if (waitLSNState &&
(XLogRecoveryCtl->lastReplayedEndRecPtr >=
pg_atomic_read_u64(&waitLSNState->minWaitedLSN)))
- WaitLSNSetLatches(XLogRecoveryCtl->lastReplayedEndRecPtr);
+ WaitLSNWakeup(XLogRecoveryCtl->lastReplayedEndRecPtr);
/* Else, try to fetch the next WAL record */
record = ReadRecord(xlogprefetcher, LOG, false, replayTLI);
diff --git a/src/backend/commands/waitlsn.c b/src/backend/commands/waitlsn.c
index d9cf9e7d75..1937bafafc 100644
--- a/src/backend/commands/waitlsn.c
+++ b/src/backend/commands/waitlsn.c
@@ -113,7 +113,7 @@ addLSNWaiter(XLogRecPtr lsn)
Assert(!procInfo->inHeap);
- procInfo->latch = MyLatch;
+ procInfo->procno = MyProcNumber;
procInfo->waitLSN = lsn;
pairingheap_add(&waitLSNState->waitersHeap, &procInfo->phNode);
@@ -152,19 +152,21 @@ deleteLSNWaiter(void)
* and set latches for all waiters.
*/
void
-WaitLSNSetLatches(XLogRecPtr currentLSN)
+WaitLSNWakeup(XLogRecPtr currentLSN)
{
int i;
- Latch **wakeUpProcLatches;
+ ProcNumber *wakeUpProcs;
int numWakeUpProcs = 0;
- wakeUpProcLatches = palloc(sizeof(Latch *) * MaxBackends);
+ wakeUpProcs = palloc(sizeof(ProcNumber) * MaxBackends);
LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE);
/*
* Iterate the pairing heap of waiting processes till we find LSN not yet
- * replayed. Record the process latches to set them later.
+ * replayed. Record the process numbers to interrupt, but to avoid
+ * holding the lock for too long, send the interrupts to them only after
+ * releasing the lock.
*/
while (!pairingheap_is_empty(&waitLSNState->waitersHeap))
{
@@ -175,7 +177,7 @@ WaitLSNSetLatches(XLogRecPtr currentLSN)
procInfo->waitLSN > currentLSN)
break;
- wakeUpProcLatches[numWakeUpProcs++] = procInfo->latch;
+ wakeUpProcs[numWakeUpProcs++] = procInfo->procno;
(void) pairingheap_remove_first(&waitLSNState->waitersHeap);
procInfo->inHeap = false;
}
@@ -192,9 +194,9 @@ WaitLSNSetLatches(XLogRecPtr currentLSN)
*/
for (i = 0; i < numWakeUpProcs; i++)
{
- SetLatch(wakeUpProcLatches[i]);
+ SetLatch(&GetPGProcByNumber(wakeUpProcs[i])->procLatch);
}
- pfree(wakeUpProcLatches);
+ pfree(wakeUpProcs);
}
/*
diff --git a/src/include/commands/waitlsn.h b/src/include/commands/waitlsn.h
index f719feadb0..81fcb9c232 100644
--- a/src/include/commands/waitlsn.h
+++ b/src/include/commands/waitlsn.h
@@ -16,6 +16,7 @@
#include "postgres.h"
#include "port/atomics.h"
#include "storage/latch.h"
+#include "storage/procnumber.h"
#include "storage/spin.h"
#include "tcop/dest.h"
@@ -29,11 +30,8 @@ typedef struct WaitLSNProcInfo
/* LSN, which this process is waiting for */
XLogRecPtr waitLSN;
- /*
- * A pointer to the latch, which should be set once the waitLSN is
- * replayed.
- */
- Latch *latch;
+ /* Process to be woken up once the waitLSN is replayed */
+ ProcNumber procno;
/* A pairing heap node for participation in waitLSNState->waitersHeap */
pairingheap_node phNode;
@@ -74,7 +72,7 @@ extern PGDLLIMPORT WaitLSNState *waitLSNState;
extern Size WaitLSNShmemSize(void);
extern void WaitLSNShmemInit(void);
-extern void WaitLSNSetLatches(XLogRecPtr currentLSN);
+extern void WaitLSNWakeup(XLogRecPtr currentLSN);
extern void WaitLSNCleanup(void);
#endif /* WAIT_LSN_H */
--
2.39.2
[text/x-patch] v2-0007-Clean-up-existing-WaitLatch-calls-that-passed-MyL.patch (2.4K, ../[email protected]/8-v2-0007-Clean-up-existing-WaitLatch-calls-that-passed-MyL.patch)
download | inline diff:
From 777c7b0ba2fb5a538c0eb3ec0afe87b0d8997fec Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Fri, 30 Aug 2024 21:06:36 +0300
Subject: [PATCH v2 07/12] Clean up existing WaitLatch calls that passed
MyLatch without WL_LATCH_SET
The 'latch' argument is ignored if WL_LATCH_SET is not given. Clarify
these calls by not pointlessly passing MyLatch.
---
src/backend/libpq/be-secure-gssapi.c | 4 ++--
src/backend/libpq/be-secure-openssl.c | 2 +-
src/backend/postmaster/walsummarizer.c | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/src/backend/libpq/be-secure-gssapi.c b/src/backend/libpq/be-secure-gssapi.c
index 483636503c..2d36c76324 100644
--- a/src/backend/libpq/be-secure-gssapi.c
+++ b/src/backend/libpq/be-secure-gssapi.c
@@ -450,7 +450,7 @@ read_or_wait(Port *port, ssize_t len)
*/
if (ret <= 0)
{
- WaitLatchOrSocket(MyLatch,
+ WaitLatchOrSocket(NULL,
WL_SOCKET_READABLE | WL_EXIT_ON_PM_DEATH,
port->sock, 0, WAIT_EVENT_GSS_OPEN_SERVER);
@@ -668,7 +668,7 @@ secure_open_gssapi(Port *port)
/* Wait and retry if we couldn't write yet */
if (ret <= 0)
{
- WaitLatchOrSocket(MyLatch,
+ WaitLatchOrSocket(NULL,
WL_SOCKET_WRITEABLE | WL_EXIT_ON_PM_DEATH,
port->sock, 0, WAIT_EVENT_GSS_OPEN_SERVER);
continue;
diff --git a/src/backend/libpq/be-secure-openssl.c b/src/backend/libpq/be-secure-openssl.c
index 60cf5d16e7..1b36077891 100644
--- a/src/backend/libpq/be-secure-openssl.c
+++ b/src/backend/libpq/be-secure-openssl.c
@@ -519,7 +519,7 @@ aloop:
else
waitfor = WL_SOCKET_WRITEABLE | WL_EXIT_ON_PM_DEATH;
- (void) WaitLatchOrSocket(MyLatch, waitfor, port->sock, 0,
+ (void) WaitLatchOrSocket(NULL, waitfor, port->sock, 0,
WAIT_EVENT_SSL_OPEN_SERVER);
goto aloop;
case SSL_ERROR_SYSCALL:
diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c
index 3a407f8daa..48350bec52 100644
--- a/src/backend/postmaster/walsummarizer.c
+++ b/src/backend/postmaster/walsummarizer.c
@@ -315,7 +315,7 @@ WalSummarizerMain(char *startup_data, size_t startup_data_len)
* So a really fast retry time doesn't seem to be especially
* beneficial, and it will clutter the logs.
*/
- (void) WaitLatch(MyLatch,
+ (void) WaitLatch(NULL,
WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
10000,
WAIT_EVENT_WAL_SUMMARIZER_ERROR);
--
2.39.2
[text/x-patch] v2-0008-Replace-Latches-with-Interrupts.patch (134.0K, ../[email protected]/9-v2-0008-Replace-Latches-with-Interrupts.patch)
download | inline diff:
From 519067cd12ad89abef3b1b2c8ed2efa31f7ffa1d Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
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 <arpa/inet.h>
#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 <unistd.h>
+
+#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 <signal.h>
+
#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 <signal.h>
+
+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 <signal.h>
+#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 <signal.h>
+
+#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
[text/x-patch] v2-0009-Replace-ProcSendSignal-ProcWaitForSignal-with-new.patch (5.8K, ../[email protected]/10-v2-0009-Replace-ProcSendSignal-ProcWaitForSignal-with-new.patch)
download | inline diff:
From 3cf023c0defd935ec700963efba2b60a62aa0453 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Sat, 24 Aug 2024 00:58:05 +0300
Subject: [PATCH v2 09/12] Replace ProcSendSignal/ProcWaitForSignal with new
interrupt functions
---
src/backend/storage/buffer/bufmgr.c | 9 +++++++--
src/backend/storage/ipc/standby.c | 15 ++++++++++++---
src/backend/storage/lmgr/predicate.c | 7 +++++--
src/backend/storage/lmgr/proc.c | 28 ----------------------------
src/include/storage/proc.h | 3 ---
5 files changed, 24 insertions(+), 38 deletions(-)
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 5cdd2f10fc..43250ec22d 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -2891,7 +2891,7 @@ UnpinBufferNoOwner(BufferDesc *buf)
buf_state &= ~BM_PIN_COUNT_WAITER;
UnlockBufHdr(buf, buf_state);
- ProcSendSignal(wait_backend_pgprocno);
+ SendInterrupt(INTERRUPT_GENERAL_WAKEUP, wait_backend_pgprocno);
}
else
UnlockBufHdr(buf, buf_state);
@@ -5353,7 +5353,12 @@ LockBufferForCleanup(Buffer buffer)
SetStartupBufferPinWaitBufId(-1);
}
else
- ProcWaitForSignal(WAIT_EVENT_BUFFER_PIN);
+ {
+ WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_BUFFER_PIN);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
+ CHECK_FOR_INTERRUPTS();
+ }
/*
* Remove flag marking us as waiter. Normally this will not be set
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 872679ca44..8d843a200e 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -696,7 +696,10 @@ ResolveRecoveryConflictWithLock(LOCKTAG locktag, bool logging_conflict)
}
/* Wait to be signaled by the release of the Relation Lock */
- ProcWaitForSignal(PG_WAIT_LOCK | locktag.locktag_type);
+ WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ PG_WAIT_LOCK | locktag.locktag_type);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
+ CHECK_FOR_INTERRUPTS();
/*
* Exit if ltime is reached. Then all the backends holding conflicting
@@ -745,7 +748,10 @@ ResolveRecoveryConflictWithLock(LOCKTAG locktag, bool logging_conflict)
* until the relation locks are released or ltime is reached.
*/
got_standby_deadlock_timeout = false;
- ProcWaitForSignal(PG_WAIT_LOCK | locktag.locktag_type);
+ WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ PG_WAIT_LOCK | locktag.locktag_type);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
+ CHECK_FOR_INTERRUPTS();
}
cleanup:
@@ -839,7 +845,10 @@ ResolveRecoveryConflictWithBufferPin(void)
* SIGHUP signal handler, etc cannot do that because it uses the different
* latch from that ProcWaitForSignal() waits on.
*/
- ProcWaitForSignal(WAIT_EVENT_BUFFER_PIN);
+ WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_BUFFER_PIN);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
+ CHECK_FOR_INTERRUPTS();
if (got_standby_delay_timeout)
SendRecoveryConflictWithBufferPin(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c
index e24a0f2fdb..93715b9bdf 100644
--- a/src/backend/storage/lmgr/predicate.c
+++ b/src/backend/storage/lmgr/predicate.c
@@ -1571,7 +1571,10 @@ GetSafeSnapshot(Snapshot origSnapshot)
SxactIsROUnsafe(MySerializableXact)))
{
LWLockRelease(SerializableXactHashLock);
- ProcWaitForSignal(WAIT_EVENT_SAFE_SNAPSHOT);
+ WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_SAFE_SNAPSHOT);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
+ CHECK_FOR_INTERRUPTS();
LWLockAcquire(SerializableXactHashLock, LW_EXCLUSIVE);
}
MySerializableXact->flags &= ~SXACT_FLAG_DEFERRABLE_WAITING;
@@ -3604,7 +3607,7 @@ ReleasePredicateLocks(bool isCommit, bool isReadOnlySafe)
*/
if (SxactIsDeferrableWaiting(roXact) &&
(SxactIsROUnsafe(roXact) || SxactIsROSafe(roXact)))
- ProcSendSignal(roXact->pgprocno);
+ SendInterrupt(INTERRUPT_GENERAL_WAKEUP, roXact->pgprocno);
}
}
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 0c1b238e0e..b6ef418617 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -1866,34 +1866,6 @@ CheckDeadLockAlert(void)
errno = save_errno;
}
-/*
- * ProcWaitForSignal - wait for a signal from another backend.
- *
- * 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) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
- wait_event_info);
- ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
- CHECK_FOR_INTERRUPTS();
-}
-
-/*
- * ProcSendSignal - send the interrupt of a backend identified by ProcNumber
- */
-void
-ProcSendSignal(ProcNumber procNumber)
-{
- if (procNumber < 0 || procNumber >= ProcGlobal->allProcCount)
- elog(ERROR, "procNumber out of range");
-
- SendInterrupt(INTERRUPT_GENERAL_WAKEUP, procNumber);
-}
-
/*
* BecomeLockGroupLeader - designate process as lock group leader
*
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 15b3c7d0e2..83bbfe3c3b 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -511,9 +511,6 @@ extern void CheckDeadLockAlert(void);
extern bool IsWaitingForLock(void);
extern void LockErrorCleanup(void);
-extern void ProcWaitForSignal(uint32 wait_event_info);
-extern void ProcSendSignal(ProcNumber procNumber);
-
extern PGPROC *AuxiliaryPidGetProc(int pid);
extern void BecomeLockGroupLeader(void);
--
2.39.2
[text/x-patch] v2-0010-Replace-all-remaining-Latch-calls-with-new-Interr.patch (70.2K, ../[email protected]/11-v2-0010-Replace-all-remaining-Latch-calls-with-new-Interr.patch)
download | inline diff:
From 994da6f0c7fb7b45e52729db91fc0b55850804cb Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Fri, 30 Aug 2024 19:48:57 +0300
Subject: [PATCH v2 10/12] Replace all remaining Latch calls with new Interrupt
calls
This is a mechanical patch that replaces all remaining Latch calls,
and removes the "latch.h" compatibility macros.
XXX: We might want to keep the compatibility macros for extensions though.
---
contrib/pg_prewarm/autoprewarm.c | 20 ++---
contrib/postgres_fdw/connection.c | 22 +++---
contrib/postgres_fdw/postgres_fdw.c | 2 +-
src/backend/access/heap/vacuumlazy.c | 11 +--
src/backend/access/transam/parallel.c | 20 ++---
src/backend/access/transam/xlogfuncs.c | 12 +--
src/backend/backup/basebackup_throttle.c | 14 ++--
src/backend/commands/async.c | 3 +-
src/backend/commands/vacuum.c | 4 +-
src/backend/executor/nodeGather.c | 8 +-
src/backend/libpq/auth.c | 8 +-
src/backend/libpq/be-secure-gssapi.c | 17 ++--
src/backend/libpq/be-secure-openssl.c | 6 +-
src/backend/libpq/pqmq.c | 7 +-
src/backend/libpq/pqsignal.c | 2 +-
src/backend/postmaster/autovacuum.c | 22 +++---
src/backend/postmaster/bgworker.c | 18 ++---
src/backend/postmaster/bgwriter.c | 17 ++--
src/backend/postmaster/syslogger.c | 16 ++--
src/backend/postmaster/walsummarizer.c | 24 +++---
.../libpqwalreceiver/libpqwalreceiver.c | 32 ++++----
.../replication/logical/applyparallelworker.c | 37 ++++-----
src/backend/replication/logical/launcher.c | 43 +++++-----
src/backend/replication/logical/slotsync.c | 23 +++---
src/backend/replication/logical/tablesync.c | 29 +++----
src/backend/replication/logical/worker.c | 15 ++--
src/backend/replication/syncrep.c | 7 +-
src/backend/replication/walreceiver.c | 23 +++---
src/backend/replication/walreceiverfuncs.c | 2 +-
src/backend/storage/buffer/bufmgr.c | 1 +
src/backend/storage/buffer/freelist.c | 25 +++---
src/backend/storage/ipc/procsignal.c | 4 +-
src/backend/storage/ipc/signalfuncs.c | 11 +--
src/backend/storage/ipc/sinval.c | 4 +-
src/backend/storage/ipc/standby.c | 5 +-
src/backend/storage/lmgr/condition_variable.c | 9 ++-
src/backend/storage/lmgr/predicate.c | 1 +
src/backend/storage/sync/sync.c | 2 +-
src/backend/tcop/postgres.c | 9 ++-
src/backend/utils/adt/misc.c | 26 +++---
src/backend/utils/adt/timestamp.c | 4 +-
src/backend/utils/init/postinit.c | 11 +--
src/backend/utils/misc/timeout.c | 4 +-
src/include/libpq/libpq-be-fe-helpers.h | 45 +++++------
src/include/storage/latch.h | 79 -------------------
src/include/storage/proc.h | 1 -
src/port/pgsleep.c | 6 +-
src/test/modules/worker_spi/worker_spi.c | 12 +--
48 files changed, 333 insertions(+), 390 deletions(-)
delete mode 100644 src/include/storage/latch.h
diff --git a/contrib/pg_prewarm/autoprewarm.c b/contrib/pg_prewarm/autoprewarm.c
index d061731706..4dbce7673b 100644
--- a/contrib/pg_prewarm/autoprewarm.c
+++ b/contrib/pg_prewarm/autoprewarm.c
@@ -39,8 +39,8 @@
#include "storage/dsm.h"
#include "storage/dsm_registry.h"
#include "storage/fd.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/procsignal.h"
@@ -223,10 +223,10 @@ autoprewarm_main(Datum main_arg)
if (autoprewarm_interval <= 0)
{
/* We're only dumping at shutdown, so just wait forever. */
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
- -1L,
- PG_WAIT_EXTENSION);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_EXIT_ON_PM_DEATH,
+ -1L,
+ PG_WAIT_EXTENSION);
}
else
{
@@ -250,14 +250,14 @@ autoprewarm_main(Datum main_arg)
}
/* Sleep until the next dump time. */
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- delay_in_ms,
- PG_WAIT_EXTENSION);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ delay_in_ms,
+ PG_WAIT_EXTENSION);
}
/* Reset the latch, loop. */
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 12d1fec0e8..c9adec13fc 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -28,7 +28,7 @@
#include "pgstat.h"
#include "postgres_fdw.h"
#include "storage/fd.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "utils/builtins.h"
#include "utils/datetime.h"
#include "utils/hsearch.h"
@@ -735,8 +735,8 @@ do_sql_command_end(PGconn *conn, const char *sql, bool consume_input)
/*
* If requested, consume whatever data is available from the socket. (Note
* that if all data is available, this allows pgfdw_get_result to call
- * PQgetResult without forcing the overhead of WaitLatchOrSocket, which
- * would be large compared to the overhead of PQconsumeInput.)
+ * PQgetResult without forcing the overhead of WaitInterruptOrSocket,
+ * which would be large compared to the overhead of PQconsumeInput.)
*/
if (consume_input && !PQconsumeInput(conn))
pgfdw_report_error(ERROR, NULL, conn, false, sql);
@@ -1371,7 +1371,7 @@ pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input)
/*
* If requested, consume whatever data is available from the socket. (Note
* that if all data is available, this allows pgfdw_get_cleanup_result to
- * call PQgetResult without forcing the overhead of WaitLatchOrSocket,
+ * call PQgetResult without forcing the overhead of WaitInterruptOrSocket,
* which would be large compared to the overhead of PQconsumeInput.)
*/
if (consume_input && !PQconsumeInput(conn))
@@ -1465,7 +1465,7 @@ pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query,
/*
* If requested, consume whatever data is available from the socket. (Note
* that if all data is available, this allows pgfdw_get_cleanup_result to
- * call PQgetResult without forcing the overhead of WaitLatchOrSocket,
+ * call PQgetResult without forcing the overhead of WaitInterruptOrSocket,
* which would be large compared to the overhead of PQconsumeInput.)
*/
if (consume_input && !PQconsumeInput(conn))
@@ -1545,12 +1545,12 @@ pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result,
pgfdw_we_cleanup_result = WaitEventExtensionNew("PostgresFdwCleanupResult");
/* Sleep until there's something to do */
- wc = WaitLatchOrSocket(MyLatch,
- WL_LATCH_SET | WL_SOCKET_READABLE |
- WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- PQsocket(conn),
- cur_timeout, pgfdw_we_cleanup_result);
- ResetLatch(MyLatch);
+ wc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_SOCKET_READABLE |
+ WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ PQsocket(conn),
+ cur_timeout, pgfdw_we_cleanup_result);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 4652851b50..771b6e9846 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -41,7 +41,7 @@
#include "optimizer/tlist.h"
#include "parser/parsetree.h"
#include "postgres_fdw.h"
-#include "storage/latch.h"
+#include "storage/waiteventset.h"
#include "utils/builtins.h"
#include "utils/float.h"
#include "utils/guc.h"
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index d82aa3d489..0f0229ff25 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -54,6 +54,7 @@
#include "postmaster/autovacuum.h"
#include "storage/bufmgr.h"
#include "storage/freespace.h"
+#include "storage/interrupt.h"
#include "storage/lmgr.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
@@ -2609,11 +2610,11 @@ lazy_truncate_heap(LVRelState *vacrel)
return;
}
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL,
- WAIT_EVENT_VACUUM_TRUNCATE);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL,
+ WAIT_EVENT_VACUUM_TRUNCATE);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
index 9aba17bd5e..49cd357025 100644
--- a/src/backend/access/transam/parallel.c
+++ b/src/backend/access/transam/parallel.c
@@ -33,6 +33,7 @@
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/spin.h"
@@ -742,12 +743,12 @@ WaitForParallelWorkersToAttach(ParallelContext *pcxt)
* might also get set for some other reason, but if so we'll
* just end up waiting for the same worker again.
*/
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
- -1, WAIT_EVENT_BGWORKER_STARTUP);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_EXIT_ON_PM_DEATH,
+ -1, WAIT_EVENT_BGWORKER_STARTUP);
- if (rc & WL_LATCH_SET)
- ResetLatch(MyLatch);
+ if (rc & WL_INTERRUPT)
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
}
@@ -862,9 +863,10 @@ WaitForParallelWorkersToFinish(ParallelContext *pcxt)
}
}
- (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, -1,
- WAIT_EVENT_PARALLEL_FINISH);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, -1,
+ WAIT_EVENT_PARALLEL_FINISH);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
if (pcxt->toc != NULL)
@@ -1017,7 +1019,7 @@ HandleParallelMessageInterrupt(void)
{
InterruptPending = true;
ParallelMessagePending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/src/backend/access/transam/xlogfuncs.c b/src/backend/access/transam/xlogfuncs.c
index b0c6d7c687..77d25447e9 100644
--- a/src/backend/access/transam/xlogfuncs.c
+++ b/src/backend/access/transam/xlogfuncs.c
@@ -28,7 +28,7 @@
#include "pgstat.h"
#include "replication/walreceiver.h"
#include "storage/fd.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "storage/standby.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
@@ -718,17 +718,17 @@ pg_promote(PG_FUNCTION_ARGS)
{
int rc;
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
if (!RecoveryInProgress())
PG_RETURN_BOOL(true);
CHECK_FOR_INTERRUPTS();
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
- 1000L / WAITS_PER_SECOND,
- WAIT_EVENT_PROMOTE);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+ 1000L / WAITS_PER_SECOND,
+ WAIT_EVENT_PROMOTE);
/*
* Emergency bailout if postmaster has died. This is to avoid the
diff --git a/src/backend/backup/basebackup_throttle.c b/src/backend/backup/basebackup_throttle.c
index 4477945e61..16bc93dda9 100644
--- a/src/backend/backup/basebackup_throttle.c
+++ b/src/backend/backup/basebackup_throttle.c
@@ -17,7 +17,7 @@
#include "backup/basebackup_sink.h"
#include "miscadmin.h"
#include "pgstat.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "utils/timestamp.h"
typedef struct bbsink_throttle
@@ -163,7 +163,7 @@ throttle(bbsink_throttle *sink, size_t increment)
if (sleep <= 0)
break;
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
/* We're eating a potentially set latch, so check for interrupts */
CHECK_FOR_INTERRUPTS();
@@ -172,12 +172,12 @@ throttle(bbsink_throttle *sink, size_t increment)
* (TAR_SEND_SIZE / throttling_sample * elapsed_min_unit) should be
* the maximum time to sleep. Thus the cast to long is safe.
*/
- wait_result = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- (long) (sleep / 1000),
- WAIT_EVENT_BASE_BACKUP_THROTTLE);
+ wait_result = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ (long) (sleep / 1000),
+ WAIT_EVENT_BASE_BACKUP_THROTTLE);
- if (wait_result & WL_LATCH_SET)
+ if (wait_result & WL_INTERRUPT)
CHECK_FOR_INTERRUPTS();
/* Done waiting? */
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 8ed503e1c1..0fd323eb1a 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -140,6 +140,7 @@
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/procsignal.h"
@@ -1812,7 +1813,7 @@ HandleNotifyInterrupt(void)
notifyInterruptPending = true;
/* make sure the event is processed in due course */
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 7d8e9d2045..4fe355ddb8 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -2387,8 +2387,8 @@ vacuum_delay_point(void)
/*
* We don't want to ignore postmaster death during very long vacuums
* with vacuum_cost_delay configured. We can't use the usual
- * WaitLatch() approach here because we want microsecond-based sleep
- * durations above.
+ * WaitInterrupt() approach here because we want microsecond-based
+ * sleep durations above.
*/
if (IsUnderPostmaster && !PostmasterIsAlive())
exit(1);
diff --git a/src/backend/executor/nodeGather.c b/src/backend/executor/nodeGather.c
index 5d4ffe989c..9751dfeb29 100644
--- a/src/backend/executor/nodeGather.c
+++ b/src/backend/executor/nodeGather.c
@@ -35,6 +35,7 @@
#include "executor/nodeGather.h"
#include "executor/tqueue.h"
#include "miscadmin.h"
+#include "storage/interrupt.h"
#include "optimizer/optimizer.h"
#include "utils/wait_event.h"
@@ -375,9 +376,10 @@ gather_readnext(GatherState *gatherstate)
return NULL;
/* Nothing to do except wait for developments. */
- (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
- WAIT_EVENT_EXECUTE_GATHER);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_EXECUTE_GATHER);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
nvisited = 0;
}
}
diff --git a/src/backend/libpq/auth.c b/src/backend/libpq/auth.c
index 2b607c5270..0929a9874d 100644
--- a/src/backend/libpq/auth.c
+++ b/src/backend/libpq/auth.c
@@ -1663,8 +1663,8 @@ interpret_ident_response(const char *ident_response,
* owns the tcp connection to "local_addr"
* If the username is successfully retrieved, check the usermap.
*
- * XXX: Using WaitLatchOrSocket() and doing a CHECK_FOR_INTERRUPTS() if the
- * latch was set would improve the responsiveness to timeouts/cancellations.
+ * XXX: Using WaitInterruptOrSocket() and doing a CHECK_FOR_INTERRUPTS() if the
+ * interrupt was pending would improve the responsiveness to timeouts/cancellations.
*/
static int
ident_inet(hbaPort *port)
@@ -3093,8 +3093,8 @@ PerformRadiusTransaction(const char *server, const char *secret, const char *por
* packets to our port thus causing us to retry in a loop and never time
* out.
*
- * XXX: Using WaitLatchOrSocket() and doing a CHECK_FOR_INTERRUPTS() if
- * the latch was set would improve the responsiveness to
+ * XXX: Using WaitInterruptOrSocket() and doing a CHECK_FOR_INTERRUPTS()
+ * if the interrupt was pending would improve the responsiveness to
* timeouts/cancellations.
*/
gettimeofday(&endtime, NULL);
diff --git a/src/backend/libpq/be-secure-gssapi.c b/src/backend/libpq/be-secure-gssapi.c
index 2d36c76324..de6030b401 100644
--- a/src/backend/libpq/be-secure-gssapi.c
+++ b/src/backend/libpq/be-secure-gssapi.c
@@ -21,6 +21,7 @@
#include "libpq/pqformat.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "storage/interrupt.h"
#include "utils/injection_point.h"
#include "utils/memutils.h"
@@ -414,7 +415,7 @@ be_gssapi_read(Port *port, void *ptr, size_t len)
/*
* Read the specified number of bytes off the wire, waiting using
- * WaitLatchOrSocket if we would block.
+ * WaitInterruptOrSocket if we would block.
*
* Results are read into PqGSSRecvBuffer.
*
@@ -450,9 +451,9 @@ read_or_wait(Port *port, ssize_t len)
*/
if (ret <= 0)
{
- WaitLatchOrSocket(NULL,
- WL_SOCKET_READABLE | WL_EXIT_ON_PM_DEATH,
- port->sock, 0, WAIT_EVENT_GSS_OPEN_SERVER);
+ WaitInterruptOrSocket(0,
+ WL_SOCKET_READABLE | WL_EXIT_ON_PM_DEATH,
+ port->sock, 0, WAIT_EVENT_GSS_OPEN_SERVER);
/*
* If we got back zero bytes, and then waited on the socket to be
@@ -489,7 +490,7 @@ read_or_wait(Port *port, ssize_t len)
*
* Note that unlike the be_gssapi_read/be_gssapi_write functions, this
* function WILL block on the socket to be ready for read/write (using
- * WaitLatchOrSocket) as appropriate while establishing the GSSAPI
+ * WaitInterruptOrSocket) as appropriate while establishing the GSSAPI
* session.
*/
ssize_t
@@ -668,9 +669,9 @@ secure_open_gssapi(Port *port)
/* Wait and retry if we couldn't write yet */
if (ret <= 0)
{
- WaitLatchOrSocket(NULL,
- WL_SOCKET_WRITEABLE | WL_EXIT_ON_PM_DEATH,
- port->sock, 0, WAIT_EVENT_GSS_OPEN_SERVER);
+ WaitInterruptOrSocket(0,
+ WL_SOCKET_WRITEABLE | WL_EXIT_ON_PM_DEATH,
+ port->sock, 0, WAIT_EVENT_GSS_OPEN_SERVER);
continue;
}
diff --git a/src/backend/libpq/be-secure-openssl.c b/src/backend/libpq/be-secure-openssl.c
index 1b36077891..309f03627a 100644
--- a/src/backend/libpq/be-secure-openssl.c
+++ b/src/backend/libpq/be-secure-openssl.c
@@ -32,7 +32,7 @@
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/fd.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
@@ -519,8 +519,8 @@ aloop:
else
waitfor = WL_SOCKET_WRITEABLE | WL_EXIT_ON_PM_DEATH;
- (void) WaitLatchOrSocket(NULL, waitfor, port->sock, 0,
- WAIT_EVENT_SSL_OPEN_SERVER);
+ (void) WaitInterruptOrSocket(0, waitfor, port->sock, 0,
+ WAIT_EVENT_SSL_OPEN_SERVER);
goto aloop;
case SSL_ERROR_SYSCALL:
if (r < 0 && errno != 0)
diff --git a/src/backend/libpq/pqmq.c b/src/backend/libpq/pqmq.c
index fd735e2fea..66062c4d75 100644
--- a/src/backend/libpq/pqmq.c
+++ b/src/backend/libpq/pqmq.c
@@ -19,6 +19,7 @@
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "storage/interrupt.h"
#include "replication/logicalworker.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
@@ -181,9 +182,9 @@ mq_putmessage(char msgtype, const char *s, size_t len)
if (result != SHM_MQ_WOULD_BLOCK)
break;
- (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
- WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
diff --git a/src/backend/libpq/pqsignal.c b/src/backend/libpq/pqsignal.c
index 22a16c50b2..2f5eebdfec 100644
--- a/src/backend/libpq/pqsignal.c
+++ b/src/backend/libpq/pqsignal.c
@@ -42,7 +42,7 @@ pqinitmask(void)
{
sigemptyset(&UnBlockSig);
- /* Note: InitializeLatchSupport() modifies UnBlockSig. */
+ /* Note: InitializeWaitEventSupport() modifies UnBlockSig. */
/* First set all signals, then clear some. */
sigfillset(&BlockSig);
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 7d0877c95e..7ec91b6ac8 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -89,8 +89,8 @@
#include "postmaster/interrupt.h"
#include "postmaster/postmaster.h"
#include "storage/bufmgr.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/lmgr.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
@@ -570,10 +570,10 @@ AutoVacLauncherMain(char *startup_data, size_t startup_data_len)
bool can_launch;
/*
- * This loop is a bit different from the normal use of WaitLatch,
+ * This loop is a bit different from the normal use of WaitInterrupt,
* because we'd like to sleep before the first launch of a child
- * process. So it's WaitLatch, then ResetLatch, then check for
- * wakening conditions.
+ * process. So it's WaitInterrupt, then ClearInterrupt, then check
+ * for wakening conditions.
*/
launcher_determine_sleep(!dlist_is_empty(&AutoVacuumShmem->av_freeWorkers),
@@ -581,14 +581,14 @@ AutoVacLauncherMain(char *startup_data, size_t startup_data_len)
/*
* Wait until naptime expires or we get some type of signal (all the
- * signal handlers will wake us by calling SetLatch).
+ * signal handlers will wake us by calling RaiseInterrupt).
*/
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- (nap.tv_sec * 1000L) + (nap.tv_usec / 1000L),
- WAIT_EVENT_AUTOVACUUM_MAIN);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ (nap.tv_sec * 1000L) + (nap.tv_usec / 1000L),
+ WAIT_EVENT_AUTOVACUUM_MAIN);
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
HandleAutoVacLauncherInterrupts();
@@ -1345,7 +1345,7 @@ static void
avl_sigusr2_handler(SIGNAL_ARGS)
{
got_SIGUSR2 = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index b83967cda3..32884eebca 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -21,8 +21,8 @@
#include "postmaster/postmaster.h"
#include "replication/logicallauncher.h"
#include "replication/logicalworker.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
@@ -1226,9 +1226,9 @@ WaitForBackgroundWorkerStartup(BackgroundWorkerHandle *handle, pid_t *pidp)
if (status != BGWH_NOT_YET_STARTED)
break;
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
- WAIT_EVENT_BGWORKER_STARTUP);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_POSTMASTER_DEATH, 0,
+ WAIT_EVENT_BGWORKER_STARTUP);
if (rc & WL_POSTMASTER_DEATH)
{
@@ -1236,7 +1236,7 @@ WaitForBackgroundWorkerStartup(BackgroundWorkerHandle *handle, pid_t *pidp)
break;
}
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
return status;
@@ -1269,9 +1269,9 @@ WaitForBackgroundWorkerShutdown(BackgroundWorkerHandle *handle)
if (status == BGWH_STOPPED)
break;
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
- WAIT_EVENT_BGWORKER_SHUTDOWN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_POSTMASTER_DEATH, 0,
+ WAIT_EVENT_BGWORKER_SHUTDOWN);
if (rc & WL_POSTMASTER_DEATH)
{
@@ -1279,7 +1279,7 @@ WaitForBackgroundWorkerShutdown(BackgroundWorkerHandle *handle)
break;
}
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
return status;
diff --git a/src/backend/postmaster/bgwriter.c b/src/backend/postmaster/bgwriter.c
index 0f75548759..a55c991df3 100644
--- a/src/backend/postmaster/bgwriter.c
+++ b/src/backend/postmaster/bgwriter.c
@@ -42,6 +42,7 @@
#include "storage/bufmgr.h"
#include "storage/condition_variable.h"
#include "storage/fd.h"
+#include "storage/interrupt.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/procsignal.h"
@@ -224,7 +225,7 @@ BackgroundWriterMain(char *startup_data, size_t startup_data_len)
int rc;
/* Clear any already-pending wakeups */
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
HandleMainLoopInterrupts();
@@ -302,9 +303,9 @@ BackgroundWriterMain(char *startup_data, size_t startup_data_len)
* down with latch events that are likely to happen frequently during
* normal operation.
*/
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- BgWriterDelay /* ms */ , WAIT_EVENT_BGWRITER_MAIN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ BgWriterDelay /* ms */ , WAIT_EVENT_BGWRITER_MAIN);
/*
* If no latch event and BgBufferSync says nothing's happening, extend
@@ -329,10 +330,10 @@ BackgroundWriterMain(char *startup_data, size_t startup_data_len)
/* Ask for notification at next buffer allocation */
StrategyNotifyBgWriter(MyProcNumber);
/* Sleep ... */
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- BgWriterDelay * HIBERNATE_FACTOR,
- WAIT_EVENT_BGWRITER_HIBERNATE);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ BgWriterDelay * HIBERNATE_FACTOR,
+ WAIT_EVENT_BGWRITER_HIBERNATE);
/* Reset the notification request in case we timed out */
StrategyNotifyBgWriter(-1);
}
diff --git a/src/backend/postmaster/syslogger.c b/src/backend/postmaster/syslogger.c
index b33033b290..264ef1bc40 100644
--- a/src/backend/postmaster/syslogger.c
+++ b/src/backend/postmaster/syslogger.c
@@ -44,8 +44,8 @@
#include "postmaster/syslogger.h"
#include "storage/dsm.h"
#include "storage/fd.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/pg_shmem.h"
#include "tcop/tcopprot.h"
#include "utils/guc.h"
@@ -328,7 +328,7 @@ SysLoggerMain(char *startup_data, size_t startup_data_len)
whereToSendOutput = DestNone;
/*
- * Set up a reusable WaitEventSet object we'll use to wait for our latch,
+ * Set up a reusable WaitEventSet object we'll use to wait for interrupts,
* and (except on Windows) our socket.
*
* Unlike all other postmaster child processes, we'll ignore postmaster
@@ -338,7 +338,7 @@ SysLoggerMain(char *startup_data, size_t startup_data_len)
* (including the postmaster).
*/
wes = CreateWaitEventSet(NULL, 2);
- AddWaitEventToSet(wes, WL_LATCH_SET, PGINVALID_SOCKET, 1 << INTERRUPT_GENERAL_WAKEUP, NULL);
+ AddWaitEventToSet(wes, WL_INTERRUPT, PGINVALID_SOCKET, 1 << INTERRUPT_GENERAL_WAKEUP, NULL);
#ifndef WIN32
AddWaitEventToSet(wes, WL_SOCKET_READABLE, syslogPipe[0], 0, NULL);
#endif
@@ -356,7 +356,7 @@ SysLoggerMain(char *startup_data, size_t startup_data_len)
#endif
/* Clear any already-pending wakeups */
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
/*
* Process any requests or signals received recently.
@@ -1185,7 +1185,7 @@ pipeThread(void *arg)
if (ftell(syslogFile) >= Log_RotationSize * 1024L ||
(csvlogFile != NULL && ftell(csvlogFile) >= Log_RotationSize * 1024L) ||
(jsonlogFile != NULL && ftell(jsonlogFile) >= Log_RotationSize * 1024L))
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
LeaveCriticalSection(&sysloggerSection);
}
@@ -1196,8 +1196,8 @@ pipeThread(void *arg)
/* if there's any data left then force it out now */
flush_pipe_input(logbuffer, &bytes_in_logbuffer);
- /* set the latch to waken the main thread, which will quit */
- SetLatch(MyLatch);
+ /* set the interrupt to waken the main thread, which will quit */
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
LeaveCriticalSection(&sysloggerSection);
_endthread();
@@ -1592,5 +1592,5 @@ static void
sigUsr1Handler(SIGNAL_ARGS)
{
rotation_requested = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
diff --git a/src/backend/postmaster/walsummarizer.c b/src/backend/postmaster/walsummarizer.c
index 096b1c2e03..678d86e0d6 100644
--- a/src/backend/postmaster/walsummarizer.c
+++ b/src/backend/postmaster/walsummarizer.c
@@ -38,8 +38,8 @@
#include "postmaster/walsummarizer.h"
#include "replication/walreceiver.h"
#include "storage/fd.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/procsignal.h"
@@ -315,10 +315,10 @@ WalSummarizerMain(char *startup_data, size_t startup_data_len)
* So a really fast retry time doesn't seem to be especially
* beneficial, and it will clutter the logs.
*/
- (void) WaitLatch(NULL,
- WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 10000,
- WAIT_EVENT_WAL_SUMMARIZER_ERROR);
+ (void) WaitInterrupt(0,
+ WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10000,
+ WAIT_EVENT_WAL_SUMMARIZER_ERROR);
}
/* We can now handle ereport(ERROR) */
@@ -630,8 +630,8 @@ GetOldestUnsummarizedLSN(TimeLineID *tli, bool *lsn_is_exact)
*
* This might not work, because there's no guarantee that the WAL summarizer
* process was successfully started, and it also might have started but
- * subsequently terminated. So, under normal circumstances, this will get the
- * latch set, but there's no guarantee.
+ * subsequently terminated. So, under normal circumstances, this will send
+ * the interrupt, but there's no guarantee.
*/
void
WakeupWalSummarizer(void)
@@ -1637,11 +1637,11 @@ summarizer_wait_for_wal(void)
}
/* OK, now sleep. */
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- sleep_quanta * MS_PER_SLEEP_QUANTUM,
- WAIT_EVENT_WAL_SUMMARIZER_WAL);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ sleep_quanta * MS_PER_SLEEP_QUANTUM,
+ WAIT_EVENT_WAL_SUMMARIZER_WAL);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
/* Reset count of pages read. */
pages_read_since_last_sleep = 0;
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index c74369953f..2294d5d437 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -30,7 +30,7 @@
#include "pgstat.h"
#include "pqexpbuffer.h"
#include "replication/walreceiver.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
@@ -237,16 +237,16 @@ libpqrcv_connect(const char *conninfo, bool replication, bool logical,
else
io_flag = WL_SOCKET_WRITEABLE;
- rc = WaitLatchOrSocket(MyLatch,
- WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
- PQsocket(conn->streamConn),
- 0,
- WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
+ rc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_EXIT_ON_PM_DEATH | WL_INTERRUPT | io_flag,
+ PQsocket(conn->streamConn),
+ 0,
+ WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
/* Interrupted? */
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
ProcessWalRcvInterrupts();
}
@@ -848,17 +848,17 @@ libpqrcv_PQgetResult(PGconn *streamConn)
* since we'll get interrupted by signals and can handle any
* interrupts here.
*/
- rc = WaitLatchOrSocket(MyLatch,
- WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
- WL_LATCH_SET,
- PQsocket(streamConn),
- 0,
- WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
+ rc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
+ WL_INTERRUPT,
+ PQsocket(streamConn),
+ 0,
+ WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
/* Interrupted? */
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
ProcessWalRcvInterrupts();
}
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index e7f7d4c5e4..334b5ae2f6 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -165,6 +165,7 @@
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/worker_internal.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "tcop/tcopprot.h"
@@ -804,13 +805,13 @@ LogicalParallelApplyLoop(shm_mq_handle *mqh)
int rc;
/* Wait for more work. */
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 1000L,
- WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 1000L,
+ WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN);
- if (rc & WL_LATCH_SET)
- ResetLatch(MyLatch);
+ if (rc & WL_INTERRUPT)
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
}
else
@@ -990,7 +991,7 @@ HandleParallelApplyMessageInterrupt(void)
{
InterruptPending = true;
ParallelApplyMessagePending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
@@ -1182,14 +1183,14 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
Assert(result == SHM_MQ_WOULD_BLOCK);
/* Wait before retrying. */
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- SHM_SEND_RETRY_INTERVAL_MS,
- WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ SHM_SEND_RETRY_INTERVAL_MS,
+ WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
@@ -1254,13 +1255,13 @@ pa_wait_for_xact_state(ParallelApplyWorkerInfo *winfo,
break;
/* Wait to be signalled. */
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 10L,
- WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10L,
+ WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
/* 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/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 84081dcf83..78244af7c9 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -34,6 +34,7 @@
#include "replication/slot.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
@@ -221,13 +222,13 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
* We need timeout because we generally don't get notified via latch
* about the worker attach. But we don't expect to have to wait long.
*/
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 10L, WAIT_EVENT_BGWORKER_STARTUP);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10L, WAIT_EVENT_BGWORKER_STARTUP);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
}
@@ -553,13 +554,13 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
LWLockRelease(LogicalRepWorkerLock);
/* Wait a bit --- we don't expect to have to wait long. */
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 10L, WAIT_EVENT_BGWORKER_STARTUP);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10L, WAIT_EVENT_BGWORKER_STARTUP);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
@@ -594,13 +595,13 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
LWLockRelease(LogicalRepWorkerLock);
/* Wait a bit --- we don't expect to have to wait long. */
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 10L, WAIT_EVENT_BGWORKER_SHUTDOWN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10L, WAIT_EVENT_BGWORKER_SHUTDOWN);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
@@ -1221,14 +1222,14 @@ ApplyLauncherMain(Datum main_arg)
MemoryContextDelete(subctx);
/* Wait for more work. */
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- wait_time,
- WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ wait_time,
+ WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 51072297fd..343bf3a52d 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -61,6 +61,7 @@
#include "replication/logical.h"
#include "replication/slotsync.h"
#include "replication/snapbuild.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
@@ -1254,13 +1255,13 @@ wait_for_slot_activity(bool some_slot_updated)
sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS;
}
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- sleep_ms,
- WAIT_EVENT_REPLICATION_SLOTSYNC_MAIN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ sleep_ms,
+ WAIT_EVENT_REPLICATION_SLOTSYNC_MAIN);
- if (rc & WL_LATCH_SET)
- ResetLatch(MyLatch);
+ if (rc & WL_INTERRUPT)
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
@@ -1592,13 +1593,13 @@ ShutDownSlotSync(void)
int rc;
/* Wait a bit, we don't expect to have to wait long */
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e03e761392..a59a859f5c 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -111,6 +111,7 @@
#include "replication/slot.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
@@ -210,11 +211,11 @@ wait_for_relation_state_change(Oid relid, char expected_state)
if (!worker)
break;
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
return false;
@@ -263,12 +264,12 @@ wait_for_worker_state_change(char expected_state)
* Wait. We expect to get a latch signal back from the apply worker,
* but use a timeout in case it dies without sending one.
*/
- rc = WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE);
- if (rc & WL_LATCH_SET)
- ResetLatch(MyLatch);
+ if (rc & WL_INTERRUPT)
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
return false;
@@ -773,12 +774,12 @@ copy_read_data(void *outbuf, int minread, int maxread)
/*
* Wait for more data or latch.
*/
- (void) WaitLatchOrSocket(MyLatch,
- WL_SOCKET_READABLE | WL_LATCH_SET |
- WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- fd, 1000L, WAIT_EVENT_LOGICAL_SYNC_DATA);
+ (void) WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_SOCKET_READABLE | WL_INTERRUPT |
+ WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ fd, 1000L, WAIT_EVENT_LOGICAL_SYNC_DATA);
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
return bytesread;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 0fb577d328..0050776827 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -177,6 +177,7 @@
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
#include "storage/buffile.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "tcop/tcopprot.h"
@@ -3739,15 +3740,15 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
else
wait_time = NAPTIME_PER_CYCLE;
- rc = WaitLatchOrSocket(MyLatch,
- WL_SOCKET_READABLE | WL_LATCH_SET |
- WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- fd, wait_time,
- WAIT_EVENT_LOGICAL_APPLY_MAIN);
+ rc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_SOCKET_READABLE | WL_INTERRUPT |
+ WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ fd, wait_time,
+ WAIT_EVENT_LOGICAL_APPLY_MAIN);
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c
index d8a52405b0..e74c153402 100644
--- a/src/backend/replication/syncrep.c
+++ b/src/backend/replication/syncrep.c
@@ -81,6 +81,7 @@
#include "replication/syncrep.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
+#include "storage/interrupt.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/guc_hooks.h"
@@ -230,7 +231,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
int rc;
/* Must reset the latch before testing state. */
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
/*
* Acquiring the lock is not needed, the latch ensures proper
@@ -285,8 +286,8 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
* Wait on latch. Any condition that should wake us up will set the
* latch, so no need for timeout.
*/
- rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH, -1,
- WAIT_EVENT_SYNC_REP);
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_POSTMASTER_DEATH, -1,
+ WAIT_EVENT_SYNC_REP);
/*
* If the postmaster dies, we'll probably never get an acknowledgment,
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 40d1fa4939..04eb1db6ca 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -67,6 +67,7 @@
#include "postmaster/interrupt.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
@@ -546,15 +547,15 @@ WalReceiverMain(char *startup_data, size_t startup_data_len)
* avoiding some system calls.
*/
Assert(wait_fd != PGINVALID_SOCKET);
- rc = WaitLatchOrSocket(MyLatch,
- WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
- WL_TIMEOUT | WL_LATCH_SET,
- wait_fd,
- nap,
- WAIT_EVENT_WAL_RECEIVER_MAIN);
- if (rc & WL_LATCH_SET)
+ rc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
+ WL_TIMEOUT | WL_INTERRUPT,
+ wait_fd,
+ nap,
+ WAIT_EVENT_WAL_RECEIVER_MAIN);
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
ProcessWalRcvInterrupts();
if (walrcv->force_reply)
@@ -692,7 +693,7 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
WakeupRecovery();
for (;;)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
ProcessWalRcvInterrupts();
@@ -724,8 +725,8 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
}
SpinLockRelease(&walrcv->mutex);
- (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
- WAIT_EVENT_WAL_RECEIVER_WAIT_START);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_WAL_RECEIVER_WAIT_START);
}
if (update_process_title)
diff --git a/src/backend/replication/walreceiverfuncs.c b/src/backend/replication/walreceiverfuncs.c
index d015f03244..6212e82577 100644
--- a/src/backend/replication/walreceiverfuncs.c
+++ b/src/backend/replication/walreceiverfuncs.c
@@ -26,7 +26,7 @@
#include "access/xlogrecovery.h"
#include "pgstat.h"
#include "replication/walreceiver.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "storage/shmem.h"
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 43250ec22d..cbb6c25f61 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -51,6 +51,7 @@
#include "storage/buf_internals.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
diff --git a/src/backend/storage/buffer/freelist.c b/src/backend/storage/buffer/freelist.c
index 383c598c5d..57733b03d2 100644
--- a/src/backend/storage/buffer/freelist.c
+++ b/src/backend/storage/buffer/freelist.c
@@ -19,6 +19,7 @@
#include "port/atomics.h"
#include "storage/buf_internals.h"
#include "storage/bufmgr.h"
+#include "storage/interrupt.h"
#include "storage/proc.h"
#define INT_ACCESS_ONCE(var) ((int)(*((volatile int *)&(var))))
@@ -219,25 +220,23 @@ StrategyGetBuffer(BufferAccessStrategy strategy, uint32 *buf_state, bool *from_r
/*
* If asked, we need to waken the bgwriter. Since we don't want to rely on
* a spinlock for this we force a read from shared memory once, and then
- * set the latch based on that value. We need to go through that length
- * because otherwise bgwprocno might be reset while/after we check because
- * the compiler might just reread from memory.
+ * send the interrupt based on that value. We need to go through that
+ * length because otherwise bgwprocno might be reset while/after we check
+ * because the compiler might just reread from memory.
*
- * This can possibly set the latch of the wrong process if the bgwriter
- * dies in the wrong moment. But since PGPROC->procLatch is never
- * deallocated the worst consequence of that is that we set the latch of
- * some arbitrary process.
+ * This can possibly send the interrupt to the wrong process if the
+ * bgwriter dies in the wrong moment, but that's harmless.
*/
bgwprocno = INT_ACCESS_ONCE(StrategyControl->bgwprocno);
if (bgwprocno != -1)
{
- /* reset bgwprocno first, before setting the latch */
+ /* reset bgwprocno first, before sending the interrupt */
StrategyControl->bgwprocno = -1;
/*
- * Not acquiring ProcArrayLock here which is slightly icky. It's
- * actually fine because procLatch isn't ever freed, so we just can
- * potentially set the wrong process' (or no process') latch.
+ * Not acquiring ProcArrayLock here which is slightly icky, because we
+ * can potentially send the interrupt to the wrong process (or no
+ * process), but it's harmless.
*/
SendInterrupt(INTERRUPT_GENERAL_WAKEUP, bgwprocno);
}
@@ -420,10 +419,10 @@ StrategySyncStart(uint32 *complete_passes, uint32 *num_buf_alloc)
}
/*
- * StrategyNotifyBgWriter -- set or clear allocation notification latch
+ * StrategyNotifyBgWriter -- set or clear allocation notification process
*
* If bgwprocno isn't -1, the next invocation of StrategyGetBuffer will
- * set that latch. Pass -1 to clear the pending notification before it
+ * interrupt that process. Pass -1 to clear the pending notification before it
* happens. This feature is used by the bgwriter process to wake itself up
* from hibernation, and is not meant for anybody else to use.
*/
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 87027f27eb..f51fe18cc9 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -25,8 +25,8 @@
#include "replication/logicalworker.h"
#include "replication/walsender.h"
#include "storage/condition_variable.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/shmem.h"
#include "storage/sinval.h"
#include "storage/smgr.h"
@@ -712,7 +712,7 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN))
HandleRecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/src/backend/storage/ipc/signalfuncs.c b/src/backend/storage/ipc/signalfuncs.c
index aa729a36e3..a354a7d065 100644
--- a/src/backend/storage/ipc/signalfuncs.c
+++ b/src/backend/storage/ipc/signalfuncs.c
@@ -20,6 +20,7 @@
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/syslogger.h"
+#include "storage/interrupt.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "storage/procarray.h"
@@ -204,12 +205,12 @@ pg_wait_until_termination(int pid, int64 timeout)
/* Process interrupts, if any, before waiting */
CHECK_FOR_INTERRUPTS();
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- waittime,
- WAIT_EVENT_BACKEND_TERMINATION);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ waittime,
+ WAIT_EVENT_BACKEND_TERMINATION);
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
remainingtime -= waittime;
} while (remainingtime > 0);
diff --git a/src/backend/storage/ipc/sinval.c b/src/backend/storage/ipc/sinval.c
index d9b16f84d1..c292369ff6 100644
--- a/src/backend/storage/ipc/sinval.c
+++ b/src/backend/storage/ipc/sinval.c
@@ -16,7 +16,7 @@
#include "access/xact.h"
#include "miscadmin.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "storage/sinvaladt.h"
#include "utils/inval.h"
@@ -161,7 +161,7 @@ HandleCatchupInterrupt(void)
catchupInterruptPending = true;
/* make sure the event is processed in due course */
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 8d843a200e..0f2570b234 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -26,6 +26,7 @@
#include "pgstat.h"
#include "replication/slot.h"
#include "storage/bufmgr.h"
+#include "storage/interrupt.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/sinvaladt.h"
@@ -594,7 +595,7 @@ ResolveRecoveryConflictWithDatabase(Oid dbid)
* ResolveRecoveryConflictWithLock is called from ProcSleep()
* to resolve conflicts with other backends holding relation locks.
*
- * The WaitLatch sleep normally done in ProcSleep()
+ * The WaitInterrupt sleep normally done in ProcSleep()
* (when not InHotStandby) is performed here, for code clarity.
*
* We either resolve conflicts immediately or set a timeout to wake us at
@@ -843,7 +844,7 @@ ResolveRecoveryConflictWithBufferPin(void)
* We assume that only UnpinBuffer() and the timeout requests established
* above can wake us up here. WakeupRecovery() called by walreceiver or
* SIGHUP signal handler, etc cannot do that because it uses the different
- * latch from that ProcWaitForSignal() waits on.
+ * interrupt from that ProcWaitForSignal() waits on.
*/
WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, WL_INTERRUPT | WL_EXIT_ON_PM_DEATH, 0,
WAIT_EVENT_BUFFER_PIN);
diff --git a/src/backend/storage/lmgr/condition_variable.c b/src/backend/storage/lmgr/condition_variable.c
index d5fabef171..5248a35504 100644
--- a/src/backend/storage/lmgr/condition_variable.c
+++ b/src/backend/storage/lmgr/condition_variable.c
@@ -21,6 +21,7 @@
#include "miscadmin.h"
#include "portability/instr_time.h"
#include "storage/condition_variable.h"
+#include "storage/interrupt.h"
#include "storage/proc.h"
#include "storage/proclist.h"
#include "storage/spin.h"
@@ -147,10 +148,10 @@ ConditionVariableTimedSleep(ConditionVariable *cv, long timeout,
INSTR_TIME_SET_CURRENT(start_time);
Assert(timeout >= 0 && timeout <= INT_MAX);
cur_timeout = timeout;
- wait_events = WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH;
+ wait_events = WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH;
}
else
- wait_events = WL_LATCH_SET | WL_EXIT_ON_PM_DEATH;
+ wait_events = WL_INTERRUPT | WL_EXIT_ON_PM_DEATH;
while (true)
{
@@ -160,10 +161,10 @@ ConditionVariableTimedSleep(ConditionVariable *cv, long timeout,
* Wait for latch to be set. (If we're awakened for some other
* reason, the code below will cope anyway.)
*/
- (void) WaitLatch(MyLatch, wait_events, cur_timeout, wait_event_info);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP, wait_events, cur_timeout, wait_event_info);
/* Reset latch before examining the state of the wait list. */
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
/*
* If this process has been taken out of the wait list, then we know
diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c
index 93715b9bdf..b870505768 100644
--- a/src/backend/storage/lmgr/predicate.c
+++ b/src/backend/storage/lmgr/predicate.c
@@ -207,6 +207,7 @@
#include "miscadmin.h"
#include "pgstat.h"
#include "port/pg_lfind.h"
+#include "storage/interrupt.h"
#include "storage/predicate.h"
#include "storage/predicate_internals.h"
#include "storage/proc.h"
diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.c
index cd5d1436ab..bec42f4060 100644
--- a/src/backend/storage/sync/sync.c
+++ b/src/backend/storage/sync/sync.c
@@ -27,7 +27,7 @@
#include "portability/instr_time.h"
#include "postmaster/bgwriter.h"
#include "storage/fd.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "storage/md.h"
#include "utils/hsearch.h"
#include "utils/memutils.h"
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 8bc6bea113..f2d6c334d4 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -62,6 +62,7 @@
#include "replication/walsender.h"
#include "rewrite/rewriteHandler.h"
#include "storage/bufmgr.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
@@ -540,7 +541,7 @@ ProcessClientReadInterrupt(bool blocked)
if (blocked)
CHECK_FOR_INTERRUPTS();
else
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
errno = save_errno;
@@ -592,7 +593,7 @@ ProcessClientWriteInterrupt(bool blocked)
}
}
else
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
errno = save_errno;
@@ -2991,7 +2992,7 @@ die(SIGNAL_ARGS)
pgStatSessionEndCause = DISCONNECT_KILLED;
/* If we're still here, waken anything waiting on the process latch */
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
/*
* If we're in single user mode, we want to quit immediately - we can't
@@ -3020,7 +3021,7 @@ StatementCancelHandler(SIGNAL_ARGS)
}
/* If we're still here, waken anything waiting on the process latch */
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/* signal handler for floating point exception */
diff --git a/src/backend/utils/adt/misc.c b/src/backend/utils/adt/misc.c
index 0e6c45807a..bc63ad4f8d 100644
--- a/src/backend/utils/adt/misc.c
+++ b/src/backend/utils/adt/misc.c
@@ -38,7 +38,7 @@
#include "postmaster/syslogger.h"
#include "rewrite/rewriteHandler.h"
#include "storage/fd.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
@@ -373,16 +373,16 @@ pg_sleep(PG_FUNCTION_ARGS)
float8 endtime;
/*
- * We sleep using WaitLatch, to ensure that we'll wake up promptly if an
- * important signal (such as SIGALRM or SIGINT) arrives. Because
- * WaitLatch's upper limit of delay is INT_MAX milliseconds, and the user
- * might ask for more than that, we sleep for at most 10 minutes and then
- * loop.
+ * We sleep using WaitInterrupt, to ensure that we'll wake up promptly if
+ * an important signal (such as SIGALRM or SIGINT) arrives. Because
+ * WaitInterrupt's upper limit of delay is INT_MAX milliseconds, and the
+ * user might ask for more than that, we sleep for at most 10 minutes and
+ * then loop.
*
* By computing the intended stop time initially, we avoid accumulation of
* extra delay across multiple sleeps. This also ensures we won't delay
- * less than the specified time when WaitLatch is terminated early by a
- * non-query-canceling signal such as SIGHUP.
+ * less than the specified time when WaitInterrupt is terminated early by
+ * a non-query-canceling signal such as SIGHUP.
*/
#define GetNowFloat() ((float8) GetCurrentTimestamp() / 1000000.0)
@@ -403,11 +403,11 @@ pg_sleep(PG_FUNCTION_ARGS)
else
break;
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- delay_ms,
- WAIT_EVENT_PG_SLEEP);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ delay_ms,
+ WAIT_EVENT_PG_SLEEP);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
PG_RETURN_VOID();
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index db9eea9098..22cabfb95d 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/src/backend/utils/adt/timestamp.c
@@ -1738,10 +1738,10 @@ TimestampDifference(TimestampTz start_time, TimestampTz stop_time,
* TimestampDifferenceMilliseconds -- convert the difference between two
* timestamps into integer milliseconds
*
- * This is typically used to calculate a wait timeout for WaitLatch()
+ * This is typically used to calculate a wait timeout for WaitInterrupt()
* or a related function. The choice of "long" as the result type
* is to harmonize with that; furthermore, we clamp the result to at most
- * INT_MAX milliseconds, because that's all that WaitLatch() allows.
+ * INT_MAX milliseconds, because that's all that WaitInterrupt() allows.
*
* We expect start_time <= stop_time. If not, we return zero,
* since then we're already past the previously determined stop_time.
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 3b50ce19a2..1350467268 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -45,6 +45,7 @@
#include "replication/walsender.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
@@ -1337,7 +1338,7 @@ TransactionTimeoutHandler(void)
{
TransactionTimeoutPending = true;
InterruptPending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
static void
@@ -1345,7 +1346,7 @@ IdleInTransactionSessionTimeoutHandler(void)
{
IdleInTransactionSessionTimeoutPending = true;
InterruptPending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
static void
@@ -1353,7 +1354,7 @@ IdleSessionTimeoutHandler(void)
{
IdleSessionTimeoutPending = true;
InterruptPending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
static void
@@ -1361,7 +1362,7 @@ IdleStatsUpdateTimeoutHandler(void)
{
IdleStatsUpdateTimeoutPending = true;
InterruptPending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
static void
@@ -1369,7 +1370,7 @@ ClientCheckTimeoutHandler(void)
{
CheckClientConnectionPending = true;
InterruptPending = true;
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
}
/*
diff --git a/src/backend/utils/misc/timeout.c b/src/backend/utils/misc/timeout.c
index ec7e570920..c45765c3d7 100644
--- a/src/backend/utils/misc/timeout.c
+++ b/src/backend/utils/misc/timeout.c
@@ -17,7 +17,7 @@
#include <sys/time.h>
#include "miscadmin.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "utils/timeout.h"
#include "utils/timestamp.h"
@@ -374,7 +374,7 @@ handle_sig_alarm(SIGNAL_ARGS)
* SIGALRM is always cause for waking anything waiting on the process
* latch.
*/
- SetLatch(MyLatch);
+ RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
/*
* Always reset signal_pending, even if !alarm_enabled, since indeed no
diff --git a/src/include/libpq/libpq-be-fe-helpers.h b/src/include/libpq/libpq-be-fe-helpers.h
index fe50829274..d1510dac41 100644
--- a/src/include/libpq/libpq-be-fe-helpers.h
+++ b/src/include/libpq/libpq-be-fe-helpers.h
@@ -43,7 +43,7 @@
#include "libpq-fe.h"
#include "miscadmin.h"
#include "storage/fd.h"
-#include "storage/latch.h"
+#include "storage/interrupt.h"
#include "utils/timestamp.h"
#include "utils/wait_event.h"
@@ -177,8 +177,8 @@ libpqsrv_connect_internal(PGconn *conn, uint32 wait_event_info)
return;
/*
- * WaitLatchOrSocket() can conceivably fail, handle that case here instead
- * of requiring all callers to do so.
+ * WaitInterruptOrSocket() can conceivably fail, handle that case here
+ * instead of requiring all callers to do so.
*/
PG_TRY();
{
@@ -209,16 +209,16 @@ libpqsrv_connect_internal(PGconn *conn, uint32 wait_event_info)
else
io_flag = WL_SOCKET_WRITEABLE;
- rc = WaitLatchOrSocket(MyLatch,
- WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
- PQsocket(conn),
- 0,
- wait_event_info);
+ rc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_EXIT_ON_PM_DEATH | WL_INTERRUPT | io_flag,
+ PQsocket(conn),
+ 0,
+ wait_event_info);
/* Interrupted? */
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
@@ -341,17 +341,17 @@ libpqsrv_get_result(PGconn *conn, uint32 wait_event_info)
{
int rc;
- rc = WaitLatchOrSocket(MyLatch,
- WL_EXIT_ON_PM_DEATH | WL_LATCH_SET |
- WL_SOCKET_READABLE,
- PQsocket(conn),
- 0,
- wait_event_info);
+ rc = WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_EXIT_ON_PM_DEATH | WL_INTERRUPT |
+ WL_SOCKET_READABLE,
+ PQsocket(conn),
+ 0,
+ wait_event_info);
/* Interrupted? */
- if (rc & WL_LATCH_SET)
+ if (rc & WL_INTERRUPT)
{
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
@@ -407,7 +407,7 @@ libpqsrv_cancel(PGconn *conn, TimestampTz endtime)
PostgresPollingStatusType pollres;
TimestampTz now;
long cur_timeout;
- int waitEvents = WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH;
+ int waitEvents = WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH;
pollres = PQcancelPoll(cancel_conn);
if (pollres == PGRES_POLLING_OK)
@@ -436,10 +436,11 @@ libpqsrv_cancel(PGconn *conn, TimestampTz endtime)
}
/* Sleep until there's something to do */
- WaitLatchOrSocket(MyLatch, waitEvents, PQcancelSocket(cancel_conn),
- cur_timeout, PG_WAIT_CLIENT);
+ WaitInterruptOrSocket(1 << INTERRUPT_GENERAL_WAKEUP, waitEvents,
+ PQcancelSocket(cancel_conn),
+ cur_timeout, PG_WAIT_CLIENT);
- ResetLatch(MyLatch);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
}
diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h
deleted file mode 100644
index 084d64a19b..0000000000
--- a/src/include/storage/latch.h
+++ /dev/null
@@ -1,79 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * latch.h
- * Backwards-compatibility macros for the old Latch interface
- *
- * 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
- *
- * src/include/storage/latch.h
- *
- *-------------------------------------------------------------------------
- */
-
-#ifndef LATCH_H
-#define LATCH_H
-
-#include "storage/interrupt.h"
-
-#define WL_LATCH_SET WL_INTERRUPT
-
-/*
- * 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.
- */
-
-#define MyLatch ((void *) 0xAAAA)
-
-static inline void
-SetLatch(void *dummy)
-{
- Assert(dummy == MyLatch);
- RaiseInterrupt(INTERRUPT_GENERAL_WAKEUP);
-}
-
-static inline void
-ResetLatch(void *dummy)
-{
- Assert(dummy == MyLatch);
- ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
-}
-
-static inline int
-WaitLatch(void *dummy, int wakeEvents, long timeout, uint32 wait_event_info)
-{
- uint32 interrupt_mask = 0;
-
- Assert(dummy == MyLatch || dummy == NULL);
-
- if ((wakeEvents & WL_LATCH_SET) && dummy == MyLatch)
- interrupt_mask = 1 << INTERRUPT_GENERAL_WAKEUP;
- return WaitInterrupt(interrupt_mask, wakeEvents,
- timeout, wait_event_info);
-}
-
-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);
-
- if ((wakeEvents & WL_LATCH_SET) && dummy == MyLatch)
- interrupt_mask = 1 << INTERRUPT_GENERAL_WAKEUP;
- return WaitInterruptOrSocket(interrupt_mask, wakeEvents,
- sock, timeout, wait_event_info);
-}
-
-#endif
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 83bbfe3c3b..15ce4f0464 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -17,7 +17,6 @@
#include "access/clog.h"
#include "access/xlogdefs.h"
#include "lib/ilist.h"
-#include "storage/latch.h"
#include "storage/lock.h"
#include "storage/pg_sema.h"
#include "storage/proclist_types.h"
diff --git a/src/port/pgsleep.c b/src/port/pgsleep.c
index 1284458bfc..34ece20d66 100644
--- a/src/port/pgsleep.c
+++ b/src/port/pgsleep.c
@@ -32,10 +32,10 @@
*
* CAUTION: It's not a good idea to use long sleeps in the backend. They will
* silently return early if a signal is caught, but that doesn't include
- * latches being set on most OSes, and even signal handlers that set MyLatch
+ * interrupts being set on most OSes, and even signal handlers that raise an interrupt
* might happen to run before the sleep begins, allowing the full delay.
- * Better practice is to use WaitLatch() with a timeout, so that backends
- * respond to latches and signals promptly.
+ * Better practice is to use WaitInterrupt() with a timeout, so that backends
+ * respond to interrupts and signals promptly.
*/
void
pg_usleep(long microsec)
diff --git a/src/test/modules/worker_spi/worker_spi.c b/src/test/modules/worker_spi/worker_spi.c
index de8f46902b..1e484455fc 100644
--- a/src/test/modules/worker_spi/worker_spi.c
+++ b/src/test/modules/worker_spi/worker_spi.c
@@ -26,8 +26,8 @@
#include "miscadmin.h"
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
+#include "storage/interrupt.h"
#include "storage/ipc.h"
-#include "storage/latch.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#include "storage/shmem.h"
@@ -231,11 +231,11 @@ worker_spi_main(Datum main_arg)
* necessary, but is awakened if postmaster dies. That way the
* background process goes away immediately in an emergency.
*/
- (void) WaitLatch(MyLatch,
- WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
- worker_spi_naptime * 1000L,
- worker_spi_wait_event_main);
- ResetLatch(MyLatch);
+ (void) WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ worker_spi_naptime * 1000L,
+ worker_spi_wait_event_main);
+ ClearInterrupt(INTERRUPT_GENERAL_WAKEUP);
CHECK_FOR_INTERRUPTS();
--
2.39.2
[text/x-patch] v2-0011-Fix-lost-wakeup-issue-in-logical-replication-laun.patch (4.0K, ../[email protected]/12-v2-0011-Fix-lost-wakeup-issue-in-logical-replication-laun.patch)
download | inline diff:
From 07f86a6e693c8ae1d093a67c51919b92c8ad9465 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <[email protected]>
Date: Sat, 24 Aug 2024 20:12:08 +0300
Subject: [PATCH v2 11/12] Fix lost wakeup issue in logical replication
launcher
by using a different interrupt reason for subscription changes
https://www.postgresql.org/message-id/flat/ff0663d9-8011-420f-a169-efbf57327cb5%40iki.fi#bef984f8c43d6b8a9428d2c5547fe72b
---
src/backend/replication/logical/launcher.c | 23 ++++++++++++++--------
src/include/storage/interrupt.h | 5 ++++-
2 files changed, 19 insertions(+), 9 deletions(-)
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 78244af7c9..206733681a 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -57,7 +57,7 @@ LogicalRepWorker *MyLogicalRepWorker = NULL;
typedef struct LogicalRepCtxStruct
{
/* Supervisor process. */
- pid_t launcher_pid;
+ ProcNumber launcher_procno;
/* Hash table holding last start times of subscriptions' apply workers. */
dsa_handle last_start_dsa;
@@ -813,7 +813,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
static void
logicalrep_launcher_onexit(int code, Datum arg)
{
- LogicalRepCtx->launcher_pid = 0;
+ LogicalRepCtx->launcher_procno = INVALID_PROC_NUMBER;
}
/*
@@ -973,6 +973,7 @@ ApplyLauncherShmemInit(void)
memset(LogicalRepCtx, 0, ApplyLauncherShmemSize());
+ LogicalRepCtx->launcher_procno = INVALID_PROC_NUMBER;
LogicalRepCtx->last_start_dsa = DSA_HANDLE_INVALID;
LogicalRepCtx->last_start_dsh = DSHASH_HANDLE_INVALID;
@@ -1118,8 +1119,12 @@ ApplyLauncherWakeupAtCommit(void)
static void
ApplyLauncherWakeup(void)
{
- if (LogicalRepCtx->launcher_pid != 0)
- kill(LogicalRepCtx->launcher_pid, SIGUSR1);
+ volatile LogicalRepCtxStruct *repctx = LogicalRepCtx;
+ ProcNumber launcher_procno;
+
+ launcher_procno = repctx->launcher_procno;
+ if (launcher_procno != INVALID_PROC_NUMBER)
+ SendInterrupt(INTERRUPT_SUBSCRIPTION_CHANGE, launcher_procno);
}
/*
@@ -1133,8 +1138,8 @@ ApplyLauncherMain(Datum main_arg)
before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
- Assert(LogicalRepCtx->launcher_pid == 0);
- LogicalRepCtx->launcher_pid = MyProcPid;
+ Assert(LogicalRepCtx->launcher_procno == INVALID_PROC_NUMBER);
+ LogicalRepCtx->launcher_procno = MyProcNumber;
/* Establish signal handlers. */
pqsignal(SIGHUP, SignalHandlerForConfigReload);
@@ -1166,6 +1171,7 @@ ApplyLauncherMain(Datum main_arg)
oldctx = MemoryContextSwitchTo(subctx);
/* Start any missing workers for enabled subscriptions. */
+ ClearInterrupt(INTERRUPT_SUBSCRIPTION_CHANGE);
sublist = get_subscription_list();
foreach(lc, sublist)
{
@@ -1222,7 +1228,8 @@ ApplyLauncherMain(Datum main_arg)
MemoryContextDelete(subctx);
/* Wait for more work. */
- rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP,
+ rc = WaitInterrupt(1 << INTERRUPT_GENERAL_WAKEUP |
+ 1 << INTERRUPT_SUBSCRIPTION_CHANGE,
WL_INTERRUPT | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
wait_time,
WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
@@ -1249,7 +1256,7 @@ ApplyLauncherMain(Datum main_arg)
bool
IsLogicalLauncher(void)
{
- return LogicalRepCtx->launcher_pid == MyProcPid;
+ return LogicalRepCtx->launcher_procno == MyProcNumber;
}
/*
diff --git a/src/include/storage/interrupt.h b/src/include/storage/interrupt.h
index 1dc85339a9..b1a3f7a974 100644
--- a/src/include/storage/interrupt.h
+++ b/src/include/storage/interrupt.h
@@ -98,11 +98,14 @@ typedef enum
INTERRUPT_GENERAL_WAKEUP,
/*
- * INTERRUPT_RECOVERY_WAKEUP is used to wake up startup process, to tell
+ * INTERRUPT_RECOVERY_CONTINUE 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,
+
+ /* sent to logical replication launcher, when a subscription changes */
+ INTERRUPT_SUBSCRIPTION_CHANGE,
} InterruptType;
/*
--
2.39.2
view thread (19+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Interrupts vs signals
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox