#include "postgres.h"
#include <unistd.h>
#include "access/htup_details.h"
#include "access/timeline.h"
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "access/xlogarchive.h"
#include "access/xlogrecovery.h"
#include "access/xlogwait.h"
#include "catalog/pg_authid.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/auxprocess.h"
#include "postmaster/interrupt.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/pg_lsn.h"
#include "utils/ps_status.h"
#include "utils/timestamp.h"

Include dependency graph for walreceiver.c:

Macros
#define	NUM_WALRCV_WAKEUPS (WALRCV_WAKEUP_HSFEEDBACK + 1)

Typedefs
typedef enum WalRcvWakeupReason	WalRcvWakeupReason

Enumerations
enum	WalRcvWakeupReason { WALRCV_WAKEUP_TERMINATE , WALRCV_WAKEUP_PING , WALRCV_WAKEUP_REPLY , WALRCV_WAKEUP_HSFEEDBACK }

Functions
static void	WalRcvFetchTimeLineHistoryFiles (TimeLineID first, TimeLineID last)

static void	WalRcvWaitForStartPosition (XLogRecPtr startpoint, TimeLineID startpointTLI)

static void	WalRcvDie (int code, Datum arg)

static void	XLogWalRcvProcessMsg (unsigned char type, char *buf, Size len, TimeLineID tli)

static void	XLogWalRcvWrite (char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli)

static void	XLogWalRcvFlush (bool dying, TimeLineID tli)

static void	XLogWalRcvClose (XLogRecPtr recptr, TimeLineID tli)

static void	XLogWalRcvSendReply (bool force, bool requestReply)

static void	XLogWalRcvSendHSFeedback (bool immed)

static void	ProcessWalSndrMessage (XLogRecPtr walEnd, TimestampTz sendTime)

static void	WalRcvComputeNextWakeup (WalRcvWakeupReason reason, TimestampTz now)

void	WalReceiverMain (const void *startup_data, size_t startup_data_len)

void	WalRcvForceReply (void)

static const char *	WalRcvGetStateString (WalRcvState state)

Datum	pg_stat_get_wal_receiver (PG_FUNCTION_ARGS)

Variables
int	wal_receiver_status_interval

int	wal_receiver_timeout

bool	hot_standby_feedback

static WalReceiverConn *	wrconn = NULL

WalReceiverFunctionsType *	WalReceiverFunctions = NULL

static int	recvFile = -1

static TimeLineID	recvFileTLI = 0

static XLogSegNo	recvSegNo = 0

struct {

XLogRecPtr Write

XLogRecPtr Flush

}	LogstreamResult

static TimestampTz	wakeup [NUM_WALRCV_WAKEUPS]

static StringInfoData	reply_message

Macro Definition Documentation

◆ NUM_WALRCV_WAKEUPS

#define NUM_WALRCV_WAKEUPS (WALRCV_WAKEUP_HSFEEDBACK + 1)

Definition at line 125 of file walreceiver.c.

Typedef Documentation

◆ WalRcvWakeupReason

typedef enum WalRcvWakeupReason WalRcvWakeupReason

Enumeration Type Documentation

◆ WalRcvWakeupReason

enum WalRcvWakeupReason

Enumerator
WALRCV_WAKEUP_TERMINATE
WALRCV_WAKEUP_PING
WALRCV_WAKEUP_REPLY
WALRCV_WAKEUP_HSFEEDBACK

Definition at line 119 of file walreceiver.c.

{
    WALRCV_WAKEUP_TERMINATE,
    WALRCV_WAKEUP_PING,
    WALRCV_WAKEUP_REPLY,
    WALRCV_WAKEUP_HSFEEDBACK,
#define NUM_WALRCV_WAKEUPS (WALRCV_WAKEUP_HSFEEDBACK + 1)
} WalRcvWakeupReason;

Function Documentation

◆ pg_stat_get_wal_receiver()

Datum pg_stat_get_wal_receiver ( PG_FUNCTION_ARGS )

Definition at line 1425 of file walreceiver.c.

{
    TupleDesc   tupdesc;
    Datum      *values;
    bool       *nulls;
    int         pid;
    bool        ready_to_display;
    WalRcvState state;
    XLogRecPtr  receive_start_lsn;
    TimeLineID  receive_start_tli;
    XLogRecPtr  written_lsn;
    XLogRecPtr  flushed_lsn;
    TimeLineID  received_tli;
    TimestampTz last_send_time;
    TimestampTz last_receipt_time;
    XLogRecPtr  latest_end_lsn;
    TimestampTz latest_end_time;
    char        sender_host[NI_MAXHOST];
    int         sender_port = 0;
    char        slotname[NAMEDATALEN];
    char        conninfo[MAXCONNINFO];
 
    /* Take a lock to ensure value consistency */
    SpinLockAcquire(&WalRcv->mutex);
    pid = (int) WalRcv->pid;
    ready_to_display = WalRcv->ready_to_display;
    state = WalRcv->walRcvState;
    receive_start_lsn = WalRcv->receiveStart;
    receive_start_tli = WalRcv->receiveStartTLI;
    flushed_lsn = WalRcv->flushedUpto;
    received_tli = WalRcv->receivedTLI;
    last_send_time = WalRcv->lastMsgSendTime;
    last_receipt_time = WalRcv->lastMsgReceiptTime;
    latest_end_lsn = WalRcv->latestWalEnd;
    latest_end_time = WalRcv->latestWalEndTime;
    strlcpy(slotname, WalRcv->slotname, sizeof(slotname));
    strlcpy(sender_host, WalRcv->sender_host, sizeof(sender_host));
    sender_port = WalRcv->sender_port;
    strlcpy(conninfo, WalRcv->conninfo, sizeof(conninfo));
    SpinLockRelease(&WalRcv->mutex);
 
    /*
     * No WAL receiver (or not ready yet), just return a tuple with NULL
     * values
     */
    if (pid == 0 || !ready_to_display)
        PG_RETURN_NULL();
 
    /*
     * Read "writtenUpto" without holding a spinlock.  Note that it may not be
     * consistent with the other shared variables of the WAL receiver
     * protected by a spinlock, but this should not be used for data integrity
     * checks.
     */
    written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);
 
    /* determine result type */
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    values = palloc0_array(Datum, tupdesc->natts);
    nulls = palloc0_array(bool, tupdesc->natts);
 
    /* Fetch values */
    values[0] = Int32GetDatum(pid);
 
    if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
    {
        /*
         * Only superusers and roles with privileges of pg_read_all_stats can
         * see details. Other users only get the pid value to know whether it
         * is a WAL receiver, but no details.
         */
        memset(&nulls[1], true, sizeof(bool) * (tupdesc->natts - 1));
    }
    else
    {
        values[1] = CStringGetTextDatum(WalRcvGetStateString(state));
 
        if (!XLogRecPtrIsValid(receive_start_lsn))
            nulls[2] = true;
        else
            values[2] = LSNGetDatum(receive_start_lsn);
        values[3] = Int32GetDatum(receive_start_tli);
        if (!XLogRecPtrIsValid(written_lsn))
            nulls[4] = true;
        else
            values[4] = LSNGetDatum(written_lsn);
        if (!XLogRecPtrIsValid(flushed_lsn))
            nulls[5] = true;
        else
            values[5] = LSNGetDatum(flushed_lsn);
        values[6] = Int32GetDatum(received_tli);
        if (last_send_time == 0)
            nulls[7] = true;
        else
            values[7] = TimestampTzGetDatum(last_send_time);
        if (last_receipt_time == 0)
            nulls[8] = true;
        else
            values[8] = TimestampTzGetDatum(last_receipt_time);
        if (!XLogRecPtrIsValid(latest_end_lsn))
            nulls[9] = true;
        else
            values[9] = LSNGetDatum(latest_end_lsn);
        if (latest_end_time == 0)
            nulls[10] = true;
        else
            values[10] = TimestampTzGetDatum(latest_end_time);
        if (*slotname == '\0')
            nulls[11] = true;
        else
            values[11] = CStringGetTextDatum(slotname);
        if (*sender_host == '\0')
            nulls[12] = true;
        else
            values[12] = CStringGetTextDatum(sender_host);
        if (sender_port == 0)
            nulls[13] = true;
        else
            values[13] = Int32GetDatum(sender_port);
        if (*conninfo == '\0')
            nulls[14] = true;
        else
            values[14] = CStringGetTextDatum(conninfo);
    }
 
    /* Returns the record as Datum */
    PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

