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