sys: ioq: control pending queue with semaphore

src/sys/driver.h

@@ -149,13 +149,14 @@ typedef struct
 {
     KSPIN_LOCK SpinLock;
     int Enabled;
+    KSEMAPHORE PendingSemaphore;
     LIST_ENTRY PendingIrpList, ProcessIrpList;
     IO_CSQ PendingIoCsq, ProcessIoCsq;
 } FSP_IOQ;
 VOID FspIoqInitialize(FSP_IOQ *Ioq);
 VOID FspIoqEnable(FSP_IOQ *Ioq, int Delta);
 BOOLEAN FspIoqPostIrp(FSP_IOQ *Ioq, PIRP Irp);
-PIRP FspIoqNextPendingIrp(FSP_IOQ *Ioq);
+PIRP FspIoqNextPendingIrp(FSP_IOQ *Ioq, ULONG millis);
 BOOLEAN FspIoqStartProcessingIrp(FSP_IOQ *Ioq, PIRP Irp);
 PIRP FspIoqEndProcessingIrp(FSP_IOQ *Ioq, UINT_PTR IrpHint);
 VOID FspIoqCancelAll(FSP_IOQ *Ioq);

src/sys/ioq.c

@@ -8,20 +8,20 @@
 
 /*
  * An FSP_IOQ encapsulates the main FSP mechanism for handling IRP's.
- * It has two queues: a "pending" queue for managing newly arrived IRP's
+ * It has two queues: a "Pending" queue for managing newly arrived IRP's
- * and a "processing" queue for managing IRP's currently being processed
+ * and a "Processing" queue for managing IRP's currently being processed
  * (i.e. sent to the user-mode file system for further processing).
  *
  * IRP's arrive at a MajorFunction (MJ) and are then posted to the device's
  * FSP_IOQ and marked pending. When the user-mode file system performs
- * FSP_FSCTL_TRANSACT, the IRP's are removed from the pending queue and
+ * FSP_FSCTL_TRANSACT, the IRP's are removed from the Pending queue and
  * are then marshalled to the user process; prior to that they are added
- * to the processing queue. At a later time the user-mode will perform
+ * to the Processing queue. At a later time the user-mode will perform
  * another FSP_FSCTL_TRANSACT at which time any processed IRP's will be
- * marshalled back to us and will be then removed from the processing queue
+ * marshalled back to us and will be then removed from the Processing queue
  * and completed.
  *
- * State diagram:
+ * IRP State diagram:
  *                                 +--------------------+
  *           |                     |                    | StartProcessingIrp
  *           v                     |                    v
@@ -36,13 +36,18 @@
  *     +------------+              |              |     IN     |
  *           |                     |              +------------+
  *           | NextPendingIrp      |                    |
- *           v                     |                    | CompleteIrp
+ *           v                     |                    | CompleteRequest
  *     +------------+              |                    v
  *     |  TRANSACT  |              |              +------------+
  *     |    OUT     |              |              | Completed  |
  *     +------------+              |              +------------+
  *           |                     |
  *           +---------------------+
+ *
+ * Note that the Pending queue is controlled by a semaphore object, which
+ * counts how many IRP's are in the queue. The semaphore is incremented for
+ * every IRP that is inserted into the Pending queue and decremented every
+ * time an IRP is removed from the Pending queue.
  */
 
 static NTSTATUS FspIoqPendingInsertIrpEx(PIO_CSQ IoCsq, PIRP Irp, PVOID InsertContext)
@@ -52,7 +57,11 @@ static NTSTATUS FspIoqPendingInsertIrpEx(PIO_CSQ IoCsq, PIRP Irp, PVOID InsertCo
     if (0 > Ioq->Enabled)
         return STATUS_ACCESS_DENIED;
 
+    /* SpinLock: acquired, IRQL: DISPATCH_LEVEL */
     InsertTailList(&Ioq->PendingIrpList, &Irp->Tail.Overlay.ListEntry);
+    KeReleaseSemaphore(&Ioq->PendingSemaphore, FSP_IO_INCREMENT, 1, FALSE);
+        /* KeReleaseSemaphore allowed at DISPATCH_LEVEL with Wait==FALSE */
+
     return STATUS_SUCCESS;
 }
 
@@ -90,6 +99,16 @@ static VOID FspIoqPendingReleaseLock(PIO_CSQ IoCsq, KIRQL Irql)
     KeReleaseSpinLock(&Ioq->SpinLock, Irql);
 }
 
+static VOID FspIoqPendingCompleteCanceledIrp(PIO_CSQ IoCsq, PIRP Irp)
+{
+    FSP_IOQ *Ioq = CONTAINING_RECORD(IoCsq, FSP_IOQ, PendingIoCsq);
+    LARGE_INTEGER Timeout = { 0 };
+
+    /* SpinLock: not acquired */
+    KeWaitForSingleObject(&Ioq->PendingSemaphore, Executive, KernelMode, FALSE, &Timeout);
+    FspCompleteRequest(Irp, STATUS_CANCELLED);
+}
+
 static NTSTATUS FspIoqProcessInsertIrpEx(PIO_CSQ IoCsq, PIRP Irp, PVOID InsertContext)
 {
     FSP_IOQ *Ioq = CONTAINING_RECORD(IoCsq, FSP_IOQ, ProcessIoCsq);
@@ -142,7 +161,7 @@ static VOID FspIoqProcessReleaseLock(PIO_CSQ IoCsq, KIRQL Irql)
     KeReleaseSpinLock(&Ioq->SpinLock, Irql);
 }
 
