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