#include "access/xlog.h"
#include "access/xlogdefs.h"
#include "fmgr.h"
#include "replication/logicalproto.h"
#include "replication/walsender.h"
#include "storage/latch.h"
#include "storage/spin.h"
#include "pgtime.h"
#include "utils/tuplestore.h"

Include dependency graph for walreceiver.h:

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

Go to the source code of this file.

Data Structures
struct	WalRcvData

struct	WalRcvStreamOptions

struct	WalRcvExecResult

struct	WalReceiverFunctionsType

Macros
#define	MAXCONNINFO 1024

#define	AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)

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

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

#define	walrcv_get_conninfo(conn) WalReceiverFunctions->walrcv_get_conninfo(conn)

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

#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, snapshot_action, lsn) WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn)

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

#define	walrcv_disconnect(conn) WalReceiverFunctions->walrcv_disconnect(conn)

Typedefs
typedef struct WalReceiverConn	WalReceiverConn

typedef struct WalRcvExecResult	WalRcvExecResult

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

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

typedef char (	walrcv_get_conninfo_fn )(WalReceiverConn *conn)

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

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, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)

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

typedef void(*	walrcv_disconnect_fn )(WalReceiverConn *conn)

typedef struct WalReceiverFunctionsType	WalReceiverFunctionsType

Enumerations
enum	WalRcvState { WALRCV_STOPPED, WALRCV_STARTING, WALRCV_STREAMING, WALRCV_WAITING, WALRCV_RESTARTING, WALRCV_STOPPING }

enum	WalRcvExecStatus { WALRCV_ERROR, WALRCV_OK_COMMAND, WALRCV_OK_TUPLES, WALRCV_OK_COPY_IN, WALRCV_OK_COPY_OUT, WALRCV_OK_COPY_BOTH }

Functions
static void	walrcv_clear_result (WalRcvExecResult *walres)

void	WalReceiverMain (void) pg_attribute_noreturn()

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)

XLogRecPtr	GetWalRcvWriteRecPtr (XLogRecPtr latestChunkStart, TimeLineID receiveTLI)

int	GetReplicationApplyDelay (void)

int	GetReplicationTransferLatency (void)

void	WalRcvForceReply (void)

Variables
int	wal_receiver_status_interval

int	wal_receiver_timeout

bool	hot_standby_feedback

WalRcvData *	WalRcv

PGDLLIMPORT WalReceiverFunctionsType *	WalReceiverFunctions

Macro Definition Documentation

#define AllowCascadeReplication ( ) (EnableHotStandby && max_wal_senders > 0)

Definition at line 38 of file walreceiver.h.

Referenced by StartupXLOG(), and XLogWalRcvFlush().

#define MAXCONNINFO 1024

Definition at line 35 of file walreceiver.h.

Referenced by RequestXLogStreaming(), and WalReceiverMain().

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

Definition at line 244 of file walreceiver.h.

Referenced by AlterSubscription(), and CreateSubscription().

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

Definition at line 242 of file walreceiver.h.

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

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

Definition at line 260 of file walreceiver.h.

Referenced by CreateSubscription(), and LogicalRepSyncTableStart().

#define walrcv_disconnect ( conn ) WalReceiverFunctions->walrcv_disconnect(conn)

Definition at line 264 of file walreceiver.h.

Referenced by AlterSubscription_refresh(), ApplyWorkerMain(), CreateSubscription(), DropSubscription(), pg_attribute_noreturn(), reread_subscription(), and WalRcvDie().

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

Definition at line 254 of file walreceiver.h.

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

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

Definition at line 262 of file walreceiver.h.

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

#define walrcv_get_conninfo ( conn ) WalReceiverFunctions->walrcv_get_conninfo(conn)

Definition at line 246 of file walreceiver.h.

Referenced by WalReceiverMain().

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

Definition at line 248 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and WalReceiverMain().

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

Definition at line 250 of file walreceiver.h.

