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

Go to the source code of this file.

Data Structures

struct  WalRcvData
 
struct  WalRcvStreamOptions
 
struct  WalRcvExecResult
 
struct  WalReceiverFunctionsType
 

Macros

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

Typedefs

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

Enumerations

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

Functions

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

Variables

PGDLLIMPORT int wal_receiver_status_interval
 
PGDLLIMPORT int wal_receiver_timeout
 
PGDLLIMPORT bool hot_standby_feedback
 
PGDLLIMPORT WalRcvDataWalRcv
 
PGDLLIMPORT WalReceiverFunctionsTypeWalReceiverFunctions
 

Macro Definition Documentation

◆ AllowCascadeReplication

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

Definition at line 42 of file walreceiver.h.

◆ MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 39 of file walreceiver.h.

◆ walrcv_check_conninfo

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

Definition at line 410 of file walreceiver.h.

◆ walrcv_connect

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

Definition at line 408 of file walreceiver.h.

◆ walrcv_create_slot

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

Definition at line 430 of file walreceiver.h.

◆ walrcv_disconnect

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

Definition at line 436 of file walreceiver.h.

◆ walrcv_endstreaming

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

Definition at line 424 of file walreceiver.h.

◆ walrcv_exec

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

Definition at line 434 of file walreceiver.h.

◆ walrcv_get_backend_pid

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

Definition at line 432 of file walreceiver.h.

◆ walrcv_get_conninfo

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

Definition at line 412 of file walreceiver.h.

◆ walrcv_get_senderinfo

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

Definition at line 414 of file walreceiver.h.

◆ walrcv_identify_system

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

Definition at line 416 of file walreceiver.h.

◆ walrcv_readtimelinehistoryfile

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

Definition at line 420 of file walreceiver.h.

◆ walrcv_receive

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

Definition at line 426 of file walreceiver.h.

◆ walrcv_send

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

Definition at line 428 of file walreceiver.h.

◆ walrcv_server_version

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

Definition at line 418 of file walreceiver.h.

◆ walrcv_startstreaming

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

Definition at line 422 of file walreceiver.h.

Typedef Documentation

◆ walrcv_check_conninfo_fn

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

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

◆ walrcv_create_slot_fn

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

Definition at line 353 of file walreceiver.h.

◆ walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 385 of file walreceiver.h.

◆ walrcv_endstreaming_fn

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

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

◆ walrcv_get_backend_pid_fn

typedef pid_t(* walrcv_get_backend_pid_fn) (WalReceiverConn *conn)

Definition at line 365 of file walreceiver.h.

◆ walrcv_get_conninfo_fn

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

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

◆ walrcv_identify_system_fn

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

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

◆ walrcv_receive_fn

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

Definition at line 329 of file walreceiver.h.

◆ walrcv_send_fn

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

Definition at line 339 of file walreceiver.h.

◆ walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 286 of file walreceiver.h.

◆ walrcv_startstreaming_fn

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

Definition at line 309 of file walreceiver.h.

◆ WalRcvExecResult

◆ WalReceiverConn

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

203 {
204  WALRCV_ERROR, /* There was error when executing the query. */
205  WALRCV_OK_COMMAND, /* Query executed utility or replication
206  * command. */
207  WALRCV_OK_TUPLES, /* Query returned tuples. */
208  WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
209  WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
210  WALRCV_OK_COPY_BOTH /* Query started COPY BOTH replication
211  * protocol. */
WalRcvExecStatus
Definition: walreceiver.h:203
@ WALRCV_OK_COPY_IN
Definition: walreceiver.h:208
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ WALRCV_ERROR
Definition: walreceiver.h:204
@ WALRCV_OK_TUPLES
Definition: walreceiver.h:207
@ WALRCV_OK_COPY_OUT
Definition: walreceiver.h:209
@ WALRCV_OK_COPY_BOTH
Definition: walreceiver.h:210

◆ WalRcvState

Enumerator
WALRCV_STOPPED 
WALRCV_STARTING 
WALRCV_STREAMING 
WALRCV_WAITING 
WALRCV_RESTARTING 
WALRCV_STOPPING 

Definition at line 47 of file walreceiver.h.

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

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void  )

