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

Go to the source code of this file.

Data Structures

struct   WalRcvData
 
struct   WalRcvStreamOptions
 
struct   WalRcvExecResult
 
struct   WalReceiverFunctionsType
 

Macros

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

Typedefs

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

Enumerations

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

Functions

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

Variables

PGDLLIMPORT int  wal_receiver_status_interval
 
PGDLLIMPORT int  wal_receiver_timeout
 
PGDLLIMPORT bool  hot_standby_feedback
 
PGDLLIMPORT WalRcvData *  WalRcv
 
PGDLLIMPORT WalReceiverFunctionsType *  WalReceiverFunctions
 

Macro Definition Documentation

◆ AllowCascadeReplication

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

Definition at line 40 of file walreceiver.h.

◆ MAXCONNINFO

#define MAXCONNINFO   1024

Definition at line 37 of file walreceiver.h.

◆ walrcv_alter_slot

#define walrcv_alter_slot (   conn,
  slotname,
  failover,
  two_phase 
)     WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)

Definition at line 461 of file walreceiver.h.

◆ walrcv_check_conninfo

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

Definition at line 437 of file walreceiver.h.

◆ walrcv_connect

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

Definition at line 435 of file walreceiver.h.

◆ walrcv_create_slot

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

Definition at line 459 of file walreceiver.h.

◆ walrcv_disconnect

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

Definition at line 467 of file walreceiver.h.

◆ walrcv_endstreaming

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

Definition at line 453 of file walreceiver.h.

◆ walrcv_exec

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

Definition at line 465 of file walreceiver.h.

◆ walrcv_get_backend_pid

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

Definition at line 463 of file walreceiver.h.

◆ walrcv_get_conninfo

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

Definition at line 439 of file walreceiver.h.

◆ walrcv_get_dbname_from_conninfo

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

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

◆ walrcv_identify_system

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

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

◆ walrcv_receive

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

Definition at line 455 of file walreceiver.h.

◆ walrcv_send

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

Definition at line 457 of file walreceiver.h.

◆ walrcv_server_version

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

Definition at line 447 of file walreceiver.h.

◆ walrcv_startstreaming

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

Definition at line 451 of file walreceiver.h.

Typedef Documentation

◆ walrcv_alter_slot_fn

typedef void(* walrcv_alter_slot_fn) (WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)

Definition at line 379 of file walreceiver.h.

◆ walrcv_check_conninfo_fn

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

Definition at line 254 of file walreceiver.h.

◆ walrcv_connect_fn

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

Definition at line 242 of file walreceiver.h.

◆ walrcv_create_slot_fn

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

Definition at line 365 of file walreceiver.h.

◆ walrcv_disconnect_fn

typedef void(* walrcv_disconnect_fn) (WalReceiverConn *conn)

Definition at line 410 of file walreceiver.h.

◆ walrcv_endstreaming_fn

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

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

◆ walrcv_get_backend_pid_fn

typedef pid_t(* walrcv_get_backend_pid_fn) (WalReceiverConn *conn)

Definition at line 390 of file walreceiver.h.

◆ walrcv_get_conninfo_fn

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

Definition at line 263 of file walreceiver.h.

◆ walrcv_get_dbname_from_conninfo_fn

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

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

◆ walrcv_identify_system_fn

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

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

◆ walrcv_receive_fn

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

Definition at line 341 of file walreceiver.h.

◆ walrcv_send_fn

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

Definition at line 351 of file walreceiver.h.

◆ walrcv_server_version_fn

typedef int(* walrcv_server_version_fn) (WalReceiverConn *conn)

Definition at line 298 of file walreceiver.h.

◆ walrcv_startstreaming_fn

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

Definition at line 321 of file walreceiver.h.

◆ WalRcvExecResult

typedef struct WalRcvExecResult WalRcvExecResult

◆ WalReceiverConn

typedef struct WalReceiverConn WalReceiverConn

Definition at line 195 of file walreceiver.h.

◆ WalReceiverFunctionsType

typedef struct WalReceiverFunctionsType WalReceiverFunctionsType

Enumeration Type Documentation

◆ WalRcvExecStatus

