v5-0002-rework.patch

From 4a14b9edbcf79075c050cd60fc28a80c53452bc1 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@2ndquadrant.com>
Date: Fri, 10 Nov 2023 17:37:41 +0100
Subject: [PATCH v5 2/2] rework

---
 src/backend/access/transam/xact.c   |   3 +
 src/backend/access/transam/xlog.c   | 338 +++++++++++++++++++++++++---
 src/backend/utils/misc/guc_tables.c |  28 ++-
 src/include/access/xlog.h           |   6 +-
 4 files changed, 345 insertions(+), 30 deletions(-)

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 536edb3792f..9a333e8533e 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -2139,6 +2139,9 @@ StartTransaction(void)
 	 */
 	s->state = TRANS_INPROGRESS;
 
+	/* reset limits for WAL throttling */
+	ResetXLogThrottling();
+
 	ShowTransactionState("StartTransaction");
 }
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 5dab46e17e1..b44e3c84d3d 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -140,7 +140,16 @@ int			wal_retrieve_retry_interval = 5000;
 int			max_slot_wal_keep_size_mb = -1;
 int			wal_decode_buffer_size = 512 * 1024;
 bool		track_wal_io_timing = false;
-int			wal_throttle_threshold = 0;
+
+/* user-settable parameters for WAL throttling */
+int			wal_write_after = 0;
+int			wal_flush_after_local = 0;
+int			wal_flush_after_remote = 0;
+
+/* auto-adjusted version of the throttling parameters */
+int			wal_write_after_current = 0;
+int			wal_flush_after_local_current = 0;
+int			wal_flush_after_remote_current = 0;
 
 #ifdef WAL_DEBUG
 bool		XLOG_DEBUG = false;
@@ -255,16 +264,10 @@ static int	LocalXLogInsertAllowed = -1;
  * parallel backends may have written WAL records at later LSNs than the value
  * stored here.  The parallel leader advances its own copy, when necessary,
  * in WaitForParallelWorkersToFinish.
- *
- * XactLastThrottledRecEnd points to the last XLOG record that should be throttled
- * as the additional WAL records could be generated before processing interrupts.
- *
- * XXX I'm not sure I understand what "record to be throttled" means?
  */
 XLogRecPtr	ProcLastRecPtr = InvalidXLogRecPtr;
 XLogRecPtr	XactLastRecEnd = InvalidXLogRecPtr;
 XLogRecPtr	XactLastCommitEnd = InvalidXLogRecPtr;
-static XLogRecPtr	XactLastThrottledRecEnd = InvalidXLogRecPtr;
 
 /*
  * RedoRecPtr is this backend's local copy of the REDO record pointer
@@ -663,7 +666,8 @@ static MemoryContext walDebugCxt = NULL;
  *
  * XXX Not sure it should refer to "backend", it's really about xact, no?
  */