References WalRcvData::conninfo, CStringGetTextDatum, elog, ERROR, fb(), WalRcvData::flushedUpto, get_call_result_type(), GetUserId(), has_privs_of_role(), heap_form_tuple(), HeapTupleGetDatum(), Int32GetDatum(), WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::latestWalEnd, WalRcvData::latestWalEndTime, LSNGetDatum(), MAXCONNINFO, WalRcvData::mutex, NAMEDATALEN, TupleDescData::natts, palloc0_array, pg_atomic_read_u64(), PG_RETURN_DATUM, PG_RETURN_NULL, WalRcvData::pid, WalRcvData::ready_to_display, WalRcvData::receivedTLI, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, WalRcvData::sender_host, WalRcvData::sender_port, WalRcvData::slotname, SpinLockAcquire, SpinLockRelease, strlcpy(), TimestampTzGetDatum(), TYPEFUNC_COMPOSITE, values, WalRcv, WalRcvGetStateString(), WalRcvData::walRcvState, WalRcvData::writtenUpto, and XLogRecPtrIsValid.

◆ ProcessWalSndrMessage()

static void ProcessWalSndrMessage	(	XLogRecPtr	walEnd,
		TimestampTz	sendTime
	)

static

Definition at line 1287 of file walreceiver.c.

{
    WalRcvData *walrcv = WalRcv;
    TimestampTz lastMsgReceiptTime = GetCurrentTimestamp();
 
    /* Update shared-memory status */
    SpinLockAcquire(&walrcv->mutex);
    if (walrcv->latestWalEnd < walEnd)
        walrcv->latestWalEndTime = sendTime;
    walrcv->latestWalEnd = walEnd;
    walrcv->lastMsgSendTime = sendTime;
    walrcv->lastMsgReceiptTime = lastMsgReceiptTime;
    SpinLockRelease(&walrcv->mutex);
 
    if (message_level_is_interesting(DEBUG2))
    {
        char       *sendtime;
        char       *receipttime;
        int         applyDelay;
 
        /* Copy because timestamptz_to_str returns a static buffer */
        sendtime = pstrdup(timestamptz_to_str(sendTime));
        receipttime = pstrdup(timestamptz_to_str(lastMsgReceiptTime));
        applyDelay = GetReplicationApplyDelay();
 
        /* apply delay is not available */
        if (applyDelay == -1)
            elog(DEBUG2, "sendtime %s receipttime %s replication apply delay (N/A) transfer latency %d ms",
                 sendtime,
                 receipttime,
                 GetReplicationTransferLatency());
        else
            elog(DEBUG2, "sendtime %s receipttime %s replication apply delay %d ms transfer latency %d ms",
                 sendtime,
                 receipttime,
                 applyDelay,
                 GetReplicationTransferLatency());
 
        pfree(sendtime);
        pfree(receipttime);
    }
}

References DEBUG2, elog, fb(), GetCurrentTimestamp(), GetReplicationApplyDelay(), GetReplicationTransferLatency(), message_level_is_interesting(), pfree(), pstrdup(), SpinLockAcquire, SpinLockRelease, timestamptz_to_str(), and WalRcv.

Referenced by XLogWalRcvProcessMsg().

◆ WalRcvComputeNextWakeup()

static void WalRcvComputeNextWakeup	(	WalRcvWakeupReason	reason,
		TimestampTz	now
	)

static

Definition at line 1339 of file walreceiver.c.

{
    switch (reason)
    {
        case WALRCV_WAKEUP_TERMINATE:
            if (wal_receiver_timeout <= 0)
                wakeup[reason] = TIMESTAMP_INFINITY;
            else
                wakeup[reason] = TimestampTzPlusMilliseconds(now, wal_receiver_timeout);
            break;
        case WALRCV_WAKEUP_PING:
            if (wal_receiver_timeout <= 0)
                wakeup[reason] = TIMESTAMP_INFINITY;
            else
                wakeup[reason] = TimestampTzPlusMilliseconds(now, wal_receiver_timeout / 2);
            break;
        case WALRCV_WAKEUP_HSFEEDBACK:
            if (!hot_standby_feedback || wal_receiver_status_interval <= 0)
                wakeup[reason] = TIMESTAMP_INFINITY;
            else
                wakeup[reason] = TimestampTzPlusSeconds(now, wal_receiver_status_interval);
            break;
        case WALRCV_WAKEUP_REPLY:
            if (wal_receiver_status_interval <= 0)
                wakeup[reason] = TIMESTAMP_INFINITY;
            else
                wakeup[reason] = TimestampTzPlusSeconds(now, wal_receiver_status_interval);
            break;
            /* there's intentionally no default: here */
    }
}

References hot_standby_feedback, now(), TIMESTAMP_INFINITY, TimestampTzPlusMilliseconds, TimestampTzPlusSeconds, wakeup, wal_receiver_status_interval, wal_receiver_timeout, WALRCV_WAKEUP_HSFEEDBACK, WALRCV_WAKEUP_PING, WALRCV_WAKEUP_REPLY, and WALRCV_WAKEUP_TERMINATE.

Referenced by WalReceiverMain(), XLogWalRcvSendHSFeedback(), and XLogWalRcvSendReply().

◆ WalRcvDie()

static void WalRcvDie	(	int	code,
		Datum	arg
	)

static

Definition at line 793 of file walreceiver.c.

