mirror of
https://git.postgresql.org/git/postgresql.git
synced 2025-01-18 18:44:06 +08:00
Add condition variable for walreceiver shutdown.
Use this new CV to wait for walreceiver shutdown without a sleep/poll loop, while also benefiting from standard postmaster death handling. Discussion: https://postgr.es/m/CA%2BhUKGK1607VmtrDUHQXrsooU%3Dap4g4R2yaoByWOOA3m8xevUQ%40mail.gmail.com
This commit is contained in:
parent
600f2f50b7
commit
de829ddf23
@ -1766,6 +1766,10 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser
|
||||
<entry>Waiting for confirmation from a remote server during synchronous
|
||||
replication.</entry>
|
||||
</row>
|
||||
<row>
|
||||
<entry><literal>WalrcvExit</literal></entry>
|
||||
<entry>Waiting for the walreceiver to exit.</entry>
|
||||
</row>
|
||||
<row>
|
||||
<entry><literal>XactGroupUpdate</literal></entry>
|
||||
<entry>Waiting for the group leader to update transaction status at
|
||||
|
@ -4124,6 +4124,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
|
||||
case WAIT_EVENT_SYNC_REP:
|
||||
event_name = "SyncRep";
|
||||
break;
|
||||
case WAIT_EVENT_WALRCV_EXIT:
|
||||
event_name = "WalrcvExit";
|
||||
break;
|
||||
case WAIT_EVENT_XACT_GROUP_UPDATE:
|
||||
event_name = "XactGroupUpdate";
|
||||
break;
|
||||
|
@ -207,6 +207,7 @@ WalReceiverMain(void)
|
||||
|
||||
case WALRCV_STOPPED:
|
||||
SpinLockRelease(&walrcv->mutex);
|
||||
ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
|
||||
proc_exit(1);
|
||||
break;
|
||||
|
||||
@ -784,6 +785,8 @@ WalRcvDie(int code, Datum arg)
|
||||
walrcv->latch = NULL;
|
||||
SpinLockRelease(&walrcv->mutex);
|
||||
|
||||
ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
|
||||
|
||||
/* Terminate the connection gracefully. */
|
||||
if (wrconn != NULL)
|
||||
walrcv_disconnect(wrconn);
|
||||
|
@ -23,6 +23,7 @@
|
||||
#include <signal.h>
|
||||
|
||||
#include "access/xlog_internal.h"
|
||||
#include "pgstat.h"
|
||||
#include "postmaster/startup.h"
|
||||
#include "replication/walreceiver.h"
|
||||
#include "storage/pmsignal.h"
|
||||
@ -62,6 +63,7 @@ WalRcvShmemInit(void)
|
||||
/* First time through, so initialize */
|
||||
MemSet(WalRcv, 0, WalRcvShmemSize());
|
||||
WalRcv->walRcvState = WALRCV_STOPPED;
|
||||
ConditionVariableInit(&WalRcv->walRcvStoppedCV);
|
||||
SpinLockInit(&WalRcv->mutex);
|
||||
pg_atomic_init_u64(&WalRcv->writtenUpto, 0);
|
||||
WalRcv->latch = NULL;
|
||||
@ -95,12 +97,18 @@ WalRcvRunning(void)
|
||||
|
||||
if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
|
||||
{
|
||||
bool stopped = false;
|
||||
|
||||
SpinLockAcquire(&walrcv->mutex);
|
||||
|
||||
if (walrcv->walRcvState == WALRCV_STARTING)
|
||||
{
|
||||
state = walrcv->walRcvState = WALRCV_STOPPED;
|
||||
|
||||
stopped = true;
|
||||
}
|
||||
SpinLockRelease(&walrcv->mutex);
|
||||
|
||||
if (stopped)
|
||||
ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
|
||||
}
|
||||
}
|
||||
|
||||
@ -140,12 +148,18 @@ WalRcvStreaming(void)
|
||||
|
||||
if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
|
||||
{
|
||||
bool stopped = false;
|
||||
|
||||
SpinLockAcquire(&walrcv->mutex);
|
||||
|
||||
if (walrcv->walRcvState == WALRCV_STARTING)
|
||||
{
|
||||
state = walrcv->walRcvState = WALRCV_STOPPED;
|
||||
|
||||
stopped = true;
|
||||
}
|
||||
SpinLockRelease(&walrcv->mutex);
|
||||
|
||||
if (stopped)
|
||||
ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
|
||||
}
|
||||
}
|
||||
|
||||
@ -165,6 +179,7 @@ ShutdownWalRcv(void)
|
||||
{
|
||||
WalRcvData *walrcv = WalRcv;
|
||||
pid_t walrcvpid = 0;
|
||||
bool stopped = false;
|
||||
|
||||
/*
|
||||
* Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
|
||||
@ -178,6 +193,7 @@ ShutdownWalRcv(void)
|
||||
break;
|
||||
case WALRCV_STARTING:
|
||||
walrcv->walRcvState = WALRCV_STOPPED;
|
||||
stopped = true;
|
||||
break;
|
||||
|
||||
case WALRCV_STREAMING:
|
||||
@ -191,6 +207,10 @@ ShutdownWalRcv(void)
|
||||
}
|
||||
SpinLockRelease(&walrcv->mutex);
|
||||
|
||||
/* Unnecessary but consistent. */
|
||||
if (stopped)
|
||||
ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
|
||||
|
||||
/*
|
||||
* Signal walreceiver process if it was still running.
|
||||
*/
|
||||
@ -201,16 +221,11 @@ ShutdownWalRcv(void)
|
||||
* Wait for walreceiver to acknowledge its death by setting state to
|
||||
* WALRCV_STOPPED.
|
||||
*/
|
||||
ConditionVariablePrepareToSleep(&walrcv->walRcvStoppedCV);
|
||||
while (WalRcvRunning())
|
||||
{
|
||||
/*
|
||||
* This possibly-long loop needs to handle interrupts of startup
|
||||
* process.
|
||||
*/
|
||||
HandleStartupProcInterrupts();
|
||||
|
||||
pg_usleep(100000); /* 100ms */
|
||||
}
|
||||
ConditionVariableSleep(&walrcv->walRcvStoppedCV,
|
||||
WAIT_EVENT_WALRCV_EXIT);
|
||||
ConditionVariableCancelSleep();
|
||||
}
|
||||
|
||||
/*
|
||||
|
@ -1009,6 +1009,7 @@ typedef enum
|
||||
WAIT_EVENT_REPLICATION_SLOT_DROP,
|
||||
WAIT_EVENT_SAFE_SNAPSHOT,
|
||||
WAIT_EVENT_SYNC_REP,
|
||||
WAIT_EVENT_WALRCV_EXIT,
|
||||
WAIT_EVENT_XACT_GROUP_UPDATE
|
||||
} WaitEventIPC;
|
||||
|
||||
|
@ -19,6 +19,7 @@
|
||||
#include "port/atomics.h"
|
||||
#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"
|
||||
@ -62,6 +63,7 @@ typedef struct
|
||||
*/
|
||||
pid_t pid;
|
||||
WalRcvState walRcvState;
|
||||
ConditionVariable walRcvStoppedCV;
|
||||
pg_time_t startTime;
|
||||
|
||||
/*
|
||||
|
Loading…
Reference in New Issue
Block a user