-uint32	backendWalInserted = 0;
+uint32		backendWalInserted = 0;
+XLogRecPtr	backendLastFlushPtr = InvalidXLogRecPtr;
 
 static void CleanupAfterArchiveRecovery(TimeLineID EndOfLogTLI,
 										XLogRecPtr EndOfLog,
@@ -1094,17 +1098,48 @@ XLogInsertRecord(XLogRecData *rdata,
 		/*
 		 * Decide if we need to throttle this backend, so that it does not write
 		 * WAL too fast, causing lag against the sync standby (which in turn
-		 * increases latency for standby confirmations). We may be holding locks
-		 * and blocking interrupts here, so we only make the decision, but the
-		 * wait (for sync standby confirmation) happens elsewhere.
+		 * increases latency both for local commits and for standby confirmations).
+		 * We may be holding locks and blocking interrupts here, so we only make
+		 * the decisions here, but the wait (for sync standby confirmation)
+		 * happens elsewhere in processing interrupts.
+		 *
+		 * This is somewhat similar to what walwriter does, except that walwriter
+		 * is driven by time (not by amount of WAL written), and does not throttle
+		 * the backends (in a way, the intention is to do exactly the opposite by
+		 * taking away the responsibility for writing WAL from backends).
+		 *
+		 * We may decide to do one or more of these things:
+		 *
+		 * - write the WAL locally (in the background, no throttle)
+		 * - flush the WAL locally (throttle, if synchronous_commit == local)
+		 * - wait for sync replica (throttle, if synchronous_commit != local)
+		 *
+		 * The first point (driven by wal_write_after) is meant to start writing
+		 * WAL to disk early, so that the subsequent flush is faster. This does
+		 * not do any flushes, it merely writes WAL data to page cache.
 		 *
-		 * The throttling is applied only to large transactions (producing more
-		 * than wal_throttle_threshold kilobytes of WAL). Throttled backends
-		 * can be identified by a new wait event SYNC_REP_THROTTLED.
+		 * The second point (driven by wal_flush_after_local) forces the backend
+		 * to actually flush WAL to disk, to reduce the amount of unflushed WAL.
+		 * This is a synchronous operation, and is applied only to transactions
+		 * that generate enough WAL (e.g. 256kB). The goal is to limit impact of
+		 * large transactions (producing a lot of WAL) on small OLTP ones.
+		 *
+		 * The third point (driven by wal_flush_after_remote) is similar, except
+		 * that it throttles based on confirmations from a sync replica.
+		 *
+		 * The throttling is applied only to large transactions, and the backends
+		 * affected by the throttling can be identified by a new wait event
+		 * SYNC_REP_THROTTLED.
 		 *
 		 * Small transactions (by amount of produced WAL) are still subject to
 		 * the sync replication, so the same wait happens at commit time.
 		 *
+		 * All of those thresholds are auto-tuned, i.e. the value is adjusted
+		 * based on concurrent activity. All the large transactions need to
+		 * share the WAL budgets, somehow. We do that by adjusting the value
+		 * by observing the distance since the last write/flush, and comparing
+		 * it to the current limit.
+		 *
 		 * XXX Not sure this is the right place for a comment explaining how the
 		 * throttling works. This place is way too low level, and rather far from
 		 * the place where the wait actually happens.
@@ -1112,14 +1147,47 @@ XLogInsertRecord(XLogRecData *rdata,
 		 * XXX Should this be done even if XLogDelayPending is already set? Maybe
 		 * that should only update XactLastThrottledRecEnd, withoug incrementing
 		 * the pgWalUsage.wal_throttled counter?
+		 *
+		 * XXX Maybe the different cases should have separate wait events?
+		 *
+		 * XXX Note that for the remote WAL, we need to also enforce the local
+		 * flush throttling, because we only send data to replica after it was
+		 * flushed locally.
+		 *
+		 * XXX I guess to make this work well, we should ensure
+		 * (wal_write_after < wal_flush_after_local < wal_flush_after_remove)
 		 */
 		backendWalInserted += rechdr->xl_tot_len;
 
-		if ((synchronous_commit >= SYNCHRONOUS_COMMIT_REMOTE_WRITE) &&
-			(wal_throttle_threshold > 0) &&
-			(backendWalInserted >= wal_throttle_threshold * 1024L))
+		/*
+		 * Should we issue the WAL write to local disk (without a flush)?
+		 */
+		if ((wal_write_after_current > 0) &&
+			(backendWalInserted >= wal_write_after_current * 1024L))
+		{
+			InterruptPending = true;
+			XLogDelayPending = true;
+		}
+
+		/*
+		 * Should we flush the WAL to local disk?
+		 */
+		else if ((synchronous_commit >= SYNCHRONOUS_COMMIT_LOCAL_FLUSH) &&
+			(wal_flush_after_local_current > 0) &&
+			(backendWalInserted >= wal_flush_after_local_current * 1024L))
+		{
+			InterruptPending = true;
+			XLogDelayPending = true;
+			pgWalUsage.wal_throttled++;
+		}
+
+		/*
+		 * Should we flush the WAL to sync replica?
+		 */
+		else if ((synchronous_commit >= SYNCHRONOUS_COMMIT_REMOTE_WRITE) &&
+			(wal_flush_after_remote_current > 0) &&
+			(backendWalInserted >= wal_flush_after_remote_current * 1024L))
 		{
-			XactLastThrottledRecEnd = XactLastRecEnd;
 			InterruptPending = true;
 			XLogDelayPending = true;
 			pgWalUsage.wal_throttled++;
@@ -2703,6 +2771,7 @@ XLogFlush(XLogRecPtr record)
 	 * XXX For the approximation we should probably use LogwrtResult.Flush.
 	 */
 	backendWalInserted = 0;
+	backendLastFlushPtr = record;
 
 	/* Quick exit if already known flushed */
 	if (record <= LogwrtResult.Flush)
@@ -9234,19 +9303,234 @@ SetWalWriterSleeping(bool sleeping)
  * XXX Not sure why we use XactLastThrottledRecEnd and not simply XLogRecEnd?
  */
 void
-HandleXLogDelayPending()
+HandleXLogDelayPending(void)
 {
-	XLogRecPtr 	lsn;
+	/* need to remember the values, because XLogFlush may reset them */
+	uint32		wal_inserted = backendWalInserted;
+	XLogRecPtr	last_flush_lsn = backendLastFlushPtr;
+	XLogRecPtr	flush_lsn = InvalidXLogRecPtr;
+
+	/* make sure we're not in recovery - timeline could change, etc. */
+	Assert(!InRecovery);
+
+	/*
+	 * XactLastRecEnd gets reset once in a while (e.g. after commit or abort),
+	 * and if such record just gets us over the throttling threshold, we'll see
+	 * 0 here. We just ignore the trottling request in that case - one might
+	 * argue the commit already does the throttling anyway (flush and wait
+	 * for sync replica), and it should be a fairly rare case (How likely is
+	 * it that a large transaction crosses the limit just with when writing
+	 * the commit record?).
+	 *
+	 * XXX An alternative would be to have a separate variable, storing the
+	 * XactLastRecEnd value we've seen while deciding to throttle (in fact an
+	 * earlier patch did that). But it does not seem worth it.
+	 */
+	if (XactLastRecEnd == InvalidXLogRecPtr)
+	{
+		XLogDelayPending = false;
+		return;
+	}
+
+	/*
+	 * read LogwrtResult and update local state
+	 *
+	 * XXX Maybe we should use the cached local LogwrtResult copy to determine if
+	 * we need to a write/flush? That wouldn't require a spinlock, and in the worst
+	 * case we'd see a bit stale values and perhaps issue a write/flush early.
+	 * However, we shouldn't to this throttling very often, so seems fine.
+	 */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	LogwrtResult = XLogCtl->LogwrtResult;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	/*
+	 * First decide if we need to issue a write to local WAL. We assume this LSN
+	 * will be ahead of LSN for either flush (local or remote), so let's do this
+	 * first.
+	 */
+	if ((wal_write_after_current > 0) &&
+		(wal_inserted >= wal_write_after_current * 1024L))
+	{
+		/*
+		 * We want to keep write lag below wal_write_after_current, but we don't
+		 * want to write partial pages (because of write amplification).
+		 */
+		XLogRecPtr 	lsn = XactLastRecEnd - wal_write_after_current * 1024L;
+
+		/* truncate to the last full page */
+		lsn = lsn - (lsn % XLOG_BLCKSZ);
+
+		/* Do the WAL write unless someone already issued the write earlier. */
+		if (lsn > LogwrtResult.Write)
+		{
+			START_CRIT_SECTION();
+
+			/*
+			 * Before actually performing the write, wait for all in-flight
+			 * insertions to the pages we're about to write to finish.
+			 */
+			lsn = WaitXLogInsertionsToFinish(lsn);
+
+			LWLockAcquire(WALWriteLock, LW_EXCLUSIVE);
+
+			LogwrtResult = XLogCtl->LogwrtResult;
+			if (LogwrtResult.Write >= lsn)
+			{
+				/* OK, someone wrote it already */
+				LWLockRelease(WALWriteLock);
+			}
+			else
+			{
+				XLogwrtRqst WriteRqst;
+
+				/* Have to write it ourselves */
+				TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_START();
+				WriteRqst.Write = lsn;
+				WriteRqst.Flush = 0;
+				XLogWrite(WriteRqst, XLogCtl->InsertTimeLineID, false);
+				LWLockRelease(WALWriteLock);
+				PendingWalStats.wal_buffers_full++;		// FIXME
+				TRACE_POSTGRESQL_WAL_BUFFER_WRITE_DIRTY_DONE();
+			}
+
+			END_CRIT_SECTION();
+		}
+	}
+
+	/* Now consider if we should do a local flush. */
+	if ((synchronous_commit >= SYNCHRONOUS_COMMIT_LOCAL_FLUSH) &&
+		(wal_flush_after_local_current > 0) &&
+		(wal_inserted >= wal_flush_after_local_current * 1024L))
+	{
+		/* flush lag should be less than wal_flush_after_local_current (kB) */
+		XLogRecPtr 	lsn = XactLastRecEnd - wal_flush_after_local_current * 1024L;
+
+		/* truncate to the last full */
+		lsn = lsn - (lsn % XLOG_BLCKSZ);
+
+		/* do the flush (will reset backendWalInserted etc.) */
+		XLogFlush(lsn);
+
+		/* remember the LSN of this flush */
+		flush_lsn = lsn;
+	}
+
+	/*
+	 * Finally, decide if we should wait for confirmation from a sync replica.
+	 */
+	if ((synchronous_commit >= SYNCHRONOUS_COMMIT_REMOTE_WRITE) &&
+		(wal_flush_after_remote_current > 0) &&
+		(wal_inserted >= wal_flush_after_remote_current * 1024L))
+	{
+		/* sync flush lag should be less than wal_flush_after_remote_current (kB) */
+		XLogRecPtr 	lsn = XactLastRecEnd - wal_flush_after_remote_current * 1024L;
+
+		/* truncate to the last full page */
+		lsn = lsn - (lsn % XLOG_BLCKSZ);
+
+		SyncRepWaitForLSN(lsn, false);
+	}
+
+	/*
+	 * Now adjust the thresholds based on the distance since the last flush in this
+	 * backend, and the fraction of WAL this backend was responsible for.
+	 */
+	if (flush_lsn != InvalidXLogRecPtr)
+	{
+		/* Fraction of WAL between the flushes was produced by this backend. */
+		double wal_fraction = (double) wal_inserted / (flush_lsn - last_flush_lsn);
+
+		double	sum = 0;
+		int		cnt = 0;
+
+		/*
+		 * If the fraction is sufficiently far from 1.0 or close to it, we adjust
+		 * the limits to get close to the "ideal" value based on the fraction. We
+		 * don't immediately switch to the new value, but instead change the values
+		 * gradually by halving/doubling the previous value.
+		 *
+		 * The fraction is guaranteed to be less than 1.0, so we halve the limit
+		 * for low values below 0.5, double the limit for values above 0.9, and
+		 * do nothing for values in between.
+		 *
+		 * We calculate the desired values for each enabled limit (skip those set
+		 * to 0), and then use the average fraction to maintain the proportion.
+		 */
 
-	/* calculate last fully filled page */
-	lsn = XactLastThrottledRecEnd - (XactLastThrottledRecEnd % XLOG_BLCKSZ);
+		if (wal_fraction < 0.5)
+		{
+			wal_write_after_current /= 2;
+			wal_flush_after_local_current /= 2;
+			wal_flush_after_remote_current /= 2;
+		}
+		else if (wal_fraction > 0.9)
+		{
+			wal_write_after_current *= 2;
+			wal_flush_after_local_current *= 2;
+			wal_flush_after_remote_current *= 2;
+		}
+
+
+		/* clamp to correct range */
+
+		wal_write_after_current
+			= Min(wal_write_after, Max(1, wal_write_after_current));
+
+		wal_flush_after_local_current
+			= Min(wal_flush_after_local, Max(1, wal_flush_after_local_current));
+
+		wal_flush_after_remote_current
+			= Min(wal_flush_after_remote, Max(1, wal_flush_after_remote_current));
+
+
+		/* calculate the average fraction */
+
+		if (wal_write_after > 0)
+		{
+			sum += (double) wal_write_after_current / wal_write_after;
+			cnt += 1;
+		}
+
+		if (wal_flush_after_local > 0)
+		{
+			sum += (double) wal_flush_after_local_current / wal_flush_after_local;
+			cnt += 1;
+		}
+
+		if (wal_flush_after_remote > 0)
+		{
+			sum += (double) wal_flush_after_remote_current / wal_flush_after_remote;
+			cnt += 1;
+		}
+
+		Assert(cnt > 0);
+
+		/* apply the average fraction */
+
+		wal_write_after_current = wal_write_after * (sum / cnt);
+		wal_flush_after_local_current = wal_flush_after_local * (sum / cnt);
+		wal_flush_after_remote_current = wal_flush_after_remote * (sum / cnt);
 
-	Assert(wal_throttle_threshold > 0);
-	Assert(backendWalInserted >= wal_throttle_threshold * 1024L);
-	Assert(XactLastThrottledRecEnd != InvalidXLogRecPtr);
+		/* clamp to accepted ranges */
 
-	XLogFlush(lsn);
-	SyncRepWaitForLSN(lsn, false);
+		wal_write_after_current
+			= Min(wal_write_after, Max(1, wal_write_after_current));
+
+		wal_flush_after_local_current
+			= Min(wal_flush_after_local, Max(1, wal_flush_after_local_current));
+
+		wal_flush_after_remote_current
+			= Min(wal_flush_after_remote, Max(1, wal_flush_after_remote_current));
+	}
 
 	XLogDelayPending = false;
 }
+
+void
+ResetXLogThrottling(void)
+{
+	wal_write_after_current = wal_write_after;
+	wal_flush_after_local_current = wal_flush_after_local;
+	wal_flush_after_remote_current = wal_flush_after_remote;
+}
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index b14610e0b6e..5f98de1f356 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -2893,13 +2893,37 @@ struct config_int ConfigureNamesInt[] =
 	},
 
 	{
-		{"wal_throttle_after", PGC_USERSET, REPLICATION_SENDING,
+		{"wal_write_after", PGC_USERSET, REPLICATION_SENDING,
 			gettext_noop("Sets the maximum amount of WAL in kilobytes a backend generates "
 						 " before waiting for sync standby, to limit the replication lag."),
 			NULL,
 			GUC_UNIT_KB
 		},
-		&wal_throttle_threshold,
+		&wal_write_after,
+		0, 0, MAX_KILOBYTES,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"wal_flush_after_local", PGC_USERSET, REPLICATION_SENDING,
+			gettext_noop("Sets the maximum amount of WAL in kilobytes a backend generates "
+						 " before waiting for sync standby, to limit the replication lag."),
+			NULL,
+			GUC_UNIT_KB
+		},
+		&wal_flush_after_local,
+		0, 0, MAX_KILOBYTES,
+		NULL, NULL, NULL
+	},
+
+	{
+		{"wal_flush_after_remote", PGC_USERSET, REPLICATION_SENDING,
+			gettext_noop("Sets the maximum amount of WAL in kilobytes a backend generates "
+						 " before waiting for sync standby, to limit the replication lag."),
+			NULL,
+			GUC_UNIT_KB
+		},
+		&wal_flush_after_remote,
 		0, 0, MAX_KILOBYTES,
 		NULL, NULL, NULL
 	},
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 9d1175edab2..cb7a15c9520 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -54,7 +54,9 @@ extern PGDLLIMPORT char *wal_consistency_checking_string;
 extern PGDLLIMPORT bool log_checkpoints;
 extern PGDLLIMPORT bool track_wal_io_timing;
 extern PGDLLIMPORT int wal_decode_buffer_size;
-extern PGDLLIMPORT int wal_throttle_threshold;
+extern PGDLLIMPORT int wal_write_after;
+extern PGDLLIMPORT int wal_flush_after_local;
+extern PGDLLIMPORT int wal_flush_after_remote;
 extern PGDLLIMPORT uint32 backendWalInserted;
 
 extern PGDLLIMPORT int CheckPointSegments;
@@ -253,6 +255,8 @@ extern XLogRecPtr GetLastImportantRecPtr(void);
 
 extern void SetWalWriterSleeping(bool sleeping);
 extern void HandleXLogDelayPending(void);
+extern void HandleXLogDelayPending(void);
+extern void ResetXLogThrottling(void);
 
 /*
  * Routines used by xlogrecovery.c to call back into xlog.c during recovery.
-- 
2.41.0