Source code

Revision control

Other Tools

/* -*- 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 "IPCStreamDestination.h"
#include "mozilla/InputStreamLengthWrapper.h"
#include "mozilla/Mutex.h"
#include "nsIAsyncInputStream.h"
#include "nsIAsyncOutputStream.h"
#include "nsIBufferedStreams.h"
#include "nsICloneableInputStream.h"
#include "nsIPipe.h"
#include "nsThreadUtils.h"
#include "mozilla/webrender/WebRenderTypes.h"
namespace mozilla {
namespace ipc {
// ----------------------------------------------------------------------------
// IPCStreamDestination::DelayedStartInputStream
//
// When AutoIPCStream is used with delayedStart, we need to ask for data at the
// first real use of the nsIInputStream. In order to do so, we wrap the
// nsIInputStream, created by the nsIPipe, with this wrapper.
class IPCStreamDestination::DelayedStartInputStream final
: public nsIAsyncInputStream,
public nsIInputStreamCallback,
public nsISearchableInputStream,
public nsICloneableInputStream,
public nsIBufferedInputStream {
public:
NS_DECL_THREADSAFE_ISUPPORTS
DelayedStartInputStream(IPCStreamDestination* aDestination,
nsCOMPtr<nsIAsyncInputStream>&& aStream)
: mDestination(aDestination),
mStream(std::move(aStream)),
mMutex("IPCStreamDestination::DelayedStartInputStream::mMutex") {
MOZ_ASSERT(mDestination);
MOZ_ASSERT(mStream);
}
void DestinationShutdown() {
MutexAutoLock lock(mMutex);
mDestination = nullptr;
}
// nsIInputStream interface
NS_IMETHOD
Close() override {
MaybeCloseDestination();
return mStream->Close();
}
NS_IMETHOD
Available(uint64_t* aLength) override {
MaybeStartReading();
return mStream->Available(aLength);
}
NS_IMETHOD
Read(char* aBuffer, uint32_t aCount, uint32_t* aReadCount) override {
MaybeStartReading();
return mStream->Read(aBuffer, aCount, aReadCount);
}
NS_IMETHOD
ReadSegments(nsWriteSegmentFun aWriter, void* aClosure, uint32_t aCount,
uint32_t* aResult) override {
MaybeStartReading();
return mStream->ReadSegments(aWriter, aClosure, aCount, aResult);
}
NS_IMETHOD
IsNonBlocking(bool* aNonBlocking) override {
MaybeStartReading();
return mStream->IsNonBlocking(aNonBlocking);
}
// nsIAsyncInputStream interface
NS_IMETHOD
CloseWithStatus(nsresult aReason) override {
MaybeCloseDestination();
return mStream->CloseWithStatus(aReason);
}
NS_IMETHOD
AsyncWait(nsIInputStreamCallback* aCallback, uint32_t aFlags,
uint32_t aRequestedCount, nsIEventTarget* aTarget) override {
{
MutexAutoLock lock(mMutex);
if (mAsyncWaitCallback && aCallback) {
return NS_ERROR_FAILURE;
}
mAsyncWaitCallback = aCallback;
MaybeStartReading(lock);
}
nsCOMPtr<nsIInputStreamCallback> callback = aCallback ? this : nullptr;
return mStream->AsyncWait(callback, aFlags, aRequestedCount, aTarget);
}
NS_IMETHOD
Search(const char* aForString, bool aIgnoreCase, bool* aFound,
uint32_t* aOffsetSearchedTo) override {
MaybeStartReading();
nsCOMPtr<nsISearchableInputStream> searchable = do_QueryInterface(mStream);
MOZ_ASSERT(searchable);
return searchable->Search(aForString, aIgnoreCase, aFound,
aOffsetSearchedTo);
}
// nsICloneableInputStream interface
NS_IMETHOD
GetCloneable(bool* aCloneable) override {
MaybeStartReading();
nsCOMPtr<nsICloneableInputStream> cloneable = do_QueryInterface(mStream);
MOZ_ASSERT(cloneable);
return cloneable->GetCloneable(aCloneable);
}
NS_IMETHOD
Clone(nsIInputStream** aResult) override {
MaybeStartReading();
nsCOMPtr<nsICloneableInputStream> cloneable = do_QueryInterface(mStream);
MOZ_ASSERT(cloneable);
return cloneable->Clone(aResult);
}
// nsIBufferedInputStream
NS_IMETHOD
Init(nsIInputStream* aStream, uint32_t aBufferSize) override {
MaybeStartReading();
nsCOMPtr<nsIBufferedInputStream> stream = do_QueryInterface(mStream);
MOZ_ASSERT(stream);
return stream->Init(aStream, aBufferSize);
}
NS_IMETHODIMP
GetData(nsIInputStream** aResult) override {
return NS_ERROR_NOT_IMPLEMENTED;
}
// nsIInputStreamCallback
NS_IMETHOD
OnInputStreamReady(nsIAsyncInputStream* aStream) override {
nsCOMPtr<nsIInputStreamCallback> callback;
{
MutexAutoLock lock(mMutex);
// We have been canceled in the meanwhile.
if (!mAsyncWaitCallback) {
return NS_OK;
}
callback.swap(mAsyncWaitCallback);
}
callback->OnInputStreamReady(this);
return NS_OK;
}
void MaybeStartReading();
void MaybeStartReading(const MutexAutoLock& aProofOfLook);
void MaybeCloseDestination();
private:
~DelayedStartInputStream() = default;
IPCStreamDestination* mDestination;
nsCOMPtr<nsIAsyncInputStream> mStream;
nsCOMPtr<nsIInputStreamCallback> mAsyncWaitCallback;
// This protects mDestination: any method can be called by any thread.
Mutex mMutex;
class HelperRunnable;
};
class IPCStreamDestination::DelayedStartInputStream::HelperRunnable final
: public Runnable {
public:
enum Op {
eStartReading,
eCloseDestination,
};
HelperRunnable(
IPCStreamDestination::DelayedStartInputStream* aDelayedStartInputStream,
Op aOp)
: Runnable(
"ipc::IPCStreamDestination::DelayedStartInputStream::"
"HelperRunnable"),
mDelayedStartInputStream(aDelayedStartInputStream),
mOp(aOp) {
MOZ_ASSERT(aDelayedStartInputStream);
}
NS_IMETHOD
Run() override {
switch (mOp) {
case eStartReading:
mDelayedStartInputStream->MaybeStartReading();
break;
case eCloseDestination:
mDelayedStartInputStream->MaybeCloseDestination();
break;
}
return NS_OK;
}
private:
RefPtr<IPCStreamDestination::DelayedStartInputStream>
mDelayedStartInputStream;
Op mOp;
};
void IPCStreamDestination::DelayedStartInputStream::MaybeStartReading() {
MutexAutoLock lock(mMutex);
MaybeStartReading(lock);
}
void IPCStreamDestination::DelayedStartInputStream::MaybeStartReading(
const MutexAutoLock& aProofOfLook) {
if (!mDestination) {
return;
}
if (mDestination->IsOnOwningThread()) {
mDestination->StartReading();
mDestination = nullptr;
return;
}
RefPtr<Runnable> runnable =
new HelperRunnable(this, HelperRunnable::eStartReading);
mDestination->DispatchRunnable(runnable.forget());
}
void IPCStreamDestination::DelayedStartInputStream::MaybeCloseDestination() {
MutexAutoLock lock(mMutex);
if (!mDestination) {
return;
}
if (mDestination->IsOnOwningThread()) {
mDestination->RequestClose(NS_ERROR_ABORT);
mDestination = nullptr;
return;
}
RefPtr<Runnable> runnable =
new HelperRunnable(this, HelperRunnable::eCloseDestination);
mDestination->DispatchRunnable(runnable.forget());
}
NS_IMPL_ADDREF(IPCStreamDestination::DelayedStartInputStream);
NS_IMPL_RELEASE(IPCStreamDestination::DelayedStartInputStream);
NS_INTERFACE_MAP_BEGIN(IPCStreamDestination::DelayedStartInputStream)
NS_INTERFACE_MAP_ENTRY(nsIAsyncInputStream)
NS_INTERFACE_MAP_ENTRY(nsIInputStreamCallback)
NS_INTERFACE_MAP_ENTRY(nsISearchableInputStream)
NS_INTERFACE_MAP_ENTRY(nsICloneableInputStream)
NS_INTERFACE_MAP_ENTRY(nsIBufferedInputStream)
NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsIInputStream, nsIAsyncInputStream)
NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIAsyncInputStream)
NS_INTERFACE_MAP_END
// ----------------------------------------------------------------------------
// IPCStreamDestination
IPCStreamDestination::IPCStreamDestination()
: mOwningThread(NS_GetCurrentThread()),
mDelayedStart(false)
#ifdef MOZ_DEBUG
,
mLengthSet(false)
#endif
{
}
IPCStreamDestination::~IPCStreamDestination() = default;
nsresult IPCStreamDestination::Initialize() {
MOZ_ASSERT(!mReader);
MOZ_ASSERT(!mWriter);
// use async versions for both reader and writer even though we are
// opening the writer as an infinite stream. We want to be able to
// use CloseWithStatus() to communicate errors through the pipe.
// Use an "infinite" pipe because we cannot apply back-pressure through
// the async IPC layer at the moment. Blocking the IPC worker thread
// is not desirable, either.
nsresult rv = NS_NewPipe2(getter_AddRefs(mReader), getter_AddRefs(mWriter),
true, true, // non-blocking
0, // segment size
UINT32_MAX); // "infinite" pipe
if (NS_WARN_IF(NS_FAILED(rv))) {
return rv;
}
return NS_OK;
}
void IPCStreamDestination::SetDelayedStart(bool aDelayedStart) {
mDelayedStart = aDelayedStart;
}
void IPCStreamDestination::SetLength(int64_t aLength) {
MOZ_ASSERT(mReader);
MOZ_ASSERT(!mLengthSet);
#ifdef DEBUG
mLengthSet = true;
#endif
mLength = aLength;
}
already_AddRefed<nsIInputStream> IPCStreamDestination::TakeReader() {
MOZ_ASSERT(mReader);
MOZ_ASSERT(!mDelayedStartInputStream);
nsCOMPtr<nsIAsyncInputStream> reader{mReader.forget()};
if (mDelayedStart) {
mDelayedStartInputStream =
new DelayedStartInputStream(this, std::move(reader));
reader = mDelayedStartInputStream;
MOZ_ASSERT(reader);
}
if (mLength != -1) {
MOZ_ASSERT(mLengthSet);
nsCOMPtr<nsIInputStream> finalStream =
new InputStreamLengthWrapper(reader.forget(), mLength);
reader = do_QueryInterface(finalStream);
MOZ_ASSERT(reader);
}
return reader.forget();
}
bool IPCStreamDestination::IsOnOwningThread() const {
return mOwningThread == NS_GetCurrentThread();
}
void IPCStreamDestination::DispatchRunnable(
already_AddRefed<nsIRunnable>&& aRunnable) {
nsCOMPtr<nsIRunnable> runnable = aRunnable;
mOwningThread->Dispatch(runnable.forget(), NS_DISPATCH_NORMAL);
}
void IPCStreamDestination::ActorDestroyed() {
MOZ_ASSERT(mWriter);
// If we were gracefully closed we should have gotten RecvClose(). In
// that case, the writer will already be closed and this will have no
// effect. This just aborts the writer in the case where the child process
// crashes.
mWriter->CloseWithStatus(NS_ERROR_ABORT);
if (mDelayedStartInputStream) {
mDelayedStartInputStream->DestinationShutdown();
mDelayedStartInputStream = nullptr;
}
}
void IPCStreamDestination::BufferReceived(const wr::ByteBuffer& aBuffer) {
MOZ_ASSERT(mWriter);
uint32_t numWritten = 0;
// This should only fail if we hit an OOM condition.
nsresult rv = mWriter->Write(reinterpret_cast<char*>(aBuffer.mData),
aBuffer.mLength, &numWritten);
if (NS_WARN_IF(NS_FAILED(rv))) {
RequestClose(rv);
}
}
void IPCStreamDestination::CloseReceived(nsresult aRv) {
MOZ_ASSERT(mWriter);
mWriter->CloseWithStatus(aRv);
TerminateDestination();
}
} // namespace ipc
} // namespace mozilla