walreceiverfuncs.c File Reference

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

Go to the source code of this file.

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, 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 378 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 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(), WalRcvStreaming(), WalRcvWaitForStartPosition(), WalReceiverMain(), XLogWalRcvFlush(), and XLogWalRcvWrite().