[Tsavorite] Replace SafeTailAddress background refresh with per-thread inflight publish protocol

benchmark/BDN.benchmark/Embedded/EmbeddedRespServer.cs

@@ -30,7 +30,6 @@ public EmbeddedRespServer(GarnetServerOptions opts, ILoggerFactory loggerFactory
                 new SubscribeBroker(
                     null,
                     opts.PubSubPageSizeBytes(),
-                    opts.SubscriberRefreshFrequencyMs,
                     pubSubEpoch,
                     true);
         }

benchmark/Resp.benchmark/OfflineBench/AOFBench/AofBench.cs

@@ -30,7 +30,6 @@ public static GarnetServerOptions GetServerOptions(Options options)
                 AofReplayTaskCount = options.AofReplayTaskCount,
                 ReplicationOffsetMaxLag = 0,
                 CheckpointDir = OperatingSystem.IsLinux() ? "/tmp" : null,
-                AofReplicationRefreshFrequencyMs = 100,
             };
             return serverOptions;
         }

benchmark/Resp.benchmark/OfflineBench/AOFBench/AofGen.cs

@@ -234,7 +234,6 @@ public AofGen(Options options)
                 EnableFastCommit = true,
                 CommitFrequencyMs = -1,
                 FastAofTruncate = true,
-                AofReplicationRefreshFrequencyMs = options.AofReplicationRefreshFrequencyMs,
                 EnableCluster = true,
                 ReplicationOffsetMaxLag = 0,
                 AofPhysicalSublogCount = options.AofPhysicalSublogCount

benchmark/Resp.benchmark/OfflineBench/GarnetServerInstance.cs

@@ -29,7 +29,6 @@ public static GarnetServerOptions GetServerOptions(Options options)
                 AofReplayTaskCount = options.AofReplayTaskCount,
                 ReplicationOffsetMaxLag = 0,
                 CheckpointDir = OperatingSystem.IsLinux() ? "/tmp" : null,
-                AofReplicationRefreshFrequencyMs = 100,
             };
             return serverOptions;
         }

benchmark/Resp.benchmark/Options.cs

@@ -152,9 +152,6 @@ public partial class Options
         [Option("aof-page-size", Required = false, Default = "4m", HelpText = "Size of each AOF page in bytes(rounds down to power of 2)")]
         public string AofPageSize { get; set; }
 
-        [Option("aof-tail-ref-freq", Required = false, Default = 100, HelpText = "Aof Tail Refresh Frequency.")]
-        public int AofReplicationRefreshFrequencyMs { get; set; }
-
         /// <summary>
         /// Parse size from string specification
         /// </summary>

libs/cluster/Server/Replication/PrimaryOps/AofOperations/AofSyncDriverStore.cs

@@ -21,26 +21,18 @@ internal sealed class LogShiftTailCallback(int physicalSublogIdx, AofSyncDriverS
             readonly int physicalSublogIdx = physicalSublogIdx;
             readonly AofSyncDriverStore store = store;
 
-            internal void SafeTailShiftCallback(long oldTailAddress, long newTailAddress)
+            internal void SafeTailPageShiftCallback(long oldTailAddress, long newTailAddress)
             {
-                var oldPage = oldTailAddress >> store.logPageSizeBits;
-                var newPage = newTailAddress >> store.logPageSizeBits;
-
-                // Call truncate only once per page
-                if (oldPage != newPage)
-                {
-                    // Truncate 2 pages above ReadOnly mark, so that we have sufficient time to shift begin before we flush.
-                    // Make sure this is page-aligned, in case we go to a non-page-aligned ReadOnlyAddress.
-                    var truncateUntilAddress = store.clusterProvider.storeWrapper.appendOnlyFile.Log.UnsafeGetReadOnlyAddressAbove(physicalSublogIdx, newTailAddress, numPagesAbove: 2);
-                    if (truncateUntilAddress > 0)
-                        _ = store.SafeTruncateAof(truncateUntilAddress, physicalSublogIdx);
-                }
+                // Truncate 2 pages above ReadOnly mark, so that we have sufficient time to shift begin before we flush.
+                // Make sure this is page-aligned, in case we go to a non-page-aligned ReadOnlyAddress.
+                var truncateUntilAddress = store.clusterProvider.storeWrapper.appendOnlyFile.Log.UnsafeGetReadOnlyAddressAbove(physicalSublogIdx, newTailAddress, numPagesAbove: 2);
+                if (truncateUntilAddress > 0)
+                    _ = store.SafeTruncateAof(truncateUntilAddress, physicalSublogIdx);
             }
         }
 
         readonly ClusterProvider clusterProvider;
         readonly ILogger logger;
-        readonly int logPageSizeBits, logPageSizeMask;
 
         AofSyncDriver[] syncDrivers;
         int numDrivers;
