Index: /trunk/include/iprt/mangling.h =================================================================== --- /trunk/include/iprt/mangling.h (revision 70480) +++ /trunk/include/iprt/mangling.h (revision 70481) @@ -2114,4 +2114,5 @@ # 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) Index: /trunk/include/iprt/tcp.h =================================================================== --- /trunk/include/iprt/tcp.h (revision 70480) +++ /trunk/include/iprt/tcp.h (revision 70481) @@ -231,4 +231,15 @@ /** + * 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. * Index: /trunk/src/VBox/Runtime/include/internal/socket.h =================================================================== --- /trunk/src/VBox/Runtime/include/internal/socket.h (revision 70480) +++ /trunk/src/VBox/Runtime/include/internal/socket.h (revision 70481) @@ -55,4 +55,5 @@ 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); Index: /trunk/src/VBox/Runtime/r3/socket.cpp =================================================================== --- /trunk/src/VBox/Runtime/r3/socket.cpp (revision 70480) +++ /trunk/src/VBox/Runtime/r3/socket.cpp (revision 70481) @@ -165,4 +165,16 @@ /** 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,4 +201,13 @@ static uint32_t volatile g_uWinSockInitedVersion = 0; #endif + + +/********************************************************************************************************************************* +* Internal Functions * +*********************************************************************************************************************************/ +#ifdef RT_OS_WINDOWS +static void rtSocketPokePollFallbackThread(RTSOCKETINT *pThis); +#endif + @@ -480,8 +501,18 @@ #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,4 +581,164 @@ +/** + * 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,4 +771,19 @@ 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,5 +812,5 @@ #ifdef RT_OS_WINDOWS /* - * Close the event. + * Windows specific polling cleanup. */ WSAEVENT hEvent = pThis->hEvent; @@ -612,7 +818,34 @@ { 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,16 +1932,15 @@ -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,4 +2019,20 @@ #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,5 +2522,11 @@ } 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,4 +2542,157 @@ /** + * 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,33 +2705,45 @@ 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,15 +2760,15 @@ 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,10 +2776,15 @@ 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,7 +2813,8 @@ 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,16 +2845,48 @@ 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,5 +2901,5 @@ fRetEvents &= fEvents; } -#endif /* RT_OS_OS2 */ +# endif /* RT_OS_OS2 */ LogFlowFunc(("fRetEvents=%#x\n", fRetEvents)); @@ -2524,4 +2981,12 @@ 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,5 +3019,20 @@ 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 Index: /trunk/src/VBox/Runtime/testcase/tstRTPoll.cpp =================================================================== --- /trunk/src/VBox/Runtime/testcase/tstRTPoll.cpp (revision 70480) +++ /trunk/src/VBox/Runtime/testcase/tstRTPoll.cpp (revision 70481) @@ -33,8 +33,202 @@ #include #include +#include #include #include +#include #include +#include #include + + +/********************************************************************************************************************************* +* 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,7 +293,4 @@ RTTESTI_CHECK_RC_RETV(RTPollSetDestroy(hSet), VINF_SUCCESS); - - - } @@ -425,4 +616,6 @@ RTAssertSetQuiet(fQuiet); RTAssertSetMayPanic(fMayPanic); + + tstRTPoll3(); }