Implement SocketPal.Poll on Unix with poll instead of select

src/Common/src/Interop/Unix/System.Native/Interop.Poll.cs

@@ -10,10 +10,12 @@ internal static partial class Interop
 {
     internal static partial class Sys
     {
+        [Flags]
         internal enum PollEvents : short
         {
             POLLNONE = 0x0000,  // No events occurred.
-            POLLIN   = 0x0001,  // any readable data available
+            POLLIN   = 0x0001,  // non-urgent readable data available
+            POLLPRI  = 0x0002,  // urgent readable data available
             POLLOUT  = 0x0004,  // data can be written without blocked
             POLLERR  = 0x0008,  // an error occurred
             POLLHUP  = 0x0010,  // the file descriptor hung up
@@ -46,7 +48,7 @@ internal struct PollEvent
         /// <param name="timeout">The amount of time to wait; -1 for infinite, 0 for immediate return, and a positive number is the number of milliseconds</param>
         /// <param name="triggered">The events that were returned by the poll call. May be PollEvents.POLLNONE in the case of a timeout.</param>
         /// <returns>An error or Error.SUCCESS.</returns>
-        internal static unsafe Error Poll(SafeFileHandle fd, PollEvents events, int timeout, out PollEvents triggered)
+        internal static unsafe Error Poll(SafeHandle fd, PollEvents events, int timeout, out PollEvents triggered)
         {
             bool gotRef = false;
             try

src/Native/System.Native/pal_io.cpp

@@ -102,6 +102,7 @@ static_assert(PAL_SEEK_END == SEEK_END, "");
 
 // Validate our PollFlags enum values are correct for the platform
 static_assert(PAL_POLLIN == POLLIN, "");
+static_assert(PAL_POLLPRI == POLLPRI, "");
 static_assert(PAL_POLLOUT == POLLOUT, "");
 static_assert(PAL_POLLERR == POLLERR, "");
 static_assert(PAL_POLLHUP == POLLHUP, "");

src/Native/System.Native/pal_io.h

@@ -218,7 +218,8 @@ enum SysConfName : int32_t
  */
 enum PollEvents : int16_t
 {
-    PAL_POLLIN = 0x0001,   /* any readable data available */
+    PAL_POLLIN = 0x0001,   /* non-urgent readable data available */
+    PAL_POLLPRI = 0x0002,  /* urgent readable data available */
     PAL_POLLOUT = 0x0004,  /* data can be written without blocked */
     PAL_POLLERR = 0x0008,  /* an error occurred */
     PAL_POLLHUP = 0x0010,  /* the file descriptor hung up */

src/System.Net.Sockets/src/System.Net.Sockets.csproj

@@ -442,6 +442,9 @@
     <Compile Include="$(CommonPath)\Interop\Unix\System.Native\Interop.MulticastOption.cs">
       <Link>Interop\Unix\System.Native\Interop.MulticastOption.cs</Link>
     </Compile>
+    <Compile Include="$(CommonPath)\Interop\Unix\System.Native\Interop.Poll.cs">
+      <Link>Interop\Unix\System.Native\Interop.Poll.cs</Link>
+    </Compile>
     <Compile Include="$(CommonPath)\Interop\Unix\System.Native\Interop.PlatformSocketSupport.cs">
       <Link>Interop\Unix\System.Native\Interop.PlatformSocketSupport.cs</Link>
     </Compile>

src/System.Net.Sockets/src/System/Net/Sockets/SocketPal.Unix.cs

@@ -1061,54 +1061,30 @@ public static unsafe SocketError GetLingerOption(SafeCloseSocket handle, out Lin
 
         public static unsafe SocketError Poll(SafeCloseSocket handle, int microseconds, SelectMode mode, out bool status)
         {
-            uint* fdSet = stackalloc uint[Interop.Sys.FD_SETSIZE_UINTS];
-            Interop.Sys.FD_ZERO(fdSet);
-
-            bool releaseHandle = false;
-            try
+            Interop.Sys.PollEvents inEvent = Interop.Sys.PollEvents.POLLNONE;
+            switch (mode)
             {
-                handle.DangerousAddRef(ref releaseHandle);
-                int fd = (int)handle.DangerousGetHandle();
-                Interop.Sys.FD_SET(fd, fdSet);
-
-                int fdCount = 0;
-                uint* readFds = null;
-                uint* writeFds = null;
-                uint* errorFds = null;
-                switch (mode)
-                {
-                    case SelectMode.SelectRead:
-                        readFds = fdSet;
-                        fdCount = fd + 1;
-                        break;
-
-                    case SelectMode.SelectWrite:
-                        writeFds = fdSet;
-                        fdCount = fd + 1;
-                        break;
-
-                    case SelectMode.SelectError:
-                        errorFds = fdSet;
-                        fdCount = fd + 1;
-                        break;
-                }
+                case SelectMode.SelectRead: inEvent = Interop.Sys.PollEvents.POLLIN; break;
+                case SelectMode.SelectWrite: inEvent = Interop.Sys.PollEvents.POLLOUT; break;
+                case SelectMode.SelectError: inEvent = Interop.Sys.PollEvents.POLLPRI; break;
+            }
 
-                int socketCount = 0;
-                Interop.Error err = Interop.Sys.Select(fdCount, readFds, writeFds, errorFds, microseconds, &socketCount);
-                if (err != Interop.Error.SUCCESS)
-                {
-                    status = false;
-                    return GetSocketErrorForErrorCode(err);
-                }
+            int milliseconds = microseconds == -1 ? -1 : microseconds / 1000;
 
-                status = Interop.Sys.FD_ISSET(fd, fdSet);
+            Interop.Sys.PollEvents outEvents;
+            Interop.Error err = Interop.Sys.Poll(handle, inEvent, milliseconds, out outEvents);
+            if (err != Interop.Error.SUCCESS)
+            {
+                status = false;
+                return GetSocketErrorForErrorCode(err);
             }
-            finally
+
+            switch (mode)
             {
-                if (releaseHandle)
-                {
-                    handle.DangerousRelease();
-                }
+                case SelectMode.SelectRead: status = (outEvents & (Interop.Sys.PollEvents.POLLIN | Interop.Sys.PollEvents.POLLHUP)) != 0; break;
+                case SelectMode.SelectWrite: status = (outEvents & Interop.Sys.PollEvents.POLLOUT) != 0; break;
+                case SelectMode.SelectError: status = (outEvents & (Interop.Sys.PollEvents.POLLERR | Interop.Sys.PollEvents.POLLPRI)) != 0; break;
+                default: status = false; break;
             }
             return SocketError.Success;
         }

src/System.Net.Sockets/tests/FunctionalTests/SendReceive.cs

@@ -20,7 +20,15 @@ public SendReceive(ITestOutputHelper output)
             _log = output;
         }
 
-        private static void SendToRecvFromAsync_Datagram_UDP(IPAddress leftAddress, IPAddress rightAddress)
+        public static readonly object[][] SendToRecvFromAsync_Datagram_UDP_MemberData = new object[][]
+        {
+            new object[] { IPAddress.IPv6Loopback, IPAddress.IPv6Loopback },
+            new object[] { IPAddress.Loopback, IPAddress.Loopback }
+        };
+
+        [Theory]
+        [MemberData(nameof(SendToRecvFromAsync_Datagram_UDP_MemberData))]
+        public void SendToRecvFromAsync_Datagram_UDP(IPAddress leftAddress, IPAddress rightAddress)
         {
             const int DatagramSize = 256;
             const int DatagramsToSend = 256;
@@ -144,7 +152,17 @@ private static void SendToRecvFromAsync_Datagram_UDP(IPAddress leftAddress, IPAd
             }
         }
 
-        private static void SendRecvAsync_Stream_TCP(IPAddress listenAt, bool useMultipleBuffers)
+        public static readonly object[][] SendRecvAsync_Stream_TCP_MemberData = new object[][]
+        {
+            new object[] { IPAddress.IPv6Loopback, true },
+            new object[] { IPAddress.IPv6Loopback, false },
+            new object[] { IPAddress.Loopback, true },
+            new object[] { IPAddress.Loopback, false },
+        };
+
+        [Theory]
+        [MemberData(nameof(SendRecvAsync_Stream_TCP_MemberData))]
+        public void SendRecvAsync_Stream_TCP(IPAddress listenAt, bool useMultipleBuffers)
         {
             const int BytesToSend = 123456;
             const int ListenBacklog = 1;
@@ -329,7 +347,15 @@ private static void SendRecvAsync_Stream_TCP(IPAddress listenAt, bool useMultipl
             Assert.Equal(sentChecksum.Sum, receivedChecksum.Sum);
         }
 
-        private static void SendRecvAsync_TcpListener_TcpClient(IPAddress listenAt)
+        public static readonly object[][] SendRecvAsync_TcpListener_TcpClient_MemberData = new[]
+        {
+            new object[] { IPAddress.Loopback },
+            new object[] { IPAddress.IPv6Loopback },
+        };
+
+        [Theory]
+        [MemberData(nameof(SendRecvAsync_TcpListener_TcpClient_MemberData))]
+        public void SendRecvAsync_TcpListener_TcpClient(IPAddress listenAt)
         {
             const int BytesToSend = 123456;
             const int ListenBacklog = 1;
@@ -397,52 +423,94 @@ private static void SendRecvAsync_TcpListener_TcpClient(IPAddress listenAt)
             Assert.Equal(sentChecksum.Sum, receivedChecksum.Sum);
         }
 
-        [Fact]
-        public void SendToRecvFromAsync_Single_Datagram_UDP_IPv6()
+        public static readonly object[][] SendRecvPollSync_TcpListener_TcpClient_MemberData = new[]
         {
-            SendToRecvFromAsync_Datagram_UDP(IPAddress.IPv6Loopback, IPAddress.IPv6Loopback);
-        }
-
-        [Fact]
-        public void SendToRecvFromAsync_Single_Datagram_UDP_IPv4()
+            new object[] { IPAddress.Loopback, true },
+            new object[] { IPAddress.Loopback, false },
+            new object[] { IPAddress.IPv6Loopback, true },
+            new object[] { IPAddress.IPv6Loopback, false },
+        };
+
+        [Theory]
+        [MemberData(nameof(SendRecvPollSync_TcpListener_TcpClient_MemberData))]
+        public void SendRecvPollSync_TcpListener_Socket(IPAddress listenAt, bool pollBeforeOperation)
         {
-            SendToRecvFromAsync_Datagram_UDP(IPAddress.Loopback, IPAddress.Loopback);
-        }
+            const int BytesToSend = 123456;
+            const int ListenBacklog = 1;
+            const int TestTimeout = 30000;
 
-        [Fact]
-        public void SendRecvAsync_Multiple_Stream_TCP_IPv6()
-        {
-            SendRecvAsync_Stream_TCP(IPAddress.IPv6Loopback, useMultipleBuffers: true);
-        }
+            var listener = new TcpListener(listenAt, 0);
+            listener.Start(ListenBacklog);
+            try
+            {
+                int bytesReceived = 0;
+                var receivedChecksum = new Fletcher32();
+                Task serverTask = Task.Run(async () =>
+                {
+                    using (Socket remote = await listener.AcceptSocketAsync())
+                    {
+                        var recvBuffer = new byte[256];
+                        while (true)
+                        {
+                            if (pollBeforeOperation)
+                            {
+                                Assert.True(remote.Poll(-1, SelectMode.SelectRead), "Read poll before completion should have succeeded");
+                            }
+                            int received = remote.Receive(recvBuffer, 0, recvBuffer.Length, SocketFlags.None);
+                            if (received == 0)
+                            {
+                                Assert.True(remote.Poll(0, SelectMode.SelectRead), "Read poll after completion should have succeeded");
+                                break;
+                            }
 
-        [Fact]
-        public void SendRecvAsync_Single_Stream_TCP_IPv6()
-        {
-            SendRecvAsync_Stream_TCP(IPAddress.IPv6Loopback, useMultipleBuffers: false);
-        }
+                            bytesReceived += received;
+                            receivedChecksum.Add(recvBuffer, 0, received);
+                        }
+                    }
+                });
 
-        [Fact]
-        public void SendRecvAsync_TcpListener_TcpClient_IPv6()
-        {
-            SendRecvAsync_TcpListener_TcpClient(IPAddress.IPv6Loopback);
-        }
+                int bytesSent = 0;
+                var sentChecksum = new Fletcher32();
+                Task clientTask = Task.Run(async () =>
+                {
+                    var clientEndpoint = (IPEndPoint)listener.LocalEndpoint;
 
-        [Fact]
-        public void SendRecvAsync_Multiple_Stream_TCP_IPv4()
-        {
-            SendRecvAsync_Stream_TCP(IPAddress.Loopback, useMultipleBuffers: true);
-        }
+                    using (var client = new Socket(clientEndpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp))
+                    {
+                        await client.ConnectAsync(clientEndpoint.Address, clientEndpoint.Port);
 
-        [Fact]
-        public void SendRecvAsync_Single_Stream_TCP_IPv4()
-        {
-            SendRecvAsync_Stream_TCP(IPAddress.Loopback, useMultipleBuffers: false);
-        }
+                        if (pollBeforeOperation)
+                        {
+                            Assert.False(client.Poll(TestTimeout, SelectMode.SelectRead), "Expected writer's read poll to fail after timeout");
+                        }
 
-        [Fact]
-        public void SendRecvAsync_TcpListener_TcpClient_IPv4()
-        {
-            SendRecvAsync_TcpListener_TcpClient(IPAddress.Loopback);
+                        var random = new Random();
+                        var sendBuffer = new byte[512];
+                        for (int remaining = BytesToSend, sent = 0; remaining > 0; remaining -= sent)
+                        {
+                            random.NextBytes(sendBuffer);
+
+                            if (pollBeforeOperation)
+                            {
+                                Assert.True(client.Poll(-1, SelectMode.SelectWrite), "Write poll should have succeeded");
+                            }
+                            sent = client.Send(sendBuffer, 0, Math.Min(sendBuffer.Length, remaining), SocketFlags.None);
+
+                            bytesSent += sent;
+                            sentChecksum.Add(sendBuffer, 0, sent);
+                        }
+                    }
+                });
+
+                Assert.True(Task.WaitAll(new[] { serverTask, clientTask }, TestTimeout), "Wait timed out");
+
+                Assert.Equal(bytesSent, bytesReceived);
+                Assert.Equal(sentChecksum.Sum, receivedChecksum.Sum);
+            }
+            finally
+            {
+                listener.Stop();
+            }
         }
     }
 }