From 5404d4ad25d52190c7297642014e695277aee0a1 Mon Sep 17 00:00:00 2001 From: Tomas Vondra Date: Thu, 26 Sep 2019 19:01:30 +0200 Subject: [PATCH 10/13] Track statistics for streaming/spilling --- doc/src/sgml/monitoring.sgml | 47 +++++++++++++++++++ src/backend/catalog/system_views.sql | 8 +++- src/backend/replication/logical/reorderbuffer.c | 21 +++++++++ src/backend/replication/walsender.c | 62 ++++++++++++++++++++++++- src/include/catalog/pg_proc.dat | 6 +-- src/include/replication/reorderbuffer.h | 16 +++++++ src/include/replication/walsender_private.h | 10 ++++ src/test/regress/expected/rules.out | 10 +++- 8 files changed, 172 insertions(+), 8 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 828e908..8de4fbd 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -2121,6 +2121,53 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i with security-sensitive fields obfuscated. + + spill_txns + integer + Number of transactions spilled to disk after the memory used by + logical decoding exceeds logical_work_mem. The + counter gets incremented both for toplevel transactions and + subtransactions. + + + + spill_count + integer + Number of times transactions were spilled to disk. Transactions + may get spilled repeatedly, and this counter gets incremented on every + such invocation. + + + + spill_bytes + integer + Amount of decoded transaction data spilled to disk. + + + + stream_txns + integer + Number of in-progress transactions streamed to subscriber after + memory used by logical decoding exceeds logical_work_mem. + Streaming only works with toplevel transactions (subtransactions can't + be streamed independently), so the counter does not get incremented for + subtransactions. + + + + stream_count + integer + Number of times in-progress transactions were streamed to subscriber. + Transactions may get streamed repeatedly, and this counter gets incremented + on every such invocation. + + + + stream_bytes + integer + Amount of decoded in-progress transaction data streamed to subscriber. + + diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 9fe4a47..f8d0c4a 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -776,7 +776,13 @@ CREATE VIEW pg_stat_replication AS W.replay_lag, W.sync_priority, W.sync_state, - W.reply_time + W.reply_time, + W.spill_txns, + W.spill_count, + W.spill_bytes, + W.stream_txns, + W.stream_count, + W.stream_bytes FROM pg_stat_get_activity(NULL) AS S JOIN pg_stat_get_wal_senders() AS W ON (S.pid = W.pid) LEFT JOIN pg_authid AS U ON (S.usesysid = U.oid); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 963cbf3..5e91522 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -354,6 +354,14 @@ ReorderBufferAllocate(void) buffer->outbufsize = 0; buffer->size = 0; + buffer->spillCount = 0; + buffer->spillTxns = 0; + buffer->spillBytes = 0; + + buffer->streamCount = 0; + buffer->streamTxns = 0; + buffer->streamBytes = 0; + buffer->current_restart_decoding_lsn = InvalidXLogRecPtr; dlist_init(&buffer->toplevel_by_lsn); @@ -3001,6 +3009,7 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) int fd = -1; XLogSegNo curOpenSegNo = 0; Size spilled = 0; + Size size = txn->size; elog(DEBUG2, "spill %u changes in XID %u to disk", (uint32) txn->nentries_mem, txn->xid); @@ -3059,6 +3068,11 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) spilled++; } + /* update the statistics */ + rb->spillCount += 1; + rb->spillTxns += (rbtxn_is_serialized(txn)) ? 1 : 0; + rb->spillBytes += size; + Assert(spilled == txn->nentries_mem); Assert(dlist_is_empty(&txn->changes)); txn->nentries_mem = 0; @@ -3763,6 +3777,13 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) PG_END_TRY(); /* + * Update the stream statistics. + */ + rb->streamCount += 1; + rb->streamTxns += (rbtxn_is_streamed(txn)) ? 1 : 0; + rb->streamBytes += txn->size; + + /* * Discard the changes that we just streamed, and mark the transactions * as streamed (if they contained changes). */ diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 1c0c020..14abd85 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -248,6 +248,7 @@ 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 UpdateSpillStats(LogicalDecodingContext *ctx); static void XLogRead(WALSegmentContext *segcxt, char *buf, XLogRecPtr startptr, Size count); @@ -1267,7 +1268,8 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, /* * LogicalDecodingContext 'update_progress' callback. * - * Write the current position to the lag tracker (see XLogSendPhysical). + * Write the current position to the lag tracker (see XLogSendPhysical), + * and update the spill/stream statistics. */ static void WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid) @@ -1286,6 +1288,12 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId LagTrackerWrite(lsn, now); sendTime = now; + + /* + * Update statistics about transactions that spilled to disk or + * streamed to subscriber (before being committed). + */ + UpdateSpillStats(ctx); } /* @@ -2325,6 +2333,12 @@ InitWalSenderSlot(void) walsnd->state = WALSNDSTATE_STARTUP; walsnd->latch = &MyProc->procLatch; walsnd->replyTime = 0; + walsnd->spillTxns = 0; + walsnd->spillCount = 0; + walsnd->spillBytes = 0; + walsnd->streamTxns = 0; + walsnd->streamCount = 0; + walsnd->streamBytes = 0; SpinLockRelease(&walsnd->mutex); /* don't need the lock anymore */ MyWalSnd = (WalSnd *) walsnd; @@ -3236,7 +3250,7 @@ offset_to_interval(TimeOffset offset) Datum pg_stat_get_wal_senders(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_WAL_SENDERS_COLS 12 +#define PG_STAT_GET_WAL_SENDERS_COLS 18 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; TupleDesc tupdesc; Tuplestorestate *tupstore; @@ -3291,6 +3305,12 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) int pid; WalSndState state; TimestampTz replyTime; + int64 spillTxns; + int64 spillCount; + int64 spillBytes; + int64 streamTxns; + int64 streamCount; + int64 streamBytes; Datum values[PG_STAT_GET_WAL_SENDERS_COLS]; bool nulls[PG_STAT_GET_WAL_SENDERS_COLS]; @@ -3311,6 +3331,12 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) applyLag = walsnd->applyLag; priority = walsnd->sync_standby_priority; replyTime = walsnd->replyTime; + spillTxns = walsnd->spillTxns; + spillCount = walsnd->spillCount; + spillBytes = walsnd->spillBytes; + streamTxns = walsnd->streamTxns; + streamCount = walsnd->streamCount; + streamBytes = walsnd->streamBytes; SpinLockRelease(&walsnd->mutex); memset(nulls, 0, sizeof(nulls)); @@ -3392,6 +3418,16 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS) nulls[11] = true; else values[11] = TimestampTzGetDatum(replyTime); + + /* spill to disk */ + values[12] = Int64GetDatum(spillTxns); + values[13] = Int64GetDatum(spillCount); + values[14] = Int64GetDatum(spillBytes); + + /* stream over-sized transactions */ + values[15] = Int64GetDatum(streamTxns); + values[16] = Int64GetDatum(streamCount); + values[17] = Int64GetDatum(streamBytes); } tuplestore_putvalues(tupstore, tupdesc, values, nulls); @@ -3628,3 +3664,25 @@ LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now) Assert(time != 0); return now - time; } + +static void +UpdateSpillStats(LogicalDecodingContext *ctx) +{ + ReorderBuffer *rb = ctx->reorder; + + SpinLockAcquire(&MyWalSnd->mutex); + + MyWalSnd->spillTxns = rb->spillTxns; + MyWalSnd->spillCount = rb->spillCount; + MyWalSnd->spillBytes = rb->spillBytes; + + MyWalSnd->streamTxns = rb->streamTxns; + MyWalSnd->streamCount = rb->streamCount; + MyWalSnd->streamBytes = rb->streamBytes; + + elog(WARNING, "UpdateSpillStats: updating stats %p %ld %ld %ld %ld %ld %ld", + rb, rb->spillTxns, rb->spillCount, rb->spillBytes, + rb->streamTxns, rb->streamCount, rb->streamBytes); + + SpinLockRelease(&MyWalSnd->mutex); +} diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 58ea5b9..9a508bf 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5166,9 +5166,9 @@ proname => 'pg_stat_get_wal_senders', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', proparallel => 'r', prorettype => 'record', proargtypes => '', - proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time}', + proallargtypes => '{int4,text,pg_lsn,pg_lsn,pg_lsn,pg_lsn,interval,interval,interval,int4,text,timestamptz,int8,int8,int8,int8,int8,int8}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{pid,state,sent_lsn,write_lsn,flush_lsn,replay_lsn,write_lag,flush_lag,replay_lag,sync_priority,sync_state,reply_time,spill_txns,spill_count,spill_bytes,stream_txns,stream_count,stream_bytes}', prosrc => 'pg_stat_get_wal_senders' }, { oid => '3317', descr => 'statistics: information about WAL receiver', proname => 'pg_stat_get_wal_receiver', proisstrict => 'f', provolatile => 's', diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index f39b34f..53d9440 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -509,6 +509,22 @@ struct ReorderBuffer /* memory accounting */ Size size; + + /* + * Statistics about transactions streamed or spilled to disk. + * + * A single transaction may be streamed/spilled repeatedly, which is + * why we keep two different counters. For spilling, the transaction + * counter includes both toplevel transactions and subtransactions. + * For streaming, it only includes toplevel transactions (we never + * stream individual subtransactions). + */ + int64 spillCount; /* spill-to-disk invocation counter */ + int64 spillTxns; /* number of transactions spilled to disk */ + int64 spillBytes; /* amount of data spilled to disk */ + int64 streamCount; /* streaming invocation counter */ + int64 streamTxns; /* number of transactions spilled to disk */ + int64 streamBytes; /* amount of data streamed to subscriber */ }; diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index 0dd6d1c..f726f25 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -80,6 +80,16 @@ typedef struct WalSnd * Timestamp of the last message received from standby. */ TimestampTz replyTime; + + /* Statistics for transactions spilled to disk. */ + int64 spillTxns; + int64 spillCount; + int64 spillBytes; + + /* Statistics for in-progress transactions streamed to subscriber. */ + int64 streamTxns; + int64 streamCount; + int64 streamBytes; } WalSnd; extern WalSnd *MyWalSnd; diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 210e9cd..836abf0 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1951,9 +1951,15 @@ pg_stat_replication| SELECT s.pid, w.replay_lag, w.sync_priority, w.sync_state, - w.reply_time + w.reply_time, + w.spill_txns, + w.spill_count, + w.spill_bytes, + w.stream_txns, + w.stream_count, + w.stream_bytes FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc) - JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid))) + JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time, spill_txns, spill_count, spill_bytes, stream_txns, stream_count, stream_bytes) ON ((s.pid = w.pid))) LEFT JOIN pg_authid u ON ((s.usesysid = u.oid))); pg_stat_ssl| SELECT s.pid, s.ssl, -- 1.8.3.1