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

Include dependency graph for walreceiverfuncs.c:

Macros
#define	WALRCV_STARTUP_TIMEOUT 10

Functions
Size	WalRcvShmemSize (void)

void	WalRcvShmemInit (void)

bool	WalRcvRunning (void)

WalRcvState	WalRcvGetState (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 41 of file walreceiverfuncs.c.

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void )

Definition at line 381 of file walreceiverfuncs.c.

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

References fb(), GetCurrentChunkReplayStartTime(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), SpinLockAcquire(), SpinLockRelease(), TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void )

Definition at line 411 of file walreceiverfuncs.c.

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

References fb(), SpinLockAcquire(), SpinLockRelease(), TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

◆ GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 348 of file walreceiverfuncs.c.

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

References fb(), receiveTLI, SpinLockAcquire(), SpinLockRelease(), and WalRcv.

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

◆ GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( void )

Definition at line 369 of file walreceiverfuncs.c.

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

References fb(), pg_atomic_read_u64(), and WalRcv.

Referenced by GetCurrentLSNForWaitType().

◆ RequestXLogStreaming()

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

Definition at line 262 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    bool        launch = false;
    pg_time_t   now = (pg_time_t) time(NULL);
    ProcNumber  walrcv_proc;
 
    /*
     * 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(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(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 (!XLogRecPtrIsValid(walrcv->receiveStart) || walrcv->receivedTLI != tli)
    {
        walrcv->flushedUpto = recptr;
        walrcv->receivedTLI = tli;
        walrcv->latestChunkStart = recptr;
    }
    walrcv->receiveStart = recptr;
    walrcv->receiveStartTLI = tli;
 
    walrcv_proc = walrcv->procno;
 
    SpinLockRelease(&walrcv->mutex);
 
    if (launch)
        SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
    else if (walrcv_proc != INVALID_PROC_NUMBER)
        SetLatch(&GetPGProcByNumber(walrcv_proc)->procLatch);
}

References Assert, fb(), GetPGProcByNumber, INVALID_PROC_NUMBER, MAXCONNINFO, NAMEDATALEN, now(), PMSIGNAL_START_WALRECEIVER, SendPostmasterSignal(), SetLatch(), SpinLockAcquire(), SpinLockRelease(), strlcpy(), wal_segment_size, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_WAITING, XLogRecPtrIsValid, and XLogSegmentOffset.

Referenced by WaitForWALToBecomeAvailable().

◆ ShutdownWalRcv()

void ShutdownWalRcv ( void )

Definition at line 194 of file walreceiverfuncs.c.

{
    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_CONNECTING:
        case WALRCV_STREAMING:
        case WALRCV_WAITING:
        case WALRCV_RESTARTING:
            walrcv->walRcvState = WALRCV_STOPPING;
            pg_fallthrough;
        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();
}

References ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), fb(), kill, pg_fallthrough, SpinLockAcquire(), SpinLockRelease(), WalRcv, WALRCV_CONNECTING, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, and WalRcvRunning().

Referenced by XLogShutdownWalRcv().

◆ WalRcvGetState()

WalRcvState WalRcvGetState ( void )

Definition at line 125 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    WalRcvState state;
 
    SpinLockAcquire(&walrcv->mutex);
    state = walrcv->walRcvState;
    SpinLockRelease(&walrcv->mutex);
 
    return state;
}

References fb(), SpinLockAcquire(), SpinLockRelease(), and WalRcv.

Referenced by WaitForWALToBecomeAvailable().

◆ WalRcvRunning()

bool WalRcvRunning ( void )

Definition at line 77 of file walreceiverfuncs.c.

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

References ConditionVariableBroadcast(), fb(), now(), SpinLockAcquire(), SpinLockRelease(), WalRcv, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, and WALRCV_STOPPED.

Referenced by ShutdownWalRcv(), and StartupRequestWalReceiverRestart().

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void )

Definition at line 56 of file walreceiverfuncs.c.

{
    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->procno = INVALID_PROC_NUMBER;
    }
}

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

Referenced by CreateOrAttachShmemStructs().

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void )

Definition at line 45 of file walreceiverfuncs.c.

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

References add_size().

Referenced by CalculateShmemSize(), and WalRcvShmemInit().

◆ WalRcvStreaming()

bool WalRcvStreaming ( void )

Definition at line 142 of file walreceiverfuncs.c.

{
    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_CONNECTING || state == WALRCV_RESTARTING)
        return true;
    else
        return false;
}

References ConditionVariableBroadcast(), fb(), now(), SpinLockAcquire(), SpinLockRelease(), WalRcv, WALRCV_CONNECTING, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, and WALRCV_STREAMING.

Referenced by FinishWalRecovery(), and WaitForWALToBecomeAvailable().

Macros

Functions

Variables