Referenced by WalRcvFetchTimeLineHistoryFiles().

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

Definition at line 256 of file walreceiver.h.

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

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

Definition at line 258 of file walreceiver.h.

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

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

Definition at line 252 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and WalReceiverMain().

Typedef Documentation

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

Definition at line 197 of file walreceiver.h.

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

Definition at line 194 of file walreceiver.h.

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

Definition at line 214 of file walreceiver.h.

typedef void(* walrcv_disconnect_fn)(WalReceiverConn *conn)

Definition at line 222 of file walreceiver.h.

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

Definition at line 208 of file walreceiver.h.

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

Definition at line 218 of file walreceiver.h.

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

Definition at line 198 of file walreceiver.h.

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

Definition at line 199 of file walreceiver.h.

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

Definition at line 202 of file walreceiver.h.

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

Definition at line 210 of file walreceiver.h.

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

Definition at line 212 of file walreceiver.h.

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

Definition at line 206 of file walreceiver.h.

typedef struct WalRcvExecResult WalRcvExecResult

typedef struct WalReceiverConn WalReceiverConn

Definition at line 162 of file walreceiver.h.

typedef struct WalReceiverFunctionsType WalReceiverFunctionsType

Enumeration Type Documentation

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 169 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;

enum WalRcvState

Enumerator
WALRCV_STOPPED
WALRCV_STARTING
WALRCV_STREAMING
WALRCV_WAITING
WALRCV_RESTARTING
WALRCV_STOPPING

Definition at line 43 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

int GetReplicationApplyDelay ( void )

Definition at line 315 of file walreceiverfuncs.c.

References GetCurrentChunkReplayStartTime(), GetCurrentTimestamp(), GetXLogReplayRecPtr(), WalRcvData::mutex, NULL, WalRcvData::receivedUpto, SpinLockAcquire, SpinLockRelease, TimestampDifference(), and WalRcv.

Referenced by ProcessWalSndrMessage().

 {
     WalRcvData *walrcv = WalRcv;
     XLogRecPtr  receivePtr;
     XLogRecPtr  replayPtr;
 
     long        secs;
     int         usecs;
 
     TimestampTz chunkReplayStartTime;
 
     SpinLockAcquire(&walrcv->mutex);
     receivePtr = walrcv->receivedUpto;
     SpinLockRelease(&walrcv->mutex);
 
     replayPtr = GetXLogReplayRecPtr(NULL);
 
     if (receivePtr == replayPtr)
         return 0;
 
     chunkReplayStartTime = GetCurrentChunkReplayStartTime();
 
     if (chunkReplayStartTime == 0)
         return -1;
 
     TimestampDifference(chunkReplayStartTime,
                         GetCurrentTimestamp(),
                         &secs, &usecs);
 
     return (((int) secs * 1000) + (usecs / 1000));
 }

int GetReplicationTransferLatency ( void )

Definition at line 352 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage().

 {
     WalRcvData *walrcv = WalRcv;
 
     TimestampTz lastMsgSendTime;
     TimestampTz lastMsgReceiptTime;
 
     long        secs = 0;
     int         usecs = 0;
     int         ms;
 
     SpinLockAcquire(&walrcv->mutex);
     lastMsgSendTime = walrcv->lastMsgSendTime;
     lastMsgReceiptTime = walrcv->lastMsgReceiptTime;
     SpinLockRelease(&walrcv->mutex);
 
     TimestampDifference(lastMsgSendTime,
                         lastMsgReceiptTime,
                         &secs, &usecs);
 
     ms = ((int) secs * 1000) + (usecs / 1000);
 
     return ms;
 }

XLogRecPtr GetWalRcvWriteRecPtr	(	XLogRecPtr *	latestChunkStart,
		TimeLineID *	receiveTLI
	)

Definition at line 294 of file walreceiverfuncs.c.

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

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

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

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

Definition at line 223 of file walreceiverfuncs.c.

