[ros-diffs] [hbirr] 14372: - Put the wait entry into the DriverContext of the irp instead to allocated it from pool.

- Put the wait entry into the DriverContext of the irp instead to
allocated it from pool.  
- Lock the data list on both ends of the pipe, if we disconnect the
pipe.    
- Implemented a read and a write event on each end of the pipe.  
- Implemented a list for read requests to deliver the requests in the
correct sequence.  
- Do not end a read request if the pipe was connected and if the buffer
wasn't filled completely.
Modified: trunk/reactos/drivers/fs/np/create.c
Modified: trunk/reactos/drivers/fs/np/fsctrl.c
Modified: trunk/reactos/drivers/fs/np/npfs.c
Modified: trunk/reactos/drivers/fs/np/npfs.h
Modified: trunk/reactos/drivers/fs/np/rw.c
  _____  

Modified: trunk/reactos/drivers/fs/np/create.c
--- trunk/reactos/drivers/fs/np/create.c	2005-03-28 18:37:39 UTC
(rev 14371)
+++ trunk/reactos/drivers/fs/np/create.c	2005-03-28 18:42:53 UTC
(rev 14372)
@@ -50,20 +50,21 @@

   PLIST_ENTRY CurrentEntry;
   PNPFS_WAITER_ENTRY Waiter;
   KIRQL oldIrql;
