#include "postgres.h"
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>
#include <signal.h>
#include "access/xlog_internal.h"
#include "pgstat.h"
#include "postmaster/startup.h"
#include "replication/walreceiver.h"
#include "storage/pmsignal.h"
#include "storage/shmem.h"
#include "utils/timestamp.h"

Include dependency graph for walreceiverfuncs.c:

Macros
#define	WALRCV_STARTUP_TIMEOUT 10

Functions
Size	WalRcvShmemSize (void)

void	WalRcvShmemInit (void)

bool	WalRcvRunning (void)

bool	WalRcvStreaming (void)

void	ShutdownWalRcv (void)

void	RequestXLogStreaming (TimeLineID tli, XLogRecPtr recptr, const char conninfo, const char slotname, bool create_temp_slot)

XLogRecPtr	GetWalRcvFlushRecPtr (XLogRecPtr latestChunkStart, TimeLineID receiveTLI)

XLogRecPtr	GetWalRcvWriteRecPtr (void)

int	GetReplicationApplyDelay (void)

int	GetReplicationTransferLatency (void)

Variables
WalRcvData *	WalRcv = NULL

Macro Definition Documentation

◆ WALRCV_STARTUP_TIMEOUT

#define WALRCV_STARTUP_TIMEOUT 10

Definition at line 39 of file walreceiverfuncs.c.

Referenced by WalRcvRunning(), and WalRcvStreaming().

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void )

Definition at line 364 of file walreceiverfuncs.c.

References WalRcvData::flushedUpto, GetCurrentChunkReplayStartTime(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     XLogRecPtr  receivePtr;
     XLogRecPtr  replayPtr;
     TimestampTz chunkReplayStartTime;
 
     SpinLockAcquire(&walrcv->mutex);
     receivePtr = walrcv->flushedUpto;
     SpinLockRelease(&walrcv->mutex);
 
     replayPtr = GetXLogReplayRecPtr(NULL);
 
     if (receivePtr == replayPtr)
         return 0;
 
     chunkReplayStartTime = GetCurrentChunkReplayStartTime();
 
     if (chunkReplayStartTime == 0)
         return -1;
 
     return TimestampDifferenceMilliseconds(chunkReplayStartTime,
                                            GetCurrentTimestamp());
 }

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void )

Definition at line 394 of file walreceiverfuncs.c.

References WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     TimestampTz lastMsgSendTime;
     TimestampTz lastMsgReceiptTime;
 
     SpinLockAcquire(&walrcv->mutex);
     lastMsgSendTime = walrcv->lastMsgSendTime;
     lastMsgReceiptTime = walrcv->lastMsgReceiptTime;
     SpinLockRelease(&walrcv->mutex);
 
     return TimestampDifferenceMilliseconds(lastMsgSendTime,
                                            lastMsgReceiptTime);
 }

◆ GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 331 of file walreceiverfuncs.c.

References WalRcvData::flushedUpto, WalRcvData::latestChunkStart, WalRcvData::mutex, WalRcvData::receivedTLI, SpinLockAcquire, SpinLockRelease, and WalRcv.

Referenced by CreateRestartPoint(), GetStandbyFlushRecPtr(), pg_last_wal_receive_lsn(), WaitForWALToBecomeAvailable(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     XLogRecPtr  recptr;
 
     SpinLockAcquire(&walrcv->mutex);
     recptr = walrcv->flushedUpto;
     if (latestChunkStart)
         *latestChunkStart = walrcv->latestChunkStart;
     if (receiveTLI)
         *receiveTLI = walrcv->receivedTLI;
     SpinLockRelease(&walrcv->mutex);
 
     return recptr;
 }

◆ GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( void )

Definition at line 352 of file walreceiverfuncs.c.

References pg_atomic_read_u64(), WalRcv, and WalRcvData::writtenUpto.

Referenced by walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
 
     return pg_atomic_read_u64(&walrcv->writtenUpto);
 }

◆ RequestXLogStreaming()

void RequestXLogStreaming	(	TimeLineID	tli,
		XLogRecPtr	recptr,
		const char *	conninfo,
		const char *	slotname,
		bool	create_temp_slot
	)

Definition at line 245 of file walreceiverfuncs.c.

