Try using socket syscalls that accepts a single buffer to improve per…

…formance (#36371) * Try using socket syscalls that accepts a single buffer to improve performance * Remove ref from Receive calls * Prefix Pal methods invoking methods with Sys * Also use single-buffer syscalls for sync Socket Receive methods * Improve comment * PR feedback * Assert SocketAddress null instead of checking
dotnet · May 19, 2020 · cf1390c · cf1390c
1 parent 20ee02b
commit cf1390c
Show file tree

Hide file tree

Showing 10 changed files with 282 additions and 29 deletions.
diff --git a/src/libraries/Common/src/Interop/Unix/System.Native/Interop.Receive.cs b/src/libraries/Common/src/Interop/Unix/System.Native/Interop.Receive.cs
@@ -0,0 +1,16 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Net.Sockets;
+using System.Runtime.InteropServices;
+
+internal static partial class Interop
+{
+    internal static partial class Sys
+    {
+        [DllImport(Libraries.SystemNative, EntryPoint = "SystemNative_Receive")]
+        internal static extern unsafe Error Receive(SafeHandle socket, byte* buffer, int bufferLen, SocketFlags flags, int* received);
+    }
+}
diff --git a/src/libraries/Common/src/Interop/Unix/System.Native/Interop.Send.cs b/src/libraries/Common/src/Interop/Unix/System.Native/Interop.Send.cs
@@ -0,0 +1,16 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Net.Sockets;
+using System.Runtime.InteropServices;
+
+internal static partial class Interop
+{
+    internal static partial class Sys
+    {
+        [DllImport(Libraries.SystemNative, EntryPoint = "SystemNative_Send")]
+        internal static extern unsafe Error Send(SafeHandle socket, byte* buffer, int bufferLen, SocketFlags flags, int* sent);
+    }
+}
diff --git a/src/libraries/Native/Unix/System.Native/pal_networking.c b/src/libraries/Native/Unix/System.Native/pal_networking.c
@@ -1348,6 +1348,34 @@ static int32_t ConvertSocketFlagsPlatformToPal(int platformFlags)
            ((platformFlags & MSG_CTRUNC) == 0 ? 0 : SocketFlags_MSG_CTRUNC);
 }
 
