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