References Assert, WalRcvData::conninfo, WalRcvData::flushedUpto, WalRcvData::is_temp_slot, WalRcvData::latch, WalRcvData::latestChunkStart, MAXCONNINFO, WalRcvData::mutex, NAMEDATALEN, now(), PMSIGNAL_START_WALRECEIVER, WalRcvData::receivedTLI, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, SendPostmasterSignal(), SetLatch(), WalRcvData::slotname, SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, strlcpy(), wal_segment_size, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_WAITING, WalRcvData::walRcvState, and XLogSegmentOffset.

Referenced by WaitForWALToBecomeAvailable(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     bool        launch = false;
     pg_time_t   now = (pg_time_t) time(NULL);
     Latch      *latch;
 
     /*
      * We always start at the beginning of the segment. That prevents a broken
      * segment (i.e., with no records in the first half of a segment) from
      * being created by XLOG streaming, which might cause trouble later on if
      * the segment is e.g archived.
      */
     if (XLogSegmentOffset(recptr, wal_segment_size) != 0)
         recptr -= XLogSegmentOffset(recptr, wal_segment_size);
 
     SpinLockAcquire(&walrcv->mutex);
 
     /* It better be stopped if we try to restart it */
     Assert(walrcv->walRcvState == WALRCV_STOPPED ||
            walrcv->walRcvState == WALRCV_WAITING);
 
     if (conninfo != NULL)
         strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
     else
         walrcv->conninfo[0] = '\0';
 
     /*
      * Use configured replication slot if present, and ignore the value of
      * create_temp_slot as the slot name should be persistent.  Otherwise, use
      * create_temp_slot to determine whether this WAL receiver should create a
      * temporary slot by itself and use it, or not.
      */
     if (slotname != NULL && slotname[0] != '\0')
     {
         strlcpy((char *) walrcv->slotname, slotname, NAMEDATALEN);
         walrcv->is_temp_slot = false;
     }
     else
     {
         walrcv->slotname[0] = '\0';
         walrcv->is_temp_slot = create_temp_slot;
     }
 
     if (walrcv->walRcvState == WALRCV_STOPPED)
     {
         launch = true;
         walrcv->walRcvState = WALRCV_STARTING;
     }
     else
         walrcv->walRcvState = WALRCV_RESTARTING;
     walrcv->startTime = now;
 
     /*
      * If this is the first startup of walreceiver (on this timeline),
      * initialize flushedUpto and latestChunkStart to the starting point.
      */
     if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli)
     {
         walrcv->flushedUpto = recptr;
         walrcv->receivedTLI = tli;
         walrcv->latestChunkStart = recptr;
     }
     walrcv->receiveStart = recptr;
     walrcv->receiveStartTLI = tli;
 
     latch = walrcv->latch;
 
     SpinLockRelease(&walrcv->mutex);
 
     if (launch)
         SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
     else if (latch)
         SetLatch(latch);
 }

◆ ShutdownWalRcv()

void ShutdownWalRcv ( void )

Definition at line 178 of file walreceiverfuncs.c.

References ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), kill, WalRcvData::mutex, WalRcvData::pid, SpinLockAcquire, SpinLockRelease, WAIT_EVENT_WAL_RECEIVER_EXIT, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, WalRcvRunning(), WalRcvData::walRcvState, and WalRcvData::walRcvStoppedCV.

Referenced by StartupXLOG(), WaitForWALToBecomeAvailable(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     pid_t       walrcvpid = 0;
     bool        stopped = false;
 
     /*
      * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
      * mode once it's finished, and will also request postmaster to not
      * restart itself.
      */
     SpinLockAcquire(&walrcv->mutex);
     switch (walrcv->walRcvState)
     {
         case WALRCV_STOPPED:
             break;
         case WALRCV_STARTING:
             walrcv->walRcvState = WALRCV_STOPPED;
             stopped = true;
             break;
 
         case WALRCV_STREAMING:
         case WALRCV_WAITING:
         case WALRCV_RESTARTING:
             walrcv->walRcvState = WALRCV_STOPPING;
             /* fall through */
         case WALRCV_STOPPING:
             walrcvpid = walrcv->pid;
             break;
     }
     SpinLockRelease(&walrcv->mutex);
 
     /* Unnecessary but consistent. */
     if (stopped)
         ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
 
     /*
      * Signal walreceiver process if it was still running.
      */
     if (walrcvpid != 0)
         kill(walrcvpid, SIGTERM);
 
     /*
      * Wait for walreceiver to acknowledge its death by setting state to
      * WALRCV_STOPPED.
      */
     ConditionVariablePrepareToSleep(&walrcv->walRcvStoppedCV);
     while (WalRcvRunning())
         ConditionVariableSleep(&walrcv->walRcvStoppedCV,
                                WAIT_EVENT_WAL_RECEIVER_EXIT);
     ConditionVariableCancelSleep();
 }

