LCOV - code coverage report
Current view: top level - src/backend/replication - walsender.c (source / functions) Coverage Total Hit
Test: PostgreSQL 20devel Lines: 91.6 % 1448 1326
Test Date: 2026-07-26 08:15:28 Functions: 100.0 % 63 63
Legend: Lines:     hit not hit
Branches: + taken - not taken # not executed
Branches: 69.5 % 852 592

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

Generated by: LCOV version 2.0-1