#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 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 40 of file walreceiverfuncs.c.

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void )

Definition at line 379 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 WalRcvData::flushedUpto, GetCurrentChunkReplayStartTime(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void )

Definition at line 409 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 WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

◆ GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 346 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 WalRcvData::flushedUpto, WalRcvData::latestChunkStart, WalRcvData::mutex, WalRcvData::receivedTLI, receiveTLI, SpinLockAcquire, SpinLockRelease, and WalRcv.

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

◆ GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( void )

Definition at line 367 of file walreceiverfuncs.c.

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

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

◆ RequestXLogStreaming()

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

Definition at line 260 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(), WalRcvData::conninfo, WalRcvData::flushedUpto, GetPGProcByNumber, INVALID_PROC_NUMBER, WalRcvData::is_temp_slot, WalRcvData::latestChunkStart, MAXCONNINFO, WalRcvData::mutex, NAMEDATALEN, now(), PMSIGNAL_START_WALRECEIVER, WalRcvData::procno, 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, XLogRecPtrIsValid, and XLogSegmentOffset.

Referenced by WaitForWALToBecomeAvailable().

◆ ShutdownWalRcv()

void ShutdownWalRcv ( void )

Definition at line 193 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_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();
}

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

Referenced by XLogShutdownWalRcv().

◆ WalRcvGetState()

WalRcvState WalRcvGetState ( void )

Definition at line 124 of file walreceiverfuncs.c.

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

References WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, WalRcv, and WalRcvData::walRcvState.

Referenced by WaitForWALToBecomeAvailable().

◆ WalRcvRunning()

bool WalRcvRunning ( void )

Definition at line 76 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(), WalRcvData::mutex, now(), SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, WalRcv, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, WalRcvData::walRcvState, and WalRcvData::walRcvStoppedCV.

Referenced by ShutdownWalRcv(), and StartupRequestWalReceiverRestart().

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void )

Definition at line 55 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 44 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 141 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_RESTARTING)
        return true;
    else
        return false;
}

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

Variable Documentation

◆ WalRcv

WalRcvData* WalRcv = NULL

Macros

Functions

Variables