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