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

Generated by: LCOV version 1.13