-static VOID FspIoqCompleteCanceledIrp(PIO_CSQ IoCsq, PIRP Irp)
+static VOID FspIoqProcessCompleteCanceledIrp(PIO_CSQ IoCsq, PIRP Irp)
 {
     FspCompleteRequest(Irp, STATUS_CANCELLED);
 }
@@ -151,6 +170,7 @@ VOID FspIoqInitialize(FSP_IOQ *Ioq)
 {
     RtlZeroMemory(Ioq, sizeof *Ioq);
     KeInitializeSpinLock(&Ioq->SpinLock);
+    KeInitializeSemaphore(&Ioq->PendingSemaphore, 0, 1000000000);
     InitializeListHead(&Ioq->PendingIrpList);
     InitializeListHead(&Ioq->ProcessIrpList);
     IoCsqInitializeEx(&Ioq->PendingIoCsq,
@@ -159,14 +179,14 @@ VOID FspIoqInitialize(FSP_IOQ *Ioq)
         FspIoqPendingPeekNextIrp,
         FspIoqPendingAcquireLock,
         FspIoqPendingReleaseLock,
-        FspIoqCompleteCanceledIrp);
+        FspIoqPendingCompleteCanceledIrp);
     IoCsqInitializeEx(&Ioq->ProcessIoCsq,
         FspIoqProcessInsertIrpEx,
         FspIoqProcessRemoveIrp,
         FspIoqProcessPeekNextIrp,
         FspIoqProcessAcquireLock,
         FspIoqProcessReleaseLock,
-        FspIoqCompleteCanceledIrp);
+        FspIoqProcessCompleteCanceledIrp);
 }
 
 VOID FspIoqEnable(FSP_IOQ *Ioq, int Delta)
@@ -179,29 +199,49 @@ VOID FspIoqEnable(FSP_IOQ *Ioq, int Delta)
 
 BOOLEAN FspIoqPostIrp(FSP_IOQ *Ioq, PIRP Irp)
 {
-    return STATUS_SUCCESS == IoCsqInsertIrpEx(&Ioq->PendingIoCsq, Irp, 0, 0);
+    NTSTATUS Result;
+    Result = IoCsqInsertIrpEx(&Ioq->PendingIoCsq, Irp, 0, 0);
+    return NT_SUCCESS(Result);
 }
 
-PIRP FspIoqNextPendingIrp(FSP_IOQ *Ioq)
+PIRP FspIoqNextPendingIrp(FSP_IOQ *Ioq, ULONG millis)
 {
-    return IoCsqRemoveNextIrp(&Ioq->PendingIoCsq, (PVOID)1);
+    NTSTATUS Result;
+    LARGE_INTEGER Timeout;
+    PIRP Irp;
+
+    Timeout.QuadPart = (LONGLONG)millis * 10000;
+    Result = KeWaitForSingleObject(&Ioq->PendingSemaphore, Executive, KernelMode, FALSE,
+        -1 == millis ? 0 : &Timeout);
+    if (!NT_SUCCESS(Result))
+        return 0;
+
+    Irp = IoCsqRemoveNextIrp(&Ioq->PendingIoCsq, (PVOID)1);
+
+    if (0 == Irp)
+        /* signal the semaphore again; turns out we did not get an IRP! */
+        KeReleaseSemaphore(&Ioq->PendingSemaphore, FSP_IO_INCREMENT, 1, FALSE);
+
+    return Irp;
 }
 
 BOOLEAN FspIoqStartProcessingIrp(FSP_IOQ *Ioq, PIRP Irp)
 {
-    return STATUS_SUCCESS == IoCsqInsertIrpEx(&Ioq->ProcessIoCsq, Irp, 0, 0);
+    NTSTATUS Result;
+    Result = IoCsqInsertIrpEx(&Ioq->ProcessIoCsq, Irp, 0, 0);
+    return NT_SUCCESS(Result);
 }
 
 PIRP FspIoqEndProcessingIrp(FSP_IOQ *Ioq, UINT_PTR IrpHint)
 {
-    return IoCsqRemoveNextIrp(&Ioq->PendingIoCsq, (PVOID)IrpHint);
+    return IoCsqRemoveNextIrp(&Ioq->ProcessIoCsq, (PVOID)IrpHint);
 }
 
 VOID FspIoqCancelAll(FSP_IOQ *Ioq)
 {
     PIRP Irp;
     while (0 != (Irp = IoCsqRemoveNextIrp(&Ioq->PendingIoCsq, 0)))
-        FspCompleteRequest(Irp, STATUS_CANCELLED);
+        FspIoqPendingCompleteCanceledIrp(&Ioq->PendingIoCsq, Irp);
     while (0 != (Irp = IoCsqRemoveNextIrp(&Ioq->ProcessIoCsq, 0)))
-        FspCompleteRequest(Irp, STATUS_CANCELLED);
+        FspIoqProcessCompleteCanceledIrp(&Ioq->ProcessIoCsq, Irp);
 }