PostgreSQL Source Code  git master
walreceiver.h File Reference
#include "access/xlog.h"
#include "access/xlogdefs.h"
#include "getaddrinfo.h"
#include "pgtime.h"
#include "port/atomics.h"
#include "replication/logicalproto.h"
#include "replication/walsender.h"
#include "storage/condition_variable.h"
#include "storage/latch.h"
#include "storage/spin.h"
#include "utils/tuplestore.h"
Include dependency graph for walreceiver.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  WalRcvData
 
struct  WalRcvStreamOptions
 
struct  WalRcvExecResult
 
struct  WalReceiverFunctionsType
 

Macros

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

Typedefs

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

Enumerations

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

Functions

static void walrcv_clear_result (WalRcvExecResult *walres)
 
void WalReceiverMain (void) pg_attribute_noreturn()
 
void ProcessWalRcvInterrupts (void)
 
Size WalRcvShmemSize (void)
 
void WalRcvShmemInit (void)
 
void ShutdownWalRcv (void)
 
bool WalRcvStreaming (void)
 
bool WalRcvRunning (void)
 
void RequestXLogStreaming (TimeLineID tli, XLogRecPtr recptr, const char *conninfo, const char *slotname, bool create_temp_slot)
 
XLogRecPtr GetWalRcvFlushRecPtr (XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI)
 
XLogRecPtr GetWalRcvWriteRecPtr (void)
 
int GetReplicationApplyDelay (void)
 
int GetReplicationTransferLatency (void)
 
void WalRcvForceReply (void)
 

Variables

int wal_receiver_status_interval
 
int wal_receiver_timeout
 
bool hot_standby_feedback
 
WalRcvDataWalRcv
 
PGDLLIMPORT WalReceiverFunctionsTypeWalReceiverFunctions
 

Macro Definition Documentation

◆ AllowCascadeReplication

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

Definition at line 40 of file walreceiver.h.

Referenced by StartupXLOG(), and XLogWalRcvFlush().

◆ MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 37 of file walreceiver.h.

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

◆ walrcv_check_conninfo

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

Definition at line 403 of file walreceiver.h.

Referenced by AlterSubscription(), and CreateSubscription().

◆ walrcv_connect

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

◆ walrcv_create_slot

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

Definition at line 423 of file walreceiver.h.

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

◆ walrcv_disconnect

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

◆ walrcv_endstreaming

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

◆ walrcv_exec

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

◆ walrcv_get_backend_pid

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

Definition at line 425 of file walreceiver.h.

Referenced by WalReceiverMain().

◆ walrcv_get_conninfo

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

Definition at line 405 of file walreceiver.h.

Referenced by WalReceiverMain().

◆ walrcv_get_senderinfo

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

Definition at line 407 of file walreceiver.h.

Referenced by WalReceiverMain().

◆ walrcv_identify_system

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

Definition at line 409 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and WalReceiverMain().

◆ walrcv_readtimelinehistoryfile

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

Definition at line 413 of file walreceiver.h.

Referenced by WalRcvFetchTimeLineHistoryFiles().

◆ walrcv_receive

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

Definition at line 419 of file walreceiver.h.

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

◆ walrcv_send

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

Definition at line 421 of file walreceiver.h.

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

◆ walrcv_server_version

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

Definition at line 411 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and fetch_remote_table_info().

◆ walrcv_startstreaming

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

Definition at line 415 of file walreceiver.h.

Referenced by ApplyWorkerMain(), and WalReceiverMain().

Typedef Documentation

◆ walrcv_check_conninfo_fn

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

Definition at line 244 of file walreceiver.h.

◆ walrcv_connect_fn

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

Definition at line 234 of file walreceiver.h.

◆ walrcv_create_slot_fn

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

Definition at line 347 of file walreceiver.h.

◆ walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 378 of file walreceiver.h.

◆ walrcv_endstreaming_fn

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

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

◆ walrcv_get_backend_pid_fn

typedef pid_t(* walrcv_get_backend_pid_fn) (WalReceiverConn *conn)

Definition at line 358 of file walreceiver.h.

◆ walrcv_get_conninfo_fn

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

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

◆ walrcv_identify_system_fn

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

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

◆ walrcv_receive_fn

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

Definition at line 323 of file walreceiver.h.

◆ walrcv_send_fn

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

Definition at line 333 of file walreceiver.h.

◆ walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 280 of file walreceiver.h.

◆ walrcv_startstreaming_fn

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

Definition at line 303 of file walreceiver.h.

◆ WalRcvExecResult

◆ WalReceiverConn

Definition at line 189 of file walreceiver.h.

◆ WalReceiverFunctionsType

Enumeration Type Documentation

◆ 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 196 of file walreceiver.h.

197 {
198  WALRCV_ERROR, /* There was error when executing the query. */
199  WALRCV_OK_COMMAND, /* Query executed utility or replication
200  * command. */
201  WALRCV_OK_TUPLES, /* Query returned tuples. */
202  WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
203  WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
204  WALRCV_OK_COPY_BOTH /* Query started COPY BOTH replication
205  * protocol. */
WalRcvExecStatus
Definition: walreceiver.h:196

◆ WalRcvState

Enumerator
WALRCV_STOPPED 
WALRCV_STARTING 
WALRCV_STREAMING 
WALRCV_WAITING 
WALRCV_RESTARTING 
WALRCV_STOPPING 

Definition at line 45 of file walreceiver.h.

46 {
47  WALRCV_STOPPED, /* stopped and mustn't start up again */
48  WALRCV_STARTING, /* launched, but the process hasn't
49  * initialized yet */
50  WALRCV_STREAMING, /* walreceiver is streaming */
51  WALRCV_WAITING, /* stopped streaming, waiting for orders */
52  WALRCV_RESTARTING, /* asked to restart streaming */
53  WALRCV_STOPPING /* requested to stop, but still running */
54 } WalRcvState;
WalRcvState
Definition: walreceiver.h:45

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void  )

