Changeset 47122 in vbox for trunk/src/VBox

Timestamp:

Jul 12, 2013 2:34:26 PM (12 years ago)

Author:

vboxsync

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

87216

Message:

IPRT/localipc-win.cpp: Update (waiting on data now is properly getting interrupted on server destruction), more testcases (work in progress).

Location:

trunk/src/VBox/Runtime

Files:

• r3/win/localipc-win.cpp (modified) (13 diffs)
• testcase/tstRTLocalIpc.cpp (modified) (10 diffs)

trunk/src/VBox/Runtime/r3/win/localipc-win.cpp

@@ -52 +52 @@
 #include <iprt/string.h>
 #include <iprt/thread.h>
+#include <iprt/time.h>
 
 #include "internal/magics.h"
@@ -148 +149 @@
     /** Set if there is already pending I/O. */
     bool                fIOPending;
+    /** Set if the zero byte read that the poll code using is pending. */
+    bool                fZeroByteRead;
     /** Indicates that there is a pending cancel request. */
     bool volatile       fCancelled;
@@ -162 +165 @@
     /** Amount of allocated buffer space. */
     size_t              cbBounceBufAlloc;
+    /** Buffer for the zero byte read.
+     *  Used in RTLocalIpcSessionWaitForData(). */
+    uint8_t             abBuf[8];
 } RTLOCALIPCSESSIONINT;
 /** Pointer to a local IPC session instance (Windows). */
@@ -414 +420 @@
     ASMAtomicUoWriteU32(&pThis->u32Magic, ~RTLOCALIPCSERVER_MAGIC);
     ASMAtomicUoWriteBool(&pThis->fCancelled, true);
+    Assert(pThis->cRefs);
     pThis->cRefs--;
 
-    if (pThis->cRefs > 0)
+    if (pThis->cRefs)
     {
         BOOL fRc = SetEvent(pThis->hEvent);
@@ -462 +469 @@
         SetLastError(NO_ERROR);
         BOOL fRc = ConnectNamedPipe(pThis->hNmPipe, &pThis->OverlappedIO);
-        DWORD err = fRc ? NO_ERROR : GetLastError();
+        DWORD dwErr = fRc ? NO_ERROR : GetLastError();
         if (    !fRc
-            &&  err == ERROR_IO_PENDING)
+            &&  dwErr == ERROR_IO_PENDING)
         {
             WaitForSingleObject(pThis->hEvent, INFINITE);
             DWORD dwIgnored;
             fRc = GetOverlappedResult(pThis->hNmPipe, &pThis->OverlappedIO, &dwIgnored, FALSE /* bWait*/);
-            err = fRc ? NO_ERROR : GetLastError();
+            dwErr = fRc ? NO_ERROR : GetLastError();
         }
 
@@ -484 +491 @@
              */
             if (   fRc
-                || err == ERROR_PIPE_CONNECTED)
+                || dwErr == ERROR_PIPE_CONNECTED)
             {
                 HANDLE hNmPipe;
@@ -505 +512 @@
             }
             else
-                rc = RTErrConvertFromWin32(err);
+                rc = RTErrConvertFromWin32(dwErr);
         }
         else
@@ -516 +523 @@
              */
             if (    fRc
-                ||  err == ERROR_PIPE_CONNECTED)
+                ||  dwErr == ERROR_PIPE_CONNECTED)
                 fRc = DisconnectNamedPipe(pThis->hNmPipe);
-            else if (err == ERROR_IO_PENDING)
+            else if (dwErr == ERROR_IO_PENDING)
                 fRc = CancelIo(pThis->hNmPipe);
             else
@@ -550 +557 @@
      * and signal the event (to wake up anyone in/at WaitForSingleObject).
      */
-    RTCritSectEnter(&pThis->CritSect);
-
-    ASMAtomicUoWriteBool(&pThis->fCancelled, true);
-    BOOL fRc = SetEvent(pThis->hEvent);
-    AssertMsg(fRc, ("%d\n", GetLastError())); NOREF(fRc);
-
-    RTCritSectLeave(&pThis->CritSect);
-
-    return VINF_SUCCESS;
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_SUCCESS(rc))
+    {
+        ASMAtomicUoWriteBool(&pThis->fCancelled, true);
+        BOOL fRc = SetEvent(pThis->hEvent);
+        AssertMsg(fRc, ("%d\n", GetLastError())); NOREF(fRc);
+
+        rc = RTCritSectLeave(&pThis->CritSect);
+    }
+
+    return rc;
 }
 
