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