Changeset 21900 in vbox

Timestamp:

Jul 30, 2009 5:31:36 PM (15 years ago)

Author:

vboxsync

Message:

SSM: Added the read ahead thread.

File:

• trunk/src/VBox/VMM/SSM-new.cpp (modified) (15 diffs)

trunk/src/VBox/VMM/SSM-new.cpp

@@ -324 +324 @@
     /** The handle of the I/O thread. This is set to nil when not active. */
     RTTHREAD                hIoThread;
+    /** Where to seek to. */
+    uint64_t                offNeedSeekTo;
 
     /** The head of the consumer queue.
@@ -750 +752 @@
 static int                  ssmR3Register(PVM pVM, const char *pszName, uint32_t u32Instance, uint32_t u32Version, size_t cbGuess, const char *pszBefore, PSSMUNIT *ppUnit);
 static int                  ssmR3StrmWriteBuffers(PSSMSTRM pStrm);
+static int                  ssmR3StrmReadMore(PSSMSTRM pStrm);
 static int                  ssmR3CalcChecksum(RTFILE File, uint64_t cbFile, uint32_t *pu32CRC);
 static void                 ssmR3Progress(PSSMHANDLE pSSM, uint64_t cbAdvance);
@@ -1369 +1372 @@
     pStrm->rc           = VINF_SUCCESS;
     pStrm->hIoThread    = NIL_RTTHREAD;
+    pStrm->offNeedSeekTo= UINT64_MAX;
 
     pStrm->pHead        = NULL;
@@ -1655 +1659 @@
             if (RT_FAILURE(pStrm->rc))
                 return NULL;
+            if (    !pStrm->fWrite
+                &&  pStrm->hIoThread == NIL_RTTHREAD)
+            {
+                int rc = ssmR3StrmReadMore(pStrm);
+                if (RT_FAILURE(rc))
+                    return NULL;
+                continue;
+            }
+
             int rc = RTSemEventWaitNoResume(pStrm->hEvtHead, 30000);
             if (    rc == VERR_SEM_DESTROYED
@@ -1687 +1700 @@
                                                      cb - pStrm->offStreamCRC);
             pStrm->offCurStream += cb;
-            pStrm->off          = 0;
-            pStrm->offStreamCRC = 0;
+            pStrm->off           = 0;
+            pStrm->offStreamCRC  = 0;
 
             ssmR3StrmPutBuf(pStrm, pBuf);
@@ -1701 +1714 @@
                                                      cb - pStrm->offStreamCRC);
             pStrm->offCurStream += cb;
-            pStrm->off          = 0;
-            pStrm->offStreamCRC = 0;
+            pStrm->off           = 0;
+            pStrm->offStreamCRC  = 0;
 
             ssmR3StrmPutFreeBuf(pStrm, pBuf);
@@ -1767 +1780 @@
 {
     /*
-     * Flush and close the stream.
+     * Flush, terminate the I/O thread, and close the stream.
      */
     if (pStrm->fWrite)
     {
         ssmR3StrmFlushCurBuf(pStrm);
-        ssmR3StrmWriteBuffers(pStrm);
-    }
+        if (pStrm->hIoThread == NIL_RTTHREAD)
+            ssmR3StrmWriteBuffers(pStrm);
+    }
+
+    if (pStrm->hIoThread != NIL_RTTHREAD)
+    {
+        ASMAtomicWriteBool(&pStrm->fTerminating, true);
+        int rc2 = RTSemEventSignal(pStrm->fWrite ? pStrm->hEvtHead : pStrm->hEvtFree); AssertLogRelRC(rc2);
+        int rc3 = RTThreadWait(pStrm->hIoThread, RT_INDEFINITE_WAIT, NULL);            AssertLogRelRC(rc3);
+        pStrm->hIoThread = NIL_RTTHREAD;
+    }
+
     int rc = RTFileClose(pStrm->hFile);
     if (RT_FAILURE(rc))
@@ -1922 +1945 @@
  * Read more from the stream.
  *
- * @returns VBox status code. VERR_EOF is passed up.
+ * @returns VBox status code. VERR_EOF gets translated into VINF_EOF.
  * @param   pStrm       The stream handle.
  *
- * @thread  The I/O thread when we get one.
+ * @thread  The I/O thread when we got one, otherwise the stream user.
  */
 static int ssmR3StrmReadMore(PSSMSTRM pStrm)
 {
     int rc;
+    Log6(("ssmR3StrmReadMore:\n"));
 
     /*
@@ -1936 +1960 @@
     if (pStrm->fNeedSeek)
     {
-        rc = RTFileSeek(pStrm->hFile, pStrm->offCurStream + pStrm->off, RTFILE_SEEK_BEGIN, NULL);
+        rc = RTFileSeek(pStrm->hFile, pStrm->offNeedSeekTo, RTFILE_SEEK_BEGIN, NULL);
         if (RT_FAILURE(rc))
         {
             if (ssmR3StrmSetError(pStrm, rc))
-                LogRel(("ssmR3StrmReadMore: RTFileSeek(,%#llx,) failed with rc=%Rrc\n", pStrm->offCurStream + pStrm->off, rc));
+                LogRel(("ssmR3StrmReadMore: RTFileSeek(,%#llx,) failed with rc=%Rrc\n", pStrm->offNeedSeekTo, rc));
             return rc;
         }
-
+        pStrm->fNeedSeek     = false;
+        pStrm->offNeedSeekTo = UINT64_MAX;
     }
 
@@ -1953 +1978 @@
         return pStrm->rc;
 
-    pBuf->offStream = pStrm->offCurStream + pStrm->off;
-    Assert(pBuf->offStream == RTFileTell(pStrm->hFile));
+    pBuf->offStream = RTFileTell(pStrm->hFile);
     size_t cbRead   = sizeof(pBuf->abData);
     rc = RTFileRead(pStrm->hFile, &pBuf->abData[0], cbRead, &cbRead);
-    if (RT_SUCCESS(rc))
+    if (    RT_SUCCESS(rc)
+        &&  cbRead > 0)
     {
         pBuf->cb           = (uint32_t)cbRead;
-        pBuf->fEndOfStream = cbRead == 0; /** @todo Check out this EOF mess... */
+        pBuf->fEndOfStream = false;
+        Log6(("ssmR3StrmReadMore: %#010llx %#x\n", pBuf->offStream, pBuf->cb));
         ssmR3StrmPutBuf(pStrm, pBuf);
     }