@@ -58,15 +50,12 @@ public AofSyncDriverStore(ClusterProvider clusterProvider, int initialSize = 1,
             numDrivers = 0;
             if (clusterProvider.storeWrapper.appendOnlyFile != null)
             {
-                logPageSizeBits = clusterProvider.storeWrapper.appendOnlyFile.Log.UnsafeGetLogPageSizeBits();
-                var logPageSize = 1 << logPageSizeBits;
-                logPageSizeMask = logPageSize - 1;
                 if (clusterProvider.serverOptions.FastAofTruncate)
                 {
                     for (var i = 0; i < clusterProvider.serverOptions.AofPhysicalSublogCount; i++)
                     {
                         var logShiftTailCallback = new LogShiftTailCallback(i, this);
-                        clusterProvider.storeWrapper.appendOnlyFile.Log.SetLogShiftTailCallback(i, logShiftTailCallback.SafeTailShiftCallback);
+                        clusterProvider.storeWrapper.appendOnlyFile.Log.SetLogShiftTailCallback(i, logShiftTailCallback.SafeTailPageShiftCallback);
                     }
                 }
             }

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -689,7 +689,7 @@ async Task AdvanceTimeWorkerAsync(CancellationToken token)
                                     else
                                         converged = false;
                                 }
-                                await Task.Delay(storeWrapper.serverOptions.AofReplicationRefreshFrequencyMs, token).ConfigureAwait(false);
+                                await Task.Delay(storeWrapper.serverOptions.AofTailWitnessFreqMs, token).ConfigureAwait(false);
                             }
                         }
                     }

libs/host/Configuration/Options.cs

@@ -228,14 +228,6 @@ internal sealed class Options : ICloneable
         [Option("aof-size-limit-enforce-frequency", Required = false, HelpText = "Frequency (in secs) of execution of the AutoCheckpointBasedOnAofSizeLimit background task.")]
         public int AofSizeLimitEnforceFrequencySecs { get; set; }
 
-        [IntRangeValidation(0, int.MaxValue)]
-        [Option("aof-refresh-freq", Required = false, HelpText = "AOF replication (safe tail address) refresh frequency in milliseconds. 0 = auto refresh after every enqueue.")]
-        public int AofReplicationRefreshFrequencyMs { get; set; }
-
-        [IntRangeValidation(0, int.MaxValue)]
-        [Option("subscriber-refresh-freq", Required = false, HelpText = "Subscriber (safe tail address) refresh frequency in milliseconds (for pub-sub). 0 = auto refresh after every enqueue.")]
-        public int SubscriberRefreshFrequencyMs { get; set; }
-
         [IntRangeValidation(0, int.MaxValue)]
         [Option("compaction-freq", Required = false, HelpText = "Background hybrid log compaction frequency in seconds. 0 = disabled (compaction performed before checkpointing instead)")]
         public int CompactionFrequencySecs { get; set; }
@@ -856,7 +848,6 @@ endpoint is IPEndPoint listenEp && clusterAnnounceEndpoint[0] is IPEndPoint anno
                 AofPhysicalSublogCount = AofPhysicalSublogCount,
                 AofReplayTaskCount = AofReplayTaskCount,
                 AofTailWitnessFreqMs = AofTailWitnessFreqMs,
-                AofReplicationRefreshFrequencyMs = AofReplicationRefreshFrequencyMs,
                 CommitFrequencyMs = CommitFrequencyMs,
                 WaitForCommit = WaitForCommit.GetValueOrDefault(),
                 AofSizeLimit = AofSizeLimit,

libs/host/GarnetServer.cs

@@ -244,7 +244,7 @@ private void InitializeServer()
                 CreateDatabase(dbId, opts, clusterFactory, customCommandManager);
 
             if (!opts.DisablePubSub)
-                subscribeBroker = new SubscribeBroker(null, opts.PubSubPageSizeBytes(), opts.SubscriberRefreshFrequencyMs, pubSubEpoch, startFresh: true, logger);
+                subscribeBroker = new SubscribeBroker(null, opts.PubSubPageSizeBytes(), pubSubEpoch, startFresh: true, logger);
 
             logger?.LogTrace("TLS is {tlsEnabled}", opts.TlsOptions == null ? "disabled" : "enabled");
 

libs/host/defaults.conf

@@ -132,9 +132,6 @@
 	/* Size of each AOF page in bytes(rounds down to power of 2) */
 	"AofPageSize" : "4m",
 
-	/* AOF replication (safe tail address) refresh frequency in milliseconds. 0 = auto refresh after every enqueue. */
-	"AofReplicationRefreshFrequencyMs": 10,
-
 	/* Number of AOF physical sublogs (i.e. TsavoriteLog instances) used (=1 equivalent to the legacy single log implementation >1: sharded log implementation. */
 	"AofPhysicalSublogCount": 1,
 
@@ -144,9 +141,6 @@
 	/* Polling frequency of the background task responsible for moving time ahead for all physical sublogs (Used only with physical sublog value >1). */
 	"AofTailWitnessFreqMs": 10,
 
