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