Definition at line 364 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

365 {
366  WalRcvData *walrcv = WalRcv;
367  XLogRecPtr receivePtr;
368  XLogRecPtr replayPtr;
369  TimestampTz chunkReplayStartTime;
370 
371  SpinLockAcquire(&walrcv->mutex);
372  receivePtr = walrcv->flushedUpto;
373  SpinLockRelease(&walrcv->mutex);
374 
375  replayPtr = GetXLogReplayRecPtr(NULL);
376 
377  if (receivePtr == replayPtr)
378  return 0;
379 
380  chunkReplayStartTime = GetCurrentChunkReplayStartTime();
381 
382  if (chunkReplayStartTime == 0)
383  return -1;
384 
385  return TimestampDifferenceMilliseconds(chunkReplayStartTime,
387 }
slock_t mutex
Definition: walreceiver.h:145
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1580
int64 TimestampTz
Definition: timestamp.h:39
#define SpinLockAcquire(lock)
Definition: spin.h:62
TimestampTz GetCurrentChunkReplayStartTime(void)
Definition: xlog.c:6293
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11728
#define SpinLockRelease(lock)
Definition: spin.h:64
uint64 XLogRecPtr
Definition: xlogdefs.h:21
WalRcvData * WalRcv
XLogRecPtr flushedUpto
Definition: walreceiver.h:85
long TimestampDifferenceMilliseconds(TimestampTz start_time, TimestampTz stop_time)
Definition: timestamp.c:1693

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void  )

Definition at line 394 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

395 {
396  WalRcvData *walrcv = WalRcv;
397  TimestampTz lastMsgSendTime;
398  TimestampTz lastMsgReceiptTime;
399 
400  SpinLockAcquire(&walrcv->mutex);
401  lastMsgSendTime = walrcv->lastMsgSendTime;
402  lastMsgReceiptTime = walrcv->lastMsgReceiptTime;
403  SpinLockRelease(&walrcv->mutex);
404 
405  return TimestampDifferenceMilliseconds(lastMsgSendTime,
406  lastMsgReceiptTime);
407 }
slock_t mutex
Definition: walreceiver.h:145
int64 TimestampTz
Definition: timestamp.h:39
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:100
TimestampTz lastMsgSendTime
Definition: walreceiver.h:99
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
WalRcvData * WalRcv
long TimestampDifferenceMilliseconds(TimestampTz start_time, TimestampTz stop_time)
Definition: timestamp.c:1693

◆ GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr ( XLogRecPtr latestChunkStart,
TimeLineID receiveTLI 
)

Definition at line 331 of file walreceiverfuncs.c.

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

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

332 {
333  WalRcvData *walrcv = WalRcv;
334  XLogRecPtr recptr;
335 
336  SpinLockAcquire(&walrcv->mutex);
337  recptr = walrcv->flushedUpto;
338  if (latestChunkStart)
339  *latestChunkStart = walrcv->latestChunkStart;
340  if (receiveTLI)
341  *receiveTLI = walrcv->receivedTLI;
342  SpinLockRelease(&walrcv->mutex);
343 
344  return recptr;
345 }
slock_t mutex
Definition: walreceiver.h:145
TimeLineID receivedTLI
Definition: walreceiver.h:86
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogRecPtr latestChunkStart
Definition: walreceiver.h:94
#define SpinLockRelease(lock)
Definition: spin.h:64
static TimeLineID receiveTLI
Definition: xlog.c:218
uint64 XLogRecPtr
Definition: xlogdefs.h:21
WalRcvData * WalRcv
XLogRecPtr flushedUpto
Definition: walreceiver.h:85

◆ GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( void  )

Definition at line 352 of file walreceiverfuncs.c.

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

Referenced by walrcv_clear_result().

353 {
354  WalRcvData *walrcv = WalRcv;
355 
356  return pg_atomic_read_u64(&walrcv->writtenUpto);
357 }
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:153
WalRcvData * WalRcv
static uint64 pg_atomic_read_u64(volatile pg_atomic_uint64 *ptr)
Definition: atomics.h:429

◆ ProcessWalRcvInterrupts()

void ProcessWalRcvInterrupts ( void  )

Definition at line 146 of file walreceiver.c.

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

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

147 {
148  /*
149  * Although walreceiver interrupt handling doesn't use the same scheme as
150  * regular backends, call CHECK_FOR_INTERRUPTS() to make sure we receive
151  * any incoming signals on Win32, and also to make sure we process any
152  * barrier events.
153  */
155 
157  {
158  ereport(FATAL,
159  (errcode(ERRCODE_ADMIN_SHUTDOWN),
160  errmsg("terminating walreceiver process due to administrator command")));
161  }
162 }
int errcode(int sqlerrcode)
Definition: elog.c:698
#define FATAL
Definition: elog.h:49
volatile sig_atomic_t ShutdownRequestPending
Definition: interrupt.c:27
#define ereport(elevel,...)
Definition: elog.h:157
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:102

◆ RequestXLogStreaming()

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

Definition at line 245 of file walreceiverfuncs.c.

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

Referenced by WaitForWALToBecomeAvailable(), and walrcv_clear_result().

