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/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_CONNECTING , 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)
pg_noreturn void	WalReceiverMain (const void *startup_data, size_t startup_data_len)
void	WalRcvForceReply (void)
Size	WalRcvShmemSize (void)
void	WalRcvShmemInit (void)
void	ShutdownWalRcv (void)
bool	WalRcvStreaming (void)
bool	WalRcvRunning (void)
WalRcvState	WalRcvGetState (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 40 of file walreceiver.h.

MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 37 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 462 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);
}
 
/* prototypes for functions in walreceiver.c */
pg_noreturn extern void WalReceiverMain(const void *startup_data, size_t startup_data_len);
extern void WalRcvForceReply(void);
 
/* prototypes for functions in walreceiverfuncs.c */
extern Size WalRcvShmemSize(void);
extern void WalRcvShmemInit(void);
extern void ShutdownWalRcv(void);
extern bool WalRcvStreaming(void);
extern bool WalRcvRunning(void);
extern WalRcvState WalRcvGetState(void);
extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
                                 const char *conninfo, const char *slotname,
                                 bool create_temp_slot);
extern XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI);
extern XLogRecPtr GetWalRcvWriteRecPtr(void);
extern int  GetReplicationApplyDelay(void);
extern int  GetReplicationTransferLatency(void);
 
#endif                          /* _WALRECEIVER_H */

walrcv_check_conninfo

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

Definition at line 438 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 436 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 460 of file walreceiver.h.

walrcv_disconnect

#define walrcv_disconnect

(

conn

)

WalReceiverFunctions->walrcv_disconnect(conn)

Definition at line 468 of file walreceiver.h.

walrcv_endstreaming

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

Definition at line 454 of file walreceiver.h.

walrcv_exec

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

Definition at line 466 of file walreceiver.h.

walrcv_get_backend_pid

#define walrcv_get_backend_pid

(

conn

)

WalReceiverFunctions->walrcv_get_backend_pid(conn)

Definition at line 464 of file walreceiver.h.

walrcv_get_conninfo

#define walrcv_get_conninfo

(

conn

)

WalReceiverFunctions->walrcv_get_conninfo(conn)

Definition at line 440 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 446 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 442 of file walreceiver.h.

walrcv_identify_system

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

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

walrcv_receive

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

Definition at line 456 of file walreceiver.h.

walrcv_send

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

Definition at line 458 of file walreceiver.h.

walrcv_server_version

#define walrcv_server_version

(

conn

)

WalReceiverFunctions->walrcv_server_version(conn)

Definition at line 448 of file walreceiver.h.

walrcv_startstreaming

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

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

walrcv_check_conninfo_fn

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

Definition at line 255 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 243 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 366 of file walreceiver.h.

walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 411 of file walreceiver.h.

walrcv_endstreaming_fn

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

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

walrcv_get_backend_pid_fn

typedef pid_t(* walrcv_get_backend_pid_fn) (WalReceiverConn *conn)

Definition at line 391 of file walreceiver.h.

walrcv_get_conninfo_fn

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

Definition at line 264 of file walreceiver.h.

walrcv_get_dbname_from_conninfo_fn

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

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

walrcv_identify_system_fn

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

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

walrcv_receive_fn

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

Definition at line 342 of file walreceiver.h.

walrcv_send_fn

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

Definition at line 352 of file walreceiver.h.

walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 299 of file walreceiver.h.

walrcv_startstreaming_fn

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

Definition at line 322 of file walreceiver.h.

WalRcvExecResult

typedef struct WalRcvExecResult WalRcvExecResult

WalReceiverConn

typedef struct WalReceiverConn WalReceiverConn

Definition at line 196 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 203 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_CONNECTING
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_CONNECTING,          /* connecting to upstream server */
    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  )

extern

Definition at line 380 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 fb(), GetCurrentChunkReplayStartTime(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void  )

extern

Definition at line 410 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 fb(), SpinLockAcquire, SpinLockRelease, TimestampDifferenceMilliseconds(), and WalRcv.

Referenced by ProcessWalSndrMessage().

GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr ( XLogRecPtr *  latestChunkStart, TimeLineID *  receiveTLI  )

extern

Definition at line 347 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 fb(), receiveTLI, SpinLockAcquire, SpinLockRelease, and WalRcv.

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

GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( void  )

extern

Definition at line 368 of file walreceiverfuncs.c.

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

References fb(), pg_atomic_read_u64(), and WalRcv.

Referenced by GetCurrentLSNForWaitType().

RequestXLogStreaming()

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

extern

Definition at line 261 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    bool        launch = false;
    pg_time_t   now = (pg_time_t) time(NULL);
    ProcNumber  walrcv_proc;
 
    /*
     * We always start at the beginning of the segment. That prevents a broken
     * segment (i.e., with no records in the first half of a segment) from
     * being created by XLOG streaming, which might cause trouble later on if
     * the segment is e.g archived.
     */
    if (XLogSegmentOffset(recptr, wal_segment_size) != 0)
        recptr -= XLogSegmentOffset(recptr, wal_segment_size);
 
    SpinLockAcquire(&walrcv->mutex);
 
    /* It better be stopped if we try to restart it */
    Assert(walrcv->walRcvState == WALRCV_STOPPED ||
           walrcv->walRcvState == WALRCV_WAITING);
 
    if (conninfo != NULL)
        strlcpy(walrcv->conninfo, conninfo, MAXCONNINFO);
    else
        walrcv->conninfo[0] = '\0';
 
    /*
     * Use configured replication slot if present, and ignore the value of
     * create_temp_slot as the slot name should be persistent.  Otherwise, use
     * create_temp_slot to determine whether this WAL receiver should create a
     * temporary slot by itself and use it, or not.
     */
    if (slotname != NULL && slotname[0] != '\0')
    {
        strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
        walrcv->is_temp_slot = false;
    }
    else
    {
        walrcv->slotname[0] = '\0';
        walrcv->is_temp_slot = create_temp_slot;
    }
 
    if (walrcv->walRcvState == WALRCV_STOPPED)
    {
        launch = true;
        walrcv->walRcvState = WALRCV_STARTING;
    }
    else
        walrcv->walRcvState = WALRCV_RESTARTING;
    walrcv->startTime = now;
 
    /*
     * If this is the first startup of walreceiver (on this timeline),
     * initialize flushedUpto and latestChunkStart to the starting point.
     */
    if (!XLogRecPtrIsValid(walrcv->receiveStart) || walrcv->receivedTLI != tli)
    {
        walrcv->flushedUpto = recptr;
        walrcv->receivedTLI = tli;
        walrcv->latestChunkStart = recptr;
    }
    walrcv->receiveStart = recptr;
    walrcv->receiveStartTLI = tli;
 
    walrcv_proc = walrcv->procno;
 
    SpinLockRelease(&walrcv->mutex);
 
    if (launch)
        SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
    else if (walrcv_proc != INVALID_PROC_NUMBER)
        SetLatch(&GetPGProcByNumber(walrcv_proc)->procLatch);
}

References Assert, fb(), GetPGProcByNumber, INVALID_PROC_NUMBER, MAXCONNINFO, NAMEDATALEN, now(), PMSIGNAL_START_WALRECEIVER, SendPostmasterSignal(), SetLatch(), SpinLockAcquire, SpinLockRelease, strlcpy(), wal_segment_size, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_WAITING, XLogRecPtrIsValid, and XLogSegmentOffset.

Referenced by WaitForWALToBecomeAvailable().

ShutdownWalRcv()

void ShutdownWalRcv ( void  )

extern

Definition at line 193 of file walreceiverfuncs.c.

