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