walreceiver.h File Reference

#include "access/xlog.h"
#include "access/xlogdefs.h"
#include "getaddrinfo.h"
#include "pgtime.h"
#include "port/atomics.h"
#include "replication/logicalproto.h"
#include "replication/walsender.h"
#include "storage/condition_variable.h"
#include "storage/latch.h"
#include "storage/spin.h"
#include "utils/tuplestore.h"

Include dependency graph for walreceiver.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	WalRcvData
struct	WalRcvStreamOptions
struct	WalRcvExecResult
struct	WalReceiverFunctionsType

Macros
#define	MAXCONNINFO   1024
#define	AllowCascadeReplication()   (EnableHotStandby && max_wal_senders > 0)
#define	walrcv_connect(conninfo, logical, appname, err)   WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
#define	walrcv_check_conninfo(conninfo)   WalReceiverFunctions->walrcv_check_conninfo(conninfo)
#define	walrcv_get_conninfo(conn)   WalReceiverFunctions->walrcv_get_conninfo(conn)
#define	walrcv_get_senderinfo(conn, sender_host, sender_port)   WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
#define	walrcv_identify_system(conn, primary_tli)   WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
#define	walrcv_server_version(conn)   WalReceiverFunctions->walrcv_server_version(conn)
#define	walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)   WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
#define	walrcv_startstreaming(conn, options)   WalReceiverFunctions->walrcv_startstreaming(conn, options)
#define	walrcv_endstreaming(conn, next_tli)   WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
#define	walrcv_receive(conn, buffer, wait_fd)   WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
#define	walrcv_send(conn, buffer, nbytes)   WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#define	walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)   WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
#define	walrcv_get_backend_pid(conn)   WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define	walrcv_exec(conn, exec, nRetTypes, retTypes)   WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
#define	walrcv_disconnect(conn)   WalReceiverFunctions->walrcv_disconnect(conn)

Typedefs
typedef struct WalReceiverConn	WalReceiverConn
typedef struct WalRcvExecResult	WalRcvExecResult
typedef WalReceiverConn *(*	walrcv_connect_fn) (const char *conninfo, bool logical, const char *appname, char **err)
typedef void(*	walrcv_check_conninfo_fn) (const char *conninfo)
typedef char *(*	walrcv_get_conninfo_fn) (WalReceiverConn *conn)
typedef void(*	walrcv_get_senderinfo_fn) (WalReceiverConn *conn, char **sender_host, int *sender_port)
typedef char *(*	walrcv_identify_system_fn) (WalReceiverConn *conn, TimeLineID *primary_tli)
typedef int(*	walrcv_server_version_fn) (WalReceiverConn *conn)
typedef void(*	walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *size)
typedef bool(*	walrcv_startstreaming_fn) (WalReceiverConn *conn, const WalRcvStreamOptions *options)
typedef void(*	walrcv_endstreaming_fn) (WalReceiverConn *conn, TimeLineID *next_tli)
typedef int(*	walrcv_receive_fn) (WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
typedef void(*	walrcv_send_fn) (WalReceiverConn *conn, const char *buffer, int nbytes)
typedef char *(*	walrcv_create_slot_fn) (WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
typedef pid_t(*	walrcv_get_backend_pid_fn) (WalReceiverConn *conn)
typedef WalRcvExecResult *(*	walrcv_exec_fn) (WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
typedef void(*	walrcv_disconnect_fn) (WalReceiverConn *conn)
typedef struct WalReceiverFunctionsType	WalReceiverFunctionsType

Enumerations
enum	WalRcvState {   WALRCV_STOPPED, WALRCV_STARTING, WALRCV_STREAMING, WALRCV_WAITING,   WALRCV_RESTARTING, WALRCV_STOPPING }
enum	WalRcvExecStatus {   WALRCV_ERROR, WALRCV_OK_COMMAND, WALRCV_OK_TUPLES, WALRCV_OK_COPY_IN,   WALRCV_OK_COPY_OUT, WALRCV_OK_COPY_BOTH }

Functions
static void	walrcv_clear_result (WalRcvExecResult *walres)
void	WalReceiverMain (void) pg_attribute_noreturn()
void	ProcessWalRcvInterrupts (void)
Size	WalRcvShmemSize (void)
void	WalRcvShmemInit (void)
void	ShutdownWalRcv (void)
bool	WalRcvStreaming (void)
bool	WalRcvRunning (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)
void	WalRcvForceReply (void)

Variables
int	wal_receiver_status_interval
int	wal_receiver_timeout
bool	hot_standby_feedback
WalRcvData *	WalRcv
PGDLLIMPORT WalReceiverFunctionsType *	WalReceiverFunctions

Macro Definition Documentation

AllowCascadeReplication

#define AllowCascadeReplication

(

)

(EnableHotStandby && max_wal_senders > 0)

Definition at line 40 of file walreceiver.h.

Referenced by StartupXLOG(), and XLogWalRcvFlush().

MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 37 of file walreceiver.h.

Referenced by pg_stat_get_wal_receiver(), RequestXLogStreaming(), and WalReceiverMain().

walrcv_check_conninfo

#define walrcv_check_conninfo

(

conninfo

)

WalReceiverFunctions->walrcv_check_conninfo(conninfo)

Definition at line 406 of file walreceiver.h.

Referenced by AlterSubscription(), and CreateSubscription().

walrcv_connect

#define walrcv_connect	(		conninfo,
			logical,
			appname,
			err
	)		WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)

Definition at line 404 of file walreceiver.h.

Referenced by AlterSubscription_refresh(), ApplyWorkerMain(), CreateSubscription(), DropSubscription(), LogicalRepSyncTableStart(), and WalReceiverMain().

walrcv_create_slot

#define walrcv_create_slot	(		conn,
			slotname,
			temporary,
			two_phase,
			snapshot_action,
			lsn
	)		WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)

