reactos/drivers/fs/np/rw.c

   1 /*
   2  * COPYRIGHT:  See COPYING in the top level directory
   3  * PROJECT:    ReactOS kernel
   4  * FILE:       drivers/fs/np/rw.c
   5  * PURPOSE:    Named pipe filesystem
   6  * PROGRAMMER: David Welch <welch@cwcom.net>
   7  */
   8
   9 /* INCLUDES ******************************************************************/
  10
  11 #define NDEBUG
  12 #include <debug.h>
  13
  14 #include "npfs.h"
  15
  16 /* FUNCTIONS *****************************************************************/
  17
  18 #ifndef NDEBUG
  19 VOID HexDump(PUCHAR Buffer, ULONG Length)
  20 {
  21   CHAR Line[65];
  22   UCHAR ch;
  23   const char Hex[] = "0123456789ABCDEF";
  24   int i, j;
  25
  26   DbgPrint("---------------\n");
  27
  28   for (i = 0; i < Length; i+= 16)
  29     {
  30       memset(Line, ' ', 64);
  31       Line[64] = 0;
  32
  33       for (j = 0; j < 16 && j + i < Length; j++)
  34         {
  35           ch = Buffer[i + j];
  36           Line[3*j + 0] = Hex[ch >> 4];
  37           Line[3*j + 1] = Hex[ch & 0x0f];
  38           Line[48 + j] = isprint(ch) ? ch : '.';
  39         }
  40       DbgPrint("%s\n", Line);
  41     }
  42   DbgPrint("---------------\n");
  43 }
  44 #endif
  45
  46 static VOID STDCALL
  47 NpfsReadWriteCancelRoutine(IN PDEVICE_OBJECT DeviceObject,
  48                            IN PIRP Irp)
  49 {
  50    PNPFS_CONTEXT Context;
  51    PNPFS_DEVICE_EXTENSION DeviceExt;
  52    PIO_STACK_LOCATION IoStack;
  53    PNPFS_FCB Fcb;
  54    BOOLEAN Complete = FALSE;
  55
  56    DPRINT("NpfsReadWriteCancelRoutine(DeviceObject %x, Irp %x)\n", DeviceObject, Irp);
  57
  58    IoReleaseCancelSpinLock(Irp->CancelIrql);
  59
  60    Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
  61    DeviceExt = (PNPFS_DEVICE_EXTENSION)DeviceObject->DeviceExtension;
  62    IoStack = IoGetCurrentIrpStackLocation(Irp);
  63    Fcb = IoStack->FileObject->FsContext;
  64
  65    KeLockMutex(&DeviceExt->PipeListLock);
  66    ExAcquireFastMutex(&Fcb->DataListLock);
  67    switch(IoStack->MajorFunction)
  68    {
  69       case IRP_MJ_READ:
  70          if (Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
  71          {
  72             /* we are not the first in the list, remove an complete us */
  73             RemoveEntryList(&Context->ListEntry);
  74             Complete = TRUE;
  75          }
  76          else
  77          {
  78             KeSetEvent(&Fcb->ReadEvent, IO_NO_INCREMENT, FALSE);
  79          }
  80          break;
  81       default:
  82          KEBUGCHECK(0);
  83    }
  84    ExReleaseFastMutex(&Fcb->DataListLock);
  85    KeUnlockMutex(&DeviceExt->PipeListLock);
  86    if (Complete)
  87    {
  88       Irp->IoStatus.Status = STATUS_CANCELLED;
  89       Irp->IoStatus.Information = 0;
  90       IoCompleteRequest(Irp, IO_NO_INCREMENT);
  91    }
  92 }
  93
  94 static VOID STDCALL
  95 NpfsWaiterThread(PVOID InitContext)
  96 {
  97    PNPFS_THREAD_CONTEXT ThreadContext = (PNPFS_THREAD_CONTEXT) InitContext;
  98    ULONG CurrentCount;
  99    ULONG Count = 0;
 100    PIRP Irp = NULL;
 101    PIRP NextIrp;
 102    NTSTATUS Status;
 103    BOOLEAN Terminate = FALSE;
 104    BOOLEAN Cancel = FALSE;
 105    PIO_STACK_LOCATION IoStack = NULL;
 106    PNPFS_CONTEXT Context;
 107    PNPFS_CONTEXT NextContext;
 108    PNPFS_FCB Fcb;
 109
 110    KeLockMutex(&ThreadContext->DeviceExt->PipeListLock);
 111
 112    while (1)
 113      {
 114        CurrentCount = ThreadContext->Count;
 115        KeUnlockMutex(&ThreadContext->DeviceExt->PipeListLock);
 116        if (Irp)
 117          {
 118            if (Cancel)
 119              {
 120                Irp->IoStatus.Status = STATUS_CANCELLED;
 121                Irp->IoStatus.Information = 0;
 122                IoCompleteRequest(Irp, IO_NO_INCREMENT);
 123              }
 124            else
 125              {
 126                switch (IoStack->MajorFunction)
 127                  {
 128                    case IRP_MJ_READ:
 129                      NpfsRead(IoStack->DeviceObject, Irp);
 130                      break;
 131                    default:
 132                      KEBUGCHECK(0);
 133                  }
 134              }
 135          }
 136        if (Terminate)
 137          {
 138            break;
 139          }
 140        Status = KeWaitForMultipleObjects(CurrentCount,
 141                                          ThreadContext->WaitObjectArray,
 142                                          WaitAny,
 143                                          Executive,
 144                                          KernelMode,
 145                                          FALSE,
 146                                          NULL,
 147                                          ThreadContext->WaitBlockArray);
 148        if (!NT_SUCCESS(Status))
 149          {
 150            KEBUGCHECK(0);
 151          }
 152        KeLockMutex(&ThreadContext->DeviceExt->PipeListLock);
 153        Count = Status - STATUS_SUCCESS;
 154        ASSERT (Count < CurrentCount);
 155        if (Count > 0)
 156        {
 157           Irp = ThreadContext->WaitIrpArray[Count];
 158           ThreadContext->Count--;
 159           ThreadContext->DeviceExt->EmptyWaiterCount++;
 160           ThreadContext->WaitObjectArray[Count] = ThreadContext->WaitObjectArray[ThreadContext->Count];
 161           ThreadContext->WaitIrpArray[Count] = ThreadContext->WaitIrpArray[ThreadContext->Count];
 162
 163           Cancel = (NULL == IoSetCancelRoutine(Irp, NULL));
 164           Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
 165           IoStack = IoGetCurrentIrpStackLocation(Irp);
 166
 167           if (Cancel)
 168           {
 169              Fcb = IoStack->FileObject->FsContext;
 170              ExAcquireFastMutex(&Fcb->DataListLock);
 171              RemoveEntryList(&Context->ListEntry);
 172              switch (IoStack->MajorFunction)
 173              {
 174                 case IRP_MJ_READ:
 175                    if (!IsListEmpty(&Fcb->ReadRequestListHead))
 176                    {
 177                       /* put the next request on the wait list */
 178                       NextContext = CONTAINING_RECORD(Fcb->ReadRequestListHead.Flink, NPFS_CONTEXT, ListEntry);
 179                       ThreadContext->WaitObjectArray[ThreadContext->Count] = NextContext->WaitEvent;
 180                       NextIrp = CONTAINING_RECORD(NextContext, IRP, Tail.Overlay.DriverContext);
 181                       ThreadContext->WaitIrpArray[ThreadContext->Count] = NextIrp;
 182                       ThreadContext->Count++;
 183                       ThreadContext->DeviceExt->EmptyWaiterCount--;
 184                    }
 185                    KeUnlockMutex(&ThreadContext->DeviceExt->PipeListLock);
 186                    break;
 187                 default:
 188                    KEBUGCHECK(0);
 189              }
 190              ExReleaseFastMutex(&Fcb->DataListLock);
 191           }
 192        }
 193        else
 194        {
 195           /* someone has add a new wait request */
 196           Irp = NULL;
 197        }
 198        if (ThreadContext->Count == 1 && ThreadContext->DeviceExt->EmptyWaiterCount >= MAXIMUM_WAIT_OBJECTS)
 199         {
 200           /* it exist an other thread with empty wait slots, we can remove our thread from the list */
 201           RemoveEntryList(&ThreadContext->ListEntry);
 202           ThreadContext->DeviceExt->EmptyWaiterCount -= MAXIMUM_WAIT_OBJECTS - 1;
 203           Terminate = TRUE;
 204         }
 205      }
 206    ExFreePool(ThreadContext);
 207 }
 208
 209 static NTSTATUS
 210 NpfsAddWaitingReadWriteRequest(IN PDEVICE_OBJECT DeviceObject,
 211                                IN PIRP Irp)
 212 {
 213    PLIST_ENTRY ListEntry;
 214    PNPFS_THREAD_CONTEXT ThreadContext = NULL;
 215    NTSTATUS Status;
 216    HANDLE hThread;
 217    KIRQL oldIrql;
 218
 219    PNPFS_CONTEXT Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
 220    PNPFS_DEVICE_EXTENSION DeviceExt = (PNPFS_DEVICE_EXTENSION)DeviceObject->DeviceExtension;
 221
 222    DPRINT("NpfsAddWaitingReadWriteRequest(DeviceObject %p, Irp %p)\n", DeviceObject, Irp);
 223
 224    KeLockMutex(&DeviceExt->PipeListLock);
 225
 226    ListEntry = DeviceExt->ThreadListHead.Flink;
 227    while (ListEntry != &DeviceExt->ThreadListHead)
 228      {
 229        ThreadContext = CONTAINING_RECORD(ListEntry, NPFS_THREAD_CONTEXT, ListEntry);
 230        if (ThreadContext->Count < MAXIMUM_WAIT_OBJECTS)
 231          {
 232            break;
 233          }
 234        ListEntry = ListEntry->Flink;
 235      }
 236    if (ListEntry == &DeviceExt->ThreadListHead)
 237      {
 238        ThreadContext = ExAllocatePool(NonPagedPool, sizeof(NPFS_THREAD_CONTEXT));
 239        if (ThreadContext == NULL)
 240          {
 241            KeUnlockMutex(&DeviceExt->PipeListLock);
 242            return STATUS_NO_MEMORY;
 243          }
 244        ThreadContext->DeviceExt = DeviceExt;
 245        KeInitializeEvent(&ThreadContext->Event, SynchronizationEvent, FALSE);
 246        ThreadContext->Count = 1;
 247        ThreadContext->WaitObjectArray[0] = &ThreadContext->Event;
 248
 249
 250        DPRINT("Creating a new system thread for waiting read/write requests\n");
 251
 252        Status = PsCreateSystemThread(&hThread,
 253                                      THREAD_ALL_ACCESS,
 254                                      NULL,
 255                                      NULL,
 256                                      NULL,
 257                                      NpfsWaiterThread,
 258                                      (PVOID)ThreadContext);
 259        if (!NT_SUCCESS(Status))
 260          {
 261            ExFreePool(ThreadContext);
 262            KeUnlockMutex(&DeviceExt->PipeListLock);
 263            return Status;
 264          }
 265        InsertHeadList(&DeviceExt->ThreadListHead, &ThreadContext->ListEntry);
 266        DeviceExt->EmptyWaiterCount += MAXIMUM_WAIT_OBJECTS - 1;
 267      }
 268    IoMarkIrpPending(Irp);
 269
 270    IoAcquireCancelSpinLock(&oldIrql);
 271    if (Irp->Cancel)
 272      {
 273        IoReleaseCancelSpinLock(oldIrql);
 274        Status = STATUS_CANCELLED;
 275      }
 276    else
 277      {
 278        IoSetCancelRoutine(Irp, NpfsReadWriteCancelRoutine);
 279        IoReleaseCancelSpinLock(oldIrql);
 280        ThreadContext->WaitObjectArray[ThreadContext->Count] = Context->WaitEvent;
 281        ThreadContext->WaitIrpArray[ThreadContext->Count] = Irp;
 282        ThreadContext->Count++;
 283        DeviceExt->EmptyWaiterCount--;
 284        KeSetEvent(&ThreadContext->Event, IO_NO_INCREMENT, FALSE);
 285        Status = STATUS_SUCCESS;
 286      }
 287    KeUnlockMutex(&DeviceExt->PipeListLock);
 288    return Status;
 289 }
 290
 291 NTSTATUS STDCALL
 292 NpfsRead(IN PDEVICE_OBJECT DeviceObject,
 293          IN PIRP Irp)
 294 {
 295   PFILE_OBJECT FileObject;
 296   NTSTATUS Status;
 297   NTSTATUS OriginalStatus = STATUS_SUCCESS;
 298   PNPFS_FCB Fcb;
 299   PNPFS_CONTEXT Context;
 300   KEVENT Event;
 301   ULONG Length;
 302   ULONG Information;
 303   ULONG CopyLength;
 304   ULONG TempLength;
 305   BOOLEAN IsOriginalRequest = TRUE;
 306   PVOID Buffer;
 307
 308   DPRINT("NpfsRead(DeviceObject %p, Irp %p)\n", DeviceObject, Irp);
 309
 310   if (Irp->MdlAddress == NULL)
 311   {
 312      DPRINT("Irp->MdlAddress == NULL\n");
 313      Status = STATUS_UNSUCCESSFUL;
 314      Irp->IoStatus.Information = 0;
 315      goto done;
 316   }
 317
 318   FileObject = IoGetCurrentIrpStackLocation(Irp)->FileObject;
 319   Fcb = FileObject->FsContext;
 320   Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
 321
 322   if (Fcb->Data == NULL)
 323   {
 324      DPRINT1("Pipe is NOT readable!\n");
 325      Status = STATUS_UNSUCCESSFUL;
 326      Irp->IoStatus.Information = 0;
 327      goto done;
 328   }
 329
 330   ExAcquireFastMutex(&Fcb->DataListLock);
 331
 332   if (IoIsOperationSynchronous(Irp))
 333   {
 334      InsertTailList(&Fcb->ReadRequestListHead, &Context->ListEntry);
 335      if (Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
 336      {
 337         KeInitializeEvent(&Event, SynchronizationEvent, FALSE);
 338         Context->WaitEvent = &Event;
 339         ExReleaseFastMutex(&Fcb->DataListLock);
 340         Status = KeWaitForSingleObject(&Event,
 341                                        Executive,
 342                                        KernelMode,
 343                                        FALSE,
 344                                        NULL);
 345         if (!NT_SUCCESS(Status))
 346         {
 347            KEBUGCHECK(0);
 348         }
 349         ExAcquireFastMutex(&Fcb->DataListLock);
 350      }
 351      Irp->IoStatus.Information = 0;
 352   }
 353   else
 354   {
 355      KIRQL oldIrql;
 356      if (IsListEmpty(&Fcb->ReadRequestListHead) ||
 357          Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
 358      {
 359         /* this is a new request */
 360         Irp->IoStatus.Information = 0;
 361         Context->WaitEvent = &Fcb->ReadEvent;
 362         InsertTailList(&Fcb->ReadRequestListHead, &Context->ListEntry);
 363         if (Fcb->ReadRequestListHead.Flink != &Context->ListEntry)
 364         {
 365            /* there was already a request on the list */
 366            IoAcquireCancelSpinLock(&oldIrql);
 367            if (Irp->Cancel)
 368            {
 369               IoReleaseCancelSpinLock(oldIrql);
 370               RemoveEntryList(&Context->ListEntry);
 371               ExReleaseFastMutex(&Fcb->DataListLock);
 372               Status = STATUS_CANCELLED;
 373               goto done;
 374            }
 375            IoSetCancelRoutine(Irp, NpfsReadWriteCancelRoutine);
 376            IoReleaseCancelSpinLock(oldIrql);
 377            ExReleaseFastMutex(&Fcb->DataListLock);
 378            IoMarkIrpPending(Irp);
 379            Status = STATUS_PENDING;
 380            goto done;
 381         }
 382      }
 383   }
 384
 385   while (1)
 386   {
 387      Buffer = MmGetSystemAddressForMdl(Irp->MdlAddress);
 388      Information = Irp->IoStatus.Information;
 389      Length = IoGetCurrentIrpStackLocation(Irp)->Parameters.Read.Length;
 390      ASSERT (Information <= Length);
 391      Buffer = (PVOID)((ULONG_PTR)Buffer + Information);
 392      Length -= Information;
 393      Status = STATUS_SUCCESS;
 394
 395      while (1)
 396      {
 397         if (Fcb->ReadDataAvailable == 0)
 398         {
 399            if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
 400            {
 401               ASSERT(Fcb->OtherSide != NULL);
 402               KeSetEvent(&Fcb->OtherSide->WriteEvent, IO_NO_INCREMENT, FALSE);
 403            }
 404            if (Information > 0 &&
 405                (Fcb->Pipe->ReadMode != FILE_PIPE_BYTE_STREAM_MODE ||
 406                 Fcb->PipeState != FILE_PIPE_CONNECTED_STATE))
 407            {
 408               break;
 409            }
 410            if (Fcb->PipeState != FILE_PIPE_CONNECTED_STATE)
 411            {
 412               DPRINT("PipeState: %x\n", Fcb->PipeState);
 413               Status = STATUS_PIPE_BROKEN;
 414               break;
 415            }
 416            ExReleaseFastMutex(&Fcb->DataListLock);
 417            if (IoIsOperationSynchronous(Irp))
 418            {
 419               /* Wait for ReadEvent to become signaled */
 420
 421               DPRINT("Waiting for readable data (%wZ)\n", &Fcb->Pipe->PipeName);
 422               Status = KeWaitForSingleObject(&Fcb->ReadEvent,
 423                                              UserRequest,
 424                                              KernelMode,
 425                                              FALSE,
 426                                              NULL);
 427               DPRINT("Finished waiting (%wZ)! Status: %x\n", &Fcb->Pipe->PipeName, Status);
 428               ExAcquireFastMutex(&Fcb->DataListLock);
 429            }
 430            else
 431            {
 432               Context = (PNPFS_CONTEXT)&Irp->Tail.Overlay.DriverContext;
 433
 434               Context->WaitEvent = &Fcb->ReadEvent;
 435               Status = NpfsAddWaitingReadWriteRequest(DeviceObject, Irp);
 436
 437               if (NT_SUCCESS(Status))
 438               {
 439                  Status = STATUS_PENDING;
 440               }
 441               ExAcquireFastMutex(&Fcb->DataListLock);
 442               break;
 443            }
 444         }
 445         ASSERT(IoGetCurrentIrpStackLocation(Irp)->FileObject != NULL);
 446         if (Fcb->Pipe->ReadMode == FILE_PIPE_BYTE_STREAM_MODE)
 447         {
 448            DPRINT("Byte stream mode\n");
 449            /* Byte stream mode */
 450            while (Length > 0 && Fcb->ReadDataAvailable > 0)
 451            {
 452               CopyLength = min(Fcb->ReadDataAvailable, Length);
 453               if ((ULONG_PTR)Fcb->ReadPtr + CopyLength <= (ULONG_PTR)Fcb->Data + Fcb->MaxDataLength)
 454               {
 455                  memcpy(Buffer, Fcb->ReadPtr, CopyLength);
 456                  Fcb->ReadPtr = (PVOID)((ULONG_PTR)Fcb->ReadPtr + CopyLength);
 457                  if (Fcb->ReadPtr == (PVOID)((ULONG_PTR)Fcb->Data + Fcb->MaxDataLength))
 458                  {
 459                     Fcb->ReadPtr = Fcb->Data;
 460                  }
 461               }
 462               else
 463               {
 464                  TempLength = (ULONG)((ULONG_PTR)Fcb->Data + Fcb->MaxDataLength - (ULONG_PTR)Fcb->ReadPtr);
 465                  memcpy(Buffer, Fcb->ReadPtr, TempLength);
 466                  memcpy((PVOID)((ULONG_PTR)Buffer + TempLength), Fcb->Data, CopyLength - TempLength);
 467                  Fcb->ReadPtr = (PVOID)((ULONG_PTR)Fcb->Data + CopyLength - TempLength);
 468               }
 469
 470               Buffer = (PVOID)((ULONG_PTR)Buffer + CopyLength);
 471               Length -= CopyLength;
 472               Information += CopyLength;
 473
 474               Fcb->ReadDataAvailable -= CopyLength;
 475               Fcb->WriteQuotaAvailable += CopyLength;
 476            }
 477
 478            if (Length == 0)
 479            {
 480               if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
 481               {
 482                  KeSetEvent(&Fcb->OtherSide->WriteEvent, IO_NO_INCREMENT, FALSE);
 483               }
 484               KeResetEvent(&Fcb->ReadEvent);
 485               break;
 486            }
 487         }
 488         else
 489         {
 490            DPRINT("Message mode\n");
 491
 492            /* Message mode */
 493            if (Fcb->ReadDataAvailable)
 494            {
 495               /* Truncate the message if the receive buffer is too small */
 496               CopyLength = min(Fcb->ReadDataAvailable, Length);
 497               memcpy(Buffer, Fcb->Data, CopyLength);
 498
 499 #ifndef NDEBUG
 500               DPRINT("Length %d Buffer %x\n",CopyLength,Buffer);
 501               HexDump((PUCHAR)Buffer, CopyLength);
 502 #endif
 503
 504               Information = CopyLength;
 505
 506               if (Fcb->ReadDataAvailable > Length)
 507               {
 508                  memmove(Fcb->Data, (PVOID)((ULONG_PTR)Fcb->Data + Length),
 509                          Fcb->ReadDataAvailable - Length);
 510                  Fcb->ReadDataAvailable -= Length;
 511                  Status = STATUS_MORE_ENTRIES;
 512               }
 513               else
 514               {
 515                  KeResetEvent(&Fcb->ReadEvent);
 516                  if (Fcb->PipeState == FILE_PIPE_CONNECTED_STATE)
 517                  {
 518                     KeSetEvent(&Fcb->OtherSide->WriteEvent, IO_NO_INCREMENT, FALSE);
 519                  }
 520                  Fcb->ReadDataAvailable = 0;
 521                  Fcb->WriteQuotaAvailable = Fcb->MaxDataLength;
 522               }
 523            }
 524
 525            if (Information > 0)
 526            {
 527               break;
 528            }
 529         }
 530      }
 531      Irp->IoStatus.Information = Information;
 532      Irp->IoStatus.Status = Status;
 533
 534      ASSERT(IoGetCurrentIrpStackLocation(Irp)->FileObject != NULL);
 535
 536      if (IoIsOperationSynchronous(Irp))
 537      {
 538         RemoveEntryList(&Context->ListEntry);
 539         if (!IsListEmpty(&Fcb->ReadRequestListHead))
 540         {
 541            Context = CONTAINING_RECORD(Fcb->ReadRequestListHead.Flink, NPFS_CONTEXT, ListEntry);
 542            KeSetEvent(Context->WaitEvent, IO_NO_INCREMENT, FALSE);
 543         }
 544         ExReleaseFastMutex(&Fcb->DataListLock);
 545         IoCompleteRequest(Irp, IO_NO_INCREMENT);
 546
 547         DPRINT("NpfsRead done (Status %lx)\n", Status);
 548         return Status;
 549      }
 550      else
 551      {
 552         if (IsOriginalRequest)
 553         {
 554            IsOriginalRequest = FALSE;
 555            OriginalStatus = Status;
 556         }
 557         if (Status == STATUS_PENDING)
 558         {
 559            ExReleaseFastMutex(&Fcb->DataListLock);
 560            DPRINT("NpfsRead done (Status %lx)\n", OriginalStatus);
 561            return OriginalStatus;
 562         }
 563         RemoveEntryList(&Context->ListEntry);
 564         IoCompleteRequest(Irp, IO_NO_INCREMENT);
 565         if (IsListEmpty(&Fcb->ReadRequestListHead))
 566         {
 567            ExReleaseFastMutex(&Fcb->DataListLock);
 568            DPRINT("NpfsRead done (Status %lx)\n", OriginalStatus);
 569            return OriginalStatus;
 570         }
 571         Context = CONTAINING_RECORD(Fcb->ReadRequestListHead.Flink, NPFS_CONTEXT, ListEntry);
 572         Irp = CONTAINING_RECORD(Context, IRP, Tail.Overlay.DriverContext);
 573      }
 574   }
 575
 576 done:
 577   Irp->IoStatus.Status = Status;
 578
 579   if (Status != STATUS_PENDING)
 580     {
 581       IoCompleteRequest(Irp, IO_NO_INCREMENT);
 582     }
 583   DPRINT("NpfsRead done (Status %lx)\n", Status);
 584
 585   return Status;
 586 }
 587
 588 NTSTATUS STDCALL
 589 NpfsWrite(PDEVICE_OBJECT DeviceObject,
 590           PIRP Irp)
 591 {
 592   PIO_STACK_LOCATION IoStack;
 593   PFILE_OBJECT FileObject;
 594   PNPFS_FCB Fcb = NULL;
 595   PNPFS_FCB ReaderFcb;
 596   PNPFS_PIPE Pipe = NULL;
 597   PUCHAR Buffer;
 598   NTSTATUS Status = STATUS_SUCCESS;
 599   ULONG Length;
 600   ULONG Offset;
 601   ULONG Information;
 602   ULONG CopyLength;
 603   ULONG TempLength;
 604
 605   DPRINT("NpfsWrite()\n");
 606
 607   IoStack = IoGetCurrentIrpStackLocation(Irp);
 608   FileObject = IoStack->FileObject;
 609   DPRINT("FileObject %p\n", FileObject);
 610   DPRINT("Pipe name %wZ\n", &FileObject->FileName);
 611
 612   Fcb = FileObject->FsContext;
 613   ReaderFcb = Fcb->OtherSide;
 614   Pipe = Fcb->Pipe;
 615
 616   Length = IoStack->Parameters.Write.Length;
 617   Offset = IoStack->Parameters.Write.ByteOffset.u.LowPart;
 618   Information = 0;
 619
 620   if (Irp->MdlAddress == NULL)
 621     {
 622       DPRINT("Irp->MdlAddress == NULL\n");
 623       Status = STATUS_UNSUCCESSFUL;
 624       Length = 0;
 625       goto done;
 626     }
 627
 628   if (ReaderFcb == NULL)
 629     {
 630       DPRINT("Pipe is NOT connected!\n");
 631       if (Fcb->PipeState == FILE_PIPE_LISTENING_STATE)
 632         Status = STATUS_PIPE_LISTENING;
 633       else if (Fcb->PipeState == FILE_PIPE_DISCONNECTED_STATE)
 634         Status = STATUS_PIPE_DISCONNECTED;
 635       else
 636         Status = STATUS_UNSUCCESSFUL;
 637       Length = 0;
 638       goto done;
 639     }
 640
 641   if (ReaderFcb->Data == NULL)
 642     {
 643       DPRINT("Pipe is NOT writable!\n");
 644       Status = STATUS_UNSUCCESSFUL;
 645       Length = 0;
 646       goto done;
 647     }
 648
 649   Status = STATUS_SUCCESS;
 650   Buffer = MmGetSystemAddressForMdl (Irp->MdlAddress);
 651
 652   ExAcquireFastMutex(&ReaderFcb->DataListLock);
 653 #ifndef NDEBUG
 654   DPRINT("Length %d Buffer %x Offset %x\n",Length,Buffer,Offset);
 655   HexDump(Buffer, Length);
 656 #endif
 657
 658   while(1)
 659     {
 660       if (ReaderFcb->WriteQuotaAvailable == 0)
 661         {
 662           KeSetEvent(&ReaderFcb->ReadEvent, IO_NO_INCREMENT, FALSE);
 663           if (Fcb->PipeState != FILE_PIPE_CONNECTED_STATE)
 664             {
 665               Status = STATUS_PIPE_BROKEN;
 666               ExReleaseFastMutex(&ReaderFcb->DataListLock);
 667               goto done;
 668             }
 669           ExReleaseFastMutex(&ReaderFcb->DataListLock);
 670
 671           DPRINT("Waiting for buffer space (%S)\n", Pipe->PipeName.Buffer);
 672           Status = KeWaitForSingleObject(&Fcb->WriteEvent,
 673                                          UserRequest,
 674                                          KernelMode,
 675                                          FALSE,
 676                                          NULL);
 677           DPRINT("Finished waiting (%S)! Status: %x\n", Pipe->PipeName.Buffer, Status);
 678
 679           ExAcquireFastMutex(&ReaderFcb->DataListLock);
 680           /*
 681            * It's possible that the event was signaled because the
 682            * other side of pipe was closed.
 683            */
 684           if (Fcb->PipeState != FILE_PIPE_CONNECTED_STATE)
 685             {
 686               DPRINT("PipeState: %x\n", Fcb->PipeState);
 687               Status = STATUS_PIPE_BROKEN;
 688               ExReleaseFastMutex(&ReaderFcb->DataListLock);
 689               goto done;
 690             }
 691         }
 692
 693       if (Pipe->WriteMode == FILE_PIPE_BYTE_STREAM_MODE)
 694         {
 695           DPRINT("Byte stream mode\n");
 696           while (Length > 0 && ReaderFcb->WriteQuotaAvailable > 0)
 697             {
 698               CopyLength = min(Length, ReaderFcb->WriteQuotaAvailable);
 699               if ((ULONG_PTR)ReaderFcb->WritePtr + CopyLength <= (ULONG_PTR)ReaderFcb->Data + ReaderFcb->MaxDataLength)
 700                 {
 701                   memcpy(ReaderFcb->WritePtr, Buffer, CopyLength);
 702                   ReaderFcb->WritePtr = (PVOID)((ULONG_PTR)ReaderFcb->WritePtr + CopyLength);
 703                   if ((ULONG_PTR)ReaderFcb->WritePtr == (ULONG_PTR)ReaderFcb->Data + ReaderFcb->MaxDataLength)
 704                     {
 705                       ReaderFcb->WritePtr = ReaderFcb->Data;
 706                     }
 707                 }
 708               else
 709                 {
 710                   TempLength = (ULONG)((ULONG_PTR)ReaderFcb->Data + ReaderFcb->MaxDataLength - (ULONG_PTR)ReaderFcb->WritePtr);
 711                   memcpy(ReaderFcb->WritePtr, Buffer, TempLength);
 712                   memcpy(ReaderFcb->Data, Buffer + TempLength, CopyLength - TempLength);
 713                   ReaderFcb->WritePtr = (PVOID)((ULONG_PTR)ReaderFcb->Data + CopyLength - TempLength);
 714                 }
 715
 716               Buffer += CopyLength;
 717               Length -= CopyLength;
 718               Information += CopyLength;
 719
 720               ReaderFcb->ReadDataAvailable += CopyLength;
 721               ReaderFcb->WriteQuotaAvailable -= CopyLength;
 722             }
 723
 724           if (Length == 0)
 725             {
 726               KeSetEvent(&ReaderFcb->ReadEvent, IO_NO_INCREMENT, FALSE);
 727               KeResetEvent(&Fcb->WriteEvent);
 728               break;
 729             }
 730         }
 731       else
 732         {
 733           DPRINT("Message mode\n");
 734           if (Length > 0)
 735             {
 736               CopyLength = min(Length, ReaderFcb->WriteQuotaAvailable);
 737               memcpy(ReaderFcb->Data, Buffer, CopyLength);
 738
 739               Information = CopyLength;
 740               ReaderFcb->ReadDataAvailable = CopyLength;
 741               ReaderFcb->WriteQuotaAvailable = 0;
 742             }
 743
 744           if (Information > 0)
 745             {
 746               KeSetEvent(&ReaderFcb->ReadEvent, IO_NO_INCREMENT, FALSE);
 747               KeResetEvent(&Fcb->WriteEvent);
 748               break;
 749             }
 750         }
 751     }
 752
 753   ExReleaseFastMutex(&ReaderFcb->DataListLock);
 754
 755 done:
 756   Irp->IoStatus.Status = Status;
 757   Irp->IoStatus.Information = Information;
 758
 759   IoCompleteRequest(Irp, IO_NO_INCREMENT);
 760
 761   DPRINT("NpfsWrite done (Status %lx)\n", Status);
 762
 763   return Status;
 764 }
 765
 766 /* EOF */