walreceiver.h File Reference

#include <netdb.h>
#include "access/xlog.h"
#include "access/xlogdefs.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, replication, logical, must_use_password, appname, err)    WalReceiverFunctions->walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
#define	walrcv_check_conninfo(conninfo, must_use_password)    WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)
#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_get_dbname_from_conninfo(conninfo)    WalReceiverFunctions->walrcv_get_dbname_from_conninfo(conninfo)
#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, failover, snapshot_action, lsn)    WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
#define	walrcv_alter_slot(conn, slotname, failover, two_phase)    WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)
#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 replication, bool logical, bool must_use_password, const char *appname, char **err)
typedef void(*	walrcv_check_conninfo_fn) (const char *conninfo, bool must_use_password)
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 char *(*	walrcv_get_dbname_from_conninfo_fn) (const char *conninfo)
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, bool failover, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
typedef void(*	walrcv_alter_slot_fn) (WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)
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 (char *startup_data, size_t startup_data_len) pg_attribute_noreturn()
void	ProcessWalRcvInterrupts (void)
void	WalRcvForceReply (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)

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

Macro Definition Documentation

AllowCascadeReplication

#define AllowCascadeReplication

(

)

(EnableHotStandby && max_wal_senders > 0)

Definition at line 41 of file walreceiver.h.

MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 38 of file walreceiver.h.

walrcv_alter_slot

#define walrcv_alter_slot	(		conn,
			slotname,
			failover,
			two_phase
	)		WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)

Definition at line 460 of file walreceiver.h.

walrcv_check_conninfo

#define walrcv_check_conninfo	(		conninfo,
			must_use_password
	)		WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)

Definition at line 436 of file walreceiver.h.

walrcv_connect

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

Definition at line 434 of file walreceiver.h.

walrcv_create_slot

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

Definition at line 458 of file walreceiver.h.

walrcv_disconnect

#define walrcv_disconnect

(

conn

)

WalReceiverFunctions->walrcv_disconnect(conn)

Definition at line 466 of file walreceiver.h.

walrcv_endstreaming

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

Definition at line 452 of file walreceiver.h.

walrcv_exec

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

Definition at line 464 of file walreceiver.h.

walrcv_get_backend_pid

#define walrcv_get_backend_pid

(

conn

)

WalReceiverFunctions->walrcv_get_backend_pid(conn)

Definition at line 462 of file walreceiver.h.

walrcv_get_conninfo

#define walrcv_get_conninfo

(

conn

)

WalReceiverFunctions->walrcv_get_conninfo(conn)

Definition at line 438 of file walreceiver.h.

walrcv_get_dbname_from_conninfo

#define walrcv_get_dbname_from_conninfo

(

conninfo

)

WalReceiverFunctions->walrcv_get_dbname_from_conninfo(conninfo)

Definition at line 444 of file walreceiver.h.

walrcv_get_senderinfo

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

Definition at line 440 of file walreceiver.h.

walrcv_identify_system

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

Definition at line 442 of file walreceiver.h.

walrcv_readtimelinehistoryfile

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

Definition at line 448 of file walreceiver.h.

walrcv_receive

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

Definition at line 454 of file walreceiver.h.

walrcv_send

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

Definition at line 456 of file walreceiver.h.

walrcv_server_version

#define walrcv_server_version

(

conn

)

WalReceiverFunctions->walrcv_server_version(conn)

Definition at line 446 of file walreceiver.h.

walrcv_startstreaming

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

Definition at line 450 of file walreceiver.h.

Typedef Documentation

walrcv_alter_slot_fn

typedef void(* walrcv_alter_slot_fn) (WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)

Definition at line 378 of file walreceiver.h.

walrcv_check_conninfo_fn

typedef void(* walrcv_check_conninfo_fn) (const char *conninfo, bool must_use_password)

Definition at line 253 of file walreceiver.h.

walrcv_connect_fn

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

Definition at line 241 of file walreceiver.h.

walrcv_create_slot_fn

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

Definition at line 364 of file walreceiver.h.

walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 409 of file walreceiver.h.

walrcv_endstreaming_fn

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

Definition at line 329 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 399 of file walreceiver.h.

walrcv_get_backend_pid_fn

typedef pid_t(* walrcv_get_backend_pid_fn) (WalReceiverConn *conn)