Definition at line 426 of file walreceiver.h.

Referenced by CreateSubscription(), LogicalRepSyncTableStart(), and WalReceiverMain().

walrcv_disconnect

#define walrcv_disconnect

(

conn

)

WalReceiverFunctions->walrcv_disconnect(conn)

Definition at line 432 of file walreceiver.h.

Referenced by AlterSubscription_refresh(), CreateSubscription(), DropSubscription(), logicalrep_worker_onexit(), and WalRcvDie().

walrcv_endstreaming

#define walrcv_endstreaming	(		conn,
			next_tli
	)		WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)

Definition at line 420 of file walreceiver.h.

Referenced by LogicalRepApplyLoop(), process_syncing_tables_for_sync(), and WalReceiverMain().

walrcv_exec

#define walrcv_exec	(		conn,
			exec,
			nRetTypes,
			retTypes
	)		WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)

Definition at line 430 of file walreceiver.h.

Referenced by copy_table(), fetch_remote_table_info(), fetch_table_list(), LogicalRepSyncTableStart(), and ReplicationSlotDropAtPubNode().

walrcv_get_backend_pid

#define walrcv_get_backend_pid

(

conn

)

WalReceiverFunctions->walrcv_get_backend_pid(conn)

Definition at line 428 of file walreceiver.h.

Referenced by WalReceiverMain().

walrcv_get_conninfo

#define walrcv_get_conninfo

(

conn

)

WalReceiverFunctions->walrcv_get_conninfo(conn)

Definition at line 408 of file walreceiver.h.

Referenced by WalReceiverMain().

walrcv_get_senderinfo

#define walrcv_get_senderinfo	(		conn,
			sender_host,
			sender_port
	)		WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)

Definition at line 410 of file walreceiver.h.

Referenced by WalReceiverMain().

walrcv_identify_system

#define walrcv_identify_system	(		conn,
			primary_tli
	)		WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)

Definition at line 412 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and WalReceiverMain().

walrcv_readtimelinehistoryfile

#define walrcv_readtimelinehistoryfile	(		conn,
			tli,
			filename,
			content,
			size
	)		WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)

Definition at line 416 of file walreceiver.h.

Referenced by WalRcvFetchTimeLineHistoryFiles().

walrcv_receive

#define walrcv_receive	(		conn,
			buffer,
			wait_fd
	)		WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)

Definition at line 422 of file walreceiver.h.

Referenced by copy_read_data(), LogicalRepApplyLoop(), and WalReceiverMain().

walrcv_send

#define walrcv_send	(		conn,
			buffer,
			nbytes
	)		WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)

Definition at line 424 of file walreceiver.h.

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

walrcv_server_version

