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