@@ -632 +641 @@
     pThis->cRefs = 1; /* The one we return. */
     pThis->fIOPending = false;
+    pThis->fZeroByteRead = false;
     pThis->fCancelled = false;
     pThis->pbBounceBuf = NULL;
@@ -767 +777 @@
     {
         /* No concurrent readers, sorry. */
-        if (pThis->cRefs == 0)
+        if (pThis->cRefs == 1)
         {
             pThis->cRefs++;
@@ -915 +925 @@
     {
         /* No concurrent writers, sorry. */
-        if (pThis->cRefs == 0)
+        if (pThis->cRefs == 1)
         {
             pThis->cRefs++;
@@ -1006 +1016 @@
     AssertReturn(pThis->u32Magic == RTLOCALIPCSESSION_MAGIC, VERR_INVALID_HANDLE);
 
+    uint64_t const StartMsTS = RTTimeMilliTS();
+
     int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_FAILURE(rc))
+        return rc;
+    for (unsigned iLoop = 0;; iLoop++)
+    {
+        HANDLE hWait = INVALID_HANDLE_VALUE;
+
+        if (pThis->fIOPending)
+            hWait = pThis->OverlappedIO.hEvent;
+        else
+        {
+            /* Peek at the pipe buffer and see how many bytes it contains. */
+            DWORD cbAvailable;
+            BOOL fRc = PeekNamedPipe(pThis->hNmPipe, NULL, 0, NULL, &cbAvailable, NULL);
+            if (   fRc
+                && cbAvailable)
+            {
+                rc = VINF_SUCCESS;
+                break;
+            }
+            else if (!fRc)
+            {
+                rc = RTErrConvertFromWin32(GetLastError());
+                break;
+            }
+
+            /* Start a zero byte read operation that we can wait on. */
+            if (cMillies == 0)
+            {
+                rc = VERR_TIMEOUT;
+                break;
+            }
+            AssertBreakStmt(pThis->cRefs == 1, rc = VERR_WRONG_ORDER);
+            fRc = ResetEvent(pThis->OverlappedIO.hEvent); Assert(fRc == TRUE);
+            DWORD cbRead = 0;
+            if (ReadFile(pThis->hNmPipe, pThis->abBuf, 0, &cbRead, &pThis->OverlappedIO))
+            {
+                rc = VINF_SUCCESS;
+                if (iLoop > 10)
+                    RTThreadYield();
+            }
+            else if (GetLastError() == ERROR_IO_PENDING)
+            {
+                pThis->cRefs++;
+                pThis->fIOPending = true;
+                pThis->fZeroByteRead = true;
+                hWait = pThis->OverlappedIO.hEvent;
+            }
+            else
+                rc = RTErrConvertFromWin32(GetLastError());
+        }
+
+        if (RT_FAILURE(rc))
+            break;
+
+        /*
+         * Check for timeout.
+         */
+        DWORD cMsMaxWait = INFINITE;
+        if (   cMillies != RT_INDEFINITE_WAIT
+            && (   hWait != INVALID_HANDLE_VALUE
+                || iLoop > 10)
+           )
+        {
+            uint64_t cElapsed = RTTimeMilliTS() - StartMsTS;
+            if (cElapsed >= cMillies)
+            {
+                rc = VERR_TIMEOUT;
+                break;
+            }
+            cMsMaxWait = cMillies - (uint32_t)cElapsed;
+        }
+
+        /*
+         * Wait.
+         */
+        if (hWait != INVALID_HANDLE_VALUE)
+        {
+            RTCritSectLeave(&pThis->CritSect);
+
+            DWORD dwRc = WaitForSingleObject(hWait, cMsMaxWait);
+            if (dwRc == WAIT_OBJECT_0)
+                rc = VINF_SUCCESS;
+            else if (dwRc == WAIT_TIMEOUT)
+                rc = VERR_TIMEOUT;
+            else if (dwRc == WAIT_ABANDONED)
+                rc = VERR_INVALID_HANDLE;
+            else
+                rc = RTErrConvertFromWin32(GetLastError());
+
+            if (   RT_FAILURE(rc)
+                && pThis->u32Magic != RTLOCALIPCSESSION_MAGIC)
+                return rc;
+
+            int rc2 = RTCritSectEnter(&pThis->CritSect);
+            AssertRC(rc2);
+            if (pThis->fZeroByteRead)
+            {
+                Assert(pThis->cRefs);
+                pThis->cRefs--;
+                pThis->fIOPending = false;
+
+                if (rc != VINF_SUCCESS)
+                {
+                    BOOL fRc = CancelIo(pThis->hNmPipe);
+                    Assert(fRc == TRUE);
+                }
+
+                DWORD cbRead = 0;
+                BOOL fRc = GetOverlappedResult(pThis->hNmPipe, &pThis->OverlappedIO, &cbRead, TRUE /*fWait*/);
+                if (   !fRc
+                    && RT_SUCCESS(rc))
+                {
+                    DWORD dwRc = GetLastError();
+                    if (dwRc == ERROR_OPERATION_ABORTED)
+                        rc = VERR_CANCELLED;
+                    else
+                        rc = RTErrConvertFromWin32(dwRc);
+                }
+            }
+
+            if (RT_FAILURE(rc))
+                break;
+        }
+    }
+
+    int rc2 = RTCritSectLeave(&pThis->CritSect);
     if (RT_SUCCESS(rc))
-    {
-        /* No concurrent waiters, sorry. */
-        if (pThis->cRefs == 0)
-        {
-            pThis->cRefs++;
-            pThis->fCancelled = false; /* Reset canellation status. */
-            RTCritSectLeave(&pThis->CritSect);
-
-            DWORD dwTimeout = cMillies == RT_INDEFINITE_WAIT
-                            ? INFINITE : cMillies;
-            DWORD dwRc = WaitForSingleObject(pThis->OverlappedIO.hEvent, dwTimeout);
-
-            RTCritSectEnter(&pThis->CritSect);
-            if (dwRc == WAIT_TIMEOUT)
-            {
-                rc = VERR_TIMEOUT;
-            }
-            else if (dwRc == WAIT_ABANDONED)
-            {
-                rc = VERR_BROKEN_PIPE;
-            }
-            else if (dwRc == WAIT_FAILED)
-                rc = RTErrConvertFromWin32(GetLastError());
-            else if (pThis->fCancelled)
-                rc = VERR_CANCELLED;
-
-            pThis->cRefs--;
-        }
-        else
-            rc = VERR_WRONG_ORDER;
-        RTCritSectLeave(&pThis->CritSect);
-    }
+        rc = rc2;
 
     return rc;

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

@@ -71 +71 @@
 static int testServerListenAndCancel(RTTEST hTest, const char *pszExecPath)
 {
-    RTTestISub("testServerListenAndCancel");
+    RTTestSub(hTest, "testServerListenAndCancel");
 
     RTLOCALIPCSERVER ipcServer;
@@ -118 +118 @@
         RTLOCALIPCSESSION ipcSession;
         rc = RTLocalIpcServerListen(pCtx->hServer, &ipcSession);
-#ifdef DEBUG_andy
-        RTTestPrintf(pCtx->hTest, RTTESTLVL_INFO, "testSessionConnectionThread: Listening returned with rc=%Rrc\n", rc);
-#endif
+        RTTestPrintf(pCtx->hTest, RTTESTLVL_DEBUG, "testSessionConnectionThread: Listening returned with rc=%Rrc\n", rc);
         if (RT_SUCCESS(rc))
         {
@@ -151 +149 @@
 static int testSessionConnection(RTTEST hTest, const char *pszExecPath)
 {
-    RTTestISub("testSessionConnection");
+    RTTestSub(hTest, "testSessionConnection");
 
     RTLOCALIPCSERVER ipcServer;
@@ -158 +156 @@
     if (RT_SUCCESS(rc))
     {
-#if 0
+#ifndef VBOX_TESTCASES_WITH_NO_THREADING
         LOCALIPCTHREADCTX threadCtx = { ipcServer, hTest };
 
@@ -181 +179 @@
                 RTTEST_CHECK_RC(hTest, RTThreadWait(hThread,
                                                     30 * 1000 /* 30s timeout */, &threadRc), VINF_SUCCESS);
-                RTTEST_CHECK_RC_BREAK(hTest, threadRc,  VINF_SUCCESS);
+                RTTEST_CHECK_RC_BREAK(hTest, threadRc,  VERR_CANCELLED);
                 RTTestPrintf(hTest, RTTESTLVL_INFO, "Server thread terminated successfully\n");
-                RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcServerDestroy(ipcServer),  VINF_SUCCESS);
+                RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcServerDestroy(ipcServer), VINF_SUCCESS);
                 RTTEST_CHECK_BREAK(hTest, stsChild.enmReason == RTPROCEXITREASON_NORMAL);
                 RTTEST_CHECK_BREAK(hTest, stsChild.iStatus == 0);
@@ -190 +188 @@
         }
         else
-            RTTestIFailed("Unable to create thread for cancelling server, rc=%Rrc\n", rc);
+            RTTestFailed(hTest, "Unable to create thread for cancelling server, rc=%Rrc\n", rc);
 #else
         do
@@ -205 +203 @@
             }
             else
-                RTTestIFailed("Error while listening, rc=%Rrc\n", rc);
+                RTTestFailed(hTest, "Error while listening, rc=%Rrc\n", rc);
 
         } while (0);
@@ -211 +209 @@
     }
     else
-        RTTestIFailed("Unable to create IPC server, rc=%Rrc\n", rc);
+        RTTestFailed(hTest, "Unable to create IPC server, rc=%Rrc\n", rc);
 
     return VINF_SUCCESS;
 }
 