{
    WalRcvData *walrcv = WalRcv;
    pid_t       walrcvpid = 0;
    bool        stopped = false;
 
    /*
     * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
     * mode once it's finished, and will also request postmaster to not
     * restart itself.
     */
    SpinLockAcquire(&walrcv->mutex);
    switch (walrcv->walRcvState)
    {
        case WALRCV_STOPPED:
            break;
        case WALRCV_STARTING:
            walrcv->walRcvState = WALRCV_STOPPED;
            stopped = true;
            break;
 
        case WALRCV_CONNECTING:
        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(), fb(), kill, SpinLockAcquire, SpinLockRelease, WalRcv, WALRCV_CONNECTING, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, and WalRcvRunning().

Referenced by XLogShutdownWalRcv().

walrcv_clear_result()

static void walrcv_clear_result ( WalRcvExecResult *  walres )

inlinestatic

Definition at line 472 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 fb(), FreeTupleDesc(), pfree(), and tuplestore_end().

Referenced by check_pub_dead_tuple_retention(), check_publications(), check_publications_origin_sequences(), check_publications_origin_tables(), copy_sequences(), copy_table(), fetch_relation_list(), fetch_remote_slots(), fetch_remote_table_info(), LogicalRepSyncTableStart(), ReplicationSlotDropAtPubNode(), and validate_remote_info().

WalRcvForceReply()

void WalRcvForceReply ( void  )

extern

Definition at line 1381 of file walreceiver.c.

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

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

Referenced by ApplyWalRecord(), and WaitForWALToBecomeAvailable().

WalRcvGetState()

WalRcvState WalRcvGetState ( void  )

extern

Definition at line 124 of file walreceiverfuncs.c.

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

References fb(), SpinLockAcquire, SpinLockRelease, and WalRcv.

Referenced by WaitForWALToBecomeAvailable().

WalRcvRunning()

bool WalRcvRunning ( void  )

extern

Definition at line 76 of file walreceiverfuncs.c.

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

References ConditionVariableBroadcast(), fb(), now(), SpinLockAcquire, SpinLockRelease, WalRcv, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, and WALRCV_STOPPED.

Referenced by ShutdownWalRcv(), and StartupRequestWalReceiverRestart().

WalRcvShmemInit()

void WalRcvShmemInit ( void  )

extern

Definition at line 55 of file walreceiverfuncs.c.

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

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

Referenced by CreateOrAttachShmemStructs().

WalRcvShmemSize()

Size WalRcvShmemSize ( void  )

extern

Definition at line 44 of file walreceiverfuncs.c.

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

References add_size().

Referenced by CalculateShmemSize(), and WalRcvShmemInit().

WalRcvStreaming()

bool WalRcvStreaming ( void  )

extern

Definition at line 141 of file walreceiverfuncs.c.

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

References ConditionVariableBroadcast(), fb(), now(), SpinLockAcquire, SpinLockRelease, WalRcv, WALRCV_CONNECTING, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STARTUP_TIMEOUT, WALRCV_STOPPED, and WALRCV_STREAMING.

Referenced by FinishWalRecovery(), and WaitForWALToBecomeAvailable().

WalReceiverMain()

pg_noreturn void WalReceiverMain ( const void *  startup_data, size_t  startup_data_len  )

extern

Definition at line 153 of file walreceiver.c.

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

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

Variable Documentation

hot_standby_feedback

PGDLLIMPORT bool hot_standby_feedback

extern

Definition at line 91 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 89 of file walreceiver.c.

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

wal_receiver_timeout

PGDLLIMPORT int wal_receiver_timeout

extern

Definition at line 90 of file walreceiver.c.

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

WalRcv

PGDLLIMPORT WalRcvData* WalRcv

extern

Definition at line 34 of file walreceiverfuncs.c.

Referenced by GetReplicationApplyDelay(), GetReplicationTransferLatency(), GetWalRcvFlushRecPtr(), GetWalRcvWriteRecPtr(), pg_stat_get_wal_receiver(), ProcessWalSndrMessage(), RequestXLogStreaming(), ShutdownWalRcv(), WalRcvDie(), WalRcvForceReply(), WalRcvGetState(), WalRcvRunning(), WalRcvShmemInit(), WalRcvStreaming(), WalRcvWaitForStartPosition(), WalReceiverMain(), XLogWalRcvFlush(), and XLogWalRcvWrite().

WalReceiverFunctions

PGDLLIMPORT WalReceiverFunctionsType* WalReceiverFunctions

extern

Definition at line 95 of file walreceiver.c.

Referenced by _PG_init(), and WalReceiverMain().