segregate TLS and non TLS paths

benchmark/BDN.benchmark/Embedded/EmbeddedNetworkHandler.cs

@@ -25,12 +25,12 @@ public override void Dispose()
 
         public override bool TryClose() => throw new NotImplementedException();
 
-        public async ValueTask Send(Request request)
+        public void Send(Request request)
         {
             networkReceiveBuffer = request.buffer;
             unsafe { networkReceiveBufferPtr = request.bufferPtr; }
 
-            await OnNetworkReceiveAsync(request.buffer.Length);
+            OnNetworkReceive(request.buffer.Length);
 
             Debug.Assert(networkBytesRead == 0);
             Debug.Assert(networkReadHead == 0);

benchmark/BDN.benchmark/Network/BasicOperations.cs

@@ -22,9 +22,9 @@ public override void GlobalSetup()
         }
 
         [Benchmark]
-        public async ValueTask InlinePing()
+        public void InlinePing()
         {
-            await Send(ping);
+            Send(ping);
         }
     }
 }

benchmark/BDN.benchmark/Network/NetworkBase.cs

@@ -64,10 +64,7 @@ public virtual void GlobalCleanup()
             server.Dispose();
         }
 
-        protected ValueTask Send(Request request)
-        {
-            return networkHandler.Send(request);
-        }
+        protected void Send(Request request) => networkHandler.Send(request);
 
         protected unsafe void SetupOperation(ref Request request, ReadOnlySpan<byte> operation, int batchSize = batchSize)
         {

benchmark/BDN.benchmark/Network/RawStringOperations.cs

@@ -65,63 +65,63 @@ public override void GlobalSetup()
         }
 
         [Benchmark]
-        public async ValueTask Set()
+        public void Set()
         {
-            await Send(set);
+            Send(set);
         }
 
         [Benchmark]
-        public async ValueTask SetEx()
+        public void SetEx()
         {
-            await Send(setex);
+            Send(setex);
         }
 
         [Benchmark]
-        public async ValueTask SetNx()
+        public void SetNx()
         {
-            await Send(setnx);
+            Send(setnx);
         }
 
         [Benchmark]
-        public async ValueTask SetXx()
+        public void SetXx()
         {
-            await Send(setxx);
+            Send(setxx);
         }
 
         [Benchmark]
-        public async ValueTask GetFound()
+        public void GetFound()
         {
-            await Send(getf);
+            Send(getf);
         }
 
         [Benchmark]
-        public async ValueTask GetNotFound()
+        public void GetNotFound()
         {
-            await Send(getnf);
+            Send(getnf);
         }
 
         [Benchmark]
-        public async ValueTask Increment()
+        public void Increment()
         {
-            await Send(incr);
+            Send(incr);
         }
 
         [Benchmark]
-        public async ValueTask Decrement()
+        public void Decrement()
         {
-            await Send(decr);
+            Send(decr);
         }
 
         [Benchmark]
-        public async ValueTask IncrementBy()
+        public void IncrementBy()
         {
-            await Send(incrby);
+            Send(incrby);
         }
 
         [Benchmark]
-        public async ValueTask DecrementBy()
+        public void DecrementBy()
         {
-            await Send(decrby);
+            Send(decrby);
         }
     }
 }

libs/common/Networking/NetworkHandler.cs

@@ -267,11 +267,38 @@ async Task AuthenticateAsClientAsync(SslClientAuthenticationOptions sslClientOpt
             }
         }
 
+        public void OnNetworkReceive(int bytesTransferred)
+        {
+            networkBytesRead += bytesTransferred;
+            transportReceiveBuffer = networkReceiveBuffer;
+            unsafe
+            {
+                transportReceiveBufferPtr = networkReceiveBufferPtr;
+            }
+            transportBytesRead = networkBytesRead;
+
+            // We do not have an active read task, so we will process on the network thread
+            Process();
+            EndTransformNetworkToTransport();
+            UpdateNetworkBuffers();
+        }
+
+        private void UpdateNetworkBuffers()
+        {
+            // Shift network buffer after processing is done
+            if (networkReadHead > 0)
+                ShiftNetworkReceiveBuffer();
+
+            // Double network buffer if out of space after processing is complete
+            if (networkBytesRead == networkReceiveBuffer.Length)
+                DoubleNetworkReceiveBuffer();
+        }
+
         /// <summary>
         /// On network receive
         /// </summary>
         /// <param name="bytesTransferred">Number of bytes transferred</param>