-	/* Subscriber (safe tail address) refresh frequency in milliseconds (for pub-sub). 0 = auto refresh after every enqueue. */
-	"SubscriberRefreshFrequencyMs": 0,
-
 	/* Write ahead logging (append-only file) commit issue frequency in milliseconds. 0 = issue an immediate commit per operation, -1 = manually issue commits using COMMITAOF command */
 	"CommitFrequencyMs" : 0,
 

libs/server/AOF/GarnetLog.cs

@@ -309,16 +309,16 @@ public long GetBeginAddress(int sublogIdx)
         /// Set log shift tail callbacks
         /// </summary>
         /// <param name="sublogIdx"></param>
-        /// <param name="SafeTailShiftCallback"></param>
-        public void SetLogShiftTailCallback(int sublogIdx, Action<long, long> SafeTailShiftCallback)
+        /// <param name="safeTailPageShiftCallback"></param>
+        public void SetLogShiftTailCallback(int sublogIdx, Action<long, long> safeTailPageShiftCallback)
         {
             if (singleLog != null)
             {
-                singleLog.log.SafeTailShiftCallback = SafeTailShiftCallback;
+                singleLog.log.SafeTailPageShiftCallback = safeTailPageShiftCallback;
             }
             else
             {
-                shardedLog.sublog[sublogIdx].SafeTailShiftCallback = SafeTailShiftCallback;
+                shardedLog.sublog[sublogIdx].SafeTailPageShiftCallback = safeTailPageShiftCallback;
             }
         }
 

libs/server/PubSub/SubscribeBroker.cs

@@ -36,13 +36,12 @@ public sealed class SubscribeBroker : IDisposable, ILogEntryConsumer
         /// </summary>
         /// <param name="logDir">Directory where the log will be stored</param>
         /// <param name="pageSize">Page size of log used for pub/sub</param>
-        /// <param name="subscriberRefreshFrequencyMs">Subscriber log refresh frequency</param>
         /// <param name="startFresh">start the log from scratch, do not continue</param>
-        public SubscribeBroker(string logDir, long pageSize, int subscriberRefreshFrequencyMs, LightEpoch epoch, bool startFresh = true, ILogger logger = null)
+        public SubscribeBroker(string logDir, long pageSize, LightEpoch epoch, bool startFresh = true, ILogger logger = null)
         {
             device = logDir == null ? new NullDevice() : Devices.CreateLogDevice(logDir + "/pubsubkv", preallocateFile: false);
             device.Initialize((long)(1 << 30) * 64);
-            aof = new TsavoriteLog(new TsavoriteLogSettings { LogDevice = device, PageSize = pageSize, MemorySize = pageSize * 4, SafeTailRefreshFrequencyMs = subscriberRefreshFrequencyMs, Epoch = epoch });
+            aof = new TsavoriteLog(new TsavoriteLogSettings { LogDevice = device, PageSize = pageSize, MemorySize = pageSize * 4, Epoch = epoch });
             pageSizeBits = aof.UnsafeGetLogPageSizeBits();
             if (startFresh)
                 aof.TruncateUntil(aof.CommittedUntilAddress);

libs/server/Servers/GarnetServerOptions.cs

@@ -71,11 +71,6 @@ public class GarnetServerOptions : ServerOptions
         /// </summary>
         public string AofPageSize = "4m";
 
-        /// <summary>
-        /// AOF replication (safe tail address) refresh frequency in milliseconds. 0 = auto refresh after every enqueue.
-        /// </summary>
-        public int AofReplicationRefreshFrequencyMs = 10;
-
         /// <summary>
         /// Number of AOF physical sublogs (i.e. TsavoriteLog instances) used (=1 equivalent to the legacy single log implementation >1: sharded log implementation.
         /// </summary>
@@ -91,11 +86,6 @@ public class GarnetServerOptions : ServerOptions
         /// </summary>
         public int AofTailWitnessFreqMs = 100;
 
-        /// <summary>
-        /// Subscriber (safe tail address) refresh frequency in milliseconds (for pub-sub). 0 = auto refresh after every enqueue.
-        /// </summary>
-        public int SubscriberRefreshFrequencyMs = 0;
-
         /// <summary>
         /// Write ahead logging (append-only file) commit issue frequency in milliseconds.
         /// 0 = issue an immediate commit per operation
@@ -809,7 +799,6 @@ public void GetAofSettings(int dbId, out TsavoriteLogSettings[] tsavoriteLogSett
                     PageSizeBits = AofPageSizeBits(),
                     LogDevice = GetAofDevice(dbId, subLogIdx: AofPhysicalSublogCount == 1 ? -1 : i),
                     TryRecoverLatest = false,
-                    SafeTailRefreshFrequencyMs = EnableCluster ? AofReplicationRefreshFrequencyMs : -1,
                     FastCommitMode = EnableFastCommit,
                     AutoCommit = AofAutoCommit && (AofPhysicalSublogCount == 1),
                     MutableFraction = 0.9,