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