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

Go to the source code of this file.

Data Structures

struct  WalRcvData
 
struct  WalRcvStreamOptions
 
struct  WalRcvExecResult
 
struct  WalReceiverFunctionsType
 

Macros

#define MAXCONNINFO   1024
 
#define AllowCascadeReplication()   (EnableHotStandby && max_wal_senders > 0)
 
#define walrcv_connect(conninfo, logical, appname, err)   WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
 
#define walrcv_check_conninfo(conninfo)   WalReceiverFunctions->walrcv_check_conninfo(conninfo)
 
#define walrcv_get_conninfo(conn)   WalReceiverFunctions->walrcv_get_conninfo(conn)
 
#define walrcv_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_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 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)
 
XLogRecPtr GetWalRcvWriteRecPtr (XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI)
 
int GetReplicationApplyDelay (void)
 
int GetReplicationTransferLatency (void)
 
void WalRcvForceReply (void)
 

Variables

int wal_receiver_status_interval
 
int wal_receiver_timeout
 
bool hot_standby_feedback
 
WalRcvDataWalRcv
 
PGDLLIMPORT WalReceiverFunctionsTypeWalReceiverFunctions
 

Macro Definition Documentation

◆ AllowCascadeReplication

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

Definition at line 38 of file walreceiver.h.

Referenced by StartupXLOG(), and XLogWalRcvFlush().

◆ MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 35 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 257 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 277 of file walreceiver.h.

Referenced by CreateSubscription(), and LogicalRepSyncTableStart().

◆ 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_conninfo

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

Definition at line 259 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 261 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 263 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 267 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 273 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 275 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 265 of file walreceiver.h.

Referenced by fetch_remote_table_info().

◆ walrcv_startstreaming

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

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

◆ walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 233 of file walreceiver.h.

◆ walrcv_endstreaming_fn

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

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

◆ walrcv_get_conninfo_fn

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

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

◆ walrcv_identify_system_fn

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

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

◆ walrcv_receive_fn

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

Definition at line 221 of file walreceiver.h.

◆ walrcv_send_fn

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

Definition at line 223 of file walreceiver.h.

◆ walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 212 of file walreceiver.h.

◆ walrcv_startstreaming_fn

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

Definition at line 217 of file walreceiver.h.

◆ WalRcvExecResult

◆ WalReceiverConn

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