#define walrcv_server_version

(

conn

)

WalReceiverFunctions->walrcv_server_version(conn)

Definition at line 414 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and fetch_remote_table_info().

walrcv_startstreaming

#define walrcv_startstreaming	(		conn,
			options
	)		WalReceiverFunctions->walrcv_startstreaming(conn, options)

Definition at line 418 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and WalReceiverMain().

Typedef Documentation

walrcv_check_conninfo_fn

typedef void(* walrcv_check_conninfo_fn) (const char *conninfo)

Definition at line 246 of file walreceiver.h.

walrcv_connect_fn

typedef WalReceiverConn*(* walrcv_connect_fn) (const char *conninfo, bool logical, const char *appname, char **err)

Definition at line 236 of file walreceiver.h.

walrcv_create_slot_fn

typedef char*(* walrcv_create_slot_fn) (WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)

Definition at line 349 of file walreceiver.h.

walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 381 of file walreceiver.h.

walrcv_endstreaming_fn

typedef void(* walrcv_endstreaming_fn) (WalReceiverConn *conn, TimeLineID *next_tli)

Definition at line 314 of file walreceiver.h.

walrcv_exec_fn

typedef WalRcvExecResult*(* walrcv_exec_fn) (WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)

Definition at line 371 of file walreceiver.h.

walrcv_get_backend_pid_fn

typedef pid_t(* walrcv_get_backend_pid_fn) (WalReceiverConn *conn)

Definition at line 361 of file walreceiver.h.

walrcv_get_conninfo_fn

typedef char*(* walrcv_get_conninfo_fn) (WalReceiverConn *conn)

Definition at line 254 of file walreceiver.h.

walrcv_get_senderinfo_fn

typedef void(* walrcv_get_senderinfo_fn) (WalReceiverConn *conn, char **sender_host, int *sender_port)

Definition at line 263 of file walreceiver.h.

walrcv_identify_system_fn

typedef char*(* walrcv_identify_system_fn) (WalReceiverConn *conn, TimeLineID *primary_tli)

Definition at line 274 of file walreceiver.h.

walrcv_readtimelinehistoryfile_fn

typedef void(* walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *size)

Definition at line 291 of file walreceiver.h.

walrcv_receive_fn

typedef int(* walrcv_receive_fn) (WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)

Definition at line 325 of file walreceiver.h.

walrcv_send_fn

typedef void(* walrcv_send_fn) (WalReceiverConn *conn, const char *buffer, int nbytes)

Definition at line 335 of file walreceiver.h.

walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 282 of file walreceiver.h.

walrcv_startstreaming_fn

typedef bool(* walrcv_startstreaming_fn) (WalReceiverConn *conn, const WalRcvStreamOptions *options)

Definition at line 305 of file walreceiver.h.

WalRcvExecResult

typedef struct WalRcvExecResult WalRcvExecResult

WalReceiverConn

typedef struct WalReceiverConn WalReceiverConn

Definition at line 191 of file walreceiver.h.

WalReceiverFunctionsType

typedef struct WalReceiverFunctionsType WalReceiverFunctionsType

Enumeration Type Documentation

WalRcvExecStatus

enum WalRcvExecStatus

Enumerator
WALRCV_ERROR
WALRCV_OK_COMMAND
WALRCV_OK_TUPLES
WALRCV_OK_COPY_IN
WALRCV_OK_COPY_OUT
WALRCV_OK_COPY_BOTH

Definition at line 198 of file walreceiver.h.

{
    WALRCV_ERROR,               /* There was error when executing the query. */
    WALRCV_OK_COMMAND,          /* Query executed utility or replication
                                 * command. */
    WALRCV_OK_TUPLES,           /* Query returned tuples. */
    WALRCV_OK_COPY_IN,          /* Query started COPY FROM. */
    WALRCV_OK_COPY_OUT,         /* Query started COPY TO. */
    WALRCV_OK_COPY_BOTH         /* Query started COPY BOTH replication
                                 * protocol. */
} WalRcvExecStatus;

WalRcvState

enum WalRcvState

Enumerator
WALRCV_STOPPED
WALRCV_STARTING
WALRCV_STREAMING
WALRCV_WAITING
WALRCV_RESTARTING
WALRCV_STOPPING

Definition at line 45 of file walreceiver.h.