-        public async ValueTask OnNetworkReceiveAsync(int bytesTransferred)
+        public async ValueTask OnNetworkReceiveWithTLS(int bytesTransferred)
         {
             // Wait for SslStream async processing to complete, if any (e.g., authentication phase)
             while (readerStatus == TlsReaderStatus.Active)
@@ -283,26 +310,10 @@ public async ValueTask OnNetworkReceiveAsync(int bytesTransferred)
             switch (readerStatus)
             {
                 case TlsReaderStatus.Rest:
-                    // Synchronously try to process the received data
-                    if (sslStream == null)
-                    {
-                        transportReceiveBuffer = networkReceiveBuffer;
-                        unsafe
-                        {
-                            transportReceiveBufferPtr = networkReceiveBufferPtr;
-                        }
-                        transportBytesRead = networkBytesRead;
-
-                        // We do not have an active read task, so we will process on the network thread
-                        Process();
-                    }
-                    else
-                    {
-                        readerStatus = TlsReaderStatus.Active;
-                        Read();
-                        while (readerStatus == TlsReaderStatus.Active)
-                            await expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
-                    }
+                    readerStatus = TlsReaderStatus.Active;
+                    Read();
+                    while (readerStatus == TlsReaderStatus.Active)
+                        await expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
                     break;
                 case TlsReaderStatus.Waiting:
                     // We have a ReadAsync task waiting for new data, set it to active status
@@ -320,16 +331,7 @@ public async ValueTask OnNetworkReceiveAsync(int bytesTransferred)
             }
 
             Debug.Assert(readerStatus != TlsReaderStatus.Active);
-
-            EndTransformNetworkToTransport();
-
-            // Shift network buffer after processing is done
-            if (networkReadHead > 0)
-                ShiftNetworkReceiveBuffer();
-
-            // Double network buffer if out of space after processing is complete
-            if (networkBytesRead == networkReceiveBuffer.Length)
-                DoubleNetworkReceiveBuffer();
+            UpdateNetworkBuffers();
         }
 
         [MethodImpl(MethodImplOptions.NoInlining)]

libs/common/Networking/TcpNetworkHandlerBase.cs

@@ -26,6 +26,7 @@ public abstract class TcpNetworkHandlerBase<TServerHook, TNetworkSender> : Netwo
         readonly string remoteEndpoint;
         readonly string localEndpoint;
         int closeRequested;
+        readonly bool useTLS;
 
         /// <summary>
         /// Constructor
@@ -39,7 +40,7 @@ public TcpNetworkHandlerBase(TServerHook serverHook, TNetworkSender networkSende
 
             remoteEndpoint = socket.RemoteEndPoint is IPEndPoint remote ? $"{remote.Address}:{remote.Port}" : "";
             localEndpoint = socket.LocalEndPoint is IPEndPoint local ? $"{local.Address}:{local.Port}" : "";
-
+            this.useTLS = useTLS;
             AllocateNetworkReceiveBuffer();
         }
 
@@ -137,7 +138,36 @@ void Dispose(SocketAsyncEventArgs e)
         void RecvEventArg_Completed(object sender, SocketAsyncEventArgs e)
         {
             // Complete receive event and release thread while we process data async
-            _ = HandleReceiveAsync(sender, e);
+            if (this.useTLS)
+            {
+                _ = HandleReceiveAsync(sender, e);
+            }
+            else
+            {
+                HandleReceiveSync(sender, e);
+            }
+        }
+
+        private void HandleReceiveSync(object sender, SocketAsyncEventArgs e)
+        {
+            try
+            {
+                do
+                {
+                    if (e.BytesTransferred == 0 || e.SocketError != SocketError.Success || serverHook.Disposed)
+                    {
+                        // No more things to receive
+                        Dispose(e);
+                        break;
+                    }
+                    OnNetworkReceive(e.BytesTransferred);
+                    e.SetBuffer(networkReceiveBuffer, networkBytesRead, networkReceiveBuffer.Length - networkBytesRead);
+                } while (!e.AcceptSocket.ReceiveAsync(e));
+            }
+            catch (Exception ex)
+            {
+                HandleReceiveFailure(ex, e);
+            }
         }
 
         private async ValueTask HandleReceiveAsync(object sender, SocketAsyncEventArgs e)
@@ -152,7 +182,7 @@ private async ValueTask HandleReceiveAsync(object sender, SocketAsyncEventArgs e
                         Dispose(e);
                         break;
                     }
-                    var receiveTask = OnNetworkReceiveAsync(e.BytesTransferred);
+                    var receiveTask = OnNetworkReceiveWithTLS(e.BytesTransferred);
                     if (!receiveTask.IsCompletedSuccessfully)
                     {
                         await receiveTask;