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