-    else if (rc == VERR_EOF)
+    else if (   (   RT_SUCCESS_NP(rc)
+                 && cbRead == 0)
+             || rc == VERR_EOF)
     {
         pBuf->cb           = 0;
         pBuf->fEndOfStream = true;
+        Log6(("ssmR3StrmReadMore: %#010llx 0 EOF!\n", pBuf->offStream));
         ssmR3StrmPutBuf(pStrm, pBuf);
+        rc = VINF_EOF;
     }
     else
     {
+        Log6(("ssmR3StrmReadMore: %#010llx rc=%Rrc!\n", pBuf->offStream, rc));
         if (ssmR3StrmSetError(pStrm, rc))
             LogRel(("ssmR3StrmReadMore: RTFileRead(,,%#x,) -> %Rrc at offset %#llx\n",
@@ -2037 +2068 @@
          */
         ssmR3StrmFlushCurBuf(pStrm);
-        if (pStrm->hIoThread == NIL_RTTHREAD)
-        {
-            rc = ssmR3StrmReadMore(pStrm);
-            if (RT_FAILURE(rc))
-                break;
-        }
         PSSMSTRMBUF pBuf = ssmR3StrmGetBuf(pStrm);
         if (!pBuf)
@@ -2195 +2220 @@
     {
         pStrm->fNeedSeek    = false;
+        pStrm->offNeedSeekTo= UINT64_MAX;
         pStrm->offCurStream = offStream;
         pStrm->off          = 0;
@@ -2288 +2314 @@
     AssertReturn(pStrm->hIoThread == NIL_RTTHREAD, VERR_WRONG_ORDER);
 
-    pStrm->fNeedSeek = true;
+    if (!pStrm->fNeedSeek)
+    {
+        pStrm->fNeedSeek     = true;
+        pStrm->offNeedSeekTo = pStrm->offCurStream + (pStrm->pCur ? pStrm->pCur->cb : 0);
+    }
+
     int rc = RTFileSeek(pStrm->hFile, off, off >= 0 ? RTFILE_SEEK_BEGIN : RTFILE_SEEK_END, poff);
     if (RT_SUCCESS(rc))
@@ -2294 +2325 @@
 
     return rc;
+}
+
+
+/**
+ * The I/O thread.
+ *
+ * @returns VINF_SUCCESS (ignored).
+ * @param   hSelf       The thread handle.
+ * @param   pvStrm      The stream handle.
+ */
+static DECLCALLBACK(int) ssmR3StrmIoThread(RTTHREAD hSelf, void *pvStrm)
+{
+    PSSMSTRM pStrm = (PSSMSTRM)pvStrm;
+    ASMAtomicWriteHandle(&pStrm->hIoThread, hSelf); /* paranoia */
+
+    Log(("ssmR3StrmIoThread: starts working\n"));
+    if (pStrm->fWrite)
+    {
+        /*
+         * Write until error or terminated.
+         */
+        for (;;)
+        {
+
+        }
+    }
+    else
+    {
+        /*
+         * Read until end of file, error or termination.
+         */
+        for (;;)
+        {
+            if (ASMAtomicReadBool(&pStrm->fTerminating))
+            {
+                Log(("ssmR3StrmIoThread: quitting reading because of pending termination.\n"));
+                break;
+            }
+
+            int rc = ssmR3StrmReadMore(pStrm);
+            if (    RT_FAILURE(rc)
+                ||  rc == VINF_EOF)
+            {
+                Log(("ssmR3StrmIoThread: quitting reading with rc=%Rrc\n", rc));
+                break;
+            }
+            if (RT_FAILURE(pStrm->rc))
+            {
+                Log(("ssmR3StrmIoThread: quitting reading with stream rc=%Rrc\n", pStrm->rc));
+                break;
+            }
+        }
+    }
+
+    return VINF_SUCCESS;
+}
+
+
+/**
+ * Starts the I/O thread for the specified stream.
+ *
+ * @param   pStrm       The stream handle.
+ */
+static void ssmR3StrmStartIoThread(PSSMSTRM pStrm)
+{
+    Assert(pStrm->hIoThread == NIL_RTTHREAD);
+
+    RTTHREAD hThread;
+    int rc = RTThreadCreate(&hThread, ssmR3StrmIoThread, pStrm, 0, RTTHREADTYPE_IO, RTTHREADFLAGS_WAITABLE, "SSM-IO");
+    AssertRCReturnVoid(rc);
+    ASMAtomicWriteHandle(&pStrm->hIoThread, hThread); /* paranoia */
 }
 
@@ -3575 +3677 @@
     if (RT_SUCCESS(rc))
     {
+        ssmR3StrmStartIoThread(&Handle.Strm);
+
         Handle.enmAfter     = enmAfter;
         Handle.pfnProgress  = pfnProgress;