{
    WalRcvData *walrcv = WalRcv;
    TimeLineID *startpointTLI_p = (TimeLineID *) DatumGetPointer(arg);
 
    Assert(*startpointTLI_p != 0);
 
    /* Ensure that all WAL records received are flushed to disk */
    XLogWalRcvFlush(true, *startpointTLI_p);
 
    /* Mark ourselves inactive in shared memory */
    SpinLockAcquire(&walrcv->mutex);
    Assert(walrcv->walRcvState == WALRCV_STREAMING ||
           walrcv->walRcvState == WALRCV_CONNECTING ||
           walrcv->walRcvState == WALRCV_RESTARTING ||
           walrcv->walRcvState == WALRCV_STARTING ||
           walrcv->walRcvState == WALRCV_WAITING ||
           walrcv->walRcvState == WALRCV_STOPPING);
    Assert(walrcv->pid == MyProcPid);
    walrcv->walRcvState = WALRCV_STOPPED;
    walrcv->pid = 0;
    walrcv->procno = INVALID_PROC_NUMBER;
    walrcv->ready_to_display = false;
    SpinLockRelease(&walrcv->mutex);
 
    ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
 
    /* Terminate the connection gracefully. */
    if (wrconn != NULL)
        walrcv_disconnect(wrconn);
 
    /* Wake up the startup process to notice promptly that we're gone */
    WakeupRecovery();
}

References arg, Assert, ConditionVariableBroadcast(), DatumGetPointer(), fb(), INVALID_PROC_NUMBER, MyProcPid, SpinLockAcquire, SpinLockRelease, WakeupRecovery(), WalRcv, WALRCV_CONNECTING, walrcv_disconnect, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, wrconn, and XLogWalRcvFlush().

Referenced by WalReceiverMain().

◆ WalRcvFetchTimeLineHistoryFiles()

static void WalRcvFetchTimeLineHistoryFiles	(	TimeLineID	first,
		TimeLineID	last
	)

static

Definition at line 737 of file walreceiver.c.

{
    TimeLineID  tli;
 
    for (tli = first; tli <= last; tli++)
    {
        /* there's no history file for timeline 1 */
        if (tli != 1 && !existsTimeLineHistory(tli))
        {
            char       *fname;
            char       *content;
            int         len;
            char        expectedfname[MAXFNAMELEN];
 
            ereport(LOG,
                    (errmsg("fetching timeline history file for timeline %u from primary server",
                            tli)));
 
            walrcv_readtimelinehistoryfile(wrconn, tli, &fname, &content, &len);
 
            /*
             * Check that the filename on the primary matches what we
             * calculated ourselves. This is just a sanity check, it should
             * always match.
             */
            TLHistoryFileName(expectedfname, tli);
            if (strcmp(fname, expectedfname) != 0)
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg_internal("primary reported unexpected file name for timeline history file of timeline %u",
                                         tli)));
 
            /*
             * Write the file to pg_wal.
             */
            writeTimeLineHistoryFile(tli, content, len);
 
            /*
             * Mark the streamed history file as ready for archiving if
             * archive_mode is always.
             */
            if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
                XLogArchiveForceDone(fname);
            else
                XLogArchiveNotify(fname);
 
            pfree(fname);
            pfree(content);
        }
    }
}

References ARCHIVE_MODE_ALWAYS, ereport, errcode(), errmsg(), errmsg_internal(), ERROR, existsTimeLineHistory(), fb(), len, LOG, MAXFNAMELEN, pfree(), TLHistoryFileName(), walrcv_readtimelinehistoryfile, wrconn, writeTimeLineHistoryFile(), XLogArchiveForceDone(), XLogArchiveMode, and XLogArchiveNotify().

Referenced by WalReceiverMain().

◆ WalRcvForceReply()

void WalRcvForceReply ( void )

Definition at line 1380 of file walreceiver.c.

{
    ProcNumber  procno;
 
    WalRcv->force_reply = true;
    /* fetching the proc number is probably atomic, but don't rely on it */
    SpinLockAcquire(&WalRcv->mutex);
    procno = WalRcv->procno;
    SpinLockRelease(&WalRcv->mutex);
    if (procno != INVALID_PROC_NUMBER)
        SetLatch(&GetPGProcByNumber(procno)->procLatch);
}

References WalRcvData::force_reply, GetPGProcByNumber, INVALID_PROC_NUMBER, WalRcvData::mutex, WalRcvData::procno, SetLatch(), SpinLockAcquire, SpinLockRelease, and WalRcv.

Referenced by ApplyWalRecord(), and WaitForWALToBecomeAvailable().

◆ WalRcvGetStateString()

static const char * WalRcvGetStateString ( WalRcvState state )

static

Definition at line 1398 of file walreceiver.c.

{
    switch (state)
    {
        case WALRCV_STOPPED:
            return "stopped";
        case WALRCV_STARTING:
            return "starting";
        case WALRCV_CONNECTING:
            return "connecting";
        case WALRCV_STREAMING:
            return "streaming";
        case WALRCV_WAITING:
            return "waiting";
        case WALRCV_RESTARTING:
            return "restarting";
        case WALRCV_STOPPING:
            return "stopping";
    }
    return "UNKNOWN";
}

References WALRCV_CONNECTING, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, and WALRCV_WAITING.

Referenced by pg_stat_get_wal_receiver().

◆ WalRcvWaitForStartPosition()

static void WalRcvWaitForStartPosition	(	XLogRecPtr *	startpoint,
		TimeLineID *	startpointTLI
	)

static

Definition at line 657 of file walreceiver.c.

{
    WalRcvData *walrcv = WalRcv;
    int         state;
 
    SpinLockAcquire(&walrcv->mutex);
    state = walrcv->walRcvState;
    if (state != WALRCV_STREAMING && state != WALRCV_CONNECTING)
    {
        SpinLockRelease(&walrcv->mutex);
        if (state == WALRCV_STOPPING)
            proc_exit(0);
        else
            elog(FATAL, "unexpected walreceiver state");
    }
    walrcv->walRcvState = WALRCV_WAITING;
    walrcv->receiveStart = InvalidXLogRecPtr;
    walrcv->receiveStartTLI = 0;
    SpinLockRelease(&walrcv->mutex);
 
    set_ps_display("idle");
 
    /*
     * nudge startup process to notice that we've stopped streaming and are
     * now waiting for instructions.
     */
    WakeupRecovery();
    for (;;)
    {
        ResetLatch(MyLatch);
 
        CHECK_FOR_INTERRUPTS();
 
        SpinLockAcquire(&walrcv->mutex);
        Assert(walrcv->walRcvState == WALRCV_RESTARTING ||
               walrcv->walRcvState == WALRCV_WAITING ||
               walrcv->walRcvState == WALRCV_STOPPING);
        if (walrcv->walRcvState == WALRCV_RESTARTING)
        {
            /*
             * No need to handle changes in primary_conninfo or
             * primary_slot_name here. Startup process will signal us to
             * terminate in case those change.
             */
            *startpoint = walrcv->receiveStart;
            *startpointTLI = walrcv->receiveStartTLI;
            walrcv->walRcvState = WALRCV_CONNECTING;
            SpinLockRelease(&walrcv->mutex);
            break;
        }
        if (walrcv->walRcvState == WALRCV_STOPPING)
        {
            /*
             * We should've received SIGTERM if the startup process wants us
             * to die, but might as well check it here too.
             */
            SpinLockRelease(&walrcv->mutex);
            exit(1);
        }
        SpinLockRelease(&walrcv->mutex);
 
        (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
                         WAIT_EVENT_WAL_RECEIVER_WAIT_START);
    }
 
    if (update_process_title)
    {
        char        activitymsg[50];
 
        snprintf(activitymsg, sizeof(activitymsg), "restarting at %X/%08X",
                 LSN_FORMAT_ARGS(*startpoint));
        set_ps_display(activitymsg);
    }
}