+  PIRP Irp;
 
   CurrentEntry = Pipe->WaiterListHead.Flink;
   while (CurrentEntry != &Pipe->WaiterListHead)
     {
       Waiter = CONTAINING_RECORD(CurrentEntry, NPFS_WAITER_ENTRY,
Entry);
-      if (Waiter->Fcb->PipeState == FILE_PIPE_LISTENING_STATE &&
-          !Waiter->Irp->Cancel)
+      Irp = CONTAINING_RECORD(Waiter, IRP, Tail.Overlay.DriverContext);
+      if (Waiter->Fcb->PipeState == FILE_PIPE_LISTENING_STATE)
 	{
 	  DPRINT("Server found! Fcb %p\n", Waiter->Fcb);
   
 	  IoAcquireCancelSpinLock(&oldIrql);
-          if (!Waiter->Irp->Cancel)
+          if (!Irp->Cancel)
 	    {
-	      IoSetCancelRoutine(Waiter->Irp, NULL);
+	      IoSetCancelRoutine(Irp, NULL);
               IoReleaseCancelSpinLock(oldIrql);
               return Waiter->Fcb;
             }
@@ -83,6 +84,7 @@
 {
   PLIST_ENTRY CurrentEntry;
   PNPFS_WAITER_ENTRY Waiter;
+  PIRP Irp;
 
   CurrentEntry = Pipe->WaiterListHead.Flink;
   while (CurrentEntry != &Pipe->WaiterListHead)
@@ -92,13 +94,12 @@
 	{
 	  DPRINT("Server found! Fcb %p\n", Waiter->Fcb);
 
-	  Waiter->Irp->IoStatus.Status = STATUS_PIPE_CONNECTED;
-	  Waiter->Irp->IoStatus.Information = 0;
-	  IoCompleteRequest(Waiter->Irp, IO_NO_INCREMENT);
-
 	  RemoveEntryList(&Waiter->Entry);
-	  ExFreePool(Waiter);
-	  return;
+	  Irp = CONTAINING_RECORD(Waiter, IRP,
Tail.Overlay.DriverContext);
+	  Irp->IoStatus.Status = STATUS_PIPE_CONNECTED;
+	  Irp->IoStatus.Information = 0;
+	  IoCompleteRequest(Irp, IO_NO_INCREMENT);
+	  break;
 	}
       CurrentEntry = CurrentEntry->Flink;
     }
@@ -175,7 +176,10 @@
   ClientFcb->PipeEnd = FILE_PIPE_CLIENT_END;
   ClientFcb->OtherSide = NULL;
   ClientFcb->PipeState = SpecialAccess ? 0 :
FILE_PIPE_DISCONNECTED_STATE;
+  InitializeListHead(&ClientFcb->ReadRequestListHead);
 
+  DPRINT("Fcb: %x\n", ClientFcb);
+
   /* Initialize data list. */
   if (Pipe->OutboundQuota)
     {
@@ -202,8 +206,10 @@
   ClientFcb->MaxDataLength = Pipe->OutboundQuota;
   ExInitializeFastMutex(&ClientFcb->DataListLock);
   KeInitializeEvent(&ClientFcb->ConnectEvent, SynchronizationEvent,
FALSE);
-  KeInitializeEvent(&ClientFcb->Event, SynchronizationEvent, FALSE);
+  KeInitializeEvent(&ClientFcb->ReadEvent, SynchronizationEvent,
FALSE);
+  KeInitializeEvent(&ClientFcb->WriteEvent, SynchronizationEvent,
FALSE);
 
+
   /*
    * Step 3. Search for listening server FCB.
    */
@@ -489,6 +495,7 @@
    Fcb->ReadDataAvailable = 0;
    Fcb->WriteQuotaAvailable = Pipe->InboundQuota;
    Fcb->MaxDataLength = Pipe->InboundQuota;
+   InitializeListHead(&Fcb->ReadRequestListHead);
    ExInitializeFastMutex(&Fcb->DataListLock);
 
    Pipe->CurrentInstances++;
@@ -498,13 +505,11 @@
    Fcb->PipeState = FILE_PIPE_LISTENING_STATE;
    Fcb->OtherSide = NULL;
 
-   KeInitializeEvent(&Fcb->ConnectEvent,
-		     SynchronizationEvent,
-		     FALSE);
+   DPRINT("Fcb: %x\n", Fcb);
 
-   KeInitializeEvent(&Fcb->Event,
-		     SynchronizationEvent,
-		     FALSE);
+   KeInitializeEvent(&Fcb->ConnectEvent, SynchronizationEvent, FALSE);
+   KeInitializeEvent(&Fcb->ReadEvent, SynchronizationEvent, FALSE);
+   KeInitializeEvent(&Fcb->WriteEvent, SynchronizationEvent, FALSE);
 
    KeLockMutex(&Pipe->FcbListLock);
    InsertTailList(&Pipe->ServerFcbListHead, &Fcb->FcbListEntry);
@@ -528,7 +533,7 @@
    PNPFS_DEVICE_EXTENSION DeviceExt;
    PIO_STACK_LOCATION IoStack;
    PFILE_OBJECT FileObject;
-   PNPFS_FCB Fcb;
+   PNPFS_FCB Fcb, OtherSide;
    PNPFS_PIPE Pipe;
    BOOL Server;
 
@@ -566,19 +571,38 @@
    {
       DPRINT("Client\n");
    }
-
    if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
    {
-      if (Fcb->OtherSide)
+      OtherSide = Fcb->OtherSide;
+      /* Lock the server first */
+      if (Server)
       {
-         Fcb->OtherSide->PipeState = FILE_PIPE_DISCONNECTED_STATE;
-         Fcb->OtherSide->OtherSide = NULL;
-         /*
-          * Signaling the write event. If is possible that an other
-          * thread waits for an empty buffer.
-          */
-         KeSetEvent(&Fcb->OtherSide->Event, IO_NO_INCREMENT, FALSE);
+         ExAcquireFastMutex(&Fcb->DataListLock);
+	 ExAcquireFastMutex(&OtherSide->DataListLock);
       }
+      else
+      {
+	 ExAcquireFastMutex(&OtherSide->DataListLock);
+         ExAcquireFastMutex(&Fcb->DataListLock);
+      }
+      OtherSide->PipeState = FILE_PIPE_DISCONNECTED_STATE;
+      OtherSide->OtherSide = NULL;
+      /*
+       * Signaling the write event. If is possible that an other
+       * thread waits for an empty buffer.
+       */
+      KeSetEvent(&OtherSide->ReadEvent, IO_NO_INCREMENT, FALSE);
+      KeSetEvent(&OtherSide->WriteEvent, IO_NO_INCREMENT, FALSE);
+      if (Server)
+      {
+         ExReleaseFastMutex(&Fcb->DataListLock);
+	 ExReleaseFastMutex(&OtherSide->DataListLock);
+      }
+      else
+      {
+	 ExReleaseFastMutex(&OtherSide->DataListLock);
+	 ExReleaseFastMutex(&Fcb->DataListLock);
+      }
    }
    else if (Fcb->PipeState == FILE_PIPE_LISTENING_STATE)
    {
@@ -586,6 +610,7 @@
       PNPFS_WAITER_ENTRY WaitEntry = NULL;
       BOOLEAN Complete = FALSE; 
       KIRQL oldIrql;
+      PIRP tmpIrp;
 
       Entry = Fcb->Pipe->WaiterListHead.Flink;
       while (Entry != &Fcb->Pipe->WaiterListHead)
@@ -594,28 +619,25 @@
 	 if (WaitEntry->Fcb == Fcb)
 	 {
             RemoveEntryList(Entry);
+	    tmpIrp = CONTAINING_RECORD(WaitEntry, IRP,
Tail.Overlay.DriverContext);
 	    IoAcquireCancelSpinLock(&oldIrql);
-	    if (!Irp->Cancel)
+	    if (!tmpIrp->Cancel)
 	    {
-               IoSetCancelRoutine(WaitEntry->Irp, NULL);
+               IoSetCancelRoutine(tmpIrp, NULL);
 	       Complete = TRUE;
 	    }
 	    IoReleaseCancelSpinLock(oldIrql);
+            if (Complete)
+	    {
+	       tmpIrp->IoStatus.Status = STATUS_PIPE_BROKEN;
+               tmpIrp->IoStatus.Information = 0;
+               IoCompleteRequest(tmpIrp, IO_NO_INCREMENT);
+	    }
 	    break;
 	 }
 	 Entry = Entry->Flink;
       }
 
-      if (Entry != &Fcb->Pipe->WaiterListHead)
-      {
-         if (Complete)
-	 {
-	    WaitEntry->Irp->IoStatus.Status = STATUS_PIPE_BROKEN;
-            WaitEntry->Irp->IoStatus.Information = 0;
-            IoCompleteRequest(WaitEntry->Irp, IO_NO_INCREMENT);
-	 }
-         ExFreePool(WaitEntry);
-      }
    }
    Fcb->PipeState = FILE_PIPE_CLOSING_STATE;
 
  _____  

Modified: trunk/reactos/drivers/fs/np/fsctrl.c
--- trunk/reactos/drivers/fs/np/fsctrl.c	2005-03-28 18:37:39 UTC
(rev 14371)
+++ trunk/reactos/drivers/fs/np/fsctrl.c	2005-03-28 18:42:53 UTC
(rev 14372)
@@ -26,18 +26,18 @@

 
   DPRINT1("NpfsListeningCancelRoutine() called\n");
 
+  Waiter = (PNPFS_WAITER_ENTRY)&Irp->Tail.Overlay.DriverContext;
+
   IoReleaseCancelSpinLock(Irp->CancelIrql);
 
-  Waiter = Irp->Tail.Overlay.DriverContext[0];
 
-  KeLockMutex(&Waiter->Pipe->FcbListLock);
+  KeLockMutex(&Waiter->Fcb->Pipe->FcbListLock);
   RemoveEntryList(&Waiter->Entry);
-  KeUnlockMutex(&Waiter->Pipe->FcbListLock);
+  KeUnlockMutex(&Waiter->Fcb->Pipe->FcbListLock);
 
   Irp->IoStatus.Status = STATUS_CANCELLED;
   Irp->IoStatus.Information = 0;
   IoCompleteRequest(Irp, IO_NO_INCREMENT);
-  ExFreePool(Waiter);
 }
 
 
@@ -48,18 +48,13 @@
   PNPFS_WAITER_ENTRY Entry;
   KIRQL oldIrql;
 
-  Entry = ExAllocatePool(NonPagedPool, sizeof(NPFS_WAITER_ENTRY));
-  if (Entry == NULL)
-    return STATUS_INSUFFICIENT_RESOURCES;
+  Entry = (PNPFS_WAITER_ENTRY)&Irp->Tail.Overlay.DriverContext;
 
-  Entry->Irp = Irp;
   Entry->Fcb = Fcb;
-  Entry->Pipe = Fcb->Pipe;
 
   KeLockMutex(&Fcb->Pipe->FcbListLock);
 
   IoMarkIrpPending(Irp);
-  Irp->Tail.Overlay.DriverContext[0] = Entry;
   InsertTailList(&Fcb->Pipe->WaiterListHead, &Entry->Entry);
 
   IoAcquireCancelSpinLock(&oldIrql);
@@ -78,7 +73,6 @@
   Irp->IoStatus.Information = 0;
   IoCompleteRequest(Irp, IO_NO_INCREMENT);
   KeUnlockMutex(&Fcb->Pipe->FcbListLock);
-  ExFreePool(Entry);
 
   return STATUS_CANCELLED;
 }