References Assert, WalRcvData::conninfo, WalRcvData::latch, WalRcvData::latestChunkStart, MAXCONNINFO, WalRcvData::mutex, NAMEDATALEN, now(), NULL, PMSIGNAL_START_WALRECEIVER, WalRcvData::receivedTLI, WalRcvData::receivedUpto, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, SendPostmasterSignal(), SetLatch(), WalRcvData::slotname, SpinLockAcquire, SpinLockRelease, WalRcvData::startTime, strlcpy(), WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_WAITING, WalRcvData::walRcvState, and XLogSegSize.

Referenced by WaitForWALToBecomeAvailable().

 {
     WalRcvData *walrcv = WalRcv;
     bool        launch = false;
     pg_time_t   now = (pg_time_t) time(NULL);
 
     /*
      * 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 (recptr % XLogSegSize != 0)
         recptr -= recptr % XLogSegSize;
 
     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';
 
     if (slotname != NULL)
         strlcpy((char *) walrcv->slotname, slotname, NAMEDATALEN);
     else
         walrcv->slotname[0] = '\0';
 
     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 receivedUpto and latestChunkStart to the starting point.
      */
     if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli)
     {
         walrcv->receivedUpto = recptr;
         walrcv->receivedTLI = tli;
         walrcv->latestChunkStart = recptr;
     }
     walrcv->receiveStart = recptr;
     walrcv->receiveStartTLI = tli;
 
     SpinLockRelease(&walrcv->mutex);
 
     if (launch)
         SendPostmasterSignal(PMSIGNAL_START_WALRECEIVER);
     else if (walrcv->latch)
         SetLatch(walrcv->latch);
 }

void ShutdownWalRcv ( void )

Definition at line 163 of file walreceiverfuncs.c.

References HandleStartupProcInterrupts(), WalRcvData::mutex, pg_usleep(), WalRcvData::pid, SpinLockAcquire, SpinLockRelease, WalRcv, WALRCV_RESTARTING, WALRCV_STARTING, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, WalRcvRunning(), and WalRcvData::walRcvState.

Referenced by StartupXLOG(), and WaitForWALToBecomeAvailable().

 {
     WalRcvData *walrcv = WalRcv;
     pid_t       walrcvpid = 0;
 
     /*
      * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
      * mode once it's finished, and will also request postmaster to not
      * restart itself.
      */
     SpinLockAcquire(&walrcv->mutex);
     switch (walrcv->walRcvState)
     {
         case WALRCV_STOPPED:
             break;
         case WALRCV_STARTING:
             walrcv->walRcvState = WALRCV_STOPPED;
             break;
 
         case WALRCV_STREAMING:
         case WALRCV_WAITING:
         case WALRCV_RESTARTING:
             walrcv->walRcvState = WALRCV_STOPPING;
             /* fall through */
         case WALRCV_STOPPING:
             walrcvpid = walrcv->pid;
             break;
     }
     SpinLockRelease(&walrcv->mutex);
 
     /*
      * Signal walreceiver process if it was still running.
      */
     if (walrcvpid != 0)
         kill(walrcvpid, SIGTERM);
 
     /*
      * Wait for walreceiver to acknowledge its death by setting state to
      * WALRCV_STOPPED.
      */
     while (WalRcvRunning())
     {
         /*
          * This possibly-long loop needs to handle interrupts of startup
          * process.
          */
         HandleStartupProcInterrupts();
 
         pg_usleep(100000);      /* 100ms */
     }
 }

static void walrcv_clear_result ( WalRcvExecResult * walres )

inlinestatic

Definition at line 268 of file walreceiver.h.

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

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

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

void WalRcvForceReply ( void )

Definition at line 1341 of file walreceiver.c.

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

Referenced by StartupXLOG(), and WaitForWALToBecomeAvailable().

 {
     WalRcv->force_reply = true;
     if (WalRcv->latch)
         SetLatch(WalRcv->latch);
 }

bool WalRcvRunning ( void )

