reuse network pool buffers for replication

microsoft · Sep 10, 2024 · 9d65b87 · 9d65b87
1 parent a8ae7fb
commit 9d65b87
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 6 deletions.
diff --git a/libs/client/ClientSession/GarnetClientSession.cs b/libs/client/ClientSession/GarnetClientSession.cs
@@ -105,7 +105,7 @@ public GarnetClientSession(
             this.port = port;
 
             this.usingManagedNetworkBuffers = networkBuffers.IsAllocated;
-            this.networkBuffers = usingManagedNetworkBuffers ? networkBuffers : networkBuffers.Allocate(logger);
+            this.networkBuffers = usingManagedNetworkBuffers ? networkBuffers : networkBuffers.Allocate(logger: logger);
             this.bufferSizeDigits = NumUtils.NumDigits(this.networkBuffers.sendMinAllocationSize);
 
             this.logger = logger;

diff --git a/libs/cluster/Server/Replication/PrimaryOps/AofSyncTaskInfo.cs b/libs/cluster/Server/Replication/PrimaryOps/AofSyncTaskInfo.cs
@@ -48,6 +48,7 @@ public void Dispose()
         {
             iter?.Dispose();
             cts?.Dispose();
+            garnetClient?.Dispose();
         }
 
         public unsafe void Consume(byte* payloadPtr, int payloadLength, long currentAddress, long nextAddress)

diff --git a/libs/cluster/Server/Replication/PrimaryOps/AofTaskStore.cs b/libs/cluster/Server/Replication/PrimaryOps/AofTaskStore.cs
@@ -119,7 +119,7 @@ public bool TryAddReplicationTask(string remoteNodeId, long startAddress, out Ao
             aofSyncTaskInfo = null;
 
             if (startAddress == 0) startAddress = ReplicationManager.kFirstValidAofAddress;
-            bool success = false;
+            var success = false;
             var current = clusterProvider.clusterManager.CurrentConfig;
             var (address, port) = current.GetWorkerAddressFromNodeId(remoteNodeId);
 
@@ -131,7 +131,7 @@ public bool TryAddReplicationTask(string remoteNodeId, long startAddress, out Ao
                     this,
                     current.LocalNodeId,
                     remoteNodeId,
-                    new GarnetClientSession(address, port, new NetworkBuffers(1 << 22).Allocate(logger: logger), clusterProvider.serverOptions.TlsOptions?.TlsClientOptions, authUsername: clusterProvider.ClusterUsername, authPassword: clusterProvider.ClusterPassword, logger: logger),
+                    new GarnetClientSession(address, port, clusterProvider.replicationManager.GetNetworkBuffers, clusterProvider.serverOptions.TlsOptions?.TlsClientOptions, authUsername: clusterProvider.ClusterUsername, authPassword: clusterProvider.ClusterPassword, logger: logger),
                     new CancellationTokenSource(),
                     startAddress,
                     logger);

diff --git a/libs/cluster/Server/Replication/ReplicationManager.cs b/libs/cluster/Server/Replication/ReplicationManager.cs
@@ -94,7 +94,7 @@ public ReplicationManager(ClusterProvider clusterProvider, ILogger logger = null
         {
             var opts = clusterProvider.serverOptions;
 
-            this.networkBuffers = new NetworkBuffers(1 << 22, 1 << 22).Allocate(logger: logger);
+            this.networkBuffers = new NetworkBuffers(1 << 22, 1 << 12).Allocate(logger: logger);
             this.clusterProvider = clusterProvider;
             this.storeWrapper = clusterProvider.storeWrapper;
             aofProcessor = new AofProcessor(storeWrapper, recordToAof: false, logger: logger);
@@ -177,6 +177,7 @@ public void Dispose()
             replicationConfigDevice.Dispose();
             pool.Free();
 
+            aofTaskStore.Dispose();
             checkpointStore.WaitForReplicas();
             DisposeReplicaSyncSessionTasks();
             DisposeConnections();

diff --git a/libs/common/NetworkBuffers.cs b/libs/common/NetworkBuffers.cs
@@ -52,9 +52,10 @@ public NetworkBuffers(int sendBufferPoolSize = 1 << 17, int recvBufferPoolSize =
         /// <summary>
         /// Allocate network buffer pool
         /// </summary>
+        /// <param name="maxEntriesPerLevel"></param>
         /// <param name="logger"></param>
         /// <returns></returns>
-        public NetworkBuffers Allocate(ILogger logger = null)
+        public NetworkBuffers Allocate(int maxEntriesPerLevel = 16, ILogger logger = null)
         {
             Debug.Assert(bufferPool == null);
 
@@ -64,7 +65,7 @@ public NetworkBuffers Allocate(ILogger logger = null)
             var levels = maxSize / minSize;
             Debug.Assert(levels > 0);
             levels = Math.Max(4, levels);
-            bufferPool = new LimitedFixedBufferPool(sendMinAllocationSize, numLevels: levels, logger: logger);
+            bufferPool = new LimitedFixedBufferPool(sendMinAllocationSize, maxEntriesPerLevel: maxEntriesPerLevel, numLevels: levels, logger: logger);
             return this;
         }