Changeset 82383 in vbox for trunk/src/VBox/Runtime/common/ioqueue

Timestamp:

Dec 4, 2019 1:19:31 PM (5 years ago)

Author:

vboxsync

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

135217

Message:

Runtime/ioqueue: Finish the fallback file provider

File:

• trunk/src/VBox/Runtime/common/ioqueue/ioqueue-stdfile-provider.cpp (modified) (13 diffs)

trunk/src/VBox/Runtime/common/ioqueue/ioqueue-stdfile-provider.cpp

@@ -52 +52 @@
 /** The waiting thread was interrupted by the external wakeup call. */
 #define RTIOQUEUE_STDFILE_PROV_STATE_F_EVTWAIT_INTR             RT_BIT(1)
+#define RTIOQUEUE_STDFILE_PROV_STATE_F_EVTWAIT_INTR_BIT         1
 /** The I/O queue worker thread needs to be woken up to process new requests. */
 #define RTIOQUEUE_STDFILE_PROV_STATE_F_WORKER_NEED_WAKEUP       RT_BIT(2)
@@ -109 +110 @@
     /** Submission queue producer index. */
     volatile uint32_t           idxSqProd;
+    /** Submission queue producer value for any uncommitted requests. */
+    uint32_t                    idxSqProdUncommit;
     /** Submission queue consumer index. */
     volatile uint32_t           idxSqCons;
@@ -178 +181 @@
 
     /* Write the result back into the completion queue. */
-    pCqEntry->rcReq  = rcReq;
-    pCqEntry->pvUser = pSqEntry->pvUser;
+    pCqEntry->rcReq    = rcReq;
+    pCqEntry->pvUser   = pSqEntry->pvUser;
+    pCqEntry->cbXfered = RT_SUCCESS(rcReq) ? pSqEntry->cbReq : 0;
 }
 
@@ -215 +219 @@
 
         /* Process all requests. */
+        uint32_t cCqFree = 0;
+        if (idxCqCons > pThis->idxCqProd)
+            cCqFree = pThis->cCqEntries - (pThis->cCqEntries - idxCqCons) - pThis->idxCqProd;
+        else
+            cCqFree = pThis->cCqEntries - pThis->idxCqProd - idxCqCons;
         do
         {
             while (   idxSqCons != idxSqProd
-                   && idxCqCons != pThis->idxCqProd)
+                   && cCqFree)
             {
                 PCRTIOQUEUESSQENTRY pSqEntry = &pThis->paSqEntryBase[idxSqCons];
@@ -227 +236 @@
 
                 idxSqCons = (idxSqCons + 1) % pThis->cSqEntries;
+                cCqFree--;
                 pThis->idxCqProd = (pThis->idxCqProd + 1) % pThis->cCqEntries;
+                ASMAtomicWriteU32(&pThis->idxSqCons, idxSqCons);
                 ASMWriteFence();
                 if (ASMAtomicReadU32(&pThis->fState) & RTIOQUEUE_STDFILE_PROV_STATE_F_EVTWAIT_NEED_WAKEUP)
@@ -236 +247 @@
             }
 
-            ASMWriteFence();
-            ASMAtomicWriteU32(&pThis->idxSqCons, idxSqCons);
             idxSqProd = ASMAtomicReadU32(&pThis->idxSqProd);
-            idxSqCons = ASMAtomicReadU32(&pThis->idxSqCons);
-        } while (idxSqCons != idxSqProd);
+        } while (   idxSqCons != idxSqProd
+                 && cCqFree);
     }
 
@@ -264 +273 @@
     int rc = VINF_SUCCESS;
 
-    pThis->cSqEntries = cSqEntries;
-    pThis->cCqEntries = cCqEntries;
-    pThis->idxSqProd  = 0;
-    pThis->idxSqCons  = 0;
-    pThis->idxCqProd  = 0;
-    pThis->idxCqCons  = 0;
-    pThis->fShutdown  = false;
-    pThis->fState     = 0;
+    cSqEntries++;
+    cCqEntries++;
+
+    pThis->cSqEntries        = cSqEntries;
+    pThis->cCqEntries        = cCqEntries;
+    pThis->idxSqProd         = 0;
+    pThis->idxSqProdUncommit = 0;
+    pThis->idxSqCons         = 0;
+    pThis->idxCqProd         = 0;
+    pThis->idxCqCons         = 0;
+    pThis->fShutdown         = false;
+    pThis->fState            = 0;
 
     pThis->paSqEntryBase = (PRTIOQUEUESSQENTRY)RTMemAllocZ(cSqEntries * sizeof(RTIOQUEUESSQENTRY));
@@ -361 +374 @@
 {
     PRTIOQUEUEPROVINT pThis = hIoQueueProv;
-    uint32_t idxSqProd = ASMAtomicReadU32(&pThis->idxSqProd);
-    PRTIOQUEUESSQENTRY pSqEntry = &pThis->paSqEntryBase[idxSqProd];
+    PRTIOQUEUESSQENTRY pSqEntry = &pThis->paSqEntryBase[pThis->idxSqProdUncommit];
 
     pSqEntry->hFile     = pHandle->u.hFile;
@@ -372 +384 @@
     pSqEntry->fSg       = false;
     pSqEntry->u.pvBuf   = pvBuf;
-    ASMWriteFence();
-
-    idxSqProd = (idxSqProd + 1) % pThis->cSqEntries;
-    ASMAtomicWriteU32(&pThis->idxSqProd, idxSqProd);
+
+    pThis->idxSqProdUncommit = (pThis->idxSqProdUncommit + 1) % pThis->cSqEntries;
     return VINF_SUCCESS;
 }