{
    WALRCV_STOPPED,             /* stopped and mustn't start up again */
    WALRCV_STARTING,            /* launched, but the process hasn't
                                 * initialized yet */
    WALRCV_STREAMING,           /* walreceiver is streaming */
    WALRCV_WAITING,             /* stopped streaming, waiting for orders */
    WALRCV_RESTARTING,          /* asked to restart streaming */
    WALRCV_STOPPING             /* requested to stop, but still running */
} WalRcvState;

Function Documentation

GetReplicationApplyDelay()

int GetReplicationApplyDelay

(

void

)

Definition at line 364 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 394 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 331 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 352 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);
}

ProcessWalRcvInterrupts()

void ProcessWalRcvInterrupts

(

void

)

Definition at line 146 of file walreceiver.c.

References CHECK_FOR_INTERRUPTS, ereport, errcode(), errmsg(), FATAL, and ShutdownRequestPending.

Referenced by libpqrcv_connect(), libpqrcv_PQgetResult(), libpqrcv_processTuples(), walrcv_clear_result(), WalRcvWaitForStartPosition(), and WalReceiverMain().

{
    /*
     * Although walreceiver interrupt handling doesn't use the same scheme as
     * regular backends, call CHECK_FOR_INTERRUPTS() to make sure we receive
     * any incoming signals on Win32, and also to make sure we process any
     * barrier events.
     */
    CHECK_FOR_INTERRUPTS();

    if (ShutdownRequestPending)
    {
        ereport(FATAL,
                (errcode(ERRCODE_ADMIN_SHUTDOWN),
                 errmsg("terminating walreceiver process due to administrator command")));
    }
}

RequestXLogStreaming()

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

Definition at line 245 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 178 of file walreceiverfuncs.c.

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

Referenced by walrcv_clear_result(), and XLogShutdownWalRcv().

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

walrcv_clear_result()

static void walrcv_clear_result ( WalRcvExecResult *  walres )

inlinestatic

Definition at line 436 of file walreceiver.h.

References WalRcvExecResult::err, FreeTupleDesc(), GetReplicationApplyDelay(), GetReplicationTransferLatency(), GetWalRcvFlushRecPtr(), GetWalRcvWriteRecPtr(), pfree(), pg_attribute_noreturn, ProcessWalRcvInterrupts(), receiveTLI, RequestXLogStreaming(), ShutdownWalRcv(), WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_end(), WalRcvForceReply(), WalRcvRunning(), WalRcvShmemInit(), WalRcvShmemSize(), WalRcvStreaming(), and WalReceiverMain().

Referenced by copy_table(), fetch_remote_table_info(), fetch_table_list(), LogicalRepSyncTableStart(), and ReplicationSlotDropAtPubNode().

{
    if (!walres)
        return;

    if (walres->err)
        pfree(walres->err);

    if (walres->tuplestore)
        tuplestore_end(walres->tuplestore);

    if (walres->tupledesc)
        FreeTupleDesc(walres->tupledesc);

    pfree(walres);
}

WalRcvForceReply()

void WalRcvForceReply

(

void

)

Definition at line 1260 of file walreceiver.c.

References WalRcvData::force_reply, WalRcvData::latch, WalRcvData::mutex, SetLatch(), SpinLockAcquire, SpinLockRelease, and WalRcv.

Referenced by StartupXLOG(), WaitForWALToBecomeAvailable(), and walrcv_clear_result().

{
    Latch      *latch;

    WalRcv->force_reply = true;
    /* fetching the latch pointer might not be atomic, so use spinlock */
    SpinLockAcquire(&WalRcv->mutex);
    latch = WalRcv->latch;
    SpinLockRelease(&WalRcv->mutex);
    if (latch)
        SetLatch(latch);
}

WalRcvRunning()

bool WalRcvRunning

(

void

)

Definition at line 75 of file walreceiverfuncs.c.

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

WalRcvShmemInit()

void WalRcvShmemInit

(

void

)

Definition at line 54 of file walreceiverfuncs.c.

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

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;
        ConditionVariableInit(&WalRcv->walRcvStoppedCV);
        SpinLockInit(&WalRcv->mutex);
        pg_atomic_init_u64(&WalRcv->writtenUpto, 0);
        WalRcv->latch = NULL;
    }
}

