database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-09-03 15:22:11 +03:00
Code Activity

Track statistics for spilling of changes from ReorderBuffer.

Browse Source 
This adds the statistics about transactions spilled to disk from
ReorderBuffer.  Users can query the pg_stat_replication view to check
these stats.

Author: Tomas Vondra, with bug-fixes and minor changes by Dilip Kumar
Reviewed-by: Amit Kapila
Discussion: https://postgr.es/m/688b0b7f-2f6c-d827-c27b-216a8e3ea700@2ndquadrant.com
This commit is contained in:
Amit Kapila
2019-11-16 18:24:00 +05:30
parent 168d206400
commit 9290ad198b
9 changed files with 101 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/backend/replication/logical/reorderbuffer.c
View File

@@ -308,6 +308,10 @@ ReorderBufferAllocate(void)
buffer->outbufsize = 0;
buffer->size = 0;
buffer->spillCount = 0;
buffer->spillTxns = 0;
buffer->spillBytes = 0;
buffer->current_restart_decoding_lsn = InvalidXLogRecPtr;
dlist_init(&buffer->toplevel_by_lsn);
@@ -2415,6 +2419,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);
@@ -2473,6 +2478,13 @@ ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
spilled++;
}
/* update the statistics */
rb->spillCount += 1;
rb->spillBytes += size;
/* Don't consider already serialized transaction. */
rb->spillTxns += txn->serialized ? 0 : 1;
Assert(spilled == txn->nentries_mem);
Assert(dlist_is_empty(&txn->changes));
txn->nentries_mem = 0;

42
src/backend/replication/walsender.c
View File

@@ -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);
@@ -1261,7 +1262,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 statistics.
*/
static void
WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid)
@@ -1280,6 +1282,11 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
LagTrackerWrite(lsn, now);
sendTime = now;
/*
* Update statistics about transactions that spilled to disk.
*/
UpdateSpillStats(ctx);
}
/*
@@ -2318,6 +2325,9 @@ InitWalSenderSlot(void)
walsnd->state = WALSNDSTATE_STARTUP;
walsnd->latch = &MyProc->procLatch;
walsnd->replyTime = 0;
walsnd->spillTxns = 0;
walsnd->spillCount = 0;
walsnd->spillBytes = 0;
SpinLockRelease(&walsnd->mutex);
/* don't need the lock anymore */
MyWalSnd = (WalSnd *) walsnd;
@@ -3219,7 +3229,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 15
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
TupleDesc tupdesc;
Tuplestorestate *tupstore;
@@ -3274,6 +3284,9 @@ pg_stat_get_wal_senders(PG_FUNCTION_ARGS)
int pid;
WalSndState state;
TimestampTz replyTime;
int64 spillTxns;
int64 spillCount;
int64 spillBytes;
Datum values[PG_STAT_GET_WAL_SENDERS_COLS];
bool nulls[PG_STAT_GET_WAL_SENDERS_COLS];
@@ -3294,6 +3307,9 @@ 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;
SpinLockRelease(&walsnd->mutex);
memset(nulls, 0, sizeof(nulls));
@@ -3375,6 +3391,11 @@ 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);
}
tuplestore_putvalues(tupstore, tupdesc, values, nulls);
@@ -3611,3 +3632,20 @@ 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;
elog(DEBUG2, "UpdateSpillStats: updating stats %p %ld %ld %ld",
rb, rb->spillTxns, rb->spillCount, rb->spillBytes);
SpinLockRelease(&MyWalSnd->mutex);
}