Definition at line 365 of file walreceiverfuncs.c.

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

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

Referenced by ProcessWalSndrMessage().

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void  )

Definition at line 395 of file walreceiverfuncs.c.

396 {
397  WalRcvData *walrcv = WalRcv;
398  TimestampTz lastMsgSendTime;
399  TimestampTz lastMsgReceiptTime;
400 
401  SpinLockAcquire(&walrcv->mutex);
402  lastMsgSendTime = walrcv->lastMsgSendTime;
403  lastMsgReceiptTime = walrcv->lastMsgReceiptTime;
404  SpinLockRelease(&walrcv->mutex);
405 
406  return TimestampDifferenceMilliseconds(lastMsgSendTime,
407  lastMsgReceiptTime);
408 }
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:102
TimestampTz lastMsgSendTime
Definition: walreceiver.h:101

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

Referenced by ProcessWalSndrMessage().

◆ GetWalRcvFlushRecPtr()

XLogRecPtr GetWalRcvFlushRecPtr ( XLogRecPtr latestChunkStart,
TimeLineID receiveTLI 
)

Definition at line 332 of file walreceiverfuncs.c.

333 {
334  WalRcvData *walrcv = WalRcv;
335  XLogRecPtr recptr;
336 
337  SpinLockAcquire(&walrcv->mutex);
338  recptr = walrcv->flushedUpto;
339  if (latestChunkStart)
340  *latestChunkStart = walrcv->latestChunkStart;
341  if (receiveTLI)
342  *receiveTLI = walrcv->receivedTLI;
343  SpinLockRelease(&walrcv->mutex);
344 
345  return recptr;
346 }
TimeLineID receivedTLI
Definition: walreceiver.h:88
XLogRecPtr latestChunkStart
Definition: walreceiver.h:96
static TimeLineID receiveTLI
Definition: xlogrecovery.c:263

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

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

◆ GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( void  )

Definition at line 353 of file walreceiverfuncs.c.

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

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

◆ ProcessWalRcvInterrupts()

void ProcessWalRcvInterrupts ( void  )

Definition at line 166 of file walreceiver.c.

167 {
168  /*
169  * Although walreceiver interrupt handling doesn't use the same scheme as
170  * regular backends, call CHECK_FOR_INTERRUPTS() to make sure we receive
171  * any incoming signals on Win32, and also to make sure we process any
172  * barrier events.
173  */
175 
177  {
178  ereport(FATAL,
179  (errcode(ERRCODE_ADMIN_SHUTDOWN),
180  errmsg("terminating walreceiver process due to administrator command")));
181  }
182 }
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define FATAL
Definition: elog.h:41
#define ereport(elevel,...)
Definition: elog.h:149
volatile sig_atomic_t ShutdownRequestPending
Definition: interrupt.c:28
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121

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

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

◆ RequestXLogStreaming()

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

Definition at line 246 of file walreceiverfuncs.c.

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

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

Referenced by WaitForWALToBecomeAvailable().

◆ ShutdownWalRcv()

void ShutdownWalRcv ( void  )

Definition at line 179 of file walreceiverfuncs.c.

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

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

◆ walrcv_clear_result()

static void walrcv_clear_result ( WalRcvExecResult walres)
inlinestatic

Definition at line 440 of file walreceiver.h.

441 {
442  if (!walres)
443  return;
444 
445  if (walres->err)
446  pfree(walres->err);
447 
448  if (walres->tuplestore)
449  tuplestore_end(walres->tuplestore);
450 
451  if (walres->tupledesc)
452  FreeTupleDesc(walres->tupledesc);
453 
454  pfree(walres);
455 }
void pfree(void *pointer)
Definition: mcxt.c:1436
Tuplestorestate * tuplestore
Definition: walreceiver.h:223
TupleDesc tupledesc
Definition: walreceiver.h:224
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:309
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:453

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

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

◆ WalRcvForceReply()

void WalRcvForceReply ( void  )

Definition at line 1353 of file walreceiver.c.