◆ WalRcvRunning()

bool WalRcvRunning ( void )

Definition at line 75 of file walreceiverfuncs.c.

References ConditionVariableBroadcast(), WalRcvData::mutex, now(), SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, WalRcv, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, WalRcvData::walRcvState, and WalRcvData::walRcvStoppedCV.

Referenced by ShutdownWalRcv(), StartupRequestWalReceiverRestart(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     WalRcvState state;
     pg_time_t   startTime;
 
     SpinLockAcquire(&walrcv->mutex);
 
     state = walrcv->walRcvState;
     startTime = walrcv->startTime;
 
     SpinLockRelease(&walrcv->mutex);
 
     /*
      * If it has taken too long for walreceiver to start up, give up. Setting
      * the state to STOPPED ensures that if walreceiver later does start up
      * after all, it will see that it's not supposed to be running and die
      * without doing anything.
      */
     if (state == WALRCV_STARTING)
     {
         pg_time_t   now = (pg_time_t) time(NULL);
 
         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);
         }
     }
 
     if (state != WALRCV_STOPPED)
         return true;
     else
         return false;
 }

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void )

Definition at line 54 of file walreceiverfuncs.c.

References ConditionVariableInit(), WalRcvData::latch, MemSet, WalRcvData::mutex, pg_atomic_init_u64(), ShmemInitStruct(), SpinLockInit, WALRCV_STOPPED, WalRcvShmemSize(), WalRcvData::walRcvState, WalRcvData::walRcvStoppedCV, and WalRcvData::writtenUpto.

Referenced by CreateSharedMemoryAndSemaphores(), and walrcv_clear_result().

 {
     bool        found;
 
     WalRcv = (WalRcvData *)
         ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &found);
 
     if (!found)
     {
         /* 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;
     }
 }

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void )

Definition at line 43 of file walreceiverfuncs.c.

References add_size().

Referenced by CreateSharedMemoryAndSemaphores(), walrcv_clear_result(), and WalRcvShmemInit().

 {
     Size        size = 0;
 
     size = add_size(size, sizeof(WalRcvData));
 
     return size;
 }

◆ WalRcvStreaming()

bool WalRcvStreaming ( void )

Definition at line 126 of file walreceiverfuncs.c.

References ConditionVariableBroadcast(), WalRcvData::mutex, now(), SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, WALRCV_STREAMING, WalRcvData::walRcvState, and WalRcvData::walRcvStoppedCV.

Referenced by StartupXLOG(), WaitForWALToBecomeAvailable(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     WalRcvState state;
     pg_time_t   startTime;
 
     SpinLockAcquire(&walrcv->mutex);
 
     state = walrcv->walRcvState;
     startTime = walrcv->startTime;
 
     SpinLockRelease(&walrcv->mutex);
 
     /*
      * If it has taken too long for walreceiver to start up, give up. Setting
      * the state to STOPPED ensures that if walreceiver later does start up
      * after all, it will see that it's not supposed to be running and die
      * without doing anything.
      */
     if (state == WALRCV_STARTING)
     {
         pg_time_t   now = (pg_time_t) time(NULL);
 
         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);
         }
     }
 
     if (state == WALRCV_STREAMING || state == WALRCV_STARTING ||
         state == WALRCV_RESTARTING)
         return true;
     else
         return false;
 }

Variable Documentation

◆ WalRcv

WalRcvData* WalRcv = NULL

Definition at line 33 of file walreceiverfuncs.c.

Referenced by GetReplicationApplyDelay(), GetReplicationTransferLatency(), GetWalRcvFlushRecPtr(), GetWalRcvWriteRecPtr(), pg_stat_get_wal_receiver(), ProcessWalSndrMessage(), RequestXLogStreaming(), ShutdownWalRcv(), WalRcvDie(), WalRcvForceReply(), WalRcvRunning(), WalRcvStreaming(), WalRcvWaitForStartPosition(), WalReceiverMain(), XLogWalRcvFlush(), and XLogWalRcvWrite().

Macros

Functions

Variables