Definition at line 72 of file walreceiverfuncs.c.

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

Referenced by ShutdownWalRcv().

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

void WalRcvShmemInit ( void )

Definition at line 53 of file walreceiverfuncs.c.

References WalRcvData::latch, MemSet, WalRcvData::mutex, NULL, ShmemInitStruct(), SpinLockInit, WALRCV_STOPPED, WalRcvShmemSize(), and WalRcvData::walRcvState.

Referenced by CreateSharedMemoryAndSemaphores().

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

Size WalRcvShmemSize ( void )

Definition at line 42 of file walreceiverfuncs.c.

References add_size().

Referenced by CreateSharedMemoryAndSemaphores(), and WalRcvShmemInit().

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

bool WalRcvStreaming ( void )

Definition at line 117 of file walreceiverfuncs.c.

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

Referenced by WaitForWALToBecomeAvailable().

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

void WalReceiverMain ( void )

Definition at line 188 of file walreceiver.c.

References ARCHIVE_MODE_ALWAYS, Assert, BlockSig, buf, close, WalRcvData::conninfo, CurrentResourceOwner, DEBUG1, DisableWalRcvImmediateExit(), elog, EnableWalRcvImmediateExit(), ereport, errcode_for_file_access(), errdetail(), errmsg(), ERROR, FATAL, WalRcvData::force_reply, GetCurrentTimestamp(), GetSystemIdentifier(), GetXLogReplayRecPtr(), got_SIGHUP, initStringInfo(), WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::latch, WalRcvData::latestWalEndTime, load_file(), LOG, WalRcvStreamOptions::logical, LogstreamResult, MAXCONNINFO, MAXFNAMELEN, WalRcvData::mutex, MyProc, MyProcPid, NAMEDATALEN, NAPTIME_PER_CYCLE, now(), NULL, on_shmem_exit(), options, PANIC, pfree(), pg_memory_barrier, PG_SETMASK, PGC_SIGHUP, PGINVALID_SOCKET, WalRcvStreamOptions::physical, WalRcvData::pid, pqsignal(), proc_exit(), ProcessConfigFile(), ProcessWalRcvInterrupts(), PGPROC::procLatch, WalRcvStreamOptions::proto, WalRcvData::ready_to_display, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, RecoveryInProgress(), recvFile, recvFileTLI, recvSegNo, ResetLatch(), ResourceOwnerCreate(), server_version, SIG_DFL, SIG_IGN, SIGALRM, SIGCHLD, SIGCONT, SIGHUP, SIGPIPE, SIGQUIT, SIGTTIN, SIGTTOU, SIGUSR1, SIGUSR2, SIGWINCH, WalRcvData::slotname, WalRcvStreamOptions::slotname, snprintf(), SpinLockAcquire, SpinLockRelease, WalRcvStreamOptions::startpoint, strlcpy(), ThisTimeLineID, TimestampTzPlusMilliseconds, UINT64_FORMAT, UnBlockSig, WAIT_EVENT_WAL_RECEIVER_MAIN, WaitLatchOrSocket(), wal_receiver_timeout, WalRcv, walrcv_connect, walrcv_endstreaming, walrcv_get_conninfo, walrcv_identify_system, walrcv_receive, WALRCV_RESTARTING, WALRCV_STARTING, walrcv_startstreaming, WALRCV_STOPPED, WALRCV_STOPPING, WALRCV_STREAMING, WALRCV_WAITING, WalRcvDie(), WalRcvFetchTimeLineHistoryFiles(), WalRcvQuickDieHandler(), WalRcvShutdownHandler(), WalRcvSigHupHandler(), WalRcvSigUsr1Handler(), WalRcvData::walRcvState, WalRcvWaitForStartPosition(), WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_SOCKET_READABLE, WL_TIMEOUT, XLogArchiveForceDone(), XLogArchiveMode, XLogArchiveNotify(), XLogFileName, XLogFileNameP(), XLogWalRcvFlush(), XLogWalRcvProcessMsg(), XLogWalRcvSendHSFeedback(), and XLogWalRcvSendReply().

