Fix ExecuteClusterAppendLog (#1187)

* Ensure flush at ExecuteAppendLog happens optimistically outside epoch protection

* log unexpected flush

* bump version

* remove logging and make network buffer settings more clear

Author: vazois

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- VersionPrefix property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.63</VersionPrefix>
+		<VersionPrefix>1.0.64</VersionPrefix>
 	</PropertyGroup>
 </Project>

libs/client/ClientSession/GarnetClientSession.cs

@@ -328,8 +328,8 @@ public unsafe void ExecuteClusterAppendLog(string nodeId, long previousAddress,
         {
             Debug.Assert(nodeId != null);
 
-            byte* curr = offset;
-            int arraySize = 7;
+            var curr = offset;
+            var arraySize = 7;
 
             while (!RespWriteUtils.TryWriteArrayLength(arraySize, ref curr, end))
             {
@@ -389,7 +389,6 @@ public unsafe void ExecuteClusterAppendLog(string nodeId, long previousAddress,
                 curr = offset;
             }
             offset = curr;
-            Flush();
         }
 
         /// <summary>

libs/cluster/Server/Replication/PrimaryOps/AofSyncTaskInfo.cs

@@ -87,6 +87,8 @@ public unsafe void Consume(byte* payloadPtr, int payloadLength, long currentAddr
 
         public void Throttle()
         {
+            // Trigger flush while we are out of epoch protection
+            garnetClient.CompletePending(false);
             garnetClient.Throttle();
         }
 

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -114,6 +114,7 @@ public ReplicationManager(ClusterProvider clusterProvider, ILogger logger = null
             this.storeWrapper = clusterProvider.storeWrapper;
             this.pageSizeBits = storeWrapper.appendOnlyFile == null ? 0 : storeWrapper.appendOnlyFile.UnsafeGetLogPageSizeBits();
 
+            networkBufferSettings.Log(logger, nameof(ReplicationManager));
             this.networkPool = networkBufferSettings.CreateBufferPool(logger: logger);
             ValidateNetworkBufferSettings();
 

libs/cluster/Server/Replication/ReplicationNetworkBufferSettings.cs

@@ -12,9 +12,7 @@ internal sealed partial class ReplicationManager : IDisposable
         /// <summary>
         /// NetworkBufferSettings for the buffer pool maintained by the ReplicationManager
         /// </summary>
-        const int defaultSendBufferSize = 1 << 22;
-        const int defaultInitialReceiveBufferSize = 1 << 12;
-        readonly NetworkBufferSettings networkBufferSettings = new(defaultSendBufferSize, defaultInitialReceiveBufferSize);
+        NetworkBufferSettings networkBufferSettings => NetworkBufferSettings.GetInclusive([GetRSSNetworkBufferSettings, GetIRSNetworkBufferSettings, GetAofSyncNetworkBufferSettings]);
 
         /// <summary>
         /// Network pool maintained by the ReplicationManager
@@ -38,10 +36,11 @@ internal sealed partial class ReplicationManager : IDisposable
 
         /// <summary>
         /// NetworkBufferSettings for the AOF sync task clients
+        /// NOTE: double buffer size for send page to ensure payload (command header + page size) always fits into client buffer.
         /// </summary>
-        const int aofSyncSendBufferSize = 1 << 22;
+        int aofSyncSendBufferSize => 2 << clusterProvider.storeWrapper.serverOptions.AofPageSizeBits();
         const int aofSyncInitialReceiveBufferSize = 1 << 17;
-        public NetworkBufferSettings GetAofSyncNetworkBufferSettings { get; } = new(aofSyncSendBufferSize, aofSyncInitialReceiveBufferSize);
+        public NetworkBufferSettings GetAofSyncNetworkBufferSettings => new(aofSyncSendBufferSize, aofSyncInitialReceiveBufferSize);
 
         void ValidateNetworkBufferSettings()
         {

libs/common/NetworkBufferSettings.cs

@@ -47,6 +47,25 @@ public NetworkBufferSettings(int sendBufferSize = 1 << 17, int initialReceiveBuf
             this.maxReceiveBufferSize = maxReceiveBufferSize;
         }
 
+        /// <summary>
+        /// Return inclusive size for array of settings
+        /// </summary>
+        /// <param name="settings"></param>
+        /// <returns></returns>
+        public static NetworkBufferSettings GetInclusive(NetworkBufferSettings[] settings)
+        {
+            var maxSendBufferSize = 1 << 17;
+            var minInitialReceiveBufferSize = 1 << 17;
+            var maxReceiveBufferSize = 1 << 20;
+            foreach (var setting in settings)
+            {
+                maxSendBufferSize = Math.Max(maxSendBufferSize, setting.sendBufferSize);
+                minInitialReceiveBufferSize = Math.Min(minInitialReceiveBufferSize, setting.initialReceiveBufferSize);
+                maxReceiveBufferSize = Math.Min(maxReceiveBufferSize, setting.maxReceiveBufferSize);
+            }
+            return new NetworkBufferSettings(maxSendBufferSize, minInitialReceiveBufferSize, maxReceiveBufferSize);
+        }
+
         /// <summary>
         /// Allocate network buffer pool
         /// </summary>
@@ -63,5 +82,8 @@ public LimitedFixedBufferPool CreateBufferPool(int maxEntriesPerLevel = 16, ILog
             levels = Math.Max(4, levels);
             return new LimitedFixedBufferPool(minSize, maxEntriesPerLevel: maxEntriesPerLevel, numLevels: levels, logger: logger);
         }
+
+        public void Log(ILogger logger, string category)
+            => logger?.LogInformation("[{category}] network settings: {sendBufferSize}, {initialReceiveBufferSize}, {maxReceiveBufferSize}", category, sendBufferSize, initialReceiveBufferSize, maxReceiveBufferSize);
     }
 }