Convert sync over async network handling to async

benchmark/BDN.benchmark/BDN.benchmark.csproj

@@ -12,7 +12,11 @@
     <PackageReference Include="BenchmarkDotNet" />
 		<PackageReference Include="KeraLua" />
 	</ItemGroup>
-
+<ItemGroup>
+  <None Include="../../test/testcerts/testcert.pfx">
+    <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+  </None>
+</ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\libs\cluster\Garnet.cluster.csproj" />
     <ProjectReference Include="..\..\libs\host\Garnet.host.csproj" />

libs/common/Networking/NetworkHandler.cs

@@ -271,11 +271,11 @@ async Task AuthenticateAsClientAsync(SslClientAuthenticationOptions sslClientOpt
         /// On network receive
         /// </summary>
         /// <param name="bytesTransferred">Number of bytes transferred</param>
-        public unsafe void OnNetworkReceive(int bytesTransferred)
+        public async ValueTask OnNetworkReceiveAsync(int bytesTransferred)
         {
             // Wait for SslStream async processing to complete, if any (e.g., authentication phase)
             while (readerStatus == TlsReaderStatus.Active)
-                expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false).GetAwaiter().GetResult();
+                await expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
 
             // Increment network bytes read
             networkBytesRead += bytesTransferred;
@@ -287,7 +287,10 @@ public unsafe void OnNetworkReceive(int bytesTransferred)
                     if (sslStream == null)
                     {
                         transportReceiveBuffer = networkReceiveBuffer;
-                        transportReceiveBufferPtr = networkReceiveBufferPtr;
+                        unsafe
+                        {
+                            transportReceiveBufferPtr = networkReceiveBufferPtr;
+                        }
                         transportBytesRead = networkBytesRead;
 
                         // We do not have an active read task, so we will process on the network thread
@@ -298,7 +301,7 @@ public unsafe void OnNetworkReceive(int bytesTransferred)
                         readerStatus = TlsReaderStatus.Active;
                         Read();
                         while (readerStatus == TlsReaderStatus.Active)
-                            expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false).GetAwaiter().GetResult();
+                            await expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
                     }
                     break;
                 case TlsReaderStatus.Waiting:
@@ -309,7 +312,7 @@ public unsafe void OnNetworkReceive(int bytesTransferred)
                     _ = receivedData.Release();
 
                     while (readerStatus == TlsReaderStatus.Active)
-                        expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false).GetAwaiter().GetResult();
+                        await expectingData.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
                     break;
                 default:
                     ThrowInvalidOperationException($"Unexpected reader status {readerStatus}");

libs/common/Networking/TcpNetworkHandlerBase.cs

@@ -135,6 +135,12 @@ 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);
+        }
+
+        private async ValueTask HandleReceiveAsync(object sender, SocketAsyncEventArgs e)
         {
             try
             {
@@ -146,20 +152,29 @@ void RecvEventArg_Completed(object sender, SocketAsyncEventArgs e)
                         Dispose(e);
                         break;
                     }
-                    OnNetworkReceive(e.BytesTransferred);
+                    var receiveTask = OnNetworkReceiveAsync(e.BytesTransferred);
+                    if (!receiveTask.IsCompletedSuccessfully)
+                    {
+                        await receiveTask;
+                    }
                     e.SetBuffer(networkReceiveBuffer, networkBytesRead, networkReceiveBuffer.Length - networkBytesRead);
                 } while (!e.AcceptSocket.ReceiveAsync(e));
             }
             catch (Exception ex)
             {
-                if (ex is ObjectDisposedException ex2 && ex2.ObjectName == "System.Net.Sockets.Socket")
-                    logger?.LogTrace("Accept socket was disposed at RecvEventArg_Completed");
-                else
-                    logger?.LogError(ex, "An error occurred at RecvEventArg_Completed");
-                Dispose(e);
+                HandleReceiveFailure(ex, e);
             }
         }
 
+        void HandleReceiveFailure(Exception ex, SocketAsyncEventArgs e)
+        {
+            if (ex is ObjectDisposedException ex2 && ex2.ObjectName == "System.Net.Sockets.Socket")
+                logger?.LogTrace("Accept socket was disposed at RecvEventArg_Completed");
+            else
+                logger?.LogError(ex, "An error occurred at RecvEventArg_Completed");
+            Dispose(e);
+        }
+
         unsafe void AllocateNetworkReceiveBuffer()
         {
             networkReceiveBufferEntry = networkPool.Get(networkBufferSettings.initialReceiveBufferSize);