+
+ wal_receiver_status_interval (integer)
+
+
wal_receiver_status_interval configuration parameter+
+
+ Specifies the minimum frequency, in seconds, for the WAL receiver
+ process on the standby to send information about replication progress
+ to the primary, where they can be seen using the
+ pg_stat_replication view. The standby will report
+ the last transaction log position it has written, the last position it
+ has flushed to disk, and the last position it has applied. Updates are
+ sent each time the write or flush positions changed, or at least as
+ often as specified by this parameter. Thus, the apply position may
+ lag slightly behind the true position. Setting this parameter to zero
+ disables status updates completely. This parameter can only be set in
+ the postgresql.conf file or on the server command line.
+ The default value is 10 seconds.
+
+
+
+
vacuum_defer_cleanup_age (integer)
pg_stat_replicationpg_stat_replication
One row per WAL sender process, showing process ID,
user OID, user name, application name, client's address and port number,
- time at which the server process began execution, current WAL sender
- state and transaction log location. The columns detailing what exactly
+ time at which the server process began execution, and the current WAL
+ sender state and transaction log location. In addition, the standby
+ reports the last transaction log position it received and wrote, the last
+ position it flushed to disk, and the last position it replayed, and this
+ information is also displayed here. The columns detailing what exactly
the connection is doing are only visible if the user examining the view
is a superuser.
shutdown), it will send a CommandComplete message before exiting.
This might not happen during an abnormal shutdown, of course.
+
+ The receiving process can send a status update back to the sender at
+ any time, using the following message format (also in the payload of
+ a CopyData message):
+
+
+
+
+
+ Standby status update (F)
+
+
+
+
+
+ Byte1('r')
+
+
+ Identifies the message as a receiver status update.
+
+
+
+
+
+ Byte8
+
+
+ The location of the last WAL byte + 1 received and written to disk
+ in the standby, in XLogRecPtr format.
+
+
+
+
+
+ Byte8
+
+
+ The location of the last WAL byte + 1 flushed to disk in
+ the standby, in XLogRecPtr format.
+
+
+
+
+
+ Byte8
+
+
+ The location of the last WAL byte + 1 applied in the standby, in
+ XLogRecPtr format.
+
+
+
+
+
+ Byte8
+
+
+ The server's system clock at the time of transmission,
+ given in TimestampTz format.
+
+
+
+
+
+
+
+
+
PG_RETURN_TEXT_P(cstring_to_text(location));
}
+/*
+ * Get latest redo apply position.
+ *
+ * Exported to allow WALReceiver to read the pointer directly.
+ */
+XLogRecPtr
+GetXLogReplayRecPtr(void)
+{
+ /* use volatile pointer to prevent code rearrangement */
+ volatile XLogCtlData *xlogctl = XLogCtl;
+ XLogRecPtr recptr;
+
+ SpinLockAcquire(&xlogctl->info_lck);
+ recptr = xlogctl->recoveryLastRecPtr;
+ SpinLockRelease(&xlogctl->info_lck);
+
+ return recptr;
+}
+
/*
* Report the last WAL replay location (same format as pg_start_backup etc)
*
Datum
pg_last_xlog_replay_location(PG_FUNCTION_ARGS)
{
- /* use volatile pointer to prevent code rearrangement */
- volatile XLogCtlData *xlogctl = XLogCtl;
XLogRecPtr recptr;
char location[MAXFNAMELEN];
- SpinLockAcquire(&xlogctl->info_lck);
- recptr = xlogctl->recoveryLastRecPtr;
- SpinLockRelease(&xlogctl->info_lck);
+ recptr = GetXLogReplayRecPtr();
if (recptr.xlogid == 0 && recptr.xrecoff == 0)
PG_RETURN_NULL();
S.client_port,
S.backend_start,
W.state,
- W.sent_location
+ W.sent_location,
+ W.write_location,
+ W.flush_location,
+ W.apply_location
FROM pg_stat_get_activity(NULL) AS S, pg_authid U,
pg_stat_get_wal_senders() AS W
WHERE S.usesysid = U.oid AND
/* Global variable to indicate if this process is a walreceiver process */
bool am_walreceiver;
+/* GUC variable */
+int wal_receiver_status_interval;
+
/* libpqreceiver hooks to these when loaded */
walrcv_connect_type walrcv_connect = NULL;
walrcv_receive_type walrcv_receive = NULL;
XLogRecPtr Flush; /* last byte + 1 flushed in the standby */
} LogstreamResult;
+static StandbyReplyMessage reply_message;
+
/*
* About SIGTERM handling:
*
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len);
static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr);
static void XLogWalRcvFlush(void);
+static void XLogWalRcvSendReply(void);
/* Signal handlers */
static void WalRcvSigHupHandler(SIGNAL_ARGS);
while (walrcv_receive(0, &type, &buf, &len))
XLogWalRcvProcessMsg(type, buf, len);
+ /* Let the master know that we received some data. */
+ XLogWalRcvSendReply();
+
/*
* If we've written some records, flush them to disk and let the
* startup process know about them.
*/
XLogWalRcvFlush();
}
+ else
+ {
+ /*
+ * We didn't receive anything new, but send a status update to
+ * the master anyway, to report any progress in applying WAL.
+ */
+ XLogWalRcvSendReply();
+ }
}
}
LogstreamResult.Write.xrecoff);
set_ps_display(activitymsg, false);
}
+
+ /* Also let the master know that we made some progress */
+ XLogWalRcvSendReply();
}
}
+
+/*
+ * Send reply message to primary, indicating our current XLOG positions and
+ * the current time.
+ */
+static void
+XLogWalRcvSendReply(void)
+{
+ char buf[sizeof(StandbyReplyMessage) + 1];
+ TimestampTz now;
+
+ /*
+ * If the user doesn't want status to be reported to the master, be sure
+ * to exit before doing anything at all.
+ */
+ if (wal_receiver_status_interval <= 0)
+ return;
+
+ /* Get current timestamp. */
+ now = GetCurrentTimestamp();
+
+ /*
+ * We can compare the write and flush positions to the last message we
+ * sent without taking any lock, but the apply position requires a spin
+ * lock, so we don't check that unless something else has changed or 10
+ * seconds have passed. This means that the apply log position will
+ * appear, from the master's point of view, to lag slightly, but since
+ * this is only for reporting purposes and only on idle systems, that's
+ * probably OK.
+ */
+ if (XLByteEQ(reply_message.write, LogstreamResult.Write)
+ && XLByteEQ(reply_message.flush, LogstreamResult.Flush)
+ && !TimestampDifferenceExceeds(reply_message.sendTime, now,
+ wal_receiver_status_interval * 1000))
+ return;
+
+ /* Construct a new message. */
+ reply_message.write = LogstreamResult.Write;
+ reply_message.flush = LogstreamResult.Flush;
+ reply_message.apply = GetXLogReplayRecPtr();
+ reply_message.sendTime = now;
+
+ elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X",
+ reply_message.write.xlogid, reply_message.write.xrecoff,
+ reply_message.flush.xlogid, reply_message.flush.xrecoff,
+ reply_message.apply.xlogid, reply_message.apply.xrecoff);
+
+ /* Prepend with the message type and send it. */
+ buf[0] = 'r';
+ memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
+ walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
+}
#include "funcapi.h"
#include "access/xlog_internal.h"
+#include "access/transam.h"
#include "catalog/pg_type.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/pmsignal.h"
+#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/guc.h"
static void WalSndHandshake(void);
static void WalSndKill(int code, Datum arg);
static bool XLogSend(char *msgbuf, bool *caughtup);
-static void CheckClosedConnection(void);
static void IdentifySystem(void);
static void StartReplication(StartReplicationCmd * cmd);
+static void ProcessStandbyReplyMessage(void);
+static void ProcessRepliesIfAny(void);
/* Main entry point for walsender process */
* Check if the remote end has closed the connection.
*/
static void
-CheckClosedConnection(void)
+ProcessRepliesIfAny(void)
{
unsigned char firstchar;
int r;
/* Handle the very limited subset of commands expected in this phase */
switch (firstchar)
{
+ /*
+ * 'd' means a standby reply wrapped in a COPY BOTH packet.
+ */
+ case 'd':
+ ProcessStandbyReplyMessage();
+ break;
+
/*
* 'X' means that the standby is closing down the socket.
*/
}
}
+/*
+ * Process a status update message received from standby.
+ */
+static void
+ProcessStandbyReplyMessage(void)
+{
+ static StringInfoData input_message;
+ StandbyReplyMessage reply;
+ char msgtype;
+
+ initStringInfo(&input_message);
+
+ /*
+ * Read the message contents.
+ */
+ if (pq_getmessage(&input_message, 0))
+ {
+ ereport(COMMERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg("unexpected EOF on standby connection")));
+ proc_exit(0);
+ }
+
+ /*
+ * Check message type from the first byte. At the moment, there is only
+ * one type.
+ */
+ msgtype = pq_getmsgbyte(&input_message);
+ if (msgtype != 'r')
+ ereport(COMMERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg("unexpected message type %c", msgtype)));
+
+ pq_copymsgbytes(&input_message, (char *) &reply, sizeof(StandbyReplyMessage));
+
+ elog(DEBUG2, "write %X/%X flush %X/%X apply %X/%X ",
+ reply.write.xlogid, reply.write.xrecoff,
+ reply.flush.xlogid, reply.flush.xrecoff,
+ reply.apply.xlogid, reply.apply.xrecoff);
+
+ /*
+ * Update shared state for this WalSender process
+ * based on reply data from standby.
+ */
+ {
+ /* use volatile pointer to prevent code rearrangement */
+ volatile WalSnd *walsnd = MyWalSnd;
+
+ SpinLockAcquire(&walsnd->mutex);
+ walsnd->write = reply.write;
+ walsnd->flush = reply.flush;
+ walsnd->apply = reply.apply;
+ SpinLockRelease(&walsnd->mutex);
+ }
+}
+
/* Main loop of walsender process */
static int
WalSndLoop(void)
{
if (!XLogSend(output_message, &caughtup))
break;
+ ProcessRepliesIfAny();
if (caughtup)
walsender_shutdown_requested = true;
}
WaitLatchOrSocket(&MyWalSnd->latch, MyProcPort->sock,
WalSndDelay * 1000L);
}
-
- /* Check if the connection was closed */
- CheckClosedConnection();
}
else
{
/* Update our state to indicate if we're behind or not */
WalSndSetState(caughtup ? WALSNDSTATE_STREAMING : WALSNDSTATE_CATCHUP);
+ ProcessRepliesIfAny();
}
/*
Datum
pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_WAL_SENDERS_COLS 3
+#define PG_STAT_GET_WAL_SENDERS_COLS 6
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
TupleDesc tupdesc;
Tuplestorestate *tupstore;
{
/* use volatile pointer to prevent code rearrangement */
volatile WalSnd *walsnd = &WalSndCtl->walsnds[i];
- char sent_location[MAXFNAMELEN];
+ char location[MAXFNAMELEN];
XLogRecPtr sentPtr;
+ XLogRecPtr write;
+ XLogRecPtr flush;
+ XLogRecPtr apply;
WalSndState state;
Datum values[PG_STAT_GET_WAL_SENDERS_COLS];
bool nulls[PG_STAT_GET_WAL_SENDERS_COLS];
SpinLockAcquire(&walsnd->mutex);
sentPtr = walsnd->sentPtr;
state = walsnd->state;
+ write = walsnd->write;
+ flush = walsnd->flush;
+ apply = walsnd->apply;
SpinLockRelease(&walsnd->mutex);
- snprintf(sent_location, sizeof(sent_location), "%X/%X",
- sentPtr.xlogid, sentPtr.xrecoff);
-
memset(nulls, 0, sizeof(nulls));
values[0] = Int32GetDatum(walsnd->pid);
+
if (!superuser())
{
/*
*/
nulls[1] = true;
nulls[2] = true;
+ nulls[3] = true;
+ nulls[4] = true;
+ nulls[5] = true;
}
else
{
values[1] = CStringGetTextDatum(WalSndGetStateString(state));
- values[2] = CStringGetTextDatum(sent_location);
+
+ snprintf(location, sizeof(location), "%X/%X",
+ sentPtr.xlogid, sentPtr.xrecoff);
+ values[2] = CStringGetTextDatum(location);
+
+ if (write.xlogid == 0 && write.xrecoff == 0)
+ nulls[3] = true;
+ snprintf(location, sizeof(location), "%X/%X",
+ write.xlogid, write.xrecoff);
+ values[3] = CStringGetTextDatum(location);
+
+ if (flush.xlogid == 0 && flush.xrecoff == 0)
+ nulls[4] = true;
+ snprintf(location, sizeof(location), "%X/%X",
+ flush.xlogid, flush.xrecoff);
+ values[4] = CStringGetTextDatum(location);
+
+ if (apply.xlogid == 0 && apply.xrecoff == 0)
+ nulls[5] = true;
+ snprintf(location, sizeof(location), "%X/%X",
+ apply.xlogid, apply.xrecoff);
+ values[5] = CStringGetTextDatum(location);
}
tuplestore_putvalues(tupstore, tupdesc, values, nulls);
#include "postmaster/postmaster.h"
#include "postmaster/syslogger.h"
#include "postmaster/walwriter.h"
+#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "storage/bufmgr.h"
#include "storage/standby.h"
30 * 1000, -1, INT_MAX / 1000, NULL, NULL
},
+ {
+ {"wal_receiver_status_interval", PGC_SIGHUP, WAL_STANDBY_SERVERS,
+ gettext_noop("Sets the maximum interval between WAL receiver status reports to the master."),
+ NULL,
+ GUC_UNIT_S
+ },
+ &wal_receiver_status_interval,
+ 10, 0, INT_MAX/1000, NULL, NULL
+ },
+
{
{"max_connections", PGC_POSTMASTER, CONN_AUTH_SETTINGS,
gettext_noop("Sets the maximum number of concurrent connections."),
#max_standby_streaming_delay = 30s # max delay before canceling queries
# when reading streaming WAL;
# -1 allows indefinite delay
+#wal_receiver_status_interval = 10s # replies at least this often, 0 disables
#------------------------------------------------------------------------------
extern bool RecoveryInProgress(void);
extern bool XLogInsertAllowed(void);
extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
+extern XLogRecPtr GetXLogReplayRecPtr(void);
extern void UpdateControlFile(void);
extern uint64 GetSystemIdentifier(void);
DESCR("statistics: currently active backend IDs");
DATA(insert OID = 2022 ( pg_stat_get_activity PGNSP PGUID 12 1 100 0 f f f f t s 1 0 2249 "23" "{23,26,23,26,25,25,16,1184,1184,1184,869,23}" "{i,o,o,o,o,o,o,o,o,o,o,o}" "{pid,datid,procpid,usesysid,application_name,current_query,waiting,xact_start,query_start,backend_start,client_addr,client_port}" _null_ pg_stat_get_activity _null_ _null_ _null_ ));
DESCR("statistics: information about currently active backends");
-DATA(insert OID = 3099 ( pg_stat_get_wal_senders PGNSP PGUID 12 1 10 0 f f f f t s 0 0 2249 "" "{23,25,25}" "{o,o,o}" "{procpid,state,sent_location}" _null_ pg_stat_get_wal_senders _null_ _null_ _null_ ));
+DATA(insert OID = 3099 ( pg_stat_get_wal_senders PGNSP PGUID 12 1 10 0 f f f f t s 0 0 2249 "" "{23,25,25,25,25,25}" "{o,o,o,o,o,o}" "{procpid,state,sent_location,write_location,flush_location,apply_location}" _null_ pg_stat_get_wal_senders _null_ _null_ _null_ ));
DESCR("statistics: information about currently active replication");
DATA(insert OID = 2026 ( pg_backend_pid PGNSP PGUID 12 1 0 0 f f f t f s 0 0 23 "" _null_ _null_ _null_ _null_ pg_backend_pid _null_ _null_ _null_ ));
DESCR("statistics: current backend PID");
TimestampTz sendTime;
} WalDataMessageHeader;
+/*
+ * Reply message from standby (message type 'r'). This is wrapped within
+ * a CopyData message at the FE/BE protocol level.
+ *
+ * Note that the data length is not specified here.
+ */
+typedef struct
+{
+ /*
+ * The xlog locations that have been written, flushed, and applied
+ * by standby-side. These may be invalid if the standby-side is unable
+ * to or chooses not to report these.
+ */
+ XLogRecPtr write;
+ XLogRecPtr flush;
+ XLogRecPtr apply;
+
+ /* Sender's system clock at the time of transmission */
+ TimestampTz sendTime;
+} StandbyReplyMessage;
+
/*
* Maximum data payload in a WAL data message. Must be >= XLOG_BLCKSZ.
*
#include "pgtime.h"
extern bool am_walreceiver;
+extern int wal_receiver_status_interval;
/*
* MAXCONNINFO: maximum size of a connection string.
WalSndState state; /* this walsender's state */
XLogRecPtr sentPtr; /* WAL has been sent up to this point */
- slock_t mutex; /* locks shared variables shown above */
+ /*
+ * The xlog locations that have been written, flushed, and applied
+ * by standby-side. These may be invalid if the standby-side has not
+ * offered values yet.
+ */
+ XLogRecPtr write;
+ XLogRecPtr flush;
+ XLogRecPtr apply;
+
+ /* Protects shared variables shown above. */
+ slock_t mutex;
/*
* Latch used by backends to wake up this walsender when it has work
pg_stat_bgwriter | SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints_timed, pg_stat_get_bgwriter_requested_checkpoints() AS checkpoints_req, pg_stat_get_bgwriter_buf_written_checkpoints() AS buffers_checkpoint, pg_stat_get_bgwriter_buf_written_clean() AS buffers_clean, pg_stat_get_bgwriter_maxwritten_clean() AS maxwritten_clean, pg_stat_get_buf_written_backend() AS buffers_backend, pg_stat_get_buf_fsync_backend() AS buffers_backend_fsync, pg_stat_get_buf_alloc() AS buffers_alloc, pg_stat_get_bgwriter_stat_reset_time() AS stats_reset;
pg_stat_database | SELECT d.oid AS datid, d.datname, pg_stat_get_db_numbackends(d.oid) AS numbackends, pg_stat_get_db_xact_commit(d.oid) AS xact_commit, pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback, (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read, pg_stat_get_db_blocks_hit(d.oid) AS blks_hit, pg_stat_get_db_tuples_returned(d.oid) AS tup_returned, pg_stat_get_db_tuples_fetched(d.oid) AS tup_fetched, pg_stat_get_db_tuples_inserted(d.oid) AS tup_inserted, pg_stat_get_db_tuples_updated(d.oid) AS tup_updated, pg_stat_get_db_tuples_deleted(d.oid) AS tup_deleted, pg_stat_get_db_conflict_all(d.oid) AS conflicts, pg_stat_get_db_stat_reset_time(d.oid) AS stats_reset FROM pg_database d;
pg_stat_database_conflicts | SELECT d.oid AS datid, d.datname, pg_stat_get_db_conflict_tablespace(d.oid) AS confl_tablespace, pg_stat_get_db_conflict_lock(d.oid) AS confl_lock, pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot, pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin, pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock FROM pg_database d;
- pg_stat_replication | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_port, s.backend_start, w.state, w.sent_location FROM pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u, pg_stat_get_wal_senders() w(procpid, state, sent_location) WHERE ((s.usesysid = u.oid) AND (s.procpid = w.procpid));
+ pg_stat_replication | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_port, s.backend_start, w.state, w.sent_location, w.write_location, w.flush_location, w.apply_location FROM pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u, pg_stat_get_wal_senders() w(procpid, state, sent_location, write_location, flush_location, apply_location) WHERE ((s.usesysid = u.oid) AND (s.procpid = w.procpid));
pg_stat_sys_indexes | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_indexes.schemaname ~ '^pg_toast'::text));
pg_stat_sys_tables | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze, pg_stat_all_tables.vacuum_count, pg_stat_all_tables.autovacuum_count, pg_stat_all_tables.analyze_count, pg_stat_all_tables.autoanalyze_count FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_tables.schemaname ~ '^pg_toast'::text));
pg_stat_user_functions | SELECT p.oid AS funcid, n.nspname AS schemaname, p.proname AS funcname, pg_stat_get_function_calls(p.oid) AS calls, (pg_stat_get_function_time(p.oid) / 1000) AS total_time, (pg_stat_get_function_self_time(p.oid) / 1000) AS self_time FROM (pg_proc p LEFT JOIN pg_namespace n ON ((n.oid = p.pronamespace))) WHERE ((p.prolang <> (12)::oid) AND (pg_stat_get_function_calls(p.oid) IS NOT NULL));
projects / postgresql.git / commitdiff