Quellcode-Bibliothek walsender.c
Sprache: C
/*------------------------------------------------------------------------- * * walsender . c * * The WAL sender process ( walsender ) is new as of Postgres 9 . 0 . It takes * care of sending XLOG from the primary server to a single recipient . * ( Note that there can be more than one walsender process concurrently . ) * It is started by the postmaster when the walreceiver of a standby server * connects to the primary server and requests XLOG streaming replication . * * A walsender is similar to a regular backend , ie . there is a one - to - one * relationship between a connection and a walsender process , but instead * of processing SQL queries , it understands a small set of special * replication - mode commands . The START_REPLICATION command begins streaming * WAL to the client . While streaming , the walsender keeps reading XLOG * records from the disk and sends them to the standby server over the * COPY protocol , until either side ends the replication by exiting COPY * mode ( or until the connection is closed ) . * * Normal termination is by SIGTERM , which instructs the walsender to * close the connection and exit ( 0 ) at the next convenient moment . Emergency * termination is by SIGQUIT ; like any backend , the walsender will simply * abort and exit on SIGQUIT . A close of the connection and a FATAL error * are treated as not a crash but approximately normal termination ; * the walsender will exit quickly without sending any more XLOG records . * * If the server is shut down , checkpointer sends us * PROCSIG_WALSND_INIT_STOPPING after all regular backends have exited . If * the backend is idle or runs an SQL query this causes the backend to * shutdown , if logical replication is in progress all existing WAL records * are processed followed by a shutdown . Otherwise this causes the walsender * to switch to the " stopping " state . In this state , the walsender will reject * any further replication commands . The checkpointer begins the shutdown * checkpoint once all walsenders are confirmed as stopping . When the shutdown * checkpoint finishes , the postmaster sends us SIGUSR2 . This instructs * walsender to send any outstanding WAL , including the shutdown checkpoint * record , wait for it to be replicated to the standby , and then exit . * * * Portions Copyright ( c ) 2010 - 2025 , PostgreSQL Global Development Group * * IDENTIFICATION * src / backend / replication / walsender . c * * - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - #include "postgres.h" #include <signal.h>#include <unistd.h>#include "access/timeline.h" #include "access/transam.h" #include "access/xact.h" #include "access/xlog_internal.h" #include "access/xlogreader.h" #include "access/xlogrecovery.h" #include "access/xlogutils.h" #include "backup/basebackup.h" #include "backup/basebackup_incremental.h" #include "catalog/pg_authid.h" #include "catalog/pg_type.h" #include "commands/dbcommands.h" #include "commands/defrem.h" #include "funcapi.h" #include "libpq/libpq.h" #include "libpq/pqformat.h" #include "miscadmin.h" #include "nodes/replnodes.h" #include "pgstat.h" #include "postmaster/interrupt.h" #include "replication/decode.h" #include "replication/logical.h" #include "replication/slotsync.h" #include "replication/slot.h" #include "replication/snapbuild.h" #include "replication/syncrep.h" #include "replication/walreceiver.h" #include "replication/walsender.h" #include "replication/walsender_private.h" #include "storage/condition_variable.h" #include "storage/aio_subsys.h" #include "storage/fd.h" #include "storage/ipc.h" #include "storage/pmsignal.h" #include "storage/proc.h" #include "tcop/dest.h" #include "tcop/tcopprot.h" #include "utils/acl.h" #include "utils/builtins.h" #include "utils/guc.h" #include "utils/memutils.h" #include "utils/pg_lsn.h" #include "utils/pgstat_internal.h" #include "utils/ps_status.h" #include "utils/timeout.h" #include "utils/timestamp.h" /* Minimum interval used by walsender for stats flushes, in ms */ #define WALSENDER_STATS_FLUSH_INTERVAL 1000 /* * Maximum data payload in a WAL data message . Must be > = XLOG_BLCKSZ . * * We don ' t have a good idea of what a good value would be ; there ' s some * overhead per message in both walsender and walreceiver , but on the other * hand sending large batches makes walsender less responsive to signals * because signals are checked only between messages . 128 kB ( with * default 8 k blocks ) seems like a reasonable guess for now . #define MAX_SEND_SIZE (XLOG_BLCKSZ * 16 )/* Array of WalSnds in shared memory */ /* My slot in the shared memory array */ /* Global state */ bool am_walsender = false ; /* Am I a walsender process? */ bool am_cascading_walsender = false ; /* Am I cascading WAL to another bool am_db_walsender = false ; /* Connected to a database? */ /* GUC variables */ int max_wal_senders = 10 ; /* the maximum number of concurrent int wal_sender_timeout = 60 * 1000 ; /* maximum time to send one WAL bool log_replication_commands = false ;/* * State for WalSndWakeupRequest bool wake_wal_senders = false ;/* * xlogreader used for replication . Note that a WAL sender doing physical * replication does not need xlogreader to read WAL , but it needs one to * keep a state of its work . static XLogReaderState *xlogreader = NULL;/* * If the UPLOAD_MANIFEST command is used to provide a backup manifest in * preparation for an incremental backup , uploaded_manifest will be point * to an object containing information about its contexts , and * uploaded_manifest_mcxt will point to the memory context that contains * that object and all of its subordinate data . Otherwise , both values will * be NULL . static IncrementalBackupInfo *uploaded_manifest = NULL;static MemoryContext uploaded_manifest_mcxt = NULL;/* * These variables keep track of the state of the timeline we ' re currently * sending . sendTimeLine identifies the timeline . If sendTimeLineIsHistoric , * the timeline is not the latest timeline on this server , and the server ' s * history forked off from that timeline at sendTimeLineValidUpto . static TimeLineID sendTimeLine = 0 ;static TimeLineID sendTimeLineNextTLI = 0 ;static bool sendTimeLineIsHistoric = false ;static XLogRecPtr sendTimeLineValidUpto = InvalidXLogRecPtr;/* * How far have we sent WAL already ? This is also advertised in * MyWalSnd - > sentPtr . ( Actually , this is the next WAL location to send . ) static XLogRecPtr sentPtr = InvalidXLogRecPtr;/* Buffers for constructing outgoing messages and processing reply messages. */ static StringInfoData output_message;static StringInfoData reply_message;static StringInfoData tmpbuf;/* Timestamp of last ProcessRepliesIfAny(). */ static TimestampTz last_processing = 0 ;/* * Timestamp of last ProcessRepliesIfAny ( ) that saw a reply from the * standby . Set to 0 if wal_sender_timeout doesn ' t need to be active . static TimestampTz last_reply_timestamp = 0 ;/* Have we sent a heartbeat message asking for reply, since last reply? */ static bool waiting_for_ping_response = false ;/* * While streaming WAL in Copy mode , streamingDoneSending is set to true * after we have sent CopyDone . We should not send any more CopyData messages * after that . streamingDoneReceiving is set to true when we receive CopyDone * from the other end . When both become true , it ' s time to exit Copy mode . static bool streamingDoneSending;static bool streamingDoneReceiving;/* Are we there yet? */ static bool WalSndCaughtUp = false ;/* Flags set by signal handlers for later service in main loop */ static volatile sig_atomic_t got_SIGUSR2 = false ;static volatile sig_atomic_t got_STOPPING = false ;/* * This is set while we are streaming . When not set * PROCSIG_WALSND_INIT_STOPPING signal will be handled like SIGTERM . When set , * the main loop is responsible for checking got_STOPPING and terminating when * it ' s set ( after streaming any remaining WAL ) . static volatile sig_atomic_t replication_active = false ;static LogicalDecodingContext *logical_decoding_ctx = NULL;/* A sample associating a WAL location with the time it was written. */ typedef struct /* The size of our buffer of time samples. */ #define LAG_TRACKER_BUFFER_SIZE 8192 /* A mechanism for tracking replication lag. */ typedef struct int write_head;int read_heads[NUM_SYNC_REP_WAIT_MODE];/* * Overflow entries for read heads that collide with the write head . * * When the cyclic buffer fills ( write head is about to collide with a * read head ) , we save that read head ' s current sample here and mark it as * using overflow ( read_heads [ i ] = - 1 ) . This allows the write head to * continue advancing while the overflowed mode continues lag computation * using the saved sample . * * Once the standby ' s reported LSN advances past the overflow entry ' s LSN , * we transition back to normal buffer - based tracking . static LagTracker *lag_tracker;/* Signal handlers */ static void WalSndLastCycleHandler(SIGNAL_ARGS);/* Prototypes for private functions */ typedef void (*WalSndSendDataCallback) (void );static void WalSndLoop(WalSndSendDataCallback send_data);static void InitWalSenderSlot(void );static void WalSndKill(int code, Datum arg);static void WalSndShutdown(void );static void XLogSendPhysical(void );static void XLogSendLogical(void );static void WalSndDone(WalSndSendDataCallback send_data);static void IdentifySystem(void );static void UploadManifest(void );static bool HandleUploadManifestPacket(StringInfo buf, off_t *offset,static void ReadReplicationSlot(ReadReplicationSlotCmd *cmd);static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd);static void DropReplicationSlot(DropReplicationSlotCmd *cmd);static void StartReplication(StartReplicationCmd *cmd);static void StartLogicalReplication(StartReplicationCmd *cmd);static void ProcessStandbyMessage(void );static void ProcessStandbyReplyMessage(void );static void ProcessStandbyHSFeedbackMessage(void );static void ProcessRepliesIfAny(void );static void ProcessPendingWrites(void );static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr);static void WalSndKeepaliveIfNecessary(void );static void WalSndCheckTimeOut(void );static long WalSndComputeSleeptime(TimestampTz now);static void WalSndWait(uint32 socket_events, long timeout, uint32 wait_event);static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);static void WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write);static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,bool skipped_xact);static XLogRecPtr WalSndWaitForWal(XLogRecPtr loc);static void LagTrackerWrite(XLogRecPtr lsn, TimestampTz local_flush_time);static TimeOffset LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now);static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch);static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo,/* Initialize walsender process before entering the main command loop */ void void )/* Create a per-walsender data structure in shared memory */ /* need resource owner for e.g. basebackups */ /* * Let postmaster know that we ' re a WAL sender . Once we ' ve declared us as * a WAL sender process , postmaster will let us outlive the bgwriter and * kill us last in the shutdown sequence , so we get a chance to stream all * remaining WAL at shutdown , including the shutdown checkpoint . Note that * there ' s no going back , and we mustn ' t write any WAL records after this . /* * If the client didn ' t specify a database to connect to , show in PGPROC * that our advertised xmin should affect vacuum horizons in all * databases . This allows physical replication clients to send hot * standby feedback that will delay vacuum cleanup in all databases . if (MyDatabaseId == InvalidOid)/* Initialize empty timestamp buffer for lag tracking. */ sizeof (LagTracker));/* * Clean up after an error . * * WAL sender processes don ' t use transactions like regular backends do . * This function does any cleanup required after an error in a WAL sender * process , similar to what transaction abort does in a regular backend . void void )if (xlogreader != NULL && xlogreader->seg.ws_file >= 0 )if (MyReplicationSlot != NULL)false );false ;/* * If there is a transaction in progress , it will clean up our * ResourceOwner , but if a replication command set up a resource owner * without a transaction , we ' ve got to clean that up now . if (!IsTransactionOrTransactionBlock())false );if (got_STOPPING || got_SIGUSR2)0 );/* Revert back to startup state */ /* * Handle a client ' s connection abort in an orderly manner . static void void )/* * Reset whereToSendOutput to prevent ereport from attempting to send any * more messages to the standby . if (whereToSendOutput == DestRemote)0 );/* keep the compiler quiet */ /* * Handle the IDENTIFY_SYSTEM command . static void void )char sysid[32 ];char xloc[MAXFNAMELEN];char *dbname = NULL;4 ];bool nulls[4 ] = {0 };/* * Reply with a result set with one row , four columns . First col is system * ID , second is timeline ID , third is current xlog location and the * fourth contains the database name if we are connected to one . sizeof (sysid), UINT64_FORMAT,if (am_cascading_walsender)else sizeof (xloc), "%X/%X" , LSN_FORMAT_ARGS(logptr));if (MyDatabaseId != InvalidOid)/* syscache access needs a transaction env. */ /* copy dbname out of TX context */ /* need a tuple descriptor representing four columns */ 4 );1 , "systemid" ,1 , 0 );2 , "timeline" ,1 , 0 );3 , "xlogpos" ,1 , 0 );4 , "dbname" ,1 , 0 );/* prepare for projection of tuples */ /* column 1: system identifier */ 0 ] = CStringGetTextDatum(sysid);/* column 2: timeline */ 1 ] = Int64GetDatum(currTLI);/* column 3: wal location */ 2 ] = CStringGetTextDatum(xloc);/* column 4: database name, or NULL if none */ if (dbname)3 ] = CStringGetTextDatum(dbname);else 3 ] = true ;/* send it to dest */ /* Handle READ_REPLICATION_SLOT command */ static void #define READ_REPLICATION_SLOT_COLS 3 0 };bool nulls[READ_REPLICATION_SLOT_COLS];1 , "slot_type" ,1 , 0 );2 , "restart_lsn" ,1 , 0 );/* TimeLineID is unsigned, so int4 is not wide enough. */ 3 , "restart_tli" ,1 , 0 );true , READ_REPLICATION_SLOT_COLS * sizeof (bool ));false );if (slot == NULL || !slot->in_use)else int i = 0 ;/* Copy slot contents while holding spinlock */ if (OidIsValid(slot_contents.data.database))"cannot use %s with a logical replication slot" ,"READ_REPLICATION_SLOT" ));/* slot type */ "physical" );false ;/* start LSN */ if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))char xloc[64 ];sizeof (xloc), "%X/%X" ,false ;/* timeline this WAL was produced on */ if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))/* * While in recovery , use as timeline the currently - replaying one * to get the LSN position ' s history . if (RecoveryInProgress())void ) GetXLogReplayRecPtr(¤t_timeline);else false ;/* * Handle TIMELINE_HISTORY command . static void char histfname[MAXFNAMELEN];char path[MAXPGPATH];int fd;/* * Reply with a result set with one row , and two columns . The first col is * the name of the history file , 2 nd is the contents . 2 );1 , "filename" , TEXTOID, -1 , 0 );2 , "content" , TEXTOID, -1 , 0 );/* Send a RowDescription message */ /* Send a DataRow message */ 2 ); /* # of columns */ /* col1 len */ if (fd < 0 )"could not open file \" %s\": %m" , path)));/* Determine file length and send it to client */ 0 , SEEK_END);if (histfilelen < 0 )"could not seek to end of file \" %s\": %m" , path)));if (lseek(fd, 0 , SEEK_SET) != 0 )"could not seek to beginning of file \" %s\": %m" , path)));/* col2 len */ while (bytesleft > 0 )int nread;sizeof (rbuf));if (nread < 0 )"could not read file \" %s\": %m" ,else if (nread == 0 )"could not read file \" %s\": read %d of %zu" ,if (CloseTransientFile(fd) != 0 )"could not close file \" %s\": %m" , path)));/* * Handle UPLOAD_MANIFEST command . static void void )0 ;/* * parsing the manifest will use the cryptohash stuff , which requires a * resource owner /* Prepare to read manifest data into a temporary context. */ "incremental backup information" ,/* Send a CopyInResponse message */ 0 );0 );/* Receive packets from client until done. */ while (HandleUploadManifestPacket(&buf, &offset, ib))/* Finish up manifest processing. */ /* * Discard any old manifest information and arrange to preserve the new * information we just got . * * We assume that MemoryContextDelete and MemoryContextSetParent won ' t * fail , and thus we shouldn ' t end up bailing out of here in such a way as * to leave dangling pointers . if (uploaded_manifest_mcxt != NULL)/* clean up the resource owner we created */ true );/* * Process one packet received during the handling of an UPLOAD_MANIFEST * operation . * * ' buf ' is scratch space . This function expects it to be initialized , doesn ' t * care what the current contents are , and may override them with completely * new contents . * * The return value is true if the caller should continue processing * additional packets and false if the UPLOAD_MANIFEST operation is complete . static bool int mtype;int maxmsglen;if (mtype == EOF)"unexpected EOF on client connection with an open transaction" )));switch (mtype)case 'd' : /* CopyData */ break ;case 'c' : /* CopyDone */ case 'f' : /* CopyFail */ case 'H' : /* Flush */ case 'S' : /* Sync */ break ;default :"unexpected message type 0x%02X during COPY from stdin" ,0 ; /* keep compiler quiet */ break ;/* Now collect the message body */ if (pq_getmessage(buf, maxmsglen))"unexpected EOF on client connection with an open transaction" )));/* Process the message */ switch (mtype)case 'd' : /* CopyData */ return true ;case 'c' : /* CopyDone */ return false ;case 'H' : /* Sync */ case 'S' : /* Flush */ /* Ignore these while in CopyOut mode as we do elsewhere. */ return true ;case 'f' :"COPY from stdin failed: %s" ,/* Not reached. */ false );return false ;/* * Handle START_REPLICATION command . * * At the moment , this never returns , but an ereport ( ERROR ) will take us back * to the main loop . static void /* create xlogreader for physical replication */ if (!xlogreader)"out of memory" ),"Failed while allocating a WAL reading processor." )));/* * We assume here that we ' re logging enough information in the WAL for * log - shipping , since this is checked in PostmasterMain ( ) . * * NOTE : wal_level can only change at shutdown , so in most cases it is * difficult for there to be WAL data that we can still see that was * written at wal_level = ' minimal ' . if (cmd->slotname)true , true );if (SlotIsLogical(MyReplicationSlot))"cannot use a logical replication slot for physical replication" )));/* * We don ' t need to verify the slot ' s restart_lsn here ; instead we * rely on the caller requesting the starting point to use . If the * WAL segment doesn ' t exist , we ' ll fail later . /* * Select the timeline . If it was given explicitly by the client , use * that . Otherwise use the timeline of the last replayed record . if (am_cascading_walsender)else if (cmd->timeline != 0 )if (sendTimeLine == FlushTLI)false ;else true ;/* * Check that the timeline the client requested exists , and the * requested start location is on that timeline . /* * Found the requested timeline in the history . Check that * requested startpoint is on that timeline in our history . * * This is quite loose on purpose . We only check that we didn ' t * fork off the requested timeline before the switchpoint . We * don ' t check that we switched * to * it before the requested * starting point . This is because the client can legitimately * request to start replication from the beginning of the WAL * segment that contains switchpoint , but on the new timeline , so * that it doesn ' t end up with a partial segment . If you ask for * too old a starting point , you ' ll get an error later when we * fail to find the requested WAL segment in pg_wal . * * XXX : we could be more strict here and only allow a startpoint * that ' s older than the switchpoint , if it ' s still in the same * WAL segment . if (!XLogRecPtrIsInvalid(switchpoint) &&"requested starting point %X/%X on timeline %u is not in this server's history" ,"This server's history forked from timeline %u at %X/%X." ,else false ;false ;/* If there is nothing to stream, don't even enter COPY mode */ if (!sendTimeLineIsHistoric || cmd->startpoint < sendTimeLineValidUpto)/* * When we first start replication the standby will be behind the * primary . For some applications , for example synchronous * replication , it is important to have a clear state for this initial * catchup mode , so we can trigger actions when we change streaming * state later . We may stay in this state for a long time , which is * exactly why we want to be able to monitor whether or not we are * still here . /* Send a CopyBothResponse message, and start streaming */ 0 );0 );/* * Don ' t allow a request to stream from a future point in WAL that * hasn ' t been flushed to disk in this server yet . if (FlushPtr < cmd->startpoint)"requested starting point %X/%X is ahead of the WAL flush position of this server %X/%X" ,/* Start streaming from the requested point */ /* Initialize shared memory status, too */ /* Main loop of walsender */ true ;false ;if (got_STOPPING)0 );if (cmd->slotname)/* * Copy is finished now . Send a single - row result set indicating the next * timeline . if (sendTimeLineIsHistoric)char startpos_str[8 + 1 + 8 + 1 ];2 ];bool nulls[2 ] = {0 };sizeof (startpos_str), "%X/%X" ,/* * Need a tuple descriptor representing two columns . int8 may seem * like a surprising data type for this , but in theory int4 would not * be wide enough for this , as TimeLineID is unsigned . 2 );1 , "next_tli" ,1 , 0 );2 , "next_tli_startpos" ,1 , 0 );/* prepare for projection of tuple */ 0 ] = Int64GetDatum((int64) sendTimeLineNextTLI);1 ] = CStringGetTextDatum(startpos_str);/* send it to dest */ /* Send CommandComplete message */ "START_STREAMING" );/* * XLogReaderRoutine - > page_read callback for logical decoding contexts , as a * walsender process . * * Inside the walsender we can do better than read_local_xlog_page , * which has to do a plain sleep / busy loop , because the walsender ' s latch gets * set every time WAL is flushed . static int int reqLen,char *cur_page)int count;/* * Make sure we have enough WAL available before retrieving the current * timeline . /* Fail if not enough (implies we are going to shut down) */ if (flushptr < targetPagePtr + reqLen)return -1 ;/* * Since logical decoding is also permitted on a standby server , we need * to check if the server is in recovery to decide how to get the current * timeline ID ( so that it also covers the promotion or timeline change * cases ) . We must determine am_cascading_walsender after waiting for the * required WAL so that it is correct when the walsender wakes up after a * promotion . if (am_cascading_walsender)else if (targetPagePtr + XLOG_BLCKSZ <= flushptr)/* more than one block available */ else /* part of the page available */ /* now actually read the data, we know it's there */ if (!WALRead(state,/* Pass the current TLI because only * WalSndSegmentOpen controls whether new TLI /* * After reading into the buffer , check that what we read was valid . We do * this after reading , because even though the segment was present when we * opened it , it might get recycled or removed while we read it . The * read ( ) succeeds in that case , but the data we tried to read might * already have been overwritten with new WAL records . return count;/* * Process extra options given to CREATE_REPLICATION_SLOT . static void bool *reserve_wal,bool *two_phase, bool *failover)bool snapshot_action_given = false ;bool reserve_wal_given = false ;bool two_phase_given = false ;bool failover_given = false ;/* Parse options */ if (strcmp(defel->defname, "snapshot" ) == 0 )char *action;if (snapshot_action_given || cmd->kind != REPLICATION_KIND_LOGICAL)"conflicting or redundant options" )));true ;if (strcmp(action, "export" ) == 0 )else if (strcmp(action, "nothing" ) == 0 )else if (strcmp(action, "use" ) == 0 )else "unrecognized value for %s option \" %s\": \" %s\"" ,"CREATE_REPLICATION_SLOT" , defel->defname, action)));else if (strcmp(defel->defname, "reserve_wal" ) == 0 )if (reserve_wal_given || cmd->kind != REPLICATION_KIND_PHYSICAL)"conflicting or redundant options" )));true ;else if (strcmp(defel->defname, "two_phase" ) == 0 )if (two_phase_given || cmd->kind != REPLICATION_KIND_LOGICAL)"conflicting or redundant options" )));true ;else if (strcmp(defel->defname, "failover" ) == 0 )if (failover_given || cmd->kind != REPLICATION_KIND_LOGICAL)"conflicting or redundant options" )));true ;else "unrecognized option: %s" , defel->defname);/* * Create a new replication slot . static void const char *snapshot_name = NULL;char xloc[MAXFNAMELEN];char *slot_name;bool reserve_wal = false ;bool two_phase = false ;bool failover = false ;4 ];bool nulls[4 ] = {0 };if (cmd->kind == REPLICATION_KIND_PHYSICAL)false ,false , false , false );if (reserve_wal)/* Write this slot to disk if it's a permanent one. */ if (!cmd->temporary)else bool need_full_snapshot = false ;/* * Initially create persistent slot as ephemeral - that allows us to * nicely handle errors during initialization because it ' ll get * dropped if this transaction fails . We ' ll make it persistent at the * end . Temporary slots can be created as temporary from beginning as * they get dropped on error as well . true ,false );/* * Do options check early so that we can bail before calling the * DecodingContextFindStartpoint which can take long time . if (snapshot_action == CRS_EXPORT_SNAPSHOT)if (IsTransactionBlock())/*- translator: %s is a CREATE_REPLICATION_SLOT statement */ "%s must not be called inside a transaction" ,"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'export')" )));true ;else if (snapshot_action == CRS_USE_SNAPSHOT)if (!IsTransactionBlock())/*- translator: %s is a CREATE_REPLICATION_SLOT statement */ "%s must be called inside a transaction" ,"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')" )));if (XactIsoLevel != XACT_REPEATABLE_READ)/*- translator: %s is a CREATE_REPLICATION_SLOT statement */ "%s must be called in REPEATABLE READ isolation mode transaction" ,"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')" )));if (!XactReadOnly)/*- translator: %s is a CREATE_REPLICATION_SLOT statement */ "%s must be called in a read-only transaction" ,"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')" )));if (FirstSnapshotSet)/*- translator: %s is a CREATE_REPLICATION_SLOT statement */ "%s must be called before any query" ,"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')" )));if (IsSubTransaction())/*- translator: %s is a CREATE_REPLICATION_SLOT statement */ "%s must not be called in a subtransaction" ,"CREATE_REPLICATION_SLOT ... (SNAPSHOT 'use')" )));true ;/* * Signal that we don ' t need the timeout mechanism . We ' re just * creating the replication slot and don ' t yet accept feedback * messages or send keepalives . As we possibly need to wait for * further WAL the walsender would otherwise possibly be killed too * soon . 0 ;/* build initial snapshot, might take a while */ /* * Export or use the snapshot if we ' ve been asked to do so . * * NB . We will convert the snapbuild . c kind of snapshot to normal * snapshot when doing this . if (snapshot_action == CRS_EXPORT_SNAPSHOT)else if (snapshot_action == CRS_USE_SNAPSHOT)/* don't need the decoding context anymore */ if (!cmd->temporary)sizeof (xloc), "%X/%X" ,/*---------- * Need a tuple descriptor representing four columns : * - first field : the slot name * - second field : LSN at which we became consistent * - third field : exported snapshot ' s name * - fourth field : output plugin 4 );1 , "slot_name" ,1 , 0 );2 , "consistent_point" ,1 , 0 );3 , "snapshot_name" ,1 , 0 );4 , "output_plugin" ,1 , 0 );/* prepare for projection of tuples */ /* slot_name */ 0 ] = CStringGetTextDatum(slot_name);/* consistent wal location */ 1 ] = CStringGetTextDatum(xloc);/* snapshot name, or NULL if none */ if (snapshot_name != NULL)2 ] = CStringGetTextDatum(snapshot_name);else 2 ] = true ;/* plugin, or NULL if none */ if (cmd->plugin != NULL)3 ] = CStringGetTextDatum(cmd->plugin);else 3 ] = true ;/* send it to dest */ /* * Get rid of a replication slot that is no longer wanted . static void /* * Change the definition of a replication slot . static void bool failover_given = false ;bool two_phase_given = false ;bool failover;bool two_phase;/* Parse options */ if (strcmp(defel->defname, "failover" ) == 0 )if (failover_given)"conflicting or redundant options" )));true ;else if (strcmp(defel->defname, "two_phase" ) == 0 )if (two_phase_given)"conflicting or redundant options" )));true ;else "unrecognized option: %s" , defel->defname);/* * Load previously initiated logical slot and prepare for sending data ( via * WalSndLoop ) . static void /* make sure that our requirements are still fulfilled */ true , true );/* * Force a disconnect , so that the decoding code doesn ' t need to care * about an eventual switch from running in recovery , to running in a * normal environment . Client code is expected to handle reconnects . if (am_cascading_walsender && !RecoveryInProgress())"terminating walsender process after promotion" )));true ;/* * Create our decoding context , making it start at the previously ack ' ed * position . * * Do this before sending a CopyBothResponse message , so that any errors * are reported early . false ,/* Send a CopyBothResponse message, and start streaming */ 0 );0 );/* Start reading WAL from the oldest required WAL. */ /* * Report the location after which we ' ll send out further commits as the * current sentPtr . /* Also update the sent position status in shared memory */ true ;/* Main loop of walsender */ false ;if (got_STOPPING)0 );/* Get out of COPY mode (CommandComplete). */ 0 );false );/* * LogicalDecodingContext ' prepare_write ' callback . * * Prepare a write into a StringInfo . * * Don ' t do anything lasting in here , it ' s quite possible that nothing will be done * with the data . static void bool last_write)/* can't have sync rep confused by sending the same LSN several times */ if (!last_write)'w' );/* dataStart */ /* walEnd */ /* * Fill out the sendtime later , just as it ' s done in XLogSendPhysical , but * reserve space here . 0 ); /* sendtime */ /* * LogicalDecodingContext ' write ' callback . * * Actually write out data previously prepared by WalSndPrepareWrite out to * the network . Take as long as needed , but process replies from the other * side and check timeouts during that . static void bool last_write)/* * Fill the send timestamp last , so that it is taken as late as possible . * This is somewhat ugly , but the protocol is set as it ' s already used for * several releases by streaming physical replication . 1 + sizeof (int64) + sizeof (int64)],sizeof (int64));/* output previously gathered data in a CopyData packet */ 'd' , ctx->out->data, ctx->out->len);/* Try to flush pending output to the client */ if (pq_flush_if_writable() != 0 )/* Try taking fast path unless we get too close to walsender timeout. */ if (now < TimestampTzPlusMilliseconds(last_reply_timestamp,2 ) &&return ;/* If we have pending write here, go to slow path */ /* * Wait until there is no pending write . Also process replies from the other * side and check timeouts during that . static void void )for (;;)long sleeptime;/* Check for input from the client */ /* die if timeout was reached */ /* Send keepalive if the time has come */ if (!pq_is_send_pending())break ;/* Sleep until something happens or we time out */ /* Clear any already-pending wakeups */ /* Process any requests or signals received recently */ if (ConfigReloadPending)false ;/* Try to flush pending output to the client */ if (pq_flush_if_writable() != 0 )/* reactivate latch so WalSndLoop knows to continue */ /* * LogicalDecodingContext ' update_progress ' callback . * * Write the current position to the lag tracker ( see XLogSendPhysical ) . * * When skipping empty transactions , send a keepalive message if necessary . static void bool skipped_xact)static TimestampTz sendTime = 0 ;bool pending_writes = false ;bool end_xact = ctx->end_xact;/* * Track lag no more than once per WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS to * avoid flooding the lag tracker when we commit frequently . * * We don ' t have a mechanism to get the ack for any LSN other than end * xact LSN from the downstream . So , we track lag only for end of * transaction LSN . #define WALSND_LOGICAL_LAG_TRACK_INTERVAL_MS 1000 if (end_xact && TimestampDifferenceExceeds(sendTime, now,/* * When skipping empty transactions in synchronous replication , we send a * keepalive message to avoid delaying such transactions . * * It is okay to check sync_standbys_status without lock here as in the * worst case we will just send an extra keepalive message when it is * really not required . if (skipped_xact &&volatile WalSndCtlData *) WalSndCtl)->sync_standbys_status & SYNC_STANDBY_DEFINED))false , lsn);/* Try to flush pending output to the client */ if (pq_flush_if_writable() != 0 )/* If we have pending write here, make sure it's actually flushed */ if (pq_is_send_pending())true ;/* * Process pending writes if any or try to send a keepalive if required . * We don ' t need to try sending keep alive messages at the transaction end * as that will be done at a later point in time . This is required only * for large transactions where we don ' t send any changes to the * downstream and the receiver can timeout due to that . if (pending_writes || (!end_xact &&2 )))/* * Wake up the logical walsender processes with logical failover slots if the * currently acquired physical slot is specified in synchronized_standby_slots GUC . void void )/* * If we are running in a standby , there is no need to wake up walsenders . * This is because we do not support syncing slots to cascading standbys , * so , there are no walsenders waiting for standbys to catch up . if (RecoveryInProgress())return ;if (SlotExistsInSyncStandbySlots(NameStr(MyReplicationSlot->data.name)))/* * Returns true if not all standbys have caught up to the flushed position * ( flushed_lsn ) when the current acquired slot is a logical failover * slot and we are streaming ; otherwise , returns false . * * If returning true , the function sets the appropriate wait event in * wait_event ; otherwise , wait_event is set to 0 . static bool int elevel = got_STOPPING ? ERROR : WARNING;bool failover_slot;/* * Note that after receiving the shutdown signal , an ERROR is reported if * any slots are dropped , invalidated , or inactive . This measure is taken * to prevent the walsender from waiting indefinitely . if (failover_slot && !StandbySlotsHaveCaughtup(flushed_lsn, elevel))return true ;0 ;return false ;/* * Returns true if we need to wait for WALs to be flushed to disk , or if not * all standbys have caught up to the flushed position ( flushed_lsn ) when the * current acquired slot is a logical failover slot and we are * streaming ; otherwise , returns false . * * If returning true , the function sets the appropriate wait event in * wait_event ; otherwise , wait_event is set to 0 . static bool /* Check if we need to wait for WALs to be flushed to disk */ if (target_lsn > flushed_lsn)return true ;/* Check if the standby slots have caught up to the flushed position */ return NeedToWaitForStandbys(flushed_lsn, wait_event);/* * Wait till WAL < loc is flushed to disk so it can be safely sent to client . * * If the walsender holds a logical failover slot , we also wait for all the * specified streaming replication standby servers to confirm receipt of WAL * up to RecentFlushPtr . It is beneficial to wait here for the confirmation * up to RecentFlushPtr rather than waiting before transmitting each change * to logical subscribers , which is already covered by RecentFlushPtr . * * Returns end LSN of flushed WAL . Normally this will be > = loc , but if we * detect a shutdown request ( either from postmaster or client ) we will return * early , so caller must always check . static XLogRecPtrint wakeEvents;0 ;static XLogRecPtr RecentFlushPtr = InvalidXLogRecPtr;0 ;/* * Fast path to avoid acquiring the spinlock in case we already know we * have enough WAL available and all the standby servers have confirmed * receipt of WAL up to RecentFlushPtr . This is particularly interesting * if we ' re far behind . if (!XLogRecPtrIsInvalid(RecentFlushPtr) &&return RecentFlushPtr;/* * Within the loop , we wait for the necessary WALs to be flushed to disk * first , followed by waiting for standbys to catch up if there are enough * WALs ( see NeedToWaitForWal ( ) ) or upon receiving the shutdown signal . for (;;)bool wait_for_standby_at_stop = false ;long sleeptime;/* Clear any already-pending wakeups */ /* Process any requests or signals received recently */ if (ConfigReloadPending)false ;/* Check for input from the client */ /* * If we ' re shutting down , trigger pending WAL to be written out , * otherwise we ' d possibly end up waiting for WAL that never gets * written , because walwriter has shut down already . * * Note that GetXLogInsertEndRecPtr ( ) is used to obtain the WAL flush * request location instead of GetXLogInsertRecPtr ( ) . Because if the * last WAL record ends at a page boundary , GetXLogInsertRecPtr ( ) can * return an LSN pointing past the page header , which may cause * XLogFlush ( ) to report an error . if (got_STOPPING && !RecoveryInProgress())/* * To avoid the scenario where standbys need to catch up to a newer * WAL location in each iteration , we update our idea of the currently * flushed position only if we are not waiting for standbys to catch * up . if (wait_event != WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION)if (!RecoveryInProgress())else /* * If postmaster asked us to stop and the standby slots have caught up * to the flushed position , don ' t wait anymore . * * It ' s important to do this check after the recomputation of * RecentFlushPtr , so we can send all remaining data before shutting * down . if (got_STOPPING)if (NeedToWaitForStandbys(RecentFlushPtr, &wait_event))true ;else break ;/* * We only send regular messages to the client for full decoded * transactions , but a synchronous replication and walsender shutdown * possibly are waiting for a later location . So , before sleeping , we * send a ping containing the flush location . If the receiver is * otherwise idle , this keepalive will trigger a reply . Processing the * reply will update these MyWalSnd locations . if (MyWalSnd->flush < sentPtr &&false , InvalidXLogRecPtr);/* * Exit the loop if already caught up and doesn ' t need to wait for * standby slots . if (!wait_for_standby_at_stop &&break ;/* * Waiting for new WAL or waiting for standbys to catch up . Since we * need to wait , we ' re now caught up . true ;/* * Try to flush any pending output to the client . if (pq_flush_if_writable() != 0 )/* * If we have received CopyDone from the client , sent CopyDone * ourselves , and the output buffer is empty , it ' s time to exit * streaming , so fail the current WAL fetch request . if (streamingDoneReceiving && streamingDoneSending &&break ;/* die if timeout was reached */ /* Send keepalive if the time has come */ /* * Sleep until something happens or we time out . Also wait for the * socket becoming writable , if there ' s still pending output . * Otherwise we might sit on sendable output data while waiting for * new WAL to be generated . ( But if we have nothing to send , we don ' t * want to wake on socket - writable . ) if (pq_is_send_pending())0 );/* Report IO statistics, if needed */ if (TimestampDifferenceExceeds(last_flush, now,false );void ) pgstat_flush_backend(false , PGSTAT_BACKEND_FLUSH_IO);/* reactivate latch so WalSndLoop knows to continue */ return RecentFlushPtr;/* * Execute an incoming replication command . * * Returns true if the cmd_string was recognized as WalSender command , false * if not . bool const char *cmd_string)int parse_rc;const char *cmdtag;/* We save and re-use the cmd_context across calls */ static MemoryContext cmd_context = NULL;/* * If WAL sender has been told that shutdown is getting close , switch its * status accordingly to handle the next replication commands correctly . if (got_STOPPING)/* * Throw error if in stopping mode . We need prevent commands that could * generate WAL while the shutdown checkpoint is being written . To be * safe , we just prohibit all new commands . if (MyWalSnd->state == WALSNDSTATE_STOPPING)"cannot execute new commands while WAL sender is in stopping mode" )));/* * CREATE_REPLICATION_SLOT . . . LOGICAL exports a snapshot until the next * command arrives . Clean up the old stuff if there ' s anything . /* * Prepare to parse and execute the command . * * Because replication command execution can involve beginning or ending * transactions , we need a working context that will survive that , so we * make it a child of TopMemoryContext . That in turn creates a hazard of * long - lived memory leaks if we lose track of the working context . We * deal with that by creating it only once per walsender , and resetting it * for each new command . ( Normally this reset is a no - op , but if the * prior exec_replication_command call failed with an error , it won ' t be . ) * * This is subtler than it looks . The transactions we manage can extend * across replication commands , indeed SnapBuildClearExportedSnapshot * might have just ended one . Because transaction exit will revert to the * memory context that was current at transaction start , we need to be * sure that that context is still valid . That motivates re - using the * same cmd_context rather than making a new one each time . if (cmd_context == NULL)"Replication command context" ,else /* * Is it a WalSender command ? if (!replication_scanner_is_replication_command(scanner))/* Nope; clean up and get out. */ /* XXX this is a pretty random place to make this check */ if (MyDatabaseId == InvalidOid)"cannot execute SQL commands in WAL sender for physical replication" )));/* Tell the caller that this wasn't a WalSender command. */ return false ;/* * Looks like a WalSender command , so parse it . if (parse_rc != 0 )"replication command parser returned %d" ,/* * Report query to various monitoring facilities . For this purpose , we * report replication commands just like SQL commands . /* * Log replication command if log_replication_commands is enabled . Even * when it ' s disabled , log the command with DEBUG1 level for backward * compatibility . "received replication command: %s" , cmd_string)));/* * Disallow replication commands in aborted transaction blocks . if (IsAbortedTransactionBlockState())"current transaction is aborted, " "commands ignored until end of transaction block" )));/* * Allocate buffers that will be used for each outgoing and incoming * message . We do this just once per command to reduce palloc overhead . switch (cmd_node->type)case T_IdentifySystemCmd:"IDENTIFY_SYSTEM" ;break ;case T_ReadReplicationSlotCmd:"READ_REPLICATION_SLOT" ;break ;case T_BaseBackupCmd:"BASE_BACKUP" ;true , cmdtag);break ;case T_CreateReplicationSlotCmd:"CREATE_REPLICATION_SLOT" ;break ;case T_DropReplicationSlotCmd:"DROP_REPLICATION_SLOT" ;break ;case T_AlterReplicationSlotCmd:"ALTER_REPLICATION_SLOT" ;break ;case T_StartReplicationCmd:"START_REPLICATION" ;true , cmdtag);if (cmd->kind == REPLICATION_KIND_PHYSICAL)else /* dupe, but necessary per libpqrcv_endstreaming */ break ;case T_TimeLineHistoryCmd:"TIMELINE_HISTORY" ;true , cmdtag);break ;case T_VariableShowStmt:"SHOW" ;/* syscache access needs a transaction environment */ break ;case T_UploadManifestCmd:"UPLOAD_MANIFEST" ;true , cmdtag);break ;default :"unrecognized replication command node tag: %u" ,/* * Done . Revert to caller ' s memory context , and clean out the cmd_context * to recover memory right away . /* * We need not update ps display or pg_stat_activity , because PostgresMain * will reset those to " idle " . But we must reset debug_query_string to * ensure it doesn ' t become a dangling pointer . return true ;/* * Process any incoming messages while streaming . Also checks if the remote * end has closed the connection . static void void )unsigned char firstchar;int maxmsglen;int r;bool received = false ;/* * If we already received a CopyDone from the frontend , any subsequent * message is the beginning of a new command , and should be processed in * the main processing loop . while (!streamingDoneReceiving)if (r < 0 )/* unexpected error or EOF */ "unexpected EOF on standby connection" )));0 );if (r == 0 )/* no data available without blocking */ break ;/* Validate message type and set packet size limit */ switch (firstchar)case PqMsg_CopyData:break ;case PqMsg_CopyDone:case PqMsg_Terminate:break ;default :"invalid standby message type \" %c\"" ,0 ; /* keep compiler quiet */ break ;/* Read the message contents */ if (pq_getmessage(&reply_message, maxmsglen))"unexpected EOF on standby connection" )));0 );/* ... and process it */ switch (firstchar)/* * ' d ' means a standby reply wrapped in a CopyData packet . case PqMsg_CopyData:true ;break ;/* * CopyDone means the standby requested to finish streaming . * Reply with CopyDone , if we had not sent that already . case PqMsg_CopyDone:if (!streamingDoneSending)'c' , NULL, 0 );true ;true ;true ;break ;/* * ' X ' means that the standby is closing down the socket . case PqMsg_Terminate:0 );default :false ); /* NOT REACHED */ /* * Save the last reply timestamp if we ' ve received at least one reply . if (received)false ;/* * Process a status update message received from standby . static void void )char msgtype;/* * Check message type from the first byte . switch (msgtype)case 'r' :break ;case 'h' :break ;default :"unexpected message type \" %c\"" , msgtype)));0 );/* * Remember that a walreceiver just confirmed receipt of lsn ` lsn ` . static void bool changed = false ;if (slot->data.restart_lsn != lsn)true ;if (changed)/* * One could argue that the slot should be saved to disk now , but that ' d * be energy wasted - the worst thing lost information could cause here is * to give wrong information in a statistics view - we ' ll just potentially * be more conservative in removing files . /* * Regular reply from standby advising of WAL locations on standby server . static void void )bool replyRequested;bool clearLagTimes;static XLogRecPtr prevWritePtr = InvalidXLogRecPtr;static XLogRecPtr prevFlushPtr = InvalidXLogRecPtr;static XLogRecPtr prevApplyPtr = InvalidXLogRecPtr;/* the caller already consumed the msgtype byte */ if (message_level_is_interesting(DEBUG2))char *replyTimeStr;/* Copy because timestamptz_to_str returns a static buffer */ "write %X/%X flush %X/%X apply %X/%X%s reply_time %s" ," (reply requested)" : "" ,/* See if we can compute the round-trip lag for these positions. */ /* * If the standby reports that it has fully replayed the WAL , and the * write / flush / apply positions remain unchanged across two consecutive * reply messages , forget the lag times measured when it last * wrote / flushed / applied a WAL record . * * The second message with unchanged positions typically results from * wal_receiver_status_interval expiring on the standby , so lag values are * usually cleared after that interval when there is no activity . This * avoids displaying stale lag data until more WAL traffic arrives . /* Send a reply if the standby requested one. */ if (replyRequested)false , InvalidXLogRecPtr);/* * Update shared state for this WalSender process based on reply data from * standby . if (writeLag != -1 || clearLagTimes)if (flushLag != -1 || clearLagTimes)if (applyLag != -1 || clearLagTimes)if (!am_cascading_walsender)/* * Advance our local xmin horizon when the client confirmed a flush . if (MyReplicationSlot && flushPtr != InvalidXLogRecPtr)if (SlotIsLogical(MyReplicationSlot))else /* compute new replication slot xmin horizon if needed */ static void talogXmin)bool changed = false ;/* * For physical replication we don ' t need the interlock provided by xmin * and effective_xmin since the consequences of a missed increase are * limited to query cancellations , so set both at once . if (!TransactionIdIsNormal(slot->data.xmin) ||true ;if (!TransactionIdIsNormal(slot->data.catalog_xmin) ||true ;if (changed)false );/* * Check that the provided xmin / epoch are sane , that is , not in the future * and not so far back as to be already wrapped around . * * Epoch of nextXid should be same as standby , or if the counter has * wrapped , then one greater than standby . * * This check doesn ' t care about whether clog exists for these xids * at all . static bool if (xid <= nextXid)if (epoch != nextEpoch)return false ;else if (epoch + 1 != nextEpoch)return false ;if (!TransactionIdPrecedesOrEquals(xid, nextXid))return false ; /* epoch OK, but it's wrapped around */ return true ;/* * Hot Standby feedback static void void )/* * Decipher the reply message . The caller already consumed the msgtype * byte . See XLogWalRcvSendHSFeedback ( ) in walreceiver . c for the creation * of this message . 4 );4 );4 );4 );if (message_level_is_interesting(DEBUG2))char *replyTimeStr;/* Copy because timestamptz_to_str returns a static buffer */ "hot standby feedback xmin %u epoch %u, catalog_xmin %u epoch %u reply_time %s" ,/* * Update shared state for this WalSender process based on reply data from * standby . /* * Unset WalSender ' s xmins if the feedback message values are invalid . * This happens when the downstream turned hot_standby_feedback off . if (!TransactionIdIsNormal(feedbackXmin)if (MyReplicationSlot != NULL)return ;/* * Check that the provided xmin / epoch are sane , that is , not in the future * and not so far back as to be already wrapped around . Ignore if not . if (TransactionIdIsNormal(feedbackXmin) &&return ;if (TransactionIdIsNormal(feedbackCatalogXmin) &&return ;/* * Set the WalSender ' s xmin equal to the standby ' s requested xmin , so that * the xmin will be taken into account by GetSnapshotData ( ) / * ComputeXidHorizons ( ) . This will hold back the removal of dead rows and * thereby prevent the generation of cleanup conflicts on the standby * server . * * There is a small window for a race condition here : although we just * checked that feedbackXmin precedes nextXid , the nextXid could have * gotten advanced between our fetching it and applying the xmin below , * perhaps far enough to make feedbackXmin wrap around . In that case the * xmin we set here would be " in the future " and have no effect . No point * in worrying about this since it ' s too late to save the desired data * anyway . Assuming that the standby sends us an increasing sequence of * xmins , this could only happen during the first reply cycle , else our * own xmin would prevent nextXid from advancing so far . * * We don ' t bother taking the ProcArrayLock here . Setting the xmin field * is assumed atomic , and there ' s no real need to prevent concurrent * horizon determinations . ( If we ' re moving our xmin forward , this is * obviously safe , and if we ' re moving it backwards , well , the data is at * risk already since a VACUUM could already have determined the horizon . ) * * If we ' re using a replication slot we reserve the xmin via that , * otherwise via the walsender ' s PGPROC entry . We can only track the * catalog xmin separately when using a slot , so we store the least of the * two provided when not using a slot . * * XXX : It might make sense to generalize the ephemeral slot concept and * always use the slot mechanism to handle the feedback xmin . if (MyReplicationSlot != NULL) /* XXX: persistency configurable? */ else if (TransactionIdIsNormal(feedbackCatalogXmin)else /* * Compute how long send / receive loops should sleep . * * If wal_sender_timeout is enabled we want to wake up in time to send * keepalives and to abort the connection if wal_sender_timeout has been * reached . static long long sleeptime = 10000 ; /* 10 s */ if (wal_sender_timeout > 0 && last_reply_timestamp > 0 )/* * At the latest stop sleeping once wal_sender_timeout has been * reached . /* * If no ping has been sent yet , wakeup when it ' s time to do so . * WalSndKeepaliveIfNecessary ( ) wants to send a keepalive once half of * the timeout passed without a response . if (!waiting_for_ping_response)2 );/* Compute relative time until wakeup. */ return sleeptime;/* * Check whether there have been responses by the client within * wal_sender_timeout and shutdown if not . Using last_processing as the * reference point avoids counting server - side stalls against the client . * However , a long server - side stall can make WalSndKeepaliveIfNecessary ( ) * postdate last_processing by more than wal_sender_timeout . If that happens , * the client must reply almost immediately to avoid a timeout . This rarely * affects the default configuration , under which clients spontaneously send a * message every standby_message_timeout = wal_sender_timeout / 6 = 10 s . We * could eliminate that problem by recognizing timeout expiration at * wal_sender_timeout / 2 after the keepalive . static void void )/* don't bail out if we're doing something that doesn't require timeouts */ if (last_reply_timestamp <= 0 )return ;if (wal_sender_timeout > 0 && last_processing >= timeout)/* * Since typically expiration of replication timeout means * communication problem , we don ' t send the error message to the * standby . "terminating walsender process due to replication timeout" )));/* Main loop of walsender process that streams the WAL over Copy messages. */ static void 0 ;/* * Initialize the last reply timestamp . That enables timeout processing * from hereon . false ;/* * Loop until we reach the end of this timeline or the client requests to * stop streaming . for (;;)/* Clear any already-pending wakeups */ /* Process any requests or signals received recently */ if (ConfigReloadPending)false ;/* Check for input from the client */ /* * If we have received CopyDone from the client , sent CopyDone * ourselves , and the output buffer is empty , it ' s time to exit * streaming . if (streamingDoneReceiving && streamingDoneSending &&break ;/* * If we don ' t have any pending data in the output buffer , try to send * some more . If there is some , we don ' t bother to call send_data * again until we ' ve flushed it . . . but we ' d better assume we are not * caught up . if (!pq_is_send_pending())else false ;/* Try to flush pending output to the client */ if (pq_flush_if_writable() != 0 )/* If nothing remains to be sent right now ... */ if (WalSndCaughtUp && !pq_is_send_pending())/* * If we ' re in catchup state , move to streaming . This is an * important state change for users to know about , since before * this point data loss might occur if the primary dies and we * need to failover to the standby . The state change is also * important for synchronous replication , since commits that * started to wait at that point might wait for some time . if (MyWalSnd->state == WALSNDSTATE_CATCHUP)"\" %s\" has now caught up with upstream server" ,/* * When SIGUSR2 arrives , we send any outstanding logs up to the * shutdown checkpoint record ( i . e . , the latest record ) , wait for * them to be replicated to the standby , and exit . This may be a * normal termination at shutdown , or a promotion , the walsender * is not sure which . if (got_SIGUSR2)/* Check for replication timeout. */ /* Send keepalive if the time has come */ /* * Block if we have unsent data . XXX For logical replication , let * WalSndWaitForWal ( ) handle any other blocking ; idle receivers need * its additional actions . For physical replication , also block if * caught up ; its send_data does not block . * * The IO statistics are reported in WalSndWaitForWal ( ) for the * logical WAL senders . if ((WalSndCaughtUp && send_data != XLogSendLogical &&long sleeptime;int wakeEvents;if (!streamingDoneReceiving)else 0 ;/* * Use fresh timestamp , not last_processing , to reduce the chance * of reaching wal_sender_timeout before sending a keepalive . if (pq_is_send_pending())/* Report IO statistics, if needed */ if (TimestampDifferenceExceeds(last_flush, now,false );void ) pgstat_flush_backend(false , PGSTAT_BACKEND_FLUSH_IO);/* Sleep until something happens or we time out */ /* Initialize a per-walsender data structure for this walsender process */ static void void )int i;/* * WalSndCtl should be set up already ( we inherit this by fork ( ) or * EXEC_BACKEND mechanism from the postmaster ) . /* * Find a free walsender slot and reserve it . This must not fail due to * the prior check for free WAL senders in InitProcess ( ) . for (i = 0 ; i < max_wal_senders; i++)if (walsnd->pid != 0 )continue ;else /* * Found a free slot . Reserve it for us . false ;1 ;1 ;1 ;0 ;0 ;/* * The kind assignment is done here and not in StartReplication ( ) * and StartLogicalReplication ( ) . Indeed , the logical walsender * needs to read WAL records ( like snapshot of running * transactions ) during the slot creation . So it needs to be woken * up based on its kind . * * The kind assignment could also be done in StartReplication ( ) , * StartLogicalReplication ( ) and CREATE_REPLICATION_SLOT but it * seems better to set it on one place . if (MyDatabaseId == InvalidOid)else /* don't need the lock anymore */ break ;/* Arrange to clean up at walsender exit */ 0 );/* Destroy the per-walsender data structure for this walsender process */ static void int code, Datum arg)/* Mark WalSnd struct as no longer being in use. */ 0 ;/* XLogReaderRoutine->segment_open callback */ static void char path[MAXPGPATH];/*------- * When reading from a historic timeline , and there is a timeline switch * within this segment , read from the WAL segment belonging to the new * timeline . * * For example , imagine that this server is currently on timeline 5 , and * we ' re streaming timeline 4 . The switch from timeline 4 to 5 happened at * 0 / 13002088 . In pg_wal , we have these files : * * . . . * 000000040000000000000012 * 000000040000000000000013 * 000000050000000000000013 * 000000050000000000000014 * . . . * * In this situation , when requested to send the WAL from segment 0 x13 , on * timeline 4 , we read the WAL from file 000000050000000000000013 . Archive * recovery prefers files from newer timelines , so if the segment was * restored from the archive on this server , the file belonging to the old * timeline , 000000040000000000000013 , might not exist . Their contents are * equal up to the switchpoint , because at a timeline switch , the used * portion of the old segment is copied to the new file . if (sendTimeLineIsHistoric)if (nextSegNo == endSegNo)if (state->seg.ws_file >= 0 )return ;/* * If the file is not found , assume it ' s because the standby asked for a * too old WAL segment that has already been removed or recycled . if (errno == ENOENT)char xlogfname[MAXFNAMELEN];int save_errno = errno;"requested WAL segment %s has already been removed" ,else "could not open file \" %s\": %m" ,/* * Send out the WAL in its normal physical / stored form . * * Read up to MAX_SEND_SIZE bytes of WAL that ' s been flushed to disk , * but not yet sent to the client , and buffer it in the libpq output * buffer . * * If there is no unsent WAL remaining , WalSndCaughtUp is set to true , * otherwise WalSndCaughtUp is set to false . static void void )/* If requested switch the WAL sender to the stopping state. */ if (got_STOPPING)if (streamingDoneSending)true ;return ;/* Figure out how far we can safely send the WAL. */ if (sendTimeLineIsHistoric)/* * Streaming an old timeline that ' s in this server ' s history , but is * not the one we ' re currently inserting or replaying . It can be * streamed up to the point where we switched off that timeline . else if (am_cascading_walsender)/* * Streaming the latest timeline on a standby . * * Attempt to send all WAL that has already been replayed , so that we * know it ' s valid . If we ' re receiving WAL through streaming * replication , it ' s also OK to send any WAL that has been received * but not replayed . * * The timeline we ' re recovering from can change , or we can be * promoted . In either case , the current timeline becomes historic . We * need to detect that so that we don ' t try to stream past the point * where we switched to another timeline . We check for promotion or * timeline switch after calculating FlushPtr , to avoid a race * condition : if the timeline becomes historic just after we checked * that it was still current , it ' s still be OK to stream it up to the * FlushPtr that was calculated before it became historic . bool becameHistoric = false ;if (!RecoveryInProgress())/* We have been promoted. */ false ;true ;else /* * Still a cascading standby . But is the timeline we ' re sending * still the one recovery is recovering from ? if (sendTimeLine != SendRqstTLI)true ;if (becameHistoric)/* * The timeline we were sending has become historic . Read the * timeline history file of the new timeline to see where exactly * we forked off from the timeline we were sending . true ;else /* * Streaming the current timeline on a primary . * * Attempt to send all data that ' s already been written out and * fsync ' d to disk . We cannot go further than what ' s been written out * given the current implementation of WALRead ( ) . And in any case * it ' s unsafe to send WAL that is not securely down to disk on the * primary : if the primary subsequently crashes and restarts , standbys * must not have applied any WAL that got lost on the primary . /* * Record the current system time as an approximation of the time at which * this WAL location was written for the purposes of lag tracking . * * In theory we could make XLogFlush ( ) record a time in shmem whenever WAL * is flushed and we could get that time as well as the LSN when we call * GetFlushRecPtr ( ) above ( and likewise for the cascading standby * equivalent ) , but rather than putting any new code into the hot WAL path * it seems good enough to capture the time here . We should reach this * after XLogFlush ( ) runs WalSndWakeupProcessRequests ( ) , and although that * may take some time , we read the WAL flush pointer and take the time * very close to together here so that we ' ll get a later position if it is * still moving . * * Because LagTrackerWrite ignores samples when the LSN hasn ' t advanced , * this gives us a cheap approximation for the WAL flush time for this * LSN . * * Note that the LSN is not necessarily the LSN for the data contained in * the present message ; it ' s the end of the WAL , which might be further * ahead . All the lag tracking machinery cares about is finding out when * that arbitrary LSN is eventually reported as written , flushed and * applied , so that it can measure the elapsed time . /* * If this is a historic timeline and we ' ve reached the point where we * forked to the next timeline , stop streaming . * * Note : We might already have sent WAL > sendTimeLineValidUpto . The * startup process will normally replay all WAL that has been received * from the primary , before promoting , but if the WAL streaming is * terminated at a WAL page boundary , the valid portion of the timeline * might end in the middle of a WAL record . We might ' ve already sent the * first half of that partial WAL record to the cascading standby , so that * sentPtr > sendTimeLineValidUpto . That ' s OK ; the cascading standby can ' t * replay the partial WAL record either , so it can still follow our * timeline switch . if (sendTimeLineIsHistoric && sendTimeLineValidUpto <= sentPtr)/* close the current file. */ if (xlogreader->seg.ws_file >= 0 )/* Send CopyDone */ 'c' , NULL, 0 );true ;true ;"walsender reached end of timeline at %X/%X (sent up to %X/%X)" ,return ;/* Do we have any work to do? */ if (SendRqstPtr <= sentPtr)true ;return ;/* * Figure out how much to send in one message . If there ' s no more than * MAX_SEND_SIZE bytes to send , send everything . Otherwise send * MAX_SEND_SIZE bytes , but round back to logfile or page boundary . * * The rounding is not only for performance reasons . Walreceiver relies on * the fact that we never split a WAL record across two messages . Since a * long WAL record is split at page boundary into continuation records , * page boundary is always a safe cut - off point . We also assume that * SendRqstPtr never points to the middle of a WAL record . /* if we went beyond SendRqstPtr, back off */ if (SendRqstPtr <= endptr)if (sendTimeLineIsHistoric)false ;else true ;else /* round down to page boundary. */ false ;/* * OK to read and send the slice . 'w' );/* dataStart */ /* walEnd */ 0 ); /* sendtime, filled in last */ /* * Read the log directly into the output buffer to avoid extra memcpy * calls . /* attempt to read WAL from WAL buffers first */ /* now read the remaining WAL from WAL file */ if (nbytes > 0 &&/* Pass the current TLI because * only WalSndSegmentOpen controls /* See logical_read_xlog_page(). */ /* * During recovery , the currently - open WAL file might be replaced with the * file of the same name retrieved from archive . So we always need to * check what we read was valid after reading into the buffer . If it ' s * invalid , we try to open and read the file again . if (am_cascading_walsender)bool reload;false ;if (reload && xlogreader->seg.ws_file >= 0 )goto retry;'\0' ;/* * Fill the send timestamp last , so that it is taken as late as possible . 1 + sizeof (int64) + sizeof (int64)],sizeof (int64));'d' , output_message.data, output_message.len);/* Update shared memory status */ /* Report progress of XLOG streaming in PS display */ if (update_process_title)char activitymsg[50 ];sizeof (activitymsg), "streaming %X/%X" ,/* * Stream out logically decoded data . static void void )char *errm;/* * We ' ll use the current flush point to determine whether we ' ve caught up . * This variable is static in order to cache it across calls . Caching is * helpful because GetFlushRecPtr ( ) needs to acquire a heavily - contended * spinlock . static XLogRecPtr flushPtr = InvalidXLogRecPtr;/* * Don ' t know whether we ' ve caught up yet . We ' ll set WalSndCaughtUp to * true in WalSndWaitForWal , if we ' re actually waiting . We also set to * true if XLogReadRecord ( ) had to stop reading but WalSndWaitForWal * didn ' t wait - i . e . when we ' re shutting down . false ;/* xlog record was invalid */ if (errm != NULL)"could not find record while sending logically-decoded data: %s" ,if (record != NULL)/* * Note the lack of any call to LagTrackerWrite ( ) which is handled by * WalSndUpdateProgress which is called by output plugin through * logical decoding write api . /* * If first time through in this session , initialize flushPtr . Otherwise , * we only need to update flushPtr if EndRecPtr is past it . if (flushPtr == InvalidXLogRecPtr ||/* * For cascading logical WAL senders , we use the replay LSN instead of * the flush LSN , since logical decoding on a standby only processes * WAL that has been replayed . This distinction becomes particularly * important during shutdown , as new WAL is no longer replayed and the * last replayed LSN marks the furthest point up to which decoding can * proceed . if (am_cascading_walsender)else /* If EndRecPtr is still past our flushPtr, it means we caught up. */ if (logical_decoding_ctx->reader->EndRecPtr >= flushPtr)true ;/* * If we ' re caught up and have been requested to stop , have WalSndLoop ( ) * terminate the connection in an orderly manner , after writing out all * the pending data . if (WalSndCaughtUp && got_STOPPING)true ;/* Update shared memory status */ /* * Shutdown if the sender is caught up . * * NB : This should only be called when the shutdown signal has been received * from postmaster . * * Note that if we determine that there ' s still more data to send , this * function will return control to the caller . static void /* ... let's just be real sure we're caught up ... */ /* * To figure out whether all WAL has successfully been replicated , check * flush location if valid , write otherwise . Tools like pg_receivewal will * usually ( unless in synchronous mode ) return an invalid flush location . if (WalSndCaughtUp && sentPtr == replicatedPtr &&/* Inform the standby that XLOG streaming is done */ 0 );false );0 );if (!waiting_for_ping_response)true , InvalidXLogRecPtr);/* * Returns the latest point in WAL that has been safely flushed to disk . * This should only be called when in recovery . * * This is called either by cascading walsender to find WAL position to be sent * to a cascaded standby or by slot synchronization operation to validate remote * slot ' s lsn before syncing it locally . * * As a side - effect , * tli is updated to the TLI of the last * replayed WAL record . /* * We can safely send what ' s already been replayed . Also , if walreceiver * is streaming WAL from the same timeline , we can send anything that it * has streamed , but hasn ' t been replayed yet . if (tli)if (receiveTLI == replayTLI && receivePtr > replayPtr)return result;/* * Request walsenders to reload the currently - open WAL file void void )int i;for (i = 0 ; i < max_wal_senders; i++)if (walsnd->pid == 0 )continue ;true ;/* * Handle PROCSIG_WALSND_INIT_STOPPING signal . void void )/* * If replication has not yet started , die like with SIGTERM . If * replication is active , only set a flag and wake up the main loop . It * will send any outstanding WAL , wait for it to be replicated to the * standby , and then exit gracefully . if (!replication_active)else true ;/* * SIGUSR2 : set flag to do a last cycle and shut down afterwards . The WAL * sender should already have been switched to WALSNDSTATE_STOPPING at * this point . static void true ;/* Set up signal handlers */ void void )/* Set up signal handlers */ /* query cancel */ /* request shutdown */ /* SIGQUIT handler was already set up by InitPostmasterChild */ /* establishes SIGALRM handler */ /* request a last cycle and /* Reset some signals that are accepted by postmaster but not here */ /* Report shared-memory space needed by WalSndShmemInit */ void )0 ;sizeof (WalSnd)));return size;/* Allocate and initialize walsender-related shared memory */ void void )bool found;int i;"Wal Sender Ctl" , WalSndShmemSize(), &found);if (!found)/* First time through, so initialize */ 0 , WalSndShmemSize());for (i = 0 ; i < NUM_SYNC_REP_WAIT_MODE; i++)for (i = 0 ; i < max_wal_senders; i++)/* * Wake up physical , logical or both kinds of walsenders * * The distinction between physical and logical walsenders is done , because : * - physical walsenders can ' t send data until it ' s been flushed * - logical walsenders on standby can ' t decode and send data until it ' s been * applied * * For cascading replication we need to wake up physical walsenders separately * from logical walsenders ( see the comment before calling WalSndWakeup ( ) in * ApplyWalRecord ( ) for more details ) . * * This will be called inside critical sections , so throwing an error is not * advisable . void bool physical, bool logical)/* * Wake up all the walsenders waiting on WAL being flushed or replayed * respectively . Note that waiting walsender would have prepared to sleep * on the CV ( i . e . , added itself to the CV ' s waitlist ) in WalSndWait ( ) * before actually waiting . if (physical)if (logical)/* * Wait for readiness on the FeBe socket , or a timeout . The mask should be * composed of optional WL_SOCKET_WRITEABLE and WL_SOCKET_READABLE flags . Exit * on postmaster death . static void long timeout, uint32 wait_event)/* * We use a condition variable to efficiently wake up walsenders in * WalSndWakeup ( ) . * * Every walsender prepares to sleep on a shared memory CV . Note that it * just prepares to sleep on the CV ( i . e . , adds itself to the CV ' s * waitlist ) , but does not actually wait on the CV ( IOW , it never calls * ConditionVariableSleep ( ) ) . It still uses WaitEventSetWait ( ) for * waiting , because we also need to wait for socket events . The processes * ( startup process , walreceiver etc . ) wanting to wake up walsenders use * ConditionVariableBroadcast ( ) , which in turn calls SetLatch ( ) , helping * walsenders come out of WaitEventSetWait ( ) . * * This approach is simple and efficient because , one doesn ' t have to loop * through all the walsenders slots , with a spinlock acquisition and * release for every iteration , just to wake up only the waiting * walsenders . It makes WalSndWakeup ( ) callers ' life easy . * * XXX : A desirable future improvement would be to add support for CVs * into WaitEventSetWait ( ) . * * And , we use separate shared memory CVs for physical and logical * walsenders for selective wake ups , see WalSndWakeup ( ) for more details . * * If the wait event is WAIT_FOR_STANDBY_CONFIRMATION , wait on another CV * until awakened by physical walsenders after the walreceiver confirms * the receipt of the LSN . if (wait_event == WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION)else if (MyWalSnd->kind == REPLICATION_KIND_PHYSICAL)else if (MyWalSnd->kind == REPLICATION_KIND_LOGICAL)if (WaitEventSetWait(FeBeWaitSet, timeout, &event, 1 , wait_event) == 1 &&1 );/* * Signal all walsenders to move to stopping state . * * This will trigger walsenders to move to a state where no further WAL can be * generated . See this file ' s header for details . void void )int i;for (i = 0 ; i < max_wal_senders; i++)if (pid == 0 )continue ;/* * Wait that all the WAL senders have quit or reached the stopping state . This * is used by the checkpointer to control when the shutdown checkpoint can * safely be performed . void void )for (;;)int i;bool all_stopped = true ;for (i = 0 ; i < max_wal_senders; i++)if (walsnd->pid == 0 )continue ;if (walsnd->state != WALSNDSTATE_STOPPING)false ;break ;/* safe to leave if confirmation is done for all WAL senders */ if (all_stopped)return ;10000 L); /* wait for 10 msec */ /* Set state for current walsender (only called in walsender) */ void if (walsnd->state == state)return ;/* * Return a string constant representing the state . This is used * in system views , and should * not * be translated . static const char *switch (state)case WALSNDSTATE_STARTUP:return "startup" ;case WALSNDSTATE_BACKUP:return "backup" ;case WALSNDSTATE_CATCHUP:return "catchup" ;case WALSNDSTATE_STREAMING:return "streaming" ;case WALSNDSTATE_STOPPING:return "stopping" ;return "UNKNOWN" ;static Interval *sizeof (Interval));0 ;0 ;return result;/* * Returns activity of walsenders , including pids and xlog locations sent to * standby servers . #define PG_STAT_GET_WAL_SENDERS_COLS 12 int num_standbys;int i;0 );/* * Get the currently active synchronous standbys . This could be out of * date before we ' re done , but we ' ll use the data anyway . for (i = 0 ; i < max_wal_senders; i++)int priority;int pid;bool is_sync_standby;bool nulls[PG_STAT_GET_WAL_SENDERS_COLS] = {0 };int j;/* Collect data from shared memory */ if (walsnd->pid == 0 )continue ;/* * Detect whether walsender is / was considered synchronous . We can * provide some protection against stale data by checking the PID * along with walsnd_index . false ;for (j = 0 ; j < num_standbys; j++)if (sync_standbys[j].walsnd_index == i &&true ;break ;0 ] = Int32GetDatum(pid);if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))/* * Only superusers and roles with privileges of pg_read_all_stats * can see details . Other users only get the pid value to know * it ' s a walsender , but no details . 1 ], true , PG_STAT_GET_WAL_SENDERS_COLS - 1 );else 1 ] = CStringGetTextDatum(WalSndGetStateString(state));if (XLogRecPtrIsInvalid(sent_ptr))2 ] = true ;2 ] = LSNGetDatum(sent_ptr);if (XLogRecPtrIsInvalid(write))3 ] = true ;3 ] = LSNGetDatum(write);if (XLogRecPtrIsInvalid(flush))4 ] = true ;4 ] = LSNGetDatum(flush);if (XLogRecPtrIsInvalid(apply))5 ] = true ;5 ] = LSNGetDatum(apply);/* * Treat a standby such as a pg_basebackup background process * which always returns an invalid flush location , as an * asynchronous standby . 0 : priority;if (writeLag < 0 )6 ] = true ;else 6 ] = IntervalPGetDatum(offset_to_interval(writeLag));if (flushLag < 0 )7 ] = true ;else 7 ] = IntervalPGetDatum(offset_to_interval(flushLag));if (applyLag < 0 )8 ] = true ;else 8 ] = IntervalPGetDatum(offset_to_interval(applyLag));9 ] = Int32GetDatum(priority);/* * More easily understood version of standby state . This is purely * informational . * * In quorum - based sync replication , the role of each standby * listed in synchronous_standby_names can be changing very * frequently . Any standbys considered as " sync " at one moment can * be switched to " potential " ones at the next moment . So , it ' s * basically useless to report " sync " or " potential " as their sync * states . We report just " quorum " for them . if (priority == 0 )10 ] = CStringGetTextDatum("async" );else if (is_sync_standby)10 ] = SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY ?"sync" ) : CStringGetTextDatum("quorum" );else 10 ] = CStringGetTextDatum("potential" );if (replyTime == 0 )11 ] = true ;else 11 ] = TimestampTzGetDatum(replyTime);return (Datum) 0 ;/* * Send a keepalive message to standby . * * If requestReply is set , the message requests the other party to send * a message back to us , for heartbeat purposes . We also set a flag to * let nearby code know that we ' re waiting for that response , to avoid * repeated requests . * * writePtr is the location up to which the WAL is sent . It is essentially * the same as sentPtr but in some cases , we need to send keep alive before * sentPtr is updated like when skipping empty transactions . static void bool requestReply, XLogRecPtr writePtr)"sending replication keepalive" );/* construct the message... */ 'k' );1 : 0 );/* ... and send it wrapped in CopyData */ 'd' , output_message.data, output_message.len);/* Set local flag */ if (requestReply)true ;/* * Send keepalive message if too much time has elapsed . static void void )/* * Don ' t send keepalive messages if timeouts are globally disabled or * we ' re doing something not partaking in timeouts . if (wal_sender_timeout <= 0 || last_reply_timestamp <= 0 )return ;if (waiting_for_ping_response)return ;/* * If half of wal_sender_timeout has lapsed without receiving any reply * from the standby , send a keep - alive message to the standby requesting * an immediate reply . 2 );if (last_processing >= ping_time)true , InvalidXLogRecPtr);/* Try to flush pending output to the client */ if (pq_flush_if_writable() != 0 )/* * Record the end of the WAL and the time it was flushed locally , so that * LagTrackerRead can compute the elapsed time ( lag ) when this WAL location is * eventually reported to have been written , flushed and applied by the * standby in a reply message . static void int new_write_head;int i;if (!am_walsender)return ;/* * If the lsn hasn ' t advanced since last time , then do nothing . This way * we only record a new sample when new WAL has been written . if (lag_tracker->last_lsn == lsn)return ;/* * If advancing the write head of the circular buffer would crash into any * of the read heads , then the buffer is full . In other words , the * slowest reader ( presumably apply ) is the one that controls the release * of space . 1 ) % LAG_TRACKER_BUFFER_SIZE;for (i = 0 ; i < NUM_SYNC_REP_WAIT_MODE; ++i)/* * If the buffer is full , move the slowest reader to a separate * overflow entry and free its space in the buffer so the write head * can advance . if (new_write_head == lag_tracker->read_heads[i])1 ;/* Store a sample at the current write head position. */ /* * Find out how much time has elapsed between the moment WAL location ' lsn ' * ( or the highest known earlier LSN ) was flushed locally and the time ' now ' . * We have a separate read head for each of the reported LSN locations we * receive in replies from standby ; ' head ' controls which read head is * used . Whenever a read head crosses an LSN which was written into the * lag buffer with LagTrackerWrite , we can use the associated timestamp to * find out the time this LSN ( or an earlier one ) was flushed locally , and * therefore compute the lag . * * Return - 1 if no new sample data is available , and otherwise the elapsed * time in microseconds . static TimeOffsetint head, XLogRecPtr lsn, TimestampTz now)0 ;/* * If ' lsn ' has not passed the WAL position stored in the overflow entry , * return the elapsed time ( in microseconds ) since the saved local flush * time . If the flush time is in the future ( due to clock drift ) , return * - 1 to treat as no valid sample . * * Otherwise , switch back to using the buffer to control the read head and * compute the elapsed time . The read head is then reset to point to the * oldest entry in the buffer . if (lag_tracker->read_heads[head] == -1 )if (lag_tracker->overflowed[head].lsn > lsn)return (now >= lag_tracker->overflowed[head].time) ?1 ;1 ) % LAG_TRACKER_BUFFER_SIZE;/* Read all unread samples up to this LSN or end of buffer. */ while (lag_tracker->read_heads[head] != lag_tracker->write_head &&1 ) % LAG_TRACKER_BUFFER_SIZE;/* * If the lag tracker is empty , that means the standby has processed * everything we ' ve ever sent so we should now clear ' last_read ' . If we * didn ' t do that , we ' d risk using a stale and irrelevant sample for * interpolation at the beginning of the next burst of WAL after a period * of idleness . if (lag_tracker->read_heads[head] == lag_tracker->write_head)0 ;if (time > now)/* If the clock somehow went backwards, treat as not found. */ return -1 ;else if (time == 0 )/* * We didn ' t cross a time . If there is a future sample that we * haven ' t reached yet , and we ' ve already reached at least one sample , * let ' s interpolate the local flushed time . This is mainly useful * for reporting a completely stuck apply position as having * increasing lag , since otherwise we ' d have to wait for it to * eventually start moving again and cross one of our samples before * we can show the lag increasing . if (lag_tracker->read_heads[head] == lag_tracker->write_head)/* There are no future samples, so we can't interpolate. */ return -1 ;else if (lag_tracker->last_read[head].time != 0 )/* We can interpolate between last_read and the next sample. */ double fraction;if (lsn < prev.lsn)/* * Reported LSNs shouldn ' t normally go backwards , but it ' s * possible when there is a timeline change . Treat as not * found . return -1 ;if (prev.time > next.time)/* If the clock somehow went backwards, treat as not found. */ return -1 ;/* See how far we are between the previous and next samples. */ double ) (lsn - prev.lsn) / (double ) (next.lsn - prev.lsn);/* Scale the local flush time proportionally. */ double ) prev.time + (next.time - prev.time) * fraction);else /* * We have only a future sample , implying that we were entirely * caught up but and now there is a new burst of WAL and the * standby hasn ' t processed the first sample yet . Until the * standby reaches the future sample the best we can do is report * the hypothetical lag if that sample were to be replayed now . /* Return the elapsed time since local flush time in microseconds. */ 0 );return now - time;
Messung V0.5 in Prozent C=94 H=81 G=87
¤ Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.0.160Bemerkung:
(vorverarbeitet am 2026-08-08)
¤ *Bot Zugriff
2026-08-25
Neuigkeiten Aktuelles Motto des Tages letze Version des Elbe Quellennavigators
Jenseits des Üblichen ....
Besucher
Statistik #Sources=141584 #Domains=738142
