From afff68be45932c0340fda79ff1f7a154130beaf5 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas 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