247 {
248  WalRcvData *walrcv = WalRcv;
249  bool launch = false;
250  pg_time_t now = (pg_time_t) time(NULL);
251  Latch *latch;
252 
253  /*
254  * We always start at the beginning of the segment. That prevents a broken
255  * segment (i.e., with no records in the first half of a segment) from
256  * being created by XLOG streaming, which might cause trouble later on if
257  * the segment is e.g archived.
258  */
259  if (XLogSegmentOffset(recptr, wal_segment_size) != 0)
260  recptr -= XLogSegmentOffset(recptr, wal_segment_size);
261 
262  SpinLockAcquire(&walrcv->mutex);
263 
264  /* It better be stopped if we try to restart it */
265  Assert(walrcv->walRcvState == WALRCV_STOPPED ||
266  walrcv->walRcvState == WALRCV_WAITING);
267 
268  if (conninfo != NULL)
269  strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
270  else
271  walrcv->conninfo[0] = '\0';
272 
273  /*
274  * Use configured replication slot if present, and ignore the value of
275  * create_temp_slot as the slot name should be persistent. Otherwise, use
276  * create_temp_slot to determine whether this WAL receiver should create a
277  * temporary slot by itself and use it, or not.
278  */
279  if (slotname != NULL && slotname[0] != '\0')
280  {
281  strlcpy((char *) walrcv->slotname, slotname, NAMEDATALEN);
282  walrcv->is_temp_slot = false;
283  }
284  else
285  {
286  walrcv->slotname[0] = '\0';
287  walrcv->is_temp_slot = create_temp_slot;
288  }
289 
290  if (walrcv->walRcvState == WALRCV_STOPPED)
291  {
292  launch = true;
293  walrcv->walRcvState = WALRCV_STARTING;
294  }
295  else
296  walrcv->walRcvState = WALRCV_RESTARTING;
297  walrcv->startTime = now;
298 
299  /*
300  * If this is the first startup of walreceiver (on this timeline),
301  * initialize flushedUpto and latestChunkStart to the starting point.
302  */
303  if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli)
304  {
305  walrcv->flushedUpto = recptr;
306  walrcv->receivedTLI = tli;
307  walrcv->latestChunkStart = recptr;
308  }
309  walrcv->receiveStart = recptr;
310  walrcv->receiveStartTLI = tli;
311 
312  latch = walrcv->latch;
313 
314  SpinLockRelease(&walrcv->mutex);
315 
316  if (launch)
318  else if (latch)
319  SetLatch(latch);
320 }
int64 pg_time_t
Definition: pgtime.h:23
slock_t mutex
Definition: walreceiver.h:145
int wal_segment_size
Definition: xlog.c:121
WalRcvState walRcvState
Definition: walreceiver.h:65
TimeLineID receivedTLI
Definition: walreceiver.h:86
void SetLatch(Latch *latch)
Definition: latch.c:567
pg_time_t startTime
Definition: walreceiver.h:67
#define NAMEDATALEN
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define MAXCONNINFO
Definition: walreceiver.h:37
XLogRecPtr latestChunkStart
Definition: walreceiver.h:94
Definition: latch.h:110
TimeLineID receiveStartTLI
Definition: walreceiver.h:76
Latch * latch
Definition: walreceiver.h:143
#define SpinLockRelease(lock)
Definition: spin.h:64
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
size_t strlcpy(char *dst, const char *src, size_t siz)
Definition: strlcpy.c:45
#define Assert(condition)
Definition: c.h:804
bool is_temp_slot
Definition: walreceiver.h:131
WalRcvData * WalRcv
void SendPostmasterSignal(PMSignalReason reason)
Definition: pmsignal.c:153
XLogRecPtr receiveStart
Definition: walreceiver.h:75
XLogRecPtr flushedUpto
Definition: walreceiver.h:85
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1544
char slotname[NAMEDATALEN]
Definition: walreceiver.h:125
char conninfo[MAXCONNINFO]
Definition: walreceiver.h:112

◆ ShutdownWalRcv()

void ShutdownWalRcv ( void  )

Definition at line 178 of file walreceiverfuncs.c.

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

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

179 {
180  WalRcvData *walrcv = WalRcv;
181  pid_t walrcvpid = 0;
182  bool stopped = false;
183 
184  /*
185  * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
186  * mode once it's finished, and will also request postmaster to not
187  * restart itself.
188  */
189  SpinLockAcquire(&walrcv->mutex);
190  switch (walrcv->walRcvState)
191  {
192  case WALRCV_STOPPED:
193  break;
194  case WALRCV_STARTING:
195  walrcv->walRcvState = WALRCV_STOPPED;
196  stopped = true;
197  break;
198 
199  case WALRCV_STREAMING:
200  case WALRCV_WAITING:
201  case WALRCV_RESTARTING:
202  walrcv->walRcvState = WALRCV_STOPPING;
203  /* fall through */
204  case WALRCV_STOPPING:
205  walrcvpid = walrcv->pid;
206  break;
207  }
208  SpinLockRelease(&walrcv->mutex);
209 
210  /* Unnecessary but consistent. */
211  if (stopped)
213 
214  /*
215  * Signal walreceiver process if it was still running.
216  */
217  if (walrcvpid != 0)
218  kill(walrcvpid, SIGTERM);
219 
220  /*
221  * Wait for walreceiver to acknowledge its death by setting state to
222  * WALRCV_STOPPED.
223  */
225  while (WalRcvRunning())
229 }
slock_t mutex
Definition: walreceiver.h:145
WalRcvState walRcvState
Definition: walreceiver.h:65
void ConditionVariableBroadcast(ConditionVariable *cv)
#define kill(pid, sig)
Definition: win32_port.h:454
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableCancelSleep(void)
pid_t pid
Definition: walreceiver.h:64
#define SpinLockRelease(lock)
Definition: spin.h:64
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
WalRcvData * WalRcv
bool WalRcvRunning(void)
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66