+static DECLCALLBACK(int) testSessionWaitThread(RTTHREAD hSelf, void *pvUser)
+{
+    PLOCALIPCTHREADCTX pCtx = (PLOCALIPCTHREADCTX)pvUser;
+    AssertPtr(pCtx);
+
+    int rc;
+    for (;;)
+    {
+        RTTestPrintf(pCtx->hTest, RTTESTLVL_INFO, "testSessionWaitThread: Listening for incoming connections ...\n");
+        RTLOCALIPCSESSION ipcSession;
+        rc = RTLocalIpcServerListen(pCtx->hServer, &ipcSession);
+        if (RT_SUCCESS(rc))
+        {
+            RTTestPrintf(pCtx->hTest, RTTESTLVL_INFO, "testSessionWaitThread: Got new client connection, waiting a bit ...\n");
+            RTThreadSleep(2000);
+            rc = RTLocalIpcSessionClose(ipcSession);
+        }
+        else
+        {
+            RTTestPrintf(pCtx->hTest, RTTESTLVL_INFO, "testSessionWaitThread: Listening ended with rc=%Rrc\n", rc);
+            break;
+        }
+    }
+
+    RTTestPrintf(pCtx->hTest, RTTESTLVL_INFO, "testSessionWaitThread: Ended with rc=%Rrc\n", rc);
+    return rc;
+}
+
+static RTEXITCODE testSessionWaitChild(int argc, char **argv, RTTEST hTest)
+{
+    do
+    {
+        RTThreadSleep(2000); /* Fudge. */
+        RTLOCALIPCSESSION clientSession;
+        RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcSessionConnect(&clientSession, "tstRTLocalIpcSessionWait",
+                                                              0 /* Flags */), VINF_SUCCESS);
+        RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcSessionWaitForData(clientSession, 100       /* 100ms timeout */),
+                                                                  VERR_TIMEOUT);
+        /* Next, try 60s timeout. Should be returning way earlier because the server closed the
+         * connection after the first client connected. */
+        RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcSessionWaitForData(clientSession, 60 * 1000),
+                                                                  VERR_BROKEN_PIPE);
+        /* Last try, also should fail because the server should be not around anymore. */
+        RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcSessionWaitForData(clientSession, 5 * 1000),
+                                                                  VERR_BROKEN_PIPE);
+        RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcSessionClose(clientSession), VINF_SUCCESS);
+
+    } while (0);
+
+    return !RTTestErrorCount(hTest) ? RTEXITCODE_SUCCESS : RTEXITCODE_FAILURE;
+}
+
+static int testSessionWait(RTTEST hTest, const char *pszExecPath)
+{
+    RTTestSub(hTest, "testSessionWait");
+
+    RTLOCALIPCSERVER ipcServer;
+    int rc = RTLocalIpcServerCreate(&ipcServer, "tstRTLocalIpcSessionWait",
+                                    RTLOCALIPC_FLAGS_MULTI_SESSION);
+    if (RT_SUCCESS(rc))
+    {
+        LOCALIPCTHREADCTX threadCtx = { ipcServer, hTest };
+
+        /* Spawn a simple worker thread and let it listen for incoming connections.
+         * In the meanwhile we try to cancel the server and see what happens. */
+        RTTHREAD hThread;
+        rc = RTThreadCreate(&hThread, testSessionWaitThread,
+                            &threadCtx, 0 /* Stack */, RTTHREADTYPE_DEFAULT, RTTHREADFLAGS_WAITABLE, "tstIpc3");
+        if (RT_SUCCESS(rc))
+        {
+            do
+            {
+                RTPROCESS hProc;
+                const char *apszArgs[4] = { pszExecPath, "child", "tstRTLocalIpcSessionWaitFork", NULL };
+                RTTEST_CHECK_RC_BREAK(hTest, RTProcCreate(pszExecPath, apszArgs,
+                                                          RTENV_DEFAULT, 0 /* fFlags*/, &hProc), VINF_SUCCESS);
+                RTThreadSleep(5000); /* Let the server run for some time ... */
+                RTTestPrintf(hTest, RTTESTLVL_INFO, "Cancelling server listening\n");
+                RTTEST_CHECK_RC_BREAK(hTest, RTLocalIpcServerCancel(ipcServer), VINF_SUCCESS);
+                /* Wait for the server thread to terminate. */
+                int threadRc;
+                RTTEST_CHECK_RC(hTest, RTThreadWait(hThread,
+                                                    30 * 1000 /* 30s timeout */, &threadRc), VINF_SUCCESS);
+                RTTEST_CHECK_RC_BREAK(hTest, threadRc, VERR_CANCELLED);
+                RTTEST_CHECK_RC(hTest, RTLocalIpcServerDestroy(ipcServer), VINF_SUCCESS);
+                RTTestPrintf(hTest, RTTESTLVL_INFO, "Server thread terminated successfully\n");
+                /* Check if the child ran successfully. */
+                RTPROCSTATUS stsChild;
+                RTTEST_CHECK_RC_BREAK(hTest, RTProcWait(hProc, RTPROCWAIT_FLAGS_BLOCK, &stsChild), VINF_SUCCESS);
+                RTTestPrintf(hTest, RTTESTLVL_INFO, "Child terminated\n");
+                RTTEST_CHECK_BREAK(hTest, stsChild.enmReason == RTPROCEXITREASON_NORMAL);
+                RTTEST_CHECK_BREAK(hTest, stsChild.iStatus == 0);
+            }
+            while (0);
+        }
+        else
+            RTTestFailed(hTest, "Unable to create thread for cancelling server, rc=%Rrc\n", rc);
+    }
+    else
+        RTTestFailed(hTest, "Unable to create IPC server, rc=%Rrc\n", rc);
+
+    return VINF_SUCCESS;
+}
+
 static RTEXITCODE mainChild(int argc, char **argv)
 {
+    if (argc < 3) /* Safety first. */
+        return RTEXITCODE_FAILURE;
     /* Note: We assume argv[2] always contains the actual test type to perform. */
     RTTEST hTest;
@@ -225 +329 @@
     RTTestBanner(hTest);
 
+    RTAssertSetMayPanic(false);
+#ifdef DEBUG_andy
+    RTAssertSetQuiet(true);
+#endif
+
     if (!RTStrICmp(argv[2], "tstRTLocalIpcSessionConnectionFork"))
         rcExit = testSessionConnectionChild(argc, argv, hTest);
+    else if (!RTStrICmp(argv[2], "tstRTLocalIpcSessionWaitFork"))
+        rcExit = testSessionWaitChild(argc, argv, hTest);
 
     return RTTestSummaryAndDestroy(hTest);
@@ -283 +394 @@
     if (RTTestErrorCount(hTest) == 0)
     {
-#ifndef DEBUG_andy
         RTTESTI_CHECK_RC_RET(testServerListenAndCancel(hTest, szExecPath), VINF_SUCCESS, 1);
-#endif
         RTTESTI_CHECK_RC_RET(testSessionConnection(hTest, szExecPath), VINF_SUCCESS, 1);
+        RTTESTI_CHECK_RC_RET(testSessionWait(hTest, szExecPath), VINF_SUCCESS, 1);
     }