enum WalRcvExecStatus
Enumerator
WALRCV_ERROR 
WALRCV_OK_COMMAND 
WALRCV_OK_TUPLES 
WALRCV_OK_COPY_IN 
WALRCV_OK_COPY_OUT 
WALRCV_OK_COPY_BOTH 

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

enum WalRcvState
Enumerator
WALRCV_STOPPED 
WALRCV_STARTING 
WALRCV_STREAMING 
WALRCV_WAITING 
WALRCV_RESTARTING 
WALRCV_STOPPING 

Definition at line 45 of file walreceiver.h.

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

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:1757
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1645
int64 TimestampTz
Definition: timestamp.h:39
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
XLogRecPtr flushedUpto
Definition: walreceiver.h:96
slock_t mutex
Definition: walreceiver.h:147
WalRcvData * WalRcv
uint64 XLogRecPtr
Definition: xlogdefs.h:21
TimestampTz GetCurrentChunkReplayStartTime(void)
Definition: xlogrecovery.c:4648
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlogrecovery.c:4561

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:111
TimestampTz lastMsgSendTime
Definition: walreceiver.h:110

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:97
XLogRecPtr latestChunkStart
Definition: walreceiver.h:105
static TimeLineID receiveTLI
Definition: xlogrecovery.c:265

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

Referenced by CreateRestartPoint(), GetLatestLSN(), GetStandbyFlushRecPtr(), pg_last_wal_receive_lsn(), reserve_wal_for_local_slot(), 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:465
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:155

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

◆ 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 ProcNumber walrcv_proc;
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(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(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;
295 }
296 else
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 walrcv_proc = walrcv->procno;
314
315 SpinLockRelease(&walrcv->mutex);
316
317 if (launch)
319 else if (walrcv_proc != INVALID_PROC_NUMBER)
320 SetLatch(&GetPGProcByNumber(walrcv_proc)->procLatch);
321}
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1609
Assert(PointerIsAligned(start, uint64))
void SetLatch(Latch *latch)
Definition: latch.c:290
#define NAMEDATALEN
int64 pg_time_t
Definition: pgtime.h:23
void SendPostmasterSignal(PMSignalReason reason)
Definition: pmsignal.c:165
@ PMSIGNAL_START_WALRECEIVER
Definition: pmsignal.h:42
size_t strlcpy(char *dst, const char *src, size_t siz)
Definition: strlcpy.c:45
#define GetPGProcByNumber(n)
Definition: proc.h:440
#define INVALID_PROC_NUMBER
Definition: procnumber.h:26
int ProcNumber
Definition: procnumber.h:24
TimeLineID receiveStartTLI
Definition: walreceiver.h:87
char slotname[NAMEDATALEN]
Definition: walreceiver.h:136
XLogRecPtr receiveStart
Definition: walreceiver.h:86
ProcNumber procno
Definition: walreceiver.h:67
bool is_temp_slot
Definition: walreceiver.h:142
pg_time_t startTime
Definition: walreceiver.h:78
WalRcvState walRcvState
Definition: walreceiver.h:71
char conninfo[MAXCONNINFO]
Definition: walreceiver.h:123
#define MAXCONNINFO
Definition: walreceiver.h:37
int wal_segment_size
Definition: xlog.c:144
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
Definition: xlog_internal.h:106

References Assert(), WalRcvData::conninfo, WalRcvData::flushedUpto, GetPGProcByNumber, INVALID_PROC_NUMBER, WalRcvData::is_temp_slot, WalRcvData::latestChunkStart, MAXCONNINFO, WalRcvData::mutex, NAMEDATALEN, now(), PMSIGNAL_START_WALRECEIVER, WalRcvData::procno, 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:
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())
228 WAIT_EVENT_WAL_RECEIVER_EXIT);
230}
bool ConditionVariableCancelSleep(void)
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
pid_t pid
Definition: walreceiver.h:68
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:72
bool WalRcvRunning(void)
#define kill(pid, sig)
Definition: win32_port.h:493

References ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), kill, WalRcvData::mutex, WalRcvData::pid, SpinLockAcquire, SpinLockRelease, 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 471 of file walreceiver.h.