1354 {
1355  Latch *latch;
1356 
1357  WalRcv->force_reply = true;
1358  /* fetching the latch pointer might not be atomic, so use spinlock */
1360  latch = WalRcv->latch;
1362  if (latch)
1363  SetLatch(latch);
1364 }
sig_atomic_t force_reply
Definition: walreceiver.h:162

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

Referenced by ApplyWalRecord(), and WaitForWALToBecomeAvailable().

◆ WalRcvRunning()

bool WalRcvRunning ( void  )

Definition at line 76 of file walreceiverfuncs.c.

77 {
78  WalRcvData *walrcv = WalRcv;
80  pg_time_t startTime;
81 
82  SpinLockAcquire(&walrcv->mutex);
83 
84  state = walrcv->walRcvState;
85  startTime = walrcv->startTime;
86 
87  SpinLockRelease(&walrcv->mutex);
88 
89  /*
90  * If it has taken too long for walreceiver to start up, give up. Setting
91  * the state to STOPPED ensures that if walreceiver later does start up
92  * after all, it will see that it's not supposed to be running and die
93  * without doing anything.
94  */
95  if (state == WALRCV_STARTING)
96  {
97  pg_time_t now = (pg_time_t) time(NULL);
98 
99  if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
100  {
101  bool stopped = false;
102 
103  SpinLockAcquire(&walrcv->mutex);
104  if (walrcv->walRcvState == WALRCV_STARTING)
105  {
106  state = walrcv->walRcvState = WALRCV_STOPPED;
107  stopped = true;
108  }
109  SpinLockRelease(&walrcv->mutex);
110 
111  if (stopped)
113  }
114  }
115 
116  if (state != WALRCV_STOPPED)
117  return true;
118  else
119  return false;
120 }
Definition: regguts.h:318
#define WALRCV_STARTUP_TIMEOUT

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

Referenced by ShutdownWalRcv(), and StartupRequestWalReceiverRestart().

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void  )

Definition at line 55 of file walreceiverfuncs.c.

56 {
57  bool found;
58 
59  WalRcv = (WalRcvData *)
60  ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &found);
61 
62  if (!found)
63  {
64  /* First time through, so initialize */
70  WalRcv->latch = NULL;
71  }
72 }
static void pg_atomic_init_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:410
#define MemSet(start, val, len)
Definition: c.h:1004
void ConditionVariableInit(ConditionVariable *cv)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:396
#define SpinLockInit(lock)
Definition: spin.h:60
Size WalRcvShmemSize(void)

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

Referenced by CreateSharedMemoryAndSemaphores().

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void  )

Definition at line 44 of file walreceiverfuncs.c.

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

References add_size().

Referenced by CalculateShmemSize(), and WalRcvShmemInit().

◆ WalRcvStreaming()

bool WalRcvStreaming ( void  )

Definition at line 127 of file walreceiverfuncs.c.

128 {
129  WalRcvData *walrcv = WalRcv;
131  pg_time_t startTime;
132 
133  SpinLockAcquire(&walrcv->mutex);
134 
135  state = walrcv->walRcvState;
136  startTime = walrcv->startTime;
137 
138  SpinLockRelease(&walrcv->mutex);
139 
140  /*
141  * If it has taken too long for walreceiver to start up, give up. Setting
142  * the state to STOPPED ensures that if walreceiver later does start up
143  * after all, it will see that it's not supposed to be running and die
144  * without doing anything.
145  */
146  if (state == WALRCV_STARTING)
147  {
148  pg_time_t now = (pg_time_t) time(NULL);
149 
150  if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
151  {
152  bool stopped = false;
153 
154  SpinLockAcquire(&walrcv->mutex);
155  if (walrcv->walRcvState == WALRCV_STARTING)
156  {
157  state = walrcv->walRcvState = WALRCV_STOPPED;
158  stopped = true;
159  }
160  SpinLockRelease(&walrcv->mutex);
161 
162  if (stopped)
164  }
165  }
166 
169  return true;
170  else
171  return false;
172 }

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

Referenced by FinishWalRecovery(), and WaitForWALToBecomeAvailable().

◆ WalReceiverMain()