WalRcvShmemSize()

Size WalRcvShmemSize

(

void

)

Definition at line 43 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 126 of file walreceiverfuncs.c.

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

WalReceiverMain()

void WalReceiverMain

(

void

)

Definition at line 167 of file walreceiver.c.

References ARCHIVE_MODE_ALWAYS, Assert, buf, close, cluster_name, ConditionVariableBroadcast(), ConfigReloadPending, WalRcvData::conninfo, DEBUG1, elog, ereport, errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, FATAL, WalRcvData::force_reply, GetCurrentTimestamp(), GetSystemIdentifier(), GetXLogReplayRecPtr(), initStringInfo(), WalRcvData::is_temp_slot, WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::latch, WalRcvData::latestWalEndTime, load_file(), LOG, WalRcvStreamOptions::logical, LogstreamResult, LSN_FORMAT_ARGS, MAXCONNINFO, MAXFNAMELEN, WalRcvData::mutex, MyLatch, MyProc, MyProcPid, NAMEDATALEN, NAPTIME_PER_CYCLE, NI_MAXHOST, now(), on_shmem_exit(), options, PANIC, pfree(), pg_atomic_write_u64(), pg_memory_barrier, PG_SETMASK, PGC_SIGHUP, PGINVALID_SOCKET, WalRcvStreamOptions::physical, WalRcvData::pid, pqsignal(), proc_exit(), ProcessConfigFile(), ProcessWalRcvInterrupts(), PGPROC::procLatch, procsignal_sigusr1_handler(), WalRcvStreamOptions::proto, WalRcvData::ready_to_display, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, RecoveryInProgress(), recvFile, recvFileTLI, recvSegNo, ResetLatch(), WalRcvData::sender_host, WalRcvData::sender_port, SIG_DFL, SIG_IGN, SIGALRM, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SIGPIPE, SIGUSR1, SIGUSR2, WalRcvData::slotname, WalRcvStreamOptions::slotname, snprintf, SpinLockAcquire, SpinLockRelease, WalRcvStreamOptions::startpoint, strlcpy(), ThisTimeLineID, TimestampTzPlusMilliseconds, UINT64_FORMAT, UnBlockSig, WAIT_EVENT_WAL_RECEIVER_MAIN, WaitLatchOrSocket(), wal_receiver_timeout, wal_segment_size, WalRcv, walrcv_connect, 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, WalRcvDie(), WalRcvFetchTimeLineHistoryFiles(), WalRcvData::walRcvState, WalRcvData::walRcvStoppedCV, WalRcvWaitForStartPosition(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_READABLE, WL_TIMEOUT, WalRcvData::writtenUpto, XLogArchiveForceDone(), XLogArchiveMode, XLogArchiveNotify(), XLogFileName, XLogWalRcvFlush(), XLogWalRcvProcessMsg(), XLogWalRcvSendHSFeedback(), and XLogWalRcvSendReply().

Referenced by AuxiliaryProcessMain(), and walrcv_clear_result().

{
    char        conninfo[MAXCONNINFO];
    char       *tmp_conninfo;
    char        slotname[NAMEDATALEN];
    bool        is_temp_slot;
    XLogRecPtr  startpoint;
    TimeLineID  startpointTLI;
    TimeLineID  primaryTLI;
    bool        first_stream;
    WalRcvData *walrcv = WalRcv;
    TimestampTz last_recv_timestamp;
    TimestampTz now;
    bool        ping_sent;
    char       *err;
    char       *sender_host = NULL;
    int         sender_port = 0;

    /*
     * WalRcv should be set up already (if we are a backend, we inherit this
     * by fork() or EXEC_BACKEND mechanism from the postmaster).
     */
    Assert(walrcv != NULL);

    now = GetCurrentTimestamp();

    /*
     * 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_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_STREAMING;

    /* Fetch information required to start streaming */
    walrcv->ready_to_display = false;
    strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
    strlcpy(slotname, (char *) 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 */
    walrcv->lastMsgSendTime =
        walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;

    /* Report the latch to use to awaken this process */
    walrcv->latch = &MyProc->procLatch;

    SpinLockRelease(&walrcv->mutex);

    pg_atomic_write_u64(&WalRcv->writtenUpto, 0);

    /* Arrange to clean up at walreceiver exit */
    on_shmem_exit(WalRcvDie, 0);

    /* 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, SignalHandlerForShutdownRequest); /* 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) */
    PG_SETMASK(&UnBlockSig);

    /* Establish the connection to the primary for XLOG streaming */
    wrconn = walrcv_connect(conninfo, false,
                            cluster_name[0] ? cluster_name : "walreceiver",
                            &err);
    if (!wrconn)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("could not connect to the primary server: %s", 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((char *) walrcv->conninfo, tmp_conninfo, MAXCONNINFO);

    memset(walrcv->sender_host, 0, NI_MAXHOST);
    if (sender_host)
        strlcpy((char *) 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, 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;
        ThisTimeLineID = startpointTLI;
        if (walrcv_startstreaming(wrconn, &options))
        {
            if (first_stream)
                ereport(LOG,
                        (errmsg("started streaming WAL from primary at %X/%X on timeline %u",
                                LSN_FORMAT_ARGS(startpoint), startpointTLI)));
            else
                ereport(LOG,
                        (errmsg("restarted WAL streaming at %X/%X on timeline %u",
                                LSN_FORMAT_ARGS(startpoint), startpointTLI)));
            first_stream = false;

            /* Initialize LogstreamResult and buffers for processing messages */
            LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
            initStringInfo(&reply_message);
            initStringInfo(&incoming_message);

            /* Initialize the last recv timestamp */
            last_recv_timestamp = GetCurrentTimestamp();
            ping_sent = false;

            /* Loop until end-of-streaming or error */
            for (;;)
            {
                char       *buf;
                int         len;
                bool        endofwal = false;
                pgsocket    wait_fd = PGINVALID_SOCKET;
                int         rc;

                /*
                 * 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 */
                ProcessWalRcvInterrupts();

                if (ConfigReloadPending)
                {
                    ConfigReloadPending = false;
                    ProcessConfigFile(PGC_SIGHUP);
                    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 reset
                             * timeout
                             */
                            last_recv_timestamp = GetCurrentTimestamp();
                            ping_sent = false;
                            XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1);
                        }
                        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/%X.",
                                               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);
                }

                /* Check if we need to exit the streaming loop. */
                if (endofwal)
                    break;

                /*
                 * 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,
                                       NAPTIME_PER_CYCLE,
                                       WAIT_EVENT_WAL_RECEIVER_MAIN);
                if (rc & WL_LATCH_SET)
                {
                    ResetLatch(MyLatch);
                    ProcessWalRcvInterrupts();

                    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;

                    /*
                     * Check if time since last receive from primary has
                     * reached the configured limit.
                     */
                    if (wal_receiver_timeout > 0)
                    {
                        TimestampTz now = GetCurrentTimestamp();
                        TimestampTz timeout;

                        timeout =
                            TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                        wal_receiver_timeout);

                        if (now >= timeout)
                            ereport(ERROR,
                                    (errcode(ERRCODE_CONNECTION_FAILURE),
                                     errmsg("terminating walreceiver due to timeout")));

                        /*
                         * We didn't receive anything new, for half of
                         * receiver replication timeout. Ping the server.
                         */
                        if (!ping_sent)
                        {
                            timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                                  (wal_receiver_timeout / 2));
                            if (now >= timeout)
                            {
                                requestReply = true;
                                ping_sent = true;
                            }
                        }
                    }

                    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);
            XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
            if (close(recvFile) != 0)
                ereport(PANIC,
                        (errcode_for_file_access(),
                         errmsg("could not close log 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 */
}

Variable Documentation

hot_standby_feedback

bool hot_standby_feedback

Definition at line 91 of file walreceiver.c.

Referenced by XLogWalRcvSendHSFeedback().

wal_receiver_status_interval

int wal_receiver_status_interval

Definition at line 89 of file walreceiver.c.

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

wal_receiver_timeout

int wal_receiver_timeout

Definition at line 90 of file walreceiver.c.

Referenced by logicalrep_worker_launch(), LogicalRepApplyLoop(), and WalReceiverMain().

WalRcv

WalRcvData* WalRcv

Definition at line 33 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().

WalReceiverFunctions

PGDLLIMPORT WalReceiverFunctionsType* WalReceiverFunctions

Definition at line 95 of file walreceiver.c.

Referenced by _PG_init().