178 {
179  WALRCV_ERROR, /* There was error when executing the query. */
180  WALRCV_OK_COMMAND, /* Query executed utility or replication
181  * command. */
182  WALRCV_OK_TUPLES, /* Query returned tuples. */
183  WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
184  WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
185  WALRCV_OK_COPY_BOTH /* Query started COPY BOTH replication
186  * protocol. */
WalRcvExecStatus
Definition: walreceiver.h:177

◆ WalRcvState

Enumerator
WALRCV_STOPPED 
WALRCV_STARTING 
WALRCV_STREAMING 
WALRCV_WAITING 
WALRCV_RESTARTING 
WALRCV_STOPPING 

Definition at line 43 of file walreceiver.h.

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

Function Documentation

◆ GetReplicationApplyDelay()

int GetReplicationApplyDelay ( void  )

Definition at line 318 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

319 {
320  WalRcvData *walrcv = WalRcv;
321  XLogRecPtr receivePtr;
322  XLogRecPtr replayPtr;
323 
324  long secs;
325  int usecs;
326 
327  TimestampTz chunkReplayStartTime;
328 
329  SpinLockAcquire(&walrcv->mutex);
330  receivePtr = walrcv->receivedUpto;
331  SpinLockRelease(&walrcv->mutex);
332 
333  replayPtr = GetXLogReplayRecPtr(NULL);
334 
335  if (receivePtr == replayPtr)
336  return 0;
337 
338  chunkReplayStartTime = GetCurrentChunkReplayStartTime();
339 
340  if (chunkReplayStartTime == 0)
341  return -1;
342 
343  TimestampDifference(chunkReplayStartTime,
345  &secs, &usecs);
346 
347  return (((int) secs * 1000) + (usecs / 1000));
348 }
slock_t mutex
Definition: walreceiver.h:136
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1569
int64 TimestampTz
Definition: timestamp.h:39
#define SpinLockAcquire(lock)
Definition: spin.h:62
TimestampTz GetCurrentChunkReplayStartTime(void)
Definition: xlog.c:6094
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11130
XLogRecPtr receivedUpto
Definition: walreceiver.h:82
#define SpinLockRelease(lock)
Definition: spin.h:64
uint64 XLogRecPtr
Definition: xlogdefs.h:21
WalRcvData * WalRcv
void TimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: timestamp.c:1643

◆ GetReplicationTransferLatency()

int GetReplicationTransferLatency ( void  )

Definition at line 355 of file walreceiverfuncs.c.

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

Referenced by ProcessWalSndrMessage(), and walrcv_clear_result().

356 {
357  WalRcvData *walrcv = WalRcv;
358 
359  TimestampTz lastMsgSendTime;
360  TimestampTz lastMsgReceiptTime;
361 
362  long secs = 0;
363  int usecs = 0;
364  int ms;
365 
366  SpinLockAcquire(&walrcv->mutex);
367  lastMsgSendTime = walrcv->lastMsgSendTime;
368  lastMsgReceiptTime = walrcv->lastMsgReceiptTime;
369  SpinLockRelease(&walrcv->mutex);
370 
371  TimestampDifference(lastMsgSendTime,
372  lastMsgReceiptTime,
373  &secs, &usecs);
374 
375  ms = ((int) secs * 1000) + (usecs / 1000);
376 
377  return ms;
378 }
slock_t mutex
Definition: walreceiver.h:136
int64 TimestampTz
Definition: timestamp.h:39
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:97
TimestampTz lastMsgSendTime
Definition: walreceiver.h:96
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
WalRcvData * WalRcv
void TimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: timestamp.c:1643

◆ GetWalRcvWriteRecPtr()

XLogRecPtr GetWalRcvWriteRecPtr ( XLogRecPtr latestChunkStart,
TimeLineID receiveTLI 
)

Definition at line 297 of file walreceiverfuncs.c.

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

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

298 {
299  WalRcvData *walrcv = WalRcv;
300  XLogRecPtr recptr;
301 
302  SpinLockAcquire(&walrcv->mutex);
303  recptr = walrcv->receivedUpto;
304  if (latestChunkStart)
305  *latestChunkStart = walrcv->latestChunkStart;
306  if (receiveTLI)
307  *receiveTLI = walrcv->receivedTLI;
308  SpinLockRelease(&walrcv->mutex);
309 
310  return recptr;
311 }
slock_t mutex
Definition: walreceiver.h:136
TimeLineID receivedTLI
Definition: walreceiver.h:83
#define SpinLockAcquire(lock)
Definition: spin.h:62
XLogRecPtr latestChunkStart
Definition: walreceiver.h:91
XLogRecPtr receivedUpto
Definition: walreceiver.h:82
#define SpinLockRelease(lock)
Definition: spin.h:64
static TimeLineID receiveTLI
Definition: xlog.c:209
uint64 XLogRecPtr
Definition: xlogdefs.h:21
WalRcvData * WalRcv

◆ ProcessWalRcvInterrupts()

void ProcessWalRcvInterrupts ( void  )

Definition at line 147 of file walreceiver.c.

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

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

148 {
149  /*
150  * Although walreceiver interrupt handling doesn't use the same scheme as
151  * regular backends, call CHECK_FOR_INTERRUPTS() to make sure we receive
152  * any incoming signals on Win32.
153  */
155 
156  if (got_SIGTERM)
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:570
#define FATAL
Definition: elog.h:52
#define ereport(elevel, rest)
Definition: elog.h:141
int errmsg(const char *fmt,...)
Definition: elog.c:784
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
static volatile sig_atomic_t got_SIGTERM
Definition: walreceiver.c:100

◆ RequestXLogStreaming()

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

Definition at line 223 of file walreceiverfuncs.c.

References Assert, WalRcvData::conninfo, WalRcvData::latch, WalRcvData::latestChunkStart, MAXCONNINFO, WalRcvData::mutex, NAMEDATALEN, now(), PMSIGNAL_START_WALRECEIVER, WalRcvData::receivedTLI, WalRcvData::receivedUpto, 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().

225 {
226  WalRcvData *walrcv = WalRcv;
227  bool launch = false;
228  pg_time_t now = (pg_time_t) time(NULL);
229  Latch *latch;
230 
231  /*
232  * We always start at the beginning of the segment. That prevents a broken
233  * segment (i.e., with no records in the first half of a segment) from
234  * being created by XLOG streaming, which might cause trouble later on if
235  * the segment is e.g archived.
236  */
237  if (XLogSegmentOffset(recptr, wal_segment_size) != 0)
238  recptr -= XLogSegmentOffset(recptr, wal_segment_size);
239 
240  SpinLockAcquire(&walrcv->mutex);
241 
242  /* It better be stopped if we try to restart it */
243  Assert(walrcv->walRcvState == WALRCV_STOPPED ||
244  walrcv->walRcvState == WALRCV_WAITING);
245 
246  if (conninfo != NULL)
247  strlcpy((char *) walrcv->conninfo, conninfo, MAXCONNINFO);
248  else
249  walrcv->conninfo[0] = '\0';
250 
251  if (slotname != NULL)
252  strlcpy((char *) walrcv->slotname, slotname, NAMEDATALEN);
253  else
254  walrcv->slotname[0] = '\0';
255 
256  if (walrcv->walRcvState == WALRCV_STOPPED)
257  {
258  launch = true;
259  walrcv->walRcvState = WALRCV_STARTING;
260  }
261  else
262  walrcv->walRcvState = WALRCV_RESTARTING;
263  walrcv->startTime = now;
264 
265  /*
266  * If this is the first startup of walreceiver (on this timeline),
267  * initialize receivedUpto and latestChunkStart to the starting point.
268  */
269  if (walrcv->receiveStart == 0 || walrcv->receivedTLI != tli)
270  {
271  walrcv->receivedUpto = recptr;
272  walrcv->receivedTLI = tli;
273  walrcv->latestChunkStart = recptr;
274  }
275  walrcv->receiveStart = recptr;
276  walrcv->receiveStartTLI = tli;
277 
278  latch = walrcv->latch;
279 
280  SpinLockRelease(&walrcv->mutex);
281 
282  if (launch)
284  else if (latch)
285  SetLatch(latch);
286 }
int64 pg_time_t
Definition: pgtime.h:23
slock_t mutex
Definition: walreceiver.h:136
int wal_segment_size
Definition: xlog.c:112
WalRcvState walRcvState
Definition: walreceiver.h:63
TimeLineID receivedTLI
Definition: walreceiver.h:83
void SetLatch(Latch *latch)
Definition: latch.c:436
pg_time_t startTime
Definition: walreceiver.h:64
#define NAMEDATALEN
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define MAXCONNINFO
Definition: walreceiver.h:35
XLogRecPtr latestChunkStart
Definition: walreceiver.h:91
Definition: latch.h:110
XLogRecPtr receivedUpto
Definition: walreceiver.h:82
TimeLineID receiveStartTLI
Definition: walreceiver.h:73
Latch * latch
Definition: walreceiver.h:134
#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:732
WalRcvData * WalRcv
void SendPostmasterSignal(PMSignalReason reason)
Definition: pmsignal.c:146
XLogRecPtr receiveStart
Definition: walreceiver.h:72
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1533
char slotname[NAMEDATALEN]
Definition: walreceiver.h:122
char conninfo[MAXCONNINFO]
Definition: walreceiver.h:109

◆ ShutdownWalRcv()

void ShutdownWalRcv ( void  )

Definition at line 163 of file walreceiverfuncs.c.

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

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

164 {
165  WalRcvData *walrcv = WalRcv;
166  pid_t walrcvpid = 0;
167 
168  /*
169  * Request walreceiver to stop. Walreceiver will switch to WALRCV_STOPPED
170  * mode once it's finished, and will also request postmaster to not
171  * restart itself.
172  */
173  SpinLockAcquire(&walrcv->mutex);
174  switch (walrcv->walRcvState)
175  {
176  case WALRCV_STOPPED:
177  break;
178  case WALRCV_STARTING:
179  walrcv->walRcvState = WALRCV_STOPPED;
180  break;
181 
182  case WALRCV_STREAMING:
183  case WALRCV_WAITING:
184  case WALRCV_RESTARTING:
185  walrcv->walRcvState = WALRCV_STOPPING;
186  /* fall through */
187  case WALRCV_STOPPING:
188  walrcvpid = walrcv->pid;
189  break;
190  }
191  SpinLockRelease(&walrcv->mutex);
192 
193  /*
194  * Signal walreceiver process if it was still running.
195  */
196  if (walrcvpid != 0)
197  kill(walrcvpid, SIGTERM);
198 
199  /*
200  * Wait for walreceiver to acknowledge its death by setting state to
201  * WALRCV_STOPPED.
202  */
203  while (WalRcvRunning())
204  {
205  /*
206  * This possibly-long loop needs to handle interrupts of startup
207  * process.
208  */
210 
211  pg_usleep(100000); /* 100ms */
212  }
213 }
slock_t mutex
Definition: walreceiver.h:136
WalRcvState walRcvState
Definition: walreceiver.h:63
#define kill(pid, sig)
Definition: win32_port.h:435
#define SpinLockAcquire(lock)
Definition: spin.h:62
void pg_usleep(long microsec)
Definition: signal.c:53
pid_t pid
Definition: walreceiver.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
WalRcvData * WalRcv
void HandleStartupProcInterrupts(void)
Definition: startup.c:142
bool WalRcvRunning(void)

◆ walrcv_clear_result()

static void walrcv_clear_result ( WalRcvExecResult walres)
inlinestatic

Definition at line 285 of file walreceiver.h.

References WalRcvExecResult::err, FreeTupleDesc(), GetReplicationApplyDelay(), GetReplicationTransferLatency(), 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(), DropSubscription(), fetch_remote_table_info(), fetch_table_list(), and LogicalRepSyncTableStart().

286 {
287  if (!walres)
288  return;
289 
290  if (walres->err)
291  pfree(walres->err);
292 
293  if (walres->tuplestore)
294  tuplestore_end(walres->tuplestore);
295 
296  if (walres->tupledesc)
297  FreeTupleDesc(walres->tupledesc);
298 
299  pfree(walres);
300 }
void pfree(void *pointer)
Definition: mcxt.c:1031
TupleDesc tupledesc
Definition: walreceiver.h:198
Tuplestorestate * tuplestore
Definition: walreceiver.h:197
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:453
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:313

◆ WalRcvForceReply()

void WalRcvForceReply ( void  )

Definition at line 1297 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().

1298 {
1299  Latch *latch;
1300 
1301  WalRcv->force_reply = true;
1302  /* fetching the latch pointer might not be atomic, so use spinlock */
1304  latch = WalRcv->latch;
1306  if (latch)
1307  SetLatch(latch);
1308 }
slock_t mutex
Definition: walreceiver.h:136
sig_atomic_t force_reply
Definition: walreceiver.h:143
void SetLatch(Latch *latch)
Definition: latch.c:436
#define SpinLockAcquire(lock)
Definition: spin.h:62
Definition: latch.h:110
Latch * latch
Definition: walreceiver.h:134
#define SpinLockRelease(lock)
Definition: spin.h:64
WalRcvData * WalRcv

◆ WalRcvRunning()

bool WalRcvRunning ( void  )

Definition at line 72 of file walreceiverfuncs.c.

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

Referenced by ShutdownWalRcv(), and walrcv_clear_result().

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

◆ WalRcvShmemInit()

void WalRcvShmemInit ( void  )

Definition at line 53 of file walreceiverfuncs.c.

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

Referenced by CreateSharedMemoryAndSemaphores(), and walrcv_clear_result().

54 {
55  bool found;
56 
57  WalRcv = (WalRcvData *)
58  ShmemInitStruct("Wal Receiver Ctl", WalRcvShmemSize(), &found);
59 
60  if (!found)
61  {
62  /* First time through, so initialize */
66  WalRcv->latch = NULL;
67  }
68 }
slock_t mutex
Definition: walreceiver.h:136
WalRcvState walRcvState
Definition: walreceiver.h:63
#define SpinLockInit(lock)
Definition: spin.h:60
#define MemSet(start, val, len)
Definition: c.h:955
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:372
Latch * latch
Definition: walreceiver.h:134
Size WalRcvShmemSize(void)
WalRcvData * WalRcv

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void  )

Definition at line 42 of file walreceiverfuncs.c.

References add_size().

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

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

◆ WalRcvStreaming()

bool WalRcvStreaming ( void  )

Definition at line 117 of file walreceiverfuncs.c.

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

Referenced by WaitForWALToBecomeAvailable(), and walrcv_clear_result().

118 {
119  WalRcvData *walrcv = WalRcv;
121  pg_time_t startTime;
122 
123  SpinLockAcquire(&walrcv->mutex);
124 
125  state = walrcv->walRcvState;
126  startTime = walrcv->startTime;
127 
128  SpinLockRelease(&walrcv->mutex);
129 
130  /*
131  * If it has taken too long for walreceiver to start up, give up. Setting
132  * the state to STOPPED ensures that if walreceiver later does start up
133  * after all, it will see that it's not supposed to be running and die
134  * without doing anything.
135  */
136  if (state == WALRCV_STARTING)
137  {
138  pg_time_t now = (pg_time_t) time(NULL);
139 
140  if ((now - startTime) > WALRCV_STARTUP_TIMEOUT)
141  {
142  SpinLockAcquire(&walrcv->mutex);
143 
144  if (walrcv->walRcvState == WALRCV_STARTING)
145  state = walrcv->walRcvState = WALRCV_STOPPED;
146 
147  SpinLockRelease(&walrcv->mutex);
148  }
149  }
150 
151  if (state == WALRCV_STREAMING || state == WALRCV_STARTING ||
152  state == WALRCV_RESTARTING)
153  return true;
154  else
155  return false;
156 }
int64 pg_time_t
Definition: pgtime.h:23
slock_t mutex
Definition: walreceiver.h:136
WalRcvState walRcvState
Definition: walreceiver.h:63
pg_time_t startTime
Definition: walreceiver.h:64
WalRcvState
Definition: walreceiver.h:43
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define SpinLockRelease(lock)
Definition: spin.h:64
#define WALRCV_STARTUP_TIMEOUT
Definition: regguts.h:298
WalRcvData * WalRcv
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1533

◆ WalReceiverMain()

void WalReceiverMain ( void  )

Definition at line 167 of file walreceiver.c.

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

Variable Documentation

◆ hot_standby_feedback

bool hot_standby_feedback

Definition at line 77 of file walreceiver.c.

Referenced by XLogWalRcvSendHSFeedback().

◆ wal_receiver_status_interval

int wal_receiver_status_interval

Definition at line 75 of file walreceiver.c.

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

◆ wal_receiver_timeout

int wal_receiver_timeout

Definition at line 76 of file walreceiver.c.

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

◆ WalRcv

◆ WalReceiverFunctions

PGDLLIMPORT WalReceiverFunctionsType* WalReceiverFunctions

Definition at line 81 of file walreceiver.c.

Referenced by _PG_init().