472{
473 if (!walres)
474 return;
475
476 if (walres->err)
477 pfree(walres->err);
478
479 if (walres->tuplestore)
480 tuplestore_end(walres->tuplestore);
481
482 if (walres->tupledesc)
483 FreeTupleDesc(walres->tupledesc);
484
485 pfree(walres);
486}
void pfree(void *pointer)
Definition: mcxt.c:1594
Tuplestorestate * tuplestore
Definition: walreceiver.h:223
TupleDesc tupledesc
Definition: walreceiver.h:224
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:502
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:492

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

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

◆ WalRcvForceReply()

void WalRcvForceReply ( void  )

Definition at line 1350 of file walreceiver.c.

1351{
1352 ProcNumber procno;
1353
1354 WalRcv->force_reply = true;
1355 /* fetching the proc number is probably atomic, but don't rely on it */
1357 procno = WalRcv->procno;
1359 if (procno != INVALID_PROC_NUMBER)
1360 SetLatch(&GetPGProcByNumber(procno)->procLatch);
1361}
sig_atomic_t force_reply
Definition: walreceiver.h:162

References WalRcvData::force_reply, GetPGProcByNumber, INVALID_PROC_NUMBER, WalRcvData::mutex, WalRcvData::procno, 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 */
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:323
#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 */
71 }
72}
static void pg_atomic_init_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:451
#define MemSet(start, val, len)
Definition: c.h:1019
void ConditionVariableInit(ConditionVariable *cv)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:387
#define SpinLockInit(lock)
Definition: spin.h:57
Size WalRcvShmemSize(void)

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

Referenced by CreateOrAttachShmemStructs().

◆ WalRcvShmemSize()

Size WalRcvShmemSize ( void  )

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:610
Size add_size(Size s1, Size s2)
Definition: shmem.c:493

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

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

Definition at line 152 of file walreceiver.c.