@@ -173,38 +167,100 @@
 static NTSTATUS
 NpfsDisconnectPipe(PNPFS_FCB Fcb)
 {
-  DPRINT("NpfsDisconnectPipe()\n");
+   NTSTATUS Status;
+   PNPFS_FCB OtherSide;
+   PNPFS_PIPE Pipe;
+   BOOL Server;
 
-  if (Fcb->PipeState == FILE_PIPE_DISCONNECTED_STATE)
-    return STATUS_SUCCESS;
+   DPRINT("NpfsDisconnectPipe()\n");
 
-  if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
-    {
-      Fcb->PipeState = FILE_PIPE_DISCONNECTED_STATE;
-      /* FIXME: Shouldn't this be FILE_PIPE_CLOSING_STATE? */
-      Fcb->OtherSide->PipeState = FILE_PIPE_DISCONNECTED_STATE;
+   Pipe = Fcb->Pipe;
+   KeLockMutex(&Pipe->FcbListLock);
 
-      /* FIXME: remove data queue(s) */
-
-      Fcb->OtherSide->OtherSide = NULL;
+   if (Fcb->PipeState == FILE_PIPE_DISCONNECTED_STATE)
+   {
+      DPRINT("Pipe is already disconnected\n");
+      Status = STATUS_SUCCESS;
+   }
+   else if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
+   {
+      Server = (Fcb->PipeEnd == FILE_PIPE_SERVER_END);
+      OtherSide = Fcb->OtherSide;
       Fcb->OtherSide = NULL;
+      /* Lock the server first */
+      if (Server)
+      {
+         ExAcquireFastMutex(&Fcb->DataListLock);
+	 ExAcquireFastMutex(&OtherSide->DataListLock);
+      }
+      else
+      {
+	 ExAcquireFastMutex(&OtherSide->DataListLock);
+         ExAcquireFastMutex(&Fcb->DataListLock);
+      }
+      OtherSide->PipeState = FILE_PIPE_DISCONNECTED_STATE;
+      OtherSide->OtherSide = NULL;
+      /*
+       * Signaling the write event. If is possible that an other
+       * thread waits for an empty buffer.
+       */
+      KeSetEvent(&OtherSide->ReadEvent, IO_NO_INCREMENT, FALSE);
+      KeSetEvent(&OtherSide->WriteEvent, IO_NO_INCREMENT, FALSE);
+      if (Server)
+      {
+         ExReleaseFastMutex(&Fcb->DataListLock);
+	 ExReleaseFastMutex(&OtherSide->DataListLock);
+      }
+      else
+      {
+	 ExReleaseFastMutex(&OtherSide->DataListLock);
+	 ExReleaseFastMutex(&OtherSide->DataListLock);
+      }
+      Status = STATUS_SUCCESS;
+   }
+   else if (Fcb->PipeState == FILE_PIPE_LISTENING_STATE)
+   {
+      PLIST_ENTRY Entry;
+      PNPFS_WAITER_ENTRY WaitEntry = NULL;
+      BOOLEAN Complete = FALSE; 
+      PIRP Irp = NULL;
 
-      DPRINT("Pipe disconnected\n");
-      return STATUS_SUCCESS;
-    }
+      Entry = Fcb->Pipe->WaiterListHead.Flink;
+      while (Entry != &Fcb->Pipe->WaiterListHead)
+      {
+         WaitEntry = CONTAINING_RECORD(Entry, NPFS_WAITER_ENTRY,
Entry);
+	 if (WaitEntry->Fcb == Fcb)
+	 {
+            RemoveEntryList(Entry);
+	    Irp = CONTAINING_RECORD(Entry, IRP,
Tail.Overlay.DriverContext);
+	    Complete = (NULL == IoSetCancelRoutine(Irp, NULL));
+            break;
+	 }
+	 Entry = Entry->Flink;
+      }
 
-  if (Fcb->PipeState == FILE_PIPE_CLOSING_STATE)
-    {
+      if (Irp)
+      {
+         if (Complete)
+	 {
+	    Irp->IoStatus.Status = STATUS_PIPE_BROKEN;
+            Irp->IoStatus.Information = 0;
+            IoCompleteRequest(Irp, IO_NO_INCREMENT);
+	 }
+      }
       Fcb->PipeState = FILE_PIPE_DISCONNECTED_STATE;
-      Fcb->OtherSide = NULL;
-
-      /* FIXME: remove data queue(s) */
-
-      DPRINT("Pipe disconnected\n");
-      return STATUS_SUCCESS;
-    }
-
-  return STATUS_UNSUCCESSFUL;
+      Status = STATUS_SUCCESS;
+   }
+   else if (Fcb->PipeState == FILE_PIPE_CLOSING_STATE)
+   {
+      Status = STATUS_PIPE_CLOSING;
+   }
+   else
+   {
+      Status = STATUS_UNSUCCESSFUL;
+   }
+   KeUnlockMutex(&Pipe->FcbListLock);
+   return Status;
 }
 
 
  _____  

