implement replay side throttling

Author: vazois

libs/cluster/Server/Replication/ReplicaOps/AOFReplay/ReplicaReplayDriver.cs

@@ -34,6 +34,8 @@ internal sealed class ReplicaReplayDriver : IBulkLogEntryConsumer, IDisposable
         readonly TsavoriteLog physicalSublog;
         readonly bool useChannels = false;
 
+        int throttleCounter;
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public bool ResumeReplay() => activeWorkerMonitor.TryEnter();
 
@@ -270,7 +272,52 @@ private unsafe void ConsumeDirect(byte* record, int recordLength, long currentAd
             }
         }
 
-        public void Throttle() { }
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public void Throttle()
+        {
+            if (serverOptions.AofPhysicalSublogCount <= 1)
+                return;
+
+            var w = serverOptions.AofReplayMaxDrift;
+            if (w <= 0)
+                return;
+
+            if (++throttleCounter < w)
+                return;
+            throttleCounter = 0;
+            ThrottleSlow(w);
+
+            void ThrottleSlow(long w)
+            {
+                var rcm = appendOnlyFile.readConsistencyManager;
+                if (rcm == null)
+                    return;
+
+                var myMax = rcm.GetPhysicalSublogMax(physicalSublogIdx);
+
+                // If we're too far ahead, wait until all peers reach our current max
+                if (!AllPeersReached(rcm, Math.Max(myMax - w, w)))
+                {
+                    do
+                    {
+                        cts.Token.ThrowIfCancellationRequested();
+                        Thread.Yield();
+                    } while (!AllPeersReached(rcm, myMax));
+                }
+
+                bool AllPeersReached(ReadConsistencyManager rcm, long target)
+                {
+                    for (var i = 0; i < serverOptions.AofPhysicalSublogCount; i++)
+                    {
+                        if (i == physicalSublogIdx)
+                            continue;
+                        if (rcm.GetPhysicalSublogMax(i) < target)
+                            return false;
+                    }
+                    return true;
+                }
+            }
+        }
         #endregion
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]

libs/host/Configuration/Options.cs

@@ -212,6 +212,9 @@ internal sealed class Options : ICloneable
         [Option("aof-replay-task-count", Required = false, HelpText = "Number of replay tasks per physical sublog at the replica.")]
         public int AofReplayTaskCount { get; set; }
 
+        [Option("aof-replay-max-drift", Required = false, HelpText = "Maximum allowed drift in key sequence numbers between physical sublog replay drivers. When a driver is ahead of the slowest peer by more than this value, it yields. Only effective when aof-physical-sublog-count > 1. -1 = disabled.")]
+        public long AofReplayMaxDrift { get; set; }
+
         [IntRangeValidation(0, int.MaxValue)]
         [Option("aof-tail-witness-freq", Required = false, HelpText = "Polling frequency of the background task responsible for moving time ahead for all physical sublogs (Used only with physical sublog value >1).")]
         public int AofTailWitnessFreqMs { get; set; }
@@ -859,6 +862,7 @@ endpoint is IPEndPoint listenEp && clusterAnnounceEndpoint[0] is IPEndPoint anno
                 AofSegmentSize = AofSegmentSize,
                 AofPhysicalSublogCount = AofPhysicalSublogCount,
                 AofReplayTaskCount = AofReplayTaskCount,
+                AofReplayMaxDrift = AofReplayMaxDrift,
                 AofTailWitnessFreqMs = AofTailWitnessFreqMs,
                 CommitFrequencyMs = CommitFrequencyMs,
                 WaitForCommit = WaitForCommit.GetValueOrDefault(),

libs/host/defaults.conf

@@ -145,6 +145,9 @@
 	/* Number of replay tasks per physical sublog at the replica. */
 	"AofReplayTaskCount": 1,
 
+	/* Maximum allowed drift in key sequence numbers between physical sublog replay drivers. -1 = disabled. */
+	"AofReplayMaxDrift": -1,
+
 	/* Polling frequency of the background task responsible for moving time ahead for all physical sublogs (Used only with physical sublog value >1). */
 	"AofTailWitnessFreqMs": 10,
 

libs/server/AOF/ReadConsistency/ReadConsistencyManager.cs

@@ -54,6 +54,23 @@ public AofAddress GetPhysicalSublogMaxReplayedSequenceNumber()
             return maxKeySeqNumVector;
         }
 
+        /// <summary>
+        /// Gets the maximum replayed sequence number for a single physical sublog
+        /// by reading the max across all its virtual sublogs.
+        /// </summary>
+        /// <param name="physicalSublogIdx">Physical sublog index.</param>
+        /// <returns>The maximum sequence number observed across all virtual sublogs for this physical sublog.</returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public long GetPhysicalSublogMax(int physicalSublogIdx)
+        {
+            var replayTaskCount = serverOptions.AofReplayTaskCount;
+            var startIdx = appendOnlyFile.GetVirtualSublogIdx(physicalSublogIdx, 0);
+            long max = 0;
+            for (var rt = 0; rt < replayTaskCount; rt++)
+                max = Math.Max(max, Volatile.Read(ref vsrs[startIdx + rt].MaxRef));
+            return max;
+        }
+
         /// <summary>
         /// Get frontier sequence number for provided hash
         /// NOTE: Frontier sequence number is maximum sequence number between key specific sequence number and maximum observed sublog sequence number

libs/server/AOF/ReadConsistency/VirtualSublogReplayState.cs

@@ -44,6 +44,11 @@ sealed class SublogReplayStateMax
 
         public readonly long Max => sketchMax.Value;
 
+        /// <summary>
+        /// Reference to the max value for Volatile.Read access from external callers.
+        /// </summary>
+        public ref long MaxRef => ref sketchMax.Value;
+
         public VirtualSublogReplayState()
         {
             var size = SketchSlotSize;

libs/server/Servers/GarnetServerOptions.cs

@@ -87,6 +87,13 @@ public class GarnetServerOptions : ServerOptions
         /// </summary>
         public int AofReplayTaskCount = 1;
 
+        /// <summary>
+        /// Maximum allowed drift (in key sequence numbers) between the fastest and slowest physical sublog replay drivers.
+        /// When a driver is ahead of the slowest peer by more than this value, it yields until the gap closes.
+        /// Only effective when AofPhysicalSublogCount > 1. -1 = disabled (no throttling).
+        /// </summary>
+        public long AofReplayMaxDrift = -1;
+
         /// <summary>
         /// Polling frequency of the background task responsible for moving time ahead for all physical sublogs (Used only with physical sublog value >1).
         /// </summary>