◆ walrcv_clear_result()

static void walrcv_clear_result ( WalRcvExecResult walres)
inlinestatic

Definition at line 433 of file walreceiver.h.

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

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

434 {
435  if (!walres)
436  return;
437 
438  if (walres->err)
439  pfree(walres->err);
440 
441  if (walres->tuplestore)
442  tuplestore_end(walres->tuplestore);
443 
444  if (walres->tupledesc)
445  FreeTupleDesc(walres->tupledesc);
446 
447  pfree(walres);
448 }
void pfree(void *pointer)
Definition: mcxt.c:1169
TupleDesc tupledesc
Definition: walreceiver.h:218
Tuplestorestate * tuplestore
Definition: walreceiver.h:217
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:453
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:309

◆ WalRcvForceReply()

void WalRcvForceReply ( void  )

Definition at line 1256 of file walreceiver.c.

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

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

1257 {
1258  Latch *latch;
1259 
1260  WalRcv->force_reply = true;
1261  /* fetching the latch pointer might not be atomic, so use spinlock */
1263  latch = WalRcv->latch;
1265  if (latch)
1266  SetLatch(latch);
1267 }
slock_t mutex
Definition: walreceiver.h:145
sig_atomic_t force_reply
Definition: walreceiver.h:160
void SetLatch(Latch *latch)
Definition: latch.c:567
#define SpinLockAcquire(lock)
Definition: spin.h:62
Definition: latch.h:110
Latch * latch
Definition: walreceiver.h:143
#define SpinLockRelease(lock)
Definition: spin.h:64
WalRcvData * WalRcv

◆ WalRcvRunning()

bool WalRcvRunning ( void  )

Definition at line 75 of file walreceiverfuncs.c.

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

Referenced by ShutdownWalRcv(), StartupRequestWalReceiverRestart(), and walrcv_clear_result().

76 {
77  WalRcvData *walrcv = WalRcv;
79  pg_time_t startTime;
80 
81  SpinLockAcquire(&walrcv->mutex);
82 
83  state = walrcv->walRcvState;
84  startTime = walrcv->startTime;
85 
86  SpinLockRelease(&walrcv->mutex);
87 
88  /*
89  * If it has taken too long for walreceiver to start up, give up. Setting
90  * the state to STOPPED ensures that if walreceiver later does start up
91  * after all, it will see that it's not supposed to be running and die
92  * without doing anything.
93  */
94  if (state == WALRCV_STARTING)
95  {
96  pg_time_t now = (pg_time_t) time(NULL);
97 
98  if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
99  {
100  bool stopped = false;
101 
102  SpinLockAcquire(&walrcv->mutex);
103  if (walrcv->walRcvState == WALRCV_STARTING)
104  {
105  state = walrcv->walRcvState = WALRCV_STOPPED;
106  stopped = true;
107  }
108  SpinLockRelease(&walrcv->mutex);
109 
110  if (stopped)
112  }
113  }
114 
115  if (state != WALRCV_STOPPED)
116  return true;
117  else
118  return false;
119 }
int64 pg_time_t
Definition: pgtime.h:23
slock_t mutex
Definition: walreceiver.h:145
WalRcvState walRcvState
Definition: walreceiver.h:65
void ConditionVariableBroadcast(ConditionVariable *cv)
pg_time_t startTime
Definition: walreceiver.h:67
WalRcvState
Definition: walreceiver.h:45
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
#define WALRCV_STARTUP_TIMEOUT
Definition: regguts.h:317
WalRcvData * WalRcv
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1544
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void  )

Definition at line 54 of file walreceiverfuncs.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and walrcv_clear_result().

55 {
56  bool found;
57 
58  WalRcv = (WalRcvData *)
59  ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &found);
60 
61  if (!found)
62  {
63  /* First time through, so initialize */
69  WalRcv->latch = NULL;
70  }
71 }
slock_t mutex
Definition: walreceiver.h:145
WalRcvState walRcvState
Definition: walreceiver.h:65
#define SpinLockInit(lock)
Definition: spin.h:60
#define MemSet(start, val, len)
Definition: c.h:1008
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:153
void ConditionVariableInit(ConditionVariable *cv)
static void pg_atomic_init_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:415
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:396
Latch * latch
Definition: walreceiver.h:143
Size WalRcvShmemSize(void)
WalRcvData * WalRcv
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void  )

Definition at line 43 of file walreceiverfuncs.c.

References add_size().

Referenced by CreateSharedMemoryAndSemaphores(), walrcv_clear_result(), and WalRcvShmemInit().

44 {
45  Size size = 0;
46 
47  size = add_size(size, sizeof(WalRcvData));
48 
49  return size;
50 }
Size add_size(Size s1, Size s2)
Definition: shmem.c:502
size_t Size
Definition: c.h:540

◆ WalRcvStreaming()

bool WalRcvStreaming ( void  )

Definition at line 126 of file walreceiverfuncs.c.

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

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

