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