References Assert, CHECK_FOR_INTERRUPTS, elog, FATAL, fb(), InvalidXLogRecPtr, LSN_FORMAT_ARGS, MyLatch, proc_exit(), ResetLatch(), set_ps_display(), snprintf, SpinLockAcquire, SpinLockRelease, update_process_title, WaitLatch(), WakeupRecovery(), WalRcv, WALRCV_CONNECTING, WALRCV_RESTARTING, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, WL_EXIT_ON_PM_DEATH, and WL_LATCH_SET.

Referenced by WalReceiverMain().

◆ WalReceiverMain()

void WalReceiverMain	(	const void *	startup_data,
		size_t	startup_data_len
	)

Definition at line 153 of file walreceiver.c.

{
    char        conninfo[MAXCONNINFO];
    char       *tmp_conninfo;
    char        slotname[NAMEDATALEN];
    bool        is_temp_slot;
    XLogRecPtr  startpoint;
    TimeLineID  startpointTLI;
    TimeLineID  primaryTLI;
    bool        first_stream;
    WalRcvData *walrcv;
    TimestampTz now;
    char       *err;
    char       *sender_host = NULL;
    int         sender_port = 0;
    char       *appname;
 
    Assert(startup_data_len == 0);
 
    AuxiliaryProcessMainCommon();
 
    /*
     * WalRcv should be set up already (if we are a backend, we inherit this
     * by fork() or EXEC_BACKEND mechanism from the postmaster).
     */
    walrcv = WalRcv;
    Assert(walrcv != NULL);
 
    /*
     * Mark walreceiver as running in shared memory.
     *
     * Do this as early as possible, so that if we fail later on, we'll set
     * state to STOPPED. If we die before this, the startup process will keep
     * waiting for us to start up, until it times out.
     */
    SpinLockAcquire(&walrcv->mutex);
    Assert(walrcv->pid == 0);
    switch (walrcv->walRcvState)
    {
        case WALRCV_STOPPING:
            /* If we've already been requested to stop, don't start up. */
            walrcv->walRcvState = WALRCV_STOPPED;
            /* fall through */
 
        case WALRCV_STOPPED:
            SpinLockRelease(&walrcv->mutex);
            ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
            proc_exit(1);
            break;
 
        case WALRCV_STARTING:
            /* The usual case */
            break;
 
        case WALRCV_CONNECTING:
        case WALRCV_WAITING:
        case WALRCV_STREAMING:
        case WALRCV_RESTARTING:
        default:
            /* Shouldn't happen */
            SpinLockRelease(&walrcv->mutex);
            elog(PANIC, "walreceiver still running according to shared memory state");
    }
    /* Advertise our PID so that the startup process can kill us */
    walrcv->pid = MyProcPid;
    walrcv->walRcvState = WALRCV_CONNECTING;
 
    /* Fetch information required to start streaming */
    walrcv->ready_to_display = false;
    strlcpy(conninfo, walrcv->conninfo, MAXCONNINFO);
    strlcpy(slotname, walrcv->slotname, NAMEDATALEN);
    is_temp_slot = walrcv->is_temp_slot;
    startpoint = walrcv->receiveStart;
    startpointTLI = walrcv->receiveStartTLI;
 
    /*
     * At most one of is_temp_slot and slotname can be set; otherwise,
     * RequestXLogStreaming messed up.
     */
    Assert(!is_temp_slot || (slotname[0] == '\0'));
 
    /* Initialise to a sanish value */
    now = GetCurrentTimestamp();
    walrcv->lastMsgSendTime =
        walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
 
    /* Report our proc number so that others can wake us up */
    walrcv->procno = MyProcNumber;
 
    SpinLockRelease(&walrcv->mutex);
 
    pg_atomic_write_u64(&WalRcv->writtenUpto, 0);
 
    /* Arrange to clean up at walreceiver exit */
    on_shmem_exit(WalRcvDie, PointerGetDatum(&startpointTLI));
 
    /* Properly accept or ignore signals the postmaster might send us */
    pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
                                                     * file */
    pqsignal(SIGINT, SIG_IGN);
    pqsignal(SIGTERM, die);     /* request shutdown */
    /* SIGQUIT handler was already set up by InitPostmasterChild */
    pqsignal(SIGALRM, SIG_IGN);
    pqsignal(SIGPIPE, SIG_IGN);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    pqsignal(SIGUSR2, SIG_IGN);
 
    /* Reset some signals that are accepted by postmaster but not here */
    pqsignal(SIGCHLD, SIG_DFL);
 
    /* Load the libpq-specific functions */
    load_file("libpqwalreceiver", false);
    if (WalReceiverFunctions == NULL)
        elog(ERROR, "libpqwalreceiver didn't initialize correctly");
 
    /* Unblock signals (they were blocked when the postmaster forked us) */
    sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
 
    /* Establish the connection to the primary for XLOG streaming */
    appname = cluster_name[0] ? cluster_name : "walreceiver";
    wrconn = walrcv_connect(conninfo, true, false, false, appname, &err);
    if (!wrconn)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("streaming replication receiver \"%s\" could not connect to the primary server: %s",
                        appname, err)));
 
    /*
     * Save user-visible connection string.  This clobbers the original
     * conninfo, for security. Also save host and port of the sender server
     * this walreceiver is connected to.
     */
    tmp_conninfo = walrcv_get_conninfo(wrconn);
    walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
    SpinLockAcquire(&walrcv->mutex);
    memset(walrcv->conninfo, 0, MAXCONNINFO);
    if (tmp_conninfo)
        strlcpy(walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
 
    memset(walrcv->sender_host, 0, NI_MAXHOST);
    if (sender_host)
        strlcpy(walrcv->sender_host, sender_host, NI_MAXHOST);
 
    walrcv->sender_port = sender_port;
    walrcv->ready_to_display = true;
    SpinLockRelease(&walrcv->mutex);
 
    if (tmp_conninfo)
        pfree(tmp_conninfo);
 
    if (sender_host)
        pfree(sender_host);
 
    first_stream = true;
    for (;;)
    {
        char       *primary_sysid;
        char        standby_sysid[32];
        WalRcvStreamOptions options;
 
        /*
         * Check that we're connected to a valid server using the
         * IDENTIFY_SYSTEM replication command.
         */
        primary_sysid = walrcv_identify_system(wrconn, &primaryTLI);
 
        snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
                 GetSystemIdentifier());
        if (strcmp(primary_sysid, standby_sysid) != 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("database system identifier differs between the primary and standby"),
                     errdetail("The primary's identifier is %s, the standby's identifier is %s.",
                               primary_sysid, standby_sysid)));
        }
 
        /*
         * Confirm that the current timeline of the primary is the same or
         * ahead of ours.
         */
        if (primaryTLI < startpointTLI)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("highest timeline %u of the primary is behind recovery timeline %u",
                            primaryTLI, startpointTLI)));
 
        /*
         * Get any missing history files. We do this always, even when we're
         * not interested in that timeline, so that if we're promoted to
         * become the primary later on, we don't select the same timeline that
         * was already used in the current primary. This isn't bullet-proof -
         * you'll need some external software to manage your cluster if you
         * need to ensure that a unique timeline id is chosen in every case,
         * but let's avoid the confusion of timeline id collisions where we
         * can.
         */
        WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
 
        /*
         * Create temporary replication slot if requested, and update slot
         * name in shared memory.  (Note the slot name cannot already be set
         * in this case.)
         */
        if (is_temp_slot)
        {
            snprintf(slotname, sizeof(slotname),
                     "pg_walreceiver_%lld",
                     (long long int) walrcv_get_backend_pid(wrconn));
 
            walrcv_create_slot(wrconn, slotname, true, false, false, 0, NULL);
 
            SpinLockAcquire(&walrcv->mutex);
            strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
            SpinLockRelease(&walrcv->mutex);
        }
 
        /*
         * Start streaming.
         *
         * We'll try to start at the requested starting point and timeline,
         * even if it's different from the server's latest timeline. In case
         * we've already reached the end of the old timeline, the server will
         * finish the streaming immediately, and we will go back to await
         * orders from the startup process. If recovery_target_timeline is
         * 'latest', the startup process will scan pg_wal and find the new
         * history file, bump recovery target timeline, and ask us to restart
         * on the new timeline.
         */
        options.logical = false;
        options.startpoint = startpoint;
        options.slotname = slotname[0] != '\0' ? slotname : NULL;
        options.proto.physical.startpointTLI = startpointTLI;
        if (walrcv_startstreaming(wrconn, &options))
        {
            if (first_stream)
                ereport(LOG,
                        errmsg("started streaming WAL from primary at %X/%08X on timeline %u",
                               LSN_FORMAT_ARGS(startpoint), startpointTLI));
            else
                ereport(LOG,
                        errmsg("restarted WAL streaming at %X/%08X on timeline %u",
                               LSN_FORMAT_ARGS(startpoint), startpointTLI));
            first_stream = false;
 
            /*
             * Switch to STREAMING after a successful connection if current
             * state is CONNECTING.  This switch happens after an initial
             * startup, or after a restart as determined by
             * WalRcvWaitForStartPosition().
             */
            SpinLockAcquire(&walrcv->mutex);
            if (walrcv->walRcvState == WALRCV_CONNECTING)
                walrcv->walRcvState = WALRCV_STREAMING;
            SpinLockRelease(&walrcv->mutex);
 
            /* Initialize LogstreamResult and buffers for processing messages */
            LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
            initStringInfo(&reply_message);
 
            /* Initialize nap wakeup times. */
            now = GetCurrentTimestamp();
            for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
                WalRcvComputeNextWakeup(i, now);
 
            /* Send initial reply/feedback messages. */
            XLogWalRcvSendReply(true, false);
            XLogWalRcvSendHSFeedback(true);
 
            /* Loop until end-of-streaming or error */
            for (;;)
            {
                char       *buf;
                int         len;
                bool        endofwal = false;
                pgsocket    wait_fd = PGINVALID_SOCKET;
                int         rc;
                TimestampTz nextWakeup;
                long        nap;
 
                /*
                 * Exit walreceiver if we're not in recovery. This should not
                 * happen, but cross-check the status here.
                 */
                if (!RecoveryInProgress())
                    ereport(FATAL,
                            (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                             errmsg("cannot continue WAL streaming, recovery has already ended")));
 
                /* Process any requests or signals received recently */
                CHECK_FOR_INTERRUPTS();
 
                if (ConfigReloadPending)
                {
                    ConfigReloadPending = false;
                    ProcessConfigFile(PGC_SIGHUP);
                    /* recompute wakeup times */
                    now = GetCurrentTimestamp();
                    for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
                        WalRcvComputeNextWakeup(i, now);
                    XLogWalRcvSendHSFeedback(true);
                }
 
                /* See if we can read data immediately */
                len = walrcv_receive(wrconn, &buf, &wait_fd);
                if (len != 0)
                {
                    /*
                     * Process the received data, and any subsequent data we
                     * can read without blocking.
                     */
                    for (;;)
                    {
                        if (len > 0)
                        {
                            /*
                             * Something was received from primary, so adjust
                             * the ping and terminate wakeup times.
                             */
                            now = GetCurrentTimestamp();
                            WalRcvComputeNextWakeup(WALRCV_WAKEUP_TERMINATE,
                                                    now);
                            WalRcvComputeNextWakeup(WALRCV_WAKEUP_PING, now);
                            XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
                                                 startpointTLI);
                        }
                        else if (len == 0)
                            break;
                        else if (len < 0)
                        {
                            ereport(LOG,
                                    (errmsg("replication terminated by primary server"),
                                     errdetail("End of WAL reached on timeline %u at %X/%08X.",
                                               startpointTLI,
                                               LSN_FORMAT_ARGS(LogstreamResult.Write))));
                            endofwal = true;
                            break;
                        }
                        len = walrcv_receive(wrconn, &buf, &wait_fd);
                    }
 
                    /* Let the primary know that we received some data. */
                    XLogWalRcvSendReply(false, false);
 
                    /*
                     * If we've written some records, flush them to disk and
                     * let the startup process and primary server know about
                     * them.
                     */
                    XLogWalRcvFlush(false, startpointTLI);
                }
 
                /* Check if we need to exit the streaming loop. */
                if (endofwal)
                    break;
 
                /* Find the soonest wakeup time, to limit our nap. */
                nextWakeup = TIMESTAMP_INFINITY;
                for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
                    nextWakeup = Min(wakeup[i], nextWakeup);
 
                /* Calculate the nap time, clamping as necessary. */
                now = GetCurrentTimestamp();
                nap = TimestampDifferenceMilliseconds(now, nextWakeup);
 
                /*
                 * Ideally we would reuse a WaitEventSet object repeatedly
                 * here to avoid the overheads of WaitLatchOrSocket on epoll
                 * systems, but we can't be sure that libpq (or any other
                 * walreceiver implementation) has the same socket (even if
                 * the fd is the same number, it may have been closed and
                 * reopened since the last time).  In future, if there is a
                 * function for removing sockets from WaitEventSet, then we
                 * could add and remove just the socket each time, potentially
                 * avoiding some system calls.
                 */
                Assert(wait_fd != PGINVALID_SOCKET);
                rc = WaitLatchOrSocket(MyLatch,
                                       WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
                                       WL_TIMEOUT | WL_LATCH_SET,
                                       wait_fd,
                                       nap,
                                       WAIT_EVENT_WAL_RECEIVER_MAIN);
                if (rc & WL_LATCH_SET)
                {
                    ResetLatch(MyLatch);
                    CHECK_FOR_INTERRUPTS();
 
                    if (walrcv->force_reply)
                    {
                        /*
                         * The recovery process has asked us to send apply
                         * feedback now.  Make sure the flag is really set to
                         * false in shared memory before sending the reply, so
                         * we don't miss a new request for a reply.
                         */
                        walrcv->force_reply = false;
                        pg_memory_barrier();
                        XLogWalRcvSendReply(true, false);
                    }
                }
                if (rc & WL_TIMEOUT)
                {
                    /*
                     * We didn't receive anything new. If we haven't heard
                     * anything from the server for more than
                     * wal_receiver_timeout / 2, ping the server. Also, if
                     * it's been longer than wal_receiver_status_interval
                     * since the last update we sent, send a status update to
                     * the primary anyway, to report any progress in applying
                     * WAL.
                     */
                    bool        requestReply = false;
 
                    /*
                     * Report pending statistics to the cumulative stats
                     * system.  This location is useful for the report as it
                     * is not within a tight loop in the WAL receiver, to
                     * avoid bloating pgstats with requests, while also making
                     * sure that the reports happen each time a status update
                     * is sent.
                     */
                    pgstat_report_wal(false);
 
                    /*
                     * Check if time since last receive from primary has
                     * reached the configured limit.
                     */
                    now = GetCurrentTimestamp();
                    if (now >= wakeup[WALRCV_WAKEUP_TERMINATE])
                        ereport(ERROR,
                                (errcode(ERRCODE_CONNECTION_FAILURE),
                                 errmsg("terminating walreceiver due to timeout")));
 
                    /*
                     * If we didn't receive anything new for half of receiver
                     * replication timeout, then ping the server.
                     */
                    if (now >= wakeup[WALRCV_WAKEUP_PING])
                    {
                        requestReply = true;
                        wakeup[WALRCV_WAKEUP_PING] = TIMESTAMP_INFINITY;
                    }
 
                    XLogWalRcvSendReply(requestReply, requestReply);
                    XLogWalRcvSendHSFeedback(false);
                }
            }
 
            /*
             * The backend finished streaming. Exit streaming COPY-mode from
             * our side, too.
             */
            walrcv_endstreaming(wrconn, &primaryTLI);
 
            /*
             * If the server had switched to a new timeline that we didn't
             * know about when we began streaming, fetch its timeline history
             * file now.
             */
            WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
        }
        else
            ereport(LOG,
                    (errmsg("primary server contains no more WAL on requested timeline %u",
                            startpointTLI)));
 
        /*
         * End of WAL reached on the requested timeline. Close the last
         * segment, and await for new orders from the startup process.
         */
        if (recvFile >= 0)
        {
            char        xlogfname[MAXFNAMELEN];
 
            XLogWalRcvFlush(false, startpointTLI);
            XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
            if (close(recvFile) != 0)
                ereport(PANIC,
                        (errcode_for_file_access(),
                         errmsg("could not close WAL segment %s: %m",
                                xlogfname)));
 
            /*
             * Create .done file forcibly to prevent the streamed segment from
             * being archived later.
             */
            if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
                XLogArchiveForceDone(xlogfname);
            else
                XLogArchiveNotify(xlogfname);
        }
        recvFile = -1;
 
        elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
        WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
    }
    /* not reached */
}

