#include "postgres.h"
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>
#include <signal.h>
#include "access/xlog_internal.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 38 of file walreceiverfuncs.c.

Referenced by WalRcvRunning(), and WalRcvStreaming().

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void )

Definition at line 348 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
     XLogRecPtr  receivePtr;
     XLogRecPtr  replayPtr;
 
     long        secs;
     int         usecs;
 
     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;
 
     TimestampDifference(chunkReplayStartTime,
                         GetCurrentTimestamp(),
                         &secs, &usecs);
 
     return (((int) secs * 1000) + (usecs / 1000));
 }

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void )

Definition at line 385 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

 {
     WalRcvData *walrcv = WalRcv;
 
     TimestampTz lastMsgSendTime;
     TimestampTz lastMsgReceiptTime;
 
     long        secs = 0;
     int         usecs = 0;
     int         ms;
 
     SpinLockAcquire(&walrcv->mutex);
     lastMsgSendTime = walrcv->lastMsgSendTime;
     lastMsgReceiptTime = walrcv->lastMsgReceiptTime;
     SpinLockRelease(&walrcv->mutex);
 
     TimestampDifference(lastMsgSendTime,
                         lastMsgReceiptTime,
                         &secs, &usecs);
 
     ms = ((int) secs * 1000) + (usecs / 1000);
 
     return ms;
 }

◆ GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 315 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 336 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 229 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 163 of file walreceiverfuncs.c.

References HandleStartupProcInterrupts(), kill, WalRcvData::mutex, pg_usleep(), WalRcvData::pid, SpinLockAcquire, SpinLockRelease, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, WalRcvRunning(), and WalRcvData::walRcvState.

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

 {
     WalRcvData *walrcv = WalRcv;
     pid_t       walrcvpid = 0;
 
     /*
      * 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;
             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);
 
     /*
      * 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.
      */
     while (WalRcvRunning())
     {
         /*
          * This possibly-long loop needs to handle interrupts of startup
          * process.
          */
         HandleStartupProcInterrupts();
 
         pg_usleep(100000);      /* 100ms */
     }
 }

◆ WalRcvRunning()

bool WalRcvRunning ( void )

Definition at line 72 of file walreceiverfuncs.c.

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

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)
         {
             SpinLockAcquire(&walrcv->mutex);
 
             if (walrcv->walRcvState == WALRCV_STARTING)
                 state = walrcv->walRcvState = WALRCV_STOPPED;
 
             SpinLockRelease(&walrcv->mutex);
         }
     }
 
     if (state != WALRCV_STOPPED)
         return true;
     else
         return false;
 }

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void )

Definition at line 53 of file walreceiverfuncs.c.

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

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;
         SpinLockInit(&WalRcv->mutex);
         WalRcv->latch = NULL;
     }
 }

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void )

Definition at line 42 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 117 of file walreceiverfuncs.c.

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

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)
         {
             SpinLockAcquire(&walrcv->mutex);
 
             if (walrcv->walRcvState == WALRCV_STARTING)
                 state = walrcv->walRcvState = WALRCV_STOPPED;
 
             SpinLockRelease(&walrcv->mutex);
         }
     }
 
     if (state == WALRCV_STREAMING || state == WALRCV_STARTING ||
         state == WALRCV_RESTARTING)
         return true;
     else
         return false;
 }

Variable Documentation

◆ WalRcv

WalRcvData* WalRcv = NULL

Definition at line 32 of file walreceiverfuncs.c.

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

Macros

Functions

Variables