LCOV - code coverage report
Current view: top level - src/backend/replication - walsender.c (source / functions) Hit Total Coverage
Test: PostgreSQL 19devel Lines: 1274 1395 91.3 %
Date: 2025-10-31 04:18:36 Functions: 60 60 100.0 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.16