127 {
128  WalRcvData *walrcv = WalRcv;
130  pg_time_t startTime;
131 
132  SpinLockAcquire(&walrcv->mutex);
133 
134  state = walrcv->walRcvState;
135  startTime = walrcv->startTime;
136 
137  SpinLockRelease(&walrcv->mutex);
138 
139  /*
140  * If it has taken too long for walreceiver to start up, give up. Setting
141  * the state to STOPPED ensures that if walreceiver later does start up
142  * after all, it will see that it's not supposed to be running and die
143  * without doing anything.
144  */
145  if (state == WALRCV_STARTING)
146  {
147  pg_time_t now = (pg_time_t) time(NULL);
148 
149  if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
150  {
151  bool stopped = false;
152 
153  SpinLockAcquire(&walrcv->mutex);
154  if (walrcv->walRcvState == WALRCV_STARTING)
155  {
156  state = walrcv->walRcvState = WALRCV_STOPPED;
157  stopped = true;
158  }
159  SpinLockRelease(&walrcv->mutex);
160 
161  if (stopped)
163  }
164  }
165 
166  if (state == WALRCV_STREAMING || state == WALRCV_STARTING ||
167  state == WALRCV_RESTARTING)
168  return true;
169  else
170  return false;
171 }
int64 pg_time_t
Definition: pgtime.h:23
slock_t mutex
Definition: walreceiver.h:145
WalRcvState walRcvState
Definition: walreceiver.h:65
void ConditionVariableBroadcast(ConditionVariable *cv)
pg_time_t startTime
Definition: walreceiver.h:67
WalRcvState
Definition: walreceiver.h:45
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
#define WALRCV_STARTUP_TIMEOUT
Definition: regguts.h:317
WalRcvData * WalRcv
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1544
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66

◆ WalReceiverMain()

void WalReceiverMain ( void  )

Definition at line 167 of file walreceiver.c.

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

Referenced by AuxiliaryProcessMain(), and walrcv_clear_result().