References ARCHIVE_MODE_ALWAYS, Assert, AuxiliaryProcessMainCommon(), buf, CHECK_FOR_INTERRUPTS, close, cluster_name, ConditionVariableBroadcast(), ConfigReloadPending, DEBUG1, die, elog, ereport, err(), errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, FATAL, fb(), GetCurrentTimestamp(), GetSystemIdentifier(), GetXLogReplayRecPtr(), i, initStringInfo(), len, load_file(), LOG, LogstreamResult, LSN_FORMAT_ARGS, MAXCONNINFO, MAXFNAMELEN, Min, MyLatch, MyProcNumber, MyProcPid, NAMEDATALEN, now(), NUM_WALRCV_WAKEUPS, on_shmem_exit(), PANIC, pfree(), pg_atomic_write_u64(), pg_memory_barrier, PGC_SIGHUP, PGINVALID_SOCKET, pgstat_report_wal(), PointerGetDatum(), pqsignal, proc_exit(), ProcessConfigFile(), procsignal_sigusr1_handler(), RecoveryInProgress(), recvFile, recvFileTLI, recvSegNo, reply_message, ResetLatch(), SIGALRM, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SIGPIPE, SIGUSR1, SIGUSR2, snprintf, SpinLockAcquire, SpinLockRelease, strlcpy(), TIMESTAMP_INFINITY, TimestampDifferenceMilliseconds(), UINT64_FORMAT, UnBlockSig, WaitLatchOrSocket(), wakeup, wal_segment_size, WalRcv, walrcv_connect, WALRCV_CONNECTING, walrcv_create_slot, walrcv_endstreaming, walrcv_get_backend_pid, walrcv_get_conninfo, walrcv_get_senderinfo, walrcv_identify_system, walrcv_receive, WALRCV_RESTARTING, WALRCV_STARTING, walrcv_startstreaming, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, WALRCV_WAKEUP_PING, WALRCV_WAKEUP_TERMINATE, WalRcvComputeNextWakeup(), WalRcvDie(), WalRcvFetchTimeLineHistoryFiles(), WalRcvWaitForStartPosition(), WalReceiverFunctions, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_READABLE, WL_TIMEOUT, wrconn, WalRcvData::writtenUpto, XLogArchiveForceDone(), XLogArchiveMode, XLogArchiveNotify(), XLogFileName(), XLogWalRcvFlush(), XLogWalRcvProcessMsg(), XLogWalRcvSendHSFeedback(), and XLogWalRcvSendReply().