Definition at line 389 of file walreceiver.h.

walrcv_get_conninfo_fn

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

Definition at line 262 of file walreceiver.h.

walrcv_get_dbname_from_conninfo_fn

typedef char*(* walrcv_get_dbname_from_conninfo_fn) (const char *conninfo)

Definition at line 290 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 271 of file walreceiver.h.

walrcv_identify_system_fn

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

Definition at line 282 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 306 of file walreceiver.h.

walrcv_receive_fn

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

Definition at line 340 of file walreceiver.h.

walrcv_send_fn

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

Definition at line 350 of file walreceiver.h.

walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 297 of file walreceiver.h.

walrcv_startstreaming_fn

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

Definition at line 320 of file walreceiver.h.

WalRcvExecResult

typedef struct WalRcvExecResult WalRcvExecResult

WalReceiverConn

typedef struct WalReceiverConn WalReceiverConn

Definition at line 164 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 201 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 46 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.

{
    WalRcvData *walrcv = WalRcv;
    XLogRecPtr  receivePtr;
    XLogRecPtr  replayPtr;
    TimestampTz chunkReplayStartTime;
 
    SpinLockAcquire(&walrcv->mutex);
    receivePtr = walrcv->flushedUpto;
    SpinLockRelease(&walrcv->mutex);
 
    replayPtr = GetXLogReplayRecPtr(NULL);
 
    if (receivePtr == replayPtr)
        return 0;
 
    chunkReplayStartTime = GetCurrentChunkReplayStartTime();
 
    if (chunkReplayStartTime == 0)
        return -1;
 
    return TimestampDifferenceMilliseconds(chunkReplayStartTime,
                                           GetCurrentTimestamp());
}

References WalRcvData::flushedUpto, GetCurrentChunkReplayStartTime(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

GetReplicationTransferLatency()

int GetReplicationTransferLatency

(

void

)

Definition at line 394 of file walreceiverfuncs.c.

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

References WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::mutex, SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 331 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    XLogRecPtr  recptr;
 
    SpinLockAcquire(&walrcv->mutex);
    recptr = walrcv->flushedUpto;
    if (latestChunkStart)
        *latestChunkStart = walrcv->latestChunkStart;
    if (receiveTLI)
        *receiveTLI = walrcv->receivedTLI;
    SpinLockRelease(&walrcv->mutex);
 
    return recptr;
}

References WalRcvData::flushedUpto, WalRcvData::latestChunkStart, WalRcvData::mutex, WalRcvData::receivedTLI, receiveTLI, SpinLockAcquire, SpinLockRelease, and WalRcv.

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

GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr

(

void

)

Definition at line 352 of file walreceiverfuncs.c.

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

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

ProcessWalRcvInterrupts()

void ProcessWalRcvInterrupts

(

void

)

Definition at line 162 of file walreceiver.c.

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

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

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

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.

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

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().

ShutdownWalRcv()

void ShutdownWalRcv

(

void

)

Definition at line 178 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    pid_t       walrcvpid = 0;
    bool        stopped = false;
 
    /*
     * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
     * mode once it's finished, and will also request postmaster to not
     * restart itself.
     */
    SpinLockAcquire(&walrcv->mutex);
    switch (walrcv->walRcvState)
    {
        case WALRCV_STOPPED:
            break;
        case WALRCV_STARTING:
            walrcv->walRcvState = WALRCV_STOPPED;
            stopped = true;
            break;
 
        case WALRCV_STREAMING:
        case WALRCV_WAITING:
        case WALRCV_RESTARTING:
            walrcv->walRcvState = WALRCV_STOPPING;
            /* fall through */
        case WALRCV_STOPPING:
            walrcvpid = walrcv->pid;
            break;
    }
    SpinLockRelease(&walrcv->mutex);
 
    /* Unnecessary but consistent. */
    if (stopped)
        ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
 
    /*
     * Signal walreceiver process if it was still running.
     */
    if (walrcvpid != 0)
        kill(walrcvpid, SIGTERM);
 
    /*
     * Wait for walreceiver to acknowledge its death by setting state to
     * WALRCV_STOPPED.
     */
    ConditionVariablePrepareToSleep(&walrcv->walRcvStoppedCV);
    while (WalRcvRunning())
        ConditionVariableSleep(&walrcv->walRcvStoppedCV,
                               WAIT_EVENT_WAL_RECEIVER_EXIT);
    ConditionVariableCancelSleep();
}

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

