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

Generated by: LCOV version 1.14