◆ XLogWalRcvClose()

static void XLogWalRcvClose	(	XLogRecPtr	recptr,
		TimeLineID	tli
	)

static

Definition at line 1070 of file walreceiver.c.

{
    char        xlogfname[MAXFNAMELEN];
 
    Assert(recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size));
    Assert(tli != 0);
 
    /*
     * fsync() and close current file before we switch to next one. We would
     * otherwise have to reopen this file to fsync it later
     */
    XLogWalRcvFlush(false, tli);
 
    XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
 
    /*
     * XLOG segment files will be re-read by recovery in startup process soon,
     * so we don't advise the OS to release cache pages associated with the
     * file like XLogFileClose() does.
     */
    if (close(recvFile) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close WAL segment %s: %m",
                        xlogfname)));
 
    /*
     * Create .done file forcibly to prevent the streamed segment from being
     * archived later.
     */
    if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
        XLogArchiveForceDone(xlogfname);
    else
        XLogArchiveNotify(xlogfname);
 
    recvFile = -1;
}

References ARCHIVE_MODE_ALWAYS, Assert, close, ereport, errcode_for_file_access(), errmsg(), fb(), MAXFNAMELEN, PANIC, recvFile, recvFileTLI, recvSegNo, wal_segment_size, XLByteInSeg, XLogArchiveForceDone(), XLogArchiveMode, XLogArchiveNotify(), XLogFileName(), and XLogWalRcvFlush().

Referenced by XLogWalRcvWrite().

◆ XLogWalRcvFlush()

static void XLogWalRcvFlush	(	bool	dying,
		TimeLineID	tli
	)

static

Definition at line 1006 of file walreceiver.c.