Referenced by XLogShutdownWalRcv().

walrcv_clear_result()

static void walrcv_clear_result ( WalRcvExecResult *  walres )

inlinestatic

Definition at line 470 of file walreceiver.h.

{
    if (!walres)
        return;
 
    if (walres->err)
        pfree(walres->err);
 
    if (walres->tuplestore)
        tuplestore_end(walres->tuplestore);
 
    if (walres->tupledesc)
        FreeTupleDesc(walres->tupledesc);
 
    pfree(walres);
}

References WalRcvExecResult::err, FreeTupleDesc(), pfree(), WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, and tuplestore_end().

Referenced by check_publications(), check_publications_origin(), copy_table(), fetch_remote_table_info(), fetch_table_list(), LogicalRepSyncTableStart(), ReplicationSlotDropAtPubNode(), synchronize_slots(), and validate_remote_info().

WalRcvForceReply()

void WalRcvForceReply

(

void

)

Definition at line 1359 of file walreceiver.c.

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

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

Referenced by ApplyWalRecord(), and WaitForWALToBecomeAvailable().

WalRcvRunning()

bool WalRcvRunning

(

void

)

Definition at line 75 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    WalRcvState state;
    pg_time_t   startTime;
 
    SpinLockAcquire(&walrcv->mutex);
 
    state = walrcv->walRcvState;
    startTime = walrcv->startTime;
 
    SpinLockRelease(&walrcv->mutex);
 
    /*
     * If it has taken too long for walreceiver to start up, give up. Setting
     * the state to STOPPED ensures that if walreceiver later does start up
     * after all, it will see that it's not supposed to be running and die
     * without doing anything.
     */
    if (state == WALRCV_STARTING)
    {
        pg_time_t   now = (pg_time_t) time(NULL);
 
        if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
        {
            bool        stopped = false;
 
            SpinLockAcquire(&walrcv->mutex);
            if (walrcv->walRcvState == WALRCV_STARTING)
            {
                state = walrcv->walRcvState = WALRCV_STOPPED;
                stopped = true;
            }
            SpinLockRelease(&walrcv->mutex);
 
            if (stopped)
                ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
        }
    }
 
    if (state != WALRCV_STOPPED)
        return true;
    else
        return false;
}

References ConditionVariableBroadcast(), WalRcvData::mutex, now(), SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, WalRcv, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, WalRcvData::walRcvState, and WalRcvData::walRcvStoppedCV.

Referenced by ShutdownWalRcv(), and StartupRequestWalReceiverRestart().

WalRcvShmemInit()

void WalRcvShmemInit

(

void

)

Definition at line 54 of file walreceiverfuncs.c.

{
    bool        found;
 
    WalRcv = (WalRcvData *)
        ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &found);
 
    if (!found)
    {
        /* First time through, so initialize */
        MemSet(WalRcv, 0, WalRcvShmemSize());
        WalRcv->walRcvState = WALRCV_STOPPED;
        ConditionVariableInit(&WalRcv->walRcvStoppedCV);
        SpinLockInit(&WalRcv->mutex);
        pg_atomic_init_u64(&WalRcv->writtenUpto, 0);
        WalRcv->latch = NULL;
    }
}

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

Referenced by CreateOrAttachShmemStructs().

WalRcvShmemSize()

Size WalRcvShmemSize

(

void

)

Definition at line 43 of file walreceiverfuncs.c.

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

References add_size(), and size.

Referenced by CalculateShmemSize(), and WalRcvShmemInit().

WalRcvStreaming()

bool WalRcvStreaming

(

void

)

Definition at line 126 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    WalRcvState state;
    pg_time_t   startTime;
 
    SpinLockAcquire(&walrcv->mutex);
 
    state = walrcv->walRcvState;
    startTime = walrcv->startTime;
 
    SpinLockRelease(&walrcv->mutex);
 
    /*
     * If it has taken too long for walreceiver to start up, give up. Setting
     * the state to STOPPED ensures that if walreceiver later does start up
     * after all, it will see that it's not supposed to be running and die
     * without doing anything.
     */
    if (state == WALRCV_STARTING)
    {
        pg_time_t   now = (pg_time_t) time(NULL);
 
        if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
        {
            bool        stopped = false;
 
            SpinLockAcquire(&walrcv->mutex);
            if (walrcv->walRcvState == WALRCV_STARTING)
            {
                state = walrcv->walRcvState = WALRCV_STOPPED;
                stopped = true;
            }
            SpinLockRelease(&walrcv->mutex);
 
            if (stopped)
                ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
        }
    }
 
    if (state == WALRCV_STREAMING || state == WALRCV_STARTING ||
        state == WALRCV_RESTARTING)
        return true;
    else
        return false;
}