168 {
169  char conninfo[MAXCONNINFO];
170  char *tmp_conninfo;
171  char slotname[NAMEDATALEN];
172  bool is_temp_slot;
173  XLogRecPtr startpoint;
174  TimeLineID startpointTLI;
175  TimeLineID primaryTLI;
176  bool first_stream;
177  WalRcvData *walrcv = WalRcv;
178  TimestampTz last_recv_timestamp;
180  bool ping_sent;
181  char *err;
182  char *sender_host = NULL;
183  int sender_port = 0;
184 
185  /*
186  * WalRcv should be set up already (if we are a backend, we inherit this
187  * by fork() or EXEC_BACKEND mechanism from the postmaster).
188  */
189  Assert(walrcv != NULL);
190 
191  now = GetCurrentTimestamp();
192 
193  /*
194  * Mark walreceiver as running in shared memory.
195  *
196  * Do this as early as possible, so that if we fail later on, we'll set
197  * state to STOPPED. If we die before this, the startup process will keep
198  * waiting for us to start up, until it times out.
199  */
200  SpinLockAcquire(&walrcv->mutex);
201  Assert(walrcv->pid == 0);
202  switch (walrcv->walRcvState)
203  {
204  case WALRCV_STOPPING:
205  /* If we've already been requested to stop, don't start up. */
206  walrcv->walRcvState = WALRCV_STOPPED;
207  /* fall through */
208 
209  case WALRCV_STOPPED:
210  SpinLockRelease(&walrcv->mutex);
212  proc_exit(1);
213  break;
214 
215  case WALRCV_STARTING:
216  /* The usual case */
217  break;
218 
219  case WALRCV_WAITING:
220  case WALRCV_STREAMING:
221  case WALRCV_RESTARTING:
222  default:
223  /* Shouldn't happen */
224  SpinLockRelease(&walrcv->mutex);
225  elog(PANIC, "walreceiver still running according to shared memory state");
226  }
227  /* Advertise our PID so that the startup process can kill us */
228  walrcv->pid = MyProcPid;
229  walrcv->walRcvState = WALRCV_STREAMING;
230 
231  /* Fetch information required to start streaming */
232  walrcv->ready_to_display = false;
233  strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
234  strlcpy(slotname, (char *) walrcv->slotname, NAMEDATALEN);
235  is_temp_slot = walrcv->is_temp_slot;
236  startpoint = walrcv->receiveStart;
237  startpointTLI = walrcv->receiveStartTLI;
238 
239  /*
240  * At most one of is_temp_slot and slotname can be set; otherwise,
241  * RequestXLogStreaming messed up.
242  */
243  Assert(!is_temp_slot || (slotname[0] == '\0'));
244 
245  /* Initialise to a sanish value */
246  walrcv->lastMsgSendTime =
247  walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
248 
249  /* Report the latch to use to awaken this process */
250  walrcv->latch = &MyProc->procLatch;
251 
252  SpinLockRelease(&walrcv->mutex);
253 
255 
256  /* Arrange to clean up at walreceiver exit */
258 
259  /* Properly accept or ignore signals the postmaster might send us */
260  pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
261  * file */
262  pqsignal(SIGINT, SIG_IGN);
263  pqsignal(SIGTERM, SignalHandlerForShutdownRequest); /* request shutdown */
264  /* SIGQUIT handler was already set up by InitPostmasterChild */
269 
270  /* Reset some signals that are accepted by postmaster but not here */
272 
273  /* Load the libpq-specific functions */
274  load_file("libpqwalreceiver", false);
275  if (WalReceiverFunctions == NULL)
276  elog(ERROR, "libpqwalreceiver didn't initialize correctly");
277 
278  /* Unblock signals (they were blocked when the postmaster forked us) */
280 
281  /* Establish the connection to the primary for XLOG streaming */
282  wrconn = walrcv_connect(conninfo, false, cluster_name[0] ? cluster_name : "walreceiver", &err);
283  if (!wrconn)
284  ereport(ERROR,
285  (errmsg("could not connect to the primary server: %s", err)));
286 
287  /*
288  * Save user-visible connection string. This clobbers the original
289  * conninfo, for security. Also save host and port of the sender server
290  * this walreceiver is connected to.
291  */
292  tmp_conninfo = walrcv_get_conninfo(wrconn);
293  walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
294  SpinLockAcquire(&walrcv->mutex);
295  memset(walrcv->conninfo, 0, MAXCONNINFO);
296  if (tmp_conninfo)
297  strlcpy((char *) walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
298 
299  memset(walrcv->sender_host, 0, NI_MAXHOST);
300  if (sender_host)
301  strlcpy((char *) walrcv->sender_host, sender_host, NI_MAXHOST);
302 
303  walrcv->sender_port = sender_port;
304  walrcv->ready_to_display = true;
305  SpinLockRelease(&walrcv->mutex);
306 
307  if (tmp_conninfo)
308  pfree(tmp_conninfo);
309 
310  if (sender_host)
311  pfree(sender_host);
312 
313  first_stream = true;
314  for (;;)
315  {
316  char *primary_sysid;
317  char standby_sysid[32];
319 
320  /*
321  * Check that we're connected to a valid server using the
322  * IDENTIFY_SYSTEM replication command.
323  */
324  primary_sysid = walrcv_identify_system(wrconn, &primaryTLI);
325 
326  snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
328  if (strcmp(primary_sysid, standby_sysid) != 0)
329  {
330  ereport(ERROR,
331  (errmsg("database system identifier differs between the primary and standby"),
332  errdetail("The primary's identifier is %s, the standby's identifier is %s.",
333  primary_sysid, standby_sysid)));
334  }
335 
336  /*
337  * Confirm that the current timeline of the primary is the same or
338  * ahead of ours.
339  */
340  if (primaryTLI < startpointTLI)
341  ereport(ERROR,
342  (errmsg("highest timeline %u of the primary is behind recovery timeline %u",
343  primaryTLI, startpointTLI)));
344 
345  /*
346  * Get any missing history files. We do this always, even when we're
347  * not interested in that timeline, so that if we're promoted to
348  * become the primary later on, we don't select the same timeline that
349  * was already used in the current primary. This isn't bullet-proof -
350  * you'll need some external software to manage your cluster if you
351  * need to ensure that a unique timeline id is chosen in every case,
352  * but let's avoid the confusion of timeline id collisions where we
353  * can.
354  */
355  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
356 
357  /*
358  * Create temporary replication slot if requested, and update slot
359  * name in shared memory. (Note the slot name cannot already be set
360  * in this case.)
361  */
362  if (is_temp_slot)
363  {
364  snprintf(slotname, sizeof(slotname),
365  "pg_walreceiver_%lld",
366  (long long int) walrcv_get_backend_pid(wrconn));
367 
368  walrcv_create_slot(wrconn, slotname, true, 0, NULL);
369 
370  SpinLockAcquire(&walrcv->mutex);
371  strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
372  SpinLockRelease(&walrcv->mutex);
373  }
374 
375  /*
376  * Start streaming.
377  *
378  * We'll try to start at the requested starting point and timeline,
379  * even if it's different from the server's latest timeline. In case
380  * we've already reached the end of the old timeline, the server will
381  * finish the streaming immediately, and we will go back to await
382  * orders from the startup process. If recovery_target_timeline is
383  * 'latest', the startup process will scan pg_wal and find the new
384  * history file, bump recovery target timeline, and ask us to restart
385  * on the new timeline.
386  */
387  options.logical = false;
388  options.startpoint = startpoint;
389  options.slotname = slotname[0] != '\0' ? slotname : NULL;
390  options.proto.physical.startpointTLI = startpointTLI;
391  ThisTimeLineID = startpointTLI;
392  if (walrcv_startstreaming(wrconn, &options))
393  {
394  if (first_stream)
395  ereport(LOG,
396  (errmsg("started streaming WAL from primary at %X/%X on timeline %u",
397  LSN_FORMAT_ARGS(startpoint), startpointTLI)));
398  else
399  ereport(LOG,
400  (errmsg("restarted WAL streaming at %X/%X on timeline %u",
401  LSN_FORMAT_ARGS(startpoint), startpointTLI)));
402  first_stream = false;
403 
404  /* Initialize LogstreamResult and buffers for processing messages */
408 
409  /* Initialize the last recv timestamp */
410  last_recv_timestamp = GetCurrentTimestamp();
411  ping_sent = false;
412 
413  /* Loop until end-of-streaming or error */
414  for (;;)
415  {
416  char *buf;
417  int len;
418  bool endofwal = false;
419  pgsocket wait_fd = PGINVALID_SOCKET;
420  int rc;
421 
422  /*
423  * Exit walreceiver if we're not in recovery. This should not
424  * happen, but cross-check the status here.
425  */
426  if (!RecoveryInProgress())
427  ereport(FATAL,
428  (errmsg("cannot continue WAL streaming, recovery has already ended")));
429 
430  /* Process any requests or signals received recently */
432 
434  {
435  ConfigReloadPending = false;
438  }
439 
440  /* See if we can read data immediately */
441  len = walrcv_receive(wrconn, &buf, &wait_fd);
442  if (len != 0)
443  {
444  /*
445  * Process the received data, and any subsequent data we
446  * can read without blocking.
447  */
448  for (;;)
449  {
450  if (len > 0)
451  {
452  /*
453  * Something was received from primary, so reset
454  * timeout
455  */
456  last_recv_timestamp = GetCurrentTimestamp();
457  ping_sent = false;
458  XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1);
459  }
460  else if (len == 0)
461  break;
462  else if (len < 0)
463  {
464  ereport(LOG,
465  (errmsg("replication terminated by primary server"),
466  errdetail("End of WAL reached on timeline %u at %X/%X.",
467  startpointTLI,
469  endofwal = true;
470  break;
471  }
472  len = walrcv_receive(wrconn, &buf, &wait_fd);
473  }
474 
475  /* Let the primary know that we received some data. */
476  XLogWalRcvSendReply(false, false);
477 
478  /*
479  * If we've written some records, flush them to disk and
480  * let the startup process and primary server know about
481  * them.
482  */
483  XLogWalRcvFlush(false);
484  }
485 
486  /* Check if we need to exit the streaming loop. */
487  if (endofwal)
488  break;
489 
490  /*
491  * Ideally we would reuse a WaitEventSet object repeatedly
492  * here to avoid the overheads of WaitLatchOrSocket on epoll
493  * systems, but we can't be sure that libpq (or any other
494  * walreceiver implementation) has the same socket (even if
495  * the fd is the same number, it may have been closed and
496  * reopened since the last time). In future, if there is a
497  * function for removing sockets from WaitEventSet, then we
498  * could add and remove just the socket each time, potentially
499  * avoiding some system calls.
500  */
501  Assert(wait_fd != PGINVALID_SOCKET);
505  wait_fd,
508  if (rc & WL_LATCH_SET)
509  {
512 
513  if (walrcv->force_reply)
514  {
515  /*
516  * The recovery process has asked us to send apply
517  * feedback now. Make sure the flag is really set to
518  * false in shared memory before sending the reply, so
519  * we don't miss a new request for a reply.
520  */
521  walrcv->force_reply = false;
523  XLogWalRcvSendReply(true, false);
524  }
525  }
526  if (rc & WL_TIMEOUT)
527  {
528  /*
529  * We didn't receive anything new. If we haven't heard
530  * anything from the server for more than
531  * wal_receiver_timeout / 2, ping the server. Also, if
532  * it's been longer than wal_receiver_status_interval
533  * since the last update we sent, send a status update to
534  * the primary anyway, to report any progress in applying
535  * WAL.
536  */
537  bool requestReply = false;
538 
539  /*
540  * Check if time since last receive from primary has
541  * reached the configured limit.
542  */
543  if (wal_receiver_timeout > 0)
544  {
546  TimestampTz timeout;
547 
548  timeout =
549  TimestampTzPlusMilliseconds(last_recv_timestamp,
551 
552  if (now >= timeout)
553  ereport(ERROR,
554  (errmsg("terminating walreceiver due to timeout")));
555 
556  /*
557  * We didn't receive anything new, for half of
558  * receiver replication timeout. Ping the server.
559  */
560  if (!ping_sent)
561  {
562  timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
563  (wal_receiver_timeout / 2));
564  if (now >= timeout)
565  {
566  requestReply = true;
567  ping_sent = true;
568  }
569  }
570  }
571 
572  XLogWalRcvSendReply(requestReply, requestReply);
574  }
575  }
576 
577  /*
578  * The backend finished streaming. Exit streaming COPY-mode from
579  * our side, too.
580  */
581  walrcv_endstreaming(wrconn, &primaryTLI);
582 
583  /*
584  * If the server had switched to a new timeline that we didn't
585  * know about when we began streaming, fetch its timeline history
586  * file now.
587  */
588  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
589  }
590  else
591  ereport(LOG,
592  (errmsg("primary server contains no more WAL on requested timeline %u",
593  startpointTLI)));
594 
595  /*
596  * End of WAL reached on the requested timeline. Close the last
597  * segment, and await for new orders from the startup process.
598  */
599  if (recvFile >= 0)
600  {
601  char xlogfname[MAXFNAMELEN];
602 
603  XLogWalRcvFlush(false);
605  if (close(recvFile) != 0)
606  ereport(PANIC,
608  errmsg("could not close log segment %s: %m",
609  xlogfname)));
610 
611  /*
612  * Create .done file forcibly to prevent the streamed segment from
613  * being archived later.
614  */
616  XLogArchiveForceDone(xlogfname);
617  else
618  XLogArchiveNotify(xlogfname);
619  }
620  recvFile = -1;
621 
622  elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
623  WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
624  }
625  /* not reached */
626 }
int sender_port
Definition: walreceiver.h:119
static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
Definition: walreceiver.c:712
#define walrcv_get_conninfo(conn)
Definition: walreceiver.h:405
#define walrcv_endstreaming(conn, next_tli)
Definition: walreceiver.h:417
#define DEBUG1
Definition: elog.h:25
int MyProcPid
Definition: globals.c:43
uint32 TimeLineID
Definition: xlogdefs.h:59
slock_t mutex
Definition: walreceiver.h:145
#define WL_TIMEOUT
Definition: latch.h:128
void ProcessConfigFile(GucContext context)
int wal_segment_size
Definition: xlog.c:121
#define SIGUSR1
Definition: win32_port.h:171
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1580
#define SIGCHLD
Definition: win32_port.h:169
PGPROC * MyProc
Definition: proc.c:68
int64 TimestampTz
Definition: timestamp.h:39
static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
Definition: walreceiver.c:632
sig_atomic_t force_reply
Definition: walreceiver.h:160
#define walrcv_identify_system(conn, primary_tli)
Definition: walreceiver.h:409
WalRcvState walRcvState
Definition: walreceiver.h:65
static StringInfoData incoming_message
Definition: walreceiver.c:119
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:56
static void XLogWalRcvSendHSFeedback(bool immed)
Definition: walreceiver.c:1102
#define walrcv_receive(conn, buffer, wait_fd)
Definition: walreceiver.h:419
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:146
#define walrcv_startstreaming(conn, options)
Definition: walreceiver.h:415
void ConditionVariableBroadcast(ConditionVariable *cv)
void proc_exit(int code)
Definition: ipc.c:104
#define WL_SOCKET_READABLE
Definition: latch.h:126
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len)
Definition: walreceiver.c:803
#define SIGPIPE
Definition: win32_port.h:164
#define SIGUSR2
Definition: win32_port.h:172
#define LOG
Definition: elog.h:26
bool RecoveryInProgress(void)
Definition: xlog.c:8237
#define walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn)
Definition: walreceiver.h:423
#define PANIC
Definition: elog.h:50
#define NAPTIME_PER_CYCLE
Definition: walreceiver.c:97
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:100
void ResetLatch(Latch *latch)
Definition: latch.c:660
TimestampTz lastMsgSendTime
Definition: walreceiver.h:99
#define PG_SETMASK(mask)
Definition: pqsignal.h:19
static void pg_atomic_write_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:438
int wal_receiver_timeout
Definition: walreceiver.c:90
Latch procLatch
Definition: proc.h:130
static WalReceiverConn * wrconn
Definition: walreceiver.c:94
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:153
#define NAMEDATALEN
#define SpinLockAcquire(lock)
Definition: spin.h:62
static StringInfoData reply_message
Definition: walreceiver.c:118
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
void XLogArchiveNotify(const char *xlog)
Definition: xlogarchive.c:467
void pfree(void *pointer)
Definition: mcxt.c:1169
#define NI_MAXHOST
Definition: getaddrinfo.h:88
#define ERROR
Definition: elog.h:46
#define FATAL
Definition: elog.h:49
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11728
pid_t pid
Definition: walreceiver.h:64
#define MAXCONNINFO
Definition: walreceiver.h:37
void on_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:361
static char * buf
Definition: pg_test_fsync.c:68
int XLogArchiveMode
Definition: xlog.c:97
int errdetail(const char *fmt,...)
Definition: elog.c:1042
int errcode_for_file_access(void)
Definition: elog.c:721
#define SIGHUP
Definition: win32_port.h:159
XLogRecPtr startpoint
Definition: walreceiver.h:170
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:500
int pgsocket
Definition: port.h:31
sigset_t UnBlockSig
Definition: pqsignal.c:22
TimeLineID receiveStartTLI
Definition: walreceiver.h:76
static TimeLineID recvFileTLI
Definition: walreceiver.c:105
#define walrcv_get_backend_pid(conn)
Definition: walreceiver.h:425
Definition: guc.h:72
char * cluster_name
Definition: guc.c:582
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:104
Latch * latch
Definition: walreceiver.h:143
#define SIG_IGN
Definition: win32_port.h:156
struct WalRcvStreamOptions::@103::@104 physical
static char ** options
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static XLogSegNo recvSegNo
Definition: walreceiver.c:106
#define MAXFNAMELEN
#define SpinLockRelease(lock)
Definition: spin.h:64
void XLogArchiveForceDone(const char *xlog)
Definition: xlogarchive.c:517
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:95
#define PGINVALID_SOCKET
Definition: port.h:33
static void XLogWalRcvSendReply(bool force, bool requestReply)
Definition: walreceiver.c:1033
bool ready_to_display
Definition: walreceiver.h:134
TimestampTz latestWalEndTime
Definition: walreceiver.h:106
TimeLineID ThisTimeLineID
Definition: xlog.c:196
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:56
#define ereport(elevel,...)
Definition: elog.h:157
pqsigfunc pqsignal(int signum, pqsigfunc handler)
Definition: signal.c:170
size_t strlcpy(char *dst, const char *src, size_t siz)
Definition: strlcpy.c:45
#define pg_memory_barrier()
Definition: atomics.h:145
#define SIG_DFL
Definition: win32_port.h:154
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:146
#define XLogFileName(fname, tli, logSegNo, wal_segsz_bytes)
#define SIGALRM
Definition: win32_port.h:165
bool is_temp_slot
Definition: walreceiver.h:131
static void XLogWalRcvFlush(bool dying)
Definition: walreceiver.c:975
WalRcvData * WalRcv
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4965
char sender_host[NI_MAXHOST]
Definition: walreceiver.h:118
int errmsg(const char *fmt,...)
Definition: elog.c:909
static void WalRcvDie(int code, Datum arg)
Definition: walreceiver.c:768
#define elog(elevel,...)
Definition: elog.h:232
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:26
#define walrcv_get_senderinfo(conn, sender_host, sender_port)
Definition: walreceiver.h:407
struct Latch * MyLatch
Definition: globals.c:57
union WalRcvStreamOptions::@103 proto
#define close(a)
Definition: win32.h:12
XLogRecPtr receiveStart
Definition: walreceiver.h:75
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:642
static int recvFile
Definition: walreceiver.c:104
#define snprintf
Definition: port.h:216
#define WL_LATCH_SET
Definition: latch.h:125
#define UINT64_FORMAT
Definition: c.h:484
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1544
char slotname[NAMEDATALEN]
Definition: walreceiver.h:125
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66
static struct @17 LogstreamResult
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
char conninfo[MAXCONNINFO]
Definition: walreceiver.h:112
#define walrcv_connect(conninfo, logical, appname, err)
Definition: walreceiver.h:401

Variable Documentation

◆ hot_standby_feedback

bool hot_standby_feedback

Definition at line 91 of file walreceiver.c.

Referenced by XLogWalRcvSendHSFeedback().

◆ wal_receiver_status_interval

int wal_receiver_status_interval

Definition at line 89 of file walreceiver.c.

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

◆ wal_receiver_timeout

int wal_receiver_timeout

Definition at line 90 of file walreceiver.c.

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

◆ WalRcv

◆ WalReceiverFunctions

PGDLLIMPORT WalReceiverFunctionsType* WalReceiverFunctions

Definition at line 95 of file walreceiver.c.

Referenced by _PG_init().