{
    Assert(tli != 0);
 
    if (LogstreamResult.Flush < LogstreamResult.Write)
    {
        WalRcvData *walrcv = WalRcv;
 
        issue_xlog_fsync(recvFile, recvSegNo, tli);
 
        LogstreamResult.Flush = LogstreamResult.Write;
 
        /* Update shared-memory status */
        SpinLockAcquire(&walrcv->mutex);
        if (walrcv->flushedUpto < LogstreamResult.Flush)
        {
            walrcv->latestChunkStart = walrcv->flushedUpto;
            walrcv->flushedUpto = LogstreamResult.Flush;
            walrcv->receivedTLI = tli;
        }
        SpinLockRelease(&walrcv->mutex);
 
        /*
         * If we flushed an LSN that someone was waiting for, notify the
         * waiters.
         */
        if (waitLSNState &&
            (LogstreamResult.Flush >=
             pg_atomic_read_u64(&waitLSNState->minWaitedLSN[WAIT_LSN_TYPE_STANDBY_FLUSH])))
            WaitLSNWakeup(WAIT_LSN_TYPE_STANDBY_FLUSH, LogstreamResult.Flush);
 
        /* Signal the startup process and walsender that new WAL has arrived */
        WakeupRecovery();
        if (AllowCascadeReplication())
            WalSndWakeup(true, false);
 
        /* Report XLOG streaming progress in PS display */
        if (update_process_title)
        {
            char        activitymsg[50];
 
            snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%08X",
                     LSN_FORMAT_ARGS(LogstreamResult.Write));
            set_ps_display(activitymsg);
        }
 
        /* Also let the primary know that we made some progress */
        if (!dying)
        {
            XLogWalRcvSendReply(false, false);
            XLogWalRcvSendHSFeedback(false);
        }
    }
}

References AllowCascadeReplication, Assert, fb(), issue_xlog_fsync(), LogstreamResult, LSN_FORMAT_ARGS, WaitLSNState::minWaitedLSN, pg_atomic_read_u64(), recvFile, recvSegNo, set_ps_display(), snprintf, SpinLockAcquire, SpinLockRelease, update_process_title, WAIT_LSN_TYPE_STANDBY_FLUSH, waitLSNState, WaitLSNWakeup(), WakeupRecovery(), WalRcv, WalSndWakeup(), XLogWalRcvSendHSFeedback(), and XLogWalRcvSendReply().

Referenced by WalRcvDie(), WalReceiverMain(), and XLogWalRcvClose().

◆ XLogWalRcvProcessMsg()

static void XLogWalRcvProcessMsg	(	unsigned char	type,
		char *	buf,
		Size	len,
		TimeLineID	tli
	)

static

Definition at line 832 of file walreceiver.c.

{
    int         hdrlen;
    XLogRecPtr  dataStart;
    XLogRecPtr  walEnd;
    TimestampTz sendTime;
    bool        replyRequested;
 
    switch (type)
    {
        case PqReplMsg_WALData:
            {
                StringInfoData incoming_message;
 
                hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
                if (len < hdrlen)
                    ereport(ERROR,
                            (errcode(ERRCODE_PROTOCOL_VIOLATION),
                             errmsg_internal("invalid WAL message received from primary")));
 
                /* initialize a StringInfo with the given buffer */
                initReadOnlyStringInfo(&incoming_message, buf, hdrlen);
 
                /* read the fields */
                dataStart = pq_getmsgint64(&incoming_message);
                walEnd = pq_getmsgint64(&incoming_message);
                sendTime = pq_getmsgint64(&incoming_message);
                ProcessWalSndrMessage(walEnd, sendTime);
 
                buf += hdrlen;
                len -= hdrlen;
                XLogWalRcvWrite(buf, len, dataStart, tli);
                break;
            }
        case PqReplMsg_Keepalive:
            {
                StringInfoData incoming_message;
 
                hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
                if (len != hdrlen)
                    ereport(ERROR,
                            (errcode(ERRCODE_PROTOCOL_VIOLATION),
                             errmsg_internal("invalid keepalive message received from primary")));
 
                /* initialize a StringInfo with the given buffer */
                initReadOnlyStringInfo(&incoming_message, buf, hdrlen);
 
                /* read the fields */
                walEnd = pq_getmsgint64(&incoming_message);
                sendTime = pq_getmsgint64(&incoming_message);
                replyRequested = pq_getmsgbyte(&incoming_message);
 
                ProcessWalSndrMessage(walEnd, sendTime);
 
                /* If the primary requested a reply, send one immediately */
                if (replyRequested)
                    XLogWalRcvSendReply(true, false);
                break;
            }
        default:
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg_internal("invalid replication message type %d",
                                     type)));
    }
}

References buf, ereport, errcode(), errmsg_internal(), ERROR, fb(), initReadOnlyStringInfo(), len, pq_getmsgbyte(), pq_getmsgint64(), PqReplMsg_Keepalive, PqReplMsg_WALData, ProcessWalSndrMessage(), type, XLogWalRcvSendReply(), and XLogWalRcvWrite().

Referenced by WalReceiverMain().

◆ XLogWalRcvSendHSFeedback()

static void XLogWalRcvSendHSFeedback ( bool immed )

static

Definition at line 1191 of file walreceiver.c.

{
    TimestampTz now;
    FullTransactionId nextFullXid;
    TransactionId nextXid;
    uint32      xmin_epoch,
                catalog_xmin_epoch;
    TransactionId xmin,
                catalog_xmin;
 
    /* initially true so we always send at least one feedback message */
    static bool primary_has_standby_xmin = true;
 
    /*
     * If the user doesn't want status to be reported to the primary, be sure
     * to exit before doing anything at all.
     */
    if ((wal_receiver_status_interval <= 0 || !hot_standby_feedback) &&
        !primary_has_standby_xmin)
        return;
 
    /* Get current timestamp. */
    now = GetCurrentTimestamp();
 
    /* Send feedback at most once per wal_receiver_status_interval. */
    if (!immed && now < wakeup[WALRCV_WAKEUP_HSFEEDBACK])
        return;
 
    /* Make sure we wake up when it's time to send feedback again. */
    WalRcvComputeNextWakeup(WALRCV_WAKEUP_HSFEEDBACK, now);
 
    /*
     * If Hot Standby is not yet accepting connections there is nothing to
     * send. Check this after the interval has expired to reduce number of
     * calls.
     *
     * Bailing out here also ensures that we don't send feedback until we've
     * read our own replication slot state, so we don't tell the primary to
     * discard needed xmin or catalog_xmin from any slots that may exist on
     * this replica.
     */
    if (!HotStandbyActive())
        return;
 
    /*
     * Make the expensive call to get the oldest xmin once we are certain
     * everything else has been checked.
     */
    if (hot_standby_feedback)
    {
        GetReplicationHorizons(&xmin, &catalog_xmin);
    }
    else
    {
        xmin = InvalidTransactionId;
        catalog_xmin = InvalidTransactionId;
    }
 
    /*
     * Get epoch and adjust if nextXid and oldestXmin are different sides of
     * the epoch boundary.
     */
    nextFullXid = ReadNextFullTransactionId();
    nextXid = XidFromFullTransactionId(nextFullXid);
    xmin_epoch = EpochFromFullTransactionId(nextFullXid);
    catalog_xmin_epoch = xmin_epoch;
    if (nextXid < xmin)
        xmin_epoch--;
    if (nextXid < catalog_xmin)
        catalog_xmin_epoch--;
 
    elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u catalog_xmin %u catalog_xmin_epoch %u",
         xmin, xmin_epoch, catalog_xmin, catalog_xmin_epoch);
 
    /* Construct the message and send it. */
    resetStringInfo(&reply_message);
    pq_sendbyte(&reply_message, PqReplMsg_HotStandbyFeedback);
    pq_sendint64(&reply_message, GetCurrentTimestamp());
    pq_sendint32(&reply_message, xmin);
    pq_sendint32(&reply_message, xmin_epoch);
    pq_sendint32(&reply_message, catalog_xmin);
    pq_sendint32(&reply_message, catalog_xmin_epoch);
    walrcv_send(wrconn, reply_message.data, reply_message.len);
    if (TransactionIdIsValid(xmin) || TransactionIdIsValid(catalog_xmin))
        primary_has_standby_xmin = true;
    else
        primary_has_standby_xmin = false;
}

