Changeset 70481 in vbox

Timestamp:

Jan 7, 2018 6:46:08 PM (7 years ago)

Author:

vboxsync

svn:sync-xref-src-repo-rev:

120093

Message:

iprt/socket,tcp,pollset: Added RTTcpCreatePair. Implemented polling fallback for winsock 1.x. Extended tstRTPoll to cover sockets and actual waiting and receiving of events on pipe & socket events.

Location:

trunk

Files:

• include/iprt/mangling.h (modified) (1 diff)
• include/iprt/tcp.h (modified) (1 diff)
• src/VBox/Runtime/include/internal/socket.h (modified) (1 diff)
• src/VBox/Runtime/r3/socket.cpp (modified) (19 diffs)
• src/VBox/Runtime/testcase/tstRTPoll.cpp (modified) (3 diffs)

trunk/include/iprt/mangling.h

@@ -2114 +2114 @@
 # define RTTcpClientConnect                             RT_MANGLER(RTTcpClientConnect)
 # define RTTcpClientConnectEx                           RT_MANGLER(RTTcpClientConnectEx)
+# define RTTcpCreatePair                                RT_MANGLER(RTTcpCreatePair)
 # define RTTcpFlush                                     RT_MANGLER(RTTcpFlush)
 # define RTTcpGetLocalAddress                           RT_MANGLER(RTTcpGetLocalAddress)

trunk/include/iprt/tcp.h

@@ -231 +231 @@
 
 /**
+ * Creates connected pair of TCP sockets.
+ *
+ * @returns IPRT status code.
+ * @param   phServer            Where to return the "server" side of the pair.
+ * @param   phClient            Where to return the "client" side of the pair.
+ *
+ * @note    There is no server or client side, but we gotta call it something.
+ */
+RTR3DECL(int) RTTcpCreatePair(PRTSOCKET phServer, PRTSOCKET phClient, uint32_t fFlags);
+
+/**
  * Receive data from a socket.
  *

trunk/src/VBox/Runtime/include/internal/socket.h

@@ -55 +55 @@
 DECLHIDDEN(int) rtSocketCreateForNative(RTSOCKETINT **ppSocket, RTSOCKETNATIVE hNative);
 DECLHIDDEN(int) rtSocketCreate(PRTSOCKET phSocket, int iDomain, int iType, int iProtocol);
+DECLHIDDEN(int) rtSocketCreateTcpPair(RTSOCKET *phServer, RTSOCKET *phClient);
 DECLHIDDEN(int) rtSocketBind(RTSOCKET hSocket, PCRTNETADDR pAddr);
 DECLHIDDEN(int) rtSocketBindRawAddr(RTSOCKET hSocket, void const *pvAddr, size_t cbAddr);

trunk/src/VBox/Runtime/r3/socket.cpp

@@ -165 +165 @@
     /** Saved events which are only posted once. */
     uint32_t            fEventsSaved;
+    /** Set if we're using the polling fallback. */
+    bool                fPollFallback;
+    /** Set if the fallback polling is active (event not set). */
+    bool volatile       fPollFallbackActive;
+    /** Set to shut down the fallback polling thread. */
+    bool volatile       fPollFallbackShutdown;
+    /** Socket use to wake up the select thread. */
+    RTSOCKETNATIVE      hPollFallbackNotifyW;
+    /** Socket the select thread always waits on. */
+    RTSOCKETNATIVE      hPollFallbackNotifyR;
+    /** The fallback polling thread. */
+    RTTHREAD            hPollFallbackThread;
 #endif /* RT_OS_WINDOWS */
 } RTSOCKETINT;
@@ -189 +201 @@
 static uint32_t volatile g_uWinSockInitedVersion = 0;
 #endif
+
+
+/*********************************************************************************************************************************
+*   Internal Functions                                                                                                           *
+*********************************************************************************************************************************/
+#ifdef RT_OS_WINDOWS
+static void rtSocketPokePollFallbackThread(RTSOCKETINT *pThis);
+#endif
+
 
 
@@ -480 +501 @@
 #endif
 #ifdef RT_OS_WINDOWS
-    pThis->hEvent           = WSA_INVALID_EVENT;
-    pThis->fPollEvts        = 0;
-    pThis->fSubscribedEvts  = 0;
-    pThis->fEventsSaved     = 0;
+    pThis->hEvent                   = WSA_INVALID_EVENT;
+    pThis->fPollEvts                = 0;
+    pThis->fSubscribedEvts          = 0;
+    pThis->fEventsSaved             = 0;
+    pThis->fPollFallback            = g_uWinSockInitedVersion < MAKEWORD(2, 0)
+                                   || g_pfnWSACreateEvent == NULL
+                                   || g_pfnWSACloseEvent == NULL
+                                   || g_pfnWSAEventSelect == NULL
+                                   || g_pfnWSAEnumNetworkEvents == NULL;
+    pThis->fPollFallbackActive      = false;
+    pThis->fPollFallbackShutdown    = false;
+    pThis->hPollFallbackNotifyR     = NIL_RTSOCKETNATIVE;
+    pThis->hPollFallbackNotifyW     = NIL_RTSOCKETNATIVE;
+    pThis->hPollFallbackThread      = NIL_RTTHREAD;
 #endif
     *ppSocket = pThis;
