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