PostgreSQL Source Code  git master
walreceiver.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * walreceiver.c
4  *
5  * The WAL receiver process (walreceiver) is new as of Postgres 9.0. It
6  * is the process in the standby server that takes charge of receiving
7  * XLOG records from a primary server during streaming replication.
8  *
9  * When the startup process determines that it's time to start streaming,
10  * it instructs postmaster to start walreceiver. Walreceiver first connects
11  * to the primary server (it will be served by a walsender process
12  * in the primary server), and then keeps receiving XLOG records and
13  * writing them to the disk as long as the connection is alive. As XLOG
14  * records are received and flushed to disk, it updates the
15  * WalRcv->flushedUpto variable in shared memory, to inform the startup
16  * process of how far it can proceed with XLOG replay.
17  *
18  * A WAL receiver cannot directly load GUC parameters used when establishing
19  * its connection to the primary. Instead it relies on parameter values
20  * that are passed down by the startup process when streaming is requested.
21  * This applies, for example, to the replication slot and the connection
22  * string to be used for the connection with the primary.
23  *
24  * If the primary server ends streaming, but doesn't disconnect, walreceiver
25  * goes into "waiting" mode, and waits for the startup process to give new
26  * instructions. The startup process will treat that the same as
27  * disconnection, and will rescan the archive/pg_wal directory. But when the
28  * startup process wants to try streaming replication again, it will just
29  * nudge the existing walreceiver process that's waiting, instead of launching
30  * a new one.
31  *
32  * Normal termination is by SIGTERM, which instructs the walreceiver to
33  * exit(0). Emergency termination is by SIGQUIT; like any postmaster child
34  * process, the walreceiver will simply abort and exit on SIGQUIT. A close
35  * of the connection and a FATAL error are treated not as a crash but as
36  * normal operation.
37  *
38  * This file contains the server-facing parts of walreceiver. The libpq-
39  * specific parts are in the libpqwalreceiver module. It's loaded
40  * dynamically to avoid linking the server with libpq.
41  *
42  * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
43  *
44  *
45  * IDENTIFICATION
46  * src/backend/replication/walreceiver.c
47  *
48  *-------------------------------------------------------------------------
49  */
50 #include "postgres.h"
51 
52 #include <unistd.h>
53 
54 #include "access/htup_details.h"
55 #include "access/timeline.h"
56 #include "access/transam.h"
57 #include "access/xlog_internal.h"
58 #include "access/xlogarchive.h"
59 #include "catalog/pg_authid.h"
60 #include "catalog/pg_type.h"
61 #include "common/ip.h"
62 #include "funcapi.h"
63 #include "libpq/pqformat.h"
64 #include "libpq/pqsignal.h"
65 #include "miscadmin.h"
66 #include "pgstat.h"
67 #include "postmaster/interrupt.h"
69 #include "replication/walsender.h"
70 #include "storage/ipc.h"
71 #include "storage/pmsignal.h"
72 #include "storage/proc.h"
73 #include "storage/procarray.h"
74 #include "storage/procsignal.h"
75 #include "utils/acl.h"
76 #include "utils/builtins.h"
77 #include "utils/guc.h"
78 #include "utils/pg_lsn.h"
79 #include "utils/ps_status.h"
80 #include "utils/resowner.h"
81 #include "utils/timestamp.h"
82 
83 
84 /*
85  * GUC variables. (Other variables that affect walreceiver are in xlog.c
86  * because they're passed down from the startup process, for better
87  * synchronization.)
88  */
92 
93 /* libpqwalreceiver connection */
94 static WalReceiverConn *wrconn = NULL;
96 
97 #define NAPTIME_PER_CYCLE 100 /* max sleep time between cycles (100ms) */
98 
99 /*
100  * These variables are used similarly to openLogFile/SegNo,
101  * but for walreceiver to write the XLOG. recvFileTLI is the TimeLineID
102  * corresponding the filename of recvFile.
103  */
104 static int recvFile = -1;
106 static XLogSegNo recvSegNo = 0;
107 
108 /*
109  * LogstreamResult indicates the byte positions that we have already
110  * written/fsynced.
111  */
112 static struct
113 {
114  XLogRecPtr Write; /* last byte + 1 written out in the standby */
115  XLogRecPtr Flush; /* last byte + 1 flushed in the standby */
117 
120 
121 /* Prototypes for private functions */
122 static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last);
123 static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI);
124 static void WalRcvDie(int code, Datum arg);
125 static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len,
126  TimeLineID tli);
127 static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr,
128  TimeLineID tli);
129 static void XLogWalRcvFlush(bool dying, TimeLineID tli);
130 static void XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli);
131 static void XLogWalRcvSendReply(bool force, bool requestReply);
132 static void XLogWalRcvSendHSFeedback(bool immed);
133 static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
134 
135 /*
136  * Process any interrupts the walreceiver process may have received.
137  * This should be called any time the process's latch has become set.
138  *
139  * Currently, only SIGTERM is of interest. We can't just exit(1) within the
140  * SIGTERM signal handler, because the signal might arrive in the middle of
141  * some critical operation, like while we're holding a spinlock. Instead, the
142  * signal handler sets a flag variable as well as setting the process's latch.
143  * We must check the flag (by calling ProcessWalRcvInterrupts) anytime the
144  * latch has become set. Operations that could block for a long time, such as
145  * reading from a remote server, must pay attention to the latch too; see
146  * libpqrcv_PQgetResult for example.
147  */
148 void
150 {
151  /*
152  * Although walreceiver interrupt handling doesn't use the same scheme as
153  * regular backends, call CHECK_FOR_INTERRUPTS() to make sure we receive
154  * any incoming signals on Win32, and also to make sure we process any
155  * barrier events.
156  */
158 
160  {
161  ereport(FATAL,
162  (errcode(ERRCODE_ADMIN_SHUTDOWN),
163  errmsg("terminating walreceiver process due to administrator command")));
164  }
165 }
166 
167 
168 /* Main entry point for walreceiver process */
169 void
171 {
172  char conninfo[MAXCONNINFO];
173  char *tmp_conninfo;
174  char slotname[NAMEDATALEN];
175  bool is_temp_slot;
176  XLogRecPtr startpoint;
177  TimeLineID startpointTLI;
178  TimeLineID primaryTLI;
179  bool first_stream;
180  WalRcvData *walrcv = WalRcv;
181  TimestampTz last_recv_timestamp;
183  bool ping_sent;
184  char *err;
185  char *sender_host = NULL;
186  int sender_port = 0;
187 
188  /*
189  * WalRcv should be set up already (if we are a backend, we inherit this
190  * by fork() or EXEC_BACKEND mechanism from the postmaster).
191  */
192  Assert(walrcv != NULL);
193 
195 
196  /*
197  * Mark walreceiver as running in shared memory.
198  *
199  * Do this as early as possible, so that if we fail later on, we'll set
200  * state to STOPPED. If we die before this, the startup process will keep
201  * waiting for us to start up, until it times out.
202  */
203  SpinLockAcquire(&walrcv->mutex);
204  Assert(walrcv->pid == 0);
205  switch (walrcv->walRcvState)
206  {
207  case WALRCV_STOPPING:
208  /* If we've already been requested to stop, don't start up. */
209  walrcv->walRcvState = WALRCV_STOPPED;
210  /* fall through */
211 
212  case WALRCV_STOPPED:
213  SpinLockRelease(&walrcv->mutex);
215  proc_exit(1);
216  break;
217 
218  case WALRCV_STARTING:
219  /* The usual case */
220  break;
221 
222  case WALRCV_WAITING:
223  case WALRCV_STREAMING:
224  case WALRCV_RESTARTING:
225  default:
226  /* Shouldn't happen */
227  SpinLockRelease(&walrcv->mutex);
228  elog(PANIC, "walreceiver still running according to shared memory state");
229  }
230  /* Advertise our PID so that the startup process can kill us */
231  walrcv->pid = MyProcPid;
232  walrcv->walRcvState = WALRCV_STREAMING;
233 
234  /* Fetch information required to start streaming */
235  walrcv->ready_to_display = false;
236  strlcpy(conninfo, (char *) walrcv->conninfo, MAXCONNINFO);
237  strlcpy(slotname, (char *) walrcv->slotname, NAMEDATALEN);
238  is_temp_slot = walrcv->is_temp_slot;
239  startpoint = walrcv->receiveStart;
240  startpointTLI = walrcv->receiveStartTLI;
241 
242  /*
243  * At most one of is_temp_slot and slotname can be set; otherwise,
244  * RequestXLogStreaming messed up.
245  */
246  Assert(!is_temp_slot || (slotname[0] == '\0'));
247 
248  /* Initialise to a sanish value */
249  walrcv->lastMsgSendTime =
250  walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
251 
252  /* Report the latch to use to awaken this process */
253  walrcv->latch = &MyProc->procLatch;
254 
255  SpinLockRelease(&walrcv->mutex);
256 
258 
259  /* Arrange to clean up at walreceiver exit */
260  on_shmem_exit(WalRcvDie, PointerGetDatum(&startpointTLI));
261 
262  /* Properly accept or ignore signals the postmaster might send us */
263  pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
264  * file */
265  pqsignal(SIGINT, SIG_IGN);
266  pqsignal(SIGTERM, SignalHandlerForShutdownRequest); /* request shutdown */
267  /* SIGQUIT handler was already set up by InitPostmasterChild */
272 
273  /* Reset some signals that are accepted by postmaster but not here */
275 
276  /* Load the libpq-specific functions */
277  load_file("libpqwalreceiver", false);
278  if (WalReceiverFunctions == NULL)
279  elog(ERROR, "libpqwalreceiver didn't initialize correctly");
280 
281  /* Unblock signals (they were blocked when the postmaster forked us) */
283 
284  /* Establish the connection to the primary for XLOG streaming */
285  wrconn = walrcv_connect(conninfo, false,
286  cluster_name[0] ? cluster_name : "walreceiver",
287  &err);
288  if (!wrconn)
289  ereport(ERROR,
290  (errcode(ERRCODE_CONNECTION_FAILURE),
291  errmsg("could not connect to the primary server: %s", err)));
292 
293  /*
294  * Save user-visible connection string. This clobbers the original
295  * conninfo, for security. Also save host and port of the sender server
296  * this walreceiver is connected to.
297  */
298  tmp_conninfo = walrcv_get_conninfo(wrconn);
299  walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
300  SpinLockAcquire(&walrcv->mutex);
301  memset(walrcv->conninfo, 0, MAXCONNINFO);
302  if (tmp_conninfo)
303  strlcpy((char *) walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
304 
305  memset(walrcv->sender_host, 0, NI_MAXHOST);
306  if (sender_host)
307  strlcpy((char *) walrcv->sender_host, sender_host, NI_MAXHOST);
308 
309  walrcv->sender_port = sender_port;
310  walrcv->ready_to_display = true;
311  SpinLockRelease(&walrcv->mutex);
312 
313  if (tmp_conninfo)
314  pfree(tmp_conninfo);
315 
316  if (sender_host)
317  pfree(sender_host);
318 
319  first_stream = true;
320  for (;;)
321  {
322  char *primary_sysid;
323  char standby_sysid[32];
325 
326  /*
327  * Check that we're connected to a valid server using the
328  * IDENTIFY_SYSTEM replication command.
329  */
330  primary_sysid = walrcv_identify_system(wrconn, &primaryTLI);
331 
332  snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
334  if (strcmp(primary_sysid, standby_sysid) != 0)
335  {
336  ereport(ERROR,
337  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
338  errmsg("database system identifier differs between the primary and standby"),
339  errdetail("The primary's identifier is %s, the standby's identifier is %s.",
340  primary_sysid, standby_sysid)));
341  }
342 
343  /*
344  * Confirm that the current timeline of the primary is the same or
345  * ahead of ours.
346  */
347  if (primaryTLI < startpointTLI)
348  ereport(ERROR,
349  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
350  errmsg("highest timeline %u of the primary is behind recovery timeline %u",
351  primaryTLI, startpointTLI)));
352 
353  /*
354  * Get any missing history files. We do this always, even when we're
355  * not interested in that timeline, so that if we're promoted to
356  * become the primary later on, we don't select the same timeline that
357  * was already used in the current primary. This isn't bullet-proof -
358  * you'll need some external software to manage your cluster if you
359  * need to ensure that a unique timeline id is chosen in every case,
360  * but let's avoid the confusion of timeline id collisions where we
361  * can.
362  */
363  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
364 
365  /*
366  * Create temporary replication slot if requested, and update slot
367  * name in shared memory. (Note the slot name cannot already be set
368  * in this case.)
369  */
370  if (is_temp_slot)
371  {
372  snprintf(slotname, sizeof(slotname),
373  "pg_walreceiver_%lld",
374  (long long int) walrcv_get_backend_pid(wrconn));
375 
376  walrcv_create_slot(wrconn, slotname, true, false, 0, NULL);
377 
378  SpinLockAcquire(&walrcv->mutex);
379  strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
380  SpinLockRelease(&walrcv->mutex);
381  }
382 
383  /*
384  * Start streaming.
385  *
386  * We'll try to start at the requested starting point and timeline,
387  * even if it's different from the server's latest timeline. In case
388  * we've already reached the end of the old timeline, the server will
389  * finish the streaming immediately, and we will go back to await
390  * orders from the startup process. If recovery_target_timeline is
391  * 'latest', the startup process will scan pg_wal and find the new
392  * history file, bump recovery target timeline, and ask us to restart
393  * on the new timeline.
394  */
395  options.logical = false;
396  options.startpoint = startpoint;
397  options.slotname = slotname[0] != '\0' ? slotname : NULL;
398  options.proto.physical.startpointTLI = startpointTLI;
400  {
401  if (first_stream)
402  ereport(LOG,
403  (errmsg("started streaming WAL from primary at %X/%X on timeline %u",
404  LSN_FORMAT_ARGS(startpoint), startpointTLI)));
405  else
406  ereport(LOG,
407  (errmsg("restarted WAL streaming at %X/%X on timeline %u",
408  LSN_FORMAT_ARGS(startpoint), startpointTLI)));
409  first_stream = false;
410 
411  /* Initialize LogstreamResult and buffers for processing messages */
415 
416  /* Initialize the last recv timestamp */
417  last_recv_timestamp = GetCurrentTimestamp();
418  ping_sent = false;
419 
420  /* Loop until end-of-streaming or error */
421  for (;;)
422  {
423  char *buf;
424  int len;
425  bool endofwal = false;
426  pgsocket wait_fd = PGINVALID_SOCKET;
427  int rc;
428 
429  /*
430  * Exit walreceiver if we're not in recovery. This should not
431  * happen, but cross-check the status here.
432  */
433  if (!RecoveryInProgress())
434  ereport(FATAL,
435  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
436  errmsg("cannot continue WAL streaming, recovery has already ended")));
437 
438  /* Process any requests or signals received recently */
440 
442  {
443  ConfigReloadPending = false;
446  }
447 
448  /* See if we can read data immediately */
449  len = walrcv_receive(wrconn, &buf, &wait_fd);
450  if (len != 0)
451  {
452  /*
453  * Process the received data, and any subsequent data we
454  * can read without blocking.
455  */
456  for (;;)
457  {
458  if (len > 0)
459  {
460  /*
461  * Something was received from primary, so reset
462  * timeout
463  */
464  last_recv_timestamp = GetCurrentTimestamp();
465  ping_sent = false;
466  XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
467  startpointTLI);
468  }
469  else if (len == 0)
470  break;
471  else if (len < 0)
472  {
473  ereport(LOG,
474  (errmsg("replication terminated by primary server"),
475  errdetail("End of WAL reached on timeline %u at %X/%X.",
476  startpointTLI,
478  endofwal = true;
479  break;
480  }
481  len = walrcv_receive(wrconn, &buf, &wait_fd);
482  }
483 
484  /* Let the primary know that we received some data. */
485  XLogWalRcvSendReply(false, false);
486 
487  /*
488  * If we've written some records, flush them to disk and
489  * let the startup process and primary server know about
490  * them.
491  */
492  XLogWalRcvFlush(false, startpointTLI);
493  }
494 
495  /* Check if we need to exit the streaming loop. */
496  if (endofwal)
497  break;
498 
499  /*
500  * Ideally we would reuse a WaitEventSet object repeatedly
501  * here to avoid the overheads of WaitLatchOrSocket on epoll
502  * systems, but we can't be sure that libpq (or any other
503  * walreceiver implementation) has the same socket (even if
504  * the fd is the same number, it may have been closed and
505  * reopened since the last time). In future, if there is a
506  * function for removing sockets from WaitEventSet, then we
507  * could add and remove just the socket each time, potentially
508  * avoiding some system calls.
509  */
510  Assert(wait_fd != PGINVALID_SOCKET);
514  wait_fd,
517  if (rc & WL_LATCH_SET)
518  {
521 
522  if (walrcv->force_reply)
523  {
524  /*
525  * The recovery process has asked us to send apply
526  * feedback now. Make sure the flag is really set to
527  * false in shared memory before sending the reply, so
528  * we don't miss a new request for a reply.
529  */
530  walrcv->force_reply = false;
532  XLogWalRcvSendReply(true, false);
533  }
534  }
535  if (rc & WL_TIMEOUT)
536  {
537  /*
538  * We didn't receive anything new. If we haven't heard
539  * anything from the server for more than
540  * wal_receiver_timeout / 2, ping the server. Also, if
541  * it's been longer than wal_receiver_status_interval
542  * since the last update we sent, send a status update to
543  * the primary anyway, to report any progress in applying
544  * WAL.
545  */
546  bool requestReply = false;
547 
548  /*
549  * Check if time since last receive from primary has
550  * reached the configured limit.
551  */
552  if (wal_receiver_timeout > 0)
553  {
555  TimestampTz timeout;
556 
557  timeout =
558  TimestampTzPlusMilliseconds(last_recv_timestamp,
560 
561  if (now >= timeout)
562  ereport(ERROR,
563  (errcode(ERRCODE_CONNECTION_FAILURE),
564  errmsg("terminating walreceiver due to timeout")));
565 
566  /*
567  * We didn't receive anything new, for half of
568  * receiver replication timeout. Ping the server.
569  */
570  if (!ping_sent)
571  {
572  timeout = TimestampTzPlusMilliseconds(last_recv_timestamp,
573  (wal_receiver_timeout / 2));
574  if (now >= timeout)
575  {
576  requestReply = true;
577  ping_sent = true;
578  }
579  }
580  }
581 
582  XLogWalRcvSendReply(requestReply, requestReply);
584  }
585  }
586 
587  /*
588  * The backend finished streaming. Exit streaming COPY-mode from
589  * our side, too.
590  */
591  walrcv_endstreaming(wrconn, &primaryTLI);
592 
593  /*
594  * If the server had switched to a new timeline that we didn't
595  * know about when we began streaming, fetch its timeline history
596  * file now.
597  */
598  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
599  }
600  else
601  ereport(LOG,
602  (errmsg("primary server contains no more WAL on requested timeline %u",
603  startpointTLI)));
604 
605  /*
606  * End of WAL reached on the requested timeline. Close the last
607  * segment, and await for new orders from the startup process.
608  */
609  if (recvFile >= 0)
610  {
611  char xlogfname[MAXFNAMELEN];
612 
613  XLogWalRcvFlush(false, startpointTLI);
615  if (close(recvFile) != 0)
616  ereport(PANIC,
618  errmsg("could not close log segment %s: %m",
619  xlogfname)));
620 
621  /*
622  * Create .done file forcibly to prevent the streamed segment from
623  * being archived later.
624  */
626  XLogArchiveForceDone(xlogfname);
627  else
628  XLogArchiveNotify(xlogfname);
629  }
630  recvFile = -1;
631 
632  elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
633  WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
634  }
635  /* not reached */
636 }
637 
638 /*
639  * Wait for startup process to set receiveStart and receiveStartTLI.
640  */
641 static void
643 {
644  WalRcvData *walrcv = WalRcv;
645  int state;
646 
647  SpinLockAcquire(&walrcv->mutex);
648  state = walrcv->walRcvState;
649  if (state != WALRCV_STREAMING)
650  {
651  SpinLockRelease(&walrcv->mutex);
652  if (state == WALRCV_STOPPING)
653  proc_exit(0);
654  else
655  elog(FATAL, "unexpected walreceiver state");
656  }
657  walrcv->walRcvState = WALRCV_WAITING;
659  walrcv->receiveStartTLI = 0;
660  SpinLockRelease(&walrcv->mutex);
661 
662  set_ps_display("idle");
663 
664  /*
665  * nudge startup process to notice that we've stopped streaming and are
666  * now waiting for instructions.
667  */
668  WakeupRecovery();
669  for (;;)
670  {
672 
674 
675  SpinLockAcquire(&walrcv->mutex);
676  Assert(walrcv->walRcvState == WALRCV_RESTARTING ||
677  walrcv->walRcvState == WALRCV_WAITING ||
678  walrcv->walRcvState == WALRCV_STOPPING);
679  if (walrcv->walRcvState == WALRCV_RESTARTING)
680  {
681  /*
682  * No need to handle changes in primary_conninfo or
683  * primary_slotname here. Startup process will signal us to
684  * terminate in case those change.
685  */
686  *startpoint = walrcv->receiveStart;
687  *startpointTLI = walrcv->receiveStartTLI;
688  walrcv->walRcvState = WALRCV_STREAMING;
689  SpinLockRelease(&walrcv->mutex);
690  break;
691  }
692  if (walrcv->walRcvState == WALRCV_STOPPING)
693  {
694  /*
695  * We should've received SIGTERM if the startup process wants us
696  * to die, but might as well check it here too.
697  */
698  SpinLockRelease(&walrcv->mutex);
699  exit(1);
700  }
701  SpinLockRelease(&walrcv->mutex);
702 
705  }
706 
708  {
709  char activitymsg[50];
710 
711  snprintf(activitymsg, sizeof(activitymsg), "restarting at %X/%X",
712  LSN_FORMAT_ARGS(*startpoint));
713  set_ps_display(activitymsg);
714  }
715 }
716 
717 /*
718  * Fetch any missing timeline history files between 'first' and 'last'
719  * (inclusive) from the server.
720  */
721 static void
723 {
724  TimeLineID tli;
725 
726  for (tli = first; tli <= last; tli++)
727  {
728  /* there's no history file for timeline 1 */
729  if (tli != 1 && !existsTimeLineHistory(tli))
730  {
731  char *fname;
732  char *content;
733  int len;
734  char expectedfname[MAXFNAMELEN];
735 
736  ereport(LOG,
737  (errmsg("fetching timeline history file for timeline %u from primary server",
738  tli)));
739 
740  walrcv_readtimelinehistoryfile(wrconn, tli, &fname, &content, &len);
741 
742  /*
743  * Check that the filename on the primary matches what we
744  * calculated ourselves. This is just a sanity check, it should
745  * always match.
746  */
747  TLHistoryFileName(expectedfname, tli);
748  if (strcmp(fname, expectedfname) != 0)
749  ereport(ERROR,
750  (errcode(ERRCODE_PROTOCOL_VIOLATION),
751  errmsg_internal("primary reported unexpected file name for timeline history file of timeline %u",
752  tli)));
753 
754  /*
755  * Write the file to pg_wal.
756  */
757  writeTimeLineHistoryFile(tli, content, len);
758 
759  /*
760  * Mark the streamed history file as ready for archiving if
761  * archive_mode is always.
762  */
764  XLogArchiveForceDone(fname);
765  else
766  XLogArchiveNotify(fname);
767 
768  pfree(fname);
769  pfree(content);
770  }
771  }
772 }
773 
774 /*
775  * Mark us as STOPPED in shared memory at exit.
776  */
777 static void
778 WalRcvDie(int code, Datum arg)
779 {
780  WalRcvData *walrcv = WalRcv;
781  TimeLineID *startpointTLI_p = (TimeLineID *) DatumGetPointer(arg);
782 
783  Assert(*startpointTLI_p != 0);
784 
785  /* Ensure that all WAL records received are flushed to disk */
786  XLogWalRcvFlush(true, *startpointTLI_p);
787 
788  /* Mark ourselves inactive in shared memory */
789  SpinLockAcquire(&walrcv->mutex);
790  Assert(walrcv->walRcvState == WALRCV_STREAMING ||
791  walrcv->walRcvState == WALRCV_RESTARTING ||
792  walrcv->walRcvState == WALRCV_STARTING ||
793  walrcv->walRcvState == WALRCV_WAITING ||
794  walrcv->walRcvState == WALRCV_STOPPING);
795  Assert(walrcv->pid == MyProcPid);
796  walrcv->walRcvState = WALRCV_STOPPED;
797  walrcv->pid = 0;
798  walrcv->ready_to_display = false;
799  walrcv->latch = NULL;
800  SpinLockRelease(&walrcv->mutex);
801 
803 
804  /* Terminate the connection gracefully. */
805  if (wrconn != NULL)
807 
808  /* Wake up the startup process to notice promptly that we're gone */
809  WakeupRecovery();
810 }
811 
812 /*
813  * Accept the message from XLOG stream, and process it.
814  */
815 static void
816 XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
817 {
818  int hdrlen;
819  XLogRecPtr dataStart;
820  XLogRecPtr walEnd;
821  TimestampTz sendTime;
822  bool replyRequested;
823 
825 
826  switch (type)
827  {
828  case 'w': /* WAL records */
829  {
830  /* copy message to StringInfo */
831  hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
832  if (len < hdrlen)
833  ereport(ERROR,
834  (errcode(ERRCODE_PROTOCOL_VIOLATION),
835  errmsg_internal("invalid WAL message received from primary")));
837 
838  /* read the fields */
839  dataStart = pq_getmsgint64(&incoming_message);
840  walEnd = pq_getmsgint64(&incoming_message);
841  sendTime = pq_getmsgint64(&incoming_message);
842  ProcessWalSndrMessage(walEnd, sendTime);
843 
844  buf += hdrlen;
845  len -= hdrlen;
846  XLogWalRcvWrite(buf, len, dataStart, tli);
847  break;
848  }
849  case 'k': /* Keepalive */
850  {
851  /* copy message to StringInfo */
852  hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
853  if (len != hdrlen)
854  ereport(ERROR,
855  (errcode(ERRCODE_PROTOCOL_VIOLATION),
856  errmsg_internal("invalid keepalive message received from primary")));
858 
859  /* read the fields */
860  walEnd = pq_getmsgint64(&incoming_message);
861  sendTime = pq_getmsgint64(&incoming_message);
862  replyRequested = pq_getmsgbyte(&incoming_message);
863 
864  ProcessWalSndrMessage(walEnd, sendTime);
865 
866  /* If the primary requested a reply, send one immediately */
867  if (replyRequested)
868  XLogWalRcvSendReply(true, false);
869  break;
870  }
871  default:
872  ereport(ERROR,
873  (errcode(ERRCODE_PROTOCOL_VIOLATION),
874  errmsg_internal("invalid replication message type %d",
875  type)));
876  }
877 }
878 
879 /*
880  * Write XLOG data to disk.
881  */
882 static void
883 XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli)
884 {
885  int startoff;
886  int byteswritten;
887 
888  Assert(tli != 0);
889 
890  while (nbytes > 0)
891  {
892  int segbytes;
893 
894  /* Close the current segment if it's completed */
895  if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
896  XLogWalRcvClose(recptr, tli);
897 
898  if (recvFile < 0)
899  {
900  /* Create/use new log file */
903  recvFileTLI = tli;
904  }
905 
906  /* Calculate the start offset of the received logs */
907  startoff = XLogSegmentOffset(recptr, wal_segment_size);
908 
909  if (startoff + nbytes > wal_segment_size)
910  segbytes = wal_segment_size - startoff;
911  else
912  segbytes = nbytes;
913 
914  /* OK to write the logs */
915  errno = 0;
916 
917  byteswritten = pg_pwrite(recvFile, buf, segbytes, (off_t) startoff);
918  if (byteswritten <= 0)
919  {
920  char xlogfname[MAXFNAMELEN];
921  int save_errno;
922 
923  /* if write didn't set errno, assume no disk space */
924  if (errno == 0)
925  errno = ENOSPC;
926 
927  save_errno = errno;
929  errno = save_errno;
930  ereport(PANIC,
932  errmsg("could not write to log segment %s "
933  "at offset %u, length %lu: %m",
934  xlogfname, startoff, (unsigned long) segbytes)));
935  }
936 
937  /* Update state for write */
938  recptr += byteswritten;
939 
940  nbytes -= byteswritten;
941  buf += byteswritten;
942 
943  LogstreamResult.Write = recptr;
944  }
945 
946  /* Update shared-memory status */
948 
949  /*
950  * Close the current segment if it's fully written up in the last cycle of
951  * the loop, to create its archive notification file soon. Otherwise WAL
952  * archiving of the segment will be delayed until any data in the next
953  * segment is received and written.
954  */
955  if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
956  XLogWalRcvClose(recptr, tli);
957 }
958 
959 /*
960  * Flush the log to disk.
961  *
962  * If we're in the midst of dying, it's unwise to do anything that might throw
963  * an error, so we skip sending a reply in that case.
964  */
965 static void
966 XLogWalRcvFlush(bool dying, TimeLineID tli)
967 {
968  Assert(tli != 0);
969 
970  if (LogstreamResult.Flush < LogstreamResult.Write)
971  {
972  WalRcvData *walrcv = WalRcv;
973 
975 
976  LogstreamResult.Flush = LogstreamResult.Write;
977 
978  /* Update shared-memory status */
979  SpinLockAcquire(&walrcv->mutex);
980  if (walrcv->flushedUpto < LogstreamResult.Flush)
981  {
982  walrcv->latestChunkStart = walrcv->flushedUpto;
983  walrcv->flushedUpto = LogstreamResult.Flush;
984  walrcv->receivedTLI = tli;
985  }
986  SpinLockRelease(&walrcv->mutex);
987 
988  /* Signal the startup process and walsender that new WAL has arrived */
989  WakeupRecovery();
991  WalSndWakeup();
992 
993  /* Report XLOG streaming progress in PS display */
995  {
996  char activitymsg[50];
997 
998  snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
1000  set_ps_display(activitymsg);
1001  }
1002 
1003  /* Also let the primary know that we made some progress */
1004  if (!dying)
1005  {
1006  XLogWalRcvSendReply(false, false);
1007  XLogWalRcvSendHSFeedback(false);
1008  }
1009  }
1010 }
1011 
1012 /*
1013  * Close the current segment.
1014  *
1015  * Flush the segment to disk before closing it. Otherwise we have to
1016  * reopen and fsync it later.
1017  *
1018  * Create an archive notification file since the segment is known completed.
1019  */
1020 static void
1022 {
1023  char xlogfname[MAXFNAMELEN];
1024 
1026  Assert(tli != 0);
1027 
1028  /*
1029  * fsync() and close current file before we switch to next one. We would
1030  * otherwise have to reopen this file to fsync it later
1031  */
1032  XLogWalRcvFlush(false, tli);
1033 
1035 
1036  /*
1037  * XLOG segment files will be re-read by recovery in startup process soon,
1038  * so we don't advise the OS to release cache pages associated with the
1039  * file like XLogFileClose() does.
1040  */
1041  if (close(recvFile) != 0)
1042  ereport(PANIC,
1044  errmsg("could not close log segment %s: %m",
1045  xlogfname)));
1046 
1047  /*
1048  * Create .done file forcibly to prevent the streamed segment from being
1049  * archived later.
1050  */
1052  XLogArchiveForceDone(xlogfname);
1053  else
1054  XLogArchiveNotify(xlogfname);
1055 
1056  recvFile = -1;
1057 }
1058 
1059 /*
1060  * Send reply message to primary, indicating our current WAL locations, oldest
1061  * xmin and the current time.
1062  *
1063  * If 'force' is not set, the message is only sent if enough time has
1064  * passed since last status update to reach wal_receiver_status_interval.
1065  * If wal_receiver_status_interval is disabled altogether and 'force' is
1066  * false, this is a no-op.
1067  *
1068  * If 'requestReply' is true, requests the server to reply immediately upon
1069  * receiving this message. This is used for heartbeats, when approaching
1070  * wal_receiver_timeout.
1071  */
1072 static void
1073 XLogWalRcvSendReply(bool force, bool requestReply)
1074 {
1075  static XLogRecPtr writePtr = 0;
1076  static XLogRecPtr flushPtr = 0;
1077  XLogRecPtr applyPtr;
1078  static TimestampTz sendTime = 0;
1079  TimestampTz now;
1080 
1081  /*
1082  * If the user doesn't want status to be reported to the primary, be sure
1083  * to exit before doing anything at all.
1084  */
1085  if (!force && wal_receiver_status_interval <= 0)
1086  return;
1087 
1088  /* Get current timestamp. */
1090 
1091  /*
1092  * We can compare the write and flush positions to the last message we
1093  * sent without taking any lock, but the apply position requires a spin
1094  * lock, so we don't check that unless something else has changed or 10
1095  * seconds have passed. This means that the apply WAL location will
1096  * appear, from the primary's point of view, to lag slightly, but since
1097  * this is only for reporting purposes and only on idle systems, that's
1098  * probably OK.
1099  */
1100  if (!force
1101  && writePtr == LogstreamResult.Write
1102  && flushPtr == LogstreamResult.Flush
1103  && !TimestampDifferenceExceeds(sendTime, now,
1105  return;
1106  sendTime = now;
1107 
1108  /* Construct a new message */
1109  writePtr = LogstreamResult.Write;
1110  flushPtr = LogstreamResult.Flush;
1111  applyPtr = GetXLogReplayRecPtr(NULL);
1112 
1114  pq_sendbyte(&reply_message, 'r');
1115  pq_sendint64(&reply_message, writePtr);
1116  pq_sendint64(&reply_message, flushPtr);
1117  pq_sendint64(&reply_message, applyPtr);
1119  pq_sendbyte(&reply_message, requestReply ? 1 : 0);
1120 
1121  /* Send it */
1122  elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X%s",
1123  LSN_FORMAT_ARGS(writePtr),
1124  LSN_FORMAT_ARGS(flushPtr),
1125  LSN_FORMAT_ARGS(applyPtr),
1126  requestReply ? " (reply requested)" : "");
1127 
1129 }
1130 
1131 /*
1132  * Send hot standby feedback message to primary, plus the current time,
1133  * in case they don't have a watch.
1134  *
1135  * If the user disables feedback, send one final message to tell sender
1136  * to forget about the xmin on this standby. We also send this message
1137  * on first connect because a previous connection might have set xmin
1138  * on a replication slot. (If we're not using a slot it's harmless to
1139  * send a feedback message explicitly setting InvalidTransactionId).
1140  */
1141 static void
1143 {
1144  TimestampTz now;
1145  FullTransactionId nextFullXid;
1146  TransactionId nextXid;
1147  uint32 xmin_epoch,
1148  catalog_xmin_epoch;
1149  TransactionId xmin,
1150  catalog_xmin;
1151  static TimestampTz sendTime = 0;
1152 
1153  /* initially true so we always send at least one feedback message */
1154  static bool primary_has_standby_xmin = true;
1155 
1156  /*
1157  * If the user doesn't want status to be reported to the primary, be sure
1158  * to exit before doing anything at all.
1159  */
1161  !primary_has_standby_xmin)
1162  return;
1163 
1164  /* Get current timestamp. */
1166 
1167  if (!immed)
1168  {
1169  /*
1170  * Send feedback at most once per wal_receiver_status_interval.
1171  */
1172  if (!TimestampDifferenceExceeds(sendTime, now,
1174  return;
1175  sendTime = now;
1176  }
1177 
1178  /*
1179  * If Hot Standby is not yet accepting connections there is nothing to
1180  * send. Check this after the interval has expired to reduce number of
1181  * calls.
1182  *
1183  * Bailing out here also ensures that we don't send feedback until we've
1184  * read our own replication slot state, so we don't tell the primary to
1185  * discard needed xmin or catalog_xmin from any slots that may exist on
1186  * this replica.
1187  */
1188  if (!HotStandbyActive())
1189  return;
1190 
1191  /*
1192  * Make the expensive call to get the oldest xmin once we are certain
1193  * everything else has been checked.
1194  */
1196  {
1197  GetReplicationHorizons(&xmin, &catalog_xmin);
1198  }
1199  else
1200  {
1201  xmin = InvalidTransactionId;
1202  catalog_xmin = InvalidTransactionId;
1203  }
1204 
1205  /*
1206  * Get epoch and adjust if nextXid and oldestXmin are different sides of
1207  * the epoch boundary.
1208  */
1209  nextFullXid = ReadNextFullTransactionId();
1210  nextXid = XidFromFullTransactionId(nextFullXid);
1211  xmin_epoch = EpochFromFullTransactionId(nextFullXid);
1212  catalog_xmin_epoch = xmin_epoch;
1213  if (nextXid < xmin)
1214  xmin_epoch--;
1215  if (nextXid < catalog_xmin)
1216  catalog_xmin_epoch--;
1217 
1218  elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u catalog_xmin %u catalog_xmin_epoch %u",
1219  xmin, xmin_epoch, catalog_xmin, catalog_xmin_epoch);
1220 
1221  /* Construct the message and send it. */
1223  pq_sendbyte(&reply_message, 'h');
1225  pq_sendint32(&reply_message, xmin);
1226  pq_sendint32(&reply_message, xmin_epoch);
1227  pq_sendint32(&reply_message, catalog_xmin);
1228  pq_sendint32(&reply_message, catalog_xmin_epoch);
1230  if (TransactionIdIsValid(xmin) || TransactionIdIsValid(catalog_xmin))
1231  primary_has_standby_xmin = true;
1232  else
1233  primary_has_standby_xmin = false;
1234 }
1235 
1236 /*
1237  * Update shared memory status upon receiving a message from primary.
1238  *
1239  * 'walEnd' and 'sendTime' are the end-of-WAL and timestamp of the latest
1240  * message, reported by primary.
1241  */
1242 static void
1244 {
1245  WalRcvData *walrcv = WalRcv;
1246 
1247  TimestampTz lastMsgReceiptTime = GetCurrentTimestamp();
1248 
1249  /* Update shared-memory status */
1250  SpinLockAcquire(&walrcv->mutex);
1251  if (walrcv->latestWalEnd < walEnd)
1252  walrcv->latestWalEndTime = sendTime;
1253  walrcv->latestWalEnd = walEnd;
1254  walrcv->lastMsgSendTime = sendTime;
1255  walrcv->lastMsgReceiptTime = lastMsgReceiptTime;
1256  SpinLockRelease(&walrcv->mutex);
1257 
1259  {
1260  char *sendtime;
1261  char *receipttime;
1262  int applyDelay;
1263 
1264  /* Copy because timestamptz_to_str returns a static buffer */
1265  sendtime = pstrdup(timestamptz_to_str(sendTime));
1266  receipttime = pstrdup(timestamptz_to_str(lastMsgReceiptTime));
1267  applyDelay = GetReplicationApplyDelay();
1268 
1269  /* apply delay is not available */
1270  if (applyDelay == -1)
1271  elog(DEBUG2, "sendtime %s receipttime %s replication apply delay (N/A) transfer latency %d ms",
1272  sendtime,
1273  receipttime,
1275  else
1276  elog(DEBUG2, "sendtime %s receipttime %s replication apply delay %d ms transfer latency %d ms",
1277  sendtime,
1278  receipttime,
1279  applyDelay,
1281 
1282  pfree(sendtime);
1283  pfree(receipttime);
1284  }
1285 }
1286 
1287 /*
1288  * Wake up the walreceiver main loop.
1289  *
1290  * This is called by the startup process whenever interesting xlog records
1291  * are applied, so that walreceiver can check if it needs to send an apply
1292  * notification back to the primary which may be waiting in a COMMIT with
1293  * synchronous_commit = remote_apply.
1294  */
1295 void
1297 {
1298  Latch *latch;
1299 
1300  WalRcv->force_reply = true;
1301  /* fetching the latch pointer might not be atomic, so use spinlock */
1303  latch = WalRcv->latch;
1305  if (latch)
1306  SetLatch(latch);
1307 }
1308 
1309 /*
1310  * Return a string constant representing the state. This is used
1311  * in system functions and views, and should *not* be translated.
1312  */
1313 static const char *
1315 {
1316  switch (state)
1317  {
1318  case WALRCV_STOPPED:
1319  return "stopped";
1320  case WALRCV_STARTING:
1321  return "starting";
1322  case WALRCV_STREAMING:
1323  return "streaming";
1324  case WALRCV_WAITING:
1325  return "waiting";
1326  case WALRCV_RESTARTING:
1327  return "restarting";
1328  case WALRCV_STOPPING:
1329  return "stopping";
1330  }
1331  return "UNKNOWN";
1332 }
1333 
1334 /*
1335  * Returns activity of WAL receiver, including pid, state and xlog locations
1336  * received from the WAL sender of another server.
1337  */
1338 Datum
1340 {
1341  TupleDesc tupdesc;
1342  Datum *values;
1343  bool *nulls;
1344  int pid;
1345  bool ready_to_display;
1347  XLogRecPtr receive_start_lsn;
1348  TimeLineID receive_start_tli;
1349  XLogRecPtr written_lsn;
1350  XLogRecPtr flushed_lsn;
1351  TimeLineID received_tli;
1352  TimestampTz last_send_time;
1353  TimestampTz last_receipt_time;
1354  XLogRecPtr latest_end_lsn;
1355  TimestampTz latest_end_time;
1356  char sender_host[NI_MAXHOST];
1357  int sender_port = 0;
1358  char slotname[NAMEDATALEN];
1359  char conninfo[MAXCONNINFO];
1360 
1361  /* Take a lock to ensure value consistency */
1363  pid = (int) WalRcv->pid;
1364  ready_to_display = WalRcv->ready_to_display;
1366  receive_start_lsn = WalRcv->receiveStart;
1367  receive_start_tli = WalRcv->receiveStartTLI;
1368  flushed_lsn = WalRcv->flushedUpto;
1369  received_tli = WalRcv->receivedTLI;
1370  last_send_time = WalRcv->lastMsgSendTime;
1371  last_receipt_time = WalRcv->lastMsgReceiptTime;
1372  latest_end_lsn = WalRcv->latestWalEnd;
1373  latest_end_time = WalRcv->latestWalEndTime;
1374  strlcpy(slotname, (char *) WalRcv->slotname, sizeof(slotname));
1375  strlcpy(sender_host, (char *) WalRcv->sender_host, sizeof(sender_host));
1376  sender_port = WalRcv->sender_port;
1377  strlcpy(conninfo, (char *) WalRcv->conninfo, sizeof(conninfo));
1379 
1380  /*
1381  * No WAL receiver (or not ready yet), just return a tuple with NULL
1382  * values
1383  */
1384  if (pid == 0 || !ready_to_display)
1385  PG_RETURN_NULL();
1386 
1387  /*
1388  * Read "writtenUpto" without holding a spinlock. Note that it may not be
1389  * consistent with the other shared variables of the WAL receiver
1390  * protected by a spinlock, but this should not be used for data integrity
1391  * checks.
1392  */
1393  written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);
1394 
1395  /* determine result type */
1396  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
1397  elog(ERROR, "return type must be a row type");
1398 
1399  values = palloc0(sizeof(Datum) * tupdesc->natts);
1400  nulls = palloc0(sizeof(bool) * tupdesc->natts);
1401 
1402  /* Fetch values */
1403  values[0] = Int32GetDatum(pid);
1404 
1405  if (!is_member_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
1406  {
1407  /*
1408  * Only superusers and members of pg_read_all_stats can see details.
1409  * Other users only get the pid value to know whether it is a WAL
1410  * receiver, but no details.
1411  */
1412  MemSet(&nulls[1], true, sizeof(bool) * (tupdesc->natts - 1));
1413  }
1414  else
1415  {
1417 
1418  if (XLogRecPtrIsInvalid(receive_start_lsn))
1419  nulls[2] = true;
1420  else
1421  values[2] = LSNGetDatum(receive_start_lsn);
1422  values[3] = Int32GetDatum(receive_start_tli);
1423  if (XLogRecPtrIsInvalid(written_lsn))
1424  nulls[4] = true;
1425  else
1426  values[4] = LSNGetDatum(written_lsn);
1427  if (XLogRecPtrIsInvalid(flushed_lsn))
1428  nulls[5] = true;
1429  else
1430  values[5] = LSNGetDatum(flushed_lsn);
1431  values[6] = Int32GetDatum(received_tli);
1432  if (last_send_time == 0)
1433  nulls[7] = true;
1434  else
1435  values[7] = TimestampTzGetDatum(last_send_time);
1436  if (last_receipt_time == 0)
1437  nulls[8] = true;
1438  else
1439  values[8] = TimestampTzGetDatum(last_receipt_time);
1440  if (XLogRecPtrIsInvalid(latest_end_lsn))
1441  nulls[9] = true;
1442  else
1443  values[9] = LSNGetDatum(latest_end_lsn);
1444  if (latest_end_time == 0)
1445  nulls[10] = true;
1446  else
1447  values[10] = TimestampTzGetDatum(latest_end_time);
1448  if (*slotname == '\0')
1449  nulls[11] = true;
1450  else
1451  values[11] = CStringGetTextDatum(slotname);
1452  if (*sender_host == '\0')
1453  nulls[12] = true;
1454  else
1455  values[12] = CStringGetTextDatum(sender_host);
1456  if (sender_port == 0)
1457  nulls[13] = true;
1458  else
1459  values[13] = Int32GetDatum(sender_port);
1460  if (*conninfo == '\0')
1461  nulls[14] = true;
1462  else
1463  values[14] = CStringGetTextDatum(conninfo);
1464  }
1465 
1466  /* Returns the record as Datum */
1468 }
bool is_member_of_role(Oid member, Oid role)
Definition: acl.c:4869
static void pg_atomic_write_u64(volatile pg_atomic_uint64 *ptr, uint64 val)
Definition: atomics.h:438
#define pg_memory_barrier()
Definition: atomics.h:145
static uint64 pg_atomic_read_u64(volatile pg_atomic_uint64 *ptr)
Definition: atomics.h:429
void writeTimeLineHistoryFile(TimeLineID tli, char *content, int size)
Definition: timeline.c:467
bool existsTimeLineHistory(TimeLineID probeTLI)
Definition: timeline.c:222
sigset_t UnBlockSig
Definition: pqsignal.c:22
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1711
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1580
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1544
const char * timestamptz_to_str(TimestampTz t)
Definition: timestamp.c:1774
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define CStringGetTextDatum(s)
Definition: builtins.h:85
unsigned int uint32
Definition: c.h:441
#define UINT64_FORMAT
Definition: c.h:484
#define MemSet(start, val, len)
Definition: c.h:1008
uint32 TransactionId
Definition: c.h:587
size_t Size
Definition: c.h:540
void ConditionVariableBroadcast(ConditionVariable *cv)
int64 TimestampTz
Definition: timestamp.h:39
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:146
int errmsg_internal(const char *fmt,...)
Definition: elog.c:991
int errcode_for_file_access(void)
Definition: elog.c:716
int errdetail(const char *fmt,...)
Definition: elog.c:1037
bool message_level_is_interesting(int elevel)
Definition: elog.c:265
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define LOG
Definition: elog.h:25
#define FATAL
Definition: elog.h:35
#define DEBUG2
Definition: elog.h:23
#define PANIC
Definition: elog.h:36
#define DEBUG1
Definition: elog.h:24
#define ERROR
Definition: elog.h:33
#define elog(elevel,...)
Definition: elog.h:218
#define ereport(elevel,...)
Definition: elog.h:143
#define PG_RETURN_NULL()
Definition: fmgr.h:345
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:353
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:207
@ TYPEFUNC_COMPOSITE
Definition: funcapi.h:149
#define HeapTupleGetDatum(tuple)
Definition: funcapi.h:220
#define NI_MAXHOST
Definition: getaddrinfo.h:88
int MyProcPid
Definition: globals.c:43
struct Latch * MyLatch
Definition: globals.c:57
char * cluster_name
Definition: guc.c:615
@ PGC_SIGHUP
Definition: guc.h:72
void ProcessConfigFile(GucContext context)
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
#define close(a)
Definition: win32.h:12
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:109
volatile sig_atomic_t ShutdownRequestPending
Definition: interrupt.c:28
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void on_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:361
void proc_exit(int code)
Definition: ipc.c:104
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:500
void SetLatch(Latch *latch)
Definition: latch.c:567
void ResetLatch(Latch *latch)
Definition: latch.c:660
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:452
#define WL_SOCKET_READABLE
Definition: latch.h:126
#define WL_TIMEOUT
Definition: latch.h:128
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
#define WL_LATCH_SET
Definition: latch.h:125
Assert(fmt[strlen(fmt) - 1] !='\n')
exit(1)
char * pstrdup(const char *in)
Definition: mcxt.c:1299
void pfree(void *pointer)
Definition: mcxt.c:1169
void * palloc0(Size size)
Definition: mcxt.c:1093
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:120
Oid GetUserId(void)
Definition: miscinit.c:495
void * arg
#define NAMEDATALEN
const void size_t len
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
static char ** options
static char * buf
Definition: pg_test_fsync.c:70
ssize_t pg_pwrite(int fd, const void *buf, size_t nbyte, off_t offset)
Definition: pwrite.c:27
int pgsocket
Definition: port.h:29
#define snprintf
Definition: port.h:225
#define PGINVALID_SOCKET
Definition: port.h:31
size_t strlcpy(char *dst, const char *src, size_t siz)
Definition: strlcpy.c:45
uintptr_t Datum
Definition: postgres.h:411
#define DatumGetPointer(X)
Definition: postgres.h:593
#define Int32GetDatum(X)
Definition: postgres.h:523
#define PointerGetDatum(X)
Definition: postgres.h:600
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:401
int64 pq_getmsgint64(StringInfo msg)
Definition: pqformat.c:455
static void pq_sendint32(StringInfo buf, uint32 i)
Definition: pqformat.h:145
static void pq_sendbyte(StringInfo buf, uint8 byt)
Definition: pqformat.h:161
static void pq_sendint64(StringInfo buf, uint64 i)
Definition: pqformat.h:153
#define PG_SETMASK(mask)
Definition: pqsignal.h:19
void GetReplicationHorizons(TransactionId *xmin, TransactionId *catalog_xmin)
Definition: procarray.c:2047
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:642
bool update_process_title
Definition: ps_status.c:36
void set_ps_display(const char *activity)
Definition: ps_status.c:349
pqsigfunc pqsignal(int signum, pqsigfunc handler)
Definition: signal.c:180
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
PGPROC * MyProc
Definition: proc.c:68
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
void appendBinaryStringInfo(StringInfo str, const char *data, int datalen)
Definition: stringinfo.c:227
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
Definition: latch.h:111
Latch procLatch
Definition: proc.h:133
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:100
XLogRecPtr latestWalEnd
Definition: walreceiver.h:105
TimeLineID receiveStartTLI
Definition: walreceiver.h:76
TimeLineID receivedTLI
Definition: walreceiver.h:86
char slotname[NAMEDATALEN]
Definition: walreceiver.h:125
Latch * latch
Definition: walreceiver.h:143
char sender_host[NI_MAXHOST]
Definition: walreceiver.h:118
pid_t pid
Definition: walreceiver.h:64
XLogRecPtr latestChunkStart
Definition: walreceiver.h:94
XLogRecPtr receiveStart
Definition: walreceiver.h:75
XLogRecPtr flushedUpto
Definition: walreceiver.h:85
sig_atomic_t force_reply
Definition: walreceiver.h:160
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66
bool is_temp_slot
Definition: walreceiver.h:131
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:153
TimestampTz lastMsgSendTime
Definition: walreceiver.h:99
WalRcvState walRcvState
Definition: walreceiver.h:65
TimestampTz latestWalEndTime
Definition: walreceiver.h:106
bool ready_to_display
Definition: walreceiver.h:134
int sender_port
Definition: walreceiver.h:119
slock_t mutex
Definition: walreceiver.h:145
char conninfo[MAXCONNINFO]
Definition: walreceiver.h:112
Definition: regguts.h:318
#define InvalidTransactionId
Definition: transam.h:31
#define EpochFromFullTransactionId(x)
Definition: transam.h:47
#define XidFromFullTransactionId(x)
Definition: transam.h:48
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define TimestampTzGetDatum(X)
Definition: timestamp.h:32
#define TimestampTzPlusMilliseconds(tz, ms)
Definition: timestamp.h:56
FullTransactionId ReadNextFullTransactionId(void)
Definition: varsup.c:261
@ WAIT_EVENT_WAL_RECEIVER_MAIN
Definition: wait_event.h:48
@ WAIT_EVENT_WAL_RECEIVER_WAIT_START
Definition: wait_event.h:131
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:149
static WalReceiverConn * wrconn
Definition: walreceiver.c:94
void WalReceiverMain(void)
Definition: walreceiver.c:170
static StringInfoData reply_message
Definition: walreceiver.c:118
bool hot_standby_feedback
Definition: walreceiver.c:91
XLogRecPtr Flush
Definition: walreceiver.c:115
static int recvFile
Definition: walreceiver.c:104
static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
Definition: walreceiver.c:1243
int wal_receiver_status_interval
Definition: walreceiver.c:89
#define NAPTIME_PER_CYCLE
Definition: walreceiver.c:97
static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
Definition: walreceiver.c:722
XLogRecPtr Write
Definition: walreceiver.c:114
static StringInfoData incoming_message
Definition: walreceiver.c:119
static void XLogWalRcvFlush(bool dying, TimeLineID tli)
Definition: walreceiver.c:966
static TimeLineID recvFileTLI
Definition: walreceiver.c:105
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:95
static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli)
Definition: walreceiver.c:883
Datum pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
Definition: walreceiver.c:1339
int wal_receiver_timeout
Definition: walreceiver.c:90
static XLogSegNo recvSegNo
Definition: walreceiver.c:106
static void XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli)
Definition: walreceiver.c:1021
static void XLogWalRcvSendHSFeedback(bool immed)
Definition: walreceiver.c:1142
static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
Definition: walreceiver.c:642
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
Definition: walreceiver.c:816
static const char * WalRcvGetStateString(WalRcvState state)
Definition: walreceiver.c:1314
static void WalRcvDie(int code, Datum arg)
Definition: walreceiver.c:778
static void XLogWalRcvSendReply(bool force, bool requestReply)
Definition: walreceiver.c:1073
static struct @16 LogstreamResult
void WalRcvForceReply(void)
Definition: walreceiver.c:1296
#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
Definition: walreceiver.h:426
#define AllowCascadeReplication()
Definition: walreceiver.h:40
#define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
Definition: walreceiver.h:416
#define walrcv_startstreaming(conn, options)
Definition: walreceiver.h:418
#define walrcv_connect(conninfo, logical, appname, err)
Definition: walreceiver.h:404
#define walrcv_send(conn, buffer, nbytes)
Definition: walreceiver.h:424
#define walrcv_get_senderinfo(conn, sender_host, sender_port)
Definition: walreceiver.h:410
#define MAXCONNINFO
Definition: walreceiver.h:37
#define walrcv_get_conninfo(conn)
Definition: walreceiver.h:408
#define walrcv_endstreaming(conn, next_tli)
Definition: walreceiver.h:420
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
#define walrcv_identify_system(conn, primary_tli)
Definition: walreceiver.h:412
#define walrcv_disconnect(conn)
Definition: walreceiver.h:432
#define walrcv_get_backend_pid(conn)
Definition: walreceiver.h:428
#define walrcv_receive(conn, buffer, wait_fd)
Definition: walreceiver.h:422
WalRcvData * WalRcv
int GetReplicationApplyDelay(void)
int GetReplicationTransferLatency(void)
void WalSndWakeup(void)
Definition: walsender.c:3229
#define SIGCHLD
Definition: win32_port.h:177
#define SIGHUP
Definition: win32_port.h:167
#define SIG_DFL
Definition: win32_port.h:162
#define SIGPIPE
Definition: win32_port.h:172
#define SIGUSR1
Definition: win32_port.h:179
#define SIGALRM
Definition: win32_port.h:173
#define SIGUSR2
Definition: win32_port.h:180
#define SIG_IGN
Definition: win32_port.h:164
int XLogFileInit(XLogSegNo logsegno, TimeLineID logtli)
Definition: xlog.c:3521
uint64 GetSystemIdentifier(void)
Definition: xlog.c:5062
bool RecoveryInProgress(void)
Definition: xlog.c:8404
bool HotStandbyActive(void)
Definition: xlog.c:8461
int XLogArchiveMode
Definition: xlog.c:100
int wal_segment_size
Definition: xlog.c:123
void WakeupRecovery(void)
Definition: xlog.c:13222
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
Definition: xlog.c:11966
void issue_xlog_fsync(int fd, XLogSegNo segno, TimeLineID tli)
Definition: xlog.c:10859
@ ARCHIVE_MODE_ALWAYS
Definition: xlog.h:114
#define TLHistoryFileName(fname, tli)
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
#define XLogFileName(fname, tli, logSegNo, wal_segsz_bytes)
#define MAXFNAMELEN
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define XLByteInSeg(xlrp, logSegNo, wal_segsz_bytes)
void XLogArchiveForceDone(const char *xlog)
Definition: xlogarchive.c:540
void XLogArchiveNotify(const char *xlog)
Definition: xlogarchive.c:474
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
uint32 TimeLineID
Definition: xlogdefs.h:59
uint64 XLogSegNo
Definition: xlogdefs.h:48