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