153{
154 char conninfo[MAXCONNINFO];
155 char *tmp_conninfo;
156 char slotname[NAMEDATALEN];
157 bool is_temp_slot;
158 XLogRecPtr startpoint;
159 TimeLineID startpointTLI;
160 TimeLineID primaryTLI;
161 bool first_stream;
162 WalRcvData *walrcv;
164 char *err;
165 char *sender_host = NULL;
166 int sender_port = 0;
167 char *appname;
168
169 Assert(startup_data_len == 0);
170
173
174 /*
175 * WalRcv should be set up already (if we are a backend, we inherit this
176 * by fork() or EXEC_BACKEND mechanism from the postmaster).
177 */
178 walrcv = WalRcv;
179 Assert(walrcv != NULL);
180
181 /*
182 * Mark walreceiver as running in shared memory.
183 *
184 * Do this as early as possible, so that if we fail later on, we'll set
185 * state to STOPPED. If we die before this, the startup process will keep
186 * waiting for us to start up, until it times out.
187 */
188 SpinLockAcquire(&walrcv->mutex);
189 Assert(walrcv->pid == 0);
190 switch (walrcv->walRcvState)
191 {
192 case WALRCV_STOPPING:
193 /* If we've already been requested to stop, don't start up. */
194 walrcv->walRcvState = WALRCV_STOPPED;
195 /* fall through */
196
197 case WALRCV_STOPPED:
198 SpinLockRelease(&walrcv->mutex);
200 proc_exit(1);
201 break;
202
203 case WALRCV_STARTING:
204 /* The usual case */
205 break;
206
207 case WALRCV_WAITING:
208 case WALRCV_STREAMING:
210 default:
211 /* Shouldn't happen */
212 SpinLockRelease(&walrcv->mutex);
213 elog(PANIC, "walreceiver still running according to shared memory state");
214 }
215 /* Advertise our PID so that the startup process can kill us */
216 walrcv->pid = MyProcPid;
218
219 /* Fetch information required to start streaming */
220 walrcv->ready_to_display = false;
221 strlcpy(conninfo, walrcv->conninfo, MAXCONNINFO);
222 strlcpy(slotname, walrcv->slotname, NAMEDATALEN);
223 is_temp_slot = walrcv->is_temp_slot;
224 startpoint = walrcv->receiveStart;
225 startpointTLI = walrcv->receiveStartTLI;
226
227 /*
228 * At most one of is_temp_slot and slotname can be set; otherwise,
229 * RequestXLogStreaming messed up.
230 */
231 Assert(!is_temp_slot || (slotname[0] == '0円'));
232
233 /* Initialise to a sanish value */
235 walrcv->lastMsgSendTime =
236 walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
237
238 /* Report our proc number so that others can wake us up */
239 walrcv->procno = MyProcNumber;
240
241 SpinLockRelease(&walrcv->mutex);
242
244
245 /* Arrange to clean up at walreceiver exit */
246 on_shmem_exit(WalRcvDie, PointerGetDatum(&startpointTLI));
247
248 /* Properly accept or ignore signals the postmaster might send us */
249 pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
250 * file */
251 pqsignal(SIGINT, SIG_IGN);
252 pqsignal(SIGTERM, die); /* request shutdown */
253 /* SIGQUIT handler was already set up by InitPostmasterChild */
254 pqsignal(SIGALRM, SIG_IGN);
255 pqsignal(SIGPIPE, SIG_IGN);
257 pqsignal(SIGUSR2, SIG_IGN);
258
259 /* Reset some signals that are accepted by postmaster but not here */
260 pqsignal(SIGCHLD, SIG_DFL);
261
262 /* Load the libpq-specific functions */
263 load_file("libpqwalreceiver", false);
264 if (WalReceiverFunctions == NULL)
265 elog(ERROR, "libpqwalreceiver didn't initialize correctly");
266
267 /* Unblock signals (they were blocked when the postmaster forked us) */
268 sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
269
270 /* Establish the connection to the primary for XLOG streaming */
271 appname = cluster_name[0] ? cluster_name : "walreceiver";
272 wrconn = walrcv_connect(conninfo, true, false, false, appname, &err);
273 if (!wrconn)
275 (errcode(ERRCODE_CONNECTION_FAILURE),
276 errmsg("streaming replication receiver \"%s\" could not connect to the primary server: %s",
277 appname, err)));
278
279 /*
280 * Save user-visible connection string. This clobbers the original
281 * conninfo, for security. Also save host and port of the sender server
282 * this walreceiver is connected to.
283 */
284 tmp_conninfo = walrcv_get_conninfo(wrconn);
285 walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
286 SpinLockAcquire(&walrcv->mutex);
287 memset(walrcv->conninfo, 0, MAXCONNINFO);
288 if (tmp_conninfo)
289 strlcpy(walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
290
291 memset(walrcv->sender_host, 0, NI_MAXHOST);
292 if (sender_host)
293 strlcpy(walrcv->sender_host, sender_host, NI_MAXHOST);
294
295 walrcv->sender_port = sender_port;
296 walrcv->ready_to_display = true;
297 SpinLockRelease(&walrcv->mutex);
298
299 if (tmp_conninfo)
300 pfree(tmp_conninfo);
301
302 if (sender_host)
303 pfree(sender_host);
304
305 first_stream = true;
306 for (;;)
307 {
308 char *primary_sysid;
309 char standby_sysid[32];
311
312 /*
313 * Check that we're connected to a valid server using the
314 * IDENTIFY_SYSTEM replication command.
315 */
316 primary_sysid = walrcv_identify_system(wrconn, &primaryTLI);
317
318 snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
320 if (strcmp(primary_sysid, standby_sysid) != 0)
321 {
323 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
324 errmsg("database system identifier differs between the primary and standby"),
325 errdetail("The primary's identifier is %s, the standby's identifier is %s.",
326 primary_sysid, standby_sysid)));
327 }
328
329 /*
330 * Confirm that the current timeline of the primary is the same or
331 * ahead of ours.
332 */
333 if (primaryTLI < startpointTLI)
335 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
336 errmsg("highest timeline %u of the primary is behind recovery timeline %u",
337 primaryTLI, startpointTLI)));
338
339 /*
340 * Get any missing history files. We do this always, even when we're
341 * not interested in that timeline, so that if we're promoted to
342 * become the primary later on, we don't select the same timeline that
343 * was already used in the current primary. This isn't bullet-proof -
344 * you'll need some external software to manage your cluster if you
345 * need to ensure that a unique timeline id is chosen in every case,
346 * but let's avoid the confusion of timeline id collisions where we
347 * can.
348 */
349 WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
350
351 /*
352 * Create temporary replication slot if requested, and update slot
353 * name in shared memory. (Note the slot name cannot already be set
354 * in this case.)
355 */
356 if (is_temp_slot)
357 {
358 snprintf(slotname, sizeof(slotname),
359 "pg_walreceiver_%lld",
360 (long long int) walrcv_get_backend_pid(wrconn));
361
362 walrcv_create_slot(wrconn, slotname, true, false, false, 0, NULL);
363
364 SpinLockAcquire(&walrcv->mutex);
365 strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
366 SpinLockRelease(&walrcv->mutex);
367 }
368
369 /*
370 * Start streaming.
371 *
372 * We'll try to start at the requested starting point and timeline,
373 * even if it's different from the server's latest timeline. In case
374 * we've already reached the end of the old timeline, the server will
375 * finish the streaming immediately, and we will go back to await
376 * orders from the startup process. If recovery_target_timeline is
377 * 'latest', the startup process will scan pg_wal and find the new
378 * history file, bump recovery target timeline, and ask us to restart
379 * on the new timeline.
380 */
381 options.logical = false;
382 options.startpoint = startpoint;
383 options.slotname = slotname[0] != '0円' ? slotname : NULL;
384 options.proto.physical.startpointTLI = startpointTLI;
386 {
387 if (first_stream)
388 ereport(LOG,
389 errmsg("started streaming WAL from primary at %X/%08X on timeline %u",
390 LSN_FORMAT_ARGS(startpoint), startpointTLI));
391 else
392 ereport(LOG,
393 errmsg("restarted WAL streaming at %X/%08X on timeline %u",
394 LSN_FORMAT_ARGS(startpoint), startpointTLI));
395 first_stream = false;
396
397 /* Initialize LogstreamResult and buffers for processing messages */
400
401 /* Initialize nap wakeup times. */
403 for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
405
406 /* Send initial reply/feedback messages. */
407 XLogWalRcvSendReply(true, false);
409
410 /* Loop until end-of-streaming or error */
411 for (;;)
412 {
413 char *buf;
414 int len;
415 bool endofwal = false;
416 pgsocket wait_fd = PGINVALID_SOCKET;
417 int rc;
418 TimestampTz nextWakeup;
419 long nap;
420
421 /*
422 * Exit walreceiver if we're not in recovery. This should not
423 * happen, but cross-check the status here.
424 */
425 if (!RecoveryInProgress())
427 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
428 errmsg("cannot continue WAL streaming, recovery has already ended")));
429
430 /* Process any requests or signals received recently */
432
434 {
435 ConfigReloadPending = false;
437 /* recompute wakeup times */
439 for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
442 }
443
444 /* See if we can read data immediately */
445 len = walrcv_receive(wrconn, &buf, &wait_fd);
446 if (len != 0)
447 {
448 /*
449 * Process the received data, and any subsequent data we
450 * can read without blocking.
451 */
452 for (;;)
453 {
454 if (len > 0)
455 {
456 /*
457 * Something was received from primary, so adjust
458 * the ping and terminate wakeup times.
459 */
462 now);
464 XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
465 startpointTLI);
466 }
467 else if (len == 0)
468 break;
469 else if (len < 0)
470 {
471 ereport(LOG,
472 (errmsg("replication terminated by primary server"),
473 errdetail("End of WAL reached on timeline %u at %X/%08X.",
474 startpointTLI,
476 endofwal = true;
477 break;
478 }
479 len = walrcv_receive(wrconn, &buf, &wait_fd);
480 }
481
482 /* Let the primary know that we received some data. */
483 XLogWalRcvSendReply(false, false);
484
485 /*
486 * If we've written some records, flush them to disk and
487 * let the startup process and primary server know about
488 * them.
489 */
490 XLogWalRcvFlush(false, startpointTLI);
491 }
492
493 /* Check if we need to exit the streaming loop. */
494 if (endofwal)
495 break;
496
497 /* Find the soonest wakeup time, to limit our nap. */
498 nextWakeup = TIMESTAMP_INFINITY;
499 for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
500 nextWakeup = Min(wakeup[i], nextWakeup);
501
502 /* Calculate the nap time, clamping as necessary. */
504 nap = TimestampDifferenceMilliseconds(now, nextWakeup);
505
506 /*
507 * Ideally we would reuse a WaitEventSet object repeatedly
508 * here to avoid the overheads of WaitLatchOrSocket on epoll
509 * systems, but we can't be sure that libpq (or any other
510 * walreceiver implementation) has the same socket (even if
511 * the fd is the same number, it may have been closed and
512 * reopened since the last time). In future, if there is a
513 * function for removing sockets from WaitEventSet, then we
514 * could add and remove just the socket each time, potentially
515 * avoiding some system calls.
516 */
517 Assert(wait_fd != PGINVALID_SOCKET);
521 wait_fd,
522 nap,
523 WAIT_EVENT_WAL_RECEIVER_MAIN);
524 if (rc & WL_LATCH_SET)
525 {
528
529 if (walrcv->force_reply)
530 {
531 /*
532 * The recovery process has asked us to send apply
533 * feedback now. Make sure the flag is really set to
534 * false in shared memory before sending the reply, so
535 * we don't miss a new request for a reply.
536 */
537 walrcv->force_reply = false;
539 XLogWalRcvSendReply(true, false);
540 }
541 }
542 if (rc & WL_TIMEOUT)
543 {
544 /*
545 * We didn't receive anything new. If we haven't heard
546 * anything from the server for more than
547 * wal_receiver_timeout / 2, ping the server. Also, if
548 * it's been longer than wal_receiver_status_interval
549 * since the last update we sent, send a status update to
550 * the primary anyway, to report any progress in applying
551 * WAL.
552 */
553 bool requestReply = false;
554
555 /*
556 * Report pending statistics to the cumulative stats
557 * system. This location is useful for the report as it
558 * is not within a tight loop in the WAL receiver, to
559 * avoid bloating pgstats with requests, while also making
560 * sure that the reports happen each time a status update
561 * is sent.
562 */
563 pgstat_report_wal(false);
564
565 /*
566 * Check if time since last receive from primary has
567 * reached the configured limit.
568 */
572 (errcode(ERRCODE_CONNECTION_FAILURE),
573 errmsg("terminating walreceiver due to timeout")));
574
575 /*
576 * If we didn't receive anything new for half of receiver
577 * replication timeout, then ping the server.
578 */
580 {
581 requestReply = true;
583 }
584
585 XLogWalRcvSendReply(requestReply, requestReply);
587 }
588 }
589
590 /*
591 * The backend finished streaming. Exit streaming COPY-mode from
592 * our side, too.
593 */
594 walrcv_endstreaming(wrconn, &primaryTLI);
595
596 /*
597 * If the server had switched to a new timeline that we didn't
598 * know about when we began streaming, fetch its timeline history
599 * file now.
600 */
601 WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
602 }
603 else
604 ereport(LOG,
605 (errmsg("primary server contains no more WAL on requested timeline %u",
606 startpointTLI)));
607
608 /*
609 * End of WAL reached on the requested timeline. Close the last
610 * segment, and await for new orders from the startup process.
611 */
612 if (recvFile >= 0)
613 {
614 char xlogfname[MAXFNAMELEN];
615
616 XLogWalRcvFlush(false, startpointTLI);
618 if (close(recvFile) != 0)
621 errmsg("could not close WAL segment %s: %m",
622 xlogfname)));
623
624 /*
625 * Create .done file forcibly to prevent the streamed segment from
626 * being archived later.
627 */
629 XLogArchiveForceDone(xlogfname);
630 else
631 XLogArchiveNotify(xlogfname);
632 }
633 recvFile = -1;
634
635 elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
636 WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
637 }
638 /* not reached */
639}
static void pg_atomic_write_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:483
#define pg_memory_barrier()
Definition: atomics.h:141
void AuxiliaryProcessMainCommon(void)
Definition: auxprocess.c:39
sigset_t UnBlockSig
Definition: pqsignal.c:22
#define Min(x, y)
Definition: c.h:1003
#define UINT64_FORMAT
Definition: c.h:557
#define TIMESTAMP_INFINITY
Definition: timestamp.h:151
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:149
int errcode_for_file_access(void)
Definition: elog.c:877
int errdetail(const char *fmt,...)
Definition: elog.c:1207
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define LOG
Definition: elog.h:31
#define FATAL
Definition: elog.h:41
#define PANIC
Definition: elog.h:42
#define DEBUG1
Definition: elog.h:30
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:226
#define ereport(elevel,...)
Definition: elog.h:150
void err(int eval, const char *fmt,...)
Definition: err.c:43
int MyProcPid
Definition: globals.c:47
ProcNumber MyProcNumber
Definition: globals.c:90
struct Latch * MyLatch
Definition: globals.c:63
void ProcessConfigFile(GucContext context)
Definition: guc-file.l:120
@ PGC_SIGHUP
Definition: guc.h:75
char * cluster_name
Definition: guc_tables.c:555
#define close(a)
Definition: win32.h:12
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:365
void proc_exit(int code)
Definition: ipc.c:104
i
int i
Definition: isn.c:77
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:223
void ResetLatch(Latch *latch)
Definition: latch.c:374
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
@ B_WAL_RECEIVER
Definition: miscadmin.h:365
BackendType MyBackendType
Definition: miscinit.c:64
const void size_t len
static char ** options
Definition: pg_recvlogical.c:59
#define die(msg)
Definition: pg_test_fsync.c:100
static char * buf
Definition: pg_test_fsync.c:72
void pgstat_report_wal(bool force)
Definition: pgstat_wal.c:46
#define pqsignal
Definition: port.h:531
int pgsocket
Definition: port.h:29
#define snprintf
Definition: port.h:239
#define PGINVALID_SOCKET
Definition: port.h:31
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:332
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:674
void initStringInfo(StringInfo str)
Definition: stringinfo.c:97
char sender_host[NI_MAXHOST]
Definition: walreceiver.h:129
TimestampTz latestWalEndTime
Definition: walreceiver.h:117
bool ready_to_display
Definition: walreceiver.h:145
int sender_port
Definition: walreceiver.h:130
Definition: oid2name.c:30
#define WL_SOCKET_READABLE
Definition: waiteventset.h:35
#define WL_TIMEOUT
Definition: waiteventset.h:37
#define WL_EXIT_ON_PM_DEATH
Definition: waiteventset.h:39
#define WL_LATCH_SET
Definition: waiteventset.h:34
#define NUM_WALRCV_WAKEUPS
Definition: walreceiver.c:124
static WalReceiverConn * wrconn
Definition: walreceiver.c:93
static TimestampTz wakeup[NUM_WALRCV_WAKEUPS]
Definition: walreceiver.c:130
static StringInfoData reply_message
Definition: walreceiver.c:132
static int recvFile
Definition: walreceiver.c:101
static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
Definition: walreceiver.c:725
static void XLogWalRcvFlush(bool dying, TimeLineID tli)
Definition: walreceiver.c:985
static TimeLineID recvFileTLI
Definition: walreceiver.c:102
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:94
static XLogSegNo recvSegNo
Definition: walreceiver.c:103
static void XLogWalRcvSendHSFeedback(bool immed)
Definition: walreceiver.c:1161
@ WALRCV_WAKEUP_TERMINATE
Definition: walreceiver.c:120
@ WALRCV_WAKEUP_PING
Definition: walreceiver.c:121
static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
Definition: walreceiver.c:645
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
Definition: walreceiver.c:819
static void WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now)
Definition: walreceiver.c:1309
static void WalRcvDie(int code, Datum arg)
Definition: walreceiver.c:781
static void XLogWalRcvSendReply(bool force, bool requestReply)
Definition: walreceiver.c:1092
static struct @19 LogstreamResult
#define walrcv_startstreaming(conn, options)
Definition: walreceiver.h:451
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
Definition: walreceiver.h:435
#define walrcv_get_senderinfo(conn, sender_host, sender_port)
Definition: walreceiver.h:441
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
Definition: walreceiver.h:459
#define walrcv_get_conninfo(conn)
Definition: walreceiver.h:439
#define walrcv_endstreaming(conn, next_tli)
Definition: walreceiver.h:453
#define walrcv_identify_system(conn, primary_tli)
Definition: walreceiver.h:443
#define walrcv_get_backend_pid(conn)
Definition: walreceiver.h:463
#define walrcv_receive(conn, buffer, wait_fd)
Definition: walreceiver.h:455
#define SIGCHLD
Definition: win32_port.h:168
#define SIGHUP
Definition: win32_port.h:158
#define SIGPIPE
Definition: win32_port.h:163
#define SIGUSR1
Definition: win32_port.h:170
#define SIGALRM
Definition: win32_port.h:164
#define SIGUSR2
Definition: win32_port.h:171
uint64 GetSystemIdentifier(void)
Definition: xlog.c:4595
bool RecoveryInProgress(void)
Definition: xlog.c:6386
int XLogArchiveMode
Definition: xlog.c:120
@ ARCHIVE_MODE_ALWAYS
Definition: xlog.h:67
#define MAXFNAMELEN
Definition: xlog_internal.h:156
static void XLogFileName(char *fname, TimeLineID tli, XLogSegNo logSegNo, int wal_segsz_bytes)
Definition: xlog_internal.h:166
void XLogArchiveForceDone(const char *xlog)
Definition: xlogarchive.c:510
void XLogArchiveNotify(const char *xlog)
Definition: xlogarchive.c:444
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:46
uint32 TimeLineID
Definition: xlogdefs.h:62