References ConditionVariableBroadcast(), WalRcvData::mutex, now(), SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, WALRCV_STREAMING, WalRcvData::walRcvState, and WalRcvData::walRcvStoppedCV.

Referenced by FinishWalRecovery(), and WaitForWALToBecomeAvailable().

WalReceiverMain()

void WalReceiverMain	(	char *	startup_data,
		size_t	startup_data_len
	)

Definition at line 183 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);
 
    MyBackendType = B_WAL_RECEIVER;
    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_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 */
    now = GetCurrentTimestamp();
    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, 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, 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) */
    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((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, 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/%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);
 
            /* 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 */
                ProcessWalRcvInterrupts();
 
                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/%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, 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);
                    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.
                     */
                    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(), B_WAL_RECEIVER, buf, close, cluster_name, ConditionVariableBroadcast(), ConfigReloadPending, WalRcvData::conninfo, DEBUG1, elog, ereport, err(), errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, FATAL, WalRcvData::force_reply, GetCurrentTimestamp(), GetSystemIdentifier(), GetXLogReplayRecPtr(), i, initStringInfo(), WalRcvData::is_temp_slot, WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::latch, WalRcvData::latestWalEndTime, len, load_file(), LOG, LogstreamResult, LSN_FORMAT_ARGS, MAXCONNINFO, MAXFNAMELEN, Min, WalRcvData::mutex, MyBackendType, MyLatch, MyProc, MyProcPid, NAMEDATALEN, now(), NUM_WALRCV_WAKEUPS, on_shmem_exit(), options, PANIC, pfree(), pg_atomic_write_u64(), pg_memory_barrier, PGC_SIGHUP, PGINVALID_SOCKET, WalRcvData::pid, PointerGetDatum(), pqsignal(), proc_exit(), ProcessConfigFile(), ProcessWalRcvInterrupts(), PGPROC::procLatch, procsignal_sigusr1_handler(), WalRcvData::ready_to_display, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, RecoveryInProgress(), recvFile, recvFileTLI, recvSegNo, reply_message, ResetLatch(), WalRcvData::sender_host, WalRcvData::sender_port, SIG_DFL, SIG_IGN, SIGALRM, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SIGPIPE, SIGUSR1, SIGUSR2, WalRcvData::slotname, snprintf, SpinLockAcquire, SpinLockRelease, strlcpy(), TIMESTAMP_INFINITY, TimestampDifferenceMilliseconds(), UINT64_FORMAT, UnBlockSig, WaitLatchOrSocket(), wakeup, 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, WALRCV_WAKEUP_PING, WALRCV_WAKEUP_TERMINATE, WalRcvComputeNextWakeup(), WalRcvDie(), WalRcvFetchTimeLineHistoryFiles(), WalRcvData::walRcvState, WalRcvData::walRcvStoppedCV, 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().

Variable Documentation

hot_standby_feedback

PGDLLIMPORT bool hot_standby_feedback

extern

Definition at line 89 of file walreceiver.c.

Referenced by slotsync_reread_config(), ValidateSlotSyncParams(), WalRcvComputeNextWakeup(), and XLogWalRcvSendHSFeedback().

wal_receiver_status_interval

PGDLLIMPORT int wal_receiver_status_interval

extern

Definition at line 87 of file walreceiver.c.

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

wal_receiver_timeout

PGDLLIMPORT int wal_receiver_timeout

extern

Definition at line 88 of file walreceiver.c.

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

WalRcv

PGDLLIMPORT WalRcvData* WalRcv

extern

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

WalReceiverFunctions

PGDLLIMPORT WalReceiverFunctionsType* WalReceiverFunctions

extern

Definition at line 93 of file walreceiver.c.

Referenced by _PG_init(), and WalReceiverMain().