@@ -386 +396 @@
 {
     PRTIOQUEUEPROVINT pThis = hIoQueueProv;
-    uint32_t idxSqProd = ASMAtomicReadU32(&pThis->idxSqProd);
-    PRTIOQUEUESSQENTRY pSqEntry = &pThis->paSqEntryBase[idxSqProd];
+    PRTIOQUEUESSQENTRY pSqEntry = &pThis->paSqEntryBase[pThis->idxSqProdUncommit];
 
     pSqEntry->hFile     = pHandle->u.hFile;
@@ -397 +406 @@
     pSqEntry->fSg       = true;
     pSqEntry->u.pSgBuf  = pSgBuf;
-    ASMWriteFence();
-
-    idxSqProd = (idxSqProd + 1) % pThis->cSqEntries;
-    ASMAtomicWriteU32(&pThis->idxSqProd, idxSqProd);
+
+    pThis->idxSqProdUncommit = (pThis->idxSqProdUncommit + 1) % pThis->cSqEntries;
     return VINF_SUCCESS;
 }
@@ -409 +416 @@
 {
     PRTIOQUEUEPROVINT pThis = hIoQueueProv;
-    RT_NOREF(pcReqsCommitted);
-
+
+    if (pThis->idxSqProd > pThis->idxSqProdUncommit)
+        *pcReqsCommitted = pThis->cSqEntries - pThis->idxSqProd + pThis->idxSqProdUncommit;
+    else
+        *pcReqsCommitted = pThis->idxSqProdUncommit - pThis->idxSqProd;
+
+    ASMWriteFence();
+    ASMAtomicWriteU32(&pThis->idxSqProd, pThis->idxSqProdUncommit);
     return RTSemEventSignal(pThis->hSemEvtWorker);
 }
@@ -419 +432 @@
                                                       uint32_t cMinWait, uint32_t *pcCEvt, uint32_t fFlags)
 {
-    PRTIOQUEUEPROVINT pThis = hIoQueueProv;
-    RT_NOREF(pThis, paCEvt, cCEvt, cMinWait, pcCEvt, fFlags);
-
-    return VERR_NOT_IMPLEMENTED;
+    RT_NOREF(fFlags);
+
+    PRTIOQUEUEPROVINT pThis = hIoQueueProv;
+    int rc = VINF_SUCCESS;
+    uint32_t idxCEvt = 0;
+
+    while (   RT_SUCCESS(rc)
+           && cMinWait
+           && cCEvt)
+    {
+        ASMAtomicOrU32(&pThis->fState, RTIOQUEUE_STDFILE_PROV_STATE_F_EVTWAIT_NEED_WAKEUP);
+        uint32_t idxCqProd = ASMAtomicReadU32(&pThis->idxCqProd);
+        uint32_t idxCqCons = ASMAtomicReadU32(&pThis->idxCqCons);
+
+        if (idxCqCons == idxCqProd)
+        {
+            rc = RTSemEventWait(pThis->hSemEvtWaitEvts, RT_INDEFINITE_WAIT);
+            AssertRC(rc);
+            if (ASMAtomicBitTestAndClear(&pThis->fState, RTIOQUEUE_STDFILE_PROV_STATE_F_EVTWAIT_INTR_BIT))
+            {
+                rc = VERR_INTERRUPTED;
+                ASMAtomicBitTestAndClear(&pThis->fState, RTIOQUEUE_STDFILE_PROV_STATE_F_WORKER_NEED_WAKEUP_BIT);
+                break;
+            }
+
+            idxCqProd = ASMAtomicReadU32(&pThis->idxCqProd);
+            idxCqCons = ASMAtomicReadU32(&pThis->idxCqCons);
+        }
+
+        ASMAtomicBitTestAndClear(&pThis->fState, RTIOQUEUE_STDFILE_PROV_STATE_F_WORKER_NEED_WAKEUP_BIT);
+
+        /* Process all requests. */
+        while (   idxCqCons != idxCqProd
+               && cCEvt)
+        {
+            PRTIOQUEUECEVT pCqEntry = &pThis->paCqEntryBase[idxCqCons];
+
+            paCEvt[idxCEvt].rcReq    = pCqEntry->rcReq;
+            paCEvt[idxCEvt].pvUser   = pCqEntry->pvUser;
+            paCEvt[idxCEvt].cbXfered = pCqEntry->cbXfered;
+            ASMReadFence();
+
+            idxCEvt++;
+            cCEvt--;
+            cMinWait--;
+
+            idxCqCons = (idxCqCons + 1) % pThis->cCqEntries;
+            pThis->idxCqCons = (pThis->idxCqCons + 1) % pThis->cCqEntries;
+            ASMWriteFence();
+        }
+    }
+
+    *pcCEvt = idxCEvt;
+    return rc;
 }