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