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