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