+int32_t SystemNative_Receive(intptr_t socket, void* buffer, int32_t bufferLen, int32_t flags, int32_t* received)
+{
+    if (buffer == NULL || bufferLen < 0 || received == NULL)
+    {
+        return Error_EFAULT;
+    }
+
+    int fd = ToFileDescriptor(socket);
+
+    int socketFlags;
+    if (!ConvertSocketFlagsPalToPlatform(flags, &socketFlags))
+    {
+        return Error_ENOTSUP;
+    }
+
+    ssize_t res;
+    while ((res = recv(fd, buffer, (size_t)bufferLen, socketFlags)) < 0 && errno == EINTR);
+
+    if (res != -1)
+    {
+        *received = (int32_t)res;
+        return Error_SUCCESS;
+    }
+
+    *received = 0;
+    return SystemNative_ConvertErrorPlatformToPal(errno);
+}
+
 int32_t SystemNative_ReceiveMessage(intptr_t socket, MessageHeader* messageHeader, int32_t flags, int64_t* received)
 {
     if (messageHeader == NULL || received == NULL || messageHeader->SocketAddressLen < 0 ||
@@ -1391,6 +1419,38 @@ int32_t SystemNative_ReceiveMessage(intptr_t socket, MessageHeader* messageHeade
     return SystemNative_ConvertErrorPlatformToPal(errno);
 }
 
+int32_t SystemNative_Send(intptr_t socket, void* buffer, int32_t bufferLen, int32_t flags, int32_t* sent)
+{
+    if (buffer == NULL || bufferLen < 0 || sent == NULL)
+    {
+        return Error_EFAULT;
+    }
+
+    int fd = ToFileDescriptor(socket);
+
+    int socketFlags;
+    if (!ConvertSocketFlagsPalToPlatform(flags, &socketFlags))
+    {
+        return Error_ENOTSUP;
+    }
+
+    ssize_t res;
+#if defined(__APPLE__) && __APPLE__
+    // possible OSX kernel bug: https://github.com/dotnet/runtime/issues/27221
+    while ((res = send(fd, buffer, (size_t)bufferLen, socketFlags)) < 0 && (errno == EINTR || errno == EPROTOTYPE));
+#else
+    while ((res = send(fd, buffer, (size_t)bufferLen, socketFlags)) < 0 && errno == EINTR);
+#endif
+    if (res != -1)
+    {
+        *sent = (int32_t)res;
+        return Error_SUCCESS;
+    }
+
+    *sent = 0;
+    return SystemNative_ConvertErrorPlatformToPal(errno);
+}
+
 int32_t SystemNative_SendMessage(intptr_t socket, MessageHeader* messageHeader, int32_t flags, int64_t* sent)
 {
     if (messageHeader == NULL || sent == NULL || messageHeader->SocketAddressLen < 0 ||
@@ -1412,7 +1472,7 @@ int32_t SystemNative_SendMessage(intptr_t socket, MessageHeader* messageHeader,
 
     ssize_t res;
 #if defined(__APPLE__) && __APPLE__
-    // possible OSX kernel bug:  #31927
+    // possible OSX kernel bug: https://github.com/dotnet/runtime/issues/27221
     while ((res = sendmsg(fd, &header, socketFlags)) < 0 && (errno == EINTR || errno == EPROTOTYPE));
 #else
     while ((res = sendmsg(fd, &header, socketFlags)) < 0 && errno == EINTR);

diff --git a/src/libraries/Native/Unix/System.Native/pal_networking.h b/src/libraries/Native/Unix/System.Native/pal_networking.h
@@ -360,8 +360,12 @@ PALEXPORT int32_t SystemNative_SetReceiveTimeout(intptr_t socket, int32_t millis
 
 PALEXPORT int32_t SystemNative_SetSendTimeout(intptr_t socket, int32_t millisecondsTimeout);
 
+PALEXPORT int32_t SystemNative_Receive(intptr_t socket, void* buffer, int32_t bufferLen, int32_t flags, int32_t* received);
+
 PALEXPORT int32_t SystemNative_ReceiveMessage(intptr_t socket, MessageHeader* messageHeader, int32_t flags, int64_t* received);
 
+PALEXPORT int32_t SystemNative_Send(intptr_t socket, void* buffer, int32_t bufferLen, int32_t flags, int32_t* sent);
+
 PALEXPORT int32_t SystemNative_SendMessage(intptr_t socket, MessageHeader* messageHeader, int32_t flags, int64_t* sent);
 
 PALEXPORT int32_t SystemNative_Accept(intptr_t socket, uint8_t* socketAddress, int32_t* socketAddressLen, intptr_t* acceptedSocket);

diff --git a/src/libraries/System.Net.Sockets/src/System.Net.Sockets.csproj b/src/libraries/System.Net.Sockets/src/System.Net.Sockets.csproj
@@ -274,8 +274,12 @@
              Link="Common\Interop\Unix\Interop.Poll.Structs.cs" />
     <Compile Include="$(CommonPath)Interop\Unix\System.Native\Interop.PlatformSocketSupport.cs"
              Link="Common\Interop\Unix\System.Native\Interop.PlatformSocketSupport.cs" />
+    <Compile Include="$(CommonPath)Interop\Unix\System.Native\Interop.Receive.cs"
+             Link="Common\Interop\Unix\System.Native\Interop.Receive.cs" />
     <Compile Include="$(CommonPath)Interop\Unix\System.Native\Interop.ReceiveMessage.cs"
              Link="Common\Interop\Unix\System.Native\Interop.ReceiveMessage.cs" />
+    <Compile Include="$(CommonPath)Interop\Unix\System.Native\Interop.Send.cs"
+             Link="Common\Interop\Unix\System.Native\Interop.Send.cs" />
     <Compile Include="$(CommonPath)Interop\Unix\System.Native\Interop.SendMessage.cs"
              Link="Common\Interop\Unix\System.Native\Interop.SendMessage.cs" />
     <Compile Include="$(CommonPath)Interop\Unix\System.Native\Interop.SetSockOpt.cs"

diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/Socket.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/Socket.cs
@@ -1595,7 +1595,7 @@ public int Receive(IList<ArraySegment<byte>> buffers, SocketFlags socketFlags, o
             if (NetEventSource.IsEnabled) NetEventSource.Info(this, $"SRC:{LocalEndPoint} DST:{RemoteEndPoint}");
 
             int bytesTransferred;
-            errorCode = SocketPal.Receive(_handle, buffers, ref socketFlags, out bytesTransferred);
+            errorCode = SocketPal.Receive(_handle, buffers, socketFlags, out bytesTransferred);
 
 #if TRACE_VERBOSE
             if (NetEventSource.IsEnabled)

diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncContext.Unix.cs
@@ -463,6 +463,7 @@ private abstract class ReceiveOperation : ReadOperation
         private sealed class BufferMemoryReceiveOperation : ReceiveOperation
         {
             public Memory<byte> Buffer;
+            public bool SetReceivedFlags;
 
             public BufferMemoryReceiveOperation(SocketAsyncContext context) : base(context) { }
 
@@ -479,7 +480,17 @@ protected override bool DoTryComplete(SocketAsyncContext context)
                 }
                 else
                 {
-                    return SocketPal.TryCompleteReceiveFrom(context._socket, Buffer.Span, null, Flags, SocketAddress, ref SocketAddressLen, out BytesTransferred, out ReceivedFlags, out ErrorCode);
+                    if (!SetReceivedFlags)
+                    {
+                        Debug.Assert(SocketAddress == null);
+
+                        ReceivedFlags = SocketFlags.None;
+                        return SocketPal.TryCompleteReceive(context._socket, Buffer.Span, Flags, out BytesTransferred, out ErrorCode);
+                    }
+                    else
+                    {
+                        return SocketPal.TryCompleteReceiveFrom(context._socket, Buffer.Span, null, Flags, SocketAddress, ref SocketAddressLen, out BytesTransferred, out ReceivedFlags, out ErrorCode);
+                    }
                 }
             }
 
@@ -1465,13 +1476,13 @@ public SocketError ConnectAsync(byte[] socketAddress, int socketAddressLen, Acti
             return SocketError.IOPending;
         }
 
-        public SocketError Receive(Memory<byte> buffer, ref SocketFlags flags, int timeout, out int bytesReceived)
+        public SocketError Receive(Memory<byte> buffer, SocketFlags flags, int timeout, out int bytesReceived)
         {
             int socketAddressLen = 0;
             return ReceiveFrom(buffer, ref flags, null, ref socketAddressLen, timeout, out bytesReceived);
         }
 
-        public SocketError Receive(Span<byte> buffer, ref SocketFlags flags, int timeout, out int bytesReceived)
+        public SocketError Receive(Span<byte> buffer, SocketFlags flags, int timeout, out int bytesReceived)
         {
             int socketAddressLen = 0;
             return ReceiveFrom(buffer, ref flags, null, ref socketAddressLen, timeout, out bytesReceived);
@@ -1545,6 +1556,39 @@ public unsafe SocketError ReceiveFrom(Span<byte> buffer, ref SocketFlags flags,
             }
         }
 
+        public SocketError ReceiveAsync(Memory<byte> buffer, SocketFlags flags, out int bytesReceived, Action<int, byte[]?, int, SocketFlags, SocketError> callback, CancellationToken cancellationToken = default)
+        {
+            SetNonBlocking();
+
+            SocketError errorCode;
+            int observedSequenceNumber;
+            if (_receiveQueue.IsReady(this, out observedSequenceNumber) &&
+                SocketPal.TryCompleteReceive(_socket, buffer.Span, flags, out bytesReceived, out errorCode))
+            {
+                return errorCode;
+            }
+
+            BufferMemoryReceiveOperation operation = RentBufferMemoryReceiveOperation();
+            operation.SetReceivedFlags = false;
+            operation.Callback = callback;
+            operation.Buffer = buffer;
+            operation.Flags = flags;
+            operation.SocketAddress = null;
+            operation.SocketAddressLen = 0;
+
+            if (!_receiveQueue.StartAsyncOperation(this, operation, observedSequenceNumber, cancellationToken))
+            {
+                bytesReceived = operation.BytesTransferred;
+                errorCode = operation.ErrorCode;
+
+                ReturnOperation(operation);
+                return errorCode;
+            }
+
+            bytesReceived = 0;
+            return SocketError.IOPending;
+        }
+
         public SocketError ReceiveFromAsync(Memory<byte> buffer,  SocketFlags flags, byte[]? socketAddress, ref int socketAddressLen, out int bytesReceived, out SocketFlags receivedFlags, Action<int, byte[]?, int, SocketFlags, SocketError> callback, CancellationToken cancellationToken = default)
         {
             SetNonBlocking();
@@ -1558,6 +1602,7 @@ public SocketError ReceiveFromAsync(Memory<byte> buffer,  SocketFlags flags, byt
             }
 
             BufferMemoryReceiveOperation operation = RentBufferMemoryReceiveOperation();
+            operation.SetReceivedFlags = true;
             operation.Callback = callback;
             operation.Buffer = buffer;
             operation.Flags = flags;
@@ -1579,7 +1624,7 @@ public SocketError ReceiveFromAsync(Memory<byte> buffer,  SocketFlags flags, byt
             return SocketError.IOPending;
         }
 
-        public SocketError Receive(IList<ArraySegment<byte>> buffers, ref SocketFlags flags, int timeout, out int bytesReceived)
+        public SocketError Receive(IList<ArraySegment<byte>> buffers, SocketFlags flags, int timeout, out int bytesReceived)
         {
             return ReceiveFrom(buffers, ref flags, null, 0, timeout, out bytesReceived);
         }

diff --git a/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncEventArgs.Unix.cs b/src/libraries/System.Net.Sockets/src/System/Net/Sockets/SocketAsyncEventArgs.Unix.cs
@@ -128,7 +128,17 @@ internal unsafe SocketError DoOperationReceive(SafeSocketHandle handle, Cancella
             SocketError errorCode;
             if (_bufferList == null)
             {
-                errorCode = handle.AsyncContext.ReceiveAsync(_buffer.Slice(_offset, _count), _socketFlags, out bytesReceived, out flags, TransferCompletionCallback, cancellationToken);
+                // TCP has no out-going receive flags. We can use different syscalls which give better performance.
+                bool noReceivedFlags = _currentSocket!.ProtocolType == ProtocolType.Tcp;
+                if (noReceivedFlags)
+                {
+                    errorCode = handle.AsyncContext.ReceiveAsync(_buffer.Slice(_offset, _count), _socketFlags, out bytesReceived, TransferCompletionCallback, cancellationToken);
+                    flags = SocketFlags.None;
+                }
+                else
+                {
+                    errorCode = handle.AsyncContext.ReceiveAsync(_buffer.Slice(_offset, _count), _socketFlags, out bytesReceived, out flags, TransferCompletionCallback, cancellationToken);
+                }
             }
             else
             {