#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)

XLogRecPtr	GetWalRcvWriteRecPtr (XLogRecPtr latestChunkStart, TimeLineID receiveTLI)

int	GetReplicationApplyDelay (void)

int	GetReplicationTransferLatency (void)

Variables
WalRcvData *	WalRcv = NULL

Macro Definition Documentation

#define WALRCV_STARTUP_TIMEOUT 10

Definition at line 38 of file walreceiverfuncs.c.

Referenced by WalRcvRunning(), and WalRcvStreaming().

Function Documentation

int GetReplicationApplyDelay ( void )

Definition at line 315 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage().

 {
     WalRcvData *walrcv = WalRcv;
     XLogRecPtr  receivePtr;
     XLogRecPtr  replayPtr;
 
     long        secs;
     int         usecs;
 
     TimestampTz chunkReplayStartTime;
 
     SpinLockAcquire(&walrcv->mutex);
     receivePtr = walrcv->receivedUpto;
     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));
 }

int GetReplicationTransferLatency ( void )

Definition at line 352 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage().

 {
     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;
 }

XLogRecPtr GetWalRcvWriteRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 294 of file walreceiverfuncs.c.

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

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

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

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

Definition at line 223 of file walreceiverfuncs.c.

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

Referenced by WaitForWALToBecomeAvailable().

 {
     WalRcvData *walrcv = WalRcv;
     bool        launch = false;
     pg_time_t   now = (pg_time_t) time(NULL);
 
     /*
      * 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 (recptr % XLogSegSize != 0)
         recptr -= recptr % XLogSegSize;
 
     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';
 
     if (slotname != NULL)
         strlcpy((char *) walrcv->slotname, slotname, NAMEDATALEN);
     else
         walrcv->slotname[0] = '\0';
 
     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 receivedUpto and latestChunkStart to the starting point.
      */
     if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli)
     {
         walrcv->receivedUpto = recptr;
         walrcv->receivedTLI = tli;
         walrcv->latestChunkStart = recptr;
     }
     walrcv->receiveStart = recptr;
     walrcv->receiveStartTLI = tli;
 
     SpinLockRelease(&walrcv->mutex);
 
     if (launch)
         SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
     else if (walrcv->latch)
         SetLatch(walrcv->latch);
 }

void ShutdownWalRcv ( void )

Definition at line 163 of file walreceiverfuncs.c.

References HandleStartupProcInterrupts(), 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(), and WaitForWALToBecomeAvailable().

 {
     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 */
     }
 }

bool WalRcvRunning ( void )

Definition at line 72 of file walreceiverfuncs.c.

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

Referenced by ShutdownWalRcv().

 {
     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;
 }

void WalRcvShmemInit ( void )

Definition at line 53 of file walreceiverfuncs.c.

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

Referenced by CreateSharedMemoryAndSemaphores().

 {
     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;
     }
 }

Size WalRcvShmemSize ( void )

Definition at line 42 of file walreceiverfuncs.c.

References add_size().

Referenced by CreateSharedMemoryAndSemaphores(), and WalRcvShmemInit().

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

bool WalRcvStreaming ( void )

Definition at line 117 of file walreceiverfuncs.c.

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

Referenced by WaitForWALToBecomeAvailable().

 {
     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

WalRcvData* WalRcv = NULL

Definition at line 32 of file walreceiverfuncs.c.

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

Macros

Functions

Variables

Macro Definition Documentation

Function Documentation

Variable Documentation