diff options
Diffstat (limited to 'netwerk/base/nsInputStreamPump.cpp')
| -rw-r--r-- | netwerk/base/nsInputStreamPump.cpp | 786 |
1 files changed, 786 insertions, 0 deletions
diff --git a/netwerk/base/nsInputStreamPump.cpp b/netwerk/base/nsInputStreamPump.cpp new file mode 100644 index 0000000000..11aac4cd9b --- /dev/null +++ b/netwerk/base/nsInputStreamPump.cpp @@ -0,0 +1,786 @@ +/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ +/* vim:set ts=4 sts=2 sw=2 et cin: */ +/* 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 "nsIOService.h" +#include "nsInputStreamPump.h" +#include "nsIStreamTransportService.h" +#include "nsIThreadRetargetableStreamListener.h" +#include "nsThreadUtils.h" +#include "nsCOMPtr.h" +#include "mozilla/Logging.h" +#include "mozilla/NonBlockingAsyncInputStream.h" +#include "mozilla/ProfilerLabels.h" +#include "mozilla/SlicedInputStream.h" +#include "nsIStreamListener.h" +#include "nsILoadGroup.h" +#include "nsNetCID.h" +#include "nsNetUtil.h" +#include "nsStreamUtils.h" +#include <algorithm> + +// +// MOZ_LOG=nsStreamPump:5 +// +static mozilla::LazyLogModule gStreamPumpLog("nsStreamPump"); +#undef LOG +#define LOG(args) MOZ_LOG(gStreamPumpLog, mozilla::LogLevel::Debug, args) + +//----------------------------------------------------------------------------- +// nsInputStreamPump methods +//----------------------------------------------------------------------------- + +nsInputStreamPump::nsInputStreamPump() : mOffMainThread(!NS_IsMainThread()) {} + +nsresult nsInputStreamPump::Create(nsInputStreamPump** result, + nsIInputStream* stream, uint32_t segsize, + uint32_t segcount, bool closeWhenDone, + nsISerialEventTarget* mainThreadTarget) { + nsresult rv = NS_ERROR_OUT_OF_MEMORY; + RefPtr<nsInputStreamPump> pump = new nsInputStreamPump(); + if (pump) { + rv = pump->Init(stream, segsize, segcount, closeWhenDone, mainThreadTarget); + if (NS_SUCCEEDED(rv)) { + pump.forget(result); + } + } + return rv; +} + +struct PeekData { + PeekData(nsInputStreamPump::PeekSegmentFun fun, void* closure) + : mFunc(fun), mClosure(closure) {} + + nsInputStreamPump::PeekSegmentFun mFunc; + void* mClosure; +}; + +static nsresult CallPeekFunc(nsIInputStream* aInStream, void* aClosure, + const char* aFromSegment, uint32_t aToOffset, + uint32_t aCount, uint32_t* aWriteCount) { + NS_ASSERTION(aToOffset == 0, "Called more than once?"); + NS_ASSERTION(aCount > 0, "Called without data?"); + + PeekData* data = static_cast<PeekData*>(aClosure); + data->mFunc(data->mClosure, reinterpret_cast<const uint8_t*>(aFromSegment), + aCount); + return NS_BINDING_ABORTED; +} + +nsresult nsInputStreamPump::PeekStream(PeekSegmentFun callback, void* closure) { + RecursiveMutexAutoLock lock(mMutex); + + MOZ_ASSERT(mAsyncStream, "PeekStream called without stream"); + + nsresult rv = CreateBufferedStreamIfNeeded(); + NS_ENSURE_SUCCESS(rv, rv); + + // See if the pipe is closed by checking the return of Available. + uint64_t dummy64; + rv = mAsyncStream->Available(&dummy64); + if (NS_FAILED(rv)) return rv; + uint32_t dummy = (uint32_t)std::min(dummy64, (uint64_t)UINT32_MAX); + + PeekData data(callback, closure); + return mAsyncStream->ReadSegments( + CallPeekFunc, &data, mozilla::net::nsIOService::gDefaultSegmentSize, + &dummy); +} + +nsresult nsInputStreamPump::EnsureWaiting() { + mMutex.AssertCurrentThreadIn(); + + // no need to worry about multiple threads... an input stream pump lives + // on only one thread at a time. + MOZ_ASSERT(mAsyncStream); + if (!mWaitingForInputStreamReady && !mProcessingCallbacks) { + // Ensure OnStateStop is called on the main thread only when this pump is + // created on main thread. + if (mState == STATE_STOP && !mOffMainThread) { + nsCOMPtr<nsISerialEventTarget> mainThread = + mLabeledMainThreadTarget + ? mLabeledMainThreadTarget + : do_AddRef(mozilla::GetMainThreadSerialEventTarget()); + if (mTargetThread != mainThread) { + mTargetThread = mainThread; + } + } + MOZ_ASSERT(mTargetThread); + nsresult rv = mAsyncStream->AsyncWait(this, 0, 0, mTargetThread); + if (NS_FAILED(rv)) { + NS_ERROR("AsyncWait failed"); + return rv; + } + // Any retargeting during STATE_START or START_TRANSFER is complete + // after the call to AsyncWait; next callback will be on mTargetThread. + mRetargeting = false; + mWaitingForInputStreamReady = true; + } + return NS_OK; +} + +//----------------------------------------------------------------------------- +// nsInputStreamPump::nsISupports +//----------------------------------------------------------------------------- + +// although this class can only be accessed from one thread at a time, we do +// allow its ownership to move from thread to thread, assuming the consumer +// understands the limitations of this. +NS_IMPL_ADDREF(nsInputStreamPump) +NS_IMPL_RELEASE(nsInputStreamPump) +NS_INTERFACE_MAP_BEGIN(nsInputStreamPump) + NS_INTERFACE_MAP_ENTRY(nsIRequest) + NS_INTERFACE_MAP_ENTRY(nsIThreadRetargetableRequest) + NS_INTERFACE_MAP_ENTRY(nsIInputStreamCallback) + NS_INTERFACE_MAP_ENTRY(nsIInputStreamPump) + NS_INTERFACE_MAP_ENTRY_CONCRETE(nsInputStreamPump) + NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIInputStreamPump) +NS_INTERFACE_MAP_END + +//----------------------------------------------------------------------------- +// nsInputStreamPump::nsIRequest +//----------------------------------------------------------------------------- + +NS_IMETHODIMP +nsInputStreamPump::GetName(nsACString& result) { + RecursiveMutexAutoLock lock(mMutex); + + result.Truncate(); + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::IsPending(bool* result) { + RecursiveMutexAutoLock lock(mMutex); + + *result = (mState != STATE_IDLE && mState != STATE_DEAD); + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::GetStatus(nsresult* status) { + RecursiveMutexAutoLock lock(mMutex); + + *status = mStatus; + return NS_OK; +} + +NS_IMETHODIMP nsInputStreamPump::SetCanceledReason(const nsACString& aReason) { + return SetCanceledReasonImpl(aReason); +} + +NS_IMETHODIMP nsInputStreamPump::GetCanceledReason(nsACString& aReason) { + return GetCanceledReasonImpl(aReason); +} + +NS_IMETHODIMP nsInputStreamPump::CancelWithReason(nsresult aStatus, + const nsACString& aReason) { + return CancelWithReasonImpl(aStatus, aReason); +} + +NS_IMETHODIMP +nsInputStreamPump::Cancel(nsresult status) { + RecursiveMutexAutoLock lock(mMutex); + + AssertOnThread(); + + LOG(("nsInputStreamPump::Cancel [this=%p status=%" PRIx32 "]\n", this, + static_cast<uint32_t>(status))); + + if (NS_FAILED(mStatus)) { + LOG((" already canceled\n")); + return NS_OK; + } + + NS_ASSERTION(NS_FAILED(status), "cancel with non-failure status code"); + mStatus = status; + + // close input stream + if (mAsyncStream) { + // If mSuspendCount != 0, EnsureWaiting will be called by Resume(). + // Note that while suspended, OnInputStreamReady will + // not do anything, and also note that calling asyncWait + // on a closed stream works and will dispatch an event immediately. + + nsCOMPtr<nsIEventTarget> currentTarget = NS_GetCurrentThread(); + if (mTargetThread && currentTarget != mTargetThread) { + nsresult rv = mTargetThread->Dispatch(NS_NewRunnableFunction( + "nsInputStreamPump::Cancel", [self = RefPtr{this}, status] { + RecursiveMutexAutoLock lock(self->mMutex); + if (!self->mAsyncStream) { + return; + } + self->mAsyncStream->CloseWithStatus(status); + if (self->mSuspendCount == 0) { + self->EnsureWaiting(); + } + })); + NS_ENSURE_SUCCESS(rv, rv); + } else { + mAsyncStream->CloseWithStatus(status); + if (mSuspendCount == 0) { + EnsureWaiting(); + } + } + } + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::Suspend() { + RecursiveMutexAutoLock lock(mMutex); + + LOG(("nsInputStreamPump::Suspend [this=%p]\n", this)); + NS_ENSURE_TRUE(mState != STATE_IDLE && mState != STATE_DEAD, + NS_ERROR_UNEXPECTED); + ++mSuspendCount; + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::Resume() { + RecursiveMutexAutoLock lock(mMutex); + + LOG(("nsInputStreamPump::Resume [this=%p]\n", this)); + NS_ENSURE_TRUE(mSuspendCount > 0, NS_ERROR_UNEXPECTED); + NS_ENSURE_TRUE(mState != STATE_IDLE && mState != STATE_DEAD, + NS_ERROR_UNEXPECTED); + + // There is a brief in-between state when we null out mAsyncStream in + // OnStateStop() before calling OnStopRequest, and only afterwards set + // STATE_DEAD, which we need to handle gracefully. + if (--mSuspendCount == 0 && mAsyncStream) { + EnsureWaiting(); + } + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::GetLoadFlags(nsLoadFlags* aLoadFlags) { + RecursiveMutexAutoLock lock(mMutex); + + *aLoadFlags = mLoadFlags; + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::SetLoadFlags(nsLoadFlags aLoadFlags) { + RecursiveMutexAutoLock lock(mMutex); + + mLoadFlags = aLoadFlags; + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::GetTRRMode(nsIRequest::TRRMode* aTRRMode) { + return GetTRRModeImpl(aTRRMode); +} + +NS_IMETHODIMP +nsInputStreamPump::SetTRRMode(nsIRequest::TRRMode aTRRMode) { + return SetTRRModeImpl(aTRRMode); +} + +NS_IMETHODIMP +nsInputStreamPump::GetLoadGroup(nsILoadGroup** aLoadGroup) { + RecursiveMutexAutoLock lock(mMutex); + + *aLoadGroup = do_AddRef(mLoadGroup).take(); + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::SetLoadGroup(nsILoadGroup* aLoadGroup) { + RecursiveMutexAutoLock lock(mMutex); + + mLoadGroup = aLoadGroup; + return NS_OK; +} + +//----------------------------------------------------------------------------- +// nsInputStreamPump::nsIInputStreamPump implementation +//----------------------------------------------------------------------------- + +NS_IMETHODIMP +nsInputStreamPump::Init(nsIInputStream* stream, uint32_t segsize, + uint32_t segcount, bool closeWhenDone, + nsISerialEventTarget* mainThreadTarget) { + // probably we can't be multithread-accessed yet + RecursiveMutexAutoLock lock(mMutex); + NS_ENSURE_TRUE(mState == STATE_IDLE, NS_ERROR_IN_PROGRESS); + + mStream = stream; + mSegSize = segsize; + mSegCount = segcount; + mCloseWhenDone = closeWhenDone; + mLabeledMainThreadTarget = mainThreadTarget; + if (mOffMainThread && mLabeledMainThreadTarget) { + MOZ_ASSERT( + false, + "Init stream pump off main thread with a main thread event target."); + return NS_ERROR_FAILURE; + } + + return NS_OK; +} + +NS_IMETHODIMP +nsInputStreamPump::AsyncRead(nsIStreamListener* listener) { + RecursiveMutexAutoLock lock(mMutex); + + // This ensures only one thread can interact with a pump at a time + NS_ENSURE_TRUE(mState == STATE_IDLE, NS_ERROR_IN_PROGRESS); + NS_ENSURE_ARG_POINTER(listener); + MOZ_ASSERT(NS_IsMainThread() || mOffMainThread, + "nsInputStreamPump should be read from the " + "main thread only."); + + nsresult rv = NS_MakeAsyncNonBlockingInputStream( + mStream.forget(), getter_AddRefs(mAsyncStream), mCloseWhenDone, mSegSize, + mSegCount); + if (NS_WARN_IF(NS_FAILED(rv))) { + return rv; + } + + MOZ_ASSERT(mAsyncStream); + + // mStreamOffset now holds the number of bytes currently read. + mStreamOffset = 0; + + // grab event queue (we must do this here by contract, since all notifications + // must go to the thread which called AsyncRead) + if (NS_IsMainThread() && mLabeledMainThreadTarget) { + mTargetThread = mLabeledMainThreadTarget; + } else { + mTargetThread = mozilla::GetCurrentSerialEventTarget(); + } + NS_ENSURE_STATE(mTargetThread); + + rv = EnsureWaiting(); + if (NS_FAILED(rv)) return rv; + + if (mLoadGroup) mLoadGroup->AddRequest(this, nullptr); + + mState = STATE_START; + mListener = listener; + return NS_OK; +} + +//----------------------------------------------------------------------------- +// nsInputStreamPump::nsIInputStreamCallback implementation +//----------------------------------------------------------------------------- + +NS_IMETHODIMP +nsInputStreamPump::OnInputStreamReady(nsIAsyncInputStream* stream) { + LOG(("nsInputStreamPump::OnInputStreamReady [this=%p]\n", this)); + + AUTO_PROFILER_LABEL("nsInputStreamPump::OnInputStreamReady", NETWORK); + + // this function has been called from a PLEvent, so we can safely call + // any listener or progress sink methods directly from here. + + for (;;) { + // There should only be one iteration of this loop happening at a time. + // To prevent AsyncWait() (called during callbacks or on other threads) + // from creating a parallel OnInputStreamReady(), we use: + // -- a mutex; and + // -- a boolean mProcessingCallbacks to detect parallel loops + // when exiting the mutex for callbacks. + RecursiveMutexAutoLock lock(mMutex); + + // Prevent parallel execution during callbacks, while out of mutex. + if (mProcessingCallbacks) { + MOZ_ASSERT(!mProcessingCallbacks); + break; + } + mProcessingCallbacks = true; + if (mSuspendCount || mState == STATE_IDLE || mState == STATE_DEAD) { + mWaitingForInputStreamReady = false; + mProcessingCallbacks = false; + break; + } + + uint32_t nextState; + switch (mState) { + case STATE_START: + nextState = OnStateStart(); + break; + case STATE_TRANSFER: + nextState = OnStateTransfer(); + break; + case STATE_STOP: + mRetargeting = false; + nextState = OnStateStop(); + break; + default: + nextState = 0; + MOZ_ASSERT_UNREACHABLE("Unknown enum value."); + return NS_ERROR_UNEXPECTED; + } + + bool stillTransferring = + (mState == STATE_TRANSFER && nextState == STATE_TRANSFER); + if (stillTransferring) { + NS_ASSERTION(NS_SUCCEEDED(mStatus), + "Should not have failed status for ongoing transfer"); + } else { + NS_ASSERTION(mState != nextState, + "Only OnStateTransfer can be called more than once."); + } + if (mRetargeting) { + NS_ASSERTION(mState != STATE_STOP, + "Retargeting should not happen during OnStateStop."); + } + + // Set mRetargeting so EnsureWaiting will be called. It ensures that + // OnStateStop is called on the main thread. + if (nextState == STATE_STOP && !NS_IsMainThread() && !mOffMainThread) { + mRetargeting = true; + } + + // Unset mProcessingCallbacks here (while we have lock) so our own call to + // EnsureWaiting isn't blocked by it. + mProcessingCallbacks = false; + + // We must break the loop if suspended during one of the previous + // operation. + if (mSuspendCount) { + mState = nextState; + mWaitingForInputStreamReady = false; + break; + } + + // Wait asynchronously if there is still data to transfer, or we're + // switching event delivery to another thread. + if (stillTransferring || mRetargeting) { + mState = nextState; + mWaitingForInputStreamReady = false; + nsresult rv = EnsureWaiting(); + if (NS_SUCCEEDED(rv)) break; + + // Failure to start asynchronous wait: stop transfer. + // Do not set mStatus if it was previously set to report a failure. + if (NS_SUCCEEDED(mStatus)) { + mStatus = rv; + } + nextState = STATE_STOP; + } + + mState = nextState; + } + return NS_OK; +} + +uint32_t nsInputStreamPump::OnStateStart() { + mMutex.AssertCurrentThreadIn(); + + AUTO_PROFILER_LABEL("nsInputStreamPump::OnStateStart", NETWORK); + + LOG((" OnStateStart [this=%p]\n", this)); + + nsresult rv; + + // need to check the reason why the stream is ready. this is required + // so our listener can check our status from OnStartRequest. + // XXX async streams should have a GetStatus method! + if (NS_SUCCEEDED(mStatus)) { + uint64_t avail; + rv = mAsyncStream->Available(&avail); + if (NS_FAILED(rv) && rv != NS_BASE_STREAM_CLOSED) mStatus = rv; + } + + { + nsCOMPtr<nsIStreamListener> listener = mListener; + // We're on the writing thread + AssertOnThread(); + + // Note: Must exit mutex for call to OnStartRequest to avoid + // deadlocks when calls to RetargetDeliveryTo for multiple + // nsInputStreamPumps are needed (e.g. nsHttpChannel). + RecursiveMutexAutoUnlock unlock(mMutex); + rv = listener->OnStartRequest(this); + } + + // an error returned from OnStartRequest should cause us to abort; however, + // we must not stomp on mStatus if already canceled. + if (NS_FAILED(rv) && NS_SUCCEEDED(mStatus)) mStatus = rv; + + return NS_SUCCEEDED(mStatus) ? STATE_TRANSFER : STATE_STOP; +} + +uint32_t nsInputStreamPump::OnStateTransfer() { + mMutex.AssertCurrentThreadIn(); + + AUTO_PROFILER_LABEL("nsInputStreamPump::OnStateTransfer", NETWORK); + + LOG((" OnStateTransfer [this=%p]\n", this)); + + // if canceled, go directly to STATE_STOP... + if (NS_FAILED(mStatus)) return STATE_STOP; + + nsresult rv = CreateBufferedStreamIfNeeded(); + if (NS_WARN_IF(NS_FAILED(rv))) { + return STATE_STOP; + } + + uint64_t avail; + rv = mAsyncStream->Available(&avail); + LOG((" Available returned [stream=%p rv=%" PRIx32 " avail=%" PRIu64 "]\n", + mAsyncStream.get(), static_cast<uint32_t>(rv), avail)); + + if (rv == NS_BASE_STREAM_CLOSED) { + rv = NS_OK; + avail = 0; + } else if (NS_SUCCEEDED(rv) && avail) { + // we used to limit avail to 16K - we were afraid some ODA handlers + // might assume they wouldn't get more than 16K at once + // we're removing that limit since it speeds up local file access. + // Now there's an implicit 64K limit of 4 16K segments + // NOTE: ok, so the story is as follows. OnDataAvailable impls + // are by contract supposed to consume exactly |avail| bytes. + // however, many do not... mailnews... stream converters... + // cough, cough. the input stream pump is fairly tolerant + // in this regard; however, if an ODA does not consume any + // data from the stream, then we could potentially end up in + // an infinite loop. we do our best here to try to catch + // such an error. (see bug 189672) + + // in most cases this QI will succeed (mAsyncStream is almost always + // a nsPipeInputStream, which implements nsITellableStream::Tell). + int64_t offsetBefore; + nsCOMPtr<nsITellableStream> tellable = do_QueryInterface(mAsyncStream); + if (tellable && NS_FAILED(tellable->Tell(&offsetBefore))) { + MOZ_ASSERT_UNREACHABLE("Tell failed on readable stream"); + offsetBefore = 0; + } + + uint32_t odaAvail = avail > UINT32_MAX ? UINT32_MAX : uint32_t(avail); + + LOG((" calling OnDataAvailable [offset=%" PRIu64 " count=%" PRIu64 + "(%u)]\n", + mStreamOffset, avail, odaAvail)); + + { + // We may be called on non-MainThread even if mOffMainThread is + // false, due to RetargetDeliveryTo(), so don't use AssertOnThread() + if (mTargetThread) { + MOZ_ASSERT(mTargetThread->IsOnCurrentThread()); + } else { + MOZ_ASSERT(NS_IsMainThread()); + } + + nsCOMPtr<nsIStreamListener> listener = mListener; + // Note: Must exit mutex for call to OnStartRequest to avoid + // deadlocks when calls to RetargetDeliveryTo for multiple + // nsInputStreamPumps are needed (e.g. nsHttpChannel). + RecursiveMutexAutoUnlock unlock(mMutex); + // We're on the writing thread for mListener and mAsyncStream. + // mStreamOffset is only touched in OnStateTransfer, and AsyncRead + // shouldn't be called during OnDataAvailable() + + MOZ_PUSH_IGNORE_THREAD_SAFETY + rv = listener->OnDataAvailable(this, mAsyncStream, mStreamOffset, + odaAvail); + MOZ_POP_THREAD_SAFETY + } + + // don't enter this code if ODA failed or called Cancel + if (NS_SUCCEEDED(rv) && NS_SUCCEEDED(mStatus)) { + // test to see if this ODA failed to consume data + if (tellable) { + // NOTE: if Tell fails, which can happen if the stream is + // now closed, then we assume that everything was read. + int64_t offsetAfter; + if (NS_FAILED(tellable->Tell(&offsetAfter))) { + offsetAfter = offsetBefore + odaAvail; + } + if (offsetAfter > offsetBefore) { + mStreamOffset += (offsetAfter - offsetBefore); + } else if (mSuspendCount == 0) { + // + // possible infinite loop if we continue pumping data! + // + // NOTE: although not allowed by nsIStreamListener, we + // will allow the ODA impl to Suspend the pump. IMAP + // does this :-( + // + NS_ERROR("OnDataAvailable implementation consumed no data"); + mStatus = NS_ERROR_UNEXPECTED; + } + } else { + mStreamOffset += odaAvail; // assume ODA behaved well + } + } + } + + // an error returned from Available or OnDataAvailable should cause us to + // abort; however, we must not stop on mStatus if already canceled. + + if (NS_SUCCEEDED(mStatus)) { + if (NS_FAILED(rv)) { + mStatus = rv; + } else if (avail) { + // if stream is now closed, advance to STATE_STOP right away. + // Available may return 0 bytes available at the moment; that + // would not mean that we are done. + // XXX async streams should have a GetStatus method! + rv = mAsyncStream->Available(&avail); + if (NS_SUCCEEDED(rv)) return STATE_TRANSFER; + if (rv != NS_BASE_STREAM_CLOSED) mStatus = rv; + } + } + return STATE_STOP; +} + +nsresult nsInputStreamPump::CallOnStateStop() { + RecursiveMutexAutoLock lock(mMutex); + + MOZ_ASSERT(NS_IsMainThread(), + "CallOnStateStop should only be called on the main thread."); + + mState = OnStateStop(); + return NS_OK; +} + +uint32_t nsInputStreamPump::OnStateStop() { + mMutex.AssertCurrentThreadIn(); + + if (!NS_IsMainThread() && !mOffMainThread) { + // This method can be called on a different thread if nsInputStreamPump + // is used off the main-thread. + nsresult rv = mLabeledMainThreadTarget->Dispatch( + mozilla::NewRunnableMethod("nsInputStreamPump::CallOnStateStop", this, + &nsInputStreamPump::CallOnStateStop)); + NS_ENSURE_SUCCESS(rv, STATE_DEAD); + return STATE_DEAD; + } + + AUTO_PROFILER_LABEL("nsInputStreamPump::OnStateStop", NETWORK); + + LOG((" OnStateStop [this=%p status=%" PRIx32 "]\n", this, + static_cast<uint32_t>(mStatus))); + + // if an error occurred, we must be sure to pass the error onto the async + // stream. in some cases, this is redundant, but since close is idempotent, + // this is OK. otherwise, be sure to honor the "close-when-done" option. + + if (!mAsyncStream || !mListener) { + MOZ_ASSERT(mAsyncStream, "null mAsyncStream: OnStateStop called twice?"); + MOZ_ASSERT(mListener, "null mListener: OnStateStop called twice?"); + return STATE_DEAD; + } + + if (NS_FAILED(mStatus)) { + mAsyncStream->CloseWithStatus(mStatus); + } else if (mCloseWhenDone) { + mAsyncStream->Close(); + } + + mAsyncStream = nullptr; + mIsPending = false; + { + // We're on the writing thread. + // We believe that mStatus can't be changed on us here. + AssertOnThread(); + + nsCOMPtr<nsIStreamListener> listener = mListener; + nsresult status = mStatus; + // Note: Must exit mutex for call to OnStartRequest to avoid + // deadlocks when calls to RetargetDeliveryTo for multiple + // nsInputStreamPumps are needed (e.g. nsHttpChannel). + RecursiveMutexAutoUnlock unlock(mMutex); + + listener->OnStopRequest(this, status); + } + mTargetThread = nullptr; + mListener = nullptr; + + if (mLoadGroup) mLoadGroup->RemoveRequest(this, nullptr, mStatus); + + return STATE_DEAD; +} + +nsresult nsInputStreamPump::CreateBufferedStreamIfNeeded() { + if (mAsyncStreamIsBuffered) { + return NS_OK; + } + + // ReadSegments is not available for any nsIAsyncInputStream. In order to use + // it, we wrap a nsIBufferedInputStream around it, if needed. + + if (NS_InputStreamIsBuffered(mAsyncStream)) { + mAsyncStreamIsBuffered = true; + return NS_OK; + } + + nsCOMPtr<nsIInputStream> stream; + nsresult rv = NS_NewBufferedInputStream(getter_AddRefs(stream), + mAsyncStream.forget(), 4096); + NS_ENSURE_SUCCESS(rv, rv); + + // A buffered inputStream must implement nsIAsyncInputStream. + mAsyncStream = do_QueryInterface(stream); + MOZ_DIAGNOSTIC_ASSERT(mAsyncStream); + mAsyncStreamIsBuffered = true; + + return NS_OK; +} + +//----------------------------------------------------------------------------- +// nsIThreadRetargetableRequest +//----------------------------------------------------------------------------- + +NS_IMETHODIMP +nsInputStreamPump::RetargetDeliveryTo(nsISerialEventTarget* aNewTarget) { + RecursiveMutexAutoLock lock(mMutex); + + NS_ENSURE_ARG(aNewTarget); + NS_ENSURE_TRUE(mState == STATE_START || mState == STATE_TRANSFER, + NS_ERROR_UNEXPECTED); + + // If canceled, do not retarget. Return with canceled status. + if (NS_FAILED(mStatus)) { + return mStatus; + } + + if (aNewTarget == mTargetThread) { + NS_WARNING("Retargeting delivery to same thread"); + return NS_OK; + } + + if (mOffMainThread) { + // Don't support retargeting if this pump is already used off the main + // thread. + return NS_ERROR_FAILURE; + } + + // Ensure that |mListener| and any subsequent listeners can be retargeted + // to another thread. + nsresult rv = NS_OK; + nsCOMPtr<nsIThreadRetargetableStreamListener> retargetableListener = + do_QueryInterface(mListener, &rv); + if (NS_SUCCEEDED(rv) && retargetableListener) { + rv = retargetableListener->CheckListenerChain(); + if (NS_SUCCEEDED(rv)) { + mTargetThread = aNewTarget; + mRetargeting = true; + } + } + LOG( + ("nsInputStreamPump::RetargetDeliveryTo [this=%p aNewTarget=%p] " + "%s listener [%p] rv[%" PRIx32 "]", + this, aNewTarget, (mTargetThread == aNewTarget ? "success" : "failure"), + (nsIStreamListener*)mListener, static_cast<uint32_t>(rv))); + return rv; +} + +NS_IMETHODIMP +nsInputStreamPump::GetDeliveryTarget(nsISerialEventTarget** aNewTarget) { + RecursiveMutexAutoLock lock(mMutex); + + nsCOMPtr<nsISerialEventTarget> target = mTargetThread; + target.forget(aNewTarget); + return NS_OK; +} |