Age Owner Branch data TLA Line data Source code
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 : : * ereport(FATAL). Emergency termination is by SIGQUIT; like any postmaster
34 : : * child process, the walreceiver will simply abort and exit on SIGQUIT. A
35 : : * close 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-2026, 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.h"
58 : : #include "access/xlog_internal.h"
59 : : #include "access/xlogarchive.h"
60 : : #include "access/xlogrecovery.h"
61 : : #include "access/xlogwait.h"
62 : : #include "catalog/pg_authid.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/auxprocess.h"
69 : : #include "postmaster/interrupt.h"
70 : : #include "replication/walreceiver.h"
71 : : #include "replication/walsender.h"
72 : : #include "storage/ipc.h"
73 : : #include "storage/proc.h"
74 : : #include "storage/procarray.h"
75 : : #include "storage/procsignal.h"
76 : : #include "tcop/tcopprot.h"
77 : : #include "utils/acl.h"
78 : : #include "utils/builtins.h"
79 : : #include "utils/guc.h"
80 : : #include "utils/pg_lsn.h"
81 : : #include "utils/ps_status.h"
82 : : #include "utils/timestamp.h"
83 : : #include "utils/wait_event.h"
84 : :
85 : :
86 : : /*
87 : : * GUC variables. (Other variables that affect walreceiver are in xlog.c
88 : : * because they're passed down from the startup process, for better
89 : : * synchronization.)
90 : : */
91 : : int wal_receiver_status_interval;
92 : : int wal_receiver_timeout;
93 : : bool hot_standby_feedback;
94 : :
95 : : /* libpqwalreceiver connection */
96 : : static WalReceiverConn *wrconn = NULL;
97 : : WalReceiverFunctionsType *WalReceiverFunctions = NULL;
98 : :
99 : : /*
100 : : * These variables are used similarly to openLogFile/SegNo,
101 : : * but for walreceiver to write the XLOG. recvFileTLI is the TimeLineID
102 : : * corresponding the filename of recvFile.
103 : : */
104 : : static int recvFile = -1;
105 : : static TimeLineID recvFileTLI = 0;
106 : : static XLogSegNo recvSegNo = 0;
107 : :
108 : : /*
109 : : * LogstreamResult indicates the byte positions that we have already
110 : : * written/fsynced.
111 : : */
112 : : static struct
113 : : {
114 : : XLogRecPtr Write; /* last byte + 1 written out in the standby */
115 : : XLogRecPtr Flush; /* last byte + 1 flushed in the standby */
116 : : } LogstreamResult;
117 : :
118 : : /*
119 : : * Reasons to wake up and perform periodic tasks.
120 : : */
121 : : typedef enum WalRcvWakeupReason
122 : : {
123 : : WALRCV_WAKEUP_TERMINATE,
124 : : WALRCV_WAKEUP_PING,
125 : : WALRCV_WAKEUP_REPLY,
126 : : WALRCV_WAKEUP_HSFEEDBACK,
127 : : #define NUM_WALRCV_WAKEUPS (WALRCV_WAKEUP_HSFEEDBACK + 1)
128 : : } WalRcvWakeupReason;
129 : :
130 : : /*
131 : : * Wake up times for periodic tasks.
132 : : */
133 : : static TimestampTz wakeup[NUM_WALRCV_WAKEUPS];
134 : :
135 : : static StringInfoData reply_message;
136 : :
137 : : /* Prototypes for private functions */
138 : : static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last);
139 : : static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI);
140 : : static void WalRcvDie(int code, Datum arg);
141 : : static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len,
142 : : TimeLineID tli);
143 : : static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr,
144 : : TimeLineID tli);
145 : : static void XLogWalRcvFlush(bool dying, TimeLineID tli);
146 : : static void XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli);
147 : : static void XLogWalRcvSendReply(bool force, bool requestReply, bool checkApply);
148 : : static void XLogWalRcvSendHSFeedback(bool immed);
149 : : static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
150 : : static void WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now);
151 : :
152 : :
153 : : /* Main entry point for walreceiver process */
154 : : void
576 peter@eisentraut.org 155 :CBC 283 : WalReceiverMain(const void *startup_data, size_t startup_data_len)
156 : : {
157 : : char conninfo[MAXCONNINFO];
158 : : char *tmp_conninfo;
159 : : char slotname[NAMEDATALEN];
160 : : bool is_temp_slot;
23 fujii@postgresql.org 161 : 283 : bool temp_slot_created = false;
162 : : XLogRecPtr startpoint;
163 : : TimeLineID startpointTLI;
164 : : TimeLineID primaryTLI;
165 : : bool first_stream;
53 alvherre@kurilemu.de 166 : 283 : bool upstream_catchup_logged = false;
167 : 283 : TimestampTz upstream_catchup_deadline = 0;
168 : : WalRcvData *walrcv;
169 : : TimestampTz now;
170 : : char *err;
3095 fujii@postgresql.org 171 : 283 : char *sender_host = NULL;
172 : 283 : int sender_port = 0;
173 : : char *appname;
174 : :
916 heikki.linnakangas@i 175 [ - + ]: 283 : Assert(startup_data_len == 0);
176 : :
177 : 283 : AuxiliaryProcessMainCommon();
178 : :
179 : : /*
180 : : * WalRcv should be set up already (if we are a backend, we inherit this
181 : : * by fork() or EXEC_BACKEND mechanism from the postmaster).
182 : : */
1022 183 : 283 : walrcv = WalRcv;
6080 184 [ - + ]: 283 : Assert(walrcv != NULL);
185 : :
186 : : /*
187 : : * Mark walreceiver as running in shared memory.
188 : : *
189 : : * Do this as early as possible, so that if we fail later on, we'll set
190 : : * state to STOPPED. If we die before this, the startup process will keep
191 : : * waiting for us to start up, until it times out.
192 : : */
193 : 283 : SpinLockAcquire(&walrcv->mutex);
194 [ - + ]: 283 : Assert(walrcv->pid == 0);
6050 bruce@momjian.us 195 [ - + + - ]: 283 : switch (walrcv->walRcvState)
196 : : {
6080 heikki.linnakangas@i 197 :UBC 0 : case WALRCV_STOPPING:
198 : : /* If we've already been requested to stop, don't start up. */
199 : 0 : walrcv->walRcvState = WALRCV_STOPPED;
200 : : pg_fallthrough;
201 : :
6080 heikki.linnakangas@i 202 :CBC 3 : case WALRCV_STOPPED:
203 : 3 : SpinLockRelease(&walrcv->mutex);
2018 tmunro@postgresql.or 204 : 3 : ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
6080 heikki.linnakangas@i 205 : 3 : proc_exit(1);
206 : : break;
207 : :
208 : 280 : case WALRCV_STARTING:
209 : : /* The usual case */
210 : 280 : break;
211 : :
240 michael@paquier.xyz 212 :UBC 0 : case WALRCV_CONNECTING:
213 : : case WALRCV_WAITING:
214 : : case WALRCV_STREAMING:
215 : : case WALRCV_RESTARTING:
216 : : default:
217 : : /* Shouldn't happen */
3274 alvherre@alvh.no-ip. 218 : 0 : SpinLockRelease(&walrcv->mutex);
6080 heikki.linnakangas@i 219 [ # # ]: 0 : elog(PANIC, "walreceiver still running according to shared memory state");
220 : : }
221 : : /* Advertise our PID so that the startup process can kill us */
6080 heikki.linnakangas@i 222 :CBC 280 : walrcv->pid = MyProcPid;
240 michael@paquier.xyz 223 : 280 : walrcv->walRcvState = WALRCV_CONNECTING;
224 : :
225 : : /* Fetch information required to start streaming */
3733 alvherre@alvh.no-ip. 226 : 280 : walrcv->ready_to_display = false;
585 peter@eisentraut.org 227 : 280 : strlcpy(conninfo, walrcv->conninfo, MAXCONNINFO);
228 : 280 : strlcpy(slotname, walrcv->slotname, NAMEDATALEN);
2441 229 : 280 : is_temp_slot = walrcv->is_temp_slot;
5682 heikki.linnakangas@i 230 : 280 : startpoint = walrcv->receiveStart;
5029 231 : 280 : startpointTLI = walrcv->receiveStartTLI;
232 : :
233 : : /*
234 : : * At most one of is_temp_slot and slotname can be set; otherwise,
235 : : * RequestXLogStreaming messed up.
236 : : */
2368 alvherre@alvh.no-ip. 237 [ + + - + ]: 280 : Assert(!is_temp_slot || (slotname[0] == '\0'));
238 : :
239 : : /* Initialise to a sanish value */
1333 tgl@sss.pgh.pa.us 240 : 280 : now = GetCurrentTimestamp();
3274 alvherre@alvh.no-ip. 241 : 280 : walrcv->lastMsgSendTime =
1412 tmunro@postgresql.or 242 : 280 : walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;
243 : :
244 : : /* Report our proc number so that others can wake us up */
688 heikki.linnakangas@i 245 : 280 : walrcv->procno = MyProcNumber;
246 : :
6080 247 : 280 : SpinLockRelease(&walrcv->mutex);
248 : :
249 : : /* Arrange to clean up at walreceiver exit */
1780 rhaas@postgresql.org 250 : 280 : on_shmem_exit(WalRcvDie, PointerGetDatum(&startpointTLI));
251 : :
252 : : /* Properly accept or ignore signals the postmaster might send us */
2138 fujii@postgresql.org 253 : 280 : pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
254 : : * file */
159 andrew@dunslane.net 255 : 280 : pqsignal(SIGINT, PG_SIG_IGN);
534 heikki.linnakangas@i 256 : 280 : pqsignal(SIGTERM, die); /* request shutdown */
257 : : /* SIGQUIT handler was already set up by InitPostmasterChild */
159 andrew@dunslane.net 258 : 280 : pqsignal(SIGALRM, PG_SIG_IGN);
259 : 280 : pqsignal(SIGPIPE, PG_SIG_IGN);
2491 rhaas@postgresql.org 260 : 280 : pqsignal(SIGUSR1, procsignal_sigusr1_handler);
159 andrew@dunslane.net 261 : 280 : pqsignal(SIGUSR2, PG_SIG_IGN);
262 : :
263 : : /* Reset some signals that are accepted by postmaster but not here */
264 : 280 : pqsignal(SIGCHLD, PG_SIG_DFL);
265 : :
266 : : /* Load the libpq-specific functions */
6080 heikki.linnakangas@i 267 : 280 : load_file("libpqwalreceiver", false);
3581 peter_e@gmx.net 268 [ - + ]: 280 : if (WalReceiverFunctions == NULL)
6080 heikki.linnakangas@i 269 [ # # ]:UBC 0 : elog(ERROR, "libpqwalreceiver didn't initialize correctly");
270 : :
271 : : /* Unblock signals (they were blocked when the postmaster forked us) */
1325 tmunro@postgresql.or 272 :CBC 280 : sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
273 : :
274 : : /*
275 : : * Switch the WAL receiver state as ready for display before doing a
276 : : * connection attempt, so as its connecting state is visible before
277 : : * attempting to contact the primary server. Note that this resets the
278 : : * original conninfo, sender_port and sender_host, for security. These
279 : : * fields are filled once the connection is fully established.
280 : : */
120 michael@paquier.xyz 281 : 280 : SpinLockAcquire(&walrcv->mutex);
282 : 280 : memset(walrcv->conninfo, 0, MAXCONNINFO);
283 : 280 : memset(walrcv->sender_host, 0, NI_MAXHOST);
284 : 280 : walrcv->sender_port = 0;
285 : 280 : walrcv->ready_to_display = true;
286 : 280 : SpinLockRelease(&walrcv->mutex);
287 : :
288 : : /* Establish the connection to the primary for XLOG streaming */
801 tgl@sss.pgh.pa.us 289 [ + + ]: 280 : appname = cluster_name[0] ? cluster_name : "walreceiver";
290 : 280 : wrconn = walrcv_connect(conninfo, true, false, false, appname, &err);
3531 peter_e@gmx.net 291 [ + + ]: 279 : if (!wrconn)
292 [ + - ]: 108 : ereport(ERROR,
293 : : (errcode(ERRCODE_CONNECTION_FAILURE),
294 : : errmsg("streaming replication receiver \"%s\" could not connect to the primary server: %s",
295 : : appname, err)));
296 : :
297 : : /*
298 : : * Save user-visible connection string, now that the connection has been
299 : : * achieved.
300 : : */
3581 301 : 171 : tmp_conninfo = walrcv_get_conninfo(wrconn);
3095 fujii@postgresql.org 302 : 171 : walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
3735 alvherre@alvh.no-ip. 303 : 171 : SpinLockAcquire(&walrcv->mutex);
304 [ + - ]: 171 : if (tmp_conninfo)
585 peter@eisentraut.org 305 : 171 : strlcpy(walrcv->conninfo, tmp_conninfo, MAXCONNINFO);
3095 fujii@postgresql.org 306 [ + - ]: 171 : if (sender_host)
585 peter@eisentraut.org 307 : 171 : strlcpy(walrcv->sender_host, sender_host, NI_MAXHOST);
3095 fujii@postgresql.org 308 : 171 : walrcv->sender_port = sender_port;
3735 alvherre@alvh.no-ip. 309 : 171 : SpinLockRelease(&walrcv->mutex);
310 : :
3274 311 [ + - ]: 171 : if (tmp_conninfo)
312 : 171 : pfree(tmp_conninfo);
313 : :
3095 fujii@postgresql.org 314 [ + - ]: 171 : if (sender_host)
315 : 171 : pfree(sender_host);
316 : :
317 : : /* Initialize buffers for processing messages */
146 michael@paquier.xyz 318 : 171 : initStringInfo(&reply_message);
319 : :
5029 heikki.linnakangas@i 320 : 171 : first_stream = true;
321 : : for (;;)
6092 322 : 15 : {
323 : : char *primary_sysid;
324 : : char standby_sysid[32];
325 : : XLogRecPtr primaryFlushPtr;
326 : : WalRcvStreamOptions options;
327 : :
328 : : /*
329 : : * Check that we're connected to a valid server using the
330 : : * IDENTIFY_SYSTEM replication command.
331 : : */
53 alvherre@kurilemu.de 332 : 186 : primary_sysid = walrcv_identify_system(wrconn, &primaryTLI,
333 : : &primaryFlushPtr);
334 : :
3581 peter_e@gmx.net 335 : 186 : snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
336 : : GetSystemIdentifier());
337 [ - + ]: 186 : if (strcmp(primary_sysid, standby_sysid) != 0)
338 : : {
3581 peter_e@gmx.net 339 [ # # ]:UBC 0 : ereport(ERROR,
340 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
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 : : }
146 michael@paquier.xyz 345 :CBC 186 : pfree(primary_sysid);
346 : :
347 : : /*
348 : : * Confirm that the current timeline of the primary is the same or
349 : : * ahead of ours.
350 : : */
5029 heikki.linnakangas@i 351 [ - + ]: 186 : if (primaryTLI < startpointTLI)
5029 heikki.linnakangas@i 352 [ # # ]:UBC 0 : ereport(ERROR,
353 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
354 : : errmsg("highest timeline %u of the primary is behind recovery timeline %u",
355 : : primaryTLI, startpointTLI)));
356 : :
357 : : /*
358 : : * If our requested startpoint is ahead of the upstream server's
359 : : * current WAL flush position, we cannot start streaming yet. (We say
360 : : * "upstream" here and not "primary" because this condition can only
361 : : * happen on a cascading standby.) This can happen when such a
362 : : * cascading standby has advanced past the upstream via archive
363 : : * recovery but the intermediate standby has not caught up with that
364 : : * yet. In this case, wait for the upstream to catch up before
365 : : * attempting START_REPLICATION, because that would fail with
366 : : * "requested starting point is ahead of the WAL flush position".
367 : : *
368 : : * We only perform this check when we're on the same timeline as the
369 : : * primary; when timelines differ, let START_REPLICATION handle the
370 : : * timeline negotiation.
371 : : *
372 : : * We also only wait if the gap is within one WAL segment. This is
373 : : * the expected case because archive recovery processes whole segment
374 : : * files: the cascade's next read position lands at the start of the
375 : : * following segment while the upstream's flush position is still
376 : : * inside the just-replayed one, producing at most a sub-segment gap.
377 : : * A larger gap means the upstream is genuinely behind, so we let
378 : : * START_REPLICATION fail normally and allow the startup process to
379 : : * fall back to other WAL sources.
380 : : *
381 : : * Honor wal_receiver_timeout so the walreceiver doesn't wait
382 : : * indefinitely: if the upstream hasn't caught up within the timeout,
383 : : * exit and let the startup process retry normally.
384 : : */
53 alvherre@kurilemu.de 385 [ + + ]:CBC 186 : if (startpointTLI == primaryTLI &&
386 [ + + ]: 175 : startpoint > primaryFlushPtr &&
387 [ + - ]: 1 : startpoint - primaryFlushPtr <= wal_segment_size)
388 : : {
389 : : /* Set deadline on first iteration */
390 [ + - + - ]: 1 : if (!upstream_catchup_logged && wal_receiver_timeout > 0)
391 : 1 : upstream_catchup_deadline =
392 : 1 : TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
393 : : wal_receiver_timeout);
394 : :
395 [ - + + - ]: 1 : ereport(upstream_catchup_logged ? DEBUG1 : LOG,
396 : : errmsg("walreceiver requested start point %X/%08X on timeline %u is ahead of the upstream server's flush position %X/%08X, waiting",
397 : : LSN_FORMAT_ARGS(startpoint), startpointTLI,
398 : : LSN_FORMAT_ARGS(primaryFlushPtr)));
399 : 1 : upstream_catchup_logged = true;
400 : :
401 : 1 : (void) WaitLatch(MyLatch,
402 : : WL_EXIT_ON_PM_DEATH | WL_TIMEOUT | WL_LATCH_SET,
403 : : wal_retrieve_retry_interval,
404 : : WAIT_EVENT_WAL_RECEIVER_UPSTREAM_CATCHUP);
405 : 1 : ResetLatch(MyLatch);
406 : :
407 [ + - - + ]: 2 : if (upstream_catchup_deadline > 0 &&
408 : 1 : GetCurrentTimestamp() >= upstream_catchup_deadline)
53 alvherre@kurilemu.de 409 [ # # ]:UBC 0 : ereport(ERROR,
410 : : (errcode(ERRCODE_CONNECTION_FAILURE),
411 : : errmsg("terminating walreceiver due to timeout while waiting for upstream to catch up")));
412 : :
53 alvherre@kurilemu.de 413 [ - + ]:CBC 1 : CHECK_FOR_INTERRUPTS();
414 : 1 : continue;
415 : : }
416 : : else
417 : : {
418 : 185 : upstream_catchup_logged = false;
419 : 185 : upstream_catchup_deadline = 0;
420 : : }
421 : :
422 : : /*
423 : : * Get any missing history files. We do this always, even when we're
424 : : * not interested in that timeline, so that if we're promoted to
425 : : * become the primary later on, we don't select the same timeline that
426 : : * was already used in the current primary. This isn't bullet-proof -
427 : : * you'll need some external software to manage your cluster if you
428 : : * need to ensure that a unique timeline id is chosen in every case,
429 : : * but let's avoid the confusion of timeline id collisions where we
430 : : * can.
431 : : */
5008 heikki.linnakangas@i 432 : 185 : WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
433 : :
434 : : /*
435 : : * Create a temporary replication slot if requested. This only needs
436 : : * to be done for the first streaming attempt on this connection
437 : : * because the slot remains available while this connection is reused
438 : : * for later streaming attempts. Update the slot name in shared
439 : : * memory each time because RequestXLogStreaming() clears it when
440 : : * restarting streaming.
441 : : */
2368 alvherre@alvh.no-ip. 442 [ + + ]: 185 : if (is_temp_slot)
443 : : {
23 fujii@postgresql.org 444 [ + + ]: 2 : if (!temp_slot_created)
445 : : {
446 : 1 : snprintf(slotname, sizeof(slotname),
447 : : "pg_walreceiver_%lld",
448 : 1 : (long long int) walrcv_get_backend_pid(wrconn));
449 : :
450 : 1 : walrcv_create_slot(wrconn, slotname, true, false, false, 0, NULL);
451 : :
452 : 1 : temp_slot_created = true;
453 : : }
454 : :
2368 alvherre@alvh.no-ip. 455 : 2 : SpinLockAcquire(&walrcv->mutex);
456 : 2 : strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
457 : 2 : SpinLockRelease(&walrcv->mutex);
458 : : }
459 : :
460 : : /*
461 : : * Start streaming.
462 : : *
463 : : * We'll try to start at the requested starting point and timeline,
464 : : * even if it's different from the server's latest timeline. In case
465 : : * we've already reached the end of the old timeline, the server will
466 : : * finish the streaming immediately, and we will go back to await
467 : : * orders from the startup process. If recovery_target_timeline is
468 : : * 'latest', the startup process will scan pg_wal and find the new
469 : : * history file, bump recovery target timeline, and ask us to restart
470 : : * on the new timeline.
471 : : */
3531 peter_e@gmx.net 472 : 185 : options.logical = false;
473 : 185 : options.startpoint = startpoint;
474 [ + + ]: 185 : options.slotname = slotname[0] != '\0' ? slotname : NULL;
475 : 185 : options.proto.physical.startpointTLI = startpointTLI;
476 [ + - ]: 185 : if (walrcv_startstreaming(wrconn, &options))
477 : : {
5029 heikki.linnakangas@i 478 [ + + ]: 183 : if (first_stream)
479 [ + - ]: 169 : ereport(LOG,
480 : : errmsg("started streaming WAL from primary at %X/%08X on timeline %u",
481 : : LSN_FORMAT_ARGS(startpoint), startpointTLI));
482 : : else
483 [ + - ]: 14 : ereport(LOG,
484 : : errmsg("restarted WAL streaming at %X/%08X on timeline %u",
485 : : LSN_FORMAT_ARGS(startpoint), startpointTLI));
486 : 183 : first_stream = false;
487 : :
488 : : /*
489 : : * Switch to STREAMING after a successful connection if current
490 : : * state is CONNECTING. This switch happens after an initial
491 : : * startup, or after a restart as determined by
492 : : * WalRcvWaitForStartPosition().
493 : : */
240 michael@paquier.xyz 494 : 183 : SpinLockAcquire(&walrcv->mutex);
495 [ + - ]: 183 : if (walrcv->walRcvState == WALRCV_CONNECTING)
496 : 183 : walrcv->walRcvState = WALRCV_STREAMING;
497 : 183 : SpinLockRelease(&walrcv->mutex);
498 : :
499 : : /* Initialize LogstreamResult for processing messages */
5022 heikki.linnakangas@i 500 : 183 : LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
501 : :
502 : : /* Initialize nap wakeup times. */
1412 tmunro@postgresql.or 503 : 183 : now = GetCurrentTimestamp();
504 [ + + ]: 915 : for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
505 : 732 : WalRcvComputeNextWakeup(i, now);
506 : :
507 : : /* Send initial reply/feedback messages. */
178 fujii@postgresql.org 508 : 183 : XLogWalRcvSendReply(true, false, false);
1403 tmunro@postgresql.or 509 : 183 : XLogWalRcvSendHSFeedback(true);
510 : :
511 : : /* Loop until end-of-streaming or error */
512 : : for (;;)
5092 heikki.linnakangas@i 513 : 70556 : {
514 : : char *buf;
515 : : int len;
3827 rhaas@postgresql.org 516 : 70739 : bool endofwal = false;
3811 tgl@sss.pgh.pa.us 517 : 70739 : pgsocket wait_fd = PGINVALID_SOCKET;
518 : : int rc;
519 : : TimestampTz nextWakeup;
520 : : long nap;
521 : :
522 : : /*
523 : : * Exit walreceiver if we're not in recovery. This should not
524 : : * happen, but cross-check the status here.
525 : : */
5029 heikki.linnakangas@i 526 [ - + ]: 70739 : if (!RecoveryInProgress())
5029 heikki.linnakangas@i 527 [ # # ]:UBC 0 : ereport(FATAL,
528 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
529 : : errmsg("cannot continue WAL streaming, recovery has already ended")));
530 : :
531 : : /* Process any requests or signals received recently */
534 heikki.linnakangas@i 532 [ - + ]:CBC 70739 : CHECK_FOR_INTERRUPTS();
533 : :
2138 fujii@postgresql.org 534 [ + + ]: 70739 : if (ConfigReloadPending)
535 : : {
536 : 31 : ConfigReloadPending = false;
5029 heikki.linnakangas@i 537 : 31 : ProcessConfigFile(PGC_SIGHUP);
538 : : /* recompute wakeup times */
1412 tmunro@postgresql.or 539 : 31 : now = GetCurrentTimestamp();
540 [ + + ]: 155 : for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
541 : 124 : WalRcvComputeNextWakeup(i, now);
4976 simon@2ndQuadrant.co 542 : 31 : XLogWalRcvSendHSFeedback(true);
543 : : }
544 : :
545 : : /* See if we can read data immediately */
3581 peter_e@gmx.net 546 : 70739 : len = walrcv_receive(wrconn, &buf, &wait_fd);
5029 heikki.linnakangas@i 547 [ + + ]: 70706 : if (len != 0)
548 : : {
549 : : /*
550 : : * Process the received data, and any subsequent data we
551 : : * can read without blocking.
552 : : */
553 : : for (;;)
554 : : {
555 [ + + ]: 163171 : if (len > 0)
556 : : {
557 : : /*
558 : : * Something was received from primary, so adjust
559 : : * the ping and terminate wakeup times.
560 : : */
1333 tgl@sss.pgh.pa.us 561 : 109701 : now = GetCurrentTimestamp();
1412 tmunro@postgresql.or 562 : 109701 : WalRcvComputeNextWakeup(WALRCV_WAKEUP_TERMINATE,
563 : : now);
564 : 109701 : WalRcvComputeNextWakeup(WALRCV_WAKEUP_PING, now);
1780 rhaas@postgresql.org 565 : 109701 : XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
566 : : startpointTLI);
567 : : }
5029 heikki.linnakangas@i 568 [ + + ]: 53470 : else if (len == 0)
569 : 53420 : break;
570 [ + - ]: 50 : else if (len < 0)
571 : : {
572 [ + - ]: 50 : ereport(LOG,
573 : : (errmsg("replication terminated by primary server"),
574 : : errdetail("End of WAL reached on timeline %u at %X/%08X.",
575 : : startpointTLI,
576 : : LSN_FORMAT_ARGS(LogstreamResult.Write))));
577 : 50 : endofwal = true;
578 : 50 : break;
579 : : }
3581 peter_e@gmx.net 580 : 109701 : len = walrcv_receive(wrconn, &buf, &wait_fd);
581 : : }
582 : :
583 : : /* Let the primary know that we received some data. */
178 fujii@postgresql.org 584 : 53470 : XLogWalRcvSendReply(false, false, false);
585 : :
586 : : /*
587 : : * If we've written some records, flush them to disk and
588 : : * let the startup process and primary server know about
589 : : * them.
590 : : */
1780 rhaas@postgresql.org 591 : 53469 : XLogWalRcvFlush(false, startpointTLI);
592 : : }
593 : :
594 : : /* Check if we need to exit the streaming loop. */
3827 595 [ + + ]: 70704 : if (endofwal)
596 : 49 : break;
597 : :
598 : : /* Find the soonest wakeup time, to limit our nap. */
1333 tgl@sss.pgh.pa.us 599 : 70655 : nextWakeup = TIMESTAMP_INFINITY;
1412 tmunro@postgresql.or 600 [ + + ]: 353275 : for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
601 : 282620 : nextWakeup = Min(wakeup[i], nextWakeup);
602 : :
603 : : /* Calculate the nap time, clamping as necessary. */
1333 tgl@sss.pgh.pa.us 604 : 70655 : now = GetCurrentTimestamp();
605 : 70655 : nap = TimestampDifferenceMilliseconds(now, nextWakeup);
606 : :
607 : : /*
608 : : * Ideally we would reuse a WaitEventSet object repeatedly
609 : : * here to avoid the overheads of WaitLatchOrSocket on epoll
610 : : * systems, but we can't be sure that libpq (or any other
611 : : * walreceiver implementation) has the same socket (even if
612 : : * the fd is the same number, it may have been closed and
613 : : * reopened since the last time). In future, if there is a
614 : : * function for removing sockets from WaitEventSet, then we
615 : : * could add and remove just the socket each time, potentially
616 : : * avoiding some system calls.
617 : : */
3827 rhaas@postgresql.org 618 [ - + ]: 70655 : Assert(wait_fd != PGINVALID_SOCKET);
2138 fujii@postgresql.org 619 : 70655 : rc = WaitLatchOrSocket(MyLatch,
620 : : WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
621 : : WL_TIMEOUT | WL_LATCH_SET,
622 : : wait_fd,
623 : : nap,
624 : : WAIT_EVENT_WAL_RECEIVER_MAIN);
3827 rhaas@postgresql.org 625 [ + + ]: 70655 : if (rc & WL_LATCH_SET)
626 : : {
2138 fujii@postgresql.org 627 : 12362 : ResetLatch(MyLatch);
534 heikki.linnakangas@i 628 [ + + ]: 12362 : CHECK_FOR_INTERRUPTS();
629 : :
178 fujii@postgresql.org 630 [ + + ]: 12263 : if (walrcv->apply_reply_requested)
631 : : {
632 : : /*
633 : : * The recovery process has asked us to send apply
634 : : * feedback now. Make sure the flag is really set to
635 : : * false in shared memory before sending the reply, so
636 : : * we don't miss a new request for a reply.
637 : : */
638 : 12193 : walrcv->apply_reply_requested = false;
3827 rhaas@postgresql.org 639 : 12193 : pg_memory_barrier();
178 fujii@postgresql.org 640 : 12193 : XLogWalRcvSendReply(false, false, true);
641 : : }
642 : : }
3827 rhaas@postgresql.org 643 [ + + ]: 70556 : if (rc & WL_TIMEOUT)
644 : : {
645 : : /*
646 : : * We didn't receive anything new. If we haven't heard
647 : : * anything from the server for more than
648 : : * wal_receiver_timeout / 2, ping the server. Also, if
649 : : * it's been longer than wal_receiver_status_interval
650 : : * since the last update we sent, send a status update to
651 : : * the primary anyway, to report any progress in applying
652 : : * WAL.
653 : : */
4862 bruce@momjian.us 654 : 12 : bool requestReply = false;
655 : :
656 : : /*
657 : : * Report pending statistics to the cumulative stats
658 : : * system. This location is useful for the report as it
659 : : * is not within a tight loop in the WAL receiver, to
660 : : * avoid bloating pgstats with requests, while also making
661 : : * sure that the reports happen each time a status update
662 : : * is sent.
663 : : */
564 michael@paquier.xyz 664 : 12 : pgstat_report_wal(false);
665 : :
666 : : /*
667 : : * Check if time since last receive from primary has
668 : : * reached the configured limit.
669 : : */
1333 tgl@sss.pgh.pa.us 670 : 12 : now = GetCurrentTimestamp();
1412 tmunro@postgresql.or 671 [ - + ]: 12 : if (now >= wakeup[WALRCV_WAKEUP_TERMINATE])
1412 tmunro@postgresql.or 672 [ # # ]:UBC 0 : ereport(ERROR,
673 : : (errcode(ERRCODE_CONNECTION_FAILURE),
674 : : errmsg("terminating walreceiver due to timeout")));
675 : :
676 : : /*
677 : : * If we didn't receive anything new for half of receiver
678 : : * replication timeout, then ping the server.
679 : : */
1412 tmunro@postgresql.or 680 [ - + ]:CBC 12 : if (now >= wakeup[WALRCV_WAKEUP_PING])
681 : : {
1412 tmunro@postgresql.or 682 :UBC 0 : requestReply = true;
1333 tgl@sss.pgh.pa.us 683 : 0 : wakeup[WALRCV_WAKEUP_PING] = TIMESTAMP_INFINITY;
684 : : }
685 : :
178 fujii@postgresql.org 686 :CBC 12 : XLogWalRcvSendReply(requestReply, requestReply, false);
4976 simon@2ndQuadrant.co 687 : 12 : XLogWalRcvSendHSFeedback(false);
688 : : }
689 : : }
690 : :
691 : : /*
692 : : * The backend finished streaming. Exit streaming COPY-mode from
693 : : * our side, too.
694 : : */
3581 peter_e@gmx.net 695 : 49 : walrcv_endstreaming(wrconn, &primaryTLI);
696 : :
697 : : /*
698 : : * If the server had switched to a new timeline that we didn't
699 : : * know about when we began streaming, fetch its timeline history
700 : : * file now.
701 : : */
4993 heikki.linnakangas@i 702 : 14 : WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
703 : : }
704 : : else
5029 heikki.linnakangas@i 705 [ # # ]:UBC 0 : ereport(LOG,
706 : : (errmsg("primary server contains no more WAL on requested timeline %u",
707 : : startpointTLI)));
708 : :
709 : : /*
710 : : * End of WAL reached on the requested timeline. Close the last
711 : : * segment, and await for new orders from the startup process.
712 : : */
5029 heikki.linnakangas@i 713 [ + + ]:CBC 14 : if (recvFile >= 0)
714 : : {
715 : : char xlogfname[MAXFNAMELEN];
716 : :
1780 rhaas@postgresql.org 717 : 13 : XLogWalRcvFlush(false, startpointTLI);
2483 michael@paquier.xyz 718 : 13 : XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
5029 heikki.linnakangas@i 719 [ - + ]: 13 : if (close(recvFile) != 0)
5029 heikki.linnakangas@i 720 [ # # ]:UBC 0 : ereport(PANIC,
721 : : (errcode_for_file_access(),
722 : : errmsg("could not close WAL segment %s: %m",
723 : : xlogfname)));
724 : :
725 : : /*
726 : : * Create .done file forcibly to prevent the streamed segment from
727 : : * being archived later.
728 : : */
4146 heikki.linnakangas@i 729 [ + - ]:CBC 13 : if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
730 : 13 : XLogArchiveForceDone(xlogfname);
731 : : else
1842 alvherre@alvh.no-ip. 732 :UBC 0 : XLogArchiveNotify(xlogfname);
733 : : }
5029 heikki.linnakangas@i 734 :CBC 14 : recvFile = -1;
735 : :
736 [ + + ]: 14 : elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
737 : 14 : WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
738 : : }
739 : : /* not reached */
740 : : }
741 : :
742 : : /*
743 : : * Wait for startup process to set receiveStart and receiveStartTLI.
744 : : */
745 : : static void
746 : 14 : WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
747 : : {
4002 rhaas@postgresql.org 748 : 14 : WalRcvData *walrcv = WalRcv;
749 : : int state;
750 : :
5029 heikki.linnakangas@i 751 : 14 : SpinLockAcquire(&walrcv->mutex);
752 : 14 : state = walrcv->walRcvState;
240 michael@paquier.xyz 753 [ - + - - ]: 14 : if (state != WALRCV_STREAMING && state != WALRCV_CONNECTING)
754 : : {
5029 heikki.linnakangas@i 755 :UBC 0 : SpinLockRelease(&walrcv->mutex);
756 [ # # ]: 0 : if (state == WALRCV_STOPPING)
757 : 0 : proc_exit(0);
758 : : else
759 [ # # ]: 0 : elog(FATAL, "unexpected walreceiver state");
760 : : }
5029 heikki.linnakangas@i 761 :CBC 14 : walrcv->walRcvState = WALRCV_WAITING;
762 : 14 : walrcv->receiveStart = InvalidXLogRecPtr;
763 : 14 : walrcv->receiveStartTLI = 0;
764 : 14 : SpinLockRelease(&walrcv->mutex);
765 : :
2384 peter@eisentraut.org 766 : 14 : set_ps_display("idle");
767 : :
768 : : /*
769 : : * nudge startup process to notice that we've stopped streaming and are
770 : : * now waiting for instructions.
771 : : */
5029 heikki.linnakangas@i 772 : 14 : WakeupRecovery();
773 : : for (;;)
774 : : {
2138 fujii@postgresql.org 775 : 28 : ResetLatch(MyLatch);
776 : :
534 heikki.linnakangas@i 777 [ - + ]: 28 : CHECK_FOR_INTERRUPTS();
778 : :
5029 779 : 28 : SpinLockAcquire(&walrcv->mutex);
780 [ + + - + : 28 : Assert(walrcv->walRcvState == WALRCV_RESTARTING ||
- - ]
781 : : walrcv->walRcvState == WALRCV_WAITING ||
782 : : walrcv->walRcvState == WALRCV_STOPPING);
783 [ + + ]: 28 : if (walrcv->walRcvState == WALRCV_RESTARTING)
784 : : {
785 : : /*
786 : : * No need to handle changes in primary_conninfo or
787 : : * primary_slot_name here. Startup process will signal us to
788 : : * terminate in case those change.
789 : : */
790 : 14 : *startpoint = walrcv->receiveStart;
791 : 14 : *startpointTLI = walrcv->receiveStartTLI;
240 michael@paquier.xyz 792 : 14 : walrcv->walRcvState = WALRCV_CONNECTING;
5029 heikki.linnakangas@i 793 : 14 : SpinLockRelease(&walrcv->mutex);
794 : 14 : break;
795 : : }
796 [ - + ]: 14 : if (walrcv->walRcvState == WALRCV_STOPPING)
797 : : {
798 : : /*
799 : : * We should've received SIGTERM if the startup process wants us
800 : : * to die, but might as well check it here too.
801 : : */
5029 heikki.linnakangas@i 802 :UBC 0 : SpinLockRelease(&walrcv->mutex);
157 fujii@postgresql.org 803 : 0 : proc_exit(1);
804 : : }
5029 heikki.linnakangas@i 805 :CBC 14 : SpinLockRelease(&walrcv->mutex);
806 : :
2138 fujii@postgresql.org 807 : 14 : (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
808 : : WAIT_EVENT_WAL_RECEIVER_WAIT_START);
809 : : }
810 : :
5029 heikki.linnakangas@i 811 [ + - ]: 14 : if (update_process_title)
812 : : {
813 : : char activitymsg[50];
814 : :
440 alvherre@kurilemu.de 815 : 14 : snprintf(activitymsg, sizeof(activitymsg), "restarting at %X/%08X",
2035 peter@eisentraut.org 816 : 14 : LSN_FORMAT_ARGS(*startpoint));
2384 817 : 14 : set_ps_display(activitymsg);
818 : : }
5029 heikki.linnakangas@i 819 : 14 : }
820 : :
821 : : /*
822 : : * Fetch any missing timeline history files between 'first' and 'last'
823 : : * (inclusive) from the server.
824 : : */
825 : : static void
826 : 199 : WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
827 : : {
828 : : TimeLineID tli;
829 : :
830 [ + + ]: 423 : for (tli = first; tli <= last; tli++)
831 : : {
832 : : /* there's no history file for timeline 1 */
5008 833 [ + + + + ]: 224 : if (tli != 1 && !existsTimeLineHistory(tli))
834 : : {
835 : : char *fname;
836 : : char *content;
837 : : size_t len;
838 : : char expectedfname[MAXFNAMELEN];
839 : :
5029 840 [ + - ]: 13 : ereport(LOG,
841 : : (errmsg("fetching timeline history file for timeline %u from primary server",
842 : : tli)));
843 : :
3581 peter_e@gmx.net 844 : 13 : walrcv_readtimelinehistoryfile(wrconn, tli, &fname, &content, &len);
845 : :
846 : : /*
847 : : * Check that the filename on the primary matches what we
848 : : * calculated ourselves. This is just a sanity check, it should
849 : : * always match.
850 : : */
5029 heikki.linnakangas@i 851 : 13 : TLHistoryFileName(expectedfname, tli);
852 [ - + ]: 13 : if (strcmp(fname, expectedfname) != 0)
5029 heikki.linnakangas@i 853 [ # # ]:UBC 0 : ereport(ERROR,
854 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
855 : : errmsg_internal("primary reported unexpected file name for timeline history file of timeline %u",
856 : : tli)));
857 : :
858 : : /*
859 : : * Write the file to pg_wal.
860 : : */
5029 heikki.linnakangas@i 861 :CBC 13 : writeTimeLineHistoryFile(tli, content, len);
862 : :
863 : : /*
864 : : * Mark the streamed history file as ready for archiving if
865 : : * archive_mode is always.
866 : : */
2182 fujii@postgresql.org 867 [ + - ]: 13 : if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
868 : 13 : XLogArchiveForceDone(fname);
869 : : else
1842 alvherre@alvh.no-ip. 870 :UBC 0 : XLogArchiveNotify(fname);
871 : :
5029 heikki.linnakangas@i 872 :CBC 13 : pfree(fname);
873 : 13 : pfree(content);
874 : : }
875 : : }
6092 876 : 199 : }
877 : :
878 : : /*
879 : : * Mark us as STOPPED in shared memory at exit.
880 : : */
881 : : static void
6080 882 : 280 : WalRcvDie(int code, Datum arg)
883 : : {
4002 rhaas@postgresql.org 884 : 280 : WalRcvData *walrcv = WalRcv;
1780 885 : 280 : TimeLineID *startpointTLI_p = (TimeLineID *) DatumGetPointer(arg);
886 : :
887 [ - + ]: 280 : Assert(*startpointTLI_p != 0);
888 : :
889 : : /* Ensure that all WAL records received are flushed to disk */
890 : 280 : XLogWalRcvFlush(true, *startpointTLI_p);
891 : :
892 : : /* Mark ourselves inactive in shared memory */
6092 heikki.linnakangas@i 893 : 280 : SpinLockAcquire(&walrcv->mutex);
5029 894 [ + + + + : 280 : Assert(walrcv->walRcvState == WALRCV_STREAMING ||
+ - + - +
- - + ]
895 : : walrcv->walRcvState == WALRCV_CONNECTING ||
896 : : walrcv->walRcvState == WALRCV_RESTARTING ||
897 : : walrcv->walRcvState == WALRCV_STARTING ||
898 : : walrcv->walRcvState == WALRCV_WAITING ||
899 : : walrcv->walRcvState == WALRCV_STOPPING);
900 [ - + ]: 280 : Assert(walrcv->pid == MyProcPid);
6080 901 : 280 : walrcv->walRcvState = WALRCV_STOPPED;
6092 902 : 280 : walrcv->pid = 0;
688 903 : 280 : walrcv->procno = INVALID_PROC_NUMBER;
3733 alvherre@alvh.no-ip. 904 : 280 : walrcv->ready_to_display = false;
6092 heikki.linnakangas@i 905 : 280 : SpinLockRelease(&walrcv->mutex);
906 : :
2018 tmunro@postgresql.or 907 : 280 : ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
908 : :
909 : : /* Terminate the connection gracefully. */
3581 peter_e@gmx.net 910 [ + + ]: 280 : if (wrconn != NULL)
911 : 171 : walrcv_disconnect(wrconn);
912 : :
913 : : /* Wake up the startup process to notice promptly that we're gone */
5029 heikki.linnakangas@i 914 : 280 : WakeupRecovery();
6092 915 : 280 : }
916 : :
917 : : /*
918 : : * Accept the message from XLOG stream, and process it.
919 : : */
920 : : static void
1780 rhaas@postgresql.org 921 : 109701 : XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
922 : : {
923 : : int hdrlen;
924 : : XLogRecPtr dataStart;
925 : : XLogRecPtr walEnd;
926 : : TimestampTz sendTime;
927 : : bool replyRequested;
928 : :
6073 heikki.linnakangas@i 929 [ + + - ]: 109701 : switch (type)
930 : : {
410 nathan@postgresql.or 931 : 109560 : case PqReplMsg_WALData:
932 : : {
933 : : StringInfoData incoming_message;
934 : :
5065 heikki.linnakangas@i 935 : 109560 : hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
936 [ - + ]: 109560 : if (len < hdrlen)
6050 bruce@momjian.us 937 [ # # ]:UBC 0 : ereport(ERROR,
938 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
939 : : errmsg_internal("invalid WAL message received from primary")));
940 : :
941 : : /* initialize a StringInfo with the given buffer */
1048 drowley@postgresql.o 942 :CBC 109560 : initReadOnlyStringInfo(&incoming_message, buf, hdrlen);
943 : :
944 : : /* read the fields */
5065 heikki.linnakangas@i 945 : 109560 : dataStart = pq_getmsgint64(&incoming_message);
946 : 109560 : walEnd = pq_getmsgint64(&incoming_message);
3496 tgl@sss.pgh.pa.us 947 : 109560 : sendTime = pq_getmsgint64(&incoming_message);
5065 heikki.linnakangas@i 948 : 109560 : ProcessWalSndrMessage(walEnd, sendTime);
949 : :
950 : 109560 : buf += hdrlen;
951 : 109560 : len -= hdrlen;
1780 rhaas@postgresql.org 952 : 109560 : XLogWalRcvWrite(buf, len, dataStart, tli);
6050 bruce@momjian.us 953 : 109560 : break;
954 : : }
410 nathan@postgresql.or 955 : 141 : case PqReplMsg_Keepalive:
956 : : {
957 : : StringInfoData incoming_message;
958 : :
5065 heikki.linnakangas@i 959 : 141 : hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
960 [ - + ]: 141 : if (len != hdrlen)
5377 simon@2ndQuadrant.co 961 [ # # ]:UBC 0 : ereport(ERROR,
962 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
963 : : errmsg_internal("invalid keepalive message received from primary")));
964 : :
965 : : /* initialize a StringInfo with the given buffer */
1048 drowley@postgresql.o 966 :CBC 141 : initReadOnlyStringInfo(&incoming_message, buf, hdrlen);
967 : :
968 : : /* read the fields */
5065 heikki.linnakangas@i 969 : 141 : walEnd = pq_getmsgint64(&incoming_message);
3496 tgl@sss.pgh.pa.us 970 : 141 : sendTime = pq_getmsgint64(&incoming_message);
5065 heikki.linnakangas@i 971 : 141 : replyRequested = pq_getmsgbyte(&incoming_message);
972 : :
973 : 141 : ProcessWalSndrMessage(walEnd, sendTime);
974 : :
975 : : /* If the primary requested a reply, send one immediately */
976 [ + - ]: 141 : if (replyRequested)
178 fujii@postgresql.org 977 : 141 : XLogWalRcvSendReply(true, false, false);
5377 simon@2ndQuadrant.co 978 : 141 : break;
979 : : }
6073 heikki.linnakangas@i 980 :UBC 0 : default:
981 [ # # ]: 0 : ereport(ERROR,
982 : : (errcode(ERRCODE_PROTOCOL_VIOLATION),
983 : : errmsg_internal("invalid replication message type %d",
984 : : type)));
985 : : }
6073 heikki.linnakangas@i 986 :CBC 109701 : }
987 : :
988 : : /*
989 : : * Write XLOG data to disk.
990 : : */
991 : : static void
1780 rhaas@postgresql.org 992 : 109560 : XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli)
993 : : {
994 : : int startoff;
995 : : ssize_t byteswritten;
996 : : instr_time start;
997 : :
998 [ - + ]: 109560 : Assert(tli != 0);
999 : :
6092 heikki.linnakangas@i 1000 [ + + ]: 219579 : while (nbytes > 0)
1001 : : {
1002 : : int segbytes;
1003 : :
1004 : : /* Close the current segment if it's completed */
1837 fujii@postgresql.org 1005 [ + + + + ]: 110019 : if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
1780 rhaas@postgresql.org 1006 : 459 : XLogWalRcvClose(recptr, tli);
1007 : :
1837 fujii@postgresql.org 1008 [ + + ]: 110019 : if (recvFile < 0)
1009 : : {
1010 : : /* Create/use new log file */
3288 andres@anarazel.de 1011 : 911 : XLByteToSeg(recptr, recvSegNo, wal_segment_size);
1780 rhaas@postgresql.org 1012 : 911 : recvFile = XLogFileInit(recvSegNo, tli);
1013 : 911 : recvFileTLI = tli;
1014 : : }
1015 : :
1016 : : /* Calculate the start offset of the received logs */
3288 andres@anarazel.de 1017 : 110019 : startoff = XLogSegmentOffset(recptr, wal_segment_size);
1018 : :
1019 [ + + ]: 110019 : if (startoff + nbytes > wal_segment_size)
1020 : 459 : segbytes = wal_segment_size - startoff;
1021 : : else
6092 heikki.linnakangas@i 1022 : 109560 : segbytes = nbytes;
1023 : :
1024 : : /* OK to write the logs */
1025 : 110019 : errno = 0;
1026 : :
1027 : : /*
1028 : : * Measure I/O timing to write WAL data, for pg_stat_io.
1029 : : */
563 michael@paquier.xyz 1030 : 110019 : start = pgstat_prepare_io_time(track_wal_io_timing);
1031 : :
1032 : 110019 : pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE);
311 1033 : 110019 : byteswritten = pg_pwrite(recvFile, buf, segbytes, (pgoff_t) startoff);
563 1034 : 110019 : pgstat_report_wait_end();
1035 : :
6092 heikki.linnakangas@i 1036 [ - + ]: 110019 : if (byteswritten <= 0)
1037 : : {
1038 : : char xlogfname[MAXFNAMELEN];
1039 : : int save_errno;
1040 : :
1041 : : /* if write didn't set errno, assume no disk space */
6092 heikki.linnakangas@i 1042 [ # # ]:UBC 0 : if (errno == 0)
1043 : 0 : errno = ENOSPC;
1044 : :
2483 michael@paquier.xyz 1045 : 0 : save_errno = errno;
1046 : 0 : XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
1047 : 0 : errno = save_errno;
6092 heikki.linnakangas@i 1048 [ # # ]: 0 : ereport(PANIC,
1049 : : (errcode_for_file_access(),
1050 : : errmsg("could not write to WAL segment %s "
1051 : : "at offset %d, length %d: %m",
1052 : : xlogfname, startoff, segbytes)));
1053 : : }
1054 : :
95 michael@paquier.xyz 1055 :CBC 110019 : pgstat_count_io_op_time(IOOBJECT_WAL, IOCONTEXT_NORMAL,
1056 : : IOOP_WRITE, start, 1, byteswritten);
1057 : :
1058 : : /* Update state for write */
5014 alvherre@alvh.no-ip. 1059 : 110019 : recptr += byteswritten;
1060 : :
6092 heikki.linnakangas@i 1061 : 110019 : nbytes -= byteswritten;
1062 : 110019 : buf += byteswritten;
1063 : :
6050 bruce@momjian.us 1064 : 110019 : LogstreamResult.Write = recptr;
1065 : : }
1066 : :
1067 : : /* Update shared-memory status */
140 akorotkov@postgresql 1068 : 109560 : pg_atomic_write_membarrier_u64(&WalRcv->writtenUpto, LogstreamResult.Write);
1069 : :
1070 : : /*
1071 : : * Wake up processes waiting for standby write LSN to reach current write
1072 : : * position.
1073 : : */
1074 : 109560 : WaitLSNWakeup(WAIT_LSN_TYPE_STANDBY_WRITE, LogstreamResult.Write);
1075 : :
1076 : : /*
1077 : : * Close the current segment if it's fully written up in the last cycle of
1078 : : * the loop, to create its archive notification file soon. Otherwise WAL
1079 : : * archiving of the segment will be delayed until any data in the next
1080 : : * segment is received and written.
1081 : : */
1837 fujii@postgresql.org 1082 [ + - + + ]: 109560 : if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
1780 rhaas@postgresql.org 1083 : 293 : XLogWalRcvClose(recptr, tli);
6092 heikki.linnakangas@i 1084 : 109560 : }
1085 : :
1086 : : /*
1087 : : * Flush the log to disk.
1088 : : *
1089 : : * If we're in the midst of dying, it's unwise to do anything that might throw
1090 : : * an error, so we skip sending a reply in that case.
1091 : : */
1092 : : static void
1780 rhaas@postgresql.org 1093 : 54514 : XLogWalRcvFlush(bool dying, TimeLineID tli)
1094 : : {
1095 [ - + ]: 54514 : Assert(tli != 0);
1096 : :
5014 alvherre@alvh.no-ip. 1097 [ + + ]: 54514 : if (LogstreamResult.Flush < LogstreamResult.Write)
1098 : : {
4002 rhaas@postgresql.org 1099 : 53839 : WalRcvData *walrcv = WalRcv;
1100 : :
1780 1101 : 53839 : issue_xlog_fsync(recvFile, recvSegNo, tli);
1102 : :
6092 heikki.linnakangas@i 1103 : 53839 : LogstreamResult.Flush = LogstreamResult.Write;
1104 : :
1105 : : /* Update shared-memory status */
1106 : 53839 : SpinLockAcquire(&walrcv->mutex);
2356 tmunro@postgresql.or 1107 [ + - ]: 53839 : if (walrcv->flushedUpto < LogstreamResult.Flush)
1108 : : {
1109 : 53839 : walrcv->latestChunkStart = walrcv->flushedUpto;
1110 : 53839 : walrcv->flushedUpto = LogstreamResult.Flush;
1780 rhaas@postgresql.org 1111 : 53839 : walrcv->receivedTLI = tli;
1112 : : }
6092 heikki.linnakangas@i 1113 : 53839 : SpinLockRelease(&walrcv->mutex);
1114 : :
1115 : : /*
1116 : : * Wake up processes waiting for standby flush LSN to reach current
1117 : : * flush position.
1118 : : */
140 akorotkov@postgresql 1119 : 53839 : WaitLSNWakeup(WAIT_LSN_TYPE_STANDBY_FLUSH, LogstreamResult.Flush);
1120 : :
1121 : : /* Signal the startup process and walsender that new WAL has arrived */
5849 heikki.linnakangas@i 1122 : 53839 : WakeupRecovery();
5542 simon@2ndQuadrant.co 1123 [ + - + - ]: 53839 : if (AllowCascadeReplication())
1261 andres@anarazel.de 1124 : 53839 : WalSndWakeup(true, false);
1125 : :
1126 : : /* Report XLOG streaming progress in PS display */
5949 tgl@sss.pgh.pa.us 1127 [ + - ]: 53839 : if (update_process_title)
1128 : : {
1129 : : char activitymsg[50];
1130 : :
440 alvherre@kurilemu.de 1131 : 53839 : snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%08X",
2035 peter@eisentraut.org 1132 : 53839 : LSN_FORMAT_ARGS(LogstreamResult.Write));
2384 1133 : 53839 : set_ps_display(activitymsg);
1134 : : }
1135 : :
1136 : : /* Also let the primary know that we made some progress */
5695 rhaas@postgresql.org 1137 [ + + ]: 53839 : if (!dying)
1138 : : {
178 fujii@postgresql.org 1139 : 53837 : XLogWalRcvSendReply(false, false, false);
4630 heikki.linnakangas@i 1140 : 53837 : XLogWalRcvSendHSFeedback(false);
1141 : : }
1142 : : }
6092 1143 : 54514 : }
1144 : :
1145 : : /*
1146 : : * Close the current segment.
1147 : : *
1148 : : * Flush the segment to disk before closing it. Otherwise we have to
1149 : : * reopen and fsync it later.
1150 : : *
1151 : : * Create an archive notification file since the segment is known completed.
1152 : : */
1153 : : static void
1780 rhaas@postgresql.org 1154 : 752 : XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli)
1155 : : {
1156 : : char xlogfname[MAXFNAMELEN];
1157 : :
1837 fujii@postgresql.org 1158 [ + - - + ]: 752 : Assert(recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size));
1780 rhaas@postgresql.org 1159 [ - + ]: 752 : Assert(tli != 0);
1160 : :
1161 : : /*
1162 : : * fsync() and close current file before we switch to next one. We would
1163 : : * otherwise have to reopen this file to fsync it later
1164 : : */
1165 : 752 : XLogWalRcvFlush(false, tli);
1166 : :
1837 fujii@postgresql.org 1167 : 752 : XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
1168 : :
1169 : : /*
1170 : : * XLOG segment files will be re-read by recovery in startup process soon,
1171 : : * so we don't advise the OS to release cache pages associated with the
1172 : : * file like XLogFileClose() does.
1173 : : */
1174 [ - + ]: 752 : if (close(recvFile) != 0)
1837 fujii@postgresql.org 1175 [ # # ]:UBC 0 : ereport(PANIC,
1176 : : (errcode_for_file_access(),
1177 : : errmsg("could not close WAL segment %s: %m",
1178 : : xlogfname)));
1179 : :
1180 : : /*
1181 : : * Create .done file forcibly to prevent the streamed segment from being
1182 : : * archived later.
1183 : : */
1837 fujii@postgresql.org 1184 [ + - ]:CBC 752 : if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
1185 : 752 : XLogArchiveForceDone(xlogfname);
1186 : : else
1837 fujii@postgresql.org 1187 :UBC 0 : XLogArchiveNotify(xlogfname);
1188 : :
1837 fujii@postgresql.org 1189 :CBC 752 : recvFile = -1;
1190 : 752 : }
1191 : :
1192 : : /*
1193 : : * Send reply message to primary, indicating our current WAL locations and
1194 : : * time.
1195 : : *
1196 : : * The message is sent if 'force' is set, if enough time has passed since the
1197 : : * last update to reach wal_receiver_status_interval, or if WAL locations have
1198 : : * advanced since the previous status update. If wal_receiver_status_interval
1199 : : * is disabled and 'force' is false, this function does nothing. Set 'force' to
1200 : : * send the message unconditionally.
1201 : : *
1202 : : * Whether WAL locations are considered "advanced" depends on 'checkApply'.
1203 : : * If 'checkApply' is false, only the write and flush locations are checked.
1204 : : * This should be used when the call is triggered by write/flush activity
1205 : : * (e.g., after walreceiver writes or flushes WAL), and avoids the
1206 : : * apply-location check, which requires a spinlock. If 'checkApply' is true,
1207 : : * the apply location is also considered. This should be used when the apply
1208 : : * location is expected to advance (e.g., when the startup process requests
1209 : : * an apply notification).
1210 : : *
1211 : : * If 'requestReply' is true, requests the server to reply immediately upon
1212 : : * receiving this message. This is used for heartbeats, when approaching
1213 : : * wal_receiver_timeout.
1214 : : */
1215 : : static void
178 1216 : 119836 : XLogWalRcvSendReply(bool force, bool requestReply, bool checkApply)
1217 : : {
1218 : : static XLogRecPtr writePtr = InvalidXLogRecPtr;
1219 : : static XLogRecPtr flushPtr = InvalidXLogRecPtr;
1220 : : static XLogRecPtr applyPtr = InvalidXLogRecPtr;
1221 : 119836 : XLogRecPtr latestApplyPtr = InvalidXLogRecPtr;
1222 : : TimestampTz now;
1223 : :
1224 : : /*
1225 : : * If the user doesn't want status to be reported to the primary, be sure
1226 : : * to exit before doing anything at all.
1227 : : */
5092 heikki.linnakangas@i 1228 [ + + - + ]: 119836 : if (!force && wal_receiver_status_interval <= 0)
5701 heikki.linnakangas@i 1229 :UBC 0 : return;
1230 : :
1231 : : /* Get current timestamp. */
5701 heikki.linnakangas@i 1232 :CBC 119836 : now = GetCurrentTimestamp();
1233 : :
1234 : : /*
1235 : : * We can compare the write and flush positions to the last message we
1236 : : * sent without taking any lock, but the apply position requires a spin
1237 : : * lock, so we don't check that unless it is expected to advance since the
1238 : : * previous update, i.e., when 'checkApply' is true.
1239 : : */
178 fujii@postgresql.org 1240 [ + + + + ]: 119836 : if (!force && now < wakeup[WALRCV_WAKEUP_REPLY])
1241 : : {
1242 [ + + ]: 119505 : if (checkApply)
1243 : 12193 : latestApplyPtr = GetXLogReplayRecPtr(NULL);
1244 : :
1245 [ + + ]: 119505 : if (writePtr == LogstreamResult.Write
1246 [ + + ]: 65586 : && flushPtr == LogstreamResult.Flush
1247 [ + + + + ]: 12501 : && (!checkApply || applyPtr == latestApplyPtr))
1248 : 2096 : return;
1249 : : }
1250 : :
1251 : : /* Make sure we wake up when it's time to send another reply. */
1412 tmunro@postgresql.or 1252 : 117740 : WalRcvComputeNextWakeup(WALRCV_WAKEUP_REPLY, now);
1253 : :
1254 : : /* Construct a new message */
5065 heikki.linnakangas@i 1255 : 117740 : writePtr = LogstreamResult.Write;
1256 : 117740 : flushPtr = LogstreamResult.Flush;
157 fujii@postgresql.org 1257 : 117740 : applyPtr = XLogRecPtrIsValid(latestApplyPtr) ?
1258 [ + + ]: 117740 : latestApplyPtr : GetXLogReplayRecPtr(NULL);
1259 : :
5065 heikki.linnakangas@i 1260 : 117740 : resetStringInfo(&reply_message);
410 nathan@postgresql.or 1261 : 117740 : pq_sendbyte(&reply_message, PqReplMsg_StandbyStatusUpdate);
5065 heikki.linnakangas@i 1262 : 117740 : pq_sendint64(&reply_message, writePtr);
1263 : 117740 : pq_sendint64(&reply_message, flushPtr);
1264 : 117740 : pq_sendint64(&reply_message, applyPtr);
3496 tgl@sss.pgh.pa.us 1265 : 117740 : pq_sendint64(&reply_message, GetCurrentTimestamp());
5065 heikki.linnakangas@i 1266 : 117740 : pq_sendbyte(&reply_message, requestReply ? 1 : 0);
1267 : :
1268 : : /* Send it */
440 alvherre@kurilemu.de 1269 [ + + - + ]: 117740 : elog(DEBUG2, "sending write %X/%08X flush %X/%08X apply %X/%08X%s",
1270 : : LSN_FORMAT_ARGS(writePtr),
1271 : : LSN_FORMAT_ARGS(flushPtr),
1272 : : LSN_FORMAT_ARGS(applyPtr),
1273 : : requestReply ? " (reply requested)" : "");
1274 : :
3581 peter_e@gmx.net 1275 : 117740 : walrcv_send(wrconn, reply_message.data, reply_message.len);
1276 : : }
1277 : :
1278 : : /*
1279 : : * Send hot standby feedback message to primary, plus the current time,
1280 : : * in case they don't have a watch.
1281 : : *
1282 : : * If the user disables feedback, send one final message to tell sender
1283 : : * to forget about the xmin on this standby. We also send this message
1284 : : * on first connect because a previous connection might have set xmin
1285 : : * on a replication slot. (If we're not using a slot it's harmless to
1286 : : * send a feedback message explicitly setting InvalidTransactionId).
1287 : : */
1288 : : static void
4976 simon@2ndQuadrant.co 1289 : 54063 : XLogWalRcvSendHSFeedback(bool immed)
1290 : : {
1291 : : TimestampTz now;
1292 : : FullTransactionId nextFullXid;
1293 : : TransactionId nextXid;
1294 : : uint32 xmin_epoch,
1295 : : catalog_xmin_epoch;
1296 : : TransactionId xmin,
1297 : : catalog_xmin;
1298 : :
1299 : : /* initially true so we always send at least one feedback message */
1300 : : static bool primary_has_standby_xmin = true;
1301 : :
1302 : : /*
1303 : : * If the user doesn't want status to be reported to the primary, be sure
1304 : : * to exit before doing anything at all.
1305 : : */
1306 [ + - + + ]: 54063 : if ((wal_receiver_status_interval <= 0 || !hot_standby_feedback) &&
2289 andres@anarazel.de 1307 [ + + ]: 53497 : !primary_has_standby_xmin)
5693 simon@2ndQuadrant.co 1308 : 53870 : return;
1309 : :
1310 : : /* Get current timestamp. */
1311 : 719 : now = GetCurrentTimestamp();
1312 : :
1313 : : /* Send feedback at most once per wal_receiver_status_interval. */
1412 tmunro@postgresql.or 1314 [ + + + + ]: 719 : if (!immed && now < wakeup[WALRCV_WAKEUP_HSFEEDBACK])
1315 : 524 : return;
1316 : :
1317 : : /* Make sure we wake up when it's time to send feedback again. */
1318 : 195 : WalRcvComputeNextWakeup(WALRCV_WAKEUP_HSFEEDBACK, now);
1319 : :
1320 : : /*
1321 : : * If Hot Standby is not yet accepting connections there is nothing to
1322 : : * send. Check this after the interval has expired to reduce number of
1323 : : * calls.
1324 : : *
1325 : : * Bailing out here also ensures that we don't send feedback until we've
1326 : : * read our own replication slot state, so we don't tell the primary to
1327 : : * discard needed xmin or catalog_xmin from any slots that may exist on
1328 : : * this replica.
1329 : : */
5693 simon@2ndQuadrant.co 1330 [ + + ]: 195 : if (!HotStandbyActive())
1331 : 2 : return;
1332 : :
1333 : : /*
1334 : : * Make the expensive call to get the oldest xmin once we are certain
1335 : : * everything else has been checked.
1336 : : */
4976 1337 [ + + ]: 193 : if (hot_standby_feedback)
1338 : : {
2230 andres@anarazel.de 1339 : 54 : GetReplicationHorizons(&xmin, &catalog_xmin);
1340 : : }
1341 : : else
1342 : : {
4976 simon@2ndQuadrant.co 1343 : 139 : xmin = InvalidTransactionId;
3466 1344 : 139 : catalog_xmin = InvalidTransactionId;
1345 : : }
1346 : :
1347 : : /*
1348 : : * Get epoch and adjust if nextXid and oldestXmin are different sides of
1349 : : * the epoch boundary.
1350 : : */
2733 tmunro@postgresql.or 1351 : 193 : nextFullXid = ReadNextFullTransactionId();
1352 : 193 : nextXid = XidFromFullTransactionId(nextFullXid);
1353 : 193 : xmin_epoch = EpochFromFullTransactionId(nextFullXid);
3466 simon@2ndQuadrant.co 1354 : 193 : catalog_xmin_epoch = xmin_epoch;
5693 1355 [ - + ]: 193 : if (nextXid < xmin)
3413 bruce@momjian.us 1356 :UBC 0 : xmin_epoch--;
3466 simon@2ndQuadrant.co 1357 [ - + ]:CBC 193 : if (nextXid < catalog_xmin)
3413 bruce@momjian.us 1358 :UBC 0 : catalog_xmin_epoch--;
1359 : :
3466 simon@2ndQuadrant.co 1360 [ + + ]:CBC 193 : elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u catalog_xmin %u catalog_xmin_epoch %u",
1361 : : xmin, xmin_epoch, catalog_xmin, catalog_xmin_epoch);
1362 : :
1363 : : /* Construct the message and send it. */
5065 heikki.linnakangas@i 1364 : 193 : resetStringInfo(&reply_message);
410 nathan@postgresql.or 1365 : 193 : pq_sendbyte(&reply_message, PqReplMsg_HotStandbyFeedback);
3496 tgl@sss.pgh.pa.us 1366 : 193 : pq_sendint64(&reply_message, GetCurrentTimestamp());
3266 andres@anarazel.de 1367 : 193 : pq_sendint32(&reply_message, xmin);
1368 : 193 : pq_sendint32(&reply_message, xmin_epoch);
1369 : 193 : pq_sendint32(&reply_message, catalog_xmin);
1370 : 193 : pq_sendint32(&reply_message, catalog_xmin_epoch);
3581 peter_e@gmx.net 1371 : 193 : walrcv_send(wrconn, reply_message.data, reply_message.len);
3466 simon@2ndQuadrant.co 1372 [ + + - + ]: 193 : if (TransactionIdIsValid(xmin) || TransactionIdIsValid(catalog_xmin))
2289 andres@anarazel.de 1373 : 54 : primary_has_standby_xmin = true;
1374 : : else
1375 : 139 : primary_has_standby_xmin = false;
1376 : : }
1377 : :
1378 : : /*
1379 : : * Update shared memory status upon receiving a message from primary.
1380 : : *
1381 : : * 'walEnd' and 'sendTime' are the end-of-WAL and timestamp of the latest
1382 : : * message, reported by primary.
1383 : : */
1384 : : static void
5377 simon@2ndQuadrant.co 1385 : 109701 : ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
1386 : : {
4002 rhaas@postgresql.org 1387 : 109701 : WalRcvData *walrcv = WalRcv;
5377 simon@2ndQuadrant.co 1388 : 109701 : TimestampTz lastMsgReceiptTime = GetCurrentTimestamp();
1389 : :
1390 : : /* Update shared-memory status */
1391 : 109701 : SpinLockAcquire(&walrcv->mutex);
5014 alvherre@alvh.no-ip. 1392 [ + + ]: 109701 : if (walrcv->latestWalEnd < walEnd)
5155 simon@2ndQuadrant.co 1393 : 25921 : walrcv->latestWalEndTime = sendTime;
1394 : 109701 : walrcv->latestWalEnd = walEnd;
5377 1395 : 109701 : walrcv->lastMsgSendTime = sendTime;
1396 : 109701 : walrcv->lastMsgReceiptTime = lastMsgReceiptTime;
1397 : 109701 : SpinLockRelease(&walrcv->mutex);
1398 : :
2127 tgl@sss.pgh.pa.us 1399 [ + + ]: 109701 : if (message_level_is_interesting(DEBUG2))
1400 : : {
1401 : : char *sendtime;
1402 : : char *receipttime;
1403 : : int applyDelay;
1404 : :
1405 : : /* Copy because timestamptz_to_str returns a static buffer */
4552 1406 : 418 : sendtime = pstrdup(timestamptz_to_str(sendTime));
1407 : 418 : receipttime = pstrdup(timestamptz_to_str(lastMsgReceiptTime));
4208 ishii@postgresql.org 1408 : 418 : applyDelay = GetReplicationApplyDelay();
1409 : :
1410 : : /* apply delay is not available */
1411 [ + + ]: 418 : if (applyDelay == -1)
1412 [ + - ]: 5 : elog(DEBUG2, "sendtime %s receipttime %s replication apply delay (N/A) transfer latency %d ms",
1413 : : sendtime,
1414 : : receipttime,
1415 : : GetReplicationTransferLatency());
1416 : : else
1417 [ + - ]: 413 : elog(DEBUG2, "sendtime %s receipttime %s replication apply delay %d ms transfer latency %d ms",
1418 : : sendtime,
1419 : : receipttime,
1420 : : applyDelay,
1421 : : GetReplicationTransferLatency());
1422 : :
4552 tgl@sss.pgh.pa.us 1423 : 418 : pfree(sendtime);
1424 : 418 : pfree(receipttime);
1425 : : }
5377 simon@2ndQuadrant.co 1426 : 109701 : }
1427 : :
1428 : : /*
1429 : : * Compute the next wakeup time for a given wakeup reason. Can be called to
1430 : : * initialize a wakeup time, to adjust it for the next wakeup, or to
1431 : : * reinitialize it when GUCs have changed. We ask the caller to pass in the
1432 : : * value of "now" because this frequently avoids multiple calls of
1433 : : * GetCurrentTimestamp(). It had better be a reasonably up-to-date value
1434 : : * though.
1435 : : */
1436 : : static void
1412 tmunro@postgresql.or 1437 : 338193 : WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now)
1438 : : {
1439 [ + + + + : 338193 : switch (reason)
- ]
1440 : : {
1441 : 109915 : case WALRCV_WAKEUP_TERMINATE:
1442 [ - + ]: 109915 : if (wal_receiver_timeout <= 0)
1333 tgl@sss.pgh.pa.us 1443 :UBC 0 : wakeup[reason] = TIMESTAMP_INFINITY;
1444 : : else
1333 tgl@sss.pgh.pa.us 1445 :CBC 109915 : wakeup[reason] = TimestampTzPlusMilliseconds(now, wal_receiver_timeout);
1412 tmunro@postgresql.or 1446 : 109915 : break;
1447 : 109915 : case WALRCV_WAKEUP_PING:
1448 [ - + ]: 109915 : if (wal_receiver_timeout <= 0)
1333 tgl@sss.pgh.pa.us 1449 :UBC 0 : wakeup[reason] = TIMESTAMP_INFINITY;
1450 : : else
1333 tgl@sss.pgh.pa.us 1451 :CBC 109915 : wakeup[reason] = TimestampTzPlusMilliseconds(now, wal_receiver_timeout / 2);
1412 tmunro@postgresql.or 1452 : 109915 : break;
1453 : 409 : case WALRCV_WAKEUP_HSFEEDBACK:
1454 [ + + - + ]: 409 : if (!hot_standby_feedback || wal_receiver_status_interval <= 0)
1333 tgl@sss.pgh.pa.us 1455 : 306 : wakeup[reason] = TIMESTAMP_INFINITY;
1456 : : else
1457 : 103 : wakeup[reason] = TimestampTzPlusSeconds(now, wal_receiver_status_interval);
1412 tmunro@postgresql.or 1458 : 409 : break;
1459 : 117954 : case WALRCV_WAKEUP_REPLY:
1460 [ - + ]: 117954 : if (wal_receiver_status_interval <= 0)
1333 tgl@sss.pgh.pa.us 1461 :UBC 0 : wakeup[reason] = TIMESTAMP_INFINITY;
1462 : : else
1333 tgl@sss.pgh.pa.us 1463 :CBC 117954 : wakeup[reason] = TimestampTzPlusSeconds(now, wal_receiver_status_interval);
1412 tmunro@postgresql.or 1464 : 117954 : break;
1465 : : /* there's intentionally no default: here */
1466 : : }
1467 : 338193 : }
1468 : :
1469 : : /*
1470 : : * Wake up the walreceiver main loop.
1471 : : *
1472 : : * This is called by the startup process whenever interesting xlog records
1473 : : * are applied, so that walreceiver can check if it needs to send an apply
1474 : : * notification back to the primary which may be waiting in a COMMIT with
1475 : : * synchronous_commit = remote_apply.
1476 : : */
1477 : : void
178 fujii@postgresql.org 1478 : 11966 : WalRcvRequestApplyReply(void)
1479 : : {
1480 : : ProcNumber procno;
1481 : :
1482 : 11966 : WalRcv->apply_reply_requested = true;
1483 : : /* fetching the proc number is probably atomic, but don't rely on it */
3274 tgl@sss.pgh.pa.us 1484 : 11966 : SpinLockAcquire(&WalRcv->mutex);
688 heikki.linnakangas@i 1485 : 11966 : procno = WalRcv->procno;
3274 tgl@sss.pgh.pa.us 1486 : 11966 : SpinLockRelease(&WalRcv->mutex);
688 heikki.linnakangas@i 1487 [ + + ]: 11966 : if (procno != INVALID_PROC_NUMBER)
1488 : 11752 : SetLatch(&GetPGProcByNumber(procno)->procLatch);
3827 rhaas@postgresql.org 1489 : 11966 : }
1490 : :
1491 : : /*
1492 : : * Return a string constant representing the state. This is used
1493 : : * in system functions and views, and should *not* be translated.
1494 : : */
1495 : : static const char *
3909 alvherre@alvh.no-ip. 1496 : 19 : WalRcvGetStateString(WalRcvState state)
1497 : : {
1498 [ - - + + : 19 : switch (state)
- - - - ]
1499 : : {
3909 alvherre@alvh.no-ip. 1500 :UBC 0 : case WALRCV_STOPPED:
1501 : 0 : return "stopped";
1502 : 0 : case WALRCV_STARTING:
1503 : 0 : return "starting";
240 michael@paquier.xyz 1504 :CBC 2 : case WALRCV_CONNECTING:
1505 : 2 : return "connecting";
3909 alvherre@alvh.no-ip. 1506 : 17 : case WALRCV_STREAMING:
1507 : 17 : return "streaming";
3909 alvherre@alvh.no-ip. 1508 :UBC 0 : case WALRCV_WAITING:
1509 : 0 : return "waiting";
1510 : 0 : case WALRCV_RESTARTING:
1511 : 0 : return "restarting";
1512 : 0 : case WALRCV_STOPPING:
1513 : 0 : return "stopping";
1514 : : }
1515 : 0 : return "UNKNOWN";
1516 : : }
1517 : :
1518 : : /*
1519 : : * Returns activity of WAL receiver, including pid, state and xlog locations
1520 : : * received from the WAL sender of another server.
1521 : : */
1522 : : Datum
3909 alvherre@alvh.no-ip. 1523 :CBC 32 : pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
1524 : : {
1525 : : TupleDesc tupdesc;
1526 : : Datum *values;
1527 : : bool *nulls;
1528 : : int pid;
1529 : : bool ready_to_display;
1530 : : WalRcvState state;
1531 : : XLogRecPtr receive_start_lsn;
1532 : : TimeLineID receive_start_tli;
1533 : : XLogRecPtr written_lsn;
1534 : : XLogRecPtr flushed_lsn;
1535 : : TimeLineID received_tli;
1536 : : TimestampTz last_send_time;
1537 : : TimestampTz last_receipt_time;
1538 : : XLogRecPtr latest_end_lsn;
1539 : : TimestampTz latest_end_time;
1540 : : char sender_host[NI_MAXHOST];
3095 fujii@postgresql.org 1541 : 32 : int sender_port = 0;
1542 : : char slotname[NAMEDATALEN];
1543 : : char conninfo[MAXCONNINFO];
1544 : :
1545 : : /* Take a lock to ensure value consistency */
3369 alvherre@alvh.no-ip. 1546 : 32 : SpinLockAcquire(&WalRcv->mutex);
1547 : 32 : pid = (int) WalRcv->pid;
1548 : 32 : ready_to_display = WalRcv->ready_to_display;
1549 : 32 : state = WalRcv->walRcvState;
1550 : 32 : receive_start_lsn = WalRcv->receiveStart;
1551 : 32 : receive_start_tli = WalRcv->receiveStartTLI;
2317 michael@paquier.xyz 1552 : 32 : flushed_lsn = WalRcv->flushedUpto;
3369 alvherre@alvh.no-ip. 1553 : 32 : received_tli = WalRcv->receivedTLI;
1554 : 32 : last_send_time = WalRcv->lastMsgSendTime;
1555 : 32 : last_receipt_time = WalRcv->lastMsgReceiptTime;
1556 : 32 : latest_end_lsn = WalRcv->latestWalEnd;
1557 : 32 : latest_end_time = WalRcv->latestWalEndTime;
585 peter@eisentraut.org 1558 : 32 : strlcpy(slotname, WalRcv->slotname, sizeof(slotname));
1559 : 32 : strlcpy(sender_host, WalRcv->sender_host, sizeof(sender_host));
3095 fujii@postgresql.org 1560 : 32 : sender_port = WalRcv->sender_port;
585 peter@eisentraut.org 1561 : 32 : strlcpy(conninfo, WalRcv->conninfo, sizeof(conninfo));
3369 alvherre@alvh.no-ip. 1562 : 32 : SpinLockRelease(&WalRcv->mutex);
1563 : :
1564 : : /*
1565 : : * No WAL receiver (or not ready yet), just return a tuple with NULL
1566 : : * values
1567 : : */
1568 [ + + - + ]: 32 : if (pid == 0 || !ready_to_display)
3733 1569 : 13 : PG_RETURN_NULL();
1570 : :
1571 : : /*
1572 : : * Read "writtenUpto" without holding a spinlock. Note that it may not be
1573 : : * consistent with the other shared variables of the WAL receiver
1574 : : * protected by a spinlock, but this should not be used for data integrity
1575 : : * checks.
1576 : : */
2040 fujii@postgresql.org 1577 : 19 : written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);
1578 : :
1579 : : /* determine result type */
3735 alvherre@alvh.no-ip. 1580 [ - + ]: 19 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
3735 alvherre@alvh.no-ip. 1581 [ # # ]:UBC 0 : elog(ERROR, "return type must be a row type");
1582 : :
284 michael@paquier.xyz 1583 :CBC 19 : values = palloc0_array(Datum, tupdesc->natts);
1584 : 19 : nulls = palloc0_array(bool, tupdesc->natts);
1585 : :
1586 : : /* Fetch values */
3369 alvherre@alvh.no-ip. 1587 : 19 : values[0] = Int32GetDatum(pid);
1588 : :
1637 mail@joeconway.com 1589 [ - + ]: 19 : if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
1590 : : {
1591 : : /*
1592 : : * Only superusers and roles with privileges of pg_read_all_stats can
1593 : : * see details. Other users only get the pid value to know whether it
1594 : : * is a WAL receiver, but no details.
1595 : : */
1542 peter@eisentraut.org 1596 :UBC 0 : memset(&nulls[1], true, sizeof(bool) * (tupdesc->natts - 1));
1597 : : }
1598 : : else
1599 : : {
3909 alvherre@alvh.no-ip. 1600 :CBC 19 : values[1] = CStringGetTextDatum(WalRcvGetStateString(state));
1601 : :
318 alvherre@kurilemu.de 1602 [ - + ]: 19 : if (!XLogRecPtrIsValid(receive_start_lsn))
3909 alvherre@alvh.no-ip. 1603 :UBC 0 : nulls[2] = true;
1604 : : else
3909 alvherre@alvh.no-ip. 1605 :CBC 19 : values[2] = LSNGetDatum(receive_start_lsn);
1606 : 19 : values[3] = Int32GetDatum(receive_start_tli);
318 alvherre@kurilemu.de 1607 [ - + ]: 19 : if (!XLogRecPtrIsValid(written_lsn))
3909 alvherre@alvh.no-ip. 1608 :UBC 0 : nulls[4] = true;
1609 : : else
2317 michael@paquier.xyz 1610 :CBC 19 : values[4] = LSNGetDatum(written_lsn);
318 alvherre@kurilemu.de 1611 [ - + ]: 19 : if (!XLogRecPtrIsValid(flushed_lsn))
2317 michael@paquier.xyz 1612 :UBC 0 : nulls[5] = true;
1613 : : else
2317 michael@paquier.xyz 1614 :CBC 19 : values[5] = LSNGetDatum(flushed_lsn);
1615 : 19 : values[6] = Int32GetDatum(received_tli);
3909 alvherre@alvh.no-ip. 1616 [ - + ]: 19 : if (last_send_time == 0)
2317 michael@paquier.xyz 1617 :UBC 0 : nulls[7] = true;
1618 : : else
2317 michael@paquier.xyz 1619 :CBC 19 : values[7] = TimestampTzGetDatum(last_send_time);
3909 alvherre@alvh.no-ip. 1620 [ - + ]: 19 : if (last_receipt_time == 0)
2317 michael@paquier.xyz 1621 :UBC 0 : nulls[8] = true;
1622 : : else
2317 michael@paquier.xyz 1623 :CBC 19 : values[8] = TimestampTzGetDatum(last_receipt_time);
318 alvherre@kurilemu.de 1624 [ - + ]: 19 : if (!XLogRecPtrIsValid(latest_end_lsn))
2317 michael@paquier.xyz 1625 :UBC 0 : nulls[9] = true;
1626 : : else
2317 michael@paquier.xyz 1627 :CBC 19 : values[9] = LSNGetDatum(latest_end_lsn);
3909 alvherre@alvh.no-ip. 1628 [ - + ]: 19 : if (latest_end_time == 0)
2317 michael@paquier.xyz 1629 :UBC 0 : nulls[10] = true;
1630 : : else
2317 michael@paquier.xyz 1631 :CBC 19 : values[10] = TimestampTzGetDatum(latest_end_time);
3909 alvherre@alvh.no-ip. 1632 [ + + ]: 19 : if (*slotname == '\0')
2317 michael@paquier.xyz 1633 : 16 : nulls[11] = true;
1634 : : else
1635 : 3 : values[11] = CStringGetTextDatum(slotname);
3095 fujii@postgresql.org 1636 [ + + ]: 19 : if (*sender_host == '\0')
2317 michael@paquier.xyz 1637 : 2 : nulls[12] = true;
1638 : : else
1639 : 17 : values[12] = CStringGetTextDatum(sender_host);
3095 fujii@postgresql.org 1640 [ + + ]: 19 : if (sender_port == 0)
2317 michael@paquier.xyz 1641 : 2 : nulls[13] = true;
1642 : : else
1643 : 17 : values[13] = Int32GetDatum(sender_port);
3095 fujii@postgresql.org 1644 [ + + ]: 19 : if (*conninfo == '\0')
2317 michael@paquier.xyz 1645 : 2 : nulls[14] = true;
1646 : : else
1647 : 17 : values[14] = CStringGetTextDatum(conninfo);
1648 : : }
1649 : :
1650 : : /* Returns the record as Datum */
3369 alvherre@alvh.no-ip. 1651 : 19 : PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
1652 : : }
|
