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