void WalReceiverMain ( void  )

Definition at line 187 of file walreceiver.c.

188 {
189  char conninfo[MAXCONNINFO];
190  char *tmp_conninfo;
191  char slotname[NAMEDATALEN];
192  bool is_temp_slot;
193  XLogRecPtr startpoint;
194  TimeLineID startpointTLI;
195  TimeLineID primaryTLI;
196  bool first_stream;
197  WalRcvData *walrcv = WalRcv;
199  char *err;
200  char *sender_host = NULL;
201  int sender_port = 0;
202 
203  /*
204  * WalRcv should be set up already (if we are a backend, we inherit this
205  * by fork() or EXEC_BACKEND mechanism from the postmaster).
206  */
207  Assert(walrcv != NULL);
208 
209  /*
210  * Mark walreceiver as running in shared memory.
211  *
212  * Do this as early as possible, so that if we fail later on, we'll set
213  * state to STOPPED. If we die before this, the startup process will keep
214  * waiting for us to start up, until it times out.
215  */
216  SpinLockAcquire(&walrcv->mutex);
217  Assert(walrcv->pid == 0);
218  switch (walrcv->walRcvState)
219  {
220  case WALRCV_STOPPING:
221  /* If we've already been requested to stop, don't start up. */
222  walrcv->walRcvState = WALRCV_STOPPED;
223  /* fall through */
224 
225  case WALRCV_STOPPED:
226  SpinLockRelease(&walrcv->mutex);
228  proc_exit(1);
229  break;
230 
231  case WALRCV_STARTING:
232  /* The usual case */
233  break;
234 
235  case WALRCV_WAITING:
236  case WALRCV_STREAMING:
237  case WALRCV_RESTARTING:
238  default:
239  /* Shouldn't happen */
240  SpinLockRelease(&walrcv->mutex);
241  elog(PANIC, "walreceiver still running according to shared memory state");
242  }
243  /* Advertise our PID so that the startup process can kill us */
244  walrcv->pid = MyProcPid;
245  walrcv->walRcvState = WALRCV_STREAMING;
246 
247  /* Fetch information required to start streaming */
248  walrcv->ready_to_display = false;
249  strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
250  strlcpy(slotname, (char *) walrcv->slotname, NAMEDATALEN);
251  is_temp_slot = walrcv->is_temp_slot;
252  startpoint = walrcv->receiveStart;
253  startpointTLI = walrcv->receiveStartTLI;
254 
255  /*
256  * At most one of is_temp_slot and slotname can be set; otherwise,
257  * RequestXLogStreaming messed up.
258  */
259  Assert(!is_temp_slot || (slotname[0] == '\0'));
260 
261  /* Initialise to a sanish value */
263  walrcv->lastMsgSendTime =
264  walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
265 
266  /* Report the latch to use to awaken this process */
267  walrcv->latch = &MyProc->procLatch;
268 
269  SpinLockRelease(&walrcv->mutex);
270 
272 
273  /* Arrange to clean up at walreceiver exit */
274  on_shmem_exit(WalRcvDie, PointerGetDatum(&startpointTLI));
275 
276  /* Properly accept or ignore signals the postmaster might send us */
277  pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
278  * file */
279  pqsignal(SIGINT, SIG_IGN);
280  pqsignal(SIGTERM, SignalHandlerForShutdownRequest); /* request shutdown */
281  /* SIGQUIT handler was already set up by InitPostmasterChild */
286 
287  /* Reset some signals that are accepted by postmaster but not here */
289 
290  /* Load the libpq-specific functions */
291  load_file("libpqwalreceiver", false);
292  if (WalReceiverFunctions == NULL)
293  elog(ERROR, "libpqwalreceiver didn't initialize correctly");
294 
295  /* Unblock signals (they were blocked when the postmaster forked us) */
297 
298  /* Establish the connection to the primary for XLOG streaming */
299  wrconn = walrcv_connect(conninfo, false,
300  cluster_name[0] ? cluster_name : "walreceiver",
301  &err);
302  if (!wrconn)
303  ereport(ERROR,
304  (errcode(ERRCODE_CONNECTION_FAILURE),
305  errmsg("could not connect to the primary server: %s", err)));
306 
307  /*
308  * Save user-visible connection string. This clobbers the original
309  * conninfo, for security. Also save host and port of the sender server
310  * this walreceiver is connected to.
311  */
312  tmp_conninfo = walrcv_get_conninfo(wrconn);
313  walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
314  SpinLockAcquire(&walrcv->mutex);
315  memset(walrcv->conninfo, 0, MAXCONNINFO);
316  if (tmp_conninfo)
317  strlcpy((char *) walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
318 
319  memset(walrcv->sender_host, 0, NI_MAXHOST);
320  if (sender_host)
321  strlcpy((char *) walrcv->sender_host, sender_host, NI_MAXHOST);
322 
323  walrcv->sender_port = sender_port;
324  walrcv->ready_to_display = true;
325  SpinLockRelease(&walrcv->mutex);
326 
327  if (tmp_conninfo)
328  pfree(tmp_conninfo);
329 
330  if (sender_host)
331  pfree(sender_host);
332 
333  first_stream = true;
334  for (;;)
335  {
336  char *primary_sysid;
337  char standby_sysid[32];
339 
340  /*
341  * Check that we're connected to a valid server using the
342  * IDENTIFY_SYSTEM replication command.
343  */
344  primary_sysid = walrcv_identify_system(wrconn, &primaryTLI);
345 
346  snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
348  if (strcmp(primary_sysid, standby_sysid) != 0)
349  {
350  ereport(ERROR,
351  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
352  errmsg("database system identifier differs between the primary and standby"),
353  errdetail("The primary's identifier is %s, the standby's identifier is %s.",
354  primary_sysid, standby_sysid)));
355  }
356 
357  /*
358  * Confirm that the current timeline of the primary is the same or
359  * ahead of ours.
360  */
361  if (primaryTLI < startpointTLI)
362  ereport(ERROR,
363  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
364  errmsg("highest timeline %u of the primary is behind recovery timeline %u",
365  primaryTLI, startpointTLI)));
366 
367  /*
368  * Get any missing history files. We do this always, even when we're
369  * not interested in that timeline, so that if we're promoted to
370  * become the primary later on, we don't select the same timeline that
371  * was already used in the current primary. This isn't bullet-proof -
372  * you'll need some external software to manage your cluster if you
373  * need to ensure that a unique timeline id is chosen in every case,
374  * but let's avoid the confusion of timeline id collisions where we
375  * can.
376  */
377  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
378 
379  /*
380  * Create temporary replication slot if requested, and update slot
381  * name in shared memory. (Note the slot name cannot already be set
382  * in this case.)
383  */
384  if (is_temp_slot)
385  {
386  snprintf(slotname, sizeof(slotname),
387  "pg_walreceiver_%lld",
388  (long long int) walrcv_get_backend_pid(wrconn));
389 
390  walrcv_create_slot(wrconn, slotname, true, false, 0, NULL);
391 
392  SpinLockAcquire(&walrcv->mutex);
393  strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
394  SpinLockRelease(&walrcv->mutex);
395  }
396 
397  /*
398  * Start streaming.
399  *
400  * We'll try to start at the requested starting point and timeline,
401  * even if it's different from the server's latest timeline. In case
402  * we've already reached the end of the old timeline, the server will
403  * finish the streaming immediately, and we will go back to await
404  * orders from the startup process. If recovery_target_timeline is
405  * 'latest', the startup process will scan pg_wal and find the new
406  * history file, bump recovery target timeline, and ask us to restart
407  * on the new timeline.
408  */
409  options.logical = false;
410  options.startpoint = startpoint;
411  options.slotname = slotname[0] != '\0' ? slotname : NULL;
412  options.proto.physical.startpointTLI = startpointTLI;
414  {
415  if (first_stream)
416  ereport(LOG,
417  (errmsg("started streaming WAL from primary at %X/%X on timeline %u",
418  LSN_FORMAT_ARGS(startpoint), startpointTLI)));
419  else
420  ereport(LOG,
421  (errmsg("restarted WAL streaming at %X/%X on timeline %u",
422  LSN_FORMAT_ARGS(startpoint), startpointTLI)));
423  first_stream = false;
424 
425  /* Initialize LogstreamResult and buffers for processing messages */
429 
430  /* Initialize nap wakeup times. */
432  for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
434 
435  /* Send initial reply/feedback messages. */
436  XLogWalRcvSendReply(true, false);
438 
439  /* Loop until end-of-streaming or error */
440  for (;;)
441  {
442  char *buf;
443  int len;
444  bool endofwal = false;
445  pgsocket wait_fd = PGINVALID_SOCKET;
446  int rc;
447  TimestampTz nextWakeup;
448  long nap;
449 
450  /*
451  * Exit walreceiver if we're not in recovery. This should not
452  * happen, but cross-check the status here.
453  */
454  if (!RecoveryInProgress())
455  ereport(FATAL,
456  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
457  errmsg("cannot continue WAL streaming, recovery has already ended")));
458 
459  /* Process any requests or signals received recently */
461 
463  {
464  ConfigReloadPending = false;
466  /* recompute wakeup times */
468  for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
471  }
472 
473  /* See if we can read data immediately */
474  len = walrcv_receive(wrconn, &buf, &wait_fd);
475  if (len != 0)
476  {
477  /*
478  * Process the received data, and any subsequent data we
479  * can read without blocking.
480  */
481  for (;;)
482  {
483  if (len > 0)
484  {
485  /*
486  * Something was received from primary, so adjust
487  * the ping and terminate wakeup times.
488  */
491  now);
493  XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
494  startpointTLI);
495  }
496  else if (len == 0)
497  break;
498  else if (len < 0)
499  {
500  ereport(LOG,
501  (errmsg("replication terminated by primary server"),
502  errdetail("End of WAL reached on timeline %u at %X/%X.",
503  startpointTLI,
505  endofwal = true;
506  break;
507  }
508  len = walrcv_receive(wrconn, &buf, &wait_fd);
509  }
510 
511  /* Let the primary know that we received some data. */
512  XLogWalRcvSendReply(false, false);
513 
514  /*
515  * If we've written some records, flush them to disk and
516  * let the startup process and primary server know about
517  * them.
518  */
519  XLogWalRcvFlush(false, startpointTLI);
520  }
521 
522  /* Check if we need to exit the streaming loop. */
523  if (endofwal)
524  break;
525 
526  /* Find the soonest wakeup time, to limit our nap. */
527  nextWakeup = TIMESTAMP_INFINITY;
528  for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
529  nextWakeup = Min(wakeup[i], nextWakeup);
530 
531  /* Calculate the nap time, clamping as necessary. */
533  nap = TimestampDifferenceMilliseconds(now, nextWakeup);
534 
535  /*
536  * Ideally we would reuse a WaitEventSet object repeatedly
537  * here to avoid the overheads of WaitLatchOrSocket on epoll
538  * systems, but we can't be sure that libpq (or any other
539  * walreceiver implementation) has the same socket (even if
540  * the fd is the same number, it may have been closed and
541  * reopened since the last time). In future, if there is a
542  * function for removing sockets from WaitEventSet, then we
543  * could add and remove just the socket each time, potentially
544  * avoiding some system calls.
545  */
546  Assert(wait_fd != PGINVALID_SOCKET);
550  wait_fd,
551  nap,
553  if (rc & WL_LATCH_SET)
554  {
557 
558  if (walrcv->force_reply)
559  {
560  /*
561  * The recovery process has asked us to send apply
562  * feedback now. Make sure the flag is really set to
563  * false in shared memory before sending the reply, so
564  * we don't miss a new request for a reply.
565  */
566  walrcv->force_reply = false;
568  XLogWalRcvSendReply(true, false);
569  }
570  }
571  if (rc & WL_TIMEOUT)
572  {
573  /*
574  * We didn't receive anything new. If we haven't heard
575  * anything from the server for more than
576  * wal_receiver_timeout / 2, ping the server. Also, if
577  * it's been longer than wal_receiver_status_interval
578  * since the last update we sent, send a status update to
579  * the primary anyway, to report any progress in applying
580  * WAL.
581  */
582  bool requestReply = false;
583 
584  /*
585  * Check if time since last receive from primary has
586  * reached the configured limit.
587  */
590  ereport(ERROR,
591  (errcode(ERRCODE_CONNECTION_FAILURE),
592  errmsg("terminating walreceiver due to timeout")));
593 
594  /*
595  * If we didn't receive anything new for half of receiver
596  * replication timeout, then ping the server.
597  */
598  if (now >= wakeup[WALRCV_WAKEUP_PING])
599  {
600  requestReply = true;
602  }
603 
604  XLogWalRcvSendReply(requestReply, requestReply);
606  }
607  }
608 
609  /*
610  * The backend finished streaming. Exit streaming COPY-mode from
611  * our side, too.
612  */
613  walrcv_endstreaming(wrconn, &primaryTLI);
614 
615  /*
616  * If the server had switched to a new timeline that we didn't
617  * know about when we began streaming, fetch its timeline history
618  * file now.
619  */
620  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
621  }
622  else
623  ereport(LOG,
624  (errmsg("primary server contains no more WAL on requested timeline %u",
625  startpointTLI)));
626 
627  /*
628  * End of WAL reached on the requested timeline. Close the last
629  * segment, and await for new orders from the startup process.
630  */
631  if (recvFile >= 0)
632  {
633  char xlogfname[MAXFNAMELEN];
634 
635  XLogWalRcvFlush(false, startpointTLI);
637  if (close(recvFile) != 0)
638  ereport(PANIC,
640  errmsg("could not close WAL segment %s: %m",
641  xlogfname)));
642 
643  /*
644  * Create .done file forcibly to prevent the streamed segment from
645  * being archived later.
646  */
648  XLogArchiveForceDone(xlogfname);
649  else
650  XLogArchiveNotify(xlogfname);
651  }
652  recvFile = -1;
653 
654  elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
655  WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
656  }
657  /* not reached */
658 }
static void pg_atomic_write_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:433
#define pg_memory_barrier()
Definition: atomics.h:140
sigset_t UnBlockSig
Definition: pqsignal.c:22
#define Min(x, y)
Definition: c.h:988
#define UINT64_FORMAT
Definition: c.h:533
#define TIMESTAMP_INFINITY
Definition: timestamp.h:150
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
int errcode_for_file_access(void)
Definition: elog.c:881
int errdetail(const char *fmt,...)
Definition: elog.c:1202
#define LOG
Definition: elog.h:31
#define PANIC
Definition: elog.h:42
#define DEBUG1
Definition: elog.h:30
#define ERROR
Definition: elog.h:39
int MyProcPid
Definition: globals.c:44
struct Latch * MyLatch
Definition: globals.c:58
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
char * cluster_name
Definition: guc_tables.c:508
#define close(a)
Definition: win32.h:12
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:109
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void on_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:361
void proc_exit(int code)
Definition: ipc.c:104
int i
Definition: isn.c:73
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:540
void ResetLatch(Latch *latch)
Definition: latch.c:699
#define WL_SOCKET_READABLE
Definition: latch.h:126
#define WL_TIMEOUT
Definition: latch.h:128
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
#define WL_LATCH_SET
Definition: latch.h:125
const void size_t len
static char ** options
static char * buf
Definition: pg_test_fsync.c:67
pqsigfunc pqsignal(int signo, pqsigfunc func)
int pgsocket
Definition: port.h:29
#define snprintf
Definition: port.h:238
#define PGINVALID_SOCKET
Definition: port.h:31
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
#define PG_SETMASK(mask)
Definition: pqsignal.h:18
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:639
PGPROC * MyProc
Definition: proc.c:66
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
Latch procLatch
Definition: proc.h:170
char sender_host[NI_MAXHOST]
Definition: walreceiver.h:120
TimestampTz latestWalEndTime
Definition: walreceiver.h:108
bool ready_to_display
Definition: walreceiver.h:136
int sender_port
Definition: walreceiver.h:121
@ WAIT_EVENT_WAL_RECEIVER_MAIN
Definition: wait_event.h:48
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:166
#define NUM_WALRCV_WAKEUPS
Definition: walreceiver.c:126
static WalReceiverConn * wrconn
Definition: walreceiver.c:95
static TimestampTz wakeup[NUM_WALRCV_WAKEUPS]
Definition: walreceiver.c:132
static StringInfoData reply_message
Definition: walreceiver.c:134
static int recvFile
Definition: walreceiver.c:103
static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
Definition: walreceiver.c:744
static StringInfoData incoming_message
Definition: walreceiver.c:135
static void XLogWalRcvFlush(bool dying, TimeLineID tli)
Definition: walreceiver.c:988
static struct @18 LogstreamResult
static TimeLineID recvFileTLI
Definition: walreceiver.c:104
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:96
static XLogSegNo recvSegNo
Definition: walreceiver.c:105
static void XLogWalRcvSendHSFeedback(bool immed)
Definition: walreceiver.c:1164
@ WALRCV_WAKEUP_TERMINATE
Definition: walreceiver.c:122
@ WALRCV_WAKEUP_PING
Definition: walreceiver.c:123
static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
Definition: walreceiver.c:664
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
Definition: walreceiver.c:838
static void WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now)
Definition: walreceiver.c:1312
static void WalRcvDie(int code, Datum arg)
Definition: walreceiver.c:800
static void XLogWalRcvSendReply(bool force, bool requestReply)
Definition: walreceiver.c:1095
#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
Definition: walreceiver.h:430
#define walrcv_startstreaming(conn, options)
Definition: walreceiver.h:422
#define walrcv_connect(conninfo, logical, appname, err)
Definition: walreceiver.h:408
#define walrcv_get_senderinfo(conn, sender_host, sender_port)
Definition: walreceiver.h:414
#define walrcv_get_conninfo(conn)
Definition: walreceiver.h:412
#define walrcv_endstreaming(conn, next_tli)
Definition: walreceiver.h:424
#define walrcv_identify_system(conn, primary_tli)
Definition: walreceiver.h:416
#define walrcv_get_backend_pid(conn)
Definition: walreceiver.h:432
#define walrcv_receive(conn, buffer, wait_fd)
Definition: walreceiver.h:426
#define SIGCHLD
Definition: win32_port.h:186
#define SIGHUP
Definition: win32_port.h:176
#define SIG_DFL
Definition: win32_port.h:171
#define SIGPIPE
Definition: win32_port.h:181
#define SIGUSR1
Definition: win32_port.h:188
#define SIGALRM
Definition: win32_port.h:182
#define SIGUSR2
Definition: win32_port.h:189
#define SIG_IGN
Definition: win32_port.h:173
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4175
bool RecoveryInProgress(void)
Definition: xlog.c:5908
int XLogArchiveMode
Definition: xlog.c:122
@ ARCHIVE_MODE_ALWAYS
Definition: xlog.h:62
#define MAXFNAMELEN
static void XLogFileName(char *fname, TimeLineID tli, XLogSegNo logSegNo, int wal_segsz_bytes)
void XLogArchiveForceDone(const char *xlog)
Definition: xlogarchive.c:389
void XLogArchiveNotify(const char *xlog)
Definition: xlogarchive.c:323
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
uint32 TimeLineID
Definition: xlogdefs.h:59

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

Referenced by AuxiliaryProcessMain().

Variable Documentation

◆ hot_standby_feedback

PGDLLIMPORT bool hot_standby_feedback
extern

Definition at line 92 of file walreceiver.c.

Referenced by WalRcvComputeNextWakeup(), and XLogWalRcvSendHSFeedback().

◆ wal_receiver_status_interval

PGDLLIMPORT int wal_receiver_status_interval
extern

◆ wal_receiver_timeout

PGDLLIMPORT int wal_receiver_timeout
extern

◆ WalRcv

◆ WalReceiverFunctions

PGDLLIMPORT WalReceiverFunctionsType* WalReceiverFunctions
extern

Definition at line 96 of file walreceiver.c.

Referenced by _PG_init(), and WalReceiverMain().