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