References ARCHIVE_MODE_ALWAYS, Assert(), AuxiliaryProcessMainCommon(), B_WAL_RECEIVER, buf, CHECK_FOR_INTERRUPTS, close, cluster_name, ConditionVariableBroadcast(), ConfigReloadPending, WalRcvData::conninfo, DEBUG1, die, elog, ereport, err(), errcode(), errcode_for_file_access(), errdetail(), errmsg(), ERROR, FATAL, WalRcvData::force_reply, GetCurrentTimestamp(), GetSystemIdentifier(), GetXLogReplayRecPtr(), i, initStringInfo(), WalRcvData::is_temp_slot, WalRcvData::lastMsgReceiptTime, WalRcvData::lastMsgSendTime, WalRcvData::latestWalEndTime, len, load_file(), LOG, LogstreamResult, LSN_FORMAT_ARGS, MAXCONNINFO, MAXFNAMELEN, Min, WalRcvData::mutex, MyBackendType, MyLatch, MyProcNumber, MyProcPid, NAMEDATALEN, now(), NUM_WALRCV_WAKEUPS, on_shmem_exit(), options, PANIC, pfree(), pg_atomic_write_u64(), pg_memory_barrier, PGC_SIGHUP, PGINVALID_SOCKET, pgstat_report_wal(), WalRcvData::pid, PointerGetDatum(), pqsignal, proc_exit(), ProcessConfigFile(), WalRcvData::procno, procsignal_sigusr1_handler(), WalRcvData::ready_to_display, WalRcvData::receiveStart, WalRcvData::receiveStartTLI, RecoveryInProgress(), recvFile, recvFileTLI, recvSegNo, reply_message, ResetLatch(), WalRcvData::sender_host, WalRcvData::sender_port, SIGALRM, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SIGPIPE, SIGUSR1, SIGUSR2, WalRcvData::slotname, snprintf, SpinLockAcquire, SpinLockRelease, strlcpy(), TIMESTAMP_INFINITY, TimestampDifferenceMilliseconds(), UINT64_FORMAT, UnBlockSig, 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().

Variable Documentation

◆ hot_standby_feedback

PGDLLIMPORT bool hot_standby_feedback
extern

◆ wal_receiver_status_interval

PGDLLIMPORT int wal_receiver_status_interval
extern

◆ wal_receiver_timeout

PGDLLIMPORT int wal_receiver_timeout
extern

◆ WalRcv

AltStyle によって変換されたページ (->オリジナル) /