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