Modified: trunk/reactos/drivers/fs/np/npfs.c
--- trunk/reactos/drivers/fs/np/npfs.c	2005-03-28 18:37:39 UTC (rev
14371)
+++ trunk/reactos/drivers/fs/np/npfs.c	2005-03-28 18:42:53 UTC (rev
14372)
@@ -27,6 +27,9 @@

    NTSTATUS Status;
    
    DPRINT("Named Pipe FSD 0.0.2\n");
+
+   ASSERT (sizeof(NPFS_CONTEXT) <= sizeof
(((PIRP)NULL)->Tail.Overlay.DriverContext));
+   ASSERT (sizeof(NPFS_WAITER_ENTRY) <=
sizeof(((PIRP)NULL)->Tail.Overlay.DriverContext));
    
    DriverObject->MajorFunction[IRP_MJ_CREATE] = NpfsCreate;
    DriverObject->MajorFunction[IRP_MJ_CREATE_NAMED_PIPE] =
@@ -74,8 +77,7 @@
    DeviceExtension = DeviceObject->DeviceExtension;
    InitializeListHead(&DeviceExtension->PipeListHead);
    InitializeListHead(&DeviceExtension->ThreadListHead);
-   KeInitializeMutex(&DeviceExtension->PipeListLock,
-		     0);
+   KeInitializeMutex(&DeviceExtension->PipeListLock, 0);
    DeviceExtension->EmptyWaiterCount = 0;
 
    /* set the size quotas */
  _____  

Modified: trunk/reactos/drivers/fs/np/npfs.h
--- trunk/reactos/drivers/fs/np/npfs.h	2005-03-28 18:37:39 UTC (rev
14371)
+++ trunk/reactos/drivers/fs/np/npfs.h	2005-03-28 18:42:53 UTC (rev
14372)
@@ -42,12 +42,15 @@

   struct ETHREAD *Thread;
   PNPFS_PIPE Pipe;
   KEVENT ConnectEvent;
-  KEVENT Event;
+  KEVENT ReadEvent;
+  KEVENT WriteEvent;
   ULONG PipeEnd;
   ULONG PipeState;
   ULONG ReadDataAvailable;
   ULONG WriteQuotaAvailable;
 
+  LIST_ENTRY ReadRequestListHead;
+
   PVOID Data;
   PVOID ReadPtr;
   PVOID WritePtr;
@@ -58,11 +61,8 @@
 
 typedef struct _NPFS_CONTEXT
 {
-  PDEVICE_OBJECT DeviceObject;
-  PIRP Irp;
-  PNPFS_FCB Fcb;
-  UCHAR MajorFunction;
-  BOOLEAN AllocatedFromPool;
+  LIST_ENTRY ListEntry;
+  PKEVENT WaitEvent;
 } NPFS_CONTEXT, *PNPFS_CONTEXT;
 
 typedef struct _NPFS_THREAD_CONTEXT
@@ -73,14 +73,12 @@
   LIST_ENTRY ListEntry;
   PVOID WaitObjectArray[MAXIMUM_WAIT_OBJECTS];
   KWAIT_BLOCK WaitBlockArray[MAXIMUM_WAIT_OBJECTS];
-  PNPFS_CONTEXT WaitContextArray[MAXIMUM_WAIT_OBJECTS];
+  PIRP WaitIrpArray[MAXIMUM_WAIT_OBJECTS];
 } NPFS_THREAD_CONTEXT, *PNPFS_THREAD_CONTEXT;
 
 typedef struct _NPFS_WAITER_ENTRY
 {
   LIST_ENTRY Entry;
-  PIRP Irp;
-  PNPFS_PIPE Pipe;
   PNPFS_FCB Fcb;
 } NPFS_WAITER_ENTRY, *PNPFS_WAITER_ENTRY;
 
  _____  

Modified: trunk/reactos/drivers/fs/np/rw.c
--- trunk/reactos/drivers/fs/np/rw.c	2005-03-28 18:37:39 UTC (rev
14371)
+++ trunk/reactos/drivers/fs/np/rw.c	2005-03-28 18:42:53 UTC (rev
14372)
@@ -46,61 +46,90 @@

 }
 #endif
 
-static NTSTATUS
-NpfsReadFromPipe(PNPFS_CONTEXT Context);
-
 static VOID STDCALL 
