Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * walsender.c
4 : *
5 : * The WAL sender process (walsender) is new as of Postgres 9.0. It takes
6 : * care of sending XLOG from the primary server to a single recipient.
7 : * (Note that there can be more than one walsender process concurrently.)
8 : * It is started by the postmaster when the walreceiver of a standby server
9 : * connects to the primary server and requests XLOG streaming replication.
10 : *
11 : * A walsender is similar to a regular backend, ie. there is a one-to-one
12 : * relationship between a connection and a walsender process, but instead
13 : * of processing SQL queries, it understands a small set of special
14 : * replication-mode commands. The START_REPLICATION command begins streaming
15 : * WAL to the client. While streaming, the walsender keeps reading XLOG
16 : * records from the disk and sends them to the standby server over the
17 : * COPY protocol, until either side ends the replication by exiting COPY
18 : * mode (or until the connection is closed).
19 : *
20 : * Normal termination is by SIGTERM, which instructs the walsender to
21 : * close the connection and exit(0) at the next convenient moment. Emergency
22 : * termination is by SIGQUIT; like any backend, the walsender will simply
23 : * abort and exit on SIGQUIT. A close of the connection and a FATAL error
24 : * are treated as not a crash but approximately normal termination;
25 : * the walsender will exit quickly without sending any more XLOG records.
26 : *
27 : * If the server is shut down, checkpointer sends us
28 : * PROCSIG_WALSND_INIT_STOPPING after all regular backends have exited. If
29 : * the backend is idle or runs an SQL query this causes the backend to
30 : * shutdown, if logical replication is in progress all existing WAL records
31 : * are processed followed by a shutdown. Otherwise this causes the walsender
32 : * to switch to the "stopping" state. In this state, the walsender will reject
33 : * any further replication commands. The checkpointer begins the shutdown
34 : * checkpoint once all walsenders are confirmed as stopping. When the shutdown
35 : * checkpoint finishes, the postmaster sends us SIGUSR2. This instructs
36 : * walsender to send any outstanding WAL, including the shutdown checkpoint
37 : * record, wait for it to be replicated to the standby, and then exit.
38 : *
39 : *
40 : * Portions Copyright (c) 2010-2025, PostgreSQL Global Development Group
41 : *
42 : * IDENTIFICATION
43 : * src/backend/replication/walsender.c
44 : *
45 : *-------------------------------------------------------------------------
46 : */
47 : #include "postgres.h"
48 :
49 : #include <signal.h>
50 : #include <unistd.h>
51 :
52 : #include "access/timeline.h"
53 : #include "access/transam.h"
54 : #include "access/xact.h"
55 : #include "access/xlog_internal.h"
56 : #include "access/xlogreader.h"
57 : #include "access/xlogrecovery.h"
58 : #include "access/xlogutils.h"
59 : #include "backup/basebackup.h"
60 : #include "backup/basebackup_incremental.h"
61 : #include "catalog/pg_authid.h"
62 : #include "catalog/pg_type.h"
63 : #include "commands/dbcommands.h"
64 : #include "commands/defrem.h"
65 : #include "funcapi.h"
66 : #include "libpq/libpq.h"
67 : #include "libpq/pqformat.h"
68 : #include "miscadmin.h"
69 : #include "nodes/replnodes.h"
70 : #include "pgstat.h"
71 : #include "postmaster/interrupt.h"
72 : #include "replication/decode.h"
73 : #include "replication/logical.h"
74 : #include "replication/slotsync.h"
75 : #include "replication/slot.h"
76 : #include "replication/snapbuild.h"
77 : #include "replication/syncrep.h"
78 : #include "replication/walreceiver.h"
79 : #include "replication/walsender.h"
80 : #include "replication/walsender_private.h"
81 : #include "storage/condition_variable.h"
82 : #include "storage/fd.h"
83 : #include "storage/ipc.h"
84 : #include "storage/pmsignal.h"
85 : #include "storage/proc.h"
86 : #include "tcop/dest.h"
87 : #include "tcop/tcopprot.h"
88 : #include "utils/acl.h"
89 : #include "utils/builtins.h"
90 : #include "utils/guc.h"
91 : #include "utils/memutils.h"
92 : #include "utils/pg_lsn.h"
93 : #include "utils/ps_status.h"
94 : #include "utils/timeout.h"
95 : #include "utils/timestamp.h"
96 :
97 : /*
98 : * Maximum data payload in a WAL data message. Must be >= XLOG_BLCKSZ.
99 : *
100 : * We don't have a good idea of what a good value would be; there's some
101 : * overhead per message in both walsender and walreceiver, but on the other
102 : * hand sending large batches makes walsender less responsive to signals
103 : * because signals are checked only between messages. 128kB (with
104 : * default 8k blocks) seems like a reasonable guess for now.
105 : */
106 : #define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
107 :
108 : /* Array of WalSnds in shared memory */
109 : WalSndCtlData *WalSndCtl = NULL;
110 :
111 : /* My slot in the shared memory array */
112 : WalSnd *MyWalSnd = NULL;
113 :
114 : /* Global state */
115 : bool am_walsender = false; /* Am I a walsender process? */
116 : bool am_cascading_walsender = false; /* Am I cascading WAL to another
117 : * standby? */
118 : bool am_db_walsender = false; /* Connected to a database? */
119 :
120 : /* GUC variables */
121 : int max_wal_senders = 10; /* the maximum number of concurrent
122 : * walsenders */
123 : int wal_sender_timeout = 60 * 1000; /* maximum time to send one WAL
124 : * data message */
125 : bool log_replication_commands = false;
126 :
127 : /*
128 : * State for WalSndWakeupRequest
129 : */
130 : bool wake_wal_senders = false;
131 :
132 : /*
133 : * xlogreader used for replication. Note that a WAL sender doing physical
134 : * replication does not need xlogreader to read WAL, but it needs one to
135 : * keep a state of its work.
136 : */
137 : static XLogReaderState *xlogreader = NULL;
138 :
139 : /*
140 : * If the UPLOAD_MANIFEST command is used to provide a backup manifest in
141 : * preparation for an incremental backup, uploaded_manifest will be point
142 : * to an object containing information about its contexts, and
143 : * uploaded_manifest_mcxt will point to the memory context that contains
144 : * that object and all of its subordinate data. Otherwise, both values will
145 : * be NULL.
146 : */
147 : static IncrementalBackupInfo *uploaded_manifest = NULL;
148 : static MemoryContext uploaded_manifest_mcxt = NULL;
149 :
150 : /*
151 : * These variables keep track of the state of the timeline we're currently
152 : * sending. sendTimeLine identifies the timeline. If sendTimeLineIsHistoric,
153 : * the timeline is not the latest timeline on this server, and the server's
154 : * history forked off from that timeline at sendTimeLineValidUpto.
155 : */
156 : static TimeLineID sendTimeLine = 0;
157 : static TimeLineID sendTimeLineNextTLI = 0;
158 : static bool sendTimeLineIsHistoric = false;
159 : static XLogRecPtr sendTimeLineValidUpto = InvalidXLogRecPtr;
160 :
161 : /*
162 : * How far have we sent WAL already? This is also advertised in
163 : * MyWalSnd->sentPtr. (Actually, this is the next WAL location to send.)
164 : */
165 : static XLogRecPtr sentPtr = InvalidXLogRecPtr;
166 :
167 : /* Buffers for constructing outgoing messages and processing reply messages. */
168 : static StringInfoData output_message;
169 : static StringInfoData reply_message;
170 : static StringInfoData tmpbuf;
171 :
172 : /* Timestamp of last ProcessRepliesIfAny(). */
173 : static TimestampTz last_processing = 0;
174 :
175 : /*
176 : * Timestamp of last ProcessRepliesIfAny() that saw a reply from the
177 : * standby. Set to 0 if wal_sender_timeout doesn't need to be active.
178 : */
179 : static TimestampTz last_reply_timestamp = 0;
180 :
181 : /* Have we sent a heartbeat message asking for reply, since last reply? */
182 : static bool waiting_for_ping_response = false;
183 :
184 : /*
185 : * While streaming WAL in Copy mode, streamingDoneSending is set to true
186 : * after we have sent CopyDone. We should not send any more CopyData messages
187 : * after that. streamingDoneReceiving is set to true when we receive CopyDone
188 : * from the other end. When both become true, it's time to exit Copy mode.
189 : */
190 : static bool streamingDoneSending;
191 : static bool streamingDoneReceiving;
192 :
193 : /* Are we there yet? */
194 : static bool WalSndCaughtUp = false;
195 :
196 : /* Flags set by signal handlers for later service in main loop */
197 : static volatile sig_atomic_t got_SIGUSR2 = false;
198 : static volatile sig_atomic_t got_STOPPING = false;
199 :
200 : /*
201 : * This is set while we are streaming. When not set
202 : * PROCSIG_WALSND_INIT_STOPPING signal will be handled like SIGTERM. When set,
203 : * the main loop is responsible for checking got_STOPPING and terminating when
204 : * it's set (after streaming any remaining WAL).
205 : */
206 : static volatile sig_atomic_t replication_active = false;
207 :
208 : static LogicalDecodingContext *logical_decoding_ctx = NULL;
209 :
210 : /* A sample associating a WAL location with the time it was written. */
211 : typedef struct
212 : {
213 : XLogRecPtr lsn;
214 : TimestampTz time;
215 : } WalTimeSample;
216 :
217 : /* The size of our buffer of time samples. */
218 : #define LAG_TRACKER_BUFFER_SIZE 8192
219 :
220 : /* A mechanism for tracking replication lag. */
221 : typedef struct
222 : {
223 : XLogRecPtr last_lsn;
224 : WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE];
225 : int write_head;
226 : int read_heads[NUM_SYNC_REP_WAIT_MODE];
227 : WalTimeSample last_read[NUM_SYNC_REP_WAIT_MODE];
228 : } LagTracker;
229 :
230 : static LagTracker *lag_tracker;
231 :
232 : /* Signal handlers */
233 : static void WalSndLastCycleHandler(SIGNAL_ARGS);
234 :
235 : /* Prototypes for private functions */
236 : typedef void (*WalSndSendDataCallback) (void);
237 : static void WalSndLoop(WalSndSendDataCallback send_data);
238 : static void InitWalSenderSlot(void);
239 : static void WalSndKill(int code, Datum arg);
240 : static void WalSndShutdown(void) pg_attribute_noreturn();
241 : static void XLogSendPhysical(void);
242 : static void XLogSendLogical(void);
243 : static void WalSndDone(WalSndSendDataCallback send_data);
244 : static void IdentifySystem(void);
245 : static void UploadManifest(void);
246 : static bool HandleUploadManifestPacket(StringInfo buf, off_t *offset,
247 : IncrementalBackupInfo *ib);
248 : static void ReadReplicationSlot(ReadReplicationSlotCmd *cmd);
249 : static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd);
250 : static void DropReplicationSlot(DropReplicationSlotCmd *cmd);
251 : static void StartReplication(StartReplicationCmd *cmd);
252 : static void StartLogicalReplication(StartReplicationCmd *cmd);
253 : static void ProcessStandbyMessage(void);
254 : static void ProcessStandbyReplyMessage(void);
255 : static void ProcessStandbyHSFeedbackMessage(void);
256 : static void ProcessRepliesIfAny(void);
257 : static void ProcessPendingWrites(void);
258 : static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr);
259 : static void WalSndKeepaliveIfNecessary(void);
260 : static void WalSndCheckTimeOut(void);
261 : static long WalSndComputeSleeptime(TimestampTz now);
262 : static void WalSndWait(uint32 socket_events, long timeout, uint32 wait_event);
263 : static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);
264 : static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);
265 : static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
266 : bool skipped_xact);
267 : static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);
268 : static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time);
269 : static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now);
270 : static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch);
271 :
272 : static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo,
273 : TimeLineID *tli_p);
274 :
275 :
276 : /* Initialize walsender process before entering the main command loop */
277 : void
278 2122 : InitWalSender(void)
279 : {
280 2122 : am_cascading_walsender = RecoveryInProgress();
281 :
282 : /* Create a per-walsender data structure in shared memory */
283 2122 : InitWalSenderSlot();
284 :
285 : /* need resource owner for e.g. basebackups */
286 2122 : CreateAuxProcessResourceOwner();
287 :
288 : /*
289 : * Let postmaster know that we're a WAL sender. Once we've declared us as
290 : * a WAL sender process, postmaster will let us outlive the bgwriter and
291 : * kill us last in the shutdown sequence, so we get a chance to stream all
292 : * remaining WAL at shutdown, including the shutdown checkpoint. Note that
293 : * there's no going back, and we mustn't write any WAL records after this.
294 : */
295 2122 : MarkPostmasterChildWalSender();
296 2122 : SendPostmasterSignal(PMSIGNAL_ADVANCE_STATE_MACHINE);
297 :
298 : /*
299 : * If the client didn't specify a database to connect to, show in PGPROC
300 : * that our advertised xmin should affect vacuum horizons in all
301 : * databases. This allows physical replication clients to send hot
302 : * standby feedback that will delay vacuum cleanup in all databases.
303 : */
304 2122 : if (MyDatabaseId == InvalidOid)
305 : {
306 : Assert(MyProc->xmin == InvalidTransactionId);
307 892 : LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
308 892 : MyProc->statusFlags |= PROC_AFFECTS_ALL_HORIZONS;
309 892 : ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
310 892 : LWLockRelease(ProcArrayLock);
311 : }
312 :
313 : /* Initialize empty timestamp buffer for lag tracking. */
314 2122 : lag_tracker = MemoryContextAllocZero(TopMemoryContext, sizeof(LagTracker));
315 2122 : }
316 :
317 : /*
318 : * Clean up after an error.
319 : *
320 : * WAL sender processes don't use transactions like regular backends do.
321 : * This function does any cleanup required after an error in a WAL sender
322 : * process, similar to what transaction abort does in a regular backend.
323 : */
324 : void
325 100 : WalSndErrorCleanup(void)
326 : {
327 100 : LWLockReleaseAll();
328 100 : ConditionVariableCancelSleep();
329 100 : pgstat_report_wait_end();
330 :
331 100 : if (xlogreader != NULL && xlogreader->seg.ws_file >= 0)
332 16 : wal_segment_close(xlogreader);
333 :
334 100 : if (MyReplicationSlot != NULL)
335 20 : ReplicationSlotRelease();
336 :
337 100 : ReplicationSlotCleanup(false);
338 :
339 100 : replication_active = false;
340 :
341 : /*
342 : * If there is a transaction in progress, it will clean up our
343 : * ResourceOwner, but if a replication command set up a resource owner
344 : * without a transaction, we've got to clean that up now.
345 : */
346 100 : if (!IsTransactionOrTransactionBlock())
347 98 : ReleaseAuxProcessResources(false);
348 :
349 100 : if (got_STOPPING || got_SIGUSR2)
350 0 : proc_exit(0);
351 :
352 : /* Revert back to startup state */
353 100 : WalSndSetState(WALSNDSTATE_STARTUP);
354 100 : }
355 :
356 : /*
357 : * Handle a client's connection abort in an orderly manner.
358 : */
359 : static void
360 2 : WalSndShutdown(void)
361 : {
362 : /*
363 : * Reset whereToSendOutput to prevent ereport from attempting to send any
364 : * more messages to the standby.
365 : */
366 2 : if (whereToSendOutput == DestRemote)
367 2 : whereToSendOutput = DestNone;
368 :
369 2 : proc_exit(0);
370 : abort(); /* keep the compiler quiet */
371 : }
372 :
373 : /*
374 : * Handle the IDENTIFY_SYSTEM command.
375 : */
376 : static void
377 1312 : IdentifySystem(void)
378 : {
379 : char sysid[32];
380 : char xloc[MAXFNAMELEN];
381 : XLogRecPtr logptr;
382 1312 : char *dbname = NULL;
383 : DestReceiver *dest;
384 : TupOutputState *tstate;
385 : TupleDesc tupdesc;
386 : Datum values[4];
387 1312 : bool nulls[4] = {0};
388 : TimeLineID currTLI;
389 :
390 : /*
391 : * Reply with a result set with one row, four columns. First col is system
392 : * ID, second is timeline ID, third is current xlog location and the
393 : * fourth contains the database name if we are connected to one.
394 : */
395 :
396 1312 : snprintf(sysid, sizeof(sysid), UINT64_FORMAT,
397 : GetSystemIdentifier());
398 :
399 1312 : am_cascading_walsender = RecoveryInProgress();
400 1312 : if (am_cascading_walsender)
401 112 : logptr = GetStandbyFlushRecPtr(&currTLI);
402 : else
403 1200 : logptr = GetFlushRecPtr(&currTLI);
404 :
405 1312 : snprintf(xloc, sizeof(xloc), "%X/%X", LSN_FORMAT_ARGS(logptr));
406 :
407 1312 : if (MyDatabaseId != InvalidOid)
408 : {
409 422 : MemoryContext cur = CurrentMemoryContext;
410 :
411 : /* syscache access needs a transaction env. */
412 422 : StartTransactionCommand();
413 422 : dbname = get_database_name(MyDatabaseId);
414 : /* copy dbname out of TX context */
415 422 : dbname = MemoryContextStrdup(cur, dbname);
416 422 : CommitTransactionCommand();
417 : }
418 :
419 1312 : dest = CreateDestReceiver(DestRemoteSimple);
420 :
421 : /* need a tuple descriptor representing four columns */
422 1312 : tupdesc = CreateTemplateTupleDesc(4);
423 1312 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "systemid",
424 : TEXTOID, -1, 0);
425 1312 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "timeline",
426 : INT8OID, -1, 0);
427 1312 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 3, "xlogpos",
428 : TEXTOID, -1, 0);
429 1312 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 4, "dbname",
430 : TEXTOID, -1, 0);
431 :
432 : /* prepare for projection of tuples */
433 1312 : tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
434 :
435 : /* column 1: system identifier */
436 1312 : values[0] = CStringGetTextDatum(sysid);
437 :
438 : /* column 2: timeline */
439 1312 : values[1] = Int64GetDatum(currTLI);
440 :
441 : /* column 3: wal location */
442 1312 : values[2] = CStringGetTextDatum(xloc);
443 :
444 : /* column 4: database name, or NULL if none */
445 1312 : if (dbname)
446 422 : values[3] = CStringGetTextDatum(dbname);
447 : else
448 890 : nulls[3] = true;
449 :
450 : /* send it to dest */
451 1312 : do_tup_output(tstate, values, nulls);
452 :
453 1312 : end_tup_output(tstate);
454 1312 : }
455 :
456 : /* Handle READ_REPLICATION_SLOT command */
457 : static void
458 12 : ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
459 : {
460 : #define READ_REPLICATION_SLOT_COLS 3
461 : ReplicationSlot *slot;
462 : DestReceiver *dest;
463 : TupOutputState *tstate;
464 : TupleDesc tupdesc;
465 12 : Datum values[READ_REPLICATION_SLOT_COLS] = {0};
466 : bool nulls[READ_REPLICATION_SLOT_COLS];
467 :
468 12 : tupdesc = CreateTemplateTupleDesc(READ_REPLICATION_SLOT_COLS);
469 12 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "slot_type",
470 : TEXTOID, -1, 0);
471 12 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "restart_lsn",
472 : TEXTOID, -1, 0);
473 : /* TimeLineID is unsigned, so int4 is not wide enough. */
474 12 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 3, "restart_tli",
475 : INT8OID, -1, 0);
476 :
477 12 : memset(nulls, true, READ_REPLICATION_SLOT_COLS * sizeof(bool));
478 :
479 12 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
480 12 : slot = SearchNamedReplicationSlot(cmd->slotname, false);
481 12 : if (slot == NULL || !slot->in_use)
482 : {
483 4 : LWLockRelease(ReplicationSlotControlLock);
484 : }
485 : else
486 : {
487 : ReplicationSlot slot_contents;
488 8 : int i = 0;
489 :
490 : /* Copy slot contents while holding spinlock */
491 8 : SpinLockAcquire(&slot->mutex);
492 8 : slot_contents = *slot;
493 8 : SpinLockRelease(&slot->mutex);
494 8 : LWLockRelease(ReplicationSlotControlLock);
495 :
496 8 : if (OidIsValid(slot_contents.data.database))
497 2 : ereport(ERROR,
498 : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
499 : errmsg("cannot use %s with a logical replication slot",
500 : "READ_REPLICATION_SLOT"));
501 :
502 : /* slot type */
503 6 : values[i] = CStringGetTextDatum("physical");
504 6 : nulls[i] = false;
505 6 : i++;
506 :
507 : /* start LSN */
508 6 : if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
509 : {
510 : char xloc[64];
511 :
512 6 : snprintf(xloc, sizeof(xloc), "%X/%X",
513 6 : LSN_FORMAT_ARGS(slot_contents.data.restart_lsn));
514 6 : values[i] = CStringGetTextDatum(xloc);
515 6 : nulls[i] = false;
516 : }
517 6 : i++;
518 :
519 : /* timeline this WAL was produced on */
520 6 : if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
521 : {
522 : TimeLineID slots_position_timeline;
523 : TimeLineID current_timeline;
524 6 : List *timeline_history = NIL;
525 :
526 : /*
527 : * While in recovery, use as timeline the currently-replaying one
528 : * to get the LSN position's history.
529 : */
530 6 : if (RecoveryInProgress())
531 0 : (void) GetXLogReplayRecPtr(¤t_timeline);
532 : else
533 6 : current_timeline = GetWALInsertionTimeLine();
534 :
535 6 : timeline_history = readTimeLineHistory(current_timeline);
536 6 : slots_position_timeline = tliOfPointInHistory(slot_contents.data.restart_lsn,
537 : timeline_history);
538 6 : values[i] = Int64GetDatum((int64) slots_position_timeline);
539 6 : nulls[i] = false;
540 : }
541 6 : i++;
542 :
543 : Assert(i == READ_REPLICATION_SLOT_COLS);
544 : }
545 :
546 10 : dest = CreateDestReceiver(DestRemoteSimple);
547 10 : tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
548 10 : do_tup_output(tstate, values, nulls);
549 10 : end_tup_output(tstate);
550 10 : }
551 :
552 :
553 : /*
554 : * Handle TIMELINE_HISTORY command.
555 : */
556 : static void
557 28 : SendTimeLineHistory(TimeLineHistoryCmd *cmd)
558 : {
559 : DestReceiver *dest;
560 : TupleDesc tupdesc;
561 : StringInfoData buf;
562 : char histfname[MAXFNAMELEN];
563 : char path[MAXPGPATH];
564 : int fd;
565 : off_t histfilelen;
566 : off_t bytesleft;
567 : Size len;
568 :
569 28 : dest = CreateDestReceiver(DestRemoteSimple);
570 :
571 : /*
572 : * Reply with a result set with one row, and two columns. The first col is
573 : * the name of the history file, 2nd is the contents.
574 : */
575 28 : tupdesc = CreateTemplateTupleDesc(2);
576 28 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "filename", TEXTOID, -1, 0);
577 28 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "content", TEXTOID, -1, 0);
578 :
579 28 : TLHistoryFileName(histfname, cmd->timeline);
580 28 : TLHistoryFilePath(path, cmd->timeline);
581 :
582 : /* Send a RowDescription message */
583 28 : dest->rStartup(dest, CMD_SELECT, tupdesc);
584 :
585 : /* Send a DataRow message */
586 28 : pq_beginmessage(&buf, PqMsg_DataRow);
587 28 : pq_sendint16(&buf, 2); /* # of columns */
588 28 : len = strlen(histfname);
589 28 : pq_sendint32(&buf, len); /* col1 len */
590 28 : pq_sendbytes(&buf, histfname, len);
591 :
592 28 : fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
593 28 : if (fd < 0)
594 0 : ereport(ERROR,
595 : (errcode_for_file_access(),
596 : errmsg("could not open file \"%s\": %m", path)));
597 :
598 : /* Determine file length and send it to client */
599 28 : histfilelen = lseek(fd, 0, SEEK_END);
600 28 : if (histfilelen < 0)
601 0 : ereport(ERROR,
602 : (errcode_for_file_access(),
603 : errmsg("could not seek to end of file \"%s\": %m", path)));
604 28 : if (lseek(fd, 0, SEEK_SET) != 0)
605 0 : ereport(ERROR,
606 : (errcode_for_file_access(),
607 : errmsg("could not seek to beginning of file \"%s\": %m", path)));
608 :
609 28 : pq_sendint32(&buf, histfilelen); /* col2 len */
610 :
611 28 : bytesleft = histfilelen;
612 56 : while (bytesleft > 0)
613 : {
614 : PGAlignedBlock rbuf;
615 : int nread;
616 :
617 28 : pgstat_report_wait_start(WAIT_EVENT_WALSENDER_TIMELINE_HISTORY_READ);
618 28 : nread = read(fd, rbuf.data, sizeof(rbuf));
619 28 : pgstat_report_wait_end();
620 28 : if (nread < 0)
621 0 : ereport(ERROR,
622 : (errcode_for_file_access(),
623 : errmsg("could not read file \"%s\": %m",
624 : path)));
625 28 : else if (nread == 0)
626 0 : ereport(ERROR,
627 : (errcode(ERRCODE_DATA_CORRUPTED),
628 : errmsg("could not read file \"%s\": read %d of %zu",
629 : path, nread, (Size) bytesleft)));
630 :
631 28 : pq_sendbytes(&buf, rbuf.data, nread);
632 28 : bytesleft -= nread;
633 : }
634 :
635 28 : if (CloseTransientFile(fd) != 0)
636 0 : ereport(ERROR,
637 : (errcode_for_file_access(),
638 : errmsg("could not close file \"%s\": %m", path)));
639 :
640 28 : pq_endmessage(&buf);
641 28 : }
642 :
643 : /*
644 : * Handle UPLOAD_MANIFEST command.
645 : */
646 : static void
647 22 : UploadManifest(void)
648 : {
649 : MemoryContext mcxt;
650 : IncrementalBackupInfo *ib;
651 22 : off_t offset = 0;
652 : StringInfoData buf;
653 :
654 : /*
655 : * parsing the manifest will use the cryptohash stuff, which requires a
656 : * resource owner
657 : */
658 : Assert(AuxProcessResourceOwner != NULL);
659 : Assert(CurrentResourceOwner == AuxProcessResourceOwner ||
660 : CurrentResourceOwner == NULL);
661 22 : CurrentResourceOwner = AuxProcessResourceOwner;
662 :
663 : /* Prepare to read manifest data into a temporary context. */
664 22 : mcxt = AllocSetContextCreate(CurrentMemoryContext,
665 : "incremental backup information",
666 : ALLOCSET_DEFAULT_SIZES);
667 22 : ib = CreateIncrementalBackupInfo(mcxt);
668 :
669 : /* Send a CopyInResponse message */
670 22 : pq_beginmessage(&buf, PqMsg_CopyInResponse);
671 22 : pq_sendbyte(&buf, 0);
672 22 : pq_sendint16(&buf, 0);
673 22 : pq_endmessage_reuse(&buf);
674 22 : pq_flush();
675 :
676 : /* Receive packets from client until done. */
677 86 : while (HandleUploadManifestPacket(&buf, &offset, ib))
678 : ;
679 :
680 : /* Finish up manifest processing. */
681 20 : FinalizeIncrementalManifest(ib);
682 :
683 : /*
684 : * Discard any old manifest information and arrange to preserve the new
685 : * information we just got.
686 : *
687 : * We assume that MemoryContextDelete and MemoryContextSetParent won't
688 : * fail, and thus we shouldn't end up bailing out of here in such a way as
689 : * to leave dangling pointers.
690 : */
691 20 : if (uploaded_manifest_mcxt != NULL)
692 0 : MemoryContextDelete(uploaded_manifest_mcxt);
693 20 : MemoryContextSetParent(mcxt, CacheMemoryContext);
694 20 : uploaded_manifest = ib;
695 20 : uploaded_manifest_mcxt = mcxt;
696 :
697 : /* clean up the resource owner we created */
698 20 : ReleaseAuxProcessResources(true);
699 20 : }
700 :
701 : /*
702 : * Process one packet received during the handling of an UPLOAD_MANIFEST
703 : * operation.
704 : *
705 : * 'buf' is scratch space. This function expects it to be initialized, doesn't
706 : * care what the current contents are, and may override them with completely
707 : * new contents.
708 : *
709 : * The return value is true if the caller should continue processing
710 : * additional packets and false if the UPLOAD_MANIFEST operation is complete.
711 : */
712 : static bool
713 86 : HandleUploadManifestPacket(StringInfo buf, off_t *offset,
714 : IncrementalBackupInfo *ib)
715 : {
716 : int mtype;
717 : int maxmsglen;
718 :
719 86 : HOLD_CANCEL_INTERRUPTS();
720 :
721 86 : pq_startmsgread();
722 86 : mtype = pq_getbyte();
723 86 : if (mtype == EOF)
724 0 : ereport(ERROR,
725 : (errcode(ERRCODE_CONNECTION_FAILURE),
726 : errmsg("unexpected EOF on client connection with an open transaction")));
727 :
728 86 : switch (mtype)
729 : {
730 66 : case 'd': /* CopyData */
731 66 : maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
732 66 : break;
733 20 : case 'c': /* CopyDone */
734 : case 'f': /* CopyFail */
735 : case 'H': /* Flush */
736 : case 'S': /* Sync */
737 20 : maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
738 20 : break;
739 0 : default:
740 0 : ereport(ERROR,
741 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
742 : errmsg("unexpected message type 0x%02X during COPY from stdin",
743 : mtype)));
744 : maxmsglen = 0; /* keep compiler quiet */
745 : break;
746 : }
747 :
748 : /* Now collect the message body */
749 86 : if (pq_getmessage(buf, maxmsglen))
750 0 : ereport(ERROR,
751 : (errcode(ERRCODE_CONNECTION_FAILURE),
752 : errmsg("unexpected EOF on client connection with an open transaction")));
753 86 : RESUME_CANCEL_INTERRUPTS();
754 :
755 : /* Process the message */
756 86 : switch (mtype)
757 : {
758 66 : case 'd': /* CopyData */
759 66 : AppendIncrementalManifestData(ib, buf->data, buf->len);
760 64 : return true;
761 :
762 20 : case 'c': /* CopyDone */
763 20 : return false;
764 :
765 0 : case 'H': /* Sync */
766 : case 'S': /* Flush */
767 : /* Ignore these while in CopyOut mode as we do elsewhere. */
768 0 : return true;
769 :
770 0 : case 'f':
771 0 : ereport(ERROR,
772 : (errcode(ERRCODE_QUERY_CANCELED),
773 : errmsg("COPY from stdin failed: %s",
774 : pq_getmsgstring(buf))));
775 : }
776 :
777 : /* Not reached. */
778 : Assert(false);
779 0 : return false;
780 : }
781 :
782 : /*
783 : * Handle START_REPLICATION command.
784 : *
785 : * At the moment, this never returns, but an ereport(ERROR) will take us back
786 : * to the main loop.
787 : */
788 : static void
789 512 : StartReplication(StartReplicationCmd *cmd)
790 : {
791 : StringInfoData buf;
792 : XLogRecPtr FlushPtr;
793 : TimeLineID FlushTLI;
794 :
795 : /* create xlogreader for physical replication */
796 512 : xlogreader =
797 512 : XLogReaderAllocate(wal_segment_size, NULL,
798 512 : XL_ROUTINE(.segment_open = WalSndSegmentOpen,
799 : .segment_close = wal_segment_close),
800 : NULL);
801 :
802 512 : if (!xlogreader)
803 0 : ereport(ERROR,
804 : (errcode(ERRCODE_OUT_OF_MEMORY),
805 : errmsg("out of memory"),
806 : errdetail("Failed while allocating a WAL reading processor.")));
807 :
808 : /*
809 : * We assume here that we're logging enough information in the WAL for
810 : * log-shipping, since this is checked in PostmasterMain().
811 : *
812 : * NOTE: wal_level can only change at shutdown, so in most cases it is
813 : * difficult for there to be WAL data that we can still see that was
814 : * written at wal_level='minimal'.
815 : */
816 :
817 512 : if (cmd->slotname)
818 : {
819 356 : ReplicationSlotAcquire(cmd->slotname, true, true);
820 350 : if (SlotIsLogical(MyReplicationSlot))
821 0 : ereport(ERROR,
822 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
823 : errmsg("cannot use a logical replication slot for physical replication")));
824 :
825 : /*
826 : * We don't need to verify the slot's restart_lsn here; instead we
827 : * rely on the caller requesting the starting point to use. If the
828 : * WAL segment doesn't exist, we'll fail later.
829 : */
830 : }
831 :
832 : /*
833 : * Select the timeline. If it was given explicitly by the client, use
834 : * that. Otherwise use the timeline of the last replayed record.
835 : */
836 506 : am_cascading_walsender = RecoveryInProgress();
837 506 : if (am_cascading_walsender)
838 24 : FlushPtr = GetStandbyFlushRecPtr(&FlushTLI);
839 : else
840 482 : FlushPtr = GetFlushRecPtr(&FlushTLI);
841 :
842 506 : if (cmd->timeline != 0)
843 : {
844 : XLogRecPtr switchpoint;
845 :
846 504 : sendTimeLine = cmd->timeline;
847 504 : if (sendTimeLine == FlushTLI)
848 : {
849 480 : sendTimeLineIsHistoric = false;
850 480 : sendTimeLineValidUpto = InvalidXLogRecPtr;
851 : }
852 : else
853 : {
854 : List *timeLineHistory;
855 :
856 24 : sendTimeLineIsHistoric = true;
857 :
858 : /*
859 : * Check that the timeline the client requested exists, and the
860 : * requested start location is on that timeline.
861 : */
862 24 : timeLineHistory = readTimeLineHistory(FlushTLI);
863 24 : switchpoint = tliSwitchPoint(cmd->timeline, timeLineHistory,
864 : &sendTimeLineNextTLI);
865 24 : list_free_deep(timeLineHistory);
866 :
867 : /*
868 : * Found the requested timeline in the history. Check that
869 : * requested startpoint is on that timeline in our history.
870 : *
871 : * This is quite loose on purpose. We only check that we didn't
872 : * fork off the requested timeline before the switchpoint. We
873 : * don't check that we switched *to* it before the requested
874 : * starting point. This is because the client can legitimately
875 : * request to start replication from the beginning of the WAL
876 : * segment that contains switchpoint, but on the new timeline, so
877 : * that it doesn't end up with a partial segment. If you ask for
878 : * too old a starting point, you'll get an error later when we
879 : * fail to find the requested WAL segment in pg_wal.
880 : *
881 : * XXX: we could be more strict here and only allow a startpoint
882 : * that's older than the switchpoint, if it's still in the same
883 : * WAL segment.
884 : */
885 24 : if (!XLogRecPtrIsInvalid(switchpoint) &&
886 24 : switchpoint < cmd->startpoint)
887 : {
888 0 : ereport(ERROR,
889 : (errmsg("requested starting point %X/%X on timeline %u is not in this server's history",
890 : LSN_FORMAT_ARGS(cmd->startpoint),
891 : cmd->timeline),
892 : errdetail("This server's history forked from timeline %u at %X/%X.",
893 : cmd->timeline,
894 : LSN_FORMAT_ARGS(switchpoint))));
895 : }
896 24 : sendTimeLineValidUpto = switchpoint;
897 : }
898 : }
899 : else
900 : {
901 2 : sendTimeLine = FlushTLI;
902 2 : sendTimeLineValidUpto = InvalidXLogRecPtr;
903 2 : sendTimeLineIsHistoric = false;
904 : }
905 :
906 506 : streamingDoneSending = streamingDoneReceiving = false;
907 :
908 : /* If there is nothing to stream, don't even enter COPY mode */
909 506 : if (!sendTimeLineIsHistoric || cmd->startpoint < sendTimeLineValidUpto)
910 : {
911 : /*
912 : * When we first start replication the standby will be behind the
913 : * primary. For some applications, for example synchronous
914 : * replication, it is important to have a clear state for this initial
915 : * catchup mode, so we can trigger actions when we change streaming
916 : * state later. We may stay in this state for a long time, which is
917 : * exactly why we want to be able to monitor whether or not we are
918 : * still here.
919 : */
920 506 : WalSndSetState(WALSNDSTATE_CATCHUP);
921 :
922 : /* Send a CopyBothResponse message, and start streaming */
923 506 : pq_beginmessage(&buf, PqMsg_CopyBothResponse);
924 506 : pq_sendbyte(&buf, 0);
925 506 : pq_sendint16(&buf, 0);
926 506 : pq_endmessage(&buf);
927 506 : pq_flush();
928 :
929 : /*
930 : * Don't allow a request to stream from a future point in WAL that
931 : * hasn't been flushed to disk in this server yet.
932 : */
933 506 : if (FlushPtr < cmd->startpoint)
934 : {
935 0 : ereport(ERROR,
936 : (errmsg("requested starting point %X/%X is ahead of the WAL flush position of this server %X/%X",
937 : LSN_FORMAT_ARGS(cmd->startpoint),
938 : LSN_FORMAT_ARGS(FlushPtr))));
939 : }
940 :
941 : /* Start streaming from the requested point */
942 506 : sentPtr = cmd->startpoint;
943 :
944 : /* Initialize shared memory status, too */
945 506 : SpinLockAcquire(&MyWalSnd->mutex);
946 506 : MyWalSnd->sentPtr = sentPtr;
947 506 : SpinLockRelease(&MyWalSnd->mutex);
948 :
949 506 : SyncRepInitConfig();
950 :
951 : /* Main loop of walsender */
952 506 : replication_active = true;
953 :
954 506 : WalSndLoop(XLogSendPhysical);
955 :
956 296 : replication_active = false;
957 296 : if (got_STOPPING)
958 0 : proc_exit(0);
959 296 : WalSndSetState(WALSNDSTATE_STARTUP);
960 :
961 : Assert(streamingDoneSending && streamingDoneReceiving);
962 : }
963 :
964 296 : if (cmd->slotname)
965 264 : ReplicationSlotRelease();
966 :
967 : /*
968 : * Copy is finished now. Send a single-row result set indicating the next
969 : * timeline.
970 : */
971 296 : if (sendTimeLineIsHistoric)
972 : {
973 : char startpos_str[8 + 1 + 8 + 1];
974 : DestReceiver *dest;
975 : TupOutputState *tstate;
976 : TupleDesc tupdesc;
977 : Datum values[2];
978 26 : bool nulls[2] = {0};
979 :
980 26 : snprintf(startpos_str, sizeof(startpos_str), "%X/%X",
981 26 : LSN_FORMAT_ARGS(sendTimeLineValidUpto));
982 :
983 26 : dest = CreateDestReceiver(DestRemoteSimple);
984 :
985 : /*
986 : * Need a tuple descriptor representing two columns. int8 may seem
987 : * like a surprising data type for this, but in theory int4 would not
988 : * be wide enough for this, as TimeLineID is unsigned.
989 : */
990 26 : tupdesc = CreateTemplateTupleDesc(2);
991 26 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "next_tli",
992 : INT8OID, -1, 0);
993 26 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "next_tli_startpos",
994 : TEXTOID, -1, 0);
995 :
996 : /* prepare for projection of tuple */
997 26 : tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
998 :
999 26 : values[0] = Int64GetDatum((int64) sendTimeLineNextTLI);
1000 26 : values[1] = CStringGetTextDatum(startpos_str);
1001 :
1002 : /* send it to dest */
1003 26 : do_tup_output(tstate, values, nulls);
1004 :
1005 26 : end_tup_output(tstate);
1006 : }
1007 :
1008 : /* Send CommandComplete message */
1009 296 : EndReplicationCommand("START_STREAMING");
1010 296 : }
1011 :
1012 : /*
1013 : * XLogReaderRoutine->page_read callback for logical decoding contexts, as a
1014 : * walsender process.
1015 : *
1016 : * Inside the walsender we can do better than read_local_xlog_page,
1017 : * which has to do a plain sleep/busy loop, because the walsender's latch gets
1018 : * set every time WAL is flushed.
1019 : */
1020 : static int
1021 23680 : logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen,
1022 : XLogRecPtr targetRecPtr, char *cur_page)
1023 : {
1024 : XLogRecPtr flushptr;
1025 : int count;
1026 : WALReadError errinfo;
1027 : XLogSegNo segno;
1028 : TimeLineID currTLI;
1029 :
1030 : /*
1031 : * Make sure we have enough WAL available before retrieving the current
1032 : * timeline.
1033 : */
1034 23680 : flushptr = WalSndWaitForWal(targetPagePtr + reqLen);
1035 :
1036 : /* Fail if not enough (implies we are going to shut down) */
1037 23352 : if (flushptr < targetPagePtr + reqLen)
1038 422 : return -1;
1039 :
1040 : /*
1041 : * Since logical decoding is also permitted on a standby server, we need
1042 : * to check if the server is in recovery to decide how to get the current
1043 : * timeline ID (so that it also covers the promotion or timeline change
1044 : * cases). We must determine am_cascading_walsender after waiting for the
1045 : * required WAL so that it is correct when the walsender wakes up after a
1046 : * promotion.
1047 : */
1048 22930 : am_cascading_walsender = RecoveryInProgress();
1049 :
1050 22930 : if (am_cascading_walsender)
1051 622 : GetXLogReplayRecPtr(&currTLI);
1052 : else
1053 22308 : currTLI = GetWALInsertionTimeLine();
1054 :
1055 22930 : XLogReadDetermineTimeline(state, targetPagePtr, reqLen, currTLI);
1056 22930 : sendTimeLineIsHistoric = (state->currTLI != currTLI);
1057 22930 : sendTimeLine = state->currTLI;
1058 22930 : sendTimeLineValidUpto = state->currTLIValidUntil;
1059 22930 : sendTimeLineNextTLI = state->nextTLI;
1060 :
1061 22930 : if (targetPagePtr + XLOG_BLCKSZ <= flushptr)
1062 19210 : count = XLOG_BLCKSZ; /* more than one block available */
1063 : else
1064 3720 : count = flushptr - targetPagePtr; /* part of the page available */
1065 :
1066 : /* now actually read the data, we know it's there */
1067 22930 : if (!WALRead(state,
1068 : cur_page,
1069 : targetPagePtr,
1070 : count,
1071 : currTLI, /* Pass the current TLI because only
1072 : * WalSndSegmentOpen controls whether new TLI
1073 : * is needed. */
1074 : &errinfo))
1075 0 : WALReadRaiseError(&errinfo);
1076 :
1077 : /*
1078 : * After reading into the buffer, check that what we read was valid. We do
1079 : * this after reading, because even though the segment was present when we
1080 : * opened it, it might get recycled or removed while we read it. The
1081 : * read() succeeds in that case, but the data we tried to read might
1082 : * already have been overwritten with new WAL records.
1083 : */
1084 22930 : XLByteToSeg(targetPagePtr, segno, state->segcxt.ws_segsize);
1085 22930 : CheckXLogRemoved(segno, state->seg.ws_tli);
1086 :
1087 22930 : return count;
1088 : }
1089 :
1090 : /*
1091 : * Process extra options given to CREATE_REPLICATION_SLOT.
1092 : */
1093 : static void
1094 894 : parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
1095 : bool *reserve_wal,
1096 : CRSSnapshotAction *snapshot_action,
1097 : bool *two_phase, bool *failover)
1098 : {
1099 : ListCell *lc;
1100 894 : bool snapshot_action_given = false;
1101 894 : bool reserve_wal_given = false;
1102 894 : bool two_phase_given = false;
1103 894 : bool failover_given = false;
1104 :
1105 : /* Parse options */
1106 1802 : foreach(lc, cmd->options)
1107 : {
1108 908 : DefElem *defel = (DefElem *) lfirst(lc);
1109 :
1110 908 : if (strcmp(defel->defname, "snapshot") == 0)
1111 : {
1112 : char *action;
1113 :
1114 624 : if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)
1115 0 : ereport(ERROR,
1116 : (errcode(ERRCODE_SYNTAX_ERROR),
1117 : errmsg("conflicting or redundant options")));
1118 :
1119 624 : action = defGetString(defel);
1120 624 : snapshot_action_given = true;
1121 :
1122 624 : if (strcmp(action, "export") == 0)
1123 0 : *snapshot_action = CRS_EXPORT_SNAPSHOT;
1124 624 : else if (strcmp(action, "nothing") == 0)
1125 252 : *snapshot_action = CRS_NOEXPORT_SNAPSHOT;
1126 372 : else if (strcmp(action, "use") == 0)
1127 372 : *snapshot_action = CRS_USE_SNAPSHOT;
1128 : else
1129 0 : ereport(ERROR,
1130 : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1131 : errmsg("unrecognized value for CREATE_REPLICATION_SLOT option \"%s\": \"%s\"",
1132 : defel->defname, action)));
1133 : }
1134 284 : else if (strcmp(defel->defname, "reserve_wal") == 0)
1135 : {
1136 268 : if (reserve_wal_given || cmd->kind != REPLICATION_KIND_PHYSICAL)
1137 0 : ereport(ERROR,
1138 : (errcode(ERRCODE_SYNTAX_ERROR),
1139 : errmsg("conflicting or redundant options")));
1140 :
1141 268 : reserve_wal_given = true;
1142 268 : *reserve_wal = defGetBoolean(defel);
1143 : }
1144 16 : else if (strcmp(defel->defname, "two_phase") == 0)
1145 : {
1146 4 : if (two_phase_given || cmd->kind != REPLICATION_KIND_LOGICAL)
1147 0 : ereport(ERROR,
1148 : (errcode(ERRCODE_SYNTAX_ERROR),
1149 : errmsg("conflicting or redundant options")));
1150 4 : two_phase_given = true;
1151 4 : *two_phase = defGetBoolean(defel);
1152 : }
1153 12 : else if (strcmp(defel->defname, "failover") == 0)
1154 : {
1155 12 : if (failover_given || cmd->kind != REPLICATION_KIND_LOGICAL)
1156 0 : ereport(ERROR,
1157 : (errcode(ERRCODE_SYNTAX_ERROR),
1158 : errmsg("conflicting or redundant options")));
1159 12 : failover_given = true;
1160 12 : *failover = defGetBoolean(defel);
1161 : }
1162 : else
1163 0 : elog(ERROR, "unrecognized option: %s", defel->defname);
1164 : }
1165 894 : }
1166 :
1167 : /*
1168 : * Create a new replication slot.
1169 : */
1170 : static void
1171 894 : CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
1172 : {
1173 894 : const char *snapshot_name = NULL;
1174 : char xloc[MAXFNAMELEN];
1175 : char *slot_name;
1176 894 : bool reserve_wal = false;
1177 894 : bool two_phase = false;
1178 894 : bool failover = false;
1179 894 : CRSSnapshotAction snapshot_action = CRS_EXPORT_SNAPSHOT;
1180 : DestReceiver *dest;
1181 : TupOutputState *tstate;
1182 : TupleDesc tupdesc;
1183 : Datum values[4];
1184 894 : bool nulls[4] = {0};
1185 :
1186 : Assert(!MyReplicationSlot);
1187 :
1188 894 : parseCreateReplSlotOptions(cmd, &reserve_wal, &snapshot_action, &two_phase,
1189 : &failover);
1190 :
1191 894 : if (cmd->kind == REPLICATION_KIND_PHYSICAL)
1192 : {
1193 270 : ReplicationSlotCreate(cmd->slotname, false,
1194 270 : cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
1195 : false, false, false);
1196 :
1197 268 : if (reserve_wal)
1198 : {
1199 266 : ReplicationSlotReserveWal();
1200 :
1201 266 : ReplicationSlotMarkDirty();
1202 :
1203 : /* Write this slot to disk if it's a permanent one. */
1204 266 : if (!cmd->temporary)
1205 6 : ReplicationSlotSave();
1206 : }
1207 : }
1208 : else
1209 : {
1210 : LogicalDecodingContext *ctx;
1211 624 : bool need_full_snapshot = false;
1212 :
1213 : Assert(cmd->kind == REPLICATION_KIND_LOGICAL);
1214 :
1215 624 : CheckLogicalDecodingRequirements();
1216 :
1217 : /*
1218 : * Initially create persistent slot as ephemeral - that allows us to
1219 : * nicely handle errors during initialization because it'll get
1220 : * dropped if this transaction fails. We'll make it persistent at the
1221 : * end. Temporary slots can be created as temporary from beginning as
1222 : * they get dropped on error as well.
1223 : */
1224 624 : ReplicationSlotCreate(cmd->slotname, true,
1225 624 : cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
1226 : two_phase, failover, false);
1227 :
1228 : /*
1229 : * Do options check early so that we can bail before calling the
1230 : * DecodingContextFindStartpoint which can take long time.
1231 : */
1232 624 : if (snapshot_action == CRS_EXPORT_SNAPSHOT)
1233 : {
1234 0 : if (IsTransactionBlock())
1235 0 : ereport(ERROR,
1236 : /*- translator: %s is a CREATE_REPLICATION_SLOT statement */
1237 : (errmsg("%s must not be called inside a transaction",
1238 : "CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')")));
1239 :
1240 0 : need_full_snapshot = true;
1241 : }
1242 624 : else if (snapshot_action == CRS_USE_SNAPSHOT)
1243 : {
1244 372 : if (!IsTransactionBlock())
1245 0 : ereport(ERROR,
1246 : /*- translator: %s is a CREATE_REPLICATION_SLOT statement */
1247 : (errmsg("%s must be called inside a transaction",
1248 : "CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
1249 :
1250 372 : if (XactIsoLevel != XACT_REPEATABLE_READ)
1251 0 : ereport(ERROR,
1252 : /*- translator: %s is a CREATE_REPLICATION_SLOT statement */
1253 : (errmsg("%s must be called in REPEATABLE READ isolation mode transaction",
1254 : "CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
1255 372 : if (!XactReadOnly)
1256 0 : ereport(ERROR,
1257 : /*- translator: %s is a CREATE_REPLICATION_SLOT statement */
1258 : (errmsg("%s must be called in a read-only transaction",
1259 : "CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
1260 :
1261 372 : if (FirstSnapshotSet)
1262 0 : ereport(ERROR,
1263 : /*- translator: %s is a CREATE_REPLICATION_SLOT statement */
1264 : (errmsg("%s must be called before any query",
1265 : "CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
1266 :
1267 372 : if (IsSubTransaction())
1268 0 : ereport(ERROR,
1269 : /*- translator: %s is a CREATE_REPLICATION_SLOT statement */
1270 : (errmsg("%s must not be called in a subtransaction",
1271 : "CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')")));
1272 :
1273 372 : need_full_snapshot = true;
1274 : }
1275 :
1276 624 : ctx = CreateInitDecodingContext(cmd->plugin, NIL, need_full_snapshot,
1277 : InvalidXLogRecPtr,
1278 624 : XL_ROUTINE(.page_read = logical_read_xlog_page,
1279 : .segment_open = WalSndSegmentOpen,
1280 : .segment_close = wal_segment_close),
1281 : WalSndPrepareWrite, WalSndWriteData,
1282 : WalSndUpdateProgress);
1283 :
1284 : /*
1285 : * Signal that we don't need the timeout mechanism. We're just
1286 : * creating the replication slot and don't yet accept feedback
1287 : * messages or send keepalives. As we possibly need to wait for
1288 : * further WAL the walsender would otherwise possibly be killed too
1289 : * soon.
1290 : */
1291 624 : last_reply_timestamp = 0;
1292 :
1293 : /* build initial snapshot, might take a while */
1294 624 : DecodingContextFindStartpoint(ctx);
1295 :
1296 : /*
1297 : * Export or use the snapshot if we've been asked to do so.
1298 : *
1299 : * NB. We will convert the snapbuild.c kind of snapshot to normal
1300 : * snapshot when doing this.
1301 : */
1302 624 : if (snapshot_action == CRS_EXPORT_SNAPSHOT)
1303 : {
1304 0 : snapshot_name = SnapBuildExportSnapshot(ctx->snapshot_builder);
1305 : }
1306 624 : else if (snapshot_action == CRS_USE_SNAPSHOT)
1307 : {
1308 : Snapshot snap;
1309 :
1310 372 : snap = SnapBuildInitialSnapshot(ctx->snapshot_builder);
1311 372 : RestoreTransactionSnapshot(snap, MyProc);
1312 : }
1313 :
1314 : /* don't need the decoding context anymore */
1315 624 : FreeDecodingContext(ctx);
1316 :
1317 624 : if (!cmd->temporary)
1318 624 : ReplicationSlotPersist();
1319 : }
1320 :
1321 892 : snprintf(xloc, sizeof(xloc), "%X/%X",
1322 892 : LSN_FORMAT_ARGS(MyReplicationSlot->data.confirmed_flush));
1323 :
1324 892 : dest = CreateDestReceiver(DestRemoteSimple);
1325 :
1326 : /*----------
1327 : * Need a tuple descriptor representing four columns:
1328 : * - first field: the slot name
1329 : * - second field: LSN at which we became consistent
1330 : * - third field: exported snapshot's name
1331 : * - fourth field: output plugin
1332 : */
1333 892 : tupdesc = CreateTemplateTupleDesc(4);
1334 892 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "slot_name",
1335 : TEXTOID, -1, 0);
1336 892 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "consistent_point",
1337 : TEXTOID, -1, 0);
1338 892 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 3, "snapshot_name",
1339 : TEXTOID, -1, 0);
1340 892 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 4, "output_plugin",
1341 : TEXTOID, -1, 0);
1342 :
1343 : /* prepare for projection of tuples */
1344 892 : tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
1345 :
1346 : /* slot_name */
1347 892 : slot_name = NameStr(MyReplicationSlot->data.name);
1348 892 : values[0] = CStringGetTextDatum(slot_name);
1349 :
1350 : /* consistent wal location */
1351 892 : values[1] = CStringGetTextDatum(xloc);
1352 :
1353 : /* snapshot name, or NULL if none */
1354 892 : if (snapshot_name != NULL)
1355 0 : values[2] = CStringGetTextDatum(snapshot_name);
1356 : else
1357 892 : nulls[2] = true;
1358 :
1359 : /* plugin, or NULL if none */
1360 892 : if (cmd->plugin != NULL)
1361 624 : values[3] = CStringGetTextDatum(cmd->plugin);
1362 : else
1363 268 : nulls[3] = true;
1364 :
1365 : /* send it to dest */
1366 892 : do_tup_output(tstate, values, nulls);
1367 892 : end_tup_output(tstate);
1368 :
1369 892 : ReplicationSlotRelease();
1370 892 : }
1371 :
1372 : /*
1373 : * Get rid of a replication slot that is no longer wanted.
1374 : */
1375 : static void
1376 496 : DropReplicationSlot(DropReplicationSlotCmd *cmd)
1377 : {
1378 496 : ReplicationSlotDrop(cmd->slotname, !cmd->wait);
1379 496 : }
1380 :
1381 : /*
1382 : * Change the definition of a replication slot.
1383 : */
1384 : static void
1385 12 : AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
1386 : {
1387 12 : bool failover_given = false;
1388 12 : bool two_phase_given = false;
1389 : bool failover;
1390 : bool two_phase;
1391 :
1392 : /* Parse options */
1393 36 : foreach_ptr(DefElem, defel, cmd->options)
1394 : {
1395 12 : if (strcmp(defel->defname, "failover") == 0)
1396 : {
1397 10 : if (failover_given)
1398 0 : ereport(ERROR,
1399 : (errcode(ERRCODE_SYNTAX_ERROR),
1400 : errmsg("conflicting or redundant options")));
1401 10 : failover_given = true;
1402 10 : failover = defGetBoolean(defel);
1403 : }
1404 2 : else if (strcmp(defel->defname, "two_phase") == 0)
1405 : {
1406 2 : if (two_phase_given)
1407 0 : ereport(ERROR,
1408 : (errcode(ERRCODE_SYNTAX_ERROR),
1409 : errmsg("conflicting or redundant options")));
1410 2 : two_phase_given = true;
1411 2 : two_phase = defGetBoolean(defel);
1412 : }
1413 : else
1414 0 : elog(ERROR, "unrecognized option: %s", defel->defname);
1415 : }
1416 :
1417 12 : ReplicationSlotAlter(cmd->slotname,
1418 : failover_given ? &failover : NULL,
1419 : two_phase_given ? &two_phase : NULL);
1420 8 : }
1421 :
1422 : /*
1423 : * Load previously initiated logical slot and prepare for sending data (via
1424 : * WalSndLoop).
1425 : */
1426 : static void
1427 728 : StartLogicalReplication(StartReplicationCmd *cmd)
1428 : {
1429 : StringInfoData buf;
1430 : QueryCompletion qc;
1431 :
1432 : /* make sure that our requirements are still fulfilled */
1433 728 : CheckLogicalDecodingRequirements();
1434 :
1435 : Assert(!MyReplicationSlot);
1436 :
1437 724 : ReplicationSlotAcquire(cmd->slotname, true, true);
1438 :
1439 : /*
1440 : * Force a disconnect, so that the decoding code doesn't need to care
1441 : * about an eventual switch from running in recovery, to running in a
1442 : * normal environment. Client code is expected to handle reconnects.
1443 : */
1444 714 : if (am_cascading_walsender && !RecoveryInProgress())
1445 : {
1446 0 : ereport(LOG,
1447 : (errmsg("terminating walsender process after promotion")));
1448 0 : got_STOPPING = true;
1449 : }
1450 :
1451 : /*
1452 : * Create our decoding context, making it start at the previously ack'ed
1453 : * position.
1454 : *
1455 : * Do this before sending a CopyBothResponse message, so that any errors
1456 : * are reported early.
1457 : */
1458 712 : logical_decoding_ctx =
1459 714 : CreateDecodingContext(cmd->startpoint, cmd->options, false,
1460 714 : XL_ROUTINE(.page_read = logical_read_xlog_page,
1461 : .segment_open = WalSndSegmentOpen,
1462 : .segment_close = wal_segment_close),
1463 : WalSndPrepareWrite, WalSndWriteData,
1464 : WalSndUpdateProgress);
1465 712 : xlogreader = logical_decoding_ctx->reader;
1466 :
1467 712 : WalSndSetState(WALSNDSTATE_CATCHUP);
1468 :
1469 : /* Send a CopyBothResponse message, and start streaming */
1470 712 : pq_beginmessage(&buf, PqMsg_CopyBothResponse);
1471 712 : pq_sendbyte(&buf, 0);
1472 712 : pq_sendint16(&buf, 0);
1473 712 : pq_endmessage(&buf);
1474 712 : pq_flush();
1475 :
1476 : /* Start reading WAL from the oldest required WAL. */
1477 712 : XLogBeginRead(logical_decoding_ctx->reader,
1478 712 : MyReplicationSlot->data.restart_lsn);
1479 :
1480 : /*
1481 : * Report the location after which we'll send out further commits as the
1482 : * current sentPtr.
1483 : */
1484 712 : sentPtr = MyReplicationSlot->data.confirmed_flush;
1485 :
1486 : /* Also update the sent position status in shared memory */
1487 712 : SpinLockAcquire(&MyWalSnd->mutex);
1488 712 : MyWalSnd->sentPtr = MyReplicationSlot->data.restart_lsn;
1489 712 : SpinLockRelease(&MyWalSnd->mutex);
1490 :
1491 712 : replication_active = true;
1492 :
1493 712 : SyncRepInitConfig();
1494 :
1495 : /* Main loop of walsender */
1496 712 : WalSndLoop(XLogSendLogical);
1497 :
1498 360 : FreeDecodingContext(logical_decoding_ctx);
1499 360 : ReplicationSlotRelease();
1500 :
1501 360 : replication_active = false;
1502 360 : if (got_STOPPING)
1503 0 : proc_exit(0);
1504 360 : WalSndSetState(WALSNDSTATE_STARTUP);
1505 :
1506 : /* Get out of COPY mode (CommandComplete). */
1507 360 : SetQueryCompletion(&qc, CMDTAG_COPY, 0);
1508 360 : EndCommand(&qc, DestRemote, false);
1509 360 : }
1510 :
1511 : /*
1512 : * LogicalDecodingContext 'prepare_write' callback.
1513 : *
1514 : * Prepare a write into a StringInfo.
1515 : *
1516 : * Don't do anything lasting in here, it's quite possible that nothing will be done
1517 : * with the data.
1518 : */
1519 : static void
1520 369772 : WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write)
1521 : {
1522 : /* can't have sync rep confused by sending the same LSN several times */
1523 369772 : if (!last_write)
1524 748 : lsn = InvalidXLogRecPtr;
1525 :
1526 369772 : resetStringInfo(ctx->out);
1527 :
1528 369772 : pq_sendbyte(ctx->out, 'w');
1529 369772 : pq_sendint64(ctx->out, lsn); /* dataStart */
1530 369772 : pq_sendint64(ctx->out, lsn); /* walEnd */
1531 :
1532 : /*
1533 : * Fill out the sendtime later, just as it's done in XLogSendPhysical, but
1534 : * reserve space here.
1535 : */
1536 369772 : pq_sendint64(ctx->out, 0); /* sendtime */
1537 369772 : }
1538 :
1539 : /*
1540 : * LogicalDecodingContext 'write' callback.
1541 : *
1542 : * Actually write out data previously prepared by WalSndPrepareWrite out to
1543 : * the network. Take as long as needed, but process replies from the other
1544 : * side and check timeouts during that.
1545 : */
1546 : static void
1547 369772 : WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
1548 : bool last_write)
1549 : {
1550 : TimestampTz now;
1551 :
1552 : /*
1553 : * Fill the send timestamp last, so that it is taken as late as possible.
1554 : * This is somewhat ugly, but the protocol is set as it's already used for
1555 : * several releases by streaming physical replication.
1556 : */
1557 369772 : resetStringInfo(&tmpbuf);
1558 369772 : now = GetCurrentTimestamp();
1559 369772 : pq_sendint64(&tmpbuf, now);
1560 369772 : memcpy(&ctx->out->data[1 + sizeof(int64) + sizeof(int64)],
1561 369772 : tmpbuf.data, sizeof(int64));
1562 :
1563 : /* output previously gathered data in a CopyData packet */
1564 369772 : pq_putmessage_noblock('d', ctx->out->data, ctx->out->len);
1565 :
1566 369772 : CHECK_FOR_INTERRUPTS();
1567 :
1568 : /* Try to flush pending output to the client */
1569 369772 : if (pq_flush_if_writable() != 0)
1570 2 : WalSndShutdown();
1571 :
1572 : /* Try taking fast path unless we get too close to walsender timeout. */
1573 369770 : if (now < TimestampTzPlusMilliseconds(last_reply_timestamp,
1574 369770 : wal_sender_timeout / 2) &&
1575 369770 : !pq_is_send_pending())
1576 : {
1577 369708 : return;
1578 : }
1579 :
1580 : /* If we have pending write here, go to slow path */
1581 62 : ProcessPendingWrites();
1582 : }
1583 :
1584 : /*
1585 : * Wait until there is no pending write. Also process replies from the other
1586 : * side and check timeouts during that.
1587 : */
1588 : static void
1589 160 : ProcessPendingWrites(void)
1590 : {
1591 : for (;;)
1592 98 : {
1593 : long sleeptime;
1594 :
1595 : /* Check for input from the client */
1596 160 : ProcessRepliesIfAny();
1597 :
1598 : /* die if timeout was reached */
1599 160 : WalSndCheckTimeOut();
1600 :
1601 : /* Send keepalive if the time has come */
1602 160 : WalSndKeepaliveIfNecessary();
1603 :
1604 160 : if (!pq_is_send_pending())
1605 62 : break;
1606 :
1607 98 : sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
1608 :
1609 : /* Sleep until something happens or we time out */
1610 98 : WalSndWait(WL_SOCKET_WRITEABLE | WL_SOCKET_READABLE, sleeptime,
1611 : WAIT_EVENT_WAL_SENDER_WRITE_DATA);
1612 :
1613 : /* Clear any already-pending wakeups */
1614 98 : ResetLatch(MyLatch);
1615 :
1616 98 : CHECK_FOR_INTERRUPTS();
1617 :
1618 : /* Process any requests or signals received recently */
1619 98 : if (ConfigReloadPending)
1620 : {
1621 0 : ConfigReloadPending = false;
1622 0 : ProcessConfigFile(PGC_SIGHUP);
1623 0 : SyncRepInitConfig();
1624 : }
1625 :
1626 : /* Try to flush pending output to the client */
1627 98 : if (pq_flush_if_writable() != 0)
1628 0 : WalSndShutdown();
1629 : }
1630 :
1631 : /* reactivate latch so WalSndLoop knows to continue */
1632 62 : SetLatch(MyLatch);
1633 62 : }
1634 :
1635 : /*
1636 : * LogicalDecodingContext 'update_progress' callback.
1637 : *
1638 : * Write the current position to the lag tracker (see XLogSendPhysical).
1639 : *
1640 : * When skipping empty transactions, send a keepalive message if necessary.
1641 : */
1642 : static void
1643 4688 : WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
1644 : bool skipped_xact)
1645 : {
1646 : static TimestampTz sendTime = 0;
1647 4688 : TimestampTz now = GetCurrentTimestamp();
1648 4688 : bool pending_writes = false;
1649 4688 : bool end_xact = ctx->end_xact;
1650 :
1651 : /*
1652 : * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to
1653 : * avoid flooding the lag tracker when we commit frequently.
1654 : *
1655 : * We don't have a mechanism to get the ack for any LSN other than end
1656 : * xact LSN from the downstream. So, we track lag only for end of
1657 : * transaction LSN.
1658 : */
1659 : #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000
1660 4688 : if (end_xact && TimestampDifferenceExceeds(sendTime, now,
1661 : WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS))
1662 : {
1663 352 : LagTrackerWrite(lsn, now);
1664 352 : sendTime = now;
1665 : }
1666 :
1667 : /*
1668 : * When skipping empty transactions in synchronous replication, we send a
1669 : * keepalive message to avoid delaying such transactions.
1670 : *
1671 : * It is okay to check sync_standbys_defined flag without lock here as in
1672 : * the worst case we will just send an extra keepalive message when it is
1673 : * really not required.
1674 : */
1675 4688 : if (skipped_xact &&
1676 602 : SyncRepRequested() &&
1677 602 : ((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
1678 : {
1679 0 : WalSndKeepalive(false, lsn);
1680 :
1681 : /* Try to flush pending output to the client */
1682 0 : if (pq_flush_if_writable() != 0)
1683 0 : WalSndShutdown();
1684 :
1685 : /* If we have pending write here, make sure it's actually flushed */
1686 0 : if (pq_is_send_pending())
1687 0 : pending_writes = true;
1688 : }
1689 :
1690 : /*
1691 : * Process pending writes if any or try to send a keepalive if required.
1692 : * We don't need to try sending keep alive messages at the transaction end
1693 : * as that will be done at a later point in time. This is required only
1694 : * for large transactions where we don't send any changes to the
1695 : * downstream and the receiver can timeout due to that.
1696 : */
1697 4688 : if (pending_writes || (!end_xact &&
1698 3058 : now >= TimestampTzPlusMilliseconds(last_reply_timestamp,
1699 : wal_sender_timeout / 2)))
1700 0 : ProcessPendingWrites();
1701 4688 : }
1702 :
1703 : /*
1704 : * Wake up the logical walsender processes with logical failover slots if the
1705 : * currently acquired physical slot is specified in synchronized_standby_slots GUC.
1706 : */
1707 : void
1708 15388 : PhysicalWakeupLogicalWalSnd(void)
1709 : {
1710 : Assert(MyReplicationSlot && SlotIsPhysical(MyReplicationSlot));
1711 :
1712 : /*
1713 : * If we are running in a standby, there is no need to wake up walsenders.
1714 : * This is because we do not support syncing slots to cascading standbys,
1715 : * so, there are no walsenders waiting for standbys to catch up.
1716 : */
1717 15388 : if (RecoveryInProgress())
1718 112 : return;
1719 :
1720 15276 : if (SlotExistsInSyncStandbySlots(NameStr(MyReplicationSlot->data.name)))
1721 10 : ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
1722 : }
1723 :
1724 : /*
1725 : * Returns true if not all standbys have caught up to the flushed position
1726 : * (flushed_lsn) when the current acquired slot is a logical failover
1727 : * slot and we are streaming; otherwise, returns false.
1728 : *
1729 : * If returning true, the function sets the appropriate wait event in
1730 : * wait_event; otherwise, wait_event is set to 0.
1731 : */
1732 : static bool
1733 23048 : NeedToWaitForStandbys(XLogRecPtr flushed_lsn, uint32 *wait_event)
1734 : {
1735 23048 : int elevel = got_STOPPING ? ERROR : WARNING;
1736 : bool failover_slot;
1737 :
1738 23048 : failover_slot = (replication_active && MyReplicationSlot->data.failover);
1739 :
1740 : /*
1741 : * Note that after receiving the shutdown signal, an ERROR is reported if
1742 : * any slots are dropped, invalidated, or inactive. This measure is taken
1743 : * to prevent the walsender from waiting indefinitely.
1744 : */
1745 23048 : if (failover_slot && !StandbySlotsHaveCaughtup(flushed_lsn, elevel))
1746 : {
1747 8 : *wait_event = WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION;
1748 8 : return true;
1749 : }
1750 :
1751 23040 : *wait_event = 0;
1752 23040 : return false;
1753 : }
1754 :
1755 : /*
1756 : * Returns true if we need to wait for WALs to be flushed to disk, or if not
1757 : * all standbys have caught up to the flushed position (flushed_lsn) when the
1758 : * current acquired slot is a logical failover slot and we are
1759 : * streaming; otherwise, returns false.
1760 : *
1761 : * If returning true, the function sets the appropriate wait event in
1762 : * wait_event; otherwise, wait_event is set to 0.
1763 : */
1764 : static bool
1765 31612 : NeedToWaitForWal(XLogRecPtr target_lsn, XLogRecPtr flushed_lsn,
1766 : uint32 *wait_event)
1767 : {
1768 : /* Check if we need to wait for WALs to be flushed to disk */
1769 31612 : if (target_lsn > flushed_lsn)
1770 : {
1771 8676 : *wait_event = WAIT_EVENT_WAL_SENDER_WAIT_FOR_WAL;
1772 8676 : return true;
1773 : }
1774 :
1775 : /* Check if the standby slots have caught up to the flushed position */
1776 22936 : return NeedToWaitForStandbys(flushed_lsn, wait_event);
1777 : }
1778 :
1779 : /*
1780 : * Wait till WAL < loc is flushed to disk so it can be safely sent to client.
1781 : *
1782 : * If the walsender holds a logical failover slot, we also wait for all the
1783 : * specified streaming replication standby servers to confirm receipt of WAL
1784 : * up to RecentFlushPtr. It is beneficial to wait here for the confirmation
1785 : * up to RecentFlushPtr rather than waiting before transmitting each change
1786 : * to logical subscribers, which is already covered by RecentFlushPtr.
1787 : *
1788 : * Returns end LSN of flushed WAL. Normally this will be >= loc, but if we
1789 : * detect a shutdown request (either from postmaster or client) we will return
1790 : * early, so caller must always check.
1791 : */
1792 : static XLogRecPtr
1793 23680 : WalSndWaitForWal(XLogRecPtr loc)
1794 : {
1795 : int wakeEvents;
1796 23680 : uint32 wait_event = 0;
1797 : static XLogRecPtr RecentFlushPtr = InvalidXLogRecPtr;
1798 :
1799 : /*
1800 : * Fast path to avoid acquiring the spinlock in case we already know we
1801 : * have enough WAL available and all the standby servers have confirmed
1802 : * receipt of WAL up to RecentFlushPtr. This is particularly interesting
1803 : * if we're far behind.
1804 : */
1805 23680 : if (!XLogRecPtrIsInvalid(RecentFlushPtr) &&
1806 22694 : !NeedToWaitForWal(loc, RecentFlushPtr, &wait_event))
1807 19456 : return RecentFlushPtr;
1808 :
1809 : /*
1810 : * Within the loop, we wait for the necessary WALs to be flushed to disk
1811 : * first, followed by waiting for standbys to catch up if there are enough
1812 : * WALs (see NeedToWaitForWal()) or upon receiving the shutdown signal.
1813 : */
1814 : for (;;)
1815 5134 : {
1816 9358 : bool wait_for_standby_at_stop = false;
1817 : long sleeptime;
1818 :
1819 : /* Clear any already-pending wakeups */
1820 9358 : ResetLatch(MyLatch);
1821 :
1822 9358 : CHECK_FOR_INTERRUPTS();
1823 :
1824 : /* Process any requests or signals received recently */
1825 9346 : if (ConfigReloadPending)
1826 : {
1827 14 : ConfigReloadPending = false;
1828 14 : ProcessConfigFile(PGC_SIGHUP);
1829 14 : SyncRepInitConfig();
1830 : }
1831 :
1832 : /* Check for input from the client */
1833 9346 : ProcessRepliesIfAny();
1834 :
1835 : /*
1836 : * If we're shutting down, trigger pending WAL to be written out,
1837 : * otherwise we'd possibly end up waiting for WAL that never gets
1838 : * written, because walwriter has shut down already.
1839 : */
1840 9030 : if (got_STOPPING)
1841 112 : XLogBackgroundFlush();
1842 :
1843 : /*
1844 : * To avoid the scenario where standbys need to catch up to a newer
1845 : * WAL location in each iteration, we update our idea of the currently
1846 : * flushed position only if we are not waiting for standbys to catch
1847 : * up.
1848 : */
1849 9030 : if (wait_event != WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION)
1850 : {
1851 9022 : if (!RecoveryInProgress())
1852 8242 : RecentFlushPtr = GetFlushRecPtr(NULL);
1853 : else
1854 780 : RecentFlushPtr = GetXLogReplayRecPtr(NULL);
1855 : }
1856 :
1857 : /*
1858 : * If postmaster asked us to stop and the standby slots have caught up
1859 : * to the flushed position, don't wait anymore.
1860 : *
1861 : * It's important to do this check after the recomputation of
1862 : * RecentFlushPtr, so we can send all remaining data before shutting
1863 : * down.
1864 : */
1865 9030 : if (got_STOPPING)
1866 : {
1867 112 : if (NeedToWaitForStandbys(RecentFlushPtr, &wait_event))
1868 0 : wait_for_standby_at_stop = true;
1869 : else
1870 112 : break;
1871 : }
1872 :
1873 : /*
1874 : * We only send regular messages to the client for full decoded
1875 : * transactions, but a synchronous replication and walsender shutdown
1876 : * possibly are waiting for a later location. So, before sleeping, we
1877 : * send a ping containing the flush location. If the receiver is
1878 : * otherwise idle, this keepalive will trigger a reply. Processing the
1879 : * reply will update these MyWalSnd locations.
1880 : */
1881 8918 : if (MyWalSnd->flush < sentPtr &&
1882 4760 : MyWalSnd->write < sentPtr &&
1883 3520 : !waiting_for_ping_response)
1884 3520 : WalSndKeepalive(false, InvalidXLogRecPtr);
1885 :
1886 : /*
1887 : * Exit the loop if already caught up and doesn't need to wait for
1888 : * standby slots.
1889 : */
1890 8918 : if (!wait_for_standby_at_stop &&
1891 8918 : !NeedToWaitForWal(loc, RecentFlushPtr, &wait_event))
1892 3472 : break;
1893 :
1894 : /*
1895 : * Waiting for new WAL or waiting for standbys to catch up. Since we
1896 : * need to wait, we're now caught up.
1897 : */
1898 5446 : WalSndCaughtUp = true;
1899 :
1900 : /*
1901 : * Try to flush any pending output to the client.
1902 : */
1903 5446 : if (pq_flush_if_writable() != 0)
1904 0 : WalSndShutdown();
1905 :
1906 : /*
1907 : * If we have received CopyDone from the client, sent CopyDone
1908 : * ourselves, and the output buffer is empty, it's time to exit
1909 : * streaming, so fail the current WAL fetch request.
1910 : */
1911 5446 : if (streamingDoneReceiving && streamingDoneSending &&
1912 312 : !pq_is_send_pending())
1913 312 : break;
1914 :
1915 : /* die if timeout was reached */
1916 5134 : WalSndCheckTimeOut();
1917 :
1918 : /* Send keepalive if the time has come */
1919 5134 : WalSndKeepaliveIfNecessary();
1920 :
1921 : /*
1922 : * Sleep until something happens or we time out. Also wait for the
1923 : * socket becoming writable, if there's still pending output.
1924 : * Otherwise we might sit on sendable output data while waiting for
1925 : * new WAL to be generated. (But if we have nothing to send, we don't
1926 : * want to wake on socket-writable.)
1927 : */
1928 5134 : sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
1929 :
1930 5134 : wakeEvents = WL_SOCKET_READABLE;
1931 :
1932 5134 : if (pq_is_send_pending())
1933 0 : wakeEvents |= WL_SOCKET_WRITEABLE;
1934 :
1935 : Assert(wait_event != 0);
1936 :
1937 5134 : WalSndWait(wakeEvents, sleeptime, wait_event);
1938 : }
1939 :
1940 : /* reactivate latch so WalSndLoop knows to continue */
1941 3896 : SetLatch(MyLatch);
1942 3896 : return RecentFlushPtr;
1943 : }
1944 :
1945 : /*
1946 : * Execute an incoming replication command.
1947 : *
1948 : * Returns true if the cmd_string was recognized as WalSender command, false
1949 : * if not.
1950 : */
1951 : bool
1952 9760 : exec_replication_command(const char *cmd_string)
1953 : {
1954 : yyscan_t scanner;
1955 : int parse_rc;
1956 : Node *cmd_node;
1957 : const char *cmdtag;
1958 : MemoryContext cmd_context;
1959 : MemoryContext old_context;
1960 :
1961 : /*
1962 : * If WAL sender has been told that shutdown is getting close, switch its
1963 : * status accordingly to handle the next replication commands correctly.
1964 : */
1965 9760 : if (got_STOPPING)
1966 0 : WalSndSetState(WALSNDSTATE_STOPPING);
1967 :
1968 : /*
1969 : * Throw error if in stopping mode. We need prevent commands that could
1970 : * generate WAL while the shutdown checkpoint is being written. To be
1971 : * safe, we just prohibit all new commands.
1972 : */
1973 9760 : if (MyWalSnd->state == WALSNDSTATE_STOPPING)
1974 0 : ereport(ERROR,
1975 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1976 : errmsg("cannot execute new commands while WAL sender is in stopping mode")));
1977 :
1978 : /*
1979 : * CREATE_REPLICATION_SLOT ... LOGICAL exports a snapshot until the next
1980 : * command arrives. Clean up the old stuff if there's anything.
1981 : */
1982 9760 : SnapBuildClearExportedSnapshot();
1983 :
1984 9760 : CHECK_FOR_INTERRUPTS();
1985 :
1986 : /*
1987 : * Prepare to parse and execute the command.
1988 : */
1989 9760 : cmd_context = AllocSetContextCreate(CurrentMemoryContext,
1990 : "Replication command context",
1991 : ALLOCSET_DEFAULT_SIZES);
1992 9760 : old_context = MemoryContextSwitchTo(cmd_context);
1993 :
1994 9760 : replication_scanner_init(cmd_string, &scanner);
1995 :
1996 : /*
1997 : * Is it a WalSender command?
1998 : */
1999 9760 : if (!replication_scanner_is_replication_command(scanner))
2000 : {
2001 : /* Nope; clean up and get out. */
2002 4290 : replication_scanner_finish(scanner);
2003 :
2004 4290 : MemoryContextSwitchTo(old_context);
2005 4290 : MemoryContextDelete(cmd_context);
2006 :
2007 : /* XXX this is a pretty random place to make this check */
2008 4290 : if (MyDatabaseId == InvalidOid)
2009 0 : ereport(ERROR,
2010 : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2011 : errmsg("cannot execute SQL commands in WAL sender for physical replication")));
2012 :
2013 : /* Tell the caller that this wasn't a WalSender command. */
2014 4290 : return false;
2015 : }
2016 :
2017 : /*
2018 : * Looks like a WalSender command, so parse it.
2019 : */
2020 5470 : parse_rc = replication_yyparse(&cmd_node, scanner);
2021 5470 : if (parse_rc != 0)
2022 0 : ereport(ERROR,
2023 : (errcode(ERRCODE_SYNTAX_ERROR),
2024 : errmsg_internal("replication command parser returned %d",
2025 : parse_rc)));
2026 5470 : replication_scanner_finish(scanner);
2027 :
2028 : /*
2029 : * Report query to various monitoring facilities. For this purpose, we
2030 : * report replication commands just like SQL commands.
2031 : */
2032 5470 : debug_query_string = cmd_string;
2033 :
2034 5470 : pgstat_report_activity(STATE_RUNNING, cmd_string);
2035 :
2036 : /*
2037 : * Log replication command if log_replication_commands is enabled. Even
2038 : * when it's disabled, log the command with DEBUG1 level for backward
2039 : * compatibility.
2040 : */
2041 5470 : ereport(log_replication_commands ? LOG : DEBUG1,
2042 : (errmsg("received replication command: %s", cmd_string)));
2043 :
2044 : /*
2045 : * Disallow replication commands in aborted transaction blocks.
2046 : */
2047 5470 : if (IsAbortedTransactionBlockState())
2048 0 : ereport(ERROR,
2049 : (errcode(ERRCODE_IN_FAILED_SQL_TRANSACTION),
2050 : errmsg("current transaction is aborted, "
2051 : "commands ignored until end of transaction block")));
2052 :
2053 5470 : CHECK_FOR_INTERRUPTS();
2054 :
2055 : /*
2056 : * Allocate buffers that will be used for each outgoing and incoming
2057 : * message. We do this just once per command to reduce palloc overhead.
2058 : */
2059 5470 : initStringInfo(&output_message);
2060 5470 : initStringInfo(&reply_message);
2061 5470 : initStringInfo(&tmpbuf);
2062 :
2063 5470 : switch (cmd_node->type)
2064 : {
2065 1312 : case T_IdentifySystemCmd:
2066 1312 : cmdtag = "IDENTIFY_SYSTEM";
2067 1312 : set_ps_display(cmdtag);
2068 1312 : IdentifySystem();
2069 1312 : EndReplicationCommand(cmdtag);
2070 1312 : break;
2071 :
2072 12 : case T_ReadReplicationSlotCmd:
2073 12 : cmdtag = "READ_REPLICATION_SLOT";
2074 12 : set_ps_display(cmdtag);
2075 12 : ReadReplicationSlot((ReadReplicationSlotCmd *) cmd_node);
2076 10 : EndReplicationCommand(cmdtag);
2077 10 : break;
2078 :
2079 352 : case T_BaseBackupCmd:
2080 352 : cmdtag = "BASE_BACKUP";
2081 352 : set_ps_display(cmdtag);
2082 352 : PreventInTransactionBlock(true, cmdtag);
2083 352 : SendBaseBackup((BaseBackupCmd *) cmd_node, uploaded_manifest);
2084 296 : EndReplicationCommand(cmdtag);
2085 296 : break;
2086 :
2087 894 : case T_CreateReplicationSlotCmd:
2088 894 : cmdtag = "CREATE_REPLICATION_SLOT";
2089 894 : set_ps_display(cmdtag);
2090 894 : CreateReplicationSlot((CreateReplicationSlotCmd *) cmd_node);
2091 892 : EndReplicationCommand(cmdtag);
2092 892 : break;
2093 :
2094 496 : case T_DropReplicationSlotCmd:
2095 496 : cmdtag = "DROP_REPLICATION_SLOT";
2096 496 : set_ps_display(cmdtag);
2097 496 : DropReplicationSlot((DropReplicationSlotCmd *) cmd_node);
2098 496 : EndReplicationCommand(cmdtag);
2099 496 : break;
2100 :
2101 12 : case T_AlterReplicationSlotCmd:
2102 12 : cmdtag = "ALTER_REPLICATION_SLOT";
2103 12 : set_ps_display(cmdtag);
2104 12 : AlterReplicationSlot((AlterReplicationSlotCmd *) cmd_node);
2105 8 : EndReplicationCommand(cmdtag);
2106 8 : break;
2107 :
2108 1240 : case T_StartReplicationCmd:
2109 : {
2110 1240 : StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
2111 :
2112 1240 : cmdtag = "START_REPLICATION";
2113 1240 : set_ps_display(cmdtag);
2114 1240 : PreventInTransactionBlock(true, cmdtag);
2115 :
2116 1240 : if (cmd->kind == REPLICATION_KIND_PHYSICAL)
2117 512 : StartReplication(cmd);
2118 : else
2119 728 : StartLogicalReplication(cmd);
2120 :
2121 : /* dupe, but necessary per libpqrcv_endstreaming */
2122 656 : EndReplicationCommand(cmdtag);
2123 :
2124 : Assert(xlogreader != NULL);
2125 656 : break;
2126 : }
2127 :
2128 28 : case T_TimeLineHistoryCmd:
2129 28 : cmdtag = "TIMELINE_HISTORY";
2130 28 : set_ps_display(cmdtag);
2131 28 : PreventInTransactionBlock(true, cmdtag);
2132 28 : SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);
2133 28 : EndReplicationCommand(cmdtag);
2134 28 : break;
2135 :
2136 1102 : case T_VariableShowStmt:
2137 : {
2138 1102 : DestReceiver *dest = CreateDestReceiver(DestRemoteSimple);
2139 1102 : VariableShowStmt *n = (VariableShowStmt *) cmd_node;
2140 :
2141 1102 : cmdtag = "SHOW";
2142 1102 : set_ps_display(cmdtag);
2143 :
2144 : /* syscache access needs a transaction environment */
2145 1102 : StartTransactionCommand();
2146 1102 : GetPGVariable(n->name, dest);
2147 1102 : CommitTransactionCommand();
2148 1102 : EndReplicationCommand(cmdtag);
2149 : }
2150 1102 : break;
2151 :
2152 22 : case T_UploadManifestCmd:
2153 22 : cmdtag = "UPLOAD_MANIFEST";
2154 22 : set_ps_display(cmdtag);
2155 22 : PreventInTransactionBlock(true, cmdtag);
2156 22 : UploadManifest();
2157 20 : EndReplicationCommand(cmdtag);
2158 20 : break;
2159 :
2160 0 : default:
2161 0 : elog(ERROR, "unrecognized replication command node tag: %u",
2162 : cmd_node->type);
2163 : }
2164 :
2165 : /* done */
2166 4820 : MemoryContextSwitchTo(old_context);
2167 4820 : MemoryContextDelete(cmd_context);
2168 :
2169 : /*
2170 : * We need not update ps display or pg_stat_activity, because PostgresMain
2171 : * will reset those to "idle". But we must reset debug_query_string to
2172 : * ensure it doesn't become a dangling pointer.
2173 : */
2174 4820 : debug_query_string = NULL;
2175 :
2176 4820 : return true;
2177 : }
2178 :
2179 : /*
2180 : * Process any incoming messages while streaming. Also checks if the remote
2181 : * end has closed the connection.
2182 : */
2183 : static void
2184 2424672 : ProcessRepliesIfAny(void)
2185 : {
2186 : unsigned char firstchar;
2187 : int maxmsglen;
2188 : int r;
2189 2424672 : bool received = false;
2190 :
2191 2424672 : last_processing = GetCurrentTimestamp();
2192 :
2193 : /*
2194 : * If we already received a CopyDone from the frontend, any subsequent
2195 : * message is the beginning of a new command, and should be processed in
2196 : * the main processing loop.
2197 : */
2198 2424672 : while (!streamingDoneReceiving)
2199 : {
2200 2515596 : pq_startmsgread();
2201 2515596 : r = pq_getbyte_if_available(&firstchar);
2202 2515596 : if (r < 0)
2203 : {
2204 : /* unexpected error or EOF */
2205 34 : ereport(COMMERROR,
2206 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
2207 : errmsg("unexpected EOF on standby connection")));
2208 34 : proc_exit(0);
2209 : }
2210 2515562 : if (r == 0)
2211 : {
2212 : /* no data available without blocking */
2213 2422860 : pq_endmsgread();
2214 2422860 : break;
2215 : }
2216 :
2217 : /* Validate message type and set packet size limit */
2218 92702 : switch (firstchar)
2219 : {
2220 91612 : case PqMsg_CopyData:
2221 91612 : maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
2222 91612 : break;
2223 1090 : case PqMsg_CopyDone:
2224 : case PqMsg_Terminate:
2225 1090 : maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
2226 1090 : break;
2227 0 : default:
2228 0 : ereport(FATAL,
2229 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
2230 : errmsg("invalid standby message type \"%c\"",
2231 : firstchar)));
2232 : maxmsglen = 0; /* keep compiler quiet */
2233 : break;
2234 : }
2235 :
2236 : /* Read the message contents */
2237 92702 : resetStringInfo(&reply_message);
2238 92702 : if (pq_getmessage(&reply_message, maxmsglen))
2239 : {
2240 0 : ereport(COMMERROR,
2241 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
2242 : errmsg("unexpected EOF on standby connection")));
2243 0 : proc_exit(0);
2244 : }
2245 :
2246 : /* ... and process it */
2247 92702 : switch (firstchar)
2248 : {
2249 : /*
2250 : * 'd' means a standby reply wrapped in a CopyData packet.
2251 : */
2252 91612 : case PqMsg_CopyData:
2253 91612 : ProcessStandbyMessage();
2254 91612 : received = true;
2255 91612 : break;
2256 :
2257 : /*
2258 : * CopyDone means the standby requested to finish streaming.
2259 : * Reply with CopyDone, if we had not sent that already.
2260 : */
2261 656 : case PqMsg_CopyDone:
2262 656 : if (!streamingDoneSending)
2263 : {
2264 630 : pq_putmessage_noblock('c', NULL, 0);
2265 630 : streamingDoneSending = true;
2266 : }
2267 :
2268 656 : streamingDoneReceiving = true;
2269 656 : received = true;
2270 656 : break;
2271 :
2272 : /*
2273 : * 'X' means that the standby is closing down the socket.
2274 : */
2275 434 : case PqMsg_Terminate:
2276 434 : proc_exit(0);
2277 :
2278 2516940 : default:
2279 : Assert(false); /* NOT REACHED */
2280 : }
2281 : }
2282 :
2283 : /*
2284 : * Save the last reply timestamp if we've received at least one reply.
2285 : */
2286 2424204 : if (received)
2287 : {
2288 36974 : last_reply_timestamp = last_processing;
2289 36974 : waiting_for_ping_response = false;
2290 : }
2291 2424204 : }
2292 :
2293 : /*
2294 : * Process a status update message received from standby.
2295 : */
2296 : static void
2297 91612 : ProcessStandbyMessage(void)
2298 : {
2299 : char msgtype;
2300 :
2301 : /*
2302 : * Check message type from the first byte.
2303 : */
2304 91612 : msgtype = pq_getmsgbyte(&reply_message);
2305 :
2306 91612 : switch (msgtype)
2307 : {
2308 91346 : case 'r':
2309 91346 : ProcessStandbyReplyMessage();
2310 91346 : break;
2311 :
2312 266 : case 'h':
2313 266 : ProcessStandbyHSFeedbackMessage();
2314 266 : break;
2315 :
2316 0 : default:
2317 0 : ereport(COMMERROR,
2318 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
2319 : errmsg("unexpected message type \"%c\"", msgtype)));
2320 0 : proc_exit(0);
2321 : }
2322 91612 : }
2323 :
2324 : /*
2325 : * Remember that a walreceiver just confirmed receipt of lsn `lsn`.
2326 : */
2327 : static void
2328 36196 : PhysicalConfirmReceivedLocation(XLogRecPtr lsn)
2329 : {
2330 36196 : bool changed = false;
2331 36196 : ReplicationSlot *slot = MyReplicationSlot;
2332 :
2333 : Assert(lsn != InvalidXLogRecPtr);
2334 36196 : SpinLockAcquire(&slot->mutex);
2335 36196 : if (slot->data.restart_lsn != lsn)
2336 : {
2337 15386 : changed = true;
2338 15386 : slot->data.restart_lsn = lsn;
2339 : }
2340 36196 : SpinLockRelease(&slot->mutex);
2341 :
2342 36196 : if (changed)
2343 : {
2344 15386 : ReplicationSlotMarkDirty();
2345 15386 : ReplicationSlotsComputeRequiredLSN();
2346 15386 : PhysicalWakeupLogicalWalSnd();
2347 : }
2348 :
2349 : /*
2350 : * One could argue that the slot should be saved to disk now, but that'd
2351 : * be energy wasted - the worst thing lost information could cause here is
2352 : * to give wrong information in a statistics view - we'll just potentially
2353 : * be more conservative in removing files.
2354 : */
2355 36196 : }
2356 :
2357 : /*
2358 : * Regular reply from standby advising of WAL locations on standby server.
2359 : */
2360 : static void
2361 91346 : ProcessStandbyReplyMessage(void)
2362 : {
2363 : XLogRecPtr writePtr,
2364 : flushPtr,
2365 : applyPtr;
2366 : bool replyRequested;
2367 : TimeOffset writeLag,
2368 : flushLag,
2369 : applyLag;
2370 : bool clearLagTimes;
2371 : TimestampTz now;
2372 : TimestampTz replyTime;
2373 :
2374 : static bool fullyAppliedLastTime = false;
2375 :
2376 : /* the caller already consumed the msgtype byte */
2377 91346 : writePtr = pq_getmsgint64(&reply_message);
2378 91346 : flushPtr = pq_getmsgint64(&reply_message);
2379 91346 : applyPtr = pq_getmsgint64(&reply_message);
2380 91346 : replyTime = pq_getmsgint64(&reply_message);
2381 91346 : replyRequested = pq_getmsgbyte(&reply_message);
2382 :
2383 91346 : if (message_level_is_interesting(DEBUG2))
2384 : {
2385 : char *replyTimeStr;
2386 :
2387 : /* Copy because timestamptz_to_str returns a static buffer */
2388 530 : replyTimeStr = pstrdup(timestamptz_to_str(replyTime));
2389 :
2390 530 : elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X%s reply_time %s",
2391 : LSN_FORMAT_ARGS(writePtr),
2392 : LSN_FORMAT_ARGS(flushPtr),
2393 : LSN_FORMAT_ARGS(applyPtr),
2394 : replyRequested ? " (reply requested)" : "",
2395 : replyTimeStr);
2396 :
2397 530 : pfree(replyTimeStr);
2398 : }
2399 :
2400 : /* See if we can compute the round-trip lag for these positions. */
2401 91346 : now = GetCurrentTimestamp();
2402 91346 : writeLag = LagTrackerRead(SYNC_REP_WAIT_WRITE, writePtr, now);
2403 91346 : flushLag = LagTrackerRead(SYNC_REP_WAIT_FLUSH, flushPtr, now);
2404 91346 : applyLag = LagTrackerRead(SYNC_REP_WAIT_APPLY, applyPtr, now);
2405 :
2406 : /*
2407 : * If the standby reports that it has fully replayed the WAL in two
2408 : * consecutive reply messages, then the second such message must result
2409 : * from wal_receiver_status_interval expiring on the standby. This is a
2410 : * convenient time to forget the lag times measured when it last
2411 : * wrote/flushed/applied a WAL record, to avoid displaying stale lag data
2412 : * until more WAL traffic arrives.
2413 : */
2414 91346 : clearLagTimes = false;
2415 91346 : if (applyPtr == sentPtr)
2416 : {
2417 6572 : if (fullyAppliedLastTime)
2418 1830 : clearLagTimes = true;
2419 6572 : fullyAppliedLastTime = true;
2420 : }
2421 : else
2422 84774 : fullyAppliedLastTime = false;
2423 :
2424 : /* Send a reply if the standby requested one. */
2425 91346 : if (replyRequested)
2426 0 : WalSndKeepalive(false, InvalidXLogRecPtr);
2427 :
2428 : /*
2429 : * Update shared state for this WalSender process based on reply data from
2430 : * standby.
2431 : */
2432 : {
2433 91346 : WalSnd *walsnd = MyWalSnd;
2434 :
2435 91346 : SpinLockAcquire(&walsnd->mutex);
2436 91346 : walsnd->write = writePtr;
2437 91346 : walsnd->flush = flushPtr;
2438 91346 : walsnd->apply = applyPtr;
2439 91346 : if (writeLag != -1 || clearLagTimes)
2440 26778 : walsnd->writeLag = writeLag;
2441 91346 : if (flushLag != -1 || clearLagTimes)
2442 43194 : walsnd->flushLag = flushLag;
2443 91346 : if (applyLag != -1 || clearLagTimes)
2444 47126 : walsnd->applyLag = applyLag;
2445 91346 : walsnd->replyTime = replyTime;
2446 91346 : SpinLockRelease(&walsnd->mutex);
2447 : }
2448 :
2449 91346 : if (!am_cascading_walsender)
2450 90726 : SyncRepReleaseWaiters();
2451 :
2452 : /*
2453 : * Advance our local xmin horizon when the client confirmed a flush.
2454 : */
2455 91346 : if (MyReplicationSlot && flushPtr != InvalidXLogRecPtr)
2456 : {
2457 82024 : if (SlotIsLogical(MyReplicationSlot))
2458 45828 : LogicalConfirmReceivedLocation(flushPtr);
2459 : else
2460 36196 : PhysicalConfirmReceivedLocation(flushPtr);
2461 : }
2462 91346 : }
2463 :
2464 : /* compute new replication slot xmin horizon if needed */
2465 : static void
2466 124 : PhysicalReplicationSlotNewXmin(TransactionId feedbackXmin, TransactionId feedbackCatalogXmin)
2467 : {
2468 124 : bool changed = false;
2469 124 : ReplicationSlot *slot = MyReplicationSlot;
2470 :
2471 124 : SpinLockAcquire(&slot->mutex);
2472 124 : MyProc->xmin = InvalidTransactionId;
2473 :
2474 : /*
2475 : * For physical replication we don't need the interlock provided by xmin
2476 : * and effective_xmin since the consequences of a missed increase are
2477 : * limited to query cancellations, so set both at once.
2478 : */
2479 124 : if (!TransactionIdIsNormal(slot->data.xmin) ||
2480 60 : !TransactionIdIsNormal(feedbackXmin) ||
2481 60 : TransactionIdPrecedes(slot->data.xmin, feedbackXmin))
2482 : {
2483 82 : changed = true;
2484 82 : slot->data.xmin = feedbackXmin;
2485 82 : slot->effective_xmin = feedbackXmin;
2486 : }
2487 124 : if (!TransactionIdIsNormal(slot->data.catalog_xmin) ||
2488 30 : !TransactionIdIsNormal(feedbackCatalogXmin) ||
2489 30 : TransactionIdPrecedes(slot->data.catalog_xmin, feedbackCatalogXmin))
2490 : {
2491 96 : changed = true;
2492 96 : slot->data.catalog_xmin = feedbackCatalogXmin;
2493 96 : slot->effective_catalog_xmin = feedbackCatalogXmin;
2494 : }
2495 124 : SpinLockRelease(&slot->mutex);
2496 :
2497 124 : if (changed)
2498 : {
2499 102 : ReplicationSlotMarkDirty();
2500 102 : ReplicationSlotsComputeRequiredXmin(false);
2501 : }
2502 124 : }
2503 :
2504 : /*
2505 : * Check that the provided xmin/epoch are sane, that is, not in the future
2506 : * and not so far back as to be already wrapped around.
2507 : *
2508 : * Epoch of nextXid should be same as standby, or if the counter has
2509 : * wrapped, then one greater than standby.
2510 : *
2511 : * This check doesn't care about whether clog exists for these xids
2512 : * at all.
2513 : */
2514 : static bool
2515 124 : TransactionIdInRecentPast(TransactionId xid, uint32 epoch)
2516 : {
2517 : FullTransactionId nextFullXid;
2518 : TransactionId nextXid;
2519 : uint32 nextEpoch;
2520 :
2521 124 : nextFullXid = ReadNextFullTransactionId();
2522 124 : nextXid = XidFromFullTransactionId(nextFullXid);
2523 124 : nextEpoch = EpochFromFullTransactionId(nextFullXid);
2524 :
2525 124 : if (xid <= nextXid)
2526 : {
2527 124 : if (epoch != nextEpoch)
2528 0 : return false;
2529 : }
2530 : else
2531 : {
2532 0 : if (epoch + 1 != nextEpoch)
2533 0 : return false;
2534 : }
2535 :
2536 124 : if (!TransactionIdPrecedesOrEquals(xid, nextXid))
2537 0 : return false; /* epoch OK, but it's wrapped around */
2538 :
2539 124 : return true;
2540 : }
2541 :
2542 : /*
2543 : * Hot Standby feedback
2544 : */
2545 : static void
2546 266 : ProcessStandbyHSFeedbackMessage(void)
2547 : {
2548 : TransactionId feedbackXmin;
2549 : uint32 feedbackEpoch;
2550 : TransactionId feedbackCatalogXmin;
2551 : uint32 feedbackCatalogEpoch;
2552 : TimestampTz replyTime;
2553 :
2554 : /*
2555 : * Decipher the reply message. The caller already consumed the msgtype
2556 : * byte. See XLogWalRcvSendHSFeedback() in walreceiver.c for the creation
2557 : * of this message.
2558 : */
2559 266 : replyTime = pq_getmsgint64(&reply_message);
2560 266 : feedbackXmin = pq_getmsgint(&reply_message, 4);
2561 266 : feedbackEpoch = pq_getmsgint(&reply_message, 4);
2562 266 : feedbackCatalogXmin = pq_getmsgint(&reply_message, 4);
2563 266 : feedbackCatalogEpoch = pq_getmsgint(&reply_message, 4);
2564 :
2565 266 : if (message_level_is_interesting(DEBUG2))
2566 : {
2567 : char *replyTimeStr;
2568 :
2569 : /* Copy because timestamptz_to_str returns a static buffer */
2570 8 : replyTimeStr = pstrdup(timestamptz_to_str(replyTime));
2571 :
2572 8 : elog(DEBUG2, "hot standby feedback xmin %u epoch %u, catalog_xmin %u epoch %u reply_time %s",
2573 : feedbackXmin,
2574 : feedbackEpoch,
2575 : feedbackCatalogXmin,
2576 : feedbackCatalogEpoch,
2577 : replyTimeStr);
2578 :
2579 8 : pfree(replyTimeStr);
2580 : }
2581 :
2582 : /*
2583 : * Update shared state for this WalSender process based on reply data from
2584 : * standby.
2585 : */
2586 : {
2587 266 : WalSnd *walsnd = MyWalSnd;
2588 :
2589 266 : SpinLockAcquire(&walsnd->mutex);
2590 266 : walsnd->replyTime = replyTime;
2591 266 : SpinLockRelease(&walsnd->mutex);
2592 : }
2593 :
2594 : /*
2595 : * Unset WalSender's xmins if the feedback message values are invalid.
2596 : * This happens when the downstream turned hot_standby_feedback off.
2597 : */
2598 266 : if (!TransactionIdIsNormal(feedbackXmin)
2599 184 : && !TransactionIdIsNormal(feedbackCatalogXmin))
2600 : {
2601 184 : MyProc->xmin = InvalidTransactionId;
2602 184 : if (MyReplicationSlot != NULL)
2603 44 : PhysicalReplicationSlotNewXmin(feedbackXmin, feedbackCatalogXmin);
2604 184 : return;
2605 : }
2606 :
2607 : /*
2608 : * Check that the provided xmin/epoch are sane, that is, not in the future
2609 : * and not so far back as to be already wrapped around. Ignore if not.
2610 : */
2611 82 : if (TransactionIdIsNormal(feedbackXmin) &&
2612 82 : !TransactionIdInRecentPast(feedbackXmin, feedbackEpoch))
2613 0 : return;
2614 :
2615 82 : if (TransactionIdIsNormal(feedbackCatalogXmin) &&
2616 42 : !TransactionIdInRecentPast(feedbackCatalogXmin, feedbackCatalogEpoch))
2617 0 : return;
2618 :
2619 : /*
2620 : * Set the WalSender's xmin equal to the standby's requested xmin, so that
2621 : * the xmin will be taken into account by GetSnapshotData() /
2622 : * ComputeXidHorizons(). This will hold back the removal of dead rows and
2623 : * thereby prevent the generation of cleanup conflicts on the standby
2624 : * server.
2625 : *
2626 : * There is a small window for a race condition here: although we just
2627 : * checked that feedbackXmin precedes nextXid, the nextXid could have
2628 : * gotten advanced between our fetching it and applying the xmin below,
2629 : * perhaps far enough to make feedbackXmin wrap around. In that case the
2630 : * xmin we set here would be "in the future" and have no effect. No point
2631 : * in worrying about this since it's too late to save the desired data
2632 : * anyway. Assuming that the standby sends us an increasing sequence of
2633 : * xmins, this could only happen during the first reply cycle, else our
2634 : * own xmin would prevent nextXid from advancing so far.
2635 : *
2636 : * We don't bother taking the ProcArrayLock here. Setting the xmin field
2637 : * is assumed atomic, and there's no real need to prevent concurrent
2638 : * horizon determinations. (If we're moving our xmin forward, this is
2639 : * obviously safe, and if we're moving it backwards, well, the data is at
2640 : * risk already since a VACUUM could already have determined the horizon.)
2641 : *
2642 : * If we're using a replication slot we reserve the xmin via that,
2643 : * otherwise via the walsender's PGPROC entry. We can only track the
2644 : * catalog xmin separately when using a slot, so we store the least of the
2645 : * two provided when not using a slot.
2646 : *
2647 : * XXX: It might make sense to generalize the ephemeral slot concept and
2648 : * always use the slot mechanism to handle the feedback xmin.
2649 : */
2650 82 : if (MyReplicationSlot != NULL) /* XXX: persistency configurable? */
2651 80 : PhysicalReplicationSlotNewXmin(feedbackXmin, feedbackCatalogXmin);
2652 : else
2653 : {
2654 2 : if (TransactionIdIsNormal(feedbackCatalogXmin)
2655 0 : && TransactionIdPrecedes(feedbackCatalogXmin, feedbackXmin))
2656 0 : MyProc->xmin = feedbackCatalogXmin;
2657 : else
2658 2 : MyProc->xmin = feedbackXmin;
2659 : }
2660 : }
2661 :
2662 : /*
2663 : * Compute how long send/receive loops should sleep.
2664 : *
2665 : * If wal_sender_timeout is enabled we want to wake up in time to send
2666 : * keepalives and to abort the connection if wal_sender_timeout has been
2667 : * reached.
2668 : */
2669 : static long
2670 110272 : WalSndComputeSleeptime(TimestampTz now)
2671 : {
2672 110272 : long sleeptime = 10000; /* 10 s */
2673 :
2674 110272 : if (wal_sender_timeout > 0 && last_reply_timestamp > 0)
2675 : {
2676 : TimestampTz wakeup_time;
2677 :
2678 : /*
2679 : * At the latest stop sleeping once wal_sender_timeout has been
2680 : * reached.
2681 : */
2682 110224 : wakeup_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
2683 : wal_sender_timeout);
2684 :
2685 : /*
2686 : * If no ping has been sent yet, wakeup when it's time to do so.
2687 : * WalSndKeepaliveIfNecessary() wants to send a keepalive once half of
2688 : * the timeout passed without a response.
2689 : */
2690 110224 : if (!waiting_for_ping_response)
2691 93862 : wakeup_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
2692 : wal_sender_timeout / 2);
2693 :
2694 : /* Compute relative time until wakeup. */
2695 110224 : sleeptime = TimestampDifferenceMilliseconds(now, wakeup_time);
2696 : }
2697 :
2698 110272 : return sleeptime;
2699 : }
2700 :
2701 : /*
2702 : * Check whether there have been responses by the client within
2703 : * wal_sender_timeout and shutdown if not. Using last_processing as the
2704 : * reference point avoids counting server-side stalls against the client.
2705 : * However, a long server-side stall can make WalSndKeepaliveIfNecessary()
2706 : * postdate last_processing by more than wal_sender_timeout. If that happens,
2707 : * the client must reply almost immediately to avoid a timeout. This rarely
2708 : * affects the default configuration, under which clients spontaneously send a
2709 : * message every standby_message_timeout = wal_sender_timeout/6 = 10s. We
2710 : * could eliminate that problem by recognizing timeout expiration at
2711 : * wal_sender_timeout/2 after the keepalive.
2712 : */
2713 : static void
2714 2419248 : WalSndCheckTimeOut(void)
2715 : {
2716 : TimestampTz timeout;
2717 :
2718 : /* don't bail out if we're doing something that doesn't require timeouts */
2719 2419248 : if (last_reply_timestamp <= 0)
2720 48 : return;
2721 :
2722 2419200 : timeout = TimestampTzPlusMilliseconds(last_reply_timestamp,
2723 : wal_sender_timeout);
2724 :
2725 2419200 : if (wal_sender_timeout > 0 && last_processing >= timeout)
2726 : {
2727 : /*
2728 : * Since typically expiration of replication timeout means
2729 : * communication problem, we don't send the error message to the
2730 : * standby.
2731 : */
2732 0 : ereport(COMMERROR,
2733 : (errmsg("terminating walsender process due to replication timeout")));
2734 :
2735 0 : WalSndShutdown();
2736 : }
2737 : }
2738 :
2739 : /* Main loop of walsender process that streams the WAL over Copy messages. */
2740 : static void
2741 1218 : WalSndLoop(WalSndSendDataCallback send_data)
2742 : {
2743 : /*
2744 : * Initialize the last reply timestamp. That enables timeout processing
2745 : * from hereon.
2746 : */
2747 1218 : last_reply_timestamp = GetCurrentTimestamp();
2748 1218 : waiting_for_ping_response = false;
2749 :
2750 : /*
2751 : * Loop until we reach the end of this timeline or the client requests to
2752 : * stop streaming.
2753 : */
2754 : for (;;)
2755 : {
2756 : /* Clear any already-pending wakeups */
2757 2415172 : ResetLatch(MyLatch);
2758 :
2759 2415172 : CHECK_FOR_INTERRUPTS();
2760 :
2761 : /* Process any requests or signals received recently */
2762 2415166 : if (ConfigReloadPending)
2763 : {
2764 32 : ConfigReloadPending = false;
2765 32 : ProcessConfigFile(PGC_SIGHUP);
2766 32 : SyncRepInitConfig();
2767 : }
2768 :
2769 : /* Check for input from the client */
2770 2415166 : ProcessRepliesIfAny();
2771 :
2772 : /*
2773 : * If we have received CopyDone from the client, sent CopyDone
2774 : * ourselves, and the output buffer is empty, it's time to exit
2775 : * streaming.
2776 : */
2777 2415014 : if (streamingDoneReceiving && streamingDoneSending &&
2778 1032 : !pq_is_send_pending())
2779 656 : break;
2780 :
2781 : /*
2782 : * If we don't have any pending data in the output buffer, try to send
2783 : * some more. If there is some, we don't bother to call send_data
2784 : * again until we've flushed it ... but we'd better assume we are not
2785 : * caught up.
2786 : */
2787 2414358 : if (!pq_is_send_pending())
2788 2342398 : send_data();
2789 : else
2790 71960 : WalSndCaughtUp = false;
2791 :
2792 : /* Try to flush pending output to the client */
2793 2414022 : if (pq_flush_if_writable() != 0)
2794 0 : WalSndShutdown();
2795 :
2796 : /* If nothing remains to be sent right now ... */
2797 2414022 : if (WalSndCaughtUp && !pq_is_send_pending())
2798 : {
2799 : /*
2800 : * If we're in catchup state, move to streaming. This is an
2801 : * important state change for users to know about, since before
2802 : * this point data loss might occur if the primary dies and we
2803 : * need to failover to the standby. The state change is also
2804 : * important for synchronous replication, since commits that
2805 : * started to wait at that point might wait for some time.
2806 : */
2807 506388 : if (MyWalSnd->state == WALSNDSTATE_CATCHUP)
2808 : {
2809 1186 : ereport(DEBUG1,
2810 : (errmsg_internal("\"%s\" has now caught up with upstream server",
2811 : application_name)));
2812 1186 : WalSndSetState(WALSNDSTATE_STREAMING);
2813 : }
2814 :
2815 : /*
2816 : * When SIGUSR2 arrives, we send any outstanding logs up to the
2817 : * shutdown checkpoint record (i.e., the latest record), wait for
2818 : * them to be replicated to the standby, and exit. This may be a
2819 : * normal termination at shutdown, or a promotion, the walsender
2820 : * is not sure which.
2821 : */
2822 506388 : if (got_SIGUSR2)
2823 16462 : WalSndDone(send_data);
2824 : }
2825 :
2826 : /* Check for replication timeout. */
2827 2413954 : WalSndCheckTimeOut();
2828 :
2829 : /* Send keepalive if the time has come */
2830 2413954 : WalSndKeepaliveIfNecessary();
2831 :
2832 : /*
2833 : * Block if we have unsent data. XXX For logical replication, let
2834 : * WalSndWaitForWal() handle any other blocking; idle receivers need
2835 : * its additional actions. For physical replication, also block if
2836 : * caught up; its send_data does not block.
2837 : */
2838 2413954 : if ((WalSndCaughtUp && send_data != XLogSendLogical &&
2839 2874310 : !streamingDoneSending) ||
2840 2372356 : pq_is_send_pending())
2841 : {
2842 : long sleeptime;
2843 : int wakeEvents;
2844 :
2845 105040 : if (!streamingDoneReceiving)
2846 104982 : wakeEvents = WL_SOCKET_READABLE;
2847 : else
2848 58 : wakeEvents = 0;
2849 :
2850 : /*
2851 : * Use fresh timestamp, not last_processing, to reduce the chance
2852 : * of reaching wal_sender_timeout before sending a keepalive.
2853 : */
2854 105040 : sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
2855 :
2856 105040 : if (pq_is_send_pending())
2857 71876 : wakeEvents |= WL_SOCKET_WRITEABLE;
2858 :
2859 : /* Sleep until something happens or we time out */
2860 105040 : WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_MAIN);
2861 : }
2862 : }
2863 656 : }
2864 :
2865 : /* Initialize a per-walsender data structure for this walsender process */
2866 : static void
2867 2122 : InitWalSenderSlot(void)
2868 : {
2869 : int i;
2870 :
2871 : /*
2872 : * WalSndCtl should be set up already (we inherit this by fork() or
2873 : * EXEC_BACKEND mechanism from the postmaster).
2874 : */
2875 : Assert(WalSndCtl != NULL);
2876 : Assert(MyWalSnd == NULL);
2877 :
2878 : /*
2879 : * Find a free walsender slot and reserve it. This must not fail due to
2880 : * the prior check for free WAL senders in InitProcess().
2881 : */
2882 3068 : for (i = 0; i < max_wal_senders; i++)
2883 : {
2884 3068 : WalSnd *walsnd = &WalSndCtl->walsnds[i];
2885 :
2886 3068 : SpinLockAcquire(&walsnd->mutex);
2887 :
2888 3068 : if (walsnd->pid != 0)
2889 : {
2890 946 : SpinLockRelease(&walsnd->mutex);
2891 946 : continue;
2892 : }
2893 : else
2894 : {
2895 : /*
2896 : * Found a free slot. Reserve it for us.
2897 : */
2898 2122 : walsnd->pid = MyProcPid;
2899 2122 : walsnd->state = WALSNDSTATE_STARTUP;
2900 2122 : walsnd->sentPtr = InvalidXLogRecPtr;
2901 2122 : walsnd->needreload = false;
2902 2122 : walsnd->write = InvalidXLogRecPtr;
2903 2122 : walsnd->flush = InvalidXLogRecPtr;
2904 2122 : walsnd->apply = InvalidXLogRecPtr;
2905 2122 : walsnd->writeLag = -1;
2906 2122 : walsnd->flushLag = -1;
2907 2122 : walsnd->applyLag = -1;
2908 2122 : walsnd->sync_standby_priority = 0;
2909 2122 : walsnd->replyTime = 0;
2910 :
2911 : /*
2912 : * The kind assignment is done here and not in StartReplication()
2913 : * and StartLogicalReplication(). Indeed, the logical walsender
2914 : * needs to read WAL records (like snapshot of running
2915 : * transactions) during the slot creation. So it needs to be woken
2916 : * up based on its kind.
2917 : *
2918 : * The kind assignment could also be done in StartReplication(),
2919 : * StartLogicalReplication() and CREATE_REPLICATION_SLOT but it
2920 : * seems better to set it on one place.
2921 : */
2922 2122 : if (MyDatabaseId == InvalidOid)
2923 892 : walsnd->kind = REPLICATION_KIND_PHYSICAL;
2924 : else
2925 1230 : walsnd->kind = REPLICATION_KIND_LOGICAL;
2926 :
2927 2122 : SpinLockRelease(&walsnd->mutex);
2928 : /* don't need the lock anymore */
2929 2122 : MyWalSnd = (WalSnd *) walsnd;
2930 :
2931 2122 : break;
2932 : }
2933 : }
2934 :
2935 : Assert(MyWalSnd != NULL);
2936 :
2937 : /* Arrange to clean up at walsender exit */
2938 2122 : on_shmem_exit(WalSndKill, 0);
2939 2122 : }
2940 :
2941 : /* Destroy the per-walsender data structure for this walsender process */
2942 : static void
2943 2122 : WalSndKill(int code, Datum arg)
2944 : {
2945 2122 : WalSnd *walsnd = MyWalSnd;
2946 :
2947 : Assert(walsnd != NULL);
2948 :
2949 2122 : MyWalSnd = NULL;
2950 :
2951 2122 : SpinLockAcquire(&walsnd->mutex);
2952 : /* Mark WalSnd struct as no longer being in use. */
2953 2122 : walsnd->pid = 0;
2954 2122 : SpinLockRelease(&walsnd->mutex);
2955 2122 : }
2956 :
2957 : /* XLogReaderRoutine->segment_open callback */
2958 : static void
2959 3260 : WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo,
2960 : TimeLineID *tli_p)
2961 : {
2962 : char path[MAXPGPATH];
2963 :
2964 : /*-------
2965 : * When reading from a historic timeline, and there is a timeline switch
2966 : * within this segment, read from the WAL segment belonging to the new
2967 : * timeline.
2968 : *
2969 : * For example, imagine that this server is currently on timeline 5, and
2970 : * we're streaming timeline 4. The switch from timeline 4 to 5 happened at
2971 : * 0/13002088. In pg_wal, we have these files:
2972 : *
2973 : * ...
2974 : * 000000040000000000000012
2975 : * 000000040000000000000013
2976 : * 000000050000000000000013
2977 : * 000000050000000000000014
2978 : * ...
2979 : *
2980 : * In this situation, when requested to send the WAL from segment 0x13, on
2981 : * timeline 4, we read the WAL from file 000000050000000000000013. Archive
2982 : * recovery prefers files from newer timelines, so if the segment was
2983 : * restored from the archive on this server, the file belonging to the old
2984 : * timeline, 000000040000000000000013, might not exist. Their contents are
2985 : * equal up to the switchpoint, because at a timeline switch, the used
2986 : * portion of the old segment is copied to the new file.
2987 : */
2988 3260 : *tli_p = sendTimeLine;
2989 3260 : if (sendTimeLineIsHistoric)
2990 : {
2991 : XLogSegNo endSegNo;
2992 :
2993 26 : XLByteToSeg(sendTimeLineValidUpto, endSegNo, state->segcxt.ws_segsize);
2994 26 : if (nextSegNo == endSegNo)
2995 20 : *tli_p = sendTimeLineNextTLI;
2996 : }
2997 :
2998 3260 : XLogFilePath(path, *tli_p, nextSegNo, state->segcxt.ws_segsize);
2999 3260 : state->seg.ws_file = BasicOpenFile(path, O_RDONLY | PG_BINARY);
3000 3260 : if (state->seg.ws_file >= 0)
3001 3260 : return;
3002 :
3003 : /*
3004 : * If the file is not found, assume it's because the standby asked for a
3005 : * too old WAL segment that has already been removed or recycled.
3006 : */
3007 0 : if (errno == ENOENT)
3008 : {
3009 : char xlogfname[MAXFNAMELEN];
3010 0 : int save_errno = errno;
3011 :
3012 0 : XLogFileName(xlogfname, *tli_p, nextSegNo, wal_segment_size);
3013 0 : errno = save_errno;
3014 0 : ereport(ERROR,
3015 : (errcode_for_file_access(),
3016 : errmsg("requested WAL segment %s has already been removed",
3017 : xlogfname)));
3018 : }
3019 : else
3020 0 : ereport(ERROR,
3021 : (errcode_for_file_access(),
3022 : errmsg("could not open file \"%s\": %m",
3023 : path)));
3024 : }
3025 :
3026 : /*
3027 : * Send out the WAL in its normal physical/stored form.
3028 : *
3029 : * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk,
3030 : * but not yet sent to the client, and buffer it in the libpq output
3031 : * buffer.
3032 : *
3033 : * If there is no unsent WAL remaining, WalSndCaughtUp is set to true,
3034 : * otherwise WalSndCaughtUp is set to false.
3035 : */
3036 : static void
3037 695194 : XLogSendPhysical(void)
3038 : {
3039 : XLogRecPtr SendRqstPtr;
3040 : XLogRecPtr startptr;
3041 : XLogRecPtr endptr;
3042 : Size nbytes;
3043 : XLogSegNo segno;
3044 : WALReadError errinfo;
3045 : Size rbytes;
3046 :
3047 : /* If requested switch the WAL sender to the stopping state. */
3048 695194 : if (got_STOPPING)
3049 33442 : WalSndSetState(WALSNDSTATE_STOPPING);
3050 :
3051 695194 : if (streamingDoneSending)
3052 : {
3053 460330 : WalSndCaughtUp = true;
3054 509108 : return;
3055 : }
3056 :
3057 : /* Figure out how far we can safely send the WAL. */
3058 234864 : if (sendTimeLineIsHistoric)
3059 : {
3060 : /*
3061 : * Streaming an old timeline that's in this server's history, but is
3062 : * not the one we're currently inserting or replaying. It can be
3063 : * streamed up to the point where we switched off that timeline.
3064 : */
3065 332 : SendRqstPtr = sendTimeLineValidUpto;
3066 : }
3067 234532 : else if (am_cascading_walsender)
3068 : {
3069 : TimeLineID SendRqstTLI;
3070 :
3071 : /*
3072 : * Streaming the latest timeline on a standby.
3073 : *
3074 : * Attempt to send all WAL that has already been replayed, so that we
3075 : * know it's valid. If we're receiving WAL through streaming
3076 : * replication, it's also OK to send any WAL that has been received
3077 : * but not replayed.
3078 : *
3079 : * The timeline we're recovering from can change, or we can be
3080 : * promoted. In either case, the current timeline becomes historic. We
3081 : * need to detect that so that we don't try to stream past the point
3082 : * where we switched to another timeline. We check for promotion or
3083 : * timeline switch after calculating FlushPtr, to avoid a race
3084 : * condition: if the timeline becomes historic just after we checked
3085 : * that it was still current, it's still be OK to stream it up to the
3086 : * FlushPtr that was calculated before it became historic.
3087 : */
3088 1536 : bool becameHistoric = false;
3089 :
3090 1536 : SendRqstPtr = GetStandbyFlushRecPtr(&SendRqstTLI);
3091 :
3092 1536 : if (!RecoveryInProgress())
3093 : {
3094 : /* We have been promoted. */
3095 2 : SendRqstTLI = GetWALInsertionTimeLine();
3096 2 : am_cascading_walsender = false;
3097 2 : becameHistoric = true;
3098 : }
3099 : else
3100 : {
3101 : /*
3102 : * Still a cascading standby. But is the timeline we're sending
3103 : * still the one recovery is recovering from?
3104 : */
3105 1534 : if (sendTimeLine != SendRqstTLI)
3106 0 : becameHistoric = true;
3107 : }
3108 :
3109 1536 : if (becameHistoric)
3110 : {
3111 : /*
3112 : * The timeline we were sending has become historic. Read the
3113 : * timeline history file of the new timeline to see where exactly
3114 : * we forked off from the timeline we were sending.
3115 : */
3116 : List *history;
3117 :
3118 2 : history = readTimeLineHistory(SendRqstTLI);
3119 2 : sendTimeLineValidUpto = tliSwitchPoint(sendTimeLine, history, &sendTimeLineNextTLI);
3120 :
3121 : Assert(sendTimeLine < sendTimeLineNextTLI);
3122 2 : list_free_deep(history);
3123 :
3124 2 : sendTimeLineIsHistoric = true;
3125 :
3126 2 : SendRqstPtr = sendTimeLineValidUpto;
3127 : }
3128 : }
3129 : else
3130 : {
3131 : /*
3132 : * Streaming the current timeline on a primary.
3133 : *
3134 : * Attempt to send all data that's already been written out and
3135 : * fsync'd to disk. We cannot go further than what's been written out
3136 : * given the current implementation of WALRead(). And in any case
3137 : * it's unsafe to send WAL that is not securely down to disk on the
3138 : * primary: if the primary subsequently crashes and restarts, standbys
3139 : * must not have applied any WAL that got lost on the primary.
3140 : */
3141 232996 : SendRqstPtr = GetFlushRecPtr(NULL);
3142 : }
3143 :
3144 : /*
3145 : * Record the current system time as an approximation of the time at which
3146 : * this WAL location was written for the purposes of lag tracking.
3147 : *
3148 : * In theory we could make XLogFlush() record a time in shmem whenever WAL
3149 : * is flushed and we could get that time as well as the LSN when we call
3150 : * GetFlushRecPtr() above (and likewise for the cascading standby
3151 : * equivalent), but rather than putting any new code into the hot WAL path
3152 : * it seems good enough to capture the time here. We should reach this
3153 : * after XLogFlush() runs WalSndWakeupProcessRequests(), and although that
3154 : * may take some time, we read the WAL flush pointer and take the time
3155 : * very close to together here so that we'll get a later position if it is
3156 : * still moving.
3157 : *
3158 : * Because LagTrackerWrite ignores samples when the LSN hasn't advanced,
3159 : * this gives us a cheap approximation for the WAL flush time for this
3160 : * LSN.
3161 : *
3162 : * Note that the LSN is not necessarily the LSN for the data contained in
3163 : * the present message; it's the end of the WAL, which might be further
3164 : * ahead. All the lag tracking machinery cares about is finding out when
3165 : * that arbitrary LSN is eventually reported as written, flushed and
3166 : * applied, so that it can measure the elapsed time.
3167 : */
3168 234864 : LagTrackerWrite(SendRqstPtr, GetCurrentTimestamp());
3169 :
3170 : /*
3171 : * If this is a historic timeline and we've reached the point where we
3172 : * forked to the next timeline, stop streaming.
3173 : *
3174 : * Note: We might already have sent WAL > sendTimeLineValidUpto. The
3175 : * startup process will normally replay all WAL that has been received
3176 : * from the primary, before promoting, but if the WAL streaming is
3177 : * terminated at a WAL page boundary, the valid portion of the timeline
3178 : * might end in the middle of a WAL record. We might've already sent the
3179 : * first half of that partial WAL record to the cascading standby, so that
3180 : * sentPtr > sendTimeLineValidUpto. That's OK; the cascading standby can't
3181 : * replay the partial WAL record either, so it can still follow our
3182 : * timeline switch.
3183 : */
3184 234864 : if (sendTimeLineIsHistoric && sendTimeLineValidUpto <= sentPtr)
3185 : {
3186 : /* close the current file. */
3187 26 : if (xlogreader->seg.ws_file >= 0)
3188 26 : wal_segment_close(xlogreader);
3189 :
3190 : /* Send CopyDone */
3191 26 : pq_putmessage_noblock('c', NULL, 0);
3192 26 : streamingDoneSending = true;
3193 :
3194 26 : WalSndCaughtUp = true;
3195 :
3196 26 : elog(DEBUG1, "walsender reached end of timeline at %X/%X (sent up to %X/%X)",
3197 : LSN_FORMAT_ARGS(sendTimeLineValidUpto),
3198 : LSN_FORMAT_ARGS(sentPtr));
3199 26 : return;
3200 : }
3201 :
3202 : /* Do we have any work to do? */
3203 : Assert(sentPtr <= SendRqstPtr);
3204 234838 : if (SendRqstPtr <= sentPtr)
3205 : {
3206 48752 : WalSndCaughtUp = true;
3207 48752 : return;
3208 : }
3209 :
3210 : /*
3211 : * Figure out how much to send in one message. If there's no more than
3212 : * MAX_SEND_SIZE bytes to send, send everything. Otherwise send
3213 : * MAX_SEND_SIZE bytes, but round back to logfile or page boundary.
3214 : *
3215 : * The rounding is not only for performance reasons. Walreceiver relies on
3216 : * the fact that we never split a WAL record across two messages. Since a
3217 : * long WAL record is split at page boundary into continuation records,
3218 : * page boundary is always a safe cut-off point. We also assume that
3219 : * SendRqstPtr never points to the middle of a WAL record.
3220 : */
3221 186086 : startptr = sentPtr;
3222 186086 : endptr = startptr;
3223 186086 : endptr += MAX_SEND_SIZE;
3224 :
3225 : /* if we went beyond SendRqstPtr, back off */
3226 186086 : if (SendRqstPtr <= endptr)
3227 : {
3228 9326 : endptr = SendRqstPtr;
3229 9326 : if (sendTimeLineIsHistoric)
3230 24 : WalSndCaughtUp = false;
3231 : else
3232 9302 : WalSndCaughtUp = true;
3233 : }
3234 : else
3235 : {
3236 : /* round down to page boundary. */
3237 176760 : endptr -= (endptr % XLOG_BLCKSZ);
3238 176760 : WalSndCaughtUp = false;
3239 : }
3240 :
3241 186086 : nbytes = endptr - startptr;
3242 : Assert(nbytes <= MAX_SEND_SIZE);
3243 :
3244 : /*
3245 : * OK to read and send the slice.
3246 : */
3247 186086 : resetStringInfo(&output_message);
3248 186086 : pq_sendbyte(&output_message, 'w');
3249 :
3250 186086 : pq_sendint64(&output_message, startptr); /* dataStart */
3251 186086 : pq_sendint64(&output_message, SendRqstPtr); /* walEnd */
3252 186086 : pq_sendint64(&output_message, 0); /* sendtime, filled in last */
3253 :
3254 : /*
3255 : * Read the log directly into the output buffer to avoid extra memcpy
3256 : * calls.
3257 : */
3258 186086 : enlargeStringInfo(&output_message, nbytes);
3259 :
3260 186086 : retry:
3261 : /* attempt to read WAL from WAL buffers first */
3262 186086 : rbytes = WALReadFromBuffers(&output_message.data[output_message.len],
3263 186086 : startptr, nbytes, xlogreader->seg.ws_tli);
3264 186086 : output_message.len += rbytes;
3265 186086 : startptr += rbytes;
3266 186086 : nbytes -= rbytes;
3267 :
3268 : /* now read the remaining WAL from WAL file */
3269 186086 : if (nbytes > 0 &&
3270 181732 : !WALRead(xlogreader,
3271 181732 : &output_message.data[output_message.len],
3272 : startptr,
3273 : nbytes,
3274 181732 : xlogreader->seg.ws_tli, /* Pass the current TLI because
3275 : * only WalSndSegmentOpen controls
3276 : * whether new TLI is needed. */
3277 : &errinfo))
3278 0 : WALReadRaiseError(&errinfo);
3279 :
3280 : /* See logical_read_xlog_page(). */
3281 186086 : XLByteToSeg(startptr, segno, xlogreader->segcxt.ws_segsize);
3282 186086 : CheckXLogRemoved(segno, xlogreader->seg.ws_tli);
3283 :
3284 : /*
3285 : * During recovery, the currently-open WAL file might be replaced with the
3286 : * file of the same name retrieved from archive. So we always need to
3287 : * check what we read was valid after reading into the buffer. If it's
3288 : * invalid, we try to open and read the file again.
3289 : */
3290 186086 : if (am_cascading_walsender)
3291 : {
3292 1106 : WalSnd *walsnd = MyWalSnd;
3293 : bool reload;
3294 :
3295 1106 : SpinLockAcquire(&walsnd->mutex);
3296 1106 : reload = walsnd->needreload;
3297 1106 : walsnd->needreload = false;
3298 1106 : SpinLockRelease(&walsnd->mutex);
3299 :
3300 1106 : if (reload && xlogreader->seg.ws_file >= 0)
3301 : {
3302 0 : wal_segment_close(xlogreader);
3303 :
3304 0 : goto retry;
3305 : }
3306 : }
3307 :
3308 186086 : output_message.len += nbytes;
3309 186086 : output_message.data[output_message.len] = '\0';
3310 :
3311 : /*
3312 : * Fill the send timestamp last, so that it is taken as late as possible.
3313 : */
3314 186086 : resetStringInfo(&tmpbuf);
3315 186086 : pq_sendint64(&tmpbuf, GetCurrentTimestamp());
3316 186086 : memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)],
3317 186086 : tmpbuf.data, sizeof(int64));
3318 :
3319 186086 : pq_putmessage_noblock('d', output_message.data, output_message.len);
3320 :
3321 186086 : sentPtr = endptr;
3322 :
3323 : /* Update shared memory status */
3324 : {
3325 186086 : WalSnd *walsnd = MyWalSnd;
3326 :
3327 186086 : SpinLockAcquire(&walsnd->mutex);
3328 186086 : walsnd->sentPtr = sentPtr;
3329 186086 : SpinLockRelease(&walsnd->mutex);
3330 : }
3331 :
3332 : /* Report progress of XLOG streaming in PS display */
3333 186086 : if (update_process_title)
3334 : {
3335 : char activitymsg[50];
3336 :
3337 186086 : snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
3338 186086 : LSN_FORMAT_ARGS(sentPtr));
3339 186086 : set_ps_display(activitymsg);
3340 : }
3341 : }
3342 :
3343 : /*
3344 : * Stream out logically decoded data.
3345 : */
3346 : static void
3347 1663666 : XLogSendLogical(void)
3348 : {
3349 : XLogRecord *record;
3350 : char *errm;
3351 :
3352 : /*
3353 : * We'll use the current flush point to determine whether we've caught up.
3354 : * This variable is static in order to cache it across calls. Caching is
3355 : * helpful because GetFlushRecPtr() needs to acquire a heavily-contended
3356 : * spinlock.
3357 : */
3358 : static XLogRecPtr flushPtr = InvalidXLogRecPtr;
3359 :
3360 : /*
3361 : * Don't know whether we've caught up yet. We'll set WalSndCaughtUp to
3362 : * true in WalSndWaitForWal, if we're actually waiting. We also set to
3363 : * true if XLogReadRecord() had to stop reading but WalSndWaitForWal
3364 : * didn't wait - i.e. when we're shutting down.
3365 : */
3366 1663666 : WalSndCaughtUp = false;
3367 :
3368 1663666 : record = XLogReadRecord(logical_decoding_ctx->reader, &errm);
3369 :
3370 : /* xlog record was invalid */
3371 1663338 : if (errm != NULL)
3372 0 : elog(ERROR, "could not find record while sending logically-decoded data: %s",
3373 : errm);
3374 :
3375 1663338 : if (record != NULL)
3376 : {
3377 : /*
3378 : * Note the lack of any call to LagTrackerWrite() which is handled by
3379 : * WalSndUpdateProgress which is called by output plugin through
3380 : * logical decoding write api.
3381 : */
3382 1662916 : LogicalDecodingProcessRecord(logical_decoding_ctx, logical_decoding_ctx->reader);
3383 :
3384 1662908 : sentPtr = logical_decoding_ctx->reader->EndRecPtr;
3385 : }
3386 :
3387 : /*
3388 : * If first time through in this session, initialize flushPtr. Otherwise,
3389 : * we only need to update flushPtr if EndRecPtr is past it.
3390 : */
3391 1663330 : if (flushPtr == InvalidXLogRecPtr ||
3392 1662636 : logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
3393 : {
3394 4816 : if (am_cascading_walsender)
3395 126 : flushPtr = GetStandbyFlushRecPtr(NULL);
3396 : else
3397 4690 : flushPtr = GetFlushRecPtr(NULL);
3398 : }
3399 :
3400 : /* If EndRecPtr is still past our flushPtr, it means we caught up. */
3401 1663330 : if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)
3402 3056 : WalSndCaughtUp = true;
3403 :
3404 : /*
3405 : * If we're caught up and have been requested to stop, have WalSndLoop()
3406 : * terminate the connection in an orderly manner, after writing out all
3407 : * the pending data.
3408 : */
3409 1663330 : if (WalSndCaughtUp && got_STOPPING)
3410 114 : got_SIGUSR2 = true;
3411 :
3412 : /* Update shared memory status */
3413 : {
3414 1663330 : WalSnd *walsnd = MyWalSnd;
3415 :
3416 1663330 : SpinLockAcquire(&walsnd->mutex);
3417 1663330 : walsnd->sentPtr = sentPtr;
3418 1663330 : SpinLockRelease(&walsnd->mutex);
3419 : }
3420 1663330 : }
3421 :
3422 : /*
3423 : * Shutdown if the sender is caught up.
3424 : *
3425 : * NB: This should only be called when the shutdown signal has been received
3426 : * from postmaster.
3427 : *
3428 : * Note that if we determine that there's still more data to send, this
3429 : * function will return control to the caller.
3430 : */
3431 : static void
3432 16462 : WalSndDone(WalSndSendDataCallback send_data)
3433 : {
3434 : XLogRecPtr replicatedPtr;
3435 :
3436 : /* ... let's just be real sure we're caught up ... */
3437 16462 : send_data();
3438 :
3439 : /*
3440 : * To figure out whether all WAL has successfully been replicated, check
3441 : * flush location if valid, write otherwise. Tools like pg_receivewal will
3442 : * usually (unless in synchronous mode) return an invalid flush location.
3443 : */
3444 32924 : replicatedPtr = XLogRecPtrIsInvalid(MyWalSnd->flush) ?
3445 16462 : MyWalSnd->write : MyWalSnd->flush;
3446 :
3447 16462 : if (WalSndCaughtUp && sentPtr == replicatedPtr &&
3448 68 : !pq_is_send_pending())
3449 : {
3450 : QueryCompletion qc;
3451 :
3452 : /* Inform the standby that XLOG streaming is done */
3453 68 : SetQueryCompletion(&qc, CMDTAG_COPY, 0);
3454 68 : EndCommand(&qc, DestRemote, false);
3455 68 : pq_flush();
3456 :
3457 68 : proc_exit(0);
3458 : }
3459 16394 : if (!waiting_for_ping_response)
3460 8188 : WalSndKeepalive(true, InvalidXLogRecPtr);
3461 16394 : }
3462 :
3463 : /*
3464 : * Returns the latest point in WAL that has been safely flushed to disk.
3465 : * This should only be called when in recovery.
3466 : *
3467 : * This is called either by cascading walsender to find WAL position to be sent
3468 : * to a cascaded standby or by slot synchronization operation to validate remote
3469 : * slot's lsn before syncing it locally.
3470 : *
3471 : * As a side-effect, *tli is updated to the TLI of the last
3472 : * replayed WAL record.
3473 : */
3474 : XLogRecPtr
3475 1856 : GetStandbyFlushRecPtr(TimeLineID *tli)
3476 : {
3477 : XLogRecPtr replayPtr;
3478 : TimeLineID replayTLI;
3479 : XLogRecPtr receivePtr;
3480 : TimeLineID receiveTLI;
3481 : XLogRecPtr result;
3482 :
3483 : Assert(am_cascading_walsender || IsSyncingReplicationSlots());
3484 :
3485 : /*
3486 : * We can safely send what's already been replayed. Also, if walreceiver
3487 : * is streaming WAL from the same timeline, we can send anything that it
3488 : * has streamed, but hasn't been replayed yet.
3489 : */
3490 :
3491 1856 : receivePtr = GetWalRcvFlushRecPtr(NULL, &receiveTLI);
3492 1856 : replayPtr = GetXLogReplayRecPtr(&replayTLI);
3493 :
3494 1856 : if (tli)
3495 1672 : *tli = replayTLI;
3496 :
3497 1856 : result = replayPtr;
3498 1856 : if (receiveTLI == replayTLI && receivePtr > replayPtr)
3499 110 : result = receivePtr;
3500 :
3501 1856 : return result;
3502 : }
3503 :
3504 : /*
3505 : * Request walsenders to reload the currently-open WAL file
3506 : */
3507 : void
3508 48 : WalSndRqstFileReload(void)
3509 : {
3510 : int i;
3511 :
3512 504 : for (i = 0; i < max_wal_senders; i++)
3513 : {
3514 456 : WalSnd *walsnd = &WalSndCtl->walsnds[i];
3515 :
3516 456 : SpinLockAcquire(&walsnd->mutex);
3517 456 : if (walsnd->pid == 0)
3518 : {
3519 456 : SpinLockRelease(&walsnd->mutex);
3520 456 : continue;
3521 : }
3522 0 : walsnd->needreload = true;
3523 0 : SpinLockRelease(&walsnd->mutex);
3524 : }
3525 48 : }
3526 :
3527 : /*
3528 : * Handle PROCSIG_WALSND_INIT_STOPPING signal.
3529 : */
3530 : void
3531 68 : HandleWalSndInitStopping(void)
3532 : {
3533 : Assert(am_walsender);
3534 :
3535 : /*
3536 : * If replication has not yet started, die like with SIGTERM. If
3537 : * replication is active, only set a flag and wake up the main loop. It
3538 : * will send any outstanding WAL, wait for it to be replicated to the
3539 : * standby, and then exit gracefully.
3540 : */
3541 68 : if (!replication_active)
3542 0 : kill(MyProcPid, SIGTERM);
3543 : else
3544 68 : got_STOPPING = true;
3545 68 : }
3546 :
3547 : /*
3548 : * SIGUSR2: set flag to do a last cycle and shut down afterwards. The WAL
3549 : * sender should already have been switched to WALSNDSTATE_STOPPING at
3550 : * this point.
3551 : */
3552 : static void
3553 52 : WalSndLastCycleHandler(SIGNAL_ARGS)
3554 : {
3555 52 : got_SIGUSR2 = true;
3556 52 : SetLatch(MyLatch);
3557 52 : }
3558 :
3559 : /* Set up signal handlers */
3560 : void
3561 2122 : WalSndSignals(void)
3562 : {
3563 : /* Set up signal handlers */
3564 2122 : pqsignal(SIGHUP, SignalHandlerForConfigReload);
3565 2122 : pqsignal(SIGINT, StatementCancelHandler); /* query cancel */
3566 2122 : pqsignal(SIGTERM, die); /* request shutdown */
3567 : /* SIGQUIT handler was already set up by InitPostmasterChild */
3568 2122 : InitializeTimeouts(); /* establishes SIGALRM handler */
3569 2122 : pqsignal(SIGPIPE, SIG_IGN);
3570 2122 : pqsignal(SIGUSR1, procsignal_sigusr1_handler);
3571 2122 : pqsignal(SIGUSR2, WalSndLastCycleHandler); /* request a last cycle and
3572 : * shutdown */
3573 :
3574 : /* Reset some signals that are accepted by postmaster but not here */
3575 2122 : pqsignal(SIGCHLD, SIG_DFL);
3576 2122 : }
3577 :
3578 : /* Report shared-memory space needed by WalSndShmemInit */
3579 : Size
3580 7450 : WalSndShmemSize(void)
3581 : {
3582 7450 : Size size = 0;
3583 :
3584 7450 : size = offsetof(WalSndCtlData, walsnds);
3585 7450 : size = add_size(size, mul_size(max_wal_senders, sizeof(WalSnd)));
3586 :
3587 7450 : return size;
3588 : }
3589 :
3590 : /* Allocate and initialize walsender-related shared memory */
3591 : void
3592 1930 : WalSndShmemInit(void)
3593 : {
3594 : bool found;
3595 : int i;
3596 :
3597 1930 : WalSndCtl = (WalSndCtlData *)
3598 1930 : ShmemInitStruct("Wal Sender Ctl", WalSndShmemSize(), &found);
3599 :
3600 1930 : if (!found)
3601 : {
3602 : /* First time through, so initialize */
3603 13626 : MemSet(WalSndCtl, 0, WalSndShmemSize());
3604 :
3605 7720 : for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
3606 5790 : dlist_init(&(WalSndCtl->SyncRepQueue[i]));
3607 :
3608 14790 : for (i = 0; i < max_wal_senders; i++)
3609 : {
3610 12860 : WalSnd *walsnd = &WalSndCtl->walsnds[i];
3611 :
3612 12860 : SpinLockInit(&walsnd->mutex);
3613 : }
3614 :
3615 1930 : ConditionVariableInit(&WalSndCtl->wal_flush_cv);
3616 1930 : ConditionVariableInit(&WalSndCtl->wal_replay_cv);
3617 1930 : ConditionVariableInit(&WalSndCtl->wal_confirm_rcv_cv);
3618 : }
3619 1930 : }
3620 :
3621 : /*
3622 : * Wake up physical, logical or both kinds of walsenders
3623 : *
3624 : * The distinction between physical and logical walsenders is done, because:
3625 : * - physical walsenders can't send data until it's been flushed
3626 : * - logical walsenders on standby can't decode and send data until it's been
3627 : * applied
3628 : *
3629 : * For cascading replication we need to wake up physical walsenders separately
3630 : * from logical walsenders (see the comment before calling WalSndWakeup() in
3631 : * ApplyWalRecord() for more details).
3632 : *
3633 : * This will be called inside critical sections, so throwing an error is not
3634 : * advisable.
3635 : */
3636 : void
3637 5188368 : WalSndWakeup(bool physical, bool logical)
3638 : {
3639 : /*
3640 : * Wake up all the walsenders waiting on WAL being flushed or replayed
3641 : * respectively. Note that waiting walsender would have prepared to sleep
3642 : * on the CV (i.e., added itself to the CV's waitlist) in WalSndWait()
3643 : * before actually waiting.
3644 : */
3645 5188368 : if (physical)
3646 208518 : ConditionVariableBroadcast(&WalSndCtl->wal_flush_cv);
3647 :
3648 5188368 : if (logical)
3649 5170366 : ConditionVariableBroadcast(&WalSndCtl->wal_replay_cv);
3650 5188368 : }
3651 :
3652 : /*
3653 : * Wait for readiness on the FeBe socket, or a timeout. The mask should be
3654 : * composed of optional WL_SOCKET_WRITEABLE and WL_SOCKET_READABLE flags. Exit
3655 : * on postmaster death.
3656 : */
3657 : static void
3658 110272 : WalSndWait(uint32 socket_events, long timeout, uint32 wait_event)
3659 : {
3660 : WaitEvent event;
3661 :
3662 110272 : ModifyWaitEvent(FeBeWaitSet, FeBeWaitSetSocketPos, socket_events, NULL);
3663 :
3664 : /*
3665 : * We use a condition variable to efficiently wake up walsenders in
3666 : * WalSndWakeup().
3667 : *
3668 : * Every walsender prepares to sleep on a shared memory CV. Note that it
3669 : * just prepares to sleep on the CV (i.e., adds itself to the CV's
3670 : * waitlist), but does not actually wait on the CV (IOW, it never calls
3671 : * ConditionVariableSleep()). It still uses WaitEventSetWait() for
3672 : * waiting, because we also need to wait for socket events. The processes
3673 : * (startup process, walreceiver etc.) wanting to wake up walsenders use
3674 : * ConditionVariableBroadcast(), which in turn calls SetLatch(), helping
3675 : * walsenders come out of WaitEventSetWait().
3676 : *
3677 : * This approach is simple and efficient because, one doesn't have to loop
3678 : * through all the walsenders slots, with a spinlock acquisition and
3679 : * release for every iteration, just to wake up only the waiting
3680 : * walsenders. It makes WalSndWakeup() callers' life easy.
3681 : *
3682 : * XXX: A desirable future improvement would be to add support for CVs
3683 : * into WaitEventSetWait().
3684 : *
3685 : * And, we use separate shared memory CVs for physical and logical
3686 : * walsenders for selective wake ups, see WalSndWakeup() for more details.
3687 : *
3688 : * If the wait event is WAIT_FOR_STANDBY_CONFIRMATION, wait on another CV
3689 : * until awakened by physical walsenders after the walreceiver confirms
3690 : * the receipt of the LSN.
3691 : */
3692 110272 : if (wait_event == WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION)
3693 8 : ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
3694 110264 : else if (MyWalSnd->kind == REPLICATION_KIND_PHYSICAL)
3695 105030 : ConditionVariablePrepareToSleep(&WalSndCtl->wal_flush_cv);
3696 5234 : else if (MyWalSnd->kind == REPLICATION_KIND_LOGICAL)
3697 5234 : ConditionVariablePrepareToSleep(&WalSndCtl->wal_replay_cv);
3698 :
3699 110272 : if (WaitEventSetWait(FeBeWaitSet, timeout, &event, 1, wait_event) == 1 &&
3700 110272 : (event.events & WL_POSTMASTER_DEATH))
3701 : {
3702 0 : ConditionVariableCancelSleep();
3703 0 : proc_exit(1);
3704 : }
3705 :
3706 110272 : ConditionVariableCancelSleep();
3707 110272 : }
3708 :
3709 : /*
3710 : * Signal all walsenders to move to stopping state.
3711 : *
3712 : * This will trigger walsenders to move to a state where no further WAL can be
3713 : * generated. See this file's header for details.
3714 : */
3715 : void
3716 1094 : WalSndInitStopping(void)
3717 : {
3718 : int i;
3719 :
3720 8630 : for (i = 0; i < max_wal_senders; i++)
3721 : {
3722 7536 : WalSnd *walsnd = &WalSndCtl->walsnds[i];
3723 : pid_t pid;
3724 :
3725 7536 : SpinLockAcquire(&walsnd->mutex);
3726 7536 : pid = walsnd->pid;
3727 7536 : SpinLockRelease(&walsnd->mutex);
3728 :
3729 7536 : if (pid == 0)
3730 7468 : continue;
3731 :
3732 68 : SendProcSignal(pid, PROCSIG_WALSND_INIT_STOPPING, INVALID_PROC_NUMBER);
3733 : }
3734 1094 : }
3735 :
3736 : /*
3737 : * Wait that all the WAL senders have quit or reached the stopping state. This
3738 : * is used by the checkpointer to control when the shutdown checkpoint can
3739 : * safely be performed.
3740 : */
3741 : void
3742 1158 : WalSndWaitStopping(void)
3743 : {
3744 : for (;;)
3745 64 : {
3746 : int i;
3747 1158 : bool all_stopped = true;
3748 :
3749 8696 : for (i = 0; i < max_wal_senders; i++)
3750 : {
3751 7602 : WalSnd *walsnd = &WalSndCtl->walsnds[i];
3752 :
3753 7602 : SpinLockAcquire(&walsnd->mutex);
3754 :
3755 7602 : if (walsnd->pid == 0)
3756 : {
3757 7486 : SpinLockRelease(&walsnd->mutex);
3758 7486 : continue;
3759 : }
3760 :
3761 116 : if (walsnd->state != WALSNDSTATE_STOPPING)
3762 : {
3763 64 : all_stopped = false;
3764 64 : SpinLockRelease(&walsnd->mutex);
3765 64 : break;
3766 : }
3767 52 : SpinLockRelease(&walsnd->mutex);
3768 : }
3769 :
3770 : /* safe to leave if confirmation is done for all WAL senders */
3771 1158 : if (all_stopped)
3772 1094 : return;
3773 :
3774 64 : pg_usleep(10000L); /* wait for 10 msec */
3775 : }
3776 : }
3777 :
3778 : /* Set state for current walsender (only called in walsender) */
3779 : void
3780 36924 : WalSndSetState(WalSndState state)
3781 : {
3782 36924 : WalSnd *walsnd = MyWalSnd;
3783 :
3784 : Assert(am_walsender);
3785 :
3786 36924 : if (walsnd->state == state)
3787 33456 : return;
3788 :
3789 3468 : SpinLockAcquire(&walsnd->mutex);
3790 3468 : walsnd->state = state;
3791 3468 : SpinLockRelease(&walsnd->mutex);
3792 : }
3793 :
3794 : /*
3795 : * Return a string constant representing the state. This is used
3796 : * in system views, and should *not* be translated.
3797 : */
3798 : static const char *
3799 1504 : WalSndGetStateString(WalSndState state)
3800 : {
3801 1504 : switch (state)
3802 : {
3803 10 : case WALSNDSTATE_STARTUP:
3804 10 : return "startup";
3805 0 : case WALSNDSTATE_BACKUP:
3806 0 : return "backup";
3807 16 : case WALSNDSTATE_CATCHUP:
3808 16 : return "catchup";
3809 1478 : case WALSNDSTATE_STREAMING:
3810 1478 : return "streaming";
3811 0 : case WALSNDSTATE_STOPPING:
3812 0 : return "stopping";
3813 : }
3814 0 : return "UNKNOWN";
3815 : }
3816 :
3817 : static Interval *
3818 2286 : offset_to_interval(TimeOffset offset)
3819 : {
3820 2286 : Interval *result = palloc(sizeof(Interval));
3821 :
3822 2286 : result->month = 0;
3823 2286 : result->day = 0;
3824 2286 : result->time = offset;
3825 :
3826 2286 : return result;
3827 : }
3828 :
3829 : /*
3830 : * Returns activity of walsenders, including pids and xlog locations sent to
3831 : * standby servers.
3832 : */
3833 : Datum
3834 1274 : pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
3835 : {
3836 : #define PG_STAT_GET_WAL_SENDERS_COLS 12
3837 1274 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
3838 : SyncRepStandbyData *sync_standbys;
3839 : int num_standbys;
3840 : int i;
3841 :
3842 1274 : InitMaterializedSRF(fcinfo, 0);
3843 :
3844 : /*
3845 : * Get the currently active synchronous standbys. This could be out of
3846 : * date before we're done, but we'll use the data anyway.
3847 : */
3848 1274 : num_standbys = SyncRepGetCandidateStandbys(&sync_standbys);
3849 :
3850 13626 : for (i = 0; i < max_wal_senders; i++)
3851 : {
3852 12352 : WalSnd *walsnd = &WalSndCtl->walsnds[i];
3853 : XLogRecPtr sent_ptr;
3854 : XLogRecPtr write;
3855 : XLogRecPtr flush;
3856 : XLogRecPtr apply;
3857 : TimeOffset writeLag;
3858 : TimeOffset flushLag;
3859 : TimeOffset applyLag;
3860 : int priority;
3861 : int pid;
3862 : WalSndState state;
3863 : TimestampTz replyTime;
3864 : bool is_sync_standby;
3865 : Datum values[PG_STAT_GET_WAL_SENDERS_COLS];
3866 12352 : bool nulls[PG_STAT_GET_WAL_SENDERS_COLS] = {0};
3867 : int j;
3868 :
3869 : /* Collect data from shared memory */
3870 12352 : SpinLockAcquire(&walsnd->mutex);
3871 12352 : if (walsnd->pid == 0)
3872 : {
3873 10848 : SpinLockRelease(&walsnd->mutex);
3874 10848 : continue;
3875 : }
3876 1504 : pid = walsnd->pid;
3877 1504 : sent_ptr = walsnd->sentPtr;
3878 1504 : state = walsnd->state;
3879 1504 : write = walsnd->write;
3880 1504 : flush = walsnd->flush;
3881 1504 : apply = walsnd->apply;
3882 1504 : writeLag = walsnd->writeLag;
3883 1504 : flushLag = walsnd->flushLag;
3884 1504 : applyLag = walsnd->applyLag;
3885 1504 : priority = walsnd->sync_standby_priority;
3886 1504 : replyTime = walsnd->replyTime;
3887 1504 : SpinLockRelease(&walsnd->mutex);
3888 :
3889 : /*
3890 : * Detect whether walsender is/was considered synchronous. We can
3891 : * provide some protection against stale data by checking the PID
3892 : * along with walsnd_index.
3893 : */
3894 1504 : is_sync_standby = false;
3895 1586 : for (j = 0; j < num_standbys; j++)
3896 : {
3897 136 : if (sync_standbys[j].walsnd_index == i &&
3898 54 : sync_standbys[j].pid == pid)
3899 : {
3900 54 : is_sync_standby = true;
3901 54 : break;
3902 : }
3903 : }
3904 :
3905 1504 : values[0] = Int32GetDatum(pid);
3906 :
3907 1504 : if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
3908 : {
3909 : /*
3910 : * Only superusers and roles with privileges of pg_read_all_stats
3911 : * can see details. Other users only get the pid value to know
3912 : * it's a walsender, but no details.
3913 : */
3914 0 : MemSet(&nulls[1], true, PG_STAT_GET_WAL_SENDERS_COLS - 1);
3915 : }
3916 : else
3917 : {
3918 1504 : values[1] = CStringGetTextDatum(WalSndGetStateString(state));
3919 :
3920 1504 : if (XLogRecPtrIsInvalid(sent_ptr))
3921 10 : nulls[2] = true;
3922 1504 : values[2] = LSNGetDatum(sent_ptr);
3923 :
3924 1504 : if (XLogRecPtrIsInvalid(write))
3925 16 : nulls[3] = true;
3926 1504 : values[3] = LSNGetDatum(write);
3927 :
3928 1504 : if (XLogRecPtrIsInvalid(flush))
3929 16 : nulls[4] = true;
3930 1504 : values[4] = LSNGetDatum(flush);
3931 :
3932 1504 : if (XLogRecPtrIsInvalid(apply))
3933 16 : nulls[5] = true;
3934 1504 : values[5] = LSNGetDatum(apply);
3935 :
3936 : /*
3937 : * Treat a standby such as a pg_basebackup background process
3938 : * which always returns an invalid flush location, as an
3939 : * asynchronous standby.
3940 : */
3941 1504 : priority = XLogRecPtrIsInvalid(flush) ? 0 : priority;
3942 :
3943 1504 : if (writeLag < 0)
3944 782 : nulls[6] = true;
3945 : else
3946 722 : values[6] = IntervalPGetDatum(offset_to_interval(writeLag));
3947 :
3948 1504 : if (flushLag < 0)
3949 662 : nulls[7] = true;
3950 : else
3951 842 : values[7] = IntervalPGetDatum(offset_to_interval(flushLag));
3952 :
3953 1504 : if (applyLag < 0)
3954 782 : nulls[8] = true;
3955 : else
3956 722 : values[8] = IntervalPGetDatum(offset_to_interval(applyLag));
3957 :
3958 1504 : values[9] = Int32GetDatum(priority);
3959 :
3960 : /*
3961 : * More easily understood version of standby state. This is purely
3962 : * informational.
3963 : *
3964 : * In quorum-based sync replication, the role of each standby
3965 : * listed in synchronous_standby_names can be changing very
3966 : * frequently. Any standbys considered as "sync" at one moment can
3967 : * be switched to "potential" ones at the next moment. So, it's
3968 : * basically useless to report "sync" or "potential" as their sync
3969 : * states. We report just "quorum" for them.
3970 : */
3971 1504 : if (priority == 0)
3972 1428 : values[10] = CStringGetTextDatum("async");
3973 76 : else if (is_sync_standby)
3974 54 : values[10] = SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY ?
3975 54 : CStringGetTextDatum("sync") : CStringGetTextDatum("quorum");
3976 : else
3977 22 : values[10] = CStringGetTextDatum("potential");
3978 :
3979 1504 : if (replyTime == 0)
3980 10 : nulls[11] = true;
3981 : else
3982 1494 : values[11] = TimestampTzGetDatum(replyTime);
3983 : }
3984 :
3985 1504 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
3986 : values, nulls);
3987 : }
3988 :
3989 1274 : return (Datum) 0;
3990 : }
3991 :
3992 : /*
3993 : * Send a keepalive message to standby.
3994 : *
3995 : * If requestReply is set, the message requests the other party to send
3996 : * a message back to us, for heartbeat purposes. We also set a flag to
3997 : * let nearby code know that we're waiting for that response, to avoid
3998 : * repeated requests.
3999 : *
4000 : * writePtr is the location up to which the WAL is sent. It is essentially
4001 : * the same as sentPtr but in some cases, we need to send keep alive before
4002 : * sentPtr is updated like when skipping empty transactions.
4003 : */
4004 : static void
4005 11708 : WalSndKeepalive(bool requestReply, XLogRecPtr writePtr)
4006 : {
4007 11708 : elog(DEBUG2, "sending replication keepalive");
4008 :
4009 : /* construct the message... */
4010 11708 : resetStringInfo(&output_message);
4011 11708 : pq_sendbyte(&output_message, 'k');
4012 11708 : pq_sendint64(&output_message, XLogRecPtrIsInvalid(writePtr) ? sentPtr : writePtr);
4013 11708 : pq_sendint64(&output_message, GetCurrentTimestamp());
4014 11708 : pq_sendbyte(&output_message, requestReply ? 1 : 0);
4015 :
4016 : /* ... and send it wrapped in CopyData */
4017 11708 : pq_putmessage_noblock('d', output_message.data, output_message.len);
4018 :
4019 : /* Set local flag */
4020 11708 : if (requestReply)
4021 8188 : waiting_for_ping_response = true;
4022 11708 : }
4023 :
4024 : /*
4025 : * Send keepalive message if too much time has elapsed.
4026 : */
4027 : static void
4028 2419248 : WalSndKeepaliveIfNecessary(void)
4029 : {
4030 : TimestampTz ping_time;
4031 :
4032 : /*
4033 : * Don't send keepalive messages if timeouts are globally disabled or
4034 : * we're doing something not partaking in timeouts.
4035 : */
4036 2419248 : if (wal_sender_timeout <= 0 || last_reply_timestamp <= 0)
4037 48 : return;
4038 :
4039 2419200 : if (waiting_for_ping_response)
4040 24580 : return;
4041 :
4042 : /*
4043 : * If half of wal_sender_timeout has lapsed without receiving any reply
4044 : * from the standby, send a keep-alive message to the standby requesting
4045 : * an immediate reply.
4046 : */
4047 2394620 : ping_time = TimestampTzPlusMilliseconds(last_reply_timestamp,
4048 : wal_sender_timeout / 2);
4049 2394620 : if (last_processing >= ping_time)
4050 : {
4051 0 : WalSndKeepalive(true, InvalidXLogRecPtr);
4052 :
4053 : /* Try to flush pending output to the client */
4054 0 : if (pq_flush_if_writable() != 0)
4055 0 : WalSndShutdown();
4056 : }
4057 : }
4058 :
4059 : /*
4060 : * Record the end of the WAL and the time it was flushed locally, so that
4061 : * LagTrackerRead can compute the elapsed time (lag) when this WAL location is
4062 : * eventually reported to have been written, flushed and applied by the
4063 : * standby in a reply message.
4064 : */
4065 : static void
4066 235216 : LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time)
4067 : {
4068 : bool buffer_full;
4069 : int new_write_head;
4070 : int i;
4071 :
4072 235216 : if (!am_walsender)
4073 0 : return;
4074 :
4075 : /*
4076 : * If the lsn hasn't advanced since last time, then do nothing. This way
4077 : * we only record a new sample when new WAL has been written.
4078 : */
4079 235216 : if (lag_tracker->last_lsn == lsn)
4080 221490 : return;
4081 13726 : lag_tracker->last_lsn = lsn;
4082 :
4083 : /*
4084 : * If advancing the write head of the circular buffer would crash into any
4085 : * of the read heads, then the buffer is full. In other words, the
4086 : * slowest reader (presumably apply) is the one that controls the release
4087 : * of space.
4088 : */
4089 13726 : new_write_head = (lag_tracker->write_head + 1) % LAG_TRACKER_BUFFER_SIZE;
4090 13726 : buffer_full = false;
4091 54904 : for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; ++i)
4092 : {
4093 41178 : if (new_write_head == lag_tracker->read_heads[i])
4094 0 : buffer_full = true;
4095 : }
4096 :
4097 : /*
4098 : * If the buffer is full, for now we just rewind by one slot and overwrite
4099 : * the last sample, as a simple (if somewhat uneven) way to lower the
4100 : * sampling rate. There may be better adaptive compaction algorithms.
4101 : */
4102 13726 : if (buffer_full)
4103 : {
4104 0 : new_write_head = lag_tracker->write_head;
4105 0 : if (lag_tracker->write_head > 0)
4106 0 : lag_tracker->write_head--;
4107 : else
4108 0 : lag_tracker->write_head = LAG_TRACKER_BUFFER_SIZE - 1;
4109 : }
4110 :
4111 : /* Store a sample at the current write head position. */
4112 13726 : lag_tracker->buffer[lag_tracker->write_head].lsn = lsn;
4113 13726 : lag_tracker->buffer[lag_tracker->write_head].time = local_flush_time;
4114 13726 : lag_tracker->write_head = new_write_head;
4115 : }
4116 :
4117 : /*
4118 : * Find out how much time has elapsed between the moment WAL location 'lsn'
4119 : * (or the highest known earlier LSN) was flushed locally and the time 'now'.
4120 : * We have a separate read head for each of the reported LSN locations we
4121 : * receive in replies from standby; 'head' controls which read head is
4122 : * used. Whenever a read head crosses an LSN which was written into the
4123 : * lag buffer with LagTrackerWrite, we can use the associated timestamp to
4124 : * find out the time this LSN (or an earlier one) was flushed locally, and
4125 : * therefore compute the lag.
4126 : *
4127 : * Return -1 if no new sample data is available, and otherwise the elapsed
4128 : * time in microseconds.
4129 : */
4130 : static TimeOffset
4131 274038 : LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now)
4132 : {
4133 274038 : TimestampTz time = 0;
4134 :
4135 : /* Read all unread samples up to this LSN or end of buffer. */
4136 313190 : while (lag_tracker->read_heads[head] != lag_tracker->write_head &&
4137 129838 : lag_tracker->buffer[lag_tracker->read_heads[head]].lsn <= lsn)
4138 : {
4139 39152 : time = lag_tracker->buffer[lag_tracker->read_heads[head]].time;
4140 39152 : lag_tracker->last_read[head] =
4141 39152 : lag_tracker->buffer[lag_tracker->read_heads[head]];
4142 39152 : lag_tracker->read_heads[head] =
4143 39152 : (lag_tracker->read_heads[head] + 1) % LAG_TRACKER_BUFFER_SIZE;
4144 : }
4145 :
4146 : /*
4147 : * If the lag tracker is empty, that means the standby has processed
4148 : * everything we've ever sent so we should now clear 'last_read'. If we
4149 : * didn't do that, we'd risk using a stale and irrelevant sample for
4150 : * interpolation at the beginning of the next burst of WAL after a period
4151 : * of idleness.
4152 : */
4153 274038 : if (lag_tracker->read_heads[head] == lag_tracker->write_head)
4154 183352 : lag_tracker->last_read[head].time = 0;
4155 :
4156 274038 : if (time > now)
4157 : {
4158 : /* If the clock somehow went backwards, treat as not found. */
4159 0 : return -1;
4160 : }
4161 274038 : else if (time == 0)
4162 : {
4163 : /*
4164 : * We didn't cross a time. If there is a future sample that we
4165 : * haven't reached yet, and we've already reached at least one sample,
4166 : * let's interpolate the local flushed time. This is mainly useful
4167 : * for reporting a completely stuck apply position as having
4168 : * increasing lag, since otherwise we'd have to wait for it to
4169 : * eventually start moving again and cross one of our samples before
4170 : * we can show the lag increasing.
4171 : */
4172 247774 : if (lag_tracker->read_heads[head] == lag_tracker->write_head)
4173 : {
4174 : /* There are no future samples, so we can't interpolate. */
4175 161662 : return -1;
4176 : }
4177 86112 : else if (lag_tracker->last_read[head].time != 0)
4178 : {
4179 : /* We can interpolate between last_read and the next sample. */
4180 : double fraction;
4181 49630 : WalTimeSample prev = lag_tracker->last_read[head];
4182 49630 : WalTimeSample next = lag_tracker->buffer[lag_tracker->read_heads[head]];
4183 :
4184 49630 : if (lsn < prev.lsn)
4185 : {
4186 : /*
4187 : * Reported LSNs shouldn't normally go backwards, but it's
4188 : * possible when there is a timeline change. Treat as not
4189 : * found.
4190 : */
4191 0 : return -1;
4192 : }
4193 :
4194 : Assert(prev.lsn < next.lsn);
4195 :
4196 49630 : if (prev.time > next.time)
4197 : {
4198 : /* If the clock somehow went backwards, treat as not found. */
4199 0 : return -1;
4200 : }
4201 :
4202 : /* See how far we are between the previous and next samples. */
4203 49630 : fraction =
4204 49630 : (double) (lsn - prev.lsn) / (double) (next.lsn - prev.lsn);
4205 :
4206 : /* Scale the local flush time proportionally. */
4207 49630 : time = (TimestampTz)
4208 49630 : ((double) prev.time + (next.time - prev.time) * fraction);
4209 : }
4210 : else
4211 : {
4212 : /*
4213 : * We have only a future sample, implying that we were entirely
4214 : * caught up but and now there is a new burst of WAL and the
4215 : * standby hasn't processed the first sample yet. Until the
4216 : * standby reaches the future sample the best we can do is report
4217 : * the hypothetical lag if that sample were to be replayed now.
4218 : */
4219 36482 : time = lag_tracker->buffer[lag_tracker->read_heads[head]].time;
4220 : }
4221 : }
4222 :
4223 : /* Return the elapsed time since local flush time in microseconds. */
4224 : Assert(time != 0);
4225 112376 : return now - time;
4226 : }
|
