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