-NpfsWaitingCancelRoutine(IN PDEVICE_OBJECT DeviceObject,
-                         IN PIRP Irp)
+NpfsReadWriteCancelRoutine(IN PDEVICE_OBJECT DeviceObject,
+                           IN PIRP Irp)
 {
    PNPFS_CONTEXT Context;
    PNPFS_DEVICE_EXTENSION DeviceExt;
+   PIO_STACK_LOCATION IoStack;
+   PNPFS_FCB Fcb;
+   BOOLEAN Complete = FALSE;
 
-   DPRINT1("NpfsWaitingCancelRoutine() called\n");
+   DPRINT("NpfsReadWriteCancelRoutine(DeviceObject %x, Irp %x)\n",
DeviceObject, Irp);
 
    IoReleaseCancelSpinLock(Irp->CancelIrql);
 
-   Context = Irp->Tail.Overlay.DriverContext[0];
-   DeviceExt = Context->DeviceObject->DeviceExtension;
+   Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
+   DeviceExt = (PNPFS_DEVICE_EXTENSION)DeviceObject->DeviceExtension;
+   IoStack = IoGetCurrentIrpStackLocation(Irp);
+   Fcb = IoStack->FileObject->FsContext;
 
    KeLockMutex(&DeviceExt->PipeListLock);
-   KeSetEvent(&Context->Fcb->Event, IO_NO_INCREMENT, FALSE);
+   ExAcquireFastMutex(&Fcb->DataListLock);
+   switch(IoStack->MajorFunction)   
+   {
+      case IRP_MJ_READ:
+         if (Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
+	 {
+	    /* we are not the first in the list, remove an complete us
*/
+	    RemoveEntryList(&Context->ListEntry);
+	    Complete = TRUE;
+	 }
+	 else
+	 {
+	    KeSetEvent(&Fcb->ReadEvent, IO_NO_INCREMENT, FALSE);
+	 }
+	 break;
+      default:
+         KEBUGCHECK(0);
+   }
+   ExReleaseFastMutex(&Fcb->DataListLock);
    KeUnlockMutex(&DeviceExt->PipeListLock);
+   if (Complete)
+   {
+      Irp->IoStatus.Status = STATUS_CANCELLED;
+      Irp->IoStatus.Information = 0;
+      IoCompleteRequest(Irp, IO_NO_INCREMENT);
+   }
 }
 
 static VOID STDCALL
-NpfsWaiterThread(PVOID Context)
+NpfsWaiterThread(PVOID InitContext)
 {
-   PNPFS_THREAD_CONTEXT ThreadContext = (PNPFS_THREAD_CONTEXT) Context;
-   ULONG CurrentCount, Count = 0;
-   PNPFS_CONTEXT WaitContext = NULL;
+   PNPFS_THREAD_CONTEXT ThreadContext = (PNPFS_THREAD_CONTEXT)
InitContext;
+   ULONG CurrentCount;
+   ULONG Count = 0;
+   PIRP Irp = NULL;
+   PIRP NextIrp;
    NTSTATUS Status;
    BOOLEAN Terminate = FALSE;
    BOOLEAN Cancel = FALSE;
-   KIRQL oldIrql;
+   PIO_STACK_LOCATION IoStack = NULL;
+   PNPFS_CONTEXT Context;
+   PNPFS_CONTEXT NextContext;
+   PNPFS_FCB Fcb;
 
    KeLockMutex(&ThreadContext->DeviceExt->PipeListLock);
 
    while (1)
      {
        CurrentCount = ThreadContext->Count;
-       KeResetEvent(&ThreadContext->Event);
        KeUnlockMutex(&ThreadContext->DeviceExt->PipeListLock);
-       if (WaitContext)
+       if (Irp)
          {
            if (Cancel)
              {
-	       WaitContext->Irp->IoStatus.Status = STATUS_CANCELLED;
-               WaitContext->Irp->IoStatus.Information = 0;
-               IoCompleteRequest(WaitContext->Irp, IO_NO_INCREMENT);
-	       ExFreePool(WaitContext);
+	       Irp->IoStatus.Status = STATUS_CANCELLED;
+               Irp->IoStatus.Information = 0;
+               IoCompleteRequest(Irp, IO_NO_INCREMENT);
 	     }
 	   else
 	     {
-	       switch (WaitContext->MajorFunction)
+               switch (IoStack->MajorFunction)
 	         {
 	           case IRP_MJ_READ:
-                     NpfsReadFromPipe(WaitContext);
+                     NpfsRead(IoStack->DeviceObject, Irp);
 		     break;
 		   default:
 		     KEBUGCHECK(0);
@@ -119,30 +148,56 @@
 					 FALSE,
 					 NULL,
 					 ThreadContext->WaitBlockArray);
-       KeLockMutex(&ThreadContext->DeviceExt->PipeListLock);
        if (!NT_SUCCESS(Status))
          {
            KEBUGCHECK(0);
          }
+       KeLockMutex(&ThreadContext->DeviceExt->PipeListLock);
        Count = Status - STATUS_SUCCESS;
-       ASSERT (Count <= CurrentCount);
+       ASSERT (Count < CurrentCount);
        if (Count > 0)
-         {
-	   WaitContext = ThreadContext->WaitContextArray[Count];
-	   ThreadContext->Count--;
-	   ThreadContext->DeviceExt->EmptyWaiterCount++;
-	   ThreadContext->WaitObjectArray[Count] =
ThreadContext->WaitObjectArray[ThreadContext->Count];
-	   ThreadContext->WaitContextArray[Count] =
ThreadContext->WaitContextArray[ThreadContext->Count];
-           IoAcquireCancelSpinLock(&oldIrql);
-	   Cancel = NULL == IoSetCancelRoutine(WaitContext->Irp, NULL);
-           IoReleaseCancelSpinLock(oldIrql);
-        }
-      else
-        {
+       {
+	  Irp = ThreadContext->WaitIrpArray[Count];
+	  ThreadContext->Count--;
+	  ThreadContext->DeviceExt->EmptyWaiterCount++;
+	  ThreadContext->WaitObjectArray[Count] =
ThreadContext->WaitObjectArray[ThreadContext->Count];
+	  ThreadContext->WaitIrpArray[Count] =
ThreadContext->WaitIrpArray[ThreadContext->Count];
+
+          Cancel = (NULL == IoSetCancelRoutine(Irp, NULL));
+	  Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
+	  IoStack = IoGetCurrentIrpStackLocation(Irp);
+
+	  if (Cancel)
+	  {
+	     Fcb = IoStack->FileObject->FsContext;
+ 	     ExAcquireFastMutex(&Fcb->DataListLock);
+	     RemoveEntryList(&Context->ListEntry);
+	     switch (IoStack->MajorFunction)
+	     {
+	        case IRP_MJ_READ:
+                   if (!IsListEmpty(&Fcb->ReadRequestListHead))
+		   {
+		      /* put the next request on the wait list */
+                      NextContext =
CONTAINING_RECORD(Fcb->ReadRequestListHead.Flink, NPFS_CONTEXT,
ListEntry);
+
ThreadContext->WaitObjectArray[ThreadContext->Count] =
NextContext->WaitEvent;
+		      NextIrp = CONTAINING_RECORD(NextContext, IRP,
Tail.Overlay.DriverContext);
+		      ThreadContext->WaitIrpArray[ThreadContext->Count]
= NextIrp;
+		      ThreadContext->Count++;
+                      ThreadContext->DeviceExt->EmptyWaiterCount--;
+		   }
+		   break;
+		default:
+		   KEBUGCHECK(0);
+	     }
+	     ExReleaseFastMutex(&Fcb->DataListLock);
+	  }	    
+       }
+       else
+       {  
 	  /* someone has add a new wait request */
-          WaitContext = NULL;
-        }
-      if (ThreadContext->Count == 1 &&
ThreadContext->DeviceExt->EmptyWaiterCount >= MAXIMUM_WAIT_OBJECTS)
+          Irp = NULL;
+       }
+       if (ThreadContext->Count == 1 &&
ThreadContext->DeviceExt->EmptyWaiterCount >= MAXIMUM_WAIT_OBJECTS)
         {
           /* it exist an other thread with empty wait slots, we can
remove our thread from the list */
           RemoveEntryList(&ThreadContext->ListEntry);
@@ -155,14 +210,20 @@
 }
 
 static NTSTATUS
-NpfsAddWaitingReader(PNPFS_DEVICE_EXTENSION DeviceExt, PNPFS_CONTEXT
Context, PNPFS_FCB Fcb)
+NpfsAddWaitingReadWriteRequest(IN PDEVICE_OBJECT DeviceObject,
+			       IN PIRP Irp)
 {
    PLIST_ENTRY ListEntry;
-   PNPFS_THREAD_CONTEXT ThreadContext;
+   PNPFS_THREAD_CONTEXT ThreadContext = NULL;
    NTSTATUS Status;
    HANDLE hThread;
    KIRQL oldIrql;
 
+   PNPFS_CONTEXT Context =
(PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
+   PNPFS_DEVICE_EXTENSION DeviceExt =
(PNPFS_DEVICE_EXTENSION)DeviceObject->DeviceExtension;;
+
+   DPRINT("NpfsAddWaitingReadWriteRequest(DeviceObject %p, Irp %p)\n",
DeviceObject, Irp);
+
    KeLockMutex(&DeviceExt->PipeListLock);
 
    ListEntry = DeviceExt->ThreadListHead.Flink;
@@ -184,12 +245,12 @@
            return STATUS_NO_MEMORY;
          }
        ThreadContext->DeviceExt = DeviceExt;
-       KeInitializeEvent(&ThreadContext->Event, NotificationEvent,
FALSE);
+       KeInitializeEvent(&ThreadContext->Event, SynchronizationEvent,
FALSE);
        ThreadContext->Count = 1;
        ThreadContext->WaitObjectArray[0] = &ThreadContext->Event;
 
    
-       DPRINT("Creating a new system thread for waiting read
requests\n");
+       DPRINT("Creating a new system thread for waiting read/write
requests\n");
       
        Status = PsCreateSystemThread(&hThread,
 		                     THREAD_ALL_ACCESS,
@@ -207,21 +268,20 @@
        InsertHeadList(&DeviceExt->ThreadListHead,
&ThreadContext->ListEntry);
        DeviceExt->EmptyWaiterCount += MAXIMUM_WAIT_OBJECTS - 1;       
      }
-   IoMarkIrpPending(Context->Irp);
-   Context->Irp->Tail.Overlay.DriverContext[0] = Context;
+   IoMarkIrpPending(Irp);
 
    IoAcquireCancelSpinLock(&oldIrql);
-   if (Context->Irp->Cancel)
+   if (Irp->Cancel)
      {
        IoReleaseCancelSpinLock(oldIrql);
        Status = STATUS_CANCELLED;
      }
    else
      {
-       IoSetCancelRoutine(Context->Irp, NpfsWaitingCancelRoutine);
+       IoSetCancelRoutine(Irp, NpfsReadWriteCancelRoutine);
        IoReleaseCancelSpinLock(oldIrql);
-       ThreadContext->WaitObjectArray[ThreadContext->Count] =
&Fcb->Event;
-       ThreadContext->WaitContextArray[ThreadContext->Count] = Context;
+       ThreadContext->WaitObjectArray[ThreadContext->Count] =
Context->WaitEvent;
+       ThreadContext->WaitIrpArray[ThreadContext->Count] = Irp;
        ThreadContext->Count++;
        DeviceExt->EmptyWaiterCount--;
        KeSetEvent(&ThreadContext->Event, IO_NO_INCREMENT, FALSE);
@@ -231,130 +291,182 @@
    return Status;
 }
 
-static NTSTATUS
-NpfsReadFromPipe(PNPFS_CONTEXT Context)
+NTSTATUS STDCALL
+NpfsRead(IN PDEVICE_OBJECT DeviceObject,
+	 IN PIRP Irp)
 {
-  PIO_STACK_LOCATION IoStack;
   PFILE_OBJECT FileObject;
   NTSTATUS Status;
-  ULONG Information;
+  NTSTATUS OriginalStatus = STATUS_SUCCESS;
   PNPFS_FCB Fcb;
-  PNPFS_FCB WriterFcb;
-  PNPFS_PIPE Pipe;
+  PNPFS_CONTEXT Context;
+  KEVENT Event;
   ULONG Length;
-  PVOID Buffer;
+  ULONG Information;
   ULONG CopyLength;
   ULONG TempLength;
+  BOOLEAN IsOriginalRequest = TRUE;
+  PVOID Buffer;
 
-  DPRINT("NpfsReadFromPipe(Context %p)\n", Context);
+  DPRINT("NpfsRead(DeviceObject %p, Irp %p)\n", DeviceObject, Irp);
 
-  IoStack = IoGetCurrentIrpStackLocation(Context->Irp);
-  FileObject = IoStack->FileObject;
+  if (Irp->MdlAddress == NULL)
+  {
+     DPRINT("Irp->MdlAddress == NULL\n");
+     Status = STATUS_UNSUCCESSFUL;
+     Irp->IoStatus.Information = 0;
+     goto done;
+  }
+
+  FileObject = IoGetCurrentIrpStackLocation(Irp)->FileObject;
   Fcb = FileObject->FsContext;
-  Pipe = Fcb->Pipe;
-  WriterFcb = Fcb->OtherSide;
+  Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
 
   if (Fcb->Data == NULL)
-    {
-      DPRINT("Pipe is NOT readable!\n");
-      Status = STATUS_UNSUCCESSFUL;
-      Information = 0;
-      goto done;
-    }
+  {
+     DPRINT1("Pipe is NOT readable!\n");
+     Status = STATUS_UNSUCCESSFUL;
+     Irp->IoStatus.Information = 0;
+     goto done;
+  }
 
-  Status = STATUS_SUCCESS;
-  Length = IoStack->Parameters.Read.Length;
-  Information = 0;
+  ExAcquireFastMutex(&Fcb->DataListLock);
 
-  Buffer = MmGetSystemAddressForMdl(Context->Irp->MdlAddress);
-  ExAcquireFastMutex(&Fcb->DataListLock);
-  while (1)
-    {
-      if (Fcb->ReadDataAvailable == 0)
+  if (IoIsOperationSynchronous(Irp))
+  {
+     InsertTailList(&Fcb->ReadRequestListHead, &Context->ListEntry);
+     if (Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
+     {
+        KeInitializeEvent(&Event, SynchronizationEvent, FALSE);
+	Context->WaitEvent = &Event;
+        ExReleaseFastMutex(&Fcb->DataListLock);
+        Status = KeWaitForSingleObject(&Event,
+	                               Executive,
+				       KernelMode,
+				       FALSE,
+				       NULL);
+	if (!NT_SUCCESS(Status))
 	{
-	  if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
-	    {
-	      KeSetEvent(&WriterFcb->Event, IO_NO_INCREMENT, FALSE);
-	    }
-	  ExReleaseFastMutex(&Fcb->DataListLock);
-	  if (Information > 0)
-	    {
-	      Status = STATUS_SUCCESS;
+	   KEBUGCHECK(0);
+	}
+        ExAcquireFastMutex(&Fcb->DataListLock);
+     }
+     Irp->IoStatus.Information = 0;
+  }
+  else
+  {
+     KIRQL oldIrql;
+     if (IsListEmpty(&Fcb->ReadRequestListHead) ||
+	 Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
+     {
+        /* this is a new request */
+        Irp->IoStatus.Information = 0;
+	Context->WaitEvent = &Fcb->ReadEvent;
+        InsertTailList(&Fcb->ReadRequestListHead, &Context->ListEntry);
+	if (Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
+	{
+	   /* there was already a request on the list */
+           IoAcquireCancelSpinLock(&oldIrql);
+           if (Irp->Cancel)
+           {
+	      IoReleaseCancelSpinLock(oldIrql);
+	      RemoveEntryList(&Context->ListEntry);
+	      ExReleaseFastMutex(&Fcb->DataListLock);
+	      Status = STATUS_CANCELLED;
 	      goto done;
-	    }
+           }
+           IoSetCancelRoutine(Irp, NpfsReadWriteCancelRoutine);
+           IoReleaseCancelSpinLock(oldIrql);
+	   ExReleaseFastMutex(&Fcb->DataListLock);
+           IoMarkIrpPending(Irp);
+	   Status = STATUS_PENDING;
+	   goto done;
+	}
+     }
+  }
 
-	  if (Fcb->PipeState != FILE_PIPE_CONNECTED_STATE)
-	    {
+  while (1)
+  {
+     Buffer = MmGetSystemAddressForMdl(Irp->MdlAddress);
+     Information = Irp->IoStatus.Information;
+     Length =
IoGetCurrentIrpStackLocation(Irp)->Parameters.Read.Length;
+     ASSERT (Information <= Length);
+     Buffer += Information;
+     Length -= Information;
+     Status = STATUS_SUCCESS;
+
+     while (1)
+     {
+        if (Fcb->ReadDataAvailable == 0)
+        {
+	   if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
+	   {
+	      KeSetEvent(&Fcb->OtherSide->WriteEvent, IO_NO_INCREMENT,
FALSE);
+	   }
+	   if (Information > 0 &&
+	       (Fcb->Pipe->ReadMode != FILE_PIPE_BYTE_STREAM_MODE ||
+	        Fcb->PipeState != FILE_PIPE_CONNECTED_STATE))
+	   {
+	      break;
+	   }
+      	   if (Fcb->PipeState != FILE_PIPE_CONNECTED_STATE)
+	   {
 	      DPRINT("PipeState: %x\n", Fcb->PipeState);
 	      Status = STATUS_PIPE_BROKEN;
-	      goto done;
-	    }
+	      break;
+           }
+	   ExReleaseFastMutex(&Fcb->DataListLock);
+	   if (IoIsOperationSynchronous(Irp))
+	   {
+	      /* Wait for ReadEvent to become signaled */
 
-	  if (IoIsOperationSynchronous(Context->Irp))
-	    {
-	      /* Wait for ReadEvent to become signaled */
-	      DPRINT("Waiting for readable data (%S)\n",
Pipe->PipeName.Buffer);
-	      Status = KeWaitForSingleObject(&Fcb->Event,
+	      DPRINT("Waiting for readable data (%wZ)\n",
&Fcb->Pipe->PipeName);
+	      Status = KeWaitForSingleObject(&Fcb->ReadEvent,
 				             UserRequest,
 				             KernelMode,
 				             FALSE,
 				             NULL);
-	      DPRINT("Finished waiting (%S)! Status: %x\n",
Pipe->PipeName.Buffer, Status);
-	    }
-	  else
-	    {
-	      PNPFS_CONTEXT NewContext;
-
-	      NewContext = ExAllocatePool(NonPagedPool,
sizeof(NPFS_CONTEXT));
-	      if (NewContext == NULL)
-	        {
-		   Status = STATUS_NO_MEMORY;
-		   goto done;
-		}
-	      memcpy(NewContext, Context, sizeof(NPFS_CONTEXT));
-	      NewContext->AllocatedFromPool = TRUE;
-	      NewContext->Fcb = Fcb;
-	      NewContext->MajorFunction = IRP_MJ_READ;
-
-	      Status =
NpfsAddWaitingReader(Context->DeviceObject->DeviceExtension, NewContext,
Fcb);
+	      DPRINT("Finished waiting (%wZ)! Status: %x\n",
&Fcb->Pipe->PipeName, Status);
+	      ExAcquireFastMutex(&Fcb->DataListLock);
+	   }
+	   else
+	   {
+              PNPFS_CONTEXT Context =
(PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
+	    
+              Context->WaitEvent = &Fcb->ReadEvent;
+	      Status = NpfsAddWaitingReadWriteRequest(DeviceObject,
Irp);
 	         
 	      if (NT_SUCCESS(Status))
-	        {
-		  Status = STATUS_PENDING;
-		}
-	      else
-	        {
-		  ExFreePool(NewContext);
-		}
-	      goto done;
-	    }
-
-	  ExAcquireFastMutex(&Fcb->DataListLock);
-	}
-
-      if (Pipe->ReadMode == FILE_PIPE_BYTE_STREAM_MODE)
-	{
-	  DPRINT("Byte stream mode\n");
-	  /* Byte stream mode */
-	  while (Length > 0 && Fcb->ReadDataAvailable > 0)
-	    {
+	      {
+	         Status = STATUS_PENDING;
+	      }
+	      ExAcquireFastMutex(&Fcb->DataListLock);
+	      break;
+	   }
+        } 
+        if (Fcb->Pipe->ReadMode == FILE_PIPE_BYTE_STREAM_MODE)
+        {
+	   DPRINT("Byte stream mode\n");
+	   /* Byte stream mode */
+	   while (Length > 0 && Fcb->ReadDataAvailable > 0)
+	   {
 	      CopyLength = RtlRosMin(Fcb->ReadDataAvailable, Length);
 	      if (Fcb->ReadPtr + CopyLength <= Fcb->Data +
Fcb->MaxDataLength)
-		{
-		  memcpy(Buffer, Fcb->ReadPtr, CopyLength);
-		  Fcb->ReadPtr += CopyLength;
-		  if (Fcb->ReadPtr == Fcb->Data + Fcb->MaxDataLength)
-		    {
-		      Fcb->ReadPtr = Fcb->Data;
-		    }
-		}
-	      else
-		{
-		  TempLength = Fcb->Data + Fcb->MaxDataLength -
Fcb->ReadPtr;
-		  memcpy(Buffer, Fcb->ReadPtr, TempLength);
-		  memcpy(Buffer + TempLength, Fcb->Data, CopyLength -
TempLength);
-		  Fcb->ReadPtr = Fcb->Data + CopyLength - TempLength;
-		}
+	      {
+	         memcpy(Buffer, Fcb->ReadPtr, CopyLength);
+	         Fcb->ReadPtr += CopyLength;
+	         if (Fcb->ReadPtr == Fcb->Data + Fcb->MaxDataLength)
+	         {
+		    Fcb->ReadPtr = Fcb->Data;
+	         }
+	      }
+              else
+	      {
+	         TempLength = Fcb->Data + Fcb->MaxDataLength -
Fcb->ReadPtr;
+	         memcpy(Buffer, Fcb->ReadPtr, TempLength);
+	         memcpy(Buffer + TempLength, Fcb->Data, CopyLength -
TempLength);
+	         Fcb->ReadPtr = Fcb->Data + CopyLength - TempLength;
+	      }
 
 	      Buffer += CopyLength;
 	      Length -= CopyLength;
[truncated at 1000 lines; 243 more skipped]