References StringInfoData::data, DEBUG2, elog, EpochFromFullTransactionId, fb(), GetCurrentTimestamp(), GetReplicationHorizons(), hot_standby_feedback, HotStandbyActive(), InvalidTransactionId, StringInfoData::len, now(), pq_sendbyte(), pq_sendint32(), pq_sendint64(), PqReplMsg_HotStandbyFeedback, ReadNextFullTransactionId(), reply_message, resetStringInfo(), TransactionIdIsValid, wakeup, wal_receiver_status_interval, walrcv_send, WALRCV_WAKEUP_HSFEEDBACK, WalRcvComputeNextWakeup(), wrconn, and XidFromFullTransactionId.

Referenced by WalReceiverMain(), and XLogWalRcvFlush().

◆ XLogWalRcvSendReply()

static void XLogWalRcvSendReply	(	bool	force,
		bool	requestReply
	)

static

Definition at line 1122 of file walreceiver.c.

{
    static XLogRecPtr writePtr = InvalidXLogRecPtr;
    static XLogRecPtr flushPtr = InvalidXLogRecPtr;
    XLogRecPtr  applyPtr;
    TimestampTz now;
 
    /*
     * If the user doesn't want status to be reported to the primary, be sure
     * to exit before doing anything at all.
     */
    if (!force && wal_receiver_status_interval <= 0)
        return;
 
    /* Get current timestamp. */
    now = GetCurrentTimestamp();
 
    /*
     * We can compare the write and flush positions to the last message we
     * sent without taking any lock, but the apply position requires a spin
     * lock, so we don't check that unless something else has changed or 10
     * seconds have passed.  This means that the apply WAL location will
     * appear, from the primary's point of view, to lag slightly, but since
     * this is only for reporting purposes and only on idle systems, that's
     * probably OK.
     */
    if (!force
        && writePtr == LogstreamResult.Write
        && flushPtr == LogstreamResult.Flush
        && now < wakeup[WALRCV_WAKEUP_REPLY])
        return;
 
    /* Make sure we wake up when it's time to send another reply. */
    WalRcvComputeNextWakeup(WALRCV_WAKEUP_REPLY, now);
 
    /* Construct a new message */
    writePtr = LogstreamResult.Write;
    flushPtr = LogstreamResult.Flush;
    applyPtr = GetXLogReplayRecPtr(NULL);
 
    resetStringInfo(&reply_message);
    pq_sendbyte(&reply_message, PqReplMsg_StandbyStatusUpdate);
    pq_sendint64(&reply_message, writePtr);
    pq_sendint64(&reply_message, flushPtr);
    pq_sendint64(&reply_message, applyPtr);
    pq_sendint64(&reply_message, GetCurrentTimestamp());
    pq_sendbyte(&reply_message, requestReply ? 1 : 0);
 
    /* Send it */
    elog(DEBUG2, "sending write %X/%08X flush %X/%08X apply %X/%08X%s",
         LSN_FORMAT_ARGS(writePtr),
         LSN_FORMAT_ARGS(flushPtr),
         LSN_FORMAT_ARGS(applyPtr),
         requestReply ? " (reply requested)" : "");
 
    walrcv_send(wrconn, reply_message.data, reply_message.len);
}

References StringInfoData::data, DEBUG2, elog, fb(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), InvalidXLogRecPtr, StringInfoData::len, LogstreamResult, LSN_FORMAT_ARGS, now(), pq_sendbyte(), pq_sendint64(), PqReplMsg_StandbyStatusUpdate, reply_message, resetStringInfo(), wakeup, wal_receiver_status_interval, walrcv_send, WALRCV_WAKEUP_REPLY, WalRcvComputeNextWakeup(), and wrconn.

Referenced by WalReceiverMain(), XLogWalRcvFlush(), and XLogWalRcvProcessMsg().

◆ XLogWalRcvWrite()

static void XLogWalRcvWrite	(	char *	buf,
		Size	nbytes,
		XLogRecPtr	recptr,
		TimeLineID	tli
	)

static

Definition at line 903 of file walreceiver.c.

{
    int         startoff;
    int         byteswritten;
    instr_time  start;
 
    Assert(tli != 0);
 
    while (nbytes > 0)
    {
        int         segbytes;
 
        /* Close the current segment if it's completed */
        if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
            XLogWalRcvClose(recptr, tli);
 
        if (recvFile < 0)
        {
            /* Create/use new log file */
            XLByteToSeg(recptr, recvSegNo, wal_segment_size);
            recvFile = XLogFileInit(recvSegNo, tli);
            recvFileTLI = tli;
        }
 
        /* Calculate the start offset of the received logs */
        startoff = XLogSegmentOffset(recptr, wal_segment_size);
 
        if (startoff + nbytes > wal_segment_size)
            segbytes = wal_segment_size - startoff;
        else
            segbytes = nbytes;
 
        /* OK to write the logs */
        errno = 0;
 
        /*
         * Measure I/O timing to write WAL data, for pg_stat_io.
         */
        start = pgstat_prepare_io_time(track_wal_io_timing);
 
        pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE);
        byteswritten = pg_pwrite(recvFile, buf, segbytes, (pgoff_t) startoff);
        pgstat_report_wait_end();
 
        pgstat_count_io_op_time(IOOBJECT_WAL, IOCONTEXT_NORMAL,
                                IOOP_WRITE, start, 1, byteswritten);
 
        if (byteswritten <= 0)
        {
            char        xlogfname[MAXFNAMELEN];
            int         save_errno;
 
            /* if write didn't set errno, assume no disk space */
            if (errno == 0)
                errno = ENOSPC;
 
            save_errno = errno;
            XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
            errno = save_errno;
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not write to WAL segment %s "
                            "at offset %d, length %d: %m",
                            xlogfname, startoff, segbytes)));
        }
 
        /* Update state for write */
        recptr += byteswritten;
 
        nbytes -= byteswritten;
        buf += byteswritten;
 
        LogstreamResult.Write = recptr;
    }
 
    /* Update shared-memory status */
    pg_atomic_write_u64(&WalRcv->writtenUpto, LogstreamResult.Write);
 
    /*
     * If we wrote an LSN that someone was waiting for, notify the waiters.
     */
    if (waitLSNState &&
        (LogstreamResult.Write >=
         pg_atomic_read_u64(&waitLSNState->minWaitedLSN[WAIT_LSN_TYPE_STANDBY_WRITE])))
        WaitLSNWakeup(WAIT_LSN_TYPE_STANDBY_WRITE, LogstreamResult.Write);
 
    /*
     * Close the current segment if it's fully written up in the last cycle of
     * the loop, to create its archive notification file soon. Otherwise WAL
     * archiving of the segment will be delayed until any data in the next
     * segment is received and written.
     */
    if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
        XLogWalRcvClose(recptr, tli);
}