diff options
Diffstat (limited to 'xpcom/io/nsPipe3.cpp')
| -rw-r--r-- | xpcom/io/nsPipe3.cpp | 2007 |
1 files changed, 2007 insertions, 0 deletions
diff --git a/xpcom/io/nsPipe3.cpp b/xpcom/io/nsPipe3.cpp new file mode 100644 index 000000000..56932adfc --- /dev/null +++ b/xpcom/io/nsPipe3.cpp @@ -0,0 +1,2007 @@ +/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ +/* vim: set ts=8 sts=2 et sw=2 tw=80: */ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ + +#include <algorithm> +#include "mozilla/Attributes.h" +#include "mozilla/ReentrantMonitor.h" +#include "nsIBufferedStreams.h" +#include "nsICloneableInputStream.h" +#include "nsIPipe.h" +#include "nsIEventTarget.h" +#include "nsISeekableStream.h" +#include "mozilla/RefPtr.h" +#include "nsSegmentedBuffer.h" +#include "nsStreamUtils.h" +#include "nsCOMPtr.h" +#include "nsCRT.h" +#include "mozilla/Logging.h" +#include "nsIClassInfoImpl.h" +#include "nsAlgorithm.h" +#include "nsMemory.h" +#include "nsIAsyncInputStream.h" +#include "nsIAsyncOutputStream.h" + +using namespace mozilla; + +#ifdef LOG +#undef LOG +#endif +// +// set MOZ_LOG=nsPipe:5 +// +static LazyLogModule sPipeLog("nsPipe"); +#define LOG(args) MOZ_LOG(sPipeLog, mozilla::LogLevel::Debug, args) + +#define DEFAULT_SEGMENT_SIZE 4096 +#define DEFAULT_SEGMENT_COUNT 16 + +class nsPipe; +class nsPipeEvents; +class nsPipeInputStream; +class nsPipeOutputStream; +class AutoReadSegment; + +namespace { + +enum MonitorAction +{ + DoNotNotifyMonitor, + NotifyMonitor +}; + +enum SegmentChangeResult +{ + SegmentNotChanged, + SegmentAdvanceBufferRead +}; + +} // namespace + +//----------------------------------------------------------------------------- + +// this class is used to delay notifications until the end of a particular +// scope. it helps avoid the complexity of issuing callbacks while inside +// a critical section. +class nsPipeEvents +{ +public: + nsPipeEvents() { } + ~nsPipeEvents(); + + inline void NotifyInputReady(nsIAsyncInputStream* aStream, + nsIInputStreamCallback* aCallback) + { + mInputList.AppendElement(InputEntry(aStream, aCallback)); + } + + inline void NotifyOutputReady(nsIAsyncOutputStream* aStream, + nsIOutputStreamCallback* aCallback) + { + NS_ASSERTION(!mOutputCallback, "already have an output event"); + mOutputStream = aStream; + mOutputCallback = aCallback; + } + +private: + struct InputEntry + { + InputEntry(nsIAsyncInputStream* aStream, nsIInputStreamCallback* aCallback) + : mStream(aStream) + , mCallback(aCallback) + { + MOZ_ASSERT(mStream); + MOZ_ASSERT(mCallback); + } + + nsCOMPtr<nsIAsyncInputStream> mStream; + nsCOMPtr<nsIInputStreamCallback> mCallback; + }; + + nsTArray<InputEntry> mInputList; + + nsCOMPtr<nsIAsyncOutputStream> mOutputStream; + nsCOMPtr<nsIOutputStreamCallback> mOutputCallback; +}; + +//----------------------------------------------------------------------------- + +// This class is used to maintain input stream state. Its broken out from the +// nsPipeInputStream class because generally the nsPipe should be modifying +// this state and not the input stream itself. +struct nsPipeReadState +{ + nsPipeReadState() + : mReadCursor(nullptr) + , mReadLimit(nullptr) + , mSegment(0) + , mAvailable(0) + , mActiveRead(false) + , mNeedDrain(false) + { } + + char* mReadCursor; + char* mReadLimit; + int32_t mSegment; + uint32_t mAvailable; + + // This flag is managed using the AutoReadSegment RAII stack class. + bool mActiveRead; + + // Set to indicate that the input stream has closed and should be drained, + // but that drain has been delayed due to an active read. When the read + // completes, this flag indicate the drain should then be performed. + bool mNeedDrain; +}; + +//----------------------------------------------------------------------------- + +// an input end of a pipe (maintained as a list of refs within the pipe) +class nsPipeInputStream final + : public nsIAsyncInputStream + , public nsISeekableStream + , public nsISearchableInputStream + , public nsICloneableInputStream + , public nsIClassInfo + , public nsIBufferedInputStream +{ +public: + NS_DECL_THREADSAFE_ISUPPORTS + NS_DECL_NSIINPUTSTREAM + NS_DECL_NSIASYNCINPUTSTREAM + NS_DECL_NSISEEKABLESTREAM + NS_DECL_NSISEARCHABLEINPUTSTREAM + NS_DECL_NSICLONEABLEINPUTSTREAM + NS_DECL_NSICLASSINFO + NS_DECL_NSIBUFFEREDINPUTSTREAM + + explicit nsPipeInputStream(nsPipe* aPipe) + : mPipe(aPipe) + , mLogicalOffset(0) + , mInputStatus(NS_OK) + , mBlocking(true) + , mBlocked(false) + , mCallbackFlags(0) + { } + + explicit nsPipeInputStream(const nsPipeInputStream& aOther) + : mPipe(aOther.mPipe) + , mLogicalOffset(aOther.mLogicalOffset) + , mInputStatus(aOther.mInputStatus) + , mBlocking(aOther.mBlocking) + , mBlocked(false) + , mCallbackFlags(0) + , mReadState(aOther.mReadState) + { } + + nsresult Fill(); + void SetNonBlocking(bool aNonBlocking) + { + mBlocking = !aNonBlocking; + } + + uint32_t Available(); + + // synchronously wait for the pipe to become readable. + nsresult Wait(); + + // These two don't acquire the monitor themselves. Instead they + // expect their caller to have done so and to pass the monitor as + // evidence. + MonitorAction OnInputReadable(uint32_t aBytesWritten, nsPipeEvents&, + const ReentrantMonitorAutoEnter& ev); + MonitorAction OnInputException(nsresult, nsPipeEvents&, + const ReentrantMonitorAutoEnter& ev); + + nsPipeReadState& ReadState() + { + return mReadState; + } + + const nsPipeReadState& ReadState() const + { + return mReadState; + } + + nsresult Status() const; + + // A version of Status() that doesn't acquire the monitor. + nsresult Status(const ReentrantMonitorAutoEnter& ev) const; + +private: + virtual ~nsPipeInputStream(); + + RefPtr<nsPipe> mPipe; + + int64_t mLogicalOffset; + // Individual input streams can be closed without effecting the rest of the + // pipe. So track individual input stream status separately. |mInputStatus| + // is protected by |mPipe->mReentrantMonitor|. + nsresult mInputStatus; + bool mBlocking; + + // these variables can only be accessed while inside the pipe's monitor + bool mBlocked; + nsCOMPtr<nsIInputStreamCallback> mCallback; + uint32_t mCallbackFlags; + + // requires pipe's monitor; usually treat as an opaque token to pass to nsPipe + nsPipeReadState mReadState; +}; + +//----------------------------------------------------------------------------- + +// the output end of a pipe (allocated as a member of the pipe). +class nsPipeOutputStream + : public nsIAsyncOutputStream + , public nsIClassInfo +{ +public: + // since this class will be allocated as a member of the pipe, we do not + // need our own ref count. instead, we share the lifetime (the ref count) + // of the entire pipe. this macro is just convenience since it does not + // declare a mRefCount variable; however, don't let the name fool you... + // we are not inheriting from nsPipe ;-) + NS_DECL_ISUPPORTS_INHERITED + + NS_DECL_NSIOUTPUTSTREAM + NS_DECL_NSIASYNCOUTPUTSTREAM + NS_DECL_NSICLASSINFO + + explicit nsPipeOutputStream(nsPipe* aPipe) + : mPipe(aPipe) + , mWriterRefCnt(0) + , mLogicalOffset(0) + , mBlocking(true) + , mBlocked(false) + , mWritable(true) + , mCallbackFlags(0) + { } + + void SetNonBlocking(bool aNonBlocking) + { + mBlocking = !aNonBlocking; + } + void SetWritable(bool aWritable) + { + mWritable = aWritable; + } + + // synchronously wait for the pipe to become writable. + nsresult Wait(); + + MonitorAction OnOutputWritable(nsPipeEvents&); + MonitorAction OnOutputException(nsresult, nsPipeEvents&); + +private: + nsPipe* mPipe; + + // separate refcnt so that we know when to close the producer + mozilla::ThreadSafeAutoRefCnt mWriterRefCnt; + int64_t mLogicalOffset; + bool mBlocking; + + // these variables can only be accessed while inside the pipe's monitor + bool mBlocked; + bool mWritable; + nsCOMPtr<nsIOutputStreamCallback> mCallback; + uint32_t mCallbackFlags; +}; + +//----------------------------------------------------------------------------- + +class nsPipe final : public nsIPipe +{ +public: + friend class nsPipeInputStream; + friend class nsPipeOutputStream; + friend class AutoReadSegment; + + NS_DECL_THREADSAFE_ISUPPORTS + NS_DECL_NSIPIPE + + // nsPipe methods: + nsPipe(); + +private: + ~nsPipe(); + + // + // Methods below may only be called while inside the pipe's monitor. Some + // of these methods require passing a ReentrantMonitorAutoEnter to prove the + // monitor is held. + // + + void PeekSegment(const nsPipeReadState& aReadState, uint32_t aIndex, + char*& aCursor, char*& aLimit); + SegmentChangeResult AdvanceReadSegment(nsPipeReadState& aReadState, + const ReentrantMonitorAutoEnter &ev); + bool ReadSegmentBeingWritten(nsPipeReadState& aReadState); + uint32_t CountSegmentReferences(int32_t aSegment); + void SetAllNullReadCursors(); + bool AllReadCursorsMatchWriteCursor(); + void RollBackAllReadCursors(char* aWriteCursor); + void UpdateAllReadCursors(char* aWriteCursor); + void ValidateAllReadCursors(); + uint32_t GetBufferSegmentCount(const nsPipeReadState& aReadState, + const ReentrantMonitorAutoEnter& ev) const; + bool IsAdvanceBufferFull(const ReentrantMonitorAutoEnter& ev) const; + + // + // methods below may be called while outside the pipe's monitor + // + + void DrainInputStream(nsPipeReadState& aReadState, nsPipeEvents& aEvents); + nsresult GetWriteSegment(char*& aSegment, uint32_t& aSegmentLen); + void AdvanceWriteCursor(uint32_t aCount); + + void OnInputStreamException(nsPipeInputStream* aStream, nsresult aReason); + void OnPipeException(nsresult aReason, bool aOutputOnly = false); + + nsresult CloneInputStream(nsPipeInputStream* aOriginal, + nsIInputStream** aCloneOut); + + // methods below should only be called by AutoReadSegment + nsresult GetReadSegment(nsPipeReadState& aReadState, const char*& aSegment, + uint32_t& aLength); + void ReleaseReadSegment(nsPipeReadState& aReadState, + nsPipeEvents& aEvents); + void AdvanceReadCursor(nsPipeReadState& aReadState, uint32_t aCount); + + // We can't inherit from both nsIInputStream and nsIOutputStream + // because they collide on their Close method. Consequently we nest their + // implementations to avoid the extra object allocation. + nsPipeOutputStream mOutput; + + // Since the input stream can be cloned, we may have more than one. Use + // a weak reference as the streams will clear their entry here in their + // destructor. Using a strong reference would create a reference cycle. + // Only usable while mReentrantMonitor is locked. + nsTArray<nsPipeInputStream*> mInputList; + + // But hold a strong ref to our original input stream. For backward + // compatibility we need to be able to consistently return this same + // object from GetInputStream(). Note, mOriginalInput is also stored + // in mInputList as a weak ref. + RefPtr<nsPipeInputStream> mOriginalInput; + + ReentrantMonitor mReentrantMonitor; + nsSegmentedBuffer mBuffer; + + // The maximum number of segments to allow to be buffered in advance + // of the fastest reader. This is collection of segments is called + // the "advance buffer". + uint32_t mMaxAdvanceBufferSegmentCount; + + int32_t mWriteSegment; + char* mWriteCursor; + char* mWriteLimit; + + // |mStatus| is protected by |mReentrantMonitor|. + nsresult mStatus; + bool mInited; +}; + +//----------------------------------------------------------------------------- + +// RAII class representing an active read segment. When it goes out of scope +// it automatically updates the read cursor and releases the read segment. +class MOZ_STACK_CLASS AutoReadSegment final +{ +public: + AutoReadSegment(nsPipe* aPipe, nsPipeReadState& aReadState, + uint32_t aMaxLength) + : mPipe(aPipe) + , mReadState(aReadState) + , mStatus(NS_ERROR_FAILURE) + , mSegment(nullptr) + , mLength(0) + , mOffset(0) + { + MOZ_ASSERT(mPipe); + MOZ_ASSERT(!mReadState.mActiveRead); + mStatus = mPipe->GetReadSegment(mReadState, mSegment, mLength); + if (NS_SUCCEEDED(mStatus)) { + MOZ_ASSERT(mReadState.mActiveRead); + MOZ_ASSERT(mSegment); + mLength = std::min(mLength, aMaxLength); + MOZ_ASSERT(mLength); + } + } + + ~AutoReadSegment() + { + if (NS_SUCCEEDED(mStatus)) { + if (mOffset) { + mPipe->AdvanceReadCursor(mReadState, mOffset); + } else { + nsPipeEvents events; + mPipe->ReleaseReadSegment(mReadState, events); + } + } + MOZ_ASSERT(!mReadState.mActiveRead); + } + + nsresult Status() const + { + return mStatus; + } + + const char* Data() const + { + MOZ_ASSERT(NS_SUCCEEDED(mStatus)); + MOZ_ASSERT(mSegment); + return mSegment + mOffset; + } + + uint32_t Length() const + { + MOZ_ASSERT(NS_SUCCEEDED(mStatus)); + MOZ_ASSERT(mLength >= mOffset); + return mLength - mOffset; + } + + void + Advance(uint32_t aCount) + { + MOZ_ASSERT(NS_SUCCEEDED(mStatus)); + MOZ_ASSERT(aCount <= (mLength - mOffset)); + mOffset += aCount; + } + + nsPipeReadState& + ReadState() const + { + return mReadState; + } + +private: + // guaranteed to remain alive due to limited stack lifetime of AutoReadSegment + nsPipe* mPipe; + nsPipeReadState& mReadState; + nsresult mStatus; + const char* mSegment; + uint32_t mLength; + uint32_t mOffset; +}; + +// +// NOTES on buffer architecture: +// +// +-----------------+ - - mBuffer.GetSegment(0) +// | | +// + - - - - - - - - + - - nsPipeReadState.mReadCursor +// |/////////////////| +// |/////////////////| +// |/////////////////| +// |/////////////////| +// +-----------------+ - - nsPipeReadState.mReadLimit +// | +// +-----------------+ +// |/////////////////| +// |/////////////////| +// |/////////////////| +// |/////////////////| +// |/////////////////| +// |/////////////////| +// +-----------------+ +// | +// +-----------------+ - - mBuffer.GetSegment(mWriteSegment) +// |/////////////////| +// |/////////////////| +// |/////////////////| +// + - - - - - - - - + - - mWriteCursor +// | | +// | | +// +-----------------+ - - mWriteLimit +// +// (shaded region contains data) +// +// NOTE: Each input stream produced by the nsPipe contains its own, separate +// nsPipeReadState. This means there are multiple mReadCursor and +// mReadLimit values in play. The pipe cannot discard old data until +// all mReadCursors have moved beyond that point in the stream. +// +// Likewise, each input stream reader will have it's own amount of +// buffered data. The pipe size threshold, however, is only applied +// to the input stream that is being read fastest. We call this +// the "advance buffer" in that its in advance of all readers. We +// allow slower input streams to buffer more data so that we don't +// stall processing of the faster input stream. +// +// NOTE: on some systems (notably OS/2), the heap allocator uses an arena for +// small allocations (e.g., 64 byte allocations). this means that buffers may +// be allocated back-to-back. in the diagram above, for example, mReadLimit +// would actually be pointing at the beginning of the next segment. when +// making changes to this file, please keep this fact in mind. +// + +//----------------------------------------------------------------------------- +// nsPipe methods: +//----------------------------------------------------------------------------- + +nsPipe::nsPipe() + : mOutput(this) + , mOriginalInput(new nsPipeInputStream(this)) + , mReentrantMonitor("nsPipe.mReentrantMonitor") + , mMaxAdvanceBufferSegmentCount(0) + , mWriteSegment(-1) + , mWriteCursor(nullptr) + , mWriteLimit(nullptr) + , mStatus(NS_OK) + , mInited(false) +{ + mInputList.AppendElement(mOriginalInput); +} + +nsPipe::~nsPipe() +{ +} + +NS_IMPL_ADDREF(nsPipe) +NS_IMPL_QUERY_INTERFACE(nsPipe, nsIPipe) + +NS_IMETHODIMP_(MozExternalRefCountType) +nsPipe::Release() +{ + MOZ_ASSERT(int32_t(mRefCnt) > 0, "dup release"); + nsrefcnt count = --mRefCnt; + NS_LOG_RELEASE(this, count, "nsPipe"); + if (count == 0) { + delete (this); + return 0; + } + // Avoid racing on |mOriginalInput| by only looking at it when + // the refcount is 1, that is, we are the only pointer (hence only + // thread) to access it. + if (count == 1 && mOriginalInput) { + mOriginalInput = nullptr; + return 1; + } + return count; +} + +NS_IMETHODIMP +nsPipe::Init(bool aNonBlockingIn, + bool aNonBlockingOut, + uint32_t aSegmentSize, + uint32_t aSegmentCount) +{ + mInited = true; + + if (aSegmentSize == 0) { + aSegmentSize = DEFAULT_SEGMENT_SIZE; + } + if (aSegmentCount == 0) { + aSegmentCount = DEFAULT_SEGMENT_COUNT; + } + + // protect against overflow + uint32_t maxCount = uint32_t(-1) / aSegmentSize; + if (aSegmentCount > maxCount) { + aSegmentCount = maxCount; + } + + // The internal buffer is always "infinite" so that we can allow + // the size to expand when cloned streams are read at different + // rates. We enforce a limit on how much data can be buffered + // ahead of the fastest reader in GetWriteSegment(). + nsresult rv = mBuffer.Init(aSegmentSize, UINT32_MAX); + if (NS_FAILED(rv)) { + return rv; + } + + mMaxAdvanceBufferSegmentCount = aSegmentCount; + + mOutput.SetNonBlocking(aNonBlockingOut); + mOriginalInput->SetNonBlocking(aNonBlockingIn); + + return NS_OK; +} + +NS_IMETHODIMP +nsPipe::GetInputStream(nsIAsyncInputStream** aInputStream) +{ + if (NS_WARN_IF(!mInited)) { + return NS_ERROR_NOT_INITIALIZED; + } + RefPtr<nsPipeInputStream> ref = mOriginalInput; + ref.forget(aInputStream); + return NS_OK; +} + +NS_IMETHODIMP +nsPipe::GetOutputStream(nsIAsyncOutputStream** aOutputStream) +{ + if (NS_WARN_IF(!mInited)) { + return NS_ERROR_NOT_INITIALIZED; + } + NS_ADDREF(*aOutputStream = &mOutput); + return NS_OK; +} + +void +nsPipe::PeekSegment(const nsPipeReadState& aReadState, uint32_t aIndex, + char*& aCursor, char*& aLimit) +{ + if (aIndex == 0) { + NS_ASSERTION(!aReadState.mReadCursor || mBuffer.GetSegmentCount(), + "unexpected state"); + aCursor = aReadState.mReadCursor; + aLimit = aReadState.mReadLimit; + } else { + uint32_t absoluteIndex = aReadState.mSegment + aIndex; + uint32_t numSegments = mBuffer.GetSegmentCount(); + if (absoluteIndex >= numSegments) { + aCursor = aLimit = nullptr; + } else { + aCursor = mBuffer.GetSegment(absoluteIndex); + if (mWriteSegment == (int32_t)absoluteIndex) { + aLimit = mWriteCursor; + } else { + aLimit = aCursor + mBuffer.GetSegmentSize(); + } + } + } +} + +nsresult +nsPipe::GetReadSegment(nsPipeReadState& aReadState, const char*& aSegment, + uint32_t& aLength) +{ + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + if (aReadState.mReadCursor == aReadState.mReadLimit) { + return NS_FAILED(mStatus) ? mStatus : NS_BASE_STREAM_WOULD_BLOCK; + } + + // The input stream locks the pipe while getting the buffer to read from, + // but then unlocks while actual data copying is taking place. In + // order to avoid deleting the buffer out from under this lockless read + // set a flag to indicate a read is active. This flag is only modified + // while the lock is held. + MOZ_ASSERT(!aReadState.mActiveRead); + aReadState.mActiveRead = true; + + aSegment = aReadState.mReadCursor; + aLength = aReadState.mReadLimit - aReadState.mReadCursor; + + return NS_OK; +} + +void +nsPipe::ReleaseReadSegment(nsPipeReadState& aReadState, nsPipeEvents& aEvents) +{ + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + MOZ_ASSERT(aReadState.mActiveRead); + aReadState.mActiveRead = false; + + // When a read completes and releases the mActiveRead flag, we may have blocked + // a drain from completing. This occurs when the input stream is closed during + // the read. In these cases, we need to complete the drain as soon as the + // active read completes. + if (aReadState.mNeedDrain) { + aReadState.mNeedDrain = false; + DrainInputStream(aReadState, aEvents); + } +} + +void +nsPipe::AdvanceReadCursor(nsPipeReadState& aReadState, uint32_t aBytesRead) +{ + NS_ASSERTION(aBytesRead, "don't call if no bytes read"); + + nsPipeEvents events; + { + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + LOG(("III advancing read cursor by %u\n", aBytesRead)); + NS_ASSERTION(aBytesRead <= mBuffer.GetSegmentSize(), "read too much"); + + aReadState.mReadCursor += aBytesRead; + NS_ASSERTION(aReadState.mReadCursor <= aReadState.mReadLimit, + "read cursor exceeds limit"); + + MOZ_ASSERT(aReadState.mAvailable >= aBytesRead); + aReadState.mAvailable -= aBytesRead; + + // Check to see if we're at the end of the available read data. If we + // are, and this segment is not still being written, then we can possibly + // free up the segment. + if (aReadState.mReadCursor == aReadState.mReadLimit && + !ReadSegmentBeingWritten(aReadState)) { + + // Advance the segment position. If we have read any segments from the + // advance buffer then we can potentially notify blocked writers. + if (AdvanceReadSegment(aReadState, mon) == SegmentAdvanceBufferRead && + mOutput.OnOutputWritable(events) == NotifyMonitor) { + mon.NotifyAll(); + } + } + + ReleaseReadSegment(aReadState, events); + } +} + +SegmentChangeResult +nsPipe::AdvanceReadSegment(nsPipeReadState& aReadState, + const ReentrantMonitorAutoEnter &ev) +{ + // Calculate how many segments are buffered for this stream to start. + uint32_t startBufferSegments = GetBufferSegmentCount(aReadState, ev); + + int32_t currentSegment = aReadState.mSegment; + + // Move to the next segment to read + aReadState.mSegment += 1; + + // If this was the last reference to the first segment, then remove it. + if (currentSegment == 0 && CountSegmentReferences(currentSegment) == 0) { + + // shift write and read segment index (-1 indicates an empty buffer). + mWriteSegment -= 1; + + // Directly modify the current read state. If the associated input + // stream is closed simultaneous with reading, then it may not be + // in the mInputList any more. + aReadState.mSegment -= 1; + + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + // Skip the current read state structure since we modify it manually + // before entering this loop. + if (&mInputList[i]->ReadState() == &aReadState) { + continue; + } + mInputList[i]->ReadState().mSegment -= 1; + } + + // done with this segment + mBuffer.DeleteFirstSegment(); + LOG(("III deleting first segment\n")); + } + + if (mWriteSegment < aReadState.mSegment) { + // read cursor has hit the end of written data, so reset it + MOZ_ASSERT(mWriteSegment == (aReadState.mSegment - 1)); + aReadState.mReadCursor = nullptr; + aReadState.mReadLimit = nullptr; + // also, the buffer is completely empty, so reset the write cursor + if (mWriteSegment == -1) { + mWriteCursor = nullptr; + mWriteLimit = nullptr; + } + } else { + // advance read cursor and limit to next buffer segment + aReadState.mReadCursor = mBuffer.GetSegment(aReadState.mSegment); + if (mWriteSegment == aReadState.mSegment) { + aReadState.mReadLimit = mWriteCursor; + } else { + aReadState.mReadLimit = aReadState.mReadCursor + mBuffer.GetSegmentSize(); + } + } + + // Calculate how many segments are buffered for the stream after + // reading. + uint32_t endBufferSegments = GetBufferSegmentCount(aReadState, ev); + + // If the stream has read a segment out of the set of advanced buffer + // segments, then the writer may advance. + if (startBufferSegments >= mMaxAdvanceBufferSegmentCount && + endBufferSegments < mMaxAdvanceBufferSegmentCount) { + return SegmentAdvanceBufferRead; + } + + // Otherwise there are no significant changes to the segment structure. + return SegmentNotChanged; +} + +void +nsPipe::DrainInputStream(nsPipeReadState& aReadState, nsPipeEvents& aEvents) +{ + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + // If a segment is actively being read in ReadSegments() for this input + // stream, then we cannot drain the stream. This can happen because + // ReadSegments() does not hold the lock while copying from the buffer. + // If we detect this condition, simply note that we need a drain once + // the read completes and return immediately. + if (aReadState.mActiveRead) { + MOZ_ASSERT(!aReadState.mNeedDrain); + aReadState.mNeedDrain = true; + return; + } + + aReadState.mAvailable = 0; + + while(mWriteSegment >= aReadState.mSegment) { + + // If the last segment to free is still being written to, we're done + // draining. We can't free any more. + if (ReadSegmentBeingWritten(aReadState)) { + break; + } + + // Don't bother checking if this results in an advance buffer segment + // read. Since we are draining the entire stream we will read an + // advance buffer segment no matter what. + AdvanceReadSegment(aReadState, mon); + } + + // If we have read any segments from the advance buffer then we can + // potentially notify blocked writers. + if (!IsAdvanceBufferFull(mon) && + mOutput.OnOutputWritable(aEvents) == NotifyMonitor) { + mon.NotifyAll(); + } +} + +bool +nsPipe::ReadSegmentBeingWritten(nsPipeReadState& aReadState) +{ + mReentrantMonitor.AssertCurrentThreadIn(); + bool beingWritten = mWriteSegment == aReadState.mSegment && + mWriteLimit > mWriteCursor; + NS_ASSERTION(!beingWritten || aReadState.mReadLimit == mWriteCursor, + "unexpected state"); + return beingWritten; +} + +nsresult +nsPipe::GetWriteSegment(char*& aSegment, uint32_t& aSegmentLen) +{ + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + if (NS_FAILED(mStatus)) { + return mStatus; + } + + // write cursor and limit may both be null indicating an empty buffer. + if (mWriteCursor == mWriteLimit) { + // The pipe is full if we have hit our limit on advance data buffering. + // This means the fastest reader is still reading slower than data is + // being written into the pipe. + if (IsAdvanceBufferFull(mon)) { + return NS_BASE_STREAM_WOULD_BLOCK; + } + + // The nsSegmentedBuffer is configured to be "infinite", so this + // should never return nullptr here. + char* seg = mBuffer.AppendNewSegment(); + if (!seg) { + return NS_ERROR_OUT_OF_MEMORY; + } + + LOG(("OOO appended new segment\n")); + mWriteCursor = seg; + mWriteLimit = mWriteCursor + mBuffer.GetSegmentSize(); + ++mWriteSegment; + } + + // make sure read cursor is initialized + SetAllNullReadCursors(); + + // check to see if we can roll-back our read and write cursors to the + // beginning of the current/first segment. this is purely an optimization. + if (mWriteSegment == 0 && AllReadCursorsMatchWriteCursor()) { + char* head = mBuffer.GetSegment(0); + LOG(("OOO rolling back write cursor %u bytes\n", mWriteCursor - head)); + RollBackAllReadCursors(head); + mWriteCursor = head; + } + + aSegment = mWriteCursor; + aSegmentLen = mWriteLimit - mWriteCursor; + return NS_OK; +} + +void +nsPipe::AdvanceWriteCursor(uint32_t aBytesWritten) +{ + NS_ASSERTION(aBytesWritten, "don't call if no bytes written"); + + nsPipeEvents events; + { + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + LOG(("OOO advancing write cursor by %u\n", aBytesWritten)); + + char* newWriteCursor = mWriteCursor + aBytesWritten; + NS_ASSERTION(newWriteCursor <= mWriteLimit, "write cursor exceeds limit"); + + // update read limit if reading in the same segment + UpdateAllReadCursors(newWriteCursor); + + mWriteCursor = newWriteCursor; + + ValidateAllReadCursors(); + + // update the writable flag on the output stream + if (mWriteCursor == mWriteLimit) { + mOutput.SetWritable(!IsAdvanceBufferFull(mon)); + } + + // notify input stream that pipe now contains additional data + bool needNotify = false; + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + if (mInputList[i]->OnInputReadable(aBytesWritten, events, mon) + == NotifyMonitor) { + needNotify = true; + } + } + + if (needNotify) { + mon.NotifyAll(); + } + } +} + +void +nsPipe::OnInputStreamException(nsPipeInputStream* aStream, nsresult aReason) +{ + MOZ_ASSERT(NS_FAILED(aReason)); + + nsPipeEvents events; + { + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + // Its possible to re-enter this method when we call OnPipeException() or + // OnInputExection() below. If there is a caller stuck in our synchronous + // Wait() method, then they will get woken up with a failure code which + // re-enters this method. Therefore, gracefully handle unknown streams + // here. + + // If we only have one stream open and it is the given stream, then shut + // down the entire pipe. + if (mInputList.Length() == 1) { + if (mInputList[0] == aStream) { + OnPipeException(aReason); + } + return; + } + + // Otherwise just close the particular stream that hit an exception. + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + if (mInputList[i] != aStream) { + continue; + } + + MonitorAction action = mInputList[i]->OnInputException(aReason, events, + mon); + mInputList.RemoveElementAt(i); + + // Notify after element is removed in case we re-enter as a result. + if (action == NotifyMonitor) { + mon.NotifyAll(); + } + + return; + } + } +} + +void +nsPipe::OnPipeException(nsresult aReason, bool aOutputOnly) +{ + LOG(("PPP nsPipe::OnPipeException [reason=%x output-only=%d]\n", + aReason, aOutputOnly)); + + nsPipeEvents events; + { + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + + // if we've already hit an exception, then ignore this one. + if (NS_FAILED(mStatus)) { + return; + } + + mStatus = aReason; + + bool needNotify = false; + + nsTArray<nsPipeInputStream*> tmpInputList; + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + // an output-only exception applies to the input end if the pipe has + // zero bytes available. + if (aOutputOnly && mInputList[i]->Available()) { + tmpInputList.AppendElement(mInputList[i]); + continue; + } + + if (mInputList[i]->OnInputException(aReason, events, mon) + == NotifyMonitor) { + needNotify = true; + } + } + mInputList = tmpInputList; + + if (mOutput.OnOutputException(aReason, events) == NotifyMonitor) { + needNotify = true; + } + + // Notify after we have removed any input streams from mInputList + if (needNotify) { + mon.NotifyAll(); + } + } +} + +nsresult +nsPipe::CloneInputStream(nsPipeInputStream* aOriginal, + nsIInputStream** aCloneOut) +{ + ReentrantMonitorAutoEnter mon(mReentrantMonitor); + RefPtr<nsPipeInputStream> ref = new nsPipeInputStream(*aOriginal); + mInputList.AppendElement(ref); + nsCOMPtr<nsIAsyncInputStream> downcast = ref.forget(); + downcast.forget(aCloneOut); + return NS_OK; +} + +uint32_t +nsPipe::CountSegmentReferences(int32_t aSegment) +{ + mReentrantMonitor.AssertCurrentThreadIn(); + uint32_t count = 0; + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + if (aSegment >= mInputList[i]->ReadState().mSegment) { + count += 1; + } + } + return count; +} + +void +nsPipe::SetAllNullReadCursors() +{ + mReentrantMonitor.AssertCurrentThreadIn(); + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + nsPipeReadState& readState = mInputList[i]->ReadState(); + if (!readState.mReadCursor) { + NS_ASSERTION(mWriteSegment == readState.mSegment, + "unexpected null read cursor"); + readState.mReadCursor = readState.mReadLimit = mWriteCursor; + } + } +} + +bool +nsPipe::AllReadCursorsMatchWriteCursor() +{ + mReentrantMonitor.AssertCurrentThreadIn(); + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + const nsPipeReadState& readState = mInputList[i]->ReadState(); + if (readState.mSegment != mWriteSegment || + readState.mReadCursor != mWriteCursor) { + return false; + } + } + return true; +} + +void +nsPipe::RollBackAllReadCursors(char* aWriteCursor) +{ + mReentrantMonitor.AssertCurrentThreadIn(); + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + nsPipeReadState& readState = mInputList[i]->ReadState(); + MOZ_ASSERT(mWriteSegment == readState.mSegment); + MOZ_ASSERT(mWriteCursor == readState.mReadCursor); + MOZ_ASSERT(mWriteCursor == readState.mReadLimit); + readState.mReadCursor = aWriteCursor; + readState.mReadLimit = aWriteCursor; + } +} + +void +nsPipe::UpdateAllReadCursors(char* aWriteCursor) +{ + mReentrantMonitor.AssertCurrentThreadIn(); + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + nsPipeReadState& readState = mInputList[i]->ReadState(); + if (mWriteSegment == readState.mSegment && + readState.mReadLimit == mWriteCursor) { + readState.mReadLimit = aWriteCursor; + } + } +} + +void +nsPipe::ValidateAllReadCursors() +{ + mReentrantMonitor.AssertCurrentThreadIn(); + // The only way mReadCursor == mWriteCursor is if: + // + // - mReadCursor is at the start of a segment (which, based on how + // nsSegmentedBuffer works, means that this segment is the "first" + // segment) + // - mWriteCursor points at the location past the end of the current + // write segment (so the current write filled the current write + // segment, so we've incremented mWriteCursor to point past the end + // of it) + // - the segment to which data has just been written is located + // exactly one segment's worth of bytes before the first segment + // where mReadCursor is located + // + // Consequently, the byte immediately after the end of the current + // write segment is the first byte of the first segment, so + // mReadCursor == mWriteCursor. (Another way to think about this is + // to consider the buffer architecture diagram above, but consider it + // with an arena allocator which allocates from the *end* of the + // arena to the *beginning* of the arena.) +#ifdef DEBUG + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + const nsPipeReadState& state = mInputList[i]->ReadState(); + NS_ASSERTION(state.mReadCursor != mWriteCursor || + (mBuffer.GetSegment(state.mSegment) == state.mReadCursor && + mWriteCursor == mWriteLimit), + "read cursor is bad"); + } +#endif +} + +uint32_t +nsPipe::GetBufferSegmentCount(const nsPipeReadState& aReadState, + const ReentrantMonitorAutoEnter& ev) const +{ + // The write segment can be smaller than the current reader position + // in some cases. For example, when the first write segment has not + // been allocated yet mWriteSegment is negative. In these cases + // the stream is effectively using zero segments. + if (mWriteSegment < aReadState.mSegment) { + return 0; + } + + MOZ_ASSERT(mWriteSegment >= 0); + MOZ_ASSERT(aReadState.mSegment >= 0); + + // Otherwise at least one segment is being used. We add one here + // since a single segment is being used when the write and read + // segment indices are the same. + return 1 + mWriteSegment - aReadState.mSegment; +} + +bool +nsPipe::IsAdvanceBufferFull(const ReentrantMonitorAutoEnter& ev) const +{ + // If we have fewer total segments than the limit we can immediately + // determine we are not full. Note, we must add one to mWriteSegment + // to convert from a index to a count. + MOZ_DIAGNOSTIC_ASSERT(mWriteSegment >= -1); + MOZ_DIAGNOSTIC_ASSERT(mWriteSegment < INT32_MAX); + uint32_t totalWriteSegments = mWriteSegment + 1; + if (totalWriteSegments < mMaxAdvanceBufferSegmentCount) { + return false; + } + + // Otherwise we must inspect all of our reader streams. We need + // to determine the buffer depth of the fastest reader. + uint32_t minBufferSegments = UINT32_MAX; + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + // Only count buffer segments from input streams that are open. + if (NS_FAILED(mInputList[i]->Status(ev))) { + continue; + } + const nsPipeReadState& state = mInputList[i]->ReadState(); + uint32_t bufferSegments = GetBufferSegmentCount(state, ev); + minBufferSegments = std::min(minBufferSegments, bufferSegments); + // We only care if any reader has fewer segments buffered than + // our threshold. We can stop once we hit that threshold. + if (minBufferSegments < mMaxAdvanceBufferSegmentCount) { + return false; + } + } + + // Note, its possible for minBufferSegments to exceed our + // mMaxAdvanceBufferSegmentCount here. This happens when a cloned + // reader gets far behind, but then the fastest reader stream is + // closed. This leaves us with a single stream that is buffered + // beyond our max. Naturally we continue to indicate the pipe + // is full at this point. + + return true; +} + +//----------------------------------------------------------------------------- +// nsPipeEvents methods: +//----------------------------------------------------------------------------- + +nsPipeEvents::~nsPipeEvents() +{ + // dispatch any pending events + + for (uint32_t i = 0; i < mInputList.Length(); ++i) { + mInputList[i].mCallback->OnInputStreamReady(mInputList[i].mStream); + } + mInputList.Clear(); + + if (mOutputCallback) { + mOutputCallback->OnOutputStreamReady(mOutputStream); + mOutputCallback = nullptr; + mOutputStream = nullptr; + } +} + +//----------------------------------------------------------------------------- +// nsPipeInputStream methods: +//----------------------------------------------------------------------------- + +NS_IMPL_ADDREF(nsPipeInputStream); +NS_IMPL_RELEASE(nsPipeInputStream); + +NS_INTERFACE_TABLE_HEAD(nsPipeInputStream) + NS_INTERFACE_TABLE_BEGIN + NS_INTERFACE_TABLE_ENTRY(nsPipeInputStream, nsIAsyncInputStream) + NS_INTERFACE_TABLE_ENTRY(nsPipeInputStream, nsISeekableStream) + NS_INTERFACE_TABLE_ENTRY(nsPipeInputStream, nsISearchableInputStream) + NS_INTERFACE_TABLE_ENTRY(nsPipeInputStream, nsICloneableInputStream) + NS_INTERFACE_TABLE_ENTRY(nsPipeInputStream, nsIBufferedInputStream) + NS_INTERFACE_TABLE_ENTRY(nsPipeInputStream, nsIClassInfo) + NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(nsPipeInputStream, nsIInputStream, + nsIAsyncInputStream) + NS_INTERFACE_TABLE_ENTRY_AMBIGUOUS(nsPipeInputStream, nsISupports, + nsIAsyncInputStream) + NS_INTERFACE_TABLE_END +NS_INTERFACE_TABLE_TAIL + +NS_IMPL_CI_INTERFACE_GETTER(nsPipeInputStream, + nsIInputStream, + nsIAsyncInputStream, + nsISeekableStream, + nsISearchableInputStream, + nsICloneableInputStream, + nsIBufferedInputStream) + +NS_IMPL_THREADSAFE_CI(nsPipeInputStream) + +NS_IMETHODIMP +nsPipeInputStream::Init(nsIInputStream*, uint32_t) +{ + MOZ_CRASH("nsPipeInputStream should never be initialized with " + "nsIBufferedInputStream::Init!\n"); +} + +uint32_t +nsPipeInputStream::Available() +{ + mPipe->mReentrantMonitor.AssertCurrentThreadIn(); + return mReadState.mAvailable; +} + +nsresult +nsPipeInputStream::Wait() +{ + NS_ASSERTION(mBlocking, "wait on non-blocking pipe input stream"); + + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + while (NS_SUCCEEDED(Status(mon)) && (mReadState.mAvailable == 0)) { + LOG(("III pipe input: waiting for data\n")); + + mBlocked = true; + mon.Wait(); + mBlocked = false; + + LOG(("III pipe input: woke up [status=%x available=%u]\n", + Status(mon), mReadState.mAvailable)); + } + + return Status(mon) == NS_BASE_STREAM_CLOSED ? NS_OK : Status(mon); +} + +MonitorAction +nsPipeInputStream::OnInputReadable(uint32_t aBytesWritten, + nsPipeEvents& aEvents, + const ReentrantMonitorAutoEnter& ev) +{ + MonitorAction result = DoNotNotifyMonitor; + + mPipe->mReentrantMonitor.AssertCurrentThreadIn(); + mReadState.mAvailable += aBytesWritten; + + if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) { + aEvents.NotifyInputReady(this, mCallback); + mCallback = nullptr; + mCallbackFlags = 0; + } else if (mBlocked) { + result = NotifyMonitor; + } + + return result; +} + +MonitorAction +nsPipeInputStream::OnInputException(nsresult aReason, nsPipeEvents& aEvents, + const ReentrantMonitorAutoEnter& ev) +{ + LOG(("nsPipeInputStream::OnInputException [this=%x reason=%x]\n", + this, aReason)); + + MonitorAction result = DoNotNotifyMonitor; + + NS_ASSERTION(NS_FAILED(aReason), "huh? successful exception"); + + if (NS_SUCCEEDED(mInputStatus)) { + mInputStatus = aReason; + } + + // force count of available bytes to zero. + mPipe->DrainInputStream(mReadState, aEvents); + + if (mCallback) { + aEvents.NotifyInputReady(this, mCallback); + mCallback = nullptr; + mCallbackFlags = 0; + } else if (mBlocked) { + result = NotifyMonitor; + } + + return result; +} + +NS_IMETHODIMP +nsPipeInputStream::CloseWithStatus(nsresult aReason) +{ + LOG(("III CloseWithStatus [this=%x reason=%x]\n", this, aReason)); + + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + if (NS_FAILED(mInputStatus)) { + return NS_OK; + } + + if (NS_SUCCEEDED(aReason)) { + aReason = NS_BASE_STREAM_CLOSED; + } + + mPipe->OnInputStreamException(this, aReason); + return NS_OK; +} + +NS_IMETHODIMP +nsPipeInputStream::Close() +{ + return CloseWithStatus(NS_BASE_STREAM_CLOSED); +} + +NS_IMETHODIMP +nsPipeInputStream::Available(uint64_t* aResult) +{ + // nsPipeInputStream supports under 4GB stream only + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + // return error if closed + if (!mReadState.mAvailable && NS_FAILED(Status(mon))) { + return Status(mon); + } + + *aResult = (uint64_t)mReadState.mAvailable; + return NS_OK; +} + +NS_IMETHODIMP +nsPipeInputStream::ReadSegments(nsWriteSegmentFun aWriter, + void* aClosure, + uint32_t aCount, + uint32_t* aReadCount) +{ + LOG(("III ReadSegments [this=%x count=%u]\n", this, aCount)); + + nsresult rv = NS_OK; + + *aReadCount = 0; + while (aCount) { + AutoReadSegment segment(mPipe, mReadState, aCount); + rv = segment.Status(); + if (NS_FAILED(rv)) { + // ignore this error if we've already read something. + if (*aReadCount > 0) { + rv = NS_OK; + break; + } + if (rv == NS_BASE_STREAM_WOULD_BLOCK) { + // pipe is empty + if (!mBlocking) { + break; + } + // wait for some data to be written to the pipe + rv = Wait(); + if (NS_SUCCEEDED(rv)) { + continue; + } + } + // ignore this error, just return. + if (rv == NS_BASE_STREAM_CLOSED) { + rv = NS_OK; + break; + } + mPipe->OnInputStreamException(this, rv); + break; + } + + uint32_t writeCount; + while (segment.Length()) { + writeCount = 0; + + rv = aWriter(static_cast<nsIAsyncInputStream*>(this), aClosure, + segment.Data(), *aReadCount, segment.Length(), &writeCount); + + if (NS_FAILED(rv) || writeCount == 0) { + aCount = 0; + // any errors returned from the writer end here: do not + // propagate to the caller of ReadSegments. + rv = NS_OK; + break; + } + + NS_ASSERTION(writeCount <= segment.Length(), "wrote more than expected"); + segment.Advance(writeCount); + aCount -= writeCount; + *aReadCount += writeCount; + mLogicalOffset += writeCount; + } + } + + return rv; +} + +NS_IMETHODIMP +nsPipeInputStream::Read(char* aToBuf, uint32_t aBufLen, uint32_t* aReadCount) +{ + return ReadSegments(NS_CopySegmentToBuffer, aToBuf, aBufLen, aReadCount); +} + +NS_IMETHODIMP +nsPipeInputStream::IsNonBlocking(bool* aNonBlocking) +{ + *aNonBlocking = !mBlocking; + return NS_OK; +} + +NS_IMETHODIMP +nsPipeInputStream::AsyncWait(nsIInputStreamCallback* aCallback, + uint32_t aFlags, + uint32_t aRequestedCount, + nsIEventTarget* aTarget) +{ + LOG(("III AsyncWait [this=%x]\n", this)); + + nsPipeEvents pipeEvents; + { + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + // replace a pending callback + mCallback = nullptr; + mCallbackFlags = 0; + + if (!aCallback) { + return NS_OK; + } + + nsCOMPtr<nsIInputStreamCallback> proxy; + if (aTarget) { + proxy = NS_NewInputStreamReadyEvent(aCallback, aTarget); + aCallback = proxy; + } + + if (NS_FAILED(Status(mon)) || + (mReadState.mAvailable && !(aFlags & WAIT_CLOSURE_ONLY))) { + // stream is already closed or readable; post event. + pipeEvents.NotifyInputReady(this, aCallback); + } else { + // queue up callback object to be notified when data becomes available + mCallback = aCallback; + mCallbackFlags = aFlags; + } + } + return NS_OK; +} + +NS_IMETHODIMP +nsPipeInputStream::Seek(int32_t aWhence, int64_t aOffset) +{ + NS_NOTREACHED("nsPipeInputStream::Seek"); + return NS_ERROR_NOT_IMPLEMENTED; +} + +NS_IMETHODIMP +nsPipeInputStream::Tell(int64_t* aOffset) +{ + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + // return error if closed + if (!mReadState.mAvailable && NS_FAILED(Status(mon))) { + return Status(mon); + } + + *aOffset = mLogicalOffset; + return NS_OK; +} + +NS_IMETHODIMP +nsPipeInputStream::SetEOF() +{ + NS_NOTREACHED("nsPipeInputStream::SetEOF"); + return NS_ERROR_NOT_IMPLEMENTED; +} + +static bool strings_equal(bool aIgnoreCase, + const char* aS1, const char* aS2, uint32_t aLen) +{ + return aIgnoreCase + ? !nsCRT::strncasecmp(aS1, aS2, aLen) : !nsCRT::strncmp(aS1, aS2, aLen); +} + +NS_IMETHODIMP +nsPipeInputStream::Search(const char* aForString, + bool aIgnoreCase, + bool* aFound, + uint32_t* aOffsetSearchedTo) +{ + LOG(("III Search [for=%s ic=%u]\n", aForString, aIgnoreCase)); + + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + char* cursor1; + char* limit1; + uint32_t index = 0, offset = 0; + uint32_t strLen = strlen(aForString); + + mPipe->PeekSegment(mReadState, 0, cursor1, limit1); + if (cursor1 == limit1) { + *aFound = false; + *aOffsetSearchedTo = 0; + LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo)); + return NS_OK; + } + + while (true) { + uint32_t i, len1 = limit1 - cursor1; + + // check if the string is in the buffer segment + for (i = 0; i < len1 - strLen + 1; i++) { + if (strings_equal(aIgnoreCase, &cursor1[i], aForString, strLen)) { + *aFound = true; + *aOffsetSearchedTo = offset + i; + LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo)); + return NS_OK; + } + } + + // get the next segment + char* cursor2; + char* limit2; + uint32_t len2; + + index++; + offset += len1; + + mPipe->PeekSegment(mReadState, index, cursor2, limit2); + if (cursor2 == limit2) { + *aFound = false; + *aOffsetSearchedTo = offset - strLen + 1; + LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo)); + return NS_OK; + } + len2 = limit2 - cursor2; + + // check if the string is straddling the next buffer segment + uint32_t lim = XPCOM_MIN(strLen, len2 + 1); + for (i = 0; i < lim; ++i) { + uint32_t strPart1Len = strLen - i - 1; + uint32_t strPart2Len = strLen - strPart1Len; + const char* strPart2 = &aForString[strLen - strPart2Len]; + uint32_t bufSeg1Offset = len1 - strPart1Len; + if (strings_equal(aIgnoreCase, &cursor1[bufSeg1Offset], aForString, strPart1Len) && + strings_equal(aIgnoreCase, cursor2, strPart2, strPart2Len)) { + *aFound = true; + *aOffsetSearchedTo = offset - strPart1Len; + LOG((" result [aFound=%u offset=%u]\n", *aFound, *aOffsetSearchedTo)); + return NS_OK; + } + } + + // finally continue with the next buffer + cursor1 = cursor2; + limit1 = limit2; + } + + NS_NOTREACHED("can't get here"); + return NS_ERROR_UNEXPECTED; // keep compiler happy +} + +NS_IMETHODIMP +nsPipeInputStream::GetCloneable(bool* aCloneableOut) +{ + *aCloneableOut = true; + return NS_OK; +} + +NS_IMETHODIMP +nsPipeInputStream::Clone(nsIInputStream** aCloneOut) +{ + return mPipe->CloneInputStream(this, aCloneOut); +} + +nsresult +nsPipeInputStream::Status(const ReentrantMonitorAutoEnter& ev) const +{ + if (NS_FAILED(mInputStatus)) { + return mInputStatus; + } + + if (mReadState.mAvailable) { + // Still something to read and this input stream state is OK. + return NS_OK; + } + + // Nothing to read, just fall through to the pipe's state that + // may reflect state of its output stream side (already closed). + return mPipe->mStatus; +} + +nsresult +nsPipeInputStream::Status() const +{ + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + return Status(mon); +} + +nsPipeInputStream::~nsPipeInputStream() +{ + Close(); +} + +//----------------------------------------------------------------------------- +// nsPipeOutputStream methods: +//----------------------------------------------------------------------------- + +NS_IMPL_QUERY_INTERFACE(nsPipeOutputStream, + nsIOutputStream, + nsIAsyncOutputStream, + nsIClassInfo) + +NS_IMPL_CI_INTERFACE_GETTER(nsPipeOutputStream, + nsIOutputStream, + nsIAsyncOutputStream) + +NS_IMPL_THREADSAFE_CI(nsPipeOutputStream) + +nsresult +nsPipeOutputStream::Wait() +{ + NS_ASSERTION(mBlocking, "wait on non-blocking pipe output stream"); + + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + if (NS_SUCCEEDED(mPipe->mStatus) && !mWritable) { + LOG(("OOO pipe output: waiting for space\n")); + mBlocked = true; + mon.Wait(); + mBlocked = false; + LOG(("OOO pipe output: woke up [pipe-status=%x writable=%u]\n", + mPipe->mStatus, mWritable)); + } + + return mPipe->mStatus == NS_BASE_STREAM_CLOSED ? NS_OK : mPipe->mStatus; +} + +MonitorAction +nsPipeOutputStream::OnOutputWritable(nsPipeEvents& aEvents) +{ + MonitorAction result = DoNotNotifyMonitor; + + mWritable = true; + + if (mCallback && !(mCallbackFlags & WAIT_CLOSURE_ONLY)) { + aEvents.NotifyOutputReady(this, mCallback); + mCallback = nullptr; + mCallbackFlags = 0; + } else if (mBlocked) { + result = NotifyMonitor; + } + + return result; +} + +MonitorAction +nsPipeOutputStream::OnOutputException(nsresult aReason, nsPipeEvents& aEvents) +{ + LOG(("nsPipeOutputStream::OnOutputException [this=%x reason=%x]\n", + this, aReason)); + + MonitorAction result = DoNotNotifyMonitor; + + NS_ASSERTION(NS_FAILED(aReason), "huh? successful exception"); + mWritable = false; + + if (mCallback) { + aEvents.NotifyOutputReady(this, mCallback); + mCallback = nullptr; + mCallbackFlags = 0; + } else if (mBlocked) { + result = NotifyMonitor; + } + + return result; +} + + +NS_IMETHODIMP_(MozExternalRefCountType) +nsPipeOutputStream::AddRef() +{ + ++mWriterRefCnt; + return mPipe->AddRef(); +} + +NS_IMETHODIMP_(MozExternalRefCountType) +nsPipeOutputStream::Release() +{ + if (--mWriterRefCnt == 0) { + Close(); + } + return mPipe->Release(); +} + +NS_IMETHODIMP +nsPipeOutputStream::CloseWithStatus(nsresult aReason) +{ + LOG(("OOO CloseWithStatus [this=%x reason=%x]\n", this, aReason)); + + if (NS_SUCCEEDED(aReason)) { + aReason = NS_BASE_STREAM_CLOSED; + } + + // input stream may remain open + mPipe->OnPipeException(aReason, true); + return NS_OK; +} + +NS_IMETHODIMP +nsPipeOutputStream::Close() +{ + return CloseWithStatus(NS_BASE_STREAM_CLOSED); +} + +NS_IMETHODIMP +nsPipeOutputStream::WriteSegments(nsReadSegmentFun aReader, + void* aClosure, + uint32_t aCount, + uint32_t* aWriteCount) +{ + LOG(("OOO WriteSegments [this=%x count=%u]\n", this, aCount)); + + nsresult rv = NS_OK; + + char* segment; + uint32_t segmentLen; + + *aWriteCount = 0; + while (aCount) { + rv = mPipe->GetWriteSegment(segment, segmentLen); + if (NS_FAILED(rv)) { + if (rv == NS_BASE_STREAM_WOULD_BLOCK) { + // pipe is full + if (!mBlocking) { + // ignore this error if we've already written something + if (*aWriteCount > 0) { + rv = NS_OK; + } + break; + } + // wait for the pipe to have an empty segment. + rv = Wait(); + if (NS_SUCCEEDED(rv)) { + continue; + } + } + mPipe->OnPipeException(rv); + break; + } + + // write no more than aCount + if (segmentLen > aCount) { + segmentLen = aCount; + } + + uint32_t readCount, originalLen = segmentLen; + while (segmentLen) { + readCount = 0; + + rv = aReader(this, aClosure, segment, *aWriteCount, segmentLen, &readCount); + + if (NS_FAILED(rv) || readCount == 0) { + aCount = 0; + // any errors returned from the aReader end here: do not + // propagate to the caller of WriteSegments. + rv = NS_OK; + break; + } + + NS_ASSERTION(readCount <= segmentLen, "read more than expected"); + segment += readCount; + segmentLen -= readCount; + aCount -= readCount; + *aWriteCount += readCount; + mLogicalOffset += readCount; + } + + if (segmentLen < originalLen) { + mPipe->AdvanceWriteCursor(originalLen - segmentLen); + } + } + + return rv; +} + +static nsresult +nsReadFromRawBuffer(nsIOutputStream* aOutStr, + void* aClosure, + char* aToRawSegment, + uint32_t aOffset, + uint32_t aCount, + uint32_t* aReadCount) +{ + const char* fromBuf = (const char*)aClosure; + memcpy(aToRawSegment, &fromBuf[aOffset], aCount); + *aReadCount = aCount; + return NS_OK; +} + +NS_IMETHODIMP +nsPipeOutputStream::Write(const char* aFromBuf, + uint32_t aBufLen, + uint32_t* aWriteCount) +{ + return WriteSegments(nsReadFromRawBuffer, (void*)aFromBuf, aBufLen, aWriteCount); +} + +NS_IMETHODIMP +nsPipeOutputStream::Flush(void) +{ + // nothing to do + return NS_OK; +} + +static nsresult +nsReadFromInputStream(nsIOutputStream* aOutStr, + void* aClosure, + char* aToRawSegment, + uint32_t aOffset, + uint32_t aCount, + uint32_t* aReadCount) +{ + nsIInputStream* fromStream = (nsIInputStream*)aClosure; + return fromStream->Read(aToRawSegment, aCount, aReadCount); +} + +NS_IMETHODIMP +nsPipeOutputStream::WriteFrom(nsIInputStream* aFromStream, + uint32_t aCount, + uint32_t* aWriteCount) +{ + return WriteSegments(nsReadFromInputStream, aFromStream, aCount, aWriteCount); +} + +NS_IMETHODIMP +nsPipeOutputStream::IsNonBlocking(bool* aNonBlocking) +{ + *aNonBlocking = !mBlocking; + return NS_OK; +} + +NS_IMETHODIMP +nsPipeOutputStream::AsyncWait(nsIOutputStreamCallback* aCallback, + uint32_t aFlags, + uint32_t aRequestedCount, + nsIEventTarget* aTarget) +{ + LOG(("OOO AsyncWait [this=%x]\n", this)); + + nsPipeEvents pipeEvents; + { + ReentrantMonitorAutoEnter mon(mPipe->mReentrantMonitor); + + // replace a pending callback + mCallback = nullptr; + mCallbackFlags = 0; + + if (!aCallback) { + return NS_OK; + } + + nsCOMPtr<nsIOutputStreamCallback> proxy; + if (aTarget) { + proxy = NS_NewOutputStreamReadyEvent(aCallback, aTarget); + aCallback = proxy; + } + + if (NS_FAILED(mPipe->mStatus) || + (mWritable && !(aFlags & WAIT_CLOSURE_ONLY))) { + // stream is already closed or writable; post event. + pipeEvents.NotifyOutputReady(this, aCallback); + } else { + // queue up callback object to be notified when data becomes available + mCallback = aCallback; + mCallbackFlags = aFlags; + } + } + return NS_OK; +} + +//////////////////////////////////////////////////////////////////////////////// + +nsresult +NS_NewPipe(nsIInputStream** aPipeIn, + nsIOutputStream** aPipeOut, + uint32_t aSegmentSize, + uint32_t aMaxSize, + bool aNonBlockingInput, + bool aNonBlockingOutput) +{ + if (aSegmentSize == 0) { + aSegmentSize = DEFAULT_SEGMENT_SIZE; + } + + // Handle aMaxSize of UINT32_MAX as a special case + uint32_t segmentCount; + if (aMaxSize == UINT32_MAX) { + segmentCount = UINT32_MAX; + } else { + segmentCount = aMaxSize / aSegmentSize; + } + + nsIAsyncInputStream* in; + nsIAsyncOutputStream* out; + nsresult rv = NS_NewPipe2(&in, &out, aNonBlockingInput, aNonBlockingOutput, + aSegmentSize, segmentCount); + if (NS_FAILED(rv)) { + return rv; + } + + *aPipeIn = in; + *aPipeOut = out; + return NS_OK; +} + +nsresult +NS_NewPipe2(nsIAsyncInputStream** aPipeIn, + nsIAsyncOutputStream** aPipeOut, + bool aNonBlockingInput, + bool aNonBlockingOutput, + uint32_t aSegmentSize, + uint32_t aSegmentCount) +{ + nsPipe* pipe = new nsPipe(); + nsresult rv = pipe->Init(aNonBlockingInput, + aNonBlockingOutput, + aSegmentSize, + aSegmentCount); + if (NS_FAILED(rv)) { + NS_ADDREF(pipe); + NS_RELEASE(pipe); + return rv; + } + + // These always succeed because the pipe is initialized above. + MOZ_ALWAYS_SUCCEEDS(pipe->GetInputStream(aPipeIn)); + MOZ_ALWAYS_SUCCEEDS(pipe->GetOutputStream(aPipeOut)); + return NS_OK; +} + +nsresult +nsPipeConstructor(nsISupports* aOuter, REFNSIID aIID, void** aResult) +{ + if (aOuter) { + return NS_ERROR_NO_AGGREGATION; + } + nsPipe* pipe = new nsPipe(); + NS_ADDREF(pipe); + nsresult rv = pipe->QueryInterface(aIID, aResult); + NS_RELEASE(pipe); + return rv; +} + +//////////////////////////////////////////////////////////////////////////////// |