xpcom/io/nsPipe3.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=8 sts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this
   5  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "mozilla/Attributes.h"
   8 #include "mozilla/ReentrantMonitor.h"
   9 #include "nsIPipe.h"
  10 #include "nsIEventTarget.h"
  11 #include "nsISeekableStream.h"
  12 #include "nsIProgrammingLanguage.h"
  13 #include "nsSegmentedBuffer.h"
  14 #include "nsStreamUtils.h"
  15 #include "nsCOMPtr.h"
  16 #include "nsCRT.h"
  17 #include "prlog.h"
  18 #include "nsIClassInfoImpl.h"
  19 #include "nsAlgorithm.h"
  20 #include "nsMemory.h"
  21 #include "nsIAsyncInputStream.h"
  22 #include "nsIAsyncOutputStream.h"
  23
  24 using namespace mozilla;
  25
  26 #ifdef LOG
  27 #undef LOG
  28 #endif
  29 #if defined(PR_LOGGING)
  30 //
  31 // set NSPR_LOG_MODULES=nsPipe:5
  32 //
  33 static PRLogModuleInfo*
  34 GetPipeLog()
  35 {
  36   static PRLogModuleInfo* sLog;
  37   if (!sLog) {
  38     sLog = PR_NewLogModule("nsPipe");
  39   }
  40   return sLog;
  41 }
  42 #define LOG(args) PR_LOG(GetPipeLog(), PR_LOG_DEBUG, args)
  43 #else
  44 #define LOG(args)
  45 #endif
  46
  47 #define DEFAULT_SEGMENT_SIZE  4096
  48 #define DEFAULT_SEGMENT_COUNT 16
  49
  50 class nsPipe;
  51 class nsPipeEvents;
  52 class nsPipeInputStream;
  53 class nsPipeOutputStream;
  54
  55 //-----------------------------------------------------------------------------
  56
  57 // this class is used to delay notifications until the end of a particular
  58 // scope.  it helps avoid the complexity of issuing callbacks while inside
  59 // a critical section.
  60 class nsPipeEvents
  61 {
  62 public:
  63   nsPipeEvents() { }
  64   ~nsPipeEvents();
  65
  66   inline void NotifyInputReady(nsIAsyncInputStream* aStream,
  67                                nsIInputStreamCallback* aCallback)
  68   {
  69     NS_ASSERTION(!mInputCallback, "already have an input event");
  70     mInputStream = aStream;
  71     mInputCallback = aCallback;
  72   }
  73
  74   inline void NotifyOutputReady(nsIAsyncOutputStream* aStream,
  75                                 nsIOutputStreamCallback* aCallback)
  76   {
  77     NS_ASSERTION(!mOutputCallback, "already have an output event");
  78     mOutputStream = aStream;
  79     mOutputCallback = aCallback;
  80   }
  81
  82 private:
  83   nsCOMPtr<nsIAsyncInputStream>     mInputStream;
  84   nsCOMPtr<nsIInputStreamCallback>  mInputCallback;
  85   nsCOMPtr<nsIAsyncOutputStream>    mOutputStream;
  86   nsCOMPtr<nsIOutputStreamCallback> mOutputCallback;
  87 };
  88
  89 //-----------------------------------------------------------------------------
  90
  91 // the input end of a pipe (allocated as a member of the pipe).
  92 class nsPipeInputStream
  93   : public nsIAsyncInputStream
  94   , public nsISeekableStream
  95   , public nsISearchableInputStream
  96   , public nsIClassInfo
  97 {
  98 public:
  99   // since this class will be allocated as a member of the pipe, we do not
 100   // need our own ref count.  instead, we share the lifetime (the ref count)
 101   // of the entire pipe.  this macro is just convenience since it does not
 102   // declare a mRefCount variable; however, don't let the name fool you...
 103   // we are not inheriting from nsPipe ;-)
 104   NS_DECL_ISUPPORTS_INHERITED
 105
 106   NS_DECL_NSIINPUTSTREAM
 107   NS_DECL_NSIASYNCINPUTSTREAM
 108   NS_DECL_NSISEEKABLESTREAM
 109   NS_DECL_NSISEARCHABLEINPUTSTREAM
 110   NS_DECL_NSICLASSINFO
 111
 112   explicit nsPipeInputStream(nsPipe* aPipe)
 113     : mPipe(aPipe)
 114     , mReaderRefCnt(0)
 115     , mLogicalOffset(0)
 116     , mBlocking(true)
 117     , mBlocked(false)
 118     , mAvailable(0)
 119     , mCallbackFlags(0)
 120   { }
 121
 122   nsresult Fill();
 123   void SetNonBlocking(bool aNonBlocking)
 124   {
 125     mBlocking = !aNonBlocking;
 126   }
 127
 128   uint32_t Available()
 129   {
 130     return mAvailable;
 131   }
 132   void ReduceAvailable(uint32_t aAvail)
 133   {
 134     mAvailable -= aAvail;
 135   }
 136
 137   // synchronously wait for the pipe to become readable.
 138   nsresult Wait();
 139
 140   // these functions return true to indicate that the pipe's monitor should
 141   // be notified, to wake up a blocked reader if any.
 142   bool OnInputReadable(uint32_t aBytesWritten, nsPipeEvents&);
 143   bool OnInputException(nsresult, nsPipeEvents&);
 144
 145 private:
 146   nsPipe*                        mPipe;
 147
 148   // separate refcnt so that we know when to close the consumer
 149   mozilla::ThreadSafeAutoRefCnt  mReaderRefCnt;
 150   int64_t                        mLogicalOffset;
 151   bool                           mBlocking;
 152
 153   // these variables can only be accessed while inside the pipe's monitor
 154   bool                           mBlocked;
 155   uint32_t                       mAvailable;
 156   nsCOMPtr<nsIInputStreamCallback> mCallback;
 157   uint32_t                       mCallbackFlags;
 158 };
 159
 160 //-----------------------------------------------------------------------------
 161
 162 // the output end of a pipe (allocated as a member of the pipe).
 163 class nsPipeOutputStream
 164   : public nsIAsyncOutputStream
 165   , public nsIClassInfo
 166 {
 167 public:
 168   // since this class will be allocated as a member of the pipe, we do not
 169   // need our own ref count.  instead, we share the lifetime (the ref count)
 170   // of the entire pipe.  this macro is just convenience since it does not
 171   // declare a mRefCount variable; however, don't let the name fool you...
 172   // we are not inheriting from nsPipe ;-)
 173   NS_DECL_ISUPPORTS_INHERITED
 174
 175   NS_DECL_NSIOUTPUTSTREAM
 176   NS_DECL_NSIASYNCOUTPUTSTREAM
 177   NS_DECL_NSICLASSINFO
 178
 179   explicit nsPipeOutputStream(nsPipe* aPipe)
 180     : mPipe(aPipe)
 181     , mWriterRefCnt(0)
 182     , mLogicalOffset(0)
 183     , mBlocking(true)
 184     , mBlocked(false)
 185     , mWritable(true)
 186     , mCallbackFlags(0)
 187   { }
 188
 189   void SetNonBlocking(bool aNonBlocking)
 190   {
 191     mBlocking = !aNonBlocking;
 192   }
 193   void SetWritable(bool aWritable)
 194   {
 195     mWritable = aWritable;
 196   }
 197
 198   // synchronously wait for the pipe to become writable.
 199   nsresult Wait();
 200
 201   // these functions return true to indicate that the pipe's monitor should
 202   // be notified, to wake up a blocked writer if any.
 203   bool OnOutputWritable(nsPipeEvents&);
 204   bool OnOutputException(nsresult, nsPipeEvents&);
 205
 206 private:
 207   nsPipe*                         mPipe;
 208
 209   // separate refcnt so that we know when to close the producer
 210   mozilla::ThreadSafeAutoRefCnt   mWriterRefCnt;
 211   int64_t                         mLogicalOffset;
 212   bool                            mBlocking;
 213
 214   // these variables can only be accessed while inside the pipe's monitor
 215   bool                            mBlocked;
 216   bool                            mWritable;
 217   nsCOMPtr<nsIOutputStreamCallback> mCallback;
 218   uint32_t                        mCallbackFlags;
 219 };
 220
 221 //-----------------------------------------------------------------------------
 222
 223 class nsPipe MOZ_FINAL : public nsIPipe
 224 {
 225 public:
 226   friend class nsPipeInputStream;
 227   friend class nsPipeOutputStream;
 228
 229   NS_DECL_THREADSAFE_ISUPPORTS
 230   NS_DECL_NSIPIPE
 231
 232   // nsPipe methods:
 233   nsPipe();
 234
 235 private:
 236   ~nsPipe();
 237
 238 public:
 239   //
 240   // methods below may only be called while inside the pipe's monitor
 241   //
 242
 243   void PeekSegment(uint32_t aIndex, char*& aCursor, char*& aLimit);
 244
 245   //
 246   // methods below may be called while outside the pipe's monitor
 247   //
 248
 249   nsresult GetReadSegment(const char*& aSegment, uint32_t& aSegmentLen);
 250   void     AdvanceReadCursor(uint32_t aCount);
 251
 252   nsresult GetWriteSegment(char*& aSegment, uint32_t& aSegmentLen);
 253   void     AdvanceWriteCursor(uint32_t aCount);
 254
 255   void     OnPipeException(nsresult aReason, bool aOutputOnly = false);
 256
 257 protected:
 258   // We can't inherit from both nsIInputStream and nsIOutputStream
 259   // because they collide on their Close method. Consequently we nest their
 260   // implementations to avoid the extra object allocation.
 261   nsPipeInputStream   mInput;
 262   nsPipeOutputStream  mOutput;
 263
 264   ReentrantMonitor    mReentrantMonitor;
 265   nsSegmentedBuffer   mBuffer;
 266
 267   char*               mReadCursor;
 268   char*               mReadLimit;
 269
 270   int32_t             mWriteSegment;
 271   char*               mWriteCursor;
 272   char*               mWriteLimit;
 273
 274   nsresult            mStatus;
 275   bool                mInited;
 276 };
 277
 278 //
 279 // NOTES on buffer architecture:
 280 //
 281 //       +-----------------+ - - mBuffer.GetSegment(0)
 282 //       |                 |
 283 //       + - - - - - - - - + - - mReadCursor
 284 //       |/////////////////|
 285 //       |/////////////////|
 286 //       |/////////////////|
 287 //       |/////////////////|
 288 //       +-----------------+ - - mReadLimit
 289 //                |
 290 //       +-----------------+
 291 //       |/////////////////|
 292 //       |/////////////////|
 293 //       |/////////////////|
 294 //       |/////////////////|
 295 //       |/////////////////|
 296 //       |/////////////////|
 297 //       +-----------------+
 298 //                |
 299 //       +-----------------+ - - mBuffer.GetSegment(mWriteSegment)
 300 //       |/////////////////|
 301 //       |/////////////////|
 302 //       |/////////////////|
 303 //       + - - - - - - - - + - - mWriteCursor
 304 //       |                 |
 305 //       |                 |
 306 //       +-----------------+ - - mWriteLimit
 307 //
 308 // (shaded region contains data)
 309 //
 310 // NOTE: on some systems (notably OS/2), the heap allocator uses an arena for
 311 // small allocations (e.g., 64 byte allocations).  this means that buffers may
 312 // be allocated back-to-back.  in the diagram above, for example, mReadLimit
 313 // would actually be pointing at the beginning of the next segment.  when
 314 // making changes to this file, please keep this fact in mind.
 315 //
 316
 317 //-----------------------------------------------------------------------------
 318 // nsPipe methods:
 319 //-----------------------------------------------------------------------------
 320
 321 nsPipe::nsPipe()
 322   : mInput(MOZ_THIS_IN_INITIALIZER_LIST())
 323   , mOutput(MOZ_THIS_IN_INITIALIZER_LIST())
 324   , mReentrantMonitor("nsPipe.mReentrantMonitor")
 325   , mReadCursor(nullptr)
 326   , mReadLimit(nullptr)
 327   , mWriteSegment(-1)
 328   , mWriteCursor(nullptr)
 329   , mWriteLimit(nullptr)
 330   , mStatus(NS_OK)
 331   , mInited(false)
 332 {
 333 }
 334
 335 nsPipe::~nsPipe()
 336 {
 337 }
 338
 339 NS_IMPL_ISUPPORTS(nsPipe, nsIPipe)
 340
 341 NS_IMETHODIMP
 342 nsPipe::Init(bool aNonBlockingIn,
 343              bool aNonBlockingOut,
 344              uint32_t aSegmentSize,
 345              uint32_t aSegmentCount)
 346 {
 347   mInited = true;
 348
 349   if (aSegmentSize == 0) {
 350     aSegmentSize = DEFAULT_SEGMENT_SIZE;
 351   }
 352   if (aSegmentCount == 0) {
 353     aSegmentCount = DEFAULT_SEGMENT_COUNT;
 354   }
 355
 356   // protect against overflow
 357   uint32_t maxCount = uint32_t(-1) / aSegmentSize;
 358   if (aSegmentCount > maxCount) {
 359     aSegmentCount = maxCount;
 360   }
 361
 362   nsresult rv = mBuffer.Init(aSegmentSize, aSegmentSize * aSegmentCount);
 363   if (NS_FAILED(rv)) {
 364     return rv;
 365   }
 366
 367   mInput.SetNonBlocking(aNonBlockingIn);
 368   mOutput.SetNonBlocking(aNonBlockingOut);
 369   return NS_OK;
 370 }
 371
 372 NS_IMETHODIMP
 373 nsPipe::GetInputStream(nsIAsyncInputStream** aInputStream)
 374 {
 375   NS_ADDREF(*aInputStream = &mInput);
 376   return NS_OK;
 377 }
 378
 379 NS_IMETHODIMP
 380 nsPipe::GetOutputStream(nsIAsyncOutputStream** aOutputStream)
 381 {
 382   if (NS_WARN_IF(!mInited)) {
 383     return NS_ERROR_NOT_INITIALIZED;
 384   }
 385   NS_ADDREF(*aOutputStream = &mOutput);
 386   return NS_OK;
 387 }
 388
 389 void
 390 nsPipe::PeekSegment(uint32_t aIndex, char*& aCursor, char*& aLimit)
 391 {
 392   if (aIndex == 0) {
 393     NS_ASSERTION(!mReadCursor || mBuffer.GetSegmentCount(), "unexpected state");
 394     aCursor = mReadCursor;
 395     aLimit = mReadLimit;
 396   } else {
 397     uint32_t numSegments = mBuffer.GetSegmentCount();
 398     if (aIndex >= numSegments) {
 399       aCursor = aLimit = nullptr;
 400     } else {
 401       aCursor = mBuffer.GetSegment(aIndex);
 402       if (mWriteSegment == (int32_t)aIndex) {
 403         aLimit = mWriteCursor;
 404       } else {
 405         aLimit = aCursor + mBuffer.GetSegmentSize();
 406       }
 407     }
 408   }
 409 }
 410
 411 nsresult
 412 nsPipe::GetReadSegment(const char*& aSegment, uint32_t& aSegmentLen)
 413 {
 414   ReentrantMonitorAutoEnter mon(mReentrantMonitor);
 415
 416   if (mReadCursor == mReadLimit) {
 417     return NS_FAILED(mStatus) ? mStatus : NS_BASE_STREAM_WOULD_BLOCK;
 418   }
 419
 420   aSegment    = mReadCursor;
 421   aSegmentLen = mReadLimit - mReadCursor;
 422   return NS_OK;
 423 }
 424
 425 void
 426 nsPipe::AdvanceReadCursor(uint32_t aBytesRead)
 427 {
 428   NS_ASSERTION(aBytesRead, "don't call if no bytes read");
 429
 430   nsPipeEvents events;
 431   {
 432     ReentrantMonitorAutoEnter mon(mReentrantMonitor);
 433
 434     LOG(("III advancing read cursor by %u\n", aBytesRead));
 435     NS_ASSERTION(aBytesRead <= mBuffer.GetSegmentSize(), "read too much");
 436
 437     mReadCursor += aBytesRead;
 438     NS_ASSERTION(mReadCursor <= mReadLimit, "read cursor exceeds limit");
 439
 440     mInput.ReduceAvailable(aBytesRead);
 441
 442     if (mReadCursor == mReadLimit) {
 443       // we've reached the limit of how much we can read from this segment.
 444       // if at the end of this segment, then we must discard this segment.
 445
 446       // if still writing in this segment then bail because we're not done
 447       // with the segment and have to wait for now...
 448       if (mWriteSegment == 0 && mWriteLimit > mWriteCursor) {
 449         NS_ASSERTION(mReadLimit == mWriteCursor, "unexpected state");
 450         return;
 451       }
 452
 453       // shift write segment index (-1 indicates an empty buffer).
 454       --mWriteSegment;
 455
 456       // done with this segment
 457       mBuffer.DeleteFirstSegment();
 458       LOG(("III deleting first segment\n"));
 459
 460       if (mWriteSegment == -1) {
 461         // buffer is completely empty
 462         mReadCursor = nullptr;
 463         mReadLimit = nullptr;
 464         mWriteCursor = nullptr;
 465         mWriteLimit = nullptr;
 466       } else {
 467         // advance read cursor and limit to next buffer segment
 468         mReadCursor = mBuffer.GetSegment(0);
 469         if (mWriteSegment == 0) {
 470           mReadLimit = mWriteCursor;
 471         } else {
 472           mReadLimit = mReadCursor + mBuffer.GetSegmentSize();
 473         }
 474       }
 475
 476       // we've free'd up a segment, so notify output stream that pipe has
 477       // room for a new segment.
 478       if (mOutput.OnOutputWritable(events)) {
 479         mon.Notify();
 480       }
 481     }
 482   }
 483 }
 484
 485 nsresult
 486 nsPipe::GetWriteSegment(char*& aSegment, uint32_t& aSegmentLen)
 487 {
 488   ReentrantMonitorAutoEnter mon(mReentrantMonitor);
 489
 490   if (NS_FAILED(mStatus)) {
 491     return mStatus;
 492   }
 493
 494   // write cursor and limit may both be null indicating an empty buffer.
 495   if (mWriteCursor == mWriteLimit) {
 496     char* seg = mBuffer.AppendNewSegment();
 497     // pipe is full
 498     if (!seg) {
 499       return NS_BASE_STREAM_WOULD_BLOCK;
 500     }
 501     LOG(("OOO appended new segment\n"));
 502     mWriteCursor = seg;
 503     mWriteLimit = mWriteCursor + mBuffer.GetSegmentSize();
 504     ++mWriteSegment;
 505   }
 506
 507   // make sure read cursor is initialized
 508   if (!mReadCursor) {
 509     NS_ASSERTION(mWriteSegment == 0, "unexpected null read cursor");
 510     mReadCursor = mReadLimit = mWriteCursor;
 511   }
 512
 513   // check to see if we can roll-back our read and write cursors to the
 514   // beginning of the current/first segment.  this is purely an optimization.
 515   if (mReadCursor == mWriteCursor && mWriteSegment == 0) {
 516     char* head = mBuffer.GetSegment(0);
 517     LOG(("OOO rolling back write cursor %u bytes\n", mWriteCursor - head));
 518     mWriteCursor = mReadCursor = mReadLimit = head;
 519   }
 520
 521   aSegment    = mWriteCursor;
 522   aSegmentLen = mWriteLimit - mWriteCursor;
 523   return NS_OK;
 524 }
 525
 526 void
 527 nsPipe::AdvanceWriteCursor(uint32_t aBytesWritten)
 528 {
 529   NS_ASSERTION(aBytesWritten, "don't call if no bytes written");
 530
 531   nsPipeEvents events;
 532   {
 533     ReentrantMonitorAutoEnter mon(mReentrantMonitor);
 534
 535     LOG(("OOO advancing write cursor by %u\n", aBytesWritten));
 536
 537     char* newWriteCursor = mWriteCursor + aBytesWritten;
 538     NS_ASSERTION(newWriteCursor <= mWriteLimit, "write cursor exceeds limit");
 539
 540     // update read limit if reading in the same segment
 541     if (mWriteSegment == 0 && mReadLimit == mWriteCursor) {
 542       mReadLimit = newWriteCursor;
 543     }
 544
 545     mWriteCursor = newWriteCursor;
 546
 547     // The only way mReadCursor == mWriteCursor is if:
 548     //
 549     // - mReadCursor is at the start of a segment (which, based on how
 550     //   nsSegmentedBuffer works, means that this segment is the "first"
 551     //   segment)
 552     // - mWriteCursor points at the location past the end of the current
 553     //   write segment (so the current write filled the current write
 554     //   segment, so we've incremented mWriteCursor to point past the end
 555     //   of it)
 556     // - the segment to which data has just been written is located
 557     //   exactly one segment's worth of bytes before the first segment
 558     //   where mReadCursor is located
 559     //
 560     // Consequently, the byte immediately after the end of the current
 561     // write segment is the first byte of the first segment, so
 562     // mReadCursor == mWriteCursor.  (Another way to think about this is
 563     // to consider the buffer architecture diagram above, but consider it
 564     // with an arena allocator which allocates from the *end* of the
 565     // arena to the *beginning* of the arena.)
 566     NS_ASSERTION(mReadCursor != mWriteCursor ||
 567                  (mBuffer.GetSegment(0) == mReadCursor &&
 568                   mWriteCursor == mWriteLimit),
 569                  "read cursor is bad");
 570
 571     // update the writable flag on the output stream
 572     if (mWriteCursor == mWriteLimit) {
 573       if (mBuffer.GetSize() >= mBuffer.GetMaxSize()) {
 574         mOutput.SetWritable(false);
 575       }
 576     }
 577
 578     // notify input stream that pipe now contains additional data
 579     if (mInput.OnInputReadable(aBytesWritten, events)) {
 580       mon.Notify();
 581     }
 582   }
 583 }
 584
 585 void
 586 nsPipe::OnPipeException(nsresult aReason, bool aOutputOnly)
 587 {
 588   LOG(("PPP nsPipe::OnPipeException [reason=%x output-only=%d]\n",
 589        aReason, aOutputOnly));
 590
 591   nsPipeEvents events;
 592   {
 593     ReentrantMonitorAutoEnter mon(mReentrantMonitor);
 594
 595     // if we've already hit an exception, then ignore this one.
 596     if (NS_FAILED(mStatus)) {
 597       return;
 598     }
 599
 600     mStatus = aReason;
 601
 602     // an output-only exception applies to the input end if the pipe has
 603     // zero bytes available.
 604     if (aOutputOnly && !mInput.Available()) {
 605       aOutputOnly = false;
 606     }
 607
 608     if (!aOutputOnly)
 609       if (mInput.OnInputException(aReason, events)) {
 610         mon.Notify();
 611       }
 612
 613     if (mOutput.OnOutputException(aReason, events)) {
 614       mon.Notify();
 615     }
 616   }
 617 }
 618
 619 //-----------------------------------------------------------------------------
 620 // nsPipeEvents methods:
 621 //-----------------------------------------------------------------------------
 622
 623 nsPipeEvents::~nsPipeEvents()
 624 {
 625   // dispatch any pending events
 626
 627   if (mInputCallback) {
 628     mInputCallback->OnInputStreamReady(mInputStream);
 629     mInputCallback = 0;
 630     mInputStream = 0;
 631   }
 632   if (mOutputCallback) {
 633     mOutputCallback->OnOutputStreamReady(mOutputStream);
 634     mOutputCallback = 0;
 635     mOutputStream = 0;
 636   }
 637 }
 638
 639 //-----------------------------------------------------------------------------
 640 // nsPipeInputStream methods:
 641 //-----------------------------------------------------------------------------
 642
 643 NS_IMPL_QUERY_INTERFACE(nsPipeInputStream,
 644                         nsIInputStream,
 645                         nsIAsyncInputStream,
 646                         nsISeekableStream,
 647                         nsISearchableInputStream,
 648                         nsIClassInfo)
 649
 650 NS_IMPL_CI_INTERFACE_GETTER(nsPipeInputStream,
 651                             nsIInputStream,
 652                             nsIAsyncInputStream,
 653                             nsISeekableStream,
 654                             nsISearchableInputStream)
 655
 656 NS_IMPL_THREADSAFE_CI(nsPipeInputStream)
 657
 658 nsresult
 659 nsPipeInputStream::Wait()
 660 {
 661   NS_ASSERTION(mBlocking, "wait on non-blocking pipe input stream");
 662
 663   ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
 664
 665   while (NS_SUCCEEDED(mPipe->mStatus) && (mAvailable == 0)) {
 666     LOG(("III pipe input: waiting for data\n"));
 667
 668     mBlocked = true;
 669     mon.Wait();
 670     mBlocked = false;
 671
 672     LOG(("III pipe input: woke up [pipe-status=%x available=%u]\n",
 673          mPipe->mStatus, mAvailable));
 674   }
 675
 676   return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus;
 677 }
 678
 679 bool
 680 nsPipeInputStream::OnInputReadable(uint32_t aBytesWritten, nsPipeEvents& aEvents)
 681 {
 682   bool result = false;
 683
 684   mAvailable += aBytesWritten;
 685
 686   if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) {
 687     aEvents.NotifyInputReady(this, mCallback);
 688     mCallback = 0;
 689     mCallbackFlags = 0;
 690   } else if (mBlocked) {
 691     result = true;
 692   }
 693
 694   return result;
 695 }
 696
 697 bool
 698 nsPipeInputStream::OnInputException(nsresult aReason, nsPipeEvents& aEvents)
 699 {
 700   LOG(("nsPipeInputStream::OnInputException [this=%x reason=%x]\n",
 701        this, aReason));
 702
 703   bool result = false;
 704
 705   NS_ASSERTION(NS_FAILED(aReason), "huh? successful exception");
 706
 707   // force count of available bytes to zero.
 708   mAvailable = 0;
 709
 710   if (mCallback) {
 711     aEvents.NotifyInputReady(this, mCallback);
 712     mCallback = 0;
 713     mCallbackFlags = 0;
 714   } else if (mBlocked) {
 715     result = true;
 716   }
 717
 718   return result;
 719 }
 720
 721 NS_IMETHODIMP_(MozExternalRefCountType)
 722 nsPipeInputStream::AddRef(void)
 723 {
 724   ++mReaderRefCnt;
 725   return mPipe->AddRef();
 726 }
 727
 728 NS_IMETHODIMP_(MozExternalRefCountType)
 729 nsPipeInputStream::Release(void)
 730 {
 731   if (--mReaderRefCnt == 0) {
 732     Close();
 733   }
 734   return mPipe->Release();
 735 }
 736
 737 NS_IMETHODIMP
 738 nsPipeInputStream::CloseWithStatus(nsresult aReason)
 739 {
 740   LOG(("III CloseWithStatus [this=%x reason=%x]\n", this, aReason));
 741
 742   if (NS_SUCCEEDED(aReason)) {
 743     aReason = NS_BASE_STREAM_CLOSED;
 744   }
 745
 746   mPipe->OnPipeException(aReason);
 747   return NS_OK;
 748 }
 749
 750 NS_IMETHODIMP
 751 nsPipeInputStream::Close()
 752 {
 753   return CloseWithStatus(NS_BASE_STREAM_CLOSED);
 754 }
 755
 756 NS_IMETHODIMP
 757 nsPipeInputStream::Available(uint64_t* aResult)
 758 {
 759   // nsPipeInputStream supports under 4GB stream only
 760   ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
 761
 762   // return error if pipe closed
 763   if (!mAvailable && NS_FAILED(mPipe->mStatus)) {
 764     return mPipe->mStatus;
 765   }
 766
 767   *aResult = (uint64_t)mAvailable;
 768   return NS_OK;
 769 }
 770
 771 NS_IMETHODIMP
 772 nsPipeInputStream::ReadSegments(nsWriteSegmentFun aWriter,
 773                                 void* aClosure,
 774                                 uint32_t aCount,
 775                                 uint32_t* aReadCount)
 776 {
 777   LOG(("III ReadSegments [this=%x count=%u]\n", this, aCount));
 778
 779   nsresult rv = NS_OK;
 780
 781   const char* segment;
 782   uint32_t segmentLen;
 783
 784   *aReadCount = 0;
 785   while (aCount) {
 786     rv = mPipe->GetReadSegment(segment, segmentLen);
 787     if (NS_FAILED(rv)) {
 788       // ignore this error if we've already read something.
 789       if (*aReadCount > 0) {
 790         rv = NS_OK;
 791         break;
 792       }
 793       if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
 794         // pipe is empty
 795         if (!mBlocking) {
 796           break;
 797         }
 798         // wait for some data to be written to the pipe
 799         rv = Wait();
 800         if (NS_SUCCEEDED(rv)) {
 801           continue;
 802         }
 803       }
 804       // ignore this error, just return.
 805       if (rv == NS_BASE_STREAM_CLOSED) {
 806         rv = NS_OK;
 807         break;
 808       }
 809       mPipe->OnPipeException(rv);
 810       break;
 811     }
 812
 813     // read no more than aCount
 814     if (segmentLen > aCount) {
 815       segmentLen = aCount;
 816     }
 817
 818     uint32_t writeCount, originalLen = segmentLen;
 819     while (segmentLen) {
 820       writeCount = 0;
 821
 822       rv = aWriter(this, aClosure, segment, *aReadCount, segmentLen, &writeCount);
 823
 824       if (NS_FAILED(rv) || writeCount == 0) {
 825         aCount = 0;
 826         // any errors returned from the writer end here: do not
 827         // propagate to the caller of ReadSegments.
 828         rv = NS_OK;
 829         break;
 830       }
 831
 832       NS_ASSERTION(writeCount <= segmentLen, "wrote more than expected");
 833       segment += writeCount;
 834       segmentLen -= writeCount;
 835       aCount -= writeCount;
 836       *aReadCount += writeCount;
 837       mLogicalOffset += writeCount;
 838     }
 839
 840     if (segmentLen < originalLen) {
 841       mPipe->AdvanceReadCursor(originalLen - segmentLen);
 842     }
 843   }
 844
 845   return rv;
 846 }
 847
 848 NS_IMETHODIMP
 849 nsPipeInputStream::Read(char* aToBuf, uint32_t aBufLen, uint32_t* aReadCount)
 850 {
 851   return ReadSegments(NS_CopySegmentToBuffer, aToBuf, aBufLen, aReadCount);
 852 }
 853
 854 NS_IMETHODIMP
 855 nsPipeInputStream::IsNonBlocking(bool* aNonBlocking)
 856 {
 857   *aNonBlocking = !mBlocking;
 858   return NS_OK;
 859 }
 860
 861 NS_IMETHODIMP
 862 nsPipeInputStream::AsyncWait(nsIInputStreamCallback* aCallback,
 863                              uint32_t aFlags,
 864                              uint32_t aRequestedCount,
 865                              nsIEventTarget* aTarget)
 866 {
 867   LOG(("III AsyncWait [this=%x]\n", this));
 868
 869   nsPipeEvents pipeEvents;
 870   {
 871     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
 872
 873     // replace a pending callback
 874     mCallback = 0;
 875     mCallbackFlags = 0;
 876
 877     if (!aCallback) {
 878       return NS_OK;
 879     }
 880
 881     nsCOMPtr<nsIInputStreamCallback> proxy;
 882     if (aTarget) {
 883       proxy = NS_NewInputStreamReadyEvent(aCallback, aTarget);
 884       aCallback = proxy;
 885     }
 886
 887     if (NS_FAILED(mPipe->mStatus) ||
 888         (mAvailable && !(aFlags & WAIT_CLOSURE_ONLY))) {
 889       // stream is already closed or readable; post event.
 890       pipeEvents.NotifyInputReady(this, aCallback);
 891     } else {
 892       // queue up callback object to be notified when data becomes available
 893       mCallback = aCallback;
 894       mCallbackFlags = aFlags;
 895     }
 896   }
 897   return NS_OK;
 898 }
 899
 900 NS_IMETHODIMP
 901 nsPipeInputStream::Seek(int32_t aWhence, int64_t aOffset)
 902 {
 903   NS_NOTREACHED("nsPipeInputStream::Seek");
 904   return NS_ERROR_NOT_IMPLEMENTED;
 905 }
 906
 907 NS_IMETHODIMP
 908 nsPipeInputStream::Tell(int64_t* aOffset)
 909 {
 910   ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
 911
 912   // return error if pipe closed
 913   if (!mAvailable && NS_FAILED(mPipe->mStatus)) {
 914     return mPipe->mStatus;
 915   }
 916
 917   *aOffset = mLogicalOffset;
 918   return NS_OK;
 919 }
 920
 921 NS_IMETHODIMP
 922 nsPipeInputStream::SetEOF()
 923 {
 924   NS_NOTREACHED("nsPipeInputStream::SetEOF");
 925   return NS_ERROR_NOT_IMPLEMENTED;
 926 }
 927
 928 #define COMPARE(s1, s2, i)                                               \
 929   (aIgnoreCase                                                           \
 930    ? nsCRT::strncasecmp((const char *)s1, (const char *)s2, (uint32_t)i) \
 931    : nsCRT::strncmp((const char *)s1, (const char *)s2, (uint32_t)i))
 932
 933 NS_IMETHODIMP
 934 nsPipeInputStream::Search(const char* aForString,
 935                           bool aIgnoreCase,
 936                           bool* aFound,
 937                           uint32_t* aOffsetSearchedTo)
 938 {
 939   LOG(("III Search [for=%s ic=%u]\n", aForString, aIgnoreCase));
 940
 941   ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
 942
 943   char* cursor1;
 944   char* limit1;
 945   uint32_t index = 0, offset = 0;
 946   uint32_t strLen = strlen(aForString);
 947
 948   mPipe->PeekSegment(0, cursor1, limit1);
 949   if (cursor1 == limit1) {
 950     *aFound = false;
 951     *aOffsetSearchedTo = 0;
 952     LOG(("  result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
 953     return NS_OK;
 954   }
 955
 956   while (true) {
 957     uint32_t i, len1 = limit1 - cursor1;
 958
 959     // check if the string is in the buffer segment
 960     for (i = 0; i < len1 - strLen + 1; i++) {
 961       if (COMPARE(&cursor1[i], aForString, strLen) == 0) {
 962         *aFound = true;
 963         *aOffsetSearchedTo = offset + i;
 964         LOG(("  result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
 965         return NS_OK;
 966       }
 967     }
 968
 969     // get the next segment
 970     char* cursor2;
 971     char* limit2;
 972     uint32_t len2;
 973
 974     index++;
 975     offset += len1;
 976
 977     mPipe->PeekSegment(index, cursor2, limit2);
 978     if (cursor2 == limit2) {
 979       *aFound = false;
 980       *aOffsetSearchedTo = offset - strLen + 1;
 981       LOG(("  result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
 982       return NS_OK;
 983     }
 984     len2 = limit2 - cursor2;
 985
 986     // check if the string is straddling the next buffer segment
 987     uint32_t lim = XPCOM_MIN(strLen, len2 + 1);
 988     for (i = 0; i < lim; ++i) {
 989       uint32_t strPart1Len = strLen - i - 1;
 990       uint32_t strPart2Len = strLen - strPart1Len;
 991       const char* strPart2 = &aForString[strLen - strPart2Len];
 992       uint32_t bufSeg1Offset = len1 - strPart1Len;
 993       if (COMPARE(&cursor1[bufSeg1Offset], aForString, strPart1Len) == 0 &&
 994           COMPARE(cursor2, strPart2, strPart2Len) == 0) {
 995         *aFound = true;
 996         *aOffsetSearchedTo = offset - strPart1Len;
 997         LOG(("  result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo));
 998         return NS_OK;
 999       }
1000     }
1001
1002     // finally continue with the next buffer
1003     cursor1 = cursor2;
1004     limit1 = limit2;
1005   }
1006
1007   NS_NOTREACHED("can't get here");
1008   return NS_ERROR_UNEXPECTED;    // keep compiler happy
1009 }
1010
1011 //-----------------------------------------------------------------------------
1012 // nsPipeOutputStream methods:
1013 //-----------------------------------------------------------------------------
1014
1015 NS_IMPL_QUERY_INTERFACE(nsPipeOutputStream,
1016                         nsIOutputStream,
1017                         nsIAsyncOutputStream,
1018                         nsIClassInfo)
1019
1020 NS_IMPL_CI_INTERFACE_GETTER(nsPipeOutputStream,
1021                             nsIOutputStream,
1022                             nsIAsyncOutputStream)
1023
1024 NS_IMPL_THREADSAFE_CI(nsPipeOutputStream)
1025
1026 nsresult
1027 nsPipeOutputStream::Wait()
1028 {
1029   NS_ASSERTION(mBlocking, "wait on non-blocking pipe output stream");
1030
1031   ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
1032
1033   if (NS_SUCCEEDED(mPipe->mStatus) && !mWritable) {
1034     LOG(("OOO pipe output: waiting for space\n"));
1035     mBlocked = true;
1036     mon.Wait();
1037     mBlocked = false;
1038     LOG(("OOO pipe output: woke up [pipe-status=%x writable=%u]\n",
1039          mPipe->mStatus, mWritable));
1040   }
1041
1042   return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus;
1043 }
1044
1045 bool
1046 nsPipeOutputStream::OnOutputWritable(nsPipeEvents& aEvents)
1047 {
1048   bool result = false;
1049
1050   mWritable = true;
1051
1052   if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) {
1053     aEvents.NotifyOutputReady(this, mCallback);
1054     mCallback = 0;
1055     mCallbackFlags = 0;
1056   } else if (mBlocked) {
1057     result = true;
1058   }
1059
1060   return result;
1061 }
1062
1063 bool
1064 nsPipeOutputStream::OnOutputException(nsresult aReason, nsPipeEvents& aEvents)
1065 {
1066   LOG(("nsPipeOutputStream::OnOutputException [this=%x reason=%x]\n",
1067        this, aReason));
1068
1069   bool result = false;
1070
1071   NS_ASSERTION(NS_FAILED(aReason), "huh? successful exception");
1072   mWritable = false;
1073
1074   if (mCallback) {
1075     aEvents.NotifyOutputReady(this, mCallback);
1076     mCallback = 0;
1077     mCallbackFlags = 0;
1078   } else if (mBlocked) {
1079     result = true;
1080   }
1081
1082   return result;
1083 }
1084
1085
1086 NS_IMETHODIMP_(MozExternalRefCountType)
1087 nsPipeOutputStream::AddRef()
1088 {
1089   ++mWriterRefCnt;
1090   return mPipe->AddRef();
1091 }
1092
1093 NS_IMETHODIMP_(MozExternalRefCountType)
1094 nsPipeOutputStream::Release()
1095 {
1096   if (--mWriterRefCnt == 0) {
1097     Close();
1098   }
1099   return mPipe->Release();
1100 }
1101
1102 NS_IMETHODIMP
1103 nsPipeOutputStream::CloseWithStatus(nsresult aReason)
1104 {
1105   LOG(("OOO CloseWithStatus [this=%x reason=%x]\n", this, aReason));
1106
1107   if (NS_SUCCEEDED(aReason)) {
1108     aReason = NS_BASE_STREAM_CLOSED;
1109   }
1110
1111   // input stream may remain open
1112   mPipe->OnPipeException(aReason, true);
1113   return NS_OK;
1114 }
1115
1116 NS_IMETHODIMP
1117 nsPipeOutputStream::Close()
1118 {
1119   return CloseWithStatus(NS_BASE_STREAM_CLOSED);
1120 }
1121
1122 NS_IMETHODIMP
1123 nsPipeOutputStream::WriteSegments(nsReadSegmentFun aReader,
1124                                   void* aClosure,
1125                                   uint32_t aCount,
1126                                   uint32_t* aWriteCount)
1127 {
1128   LOG(("OOO WriteSegments [this=%x count=%u]\n", this, aCount));
1129
1130   nsresult rv = NS_OK;
1131
1132   char* segment;
1133   uint32_t segmentLen;
1134
1135   *aWriteCount = 0;
1136   while (aCount) {
1137     rv = mPipe->GetWriteSegment(segment, segmentLen);
1138     if (NS_FAILED(rv)) {
1139       if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
1140         // pipe is full
1141         if (!mBlocking) {
1142           // ignore this error if we've already written something
1143           if (*aWriteCount > 0) {
1144             rv = NS_OK;
1145           }
1146           break;
1147         }
1148         // wait for the pipe to have an empty segment.
1149         rv = Wait();
1150         if (NS_SUCCEEDED(rv)) {
1151           continue;
1152         }
1153       }
1154       mPipe->OnPipeException(rv);
1155       break;
1156     }
1157
1158     // write no more than aCount
1159     if (segmentLen > aCount) {
1160       segmentLen = aCount;
1161     }
1162
1163     uint32_t readCount, originalLen = segmentLen;
1164     while (segmentLen) {
1165       readCount = 0;
1166
1167       rv = aReader(this, aClosure, segment, *aWriteCount, segmentLen, &readCount);
1168
1169       if (NS_FAILED(rv) || readCount == 0) {
1170         aCount = 0;
1171         // any errors returned from the aReader end here: do not
1172         // propagate to the caller of WriteSegments.
1173         rv = NS_OK;
1174         break;
1175       }
1176
1177       NS_ASSERTION(readCount <= segmentLen, "read more than expected");
1178       segment += readCount;
1179       segmentLen -= readCount;
1180       aCount -= readCount;
1181       *aWriteCount += readCount;
1182       mLogicalOffset += readCount;
1183     }
1184
1185     if (segmentLen < originalLen) {
1186       mPipe->AdvanceWriteCursor(originalLen - segmentLen);
1187     }
1188   }
1189
1190   return rv;
1191 }
1192
1193 static NS_METHOD
1194 nsReadFromRawBuffer(nsIOutputStream* aOutStr,
1195                     void* aClosure,
1196                     char* aToRawSegment,
1197                     uint32_t aOffset,
1198                     uint32_t aCount,
1199                     uint32_t* aReadCount)
1200 {
1201   const char* fromBuf = (const char*)aClosure;
1202   memcpy(aToRawSegment, &fromBuf[aOffset], aCount);
1203   *aReadCount = aCount;
1204   return NS_OK;
1205 }
1206
1207 NS_IMETHODIMP
1208 nsPipeOutputStream::Write(const char* aFromBuf,
1209                           uint32_t aBufLen,
1210                           uint32_t* aWriteCount)
1211 {
1212   return WriteSegments(nsReadFromRawBuffer, (void*)aFromBuf, aBufLen, aWriteCount);
1213 }
1214
1215 NS_IMETHODIMP
1216 nsPipeOutputStream::Flush(void)
1217 {
1218   // nothing to do
1219   return NS_OK;
1220 }
1221
1222 static NS_METHOD
1223 nsReadFromInputStream(nsIOutputStream* aOutStr,
1224                       void* aClosure,
1225                       char* aToRawSegment,
1226                       uint32_t aOffset,
1227                       uint32_t aCount,
1228                       uint32_t* aReadCount)
1229 {
1230   nsIInputStream* fromStream = (nsIInputStream*)aClosure;
1231   return fromStream->Read(aToRawSegment, aCount, aReadCount);
1232 }
1233
1234 NS_IMETHODIMP
1235 nsPipeOutputStream::WriteFrom(nsIInputStream* aFromStream,
1236                               uint32_t aCount,
1237                               uint32_t* aWriteCount)
1238 {
1239   return WriteSegments(nsReadFromInputStream, aFromStream, aCount, aWriteCount);
1240 }
1241
1242 NS_IMETHODIMP
1243 nsPipeOutputStream::IsNonBlocking(bool* aNonBlocking)
1244 {
1245   *aNonBlocking = !mBlocking;
1246   return NS_OK;
1247 }
1248
1249 NS_IMETHODIMP
1250 nsPipeOutputStream::AsyncWait(nsIOutputStreamCallback* aCallback,
1251                               uint32_t aFlags,
1252                               uint32_t aRequestedCount,
1253                               nsIEventTarget* aTarget)
1254 {
1255   LOG(("OOO AsyncWait [this=%x]\n", this));
1256
1257   nsPipeEvents pipeEvents;
1258   {
1259     ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor);
1260
1261     // replace a pending callback
1262     mCallback = 0;
1263     mCallbackFlags = 0;
1264
1265     if (!aCallback) {
1266       return NS_OK;
1267     }
1268
1269     nsCOMPtr<nsIOutputStreamCallback> proxy;
1270     if (aTarget) {
1271       proxy = NS_NewOutputStreamReadyEvent(aCallback, aTarget);
1272       aCallback = proxy;
1273     }
1274
1275     if (NS_FAILED(mPipe->mStatus) ||
1276         (mWritable && !(aFlags & WAIT_CLOSURE_ONLY))) {
1277       // stream is already closed or writable; post event.
1278       pipeEvents.NotifyOutputReady(this, aCallback);
1279     } else {
1280       // queue up callback object to be notified when data becomes available
1281       mCallback = aCallback;
1282       mCallbackFlags = aFlags;
1283     }
1284   }
1285   return NS_OK;
1286 }
1287
1288 ////////////////////////////////////////////////////////////////////////////////
1289
1290 nsresult
1291 NS_NewPipe(nsIInputStream** aPipeIn,
1292            nsIOutputStream** aPipeOut,
1293            uint32_t aSegmentSize,
1294            uint32_t aMaxSize,
1295            bool aNonBlockingInput,
1296            bool aNonBlockingOutput)
1297 {
1298   if (aSegmentSize == 0) {
1299     aSegmentSize = DEFAULT_SEGMENT_SIZE;
1300   }
1301
1302   // Handle aMaxSize of UINT32_MAX as a special case
1303   uint32_t segmentCount;
1304   if (aMaxSize == UINT32_MAX) {
1305     segmentCount = UINT32_MAX;
1306   } else {
1307     segmentCount = aMaxSize / aSegmentSize;
1308   }
1309
1310   nsIAsyncInputStream* in;
1311   nsIAsyncOutputStream* out;
1312   nsresult rv = NS_NewPipe2(&in, &out, aNonBlockingInput, aNonBlockingOutput,
1313                             aSegmentSize, segmentCount);
1314   if (NS_FAILED(rv)) {
1315     return rv;
1316   }
1317
1318   *aPipeIn = in;
1319   *aPipeOut = out;
1320   return NS_OK;
1321 }
1322
1323 nsresult
1324 NS_NewPipe2(nsIAsyncInputStream** aPipeIn,
1325             nsIAsyncOutputStream** aPipeOut,
1326             bool aNonBlockingInput,
1327             bool aNonBlockingOutput,
1328             uint32_t aSegmentSize,
1329             uint32_t aSegmentCount)
1330 {
1331   nsresult rv;
1332
1333   nsPipe* pipe = new nsPipe();
1334   if (!pipe) {
1335     return NS_ERROR_OUT_OF_MEMORY;
1336   }
1337
1338   rv = pipe->Init(aNonBlockingInput,
1339                   aNonBlockingOutput,
1340                   aSegmentSize,
1341                   aSegmentCount);
1342   if (NS_FAILED(rv)) {
1343     NS_ADDREF(pipe);
1344     NS_RELEASE(pipe);
1345     return rv;
1346   }
1347
1348   pipe->GetInputStream(aPipeIn);
1349   pipe->GetOutputStream(aPipeOut);
1350   return NS_OK;
1351 }
1352
1353 nsresult
1354 nsPipeConstructor(nsISupports* aOuter, REFNSIID aIID, void** aResult)
1355 {
1356   if (aOuter) {
1357     return NS_ERROR_NO_AGGREGATION;
1358   }
1359   nsPipe* pipe = new nsPipe();
1360   if (!pipe) {
1361     return NS_ERROR_OUT_OF_MEMORY;
1362   }
1363   NS_ADDREF(pipe);
1364   nsresult rv = pipe->QueryInterface(aIID, aResult);
1365   NS_RELEASE(pipe);
1366   return rv;
1367 }
1368
1369 ////////////////////////////////////////////////////////////////////////////////