@@ -550 +581 @@
 
 
+/**
+ * Wrapper around socketpair() for creating a local TCP connection.
+ *
+ * @returns IPRT status code.
+ * @param   phServer            Where to return the first native socket.
+ * @param   phClient            Where to return the second native socket.
+ */
+static int rtSocketCreateNativeTcpPair(RTSOCKETNATIVE *phServer, RTSOCKETNATIVE *phClient)
+{
+#ifdef RT_OS_WINDOWS
+    /*
+     * Initialize WinSock and make sure we got the necessary APIs.
+     */
+    int rc = rtSocketInitWinsock();
+    if (RT_FAILURE(rc))
+        return rc;
+    AssertReturn(g_pfnsocket, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfnclosesocket, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfnsetsockopt, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfnbind, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfngetsockname, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfnlisten, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfnaccept, VERR_NET_NOT_UNSUPPORTED);
+    AssertReturn(g_pfnconnect, VERR_NET_NOT_UNSUPPORTED);
+
+    /*
+     * Create the "server" listen socket and the "client" socket.
+     */
+    RTSOCKETNATIVE hListener = g_pfnsocket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+    if (hListener == NIL_RTSOCKETNATIVE)
+        return rtSocketError();
+    RTSOCKETNATIVE hClient = g_pfnsocket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+    if (hClient != NIL_RTSOCKETNATIVE)
+    {
+
+        /*
+         * We let WinSock choose a port number when we bind.
+         */
+        union
+        {
+            struct sockaddr_in  Ip;
+            struct sockaddr     Generic;
+        } uAddr;
+        RT_ZERO(uAddr);
+        uAddr.Ip.sin_family      = AF_INET;
+        uAddr.Ip.sin_addr.s_addr = RT_H2N_U32_C(INADDR_LOOPBACK);
+        //uAddr.Ip.sin_port      = 0;
+        int fReuse = 1;
+        rc = g_pfnsetsockopt(hListener, SOL_SOCKET, SO_REUSEADDR, (const char *)&fReuse, sizeof(fReuse));
+        if (rc == 0)
+        {
+            rc = g_pfnbind(hListener, &uAddr.Generic, sizeof(uAddr.Ip));
+            if (rc == 0)
+            {
+                /*
+                 * Get the address the client should connect to.  According to the docs,
+                 * we cannot assume that getsockname sets the IP and family.
+                 */
+                RT_ZERO(uAddr);
+                int cbAddr = sizeof(uAddr.Ip);
+                rc = g_pfngetsockname(hListener, &uAddr.Generic, &cbAddr);
+                if (rc == 0)
+                {
+                    uAddr.Ip.sin_family      = AF_INET;
+                    uAddr.Ip.sin_addr.s_addr = RT_H2N_U32_C(INADDR_LOOPBACK);
+
+                    /*
+                     * Listen, connect and accept.
+                     */
+                    rc = g_pfnlisten(hListener, 1 /*cBacklog*/);
+                    if (rc == 0)
+                    {
+                        rc = g_pfnconnect(hClient, &uAddr.Generic, sizeof(uAddr.Ip));
+                        if (rc == 0)
+                        {
+                            RTSOCKETNATIVE hServer = g_pfnaccept(hListener, NULL, NULL);
+                            if (hServer != NIL_RTSOCKETNATIVE)
+                            {
+                                g_pfnclosesocket(hListener);
+
+                                /*
+                                 * Done!
+                                 */
+                                *phServer = hServer;
+                                *phClient = hClient;
+                                return VINF_SUCCESS;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        rc = rtSocketError();
+        g_pfnclosesocket(hClient);
+    }
+    else
+        rc = rtSocketError();
+    g_pfnclosesocket(hListener);
+    return rc;
+
+#else
+    /*
+     * Got socket pair, so use it.
+     */
+    int aSockets[2] = { -1, -1 };
+    int (socketpair(AF_INET, SOCK_STREAM, IPPROTO_TCP, aSockets) == 0)
+    {
+        *phServer = aSockets[0];
+        *phClient = aSockets[0];
+        return VINF_SUCCESS;
+    }
+    return rtSocketError();
+#endif
+}
+
+
+/**
+ * Worker for RTTcpCreatePair.
+ *
+ * @returns IPRT status code.
+ * @param   phServer            Where to return the "server" side of the pair.
+ * @param   phClient            Where to return the "client" side of the pair.
+ * @note    There is no server or client side, but we gotta call it something.
+ */
+DECLHIDDEN(int) rtSocketCreateTcpPair(RTSOCKET *phServer, RTSOCKET *phClient)
+{
+    RTSOCKETNATIVE hServer = NIL_RTSOCKETNATIVE;
+    RTSOCKETNATIVE hClient = NIL_RTSOCKETNATIVE;
+    int rc = rtSocketCreateNativeTcpPair(&hServer, &hClient);
+    if (RT_SUCCESS(rc))
+    {
+        rc = rtSocketCreateForNative(phServer, hServer);
+        if (RT_SUCCESS(rc))
+        {
+            rc = rtSocketCreateForNative(phClient, hClient);
+            if (RT_SUCCESS(rc))
+                return VINF_SUCCESS;
+            RTSocketRelease(*phServer);
+        }
+        else
+        {
+#ifdef RT_OS_WINDOWS
+            g_pfnclosesocket(hServer);
+#else
+            close(hServer);
+#endif
+        }
+#ifdef RT_OS_WINDOWS
+        g_pfnclosesocket(hClient);
+#else
+        close(hClient);
+#endif
+    }
+
+    *phServer = NIL_RTSOCKET;
+    *phClient = NIL_RTSOCKET;
+    return rc;
+}
+
+
 RTDECL(uint32_t) RTSocketRetain(RTSOCKET hSocket)
 {
@@ -580 +771 @@
     if (ASMAtomicCmpXchgBool(&pThis->fClosed, true, false))
     {
+#ifdef RT_OS_WINDOWS
+        /*
+         * Poke the polling thread if active and give it a small chance to stop.
+         */
+        if (   pThis->fPollFallback
+            && pThis->hPollFallbackThread != NIL_RTTHREAD)
+        {
+            ASMAtomicWriteBool(&pThis->fPollFallbackShutdown, true);
+            rtSocketPokePollFallbackThread(pThis);
+            int rc2 = RTThreadWait(pThis->hPollFallbackThread, RT_MS_1SEC, NULL);
+            if (RT_SUCCESS(rc2))
+                pThis->hPollFallbackThread = NIL_RTTHREAD;
+        }
+#endif
+
         /*
          * Close the native handle.
@@ -606 +812 @@
 #ifdef RT_OS_WINDOWS
         /*
-         * Close the event.
+         * Windows specific polling cleanup.
          */
         WSAEVENT hEvent = pThis->hEvent;
@@ -612 +818 @@
         {
             pThis->hEvent = WSA_INVALID_EVENT;
-            Assert(g_pfnWSACloseEvent);
-            if (g_pfnWSACloseEvent)
-                g_pfnWSACloseEvent(hEvent);
+            if (!pThis->fPollFallback)
+            {
+                Assert(g_pfnWSACloseEvent);
+                if (g_pfnWSACloseEvent)
+                    g_pfnWSACloseEvent(hEvent);
+            }
+            else
+                CloseHandle(hEvent);
+        }
+
+        if (pThis->fPollFallback)
+        {
+            if (pThis->hPollFallbackNotifyW != NIL_RTSOCKETNATIVE)
+            {
+                g_pfnclosesocket(pThis->hPollFallbackNotifyW);
+                pThis->hPollFallbackNotifyW = NIL_RTSOCKETNATIVE;
+            }
+
+            if (pThis->hPollFallbackThread != NIL_RTTHREAD)
+            {
+                int rc2 = RTThreadWait(pThis->hPollFallbackThread, RT_MS_1MIN / 2, NULL);
+                AssertRC(rc2);
+                pThis->hPollFallbackThread = NIL_RTTHREAD;
+            }
+
+            if (pThis->hPollFallbackNotifyR != NIL_RTSOCKETNATIVE)
+            {
+                g_pfnclosesocket(pThis->hPollFallbackNotifyR);
+                pThis->hPollFallbackNotifyR = NIL_RTSOCKETNATIVE;
+            }
         }
 #endif
@@ -1699 +1932 @@
 
 
-RTDECL(int) RTSocketSelectOneEx(RTSOCKET hSocket, uint32_t fEvents, uint32_t *pfEvents, RTMSINTERVAL cMillies)
-{
-    /*
-     * Validate input.
-     */
-    RTSOCKETINT *pThis = hSocket;
-    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
-    AssertReturn(pThis->u32Magic == RTSOCKET_MAGIC, VERR_INVALID_HANDLE);
-    AssertPtrReturn(pfEvents, VERR_INVALID_PARAMETER);
-    AssertReturn(!(fEvents & ~RTSOCKET_EVT_VALID_MASK), VERR_INVALID_PARAMETER);
-    AssertReturn(RTMemPoolRefCount(pThis) >= (pThis->cUsers ? 2U : 1U), VERR_CALLER_NO_REFERENCE);
-
+/**
+ * Internal worker for RTSocketSelectOneEx and rtSocketPollCheck (fallback)
+ *
+ * @returns IPRT status code
+ * @param   pThis               The socket (valid).
+ * @param   fEvents             The events to select for.
+ * @param   pfEvents            Where to return the events.
+ * @param   cMillies            How long to select for, in milliseconds.
+ */
+static int rtSocketSelectOneEx(RTSOCKET pThis, uint32_t fEvents, uint32_t *pfEvents, RTMSINTERVAL cMillies)
+{
     RTSOCKETNATIVE hNative = pThis->hNative;
     if (hNative == NIL_RTSOCKETNATIVE)
@@ -1787 +2019 @@
 #endif
     return rc;
+}
+
+
+RTDECL(int) RTSocketSelectOneEx(RTSOCKET hSocket, uint32_t fEvents, uint32_t *pfEvents, RTMSINTERVAL cMillies)
+{
+    /*
+     * Validate input.
+     */
+    RTSOCKETINT *pThis = hSocket;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTSOCKET_MAGIC, VERR_INVALID_HANDLE);
+    AssertPtrReturn(pfEvents, VERR_INVALID_PARAMETER);
+    AssertReturn(!(fEvents & ~RTSOCKET_EVT_VALID_MASK), VERR_INVALID_PARAMETER);
+    AssertReturn(RTMemPoolRefCount(pThis) >= (pThis->cUsers ? 2U : 1U), VERR_CALLER_NO_REFERENCE);
+
+    return rtSocketSelectOneEx(pThis, fEvents, pfEvents, cMillies);
 }
 
@@ -2274 +2522 @@
     }
     else
-        rc = VERR_NET_NOT_UNSUPPORTED;
+    {
+        AssertCompile(WSA_INVALID_EVENT == (WSAEVENT)NULL);
+        pThis->hEvent = CreateEventW(NULL, TRUE /*fManualReset*/, FALSE /*fInitialState*/,  NULL /*pwszName*/);
+        *phNative = (RTHCINTPTR)pThis->hEvent;
+        if (pThis->hEvent == WSA_INVALID_EVENT)
+            rc = RTErrConvertFromWin32(GetLastError());
+    }
 
     rtSocketUnlock(pThis);
@@ -2288 +2542 @@
 
 /**
+ * Fallback poller thread.
+ *
+ * @returns VINF_SUCCESS.
+ * @param   hSelf               The thread handle.
+ * @param   pvUser              Socket instance data.
+ */
+static DECLCALLBACK(int) rtSocketPollFallbackThreadProc(RTTHREAD hSelf, void *pvUser)
+{
+    RTSOCKETINT *pThis = (RTSOCKETINT *)pvUser;
+    RT_NOREF(hSelf);
+# define __WSAFDIsSet g_pfn__WSAFDIsSet
+
+    /*
+     * The execution loop.
+     */
+    while (!ASMAtomicReadBool(&pThis->fPollFallbackShutdown))
+    {
+        /*
+         * Do the selecting (with a 15 second timeout because that seems like a good idea).
+         */
+        struct fd_set SetRead;
+        struct fd_set SetWrite;
+        struct fd_set SetXcpt;
+
+        FD_ZERO(&SetRead);
+        FD_ZERO(&SetWrite);
+        FD_ZERO(&SetXcpt);
+
+        FD_SET(pThis->hPollFallbackNotifyR, &SetRead);
+        FD_SET(pThis->hPollFallbackNotifyR, &SetXcpt);
+
+        bool     fActive = ASMAtomicReadBool(&pThis->fPollFallbackActive);
+        uint32_t fEvents;
+        if (!fActive)
+            fEvents = 0;
+        else
+        {
+            fEvents = ASMAtomicReadU32(&pThis->fSubscribedEvts);
+            if (fEvents & RTPOLL_EVT_READ)
+                FD_SET(pThis->hNative, &SetRead);
+            if (fEvents & RTPOLL_EVT_WRITE)
+                FD_SET(pThis->hNative, &SetWrite);
+            if (fEvents & RTPOLL_EVT_ERROR)
+                FD_SET(pThis->hNative, &SetXcpt);
+        }
+
+        struct timeval Timeout;
+        Timeout.tv_sec  = 15;
+        Timeout.tv_usec = 0;
+        int rc = g_pfnselect(INT_MAX /*ignored*/, &SetRead, &SetWrite, &SetXcpt, &Timeout);
+
+        /* Stop immediately if told to shut down. */
+        if (ASMAtomicReadBool(&pThis->fPollFallbackShutdown))
+            break;
+
+        /*
+         * Process the result.
+         */
+        if (rc > 0)
+        {
+            /* First the socket we're listening on. */
+            if (   fEvents
+                && (   FD_ISSET(pThis->hNative, &SetRead)
+                    || FD_ISSET(pThis->hNative, &SetWrite)
+                    || FD_ISSET(pThis->hNative, &SetXcpt)) )
+            {
+                ASMAtomicWriteBool(&pThis->fPollFallbackActive, false);
+                SetEvent(pThis->hEvent);
+            }
+
+            /* Then maintain the notification pipe.  (We only read one byte here
+               because we're overly paranoid wrt socket switching to blocking mode.) */
+            if (FD_ISSET(pThis->hPollFallbackNotifyR, &SetRead))
+            {
+                char chIgnored;
+                g_pfnrecv(pThis->hPollFallbackNotifyR, &chIgnored, sizeof(chIgnored), MSG_NOSIGNAL);
+            }
+        }
+        else
+            AssertMsg(rc == 0, ("%Rrc\n", rtSocketError()));
+    }
+
+# undef __WSAFDIsSet
+    return VINF_SUCCESS;
+}
+
+
+/**
+ * Pokes the fallback thread, making sure it gets out of whatever it's stuck in.
+ *
+ * @param   pThis               The socket handle.
+ */
+static void rtSocketPokePollFallbackThread(RTSOCKETINT *pThis)
+{
+    Assert(pThis->fPollFallback);
+    if (pThis->hPollFallbackThread != NIL_RTTHREAD)
+    {
+        int cbWritten = g_pfnsend(pThis->hPollFallbackNotifyW, "!", 1, MSG_NOSIGNAL);
+        AssertMsg(cbWritten == 1, ("cbWritten=%d err=%Rrc\n",  rtSocketError()));
+        RT_NOREF_PV(cbWritten);
+    }
+}
+
+
+/**
+ * Called by rtSocketPollStart to make the thread start selecting on the socket.
+ *
+ * @returns 0 on success, RTPOLL_EVT_ERROR on failure.
+ * @param   pThis               The socket handle.
+ */
+static uint32_t rtSocketPollFallbackStart(RTSOCKETINT *pThis)
+{
+    /*
+     * Reset the event and tell the thread to start selecting on the socket.
+     */
+    ResetEvent(pThis->hEvent);
+    ASMAtomicWriteBool(&pThis->fPollFallbackActive, true);
+
+    /*
+     * Wake up the thread the thread.
+     */
+    if (pThis->hPollFallbackThread != NIL_RTTHREAD)
+        rtSocketPokePollFallbackThread(pThis);
+    else
+    {
+        /*
+         * Not running, need to set it up and start it.
+         */
+        AssertLogRelReturn(pThis->hEvent != NULL && pThis->hEvent != INVALID_HANDLE_VALUE, RTPOLL_EVT_ERROR);
+
+        /* Create the notification socket pair. */
+        int rc;
+        if (pThis->hPollFallbackNotifyR == NIL_RTSOCKETNATIVE)
+        {
+            rc = rtSocketCreateNativeTcpPair(&pThis->hPollFallbackNotifyW, &pThis->hPollFallbackNotifyR);
+            AssertLogRelRCReturn(rc, RTPOLL_EVT_ERROR);
+
+            /* Make the read end non-blocking (not fatal). */
+            u_long fNonBlocking = 1;
+            rc = g_pfnioctlsocket(pThis->hPollFallbackNotifyR, FIONBIO, &fNonBlocking);
+            AssertLogRelMsg(rc == 0,  ("rc=%#x %Rrc\n", rc, rtSocketError()));
+        }
+
+        /* Finally, start the thread.  ASSUME we don't need too much stack. */
+        rc = RTThreadCreate(&pThis->hPollFallbackThread, rtSocketPollFallbackThreadProc, pThis,
+                            _128K, RTTHREADTYPE_IO, RTTHREADFLAGS_WAITABLE, "sockpoll");
+        AssertLogRelRCReturn(rc, RTPOLL_EVT_ERROR);
+    }
+    return 0;
+}
+
+
+/**
  * Undos the harm done by WSAEventSelect.
  *
@@ -2298 +2705 @@
     if (pThis->fSubscribedEvts)
     {
-        if (g_pfnWSAEventSelect && g_pfnioctlsocket)
-        {
-            if (g_pfnWSAEventSelect(pThis->hNative, WSA_INVALID_EVENT, 0) == 0)
+        if (!pThis->fPollFallback)
+        {
+            Assert(g_pfnWSAEventSelect && g_pfnioctlsocket);
+            if (g_pfnWSAEventSelect && g_pfnioctlsocket)
             {
-                pThis->fSubscribedEvts = 0;
-
-                /*
-                 * Switch back to blocking mode if that was the state before the
-                 * operation.
-                 */
-                if (pThis->fBlocking)
+                if (g_pfnWSAEventSelect(pThis->hNative, WSA_INVALID_EVENT, 0) == 0)
                 {
-                    u_long fNonBlocking = 0;
-                    int rc2 = g_pfnioctlsocket(pThis->hNative, FIONBIO, &fNonBlocking);
-                    if (rc2 != 0)
+                    pThis->fSubscribedEvts = 0;
+
+                    /*
+                     * Switch back to blocking mode if that was the state before the
+                     * operation.
+                     */
+                    if (pThis->fBlocking)
                     {
-                        rc = rtSocketError();
-                        AssertMsgFailed(("%Rrc; rc2=%d\n", rc, rc2));
+                        u_long fNonBlocking = 0;
+                        int rc2 = g_pfnioctlsocket(pThis->hNative, FIONBIO, &fNonBlocking);
+                        if (rc2 != 0)
+                        {
+                            rc = rtSocketError();
+                            AssertMsgFailed(("%Rrc; rc2=%d\n", rc, rc2));
+                        }
                     }
+                }
+                else
+                {
+                    rc = rtSocketError();
+                    AssertMsgFailed(("%Rrc\n", rc));
                 }
             }
             else
             {
-                rc = rtSocketError();
-                AssertMsgFailed(("%Rrc\n", rc));
+                Assert(pThis->fPollFallback);
+                rc = VINF_SUCCESS;
             }
         }
+        /*
+         * Just clear the event mask as we never started waiting if we get here.
+         */
         else
-            rc = VERR_NET_NOT_UNSUPPORTED;
+            ASMAtomicWriteU32(&pThis->fSubscribedEvts, 0);
     }
     return rc;
@@ -2341 +2760 @@
 static int rtSocketPollUpdateEvents(RTSOCKETINT *pThis, uint32_t fEvents)
 {
-    LONG fNetworkEvents = 0;
-    if (fEvents & RTPOLL_EVT_READ)
-        fNetworkEvents |= FD_READ;
-    if (fEvents & RTPOLL_EVT_WRITE)
-        fNetworkEvents |= FD_WRITE;
-    if (fEvents & RTPOLL_EVT_ERROR)
-        fNetworkEvents |= FD_CLOSE;
-    LogFlowFunc(("fNetworkEvents=%#x\n", fNetworkEvents));
-    int rc;
-    if (g_pfnWSAEventSelect)
-    {
+    if (!pThis->fPollFallback)
+    {
+        LONG fNetworkEvents = 0;
+        if (fEvents & RTPOLL_EVT_READ)
+            fNetworkEvents |= FD_READ;
+        if (fEvents & RTPOLL_EVT_WRITE)
+            fNetworkEvents |= FD_WRITE;
+        if (fEvents & RTPOLL_EVT_ERROR)
+            fNetworkEvents |= FD_CLOSE;
+        LogFlowFunc(("fNetworkEvents=%#x\n", fNetworkEvents));
+
         if (g_pfnWSAEventSelect(pThis->hNative, pThis->hEvent, fNetworkEvents) == 0)
         {
@@ -2357 +2776 @@
             return VINF_SUCCESS;
         }
-        rc = rtSocketError();
+
+        int rc = rtSocketError();
         AssertMsgFailed(("fNetworkEvents=%#x rc=%Rrc\n", fNetworkEvents, rtSocketError()));
-    }
-    else
-        rc = VERR_NET_NOT_UNSUPPORTED;
-    return rc;
+        return rc;
+    }
+
+    /*
+     * Update the events we're waiting for.  Caller will poke/start the thread. later
+     */
+    ASMAtomicWriteU32(&pThis->fSubscribedEvts, fEvents);
+    return VINF_SUCCESS;
 }
 
@@ -2389 +2813 @@
         rc = rtSocketPollUpdateEvents(pThis, pThis->fSubscribedEvts | fEvents);
 
-    /* Get the event mask, ASSUMES that WSAEnumNetworkEvents doesn't clear stuff.  */
-    if (g_pfnWSAEnumNetworkEvents)
-    {
+    if (!pThis->fPollFallback)
+    {
+        /* Atomically get pending events and reset the event semaphore. */
+        Assert(g_pfnWSAEnumNetworkEvents);
         WSANETWORKEVENTS NetEvts;
         RT_ZERO(NetEvts);
@@ -2420 +2845 @@
             rc = rtSocketError();
     }
-    else if (RT_SUCCESS(rc))
-        rc = VERR_NET_NOT_UNSUPPORTED;
-
-    /* Fall back on select if we hit an error above. */
-    if (RT_FAILURE(rc))
-    {
-        rc = RTSocketSelectOneEx(pThis, fEvents, &fRetEvents, 0);
-        if (RT_FAILURE(rc))
+
+    /* Fall back on select if we hit an error above or is using fallback polling. */
+    if (pThis->fPollFallback || RT_FAILURE(rc))
+    {
+        rc = rtSocketSelectOneEx(pThis, fEvents & RTPOLL_EVT_ERROR ? fEvents | RTPOLL_EVT_READ : fEvents, &fRetEvents, 0);
+        if (RT_SUCCESS(rc))
+        {
+            /* rtSocketSelectOneEx may return RTPOLL_EVT_READ on disconnect.  Use
+               getpeername to fix this. */
+            if ((fRetEvents & (RTPOLL_EVT_READ | RTPOLL_EVT_ERROR)) == RTPOLL_EVT_READ)
+            {
+# if 0 /* doens't work */
+                rtSocketErrorReset();
+                char chIgn;
+                rc = g_pfnrecv(pThis->hNative, &chIgn, 0, MSG_NOSIGNAL);
+                rc = rtSocketError();
+                if (RT_FAILURE(rc))
+                    fRetEvents |= RTPOLL_EVT_ERROR;
+
+                rc = g_pfnsend(pThis->hNative, &chIgn, 0, MSG_NOSIGNAL);
+                rc = rtSocketError();
+                if (RT_FAILURE(rc))
+                    fRetEvents |= RTPOLL_EVT_ERROR;
+
+                RTSOCKADDRUNION u;
+                int cbAddr = sizeof(u);
+                if (g_pfngetpeername(pThis->hNative, &u.Addr, &cbAddr) == SOCKET_ERROR)
+                    fRetEvents |= RTPOLL_EVT_ERROR;
+# endif
+                /* If no bytes are available, assume error condition. */
+                u_long cbAvail = 0;
+                rc = ioctlsocket(pThis->hNative, FIONREAD, &cbAvail);
+                if (rc == 0 && cbAvail == 0)
+                    fRetEvents |= RTPOLL_EVT_ERROR;
+            }
+            fRetEvents &= fEvents | RTPOLL_EVT_ERROR;
+        }
+        else if (rc == VERR_TIMEOUT)
             fRetEvents = 0;
-    }
-
-#else  /* RT_OS_OS2 */
+        else
+            fRetEvents |= RTPOLL_EVT_ERROR;
+    }
+
+# else  /* RT_OS_OS2 */
     int aFds[4] = { pThis->hNative, pThis->hNative, pThis->hNative, -1 };
     int rc = os2_select(aFds, 1, 1, 1, 0);
@@ -2444 +2901 @@
         fRetEvents &= fEvents;
     }
-#endif /* RT_OS_OS2 */
+# endif /* RT_OS_OS2 */
 
     LogFlowFunc(("fRetEvents=%#x\n", fRetEvents));
@@ -2524 +2981 @@
         ASMAtomicDecU32(&pThis->cUsers);
     }
+# ifdef RT_OS_WINDOWS
+    /*
+     * Kick the poller thread on if this is the final entry and we're in
+     * winsock 1.x fallback mode.
+     */
+    else if (pThis->fPollFallback && fFinalEntry)
+        fRetEvents = rtSocketPollFallbackStart(pThis);
+# endif
 
     return fRetEvents;
@@ -2554 +3019 @@
     RT_NOREF_PV(fFinalEntry);
 
-    /* Harvest events and clear the event mask for the next round of polling. */
+# ifdef RT_OS_WINDOWS
+    /*
+     * Deactivate the poll thread if we're in winsock 1.x fallback poll mode.
+     */
+    if (   pThis->fPollFallback
+        && pThis->hPollFallbackThread != NIL_RTTHREAD)
+    {
+        ASMAtomicWriteU32(&pThis->fSubscribedEvts, 0);
+        if (ASMAtomicXchgBool(&pThis->fPollFallbackActive, false))
+            rtSocketPokePollFallbackThread(pThis);
+    }
+# endif
+
+    /*
+     * Harvest events and clear the event mask for the next round of polling.
+     */
     uint32_t fRetEvents = rtSocketPollCheck(pThis, fEvents);
 # ifdef RT_OS_WINDOWS

trunk/src/VBox/Runtime/testcase/tstRTPoll.cpp

@@ -33 +33 @@
 #include <iprt/err.h>
 #include <iprt/file.h>
+#include <iprt/log.h>
 #include <iprt/mem.h>
 #include <iprt/pipe.h>
+#include <iprt/socket.h>
 #include <iprt/string.h>
+#include <iprt/tcp.h>
 #include <iprt/test.h>
+
+
+/*********************************************************************************************************************************
+*   Global Variables                                                                                                             *
+*********************************************************************************************************************************/
+/** What we write from the threads in test 3. */
+static char g_szHello[] = "hello!";
+
+
+static DECLCALLBACK(int) tstRTPoll3PipeWriteThread(RTTHREAD hSelf, void *pvUser)
+{
+    RT_NOREF_PV(hSelf);
+    RTPIPE hPipe = (RTPIPE)pvUser;
+    RTThreadSleep(RT_MS_1SEC);
+    return RTPipeWriteBlocking(hPipe, g_szHello, sizeof(g_szHello) - 1, NULL);
+}
+
+
+static DECLCALLBACK(int) tstRTPoll3SockWriteThread(RTTHREAD hSelf, void *pvUser)
+{
+    RT_NOREF_PV(hSelf);
+    RTSOCKET hSocket = (RTSOCKET)pvUser;
+    RTThreadSleep(RT_MS_1SEC);
+    return RTTcpWrite(hSocket, g_szHello, sizeof(g_szHello) - 1);
+}
+
+
+static void tstRTPoll3(void)
+{
+    RTTestISub("Pipe & Sockets");
+
+    /*
+     * Create a set and a pair of pipes and a pair of sockets.
+     */
+    RTPOLLSET hSet = NIL_RTPOLLSET;
+    RTTESTI_CHECK_RC_RETV(RTPollSetCreate(&hSet), VINF_SUCCESS);
+    RTTESTI_CHECK_RETV(hSet != NIL_RTPOLLSET);
+
+    RTTESTI_CHECK_RETV(RTPollSetGetCount(hSet) == 0);
+    RTTESTI_CHECK_RC(RTPollSetQueryHandle(hSet, 0, NULL), VERR_POLL_HANDLE_ID_NOT_FOUND);
+
+    RTPIPE hPipeR;
+    RTPIPE hPipeW;
+    RTTESTI_CHECK_RC_RETV(RTPipeCreate(&hPipeR, &hPipeW, 0/*fFlags*/), VINF_SUCCESS);
+
+    RTSOCKET hSocketR;
+    RTSOCKET hSocketW;
+    RTTESTI_CHECK_RC_RETV(RTTcpCreatePair(&hSocketR, &hSocketW, 0/*fFlags*/), VINF_SUCCESS);
+
+    /*
+     * Add them for error checking.  These must be added first if want we their IDs
+     * to show up when disconnecting.
+     */
+    RTTESTI_CHECK_RC_RETV(RTPollSetAddPipe(hSet, hPipeR, RTPOLL_EVT_ERROR, 1 /*id*/), VINF_SUCCESS);
+    RTTESTI_CHECK_RC_RETV(RTPollSetAddSocket(hSet, hSocketR, RTPOLL_EVT_ERROR, 2 /*id*/), VINF_SUCCESS);
+    RTTESTI_CHECK_RETV(RTPollSetGetCount(hSet) == 2);
+
+    /*
+     * Add the read ends.  Polling should time out.
+     */
+    RTTESTI_CHECK_RC_RETV(RTPollSetAddPipe(hSet, hPipeR, RTPOLL_EVT_READ, 11 /*id*/), VINF_SUCCESS);
+    RTTESTI_CHECK_RC_RETV(RTPollSetAddSocket(hSet, hSocketR, RTPOLL_EVT_READ, 12 /*id*/), VINF_SUCCESS);
+
+    RTTESTI_CHECK_RETV(RTPollSetGetCount(hSet) == 4);
+
+    RTTESTI_CHECK_RC(RTPollSetQueryHandle(hSet, 11 /*id*/, NULL), VINF_SUCCESS);
+    RTHANDLE Handle;
+    RTTESTI_CHECK_RC_RETV(RTPollSetQueryHandle(hSet, 11 /*id*/, &Handle), VINF_SUCCESS);
+    RTTESTI_CHECK(Handle.enmType == RTHANDLETYPE_PIPE);
+    RTTESTI_CHECK(Handle.u.hPipe == hPipeR);
+
+    RTTESTI_CHECK_RC(RTPollSetQueryHandle(hSet, 12 /*id*/, NULL), VINF_SUCCESS);
+    RTTESTI_CHECK_RC_RETV(RTPollSetQueryHandle(hSet, 12 /*id*/, &Handle), VINF_SUCCESS);
+    RTTESTI_CHECK(Handle.enmType == RTHANDLETYPE_SOCKET);
+    RTTESTI_CHECK(Handle.u.hSocket == hSocketR);
+
+    RTTESTI_CHECK_RC(RTPoll(hSet, 0, NULL,  NULL), VERR_TIMEOUT);
+    RTTESTI_CHECK_RC(RTPoll(hSet, 1, NULL,  NULL), VERR_TIMEOUT);
+
+    /*
+     * Add the write ends.  Should indicate that the first one is ready for writing.
+     */
+    RTTESTI_CHECK_RC_RETV(RTPollSetAddPipe(hSet, hPipeW, RTPOLL_EVT_WRITE, 21 /*id*/), VINF_SUCCESS);
+    RTTESTI_CHECK_RC_RETV(RTPollSetAddSocket(hSet, hSocketW, RTPOLL_EVT_WRITE, 22 /*id*/), VINF_SUCCESS);
+
+    uint32_t idReady = UINT32_MAX;
+    RTTESTI_CHECK_RC(RTPoll(hSet, 0, NULL, &idReady), VINF_SUCCESS);
+    RTTESTI_CHECK(idReady == 21 || idReady == 22);
+
+    /*
+     * Remove the write ends again.
+     */
+    RTTESTI_CHECK_RC(RTPollSetRemove(hSet, 21), VINF_SUCCESS);
+    RTTESTI_CHECK_RC(RTPollSetRemove(hSet, 22), VINF_SUCCESS);
+    RTTESTI_CHECK_RC(RTPoll(hSet, 0, NULL,  NULL), VERR_TIMEOUT);
+
+    /*
+     * Kick off a thread that writes to the socket after 1 second.
+     * This will check that we can wait and wake up.
+     */
+    for (uint32_t i = 0; i < 2; i++)
+    {
+        RTTHREAD hThread;
+        RTTESTI_CHECK_RC(RTThreadCreate(&hThread, tstRTPoll3SockWriteThread, hSocketW, 0,
+                                        RTTHREADTYPE_DEFAULT, RTTHREADFLAGS_WAITABLE, "test3sock"), VINF_SUCCESS);
+
+        uint32_t fEvents = 0;
+        idReady = 0;
+        uint64_t msStart = RTTimeSystemMilliTS();
+        RTTESTI_CHECK_RC(RTPoll(hSet, 5 * RT_MS_1SEC, &fEvents, &idReady), VINF_SUCCESS);
+        uint32_t msElapsed = RTTimeSystemMilliTS() - msStart;
+        RTTESTI_CHECK_MSG(msElapsed >= 250 && msElapsed < 4500, ("msElapsed=%RU64\n", msElapsed));
+        RTTESTI_CHECK(fEvents == RTPOLL_EVT_READ);
+        RTTESTI_CHECK(idReady == 12);
+
+        RTThreadWait(hThread, 5 * RT_MS_1SEC, NULL);
+
+        /* Drain the socket. */
+        char    achBuf[128];
+        size_t  cbRead = 0;
+        RTTESTI_CHECK_RC(RTTcpReadNB(hSocketR, achBuf, sizeof(achBuf), &cbRead), VINF_SUCCESS);
+        RTTESTI_CHECK(cbRead == sizeof(g_szHello) - 1 && memcmp(achBuf, g_szHello, sizeof(g_szHello) - 1) == 0);
+
+        RTTESTI_CHECK_RC(RTPoll(hSet, 0, NULL,  NULL), VERR_TIMEOUT);
+        RTTESTI_CHECK_RC(RTPoll(hSet, 1, NULL,  NULL), VERR_TIMEOUT);
+    }
+
+    /*
+     * Kick off a thread that writes to the pipe after 1 second.
+     * This will check that we can wait and wake up.
+     */
+    for (uint32_t i = 0; i < 2; i++)
+    {
+        RTTHREAD hThread;
+        RTTESTI_CHECK_RC(RTThreadCreate(&hThread, tstRTPoll3PipeWriteThread, hPipeW, 0,
+                                        RTTHREADTYPE_DEFAULT, RTTHREADFLAGS_WAITABLE, "test3pipe"), VINF_SUCCESS);
+
+        uint32_t fEvents = 0;
+        idReady = 0;
+        uint64_t msStart = RTTimeSystemMilliTS();
+        RTTESTI_CHECK_RC(RTPoll(hSet, 5 * RT_MS_1SEC, &fEvents, &idReady), VINF_SUCCESS);
+        uint32_t msElapsed = RTTimeSystemMilliTS() - msStart;
+        RTTESTI_CHECK_MSG(msElapsed >= 250 && msElapsed < 4500, ("msElapsed=%RU64\n", msElapsed));
+        RTTESTI_CHECK(fEvents == RTPOLL_EVT_READ);
+        RTTESTI_CHECK(idReady == 11);
+
+        RTThreadWait(hThread, 5 * RT_MS_1SEC, NULL);
+
+        /* Drain the socket. */
+        char    achBuf[128];
+        size_t  cbRead = 0;
+        RTTESTI_CHECK_RC(RTPipeRead(hPipeR, achBuf, sizeof(achBuf), &cbRead), VINF_SUCCESS);
+        RTTESTI_CHECK(cbRead == sizeof(g_szHello) - 1 && memcmp(achBuf, g_szHello, sizeof(g_szHello) - 1) == 0);
+
+//        RTTESTI_CHECK_RC(RTPoll(hSet, 0, NULL,  NULL), VERR_TIMEOUT);
+//        RTTESTI_CHECK_RC(RTPoll(hSet, 1, NULL,  NULL), VERR_TIMEOUT);
+    }
+
+
+    /*
+     * Close the write socket, checking that we get error returns.
+     */
+    RTSocketShutdown(hSocketW, true, true);
+    RTSocketClose(hSocketW);
+
+    uint32_t fEvents = 0;
+    idReady = 0;
+    RTTESTI_CHECK_RC(RTPoll(hSet, 0, &fEvents, &idReady), VINF_SUCCESS);
+    RTTESTI_CHECK_MSG(idReady == 2 || idReady == 12, ("idReady=%u\n", idReady));
+    RTTESTI_CHECK_MSG(fEvents & RTPOLL_EVT_ERROR, ("fEvents=%#x\n", fEvents));
+
+    RTTESTI_CHECK_RC(RTPollSetRemove(hSet, 2), VINF_SUCCESS);
+    RTTESTI_CHECK_RC(RTPollSetRemove(hSet, 12), VINF_SUCCESS);
+
+    RTSocketClose(hSocketR);
+
+    /*
+     * Ditto for the pipe end.
+     */
+    RTPipeClose(hPipeW);
+
+    idReady = fEvents = 0;
+    RTTESTI_CHECK_RC(RTPoll(hSet, 0, &fEvents, &idReady), VINF_SUCCESS);
+    RTTESTI_CHECK_MSG(idReady == 1 || idReady == 11, ("idReady=%u\n", idReady));
+    RTTESTI_CHECK_MSG(fEvents & RTPOLL_EVT_ERROR, ("fEvents=%#x\n", fEvents));
+
+    RTPipeClose(hPipeR);
+
+    RTTESTI_CHECK_RC(RTPollSetDestroy(hSet), VINF_SUCCESS);
+RTLogFlush(NULL);
+}
 
 
@@ -99 +293 @@
 
     RTTESTI_CHECK_RC_RETV(RTPollSetDestroy(hSet), VINF_SUCCESS);
-
-
-
 }
 
@@ -425 +616 @@
         RTAssertSetQuiet(fQuiet);
         RTAssertSetMayPanic(fMayPanic);
+
+        tstRTPoll3();
     }