Referenced by AuxiliaryProcessMain().

 {
     char        conninfo[MAXCONNINFO];
     char       *tmp_conninfo;
     char        slotname[NAMEDATALEN];
     XLogRecPtr  startpoint;
     TimeLineID  startpointTLI;
     TimeLineID  primaryTLI;
     bool        first_stream;
     WalRcvData *walrcv = WalRcv;
     TimestampTz last_recv_timestamp;
     bool        ping_sent;
     char       *err;
 
     /*
      * WalRcv should be set up already (if we are a backend, we inherit this
      * by fork() or EXEC_BACKEND mechanism from the postmaster).
      */
     Assert(walrcv != NULL);
 
     /*
      * 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);
             proc_exit(1);
             break;
 
         case WALRCV_STARTING:
             /* The usual case */
             break;
 
         case WALRCV_WAITING:
         case WALRCV_STREAMING:
         case WALRCV_RESTARTING:
         default:
             /* Shouldn't happen */
             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);
     startpoint = walrcv->receiveStart;
     startpointTLI = walrcv->receiveStartTLI;
 
     /* Initialise to a sanish value */
     walrcv->lastMsgSendTime = walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = GetCurrentTimestamp();
 
     SpinLockRelease(&walrcv->mutex);
 
     /* Arrange to clean up at walreceiver exit */
     on_shmem_exit(WalRcvDie, 0);
 
     walrcv->latch = &MyProc->procLatch;
 
     /* Properly accept or ignore signals the postmaster might send us */
     pqsignal(SIGHUP, WalRcvSigHupHandler);      /* set flag to read config
                                                  * file */
     pqsignal(SIGINT, SIG_IGN);
     pqsignal(SIGTERM, WalRcvShutdownHandler);   /* request shutdown */
     pqsignal(SIGQUIT, WalRcvQuickDieHandler);   /* hard crash time */
     pqsignal(SIGALRM, SIG_IGN);
     pqsignal(SIGPIPE, SIG_IGN);
     pqsignal(SIGUSR1, WalRcvSigUsr1Handler);
     pqsignal(SIGUSR2, SIG_IGN);
 
     /* Reset some signals that are accepted by postmaster but not here */
     pqsignal(SIGCHLD, SIG_DFL);
     pqsignal(SIGTTIN, SIG_DFL);
     pqsignal(SIGTTOU, SIG_DFL);
     pqsignal(SIGCONT, SIG_DFL);
     pqsignal(SIGWINCH, SIG_DFL);
 
     /* We allow SIGQUIT (quickdie) at all times */
     sigdelset(&BlockSig, SIGQUIT);
 
     /* Load the libpq-specific functions */
     load_file("libpqwalreceiver", false);
     if (WalReceiverFunctions == NULL)
         elog(ERROR, "libpqwalreceiver didn't initialize correctly");
 
     /*
      * Create a resource owner to keep track of our resources (not clear that
      * we need this, but may as well have one).
      */
     CurrentResourceOwner = ResourceOwnerCreate(NULL, "Wal Receiver");
 
     /* Unblock signals (they were blocked when the postmaster forked us) */
     PG_SETMASK(&UnBlockSig);
 
     /* Establish the connection to the primary for XLOG streaming */
     EnableWalRcvImmediateExit();
     wrconn = walrcv_connect(conninfo, false, "walreceiver", &err);
     if (!wrconn)
         ereport(ERROR,
                 (errmsg("could not connect to the primary server: %s", err)));
     DisableWalRcvImmediateExit();
 
     /*
      * Save user-visible connection string.  This clobbers the original
      * conninfo, for security.
      */
     tmp_conninfo = walrcv_get_conninfo(wrconn);
     SpinLockAcquire(&walrcv->mutex);
     memset(walrcv->conninfo, 0, MAXCONNINFO);
     if (tmp_conninfo)
     {
         strlcpy((char *) walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
         pfree(tmp_conninfo);
     }
     walrcv->ready_to_display = true;
     SpinLockRelease(&walrcv->mutex);
 
     first_stream = true;
     for (;;)
     {
         char       *primary_sysid;
         char        standby_sysid[32];
         int         server_version;
         WalRcvStreamOptions options;
 
         /*
          * Check that we're connected to a valid server using the
          * IDENTIFY_SYSTEM replication command.
          */
         EnableWalRcvImmediateExit();
         primary_sysid = walrcv_identify_system(wrconn, &primaryTLI,
                                                &server_version);
 
         snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
                  GetSystemIdentifier());
         if (strcmp(primary_sysid, standby_sysid) != 0)
         {
             ereport(ERROR,
                     (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)));
         }
         DisableWalRcvImmediateExit();
 
         /*
          * Confirm that the current timeline of the primary is the same or
          * ahead of ours.
          */
         if (primaryTLI < startpointTLI)
             ereport(ERROR,
                     (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 master later on, we don't select the same timeline that
          * was already used in the current master. 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);
 
         /*
          * Start streaming.
          *
          * We'll try to start at the requested starting point and timeline,
          * even if it's different from the server's latest timeline. In case
          * we've already reached the end of the old timeline, the server will
          * finish the streaming immediately, and we will go back to await
          * orders from the startup process. If recovery_target_timeline is
          * 'latest', the startup process will scan pg_wal and find the new
          * history file, bump recovery target timeline, and ask us to restart
          * on the new timeline.
          */
         options.logical = false;
         options.startpoint = startpoint;
         options.slotname = slotname[0] != '\0' ? slotname : NULL;
         options.proto.physical.startpointTLI = startpointTLI;
         ThisTimeLineID = startpointTLI;
         if (walrcv_startstreaming(wrconn, &options))
         {
             if (first_stream)
                 ereport(LOG,
                         (errmsg("started streaming WAL from primary at %X/%X on timeline %u",
                             (uint32) (startpoint >> 32), (uint32) startpoint,
                                 startpointTLI)));
             else
                 ereport(LOG,
                    (errmsg("restarted WAL streaming at %X/%X on timeline %u",
                            (uint32) (startpoint >> 32), (uint32) startpoint,
                            startpointTLI)));
             first_stream = false;
 
             /* Initialize LogstreamResult and buffers for processing messages */
             LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
             initStringInfo(&reply_message);
             initStringInfo(&incoming_message);
 
             /* Initialize the last recv timestamp */
             last_recv_timestamp = GetCurrentTimestamp();
             ping_sent = false;
 
             /* Loop until end-of-streaming or error */
             for (;;)
             {
                 char       *buf;
                 int         len;
                 bool        endofwal = false;
                 pgsocket    wait_fd = PGINVALID_SOCKET;
                 int         rc;
 
                 /*
                  * Exit walreceiver if we're not in recovery. This should not
                  * happen, but cross-check the status here.
                  */
                 if (!RecoveryInProgress())
                     ereport(FATAL,
                             (errmsg("cannot continue WAL streaming, recovery has already ended")));
 
                 /* Process any requests or signals received recently */
                 ProcessWalRcvInterrupts();
 
                 if (got_SIGHUP)
                 {
                     got_SIGHUP = false;
                     ProcessConfigFile(PGC_SIGHUP);
                     XLogWalRcvSendHSFeedback(true);
                 }
 
                 /* See if we can read data immediately */
                 len = walrcv_receive(wrconn, &buf, &wait_fd);
                 if (len != 0)
                 {
                     /*
                      * Process the received data, and any subsequent data we
                      * can read without blocking.
                      */
                     for (;;)
                     {
                         if (len > 0)
                         {
                             /*
                              * Something was received from master, so reset
                              * timeout
                              */
                             last_recv_timestamp = GetCurrentTimestamp();
                             ping_sent = false;
                             XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1);
                         }
                         else if (len == 0)
                             break;
                         else if (len < 0)
                         {
                             ereport(LOG,
                                     (errmsg("replication terminated by primary server"),
                                      errdetail("End of WAL reached on timeline %u at %X/%X.",
                                                startpointTLI,
                                                (uint32) (LogstreamResult.Write >> 32), (uint32) LogstreamResult.Write)));
                             endofwal = true;
                             break;
                         }
                         len = walrcv_receive(wrconn, &buf, &wait_fd);
                     }
 
                     /* Let the master know that we received some data. */
                     XLogWalRcvSendReply(false, false);
 
                     /*
                      * If we've written some records, flush them to disk and
                      * let the startup process and primary server know about
                      * them.
                      */
                     XLogWalRcvFlush(false);
                 }
 
                 /* Check if we need to exit the streaming loop. */
                 if (endofwal)
                     break;
 
                 /*
                  * Ideally we would reuse a WaitEventSet object repeatedly
                  * here to avoid the overheads of WaitLatchOrSocket on epoll
                  * systems, but we can't be sure that libpq (or any other
                  * walreceiver implementation) has the same socket (even if
                  * the fd is the same number, it may have been closed and
                  * reopened since the last time).  In future, if there is a
                  * function for removing sockets from WaitEventSet, then we
                  * could add and remove just the socket each time, potentially
                  * avoiding some system calls.
                  */
                 Assert(wait_fd != PGINVALID_SOCKET);
                 rc = WaitLatchOrSocket(walrcv->latch,
                                    WL_POSTMASTER_DEATH | WL_SOCKET_READABLE |
                                        WL_TIMEOUT | WL_LATCH_SET,
                                        wait_fd,
                                        NAPTIME_PER_CYCLE,
                                        WAIT_EVENT_WAL_RECEIVER_MAIN);
                 if (rc & WL_LATCH_SET)
                 {
                     ResetLatch(walrcv->latch);
                     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_POSTMASTER_DEATH)
                 {
                     /*
                      * Emergency bailout if postmaster has died.  This is to
                      * avoid the necessity for manual cleanup of all
                      * postmaster children.
                      */
                     exit(1);
                 }
                 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 master anyway, to report any progress in applying
                      * WAL.
                      */
                     bool        requestReply = false;
 
                     /*
                      * Check if time since last receive from standby has
                      * reached the configured limit.
                      */
                     if (wal_receiver_timeout > 0)
                     {
                         TimestampTz now = GetCurrentTimestamp();
                         TimestampTz timeout;
 
                         timeout =
                             TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                         wal_receiver_timeout);
 
                         if (now >= timeout)
                             ereport(ERROR,
                                     (errmsg("terminating walreceiver due to timeout")));
 
                         /*
                          * We didn't receive anything new, for half of
                          * receiver replication timeout. Ping the server.
                          */
                         if (!ping_sent)
                         {
                             timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
                                                  (wal_receiver_timeout / 2));
                             if (now >= timeout)
                             {
                                 requestReply = true;
                                 ping_sent = true;
                             }
                         }
                     }
 
                     XLogWalRcvSendReply(requestReply, requestReply);
                     XLogWalRcvSendHSFeedback(false);
                 }
             }
 
             /*
              * The backend finished streaming. Exit streaming COPY-mode from
              * our side, too.
              */
             EnableWalRcvImmediateExit();
             walrcv_endstreaming(wrconn, &primaryTLI);
             DisableWalRcvImmediateExit();
 
             /*
              * 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);
             if (close(recvFile) != 0)
                 ereport(PANIC,
                         (errcode_for_file_access(),
                          errmsg("could not close log segment %s: %m",
                                 XLogFileNameP(recvFileTLI, recvSegNo))));
 
             /*
              * Create .done file forcibly to prevent the streamed segment from
              * being archived later.
              */
             XLogFileName(xlogfname, recvFileTLI, recvSegNo);
             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 */
 }