gecko-dev/ipc/glue/IPCStreamDestination.cpp

/* -*- 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"

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 nsISearchableInputStream
  , public nsICloneableInputStream
  , public nsIBufferedInputStream
{
public:
  NS_DECL_THREADSAFE_ISUPPORTS

  DelayedStartInputStream(IPCStreamDestination* aDestination,
                          already_AddRefed<nsIAsyncInputStream>&& aStream)
    : mDestination(aDestination)
    , mStream(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
  {
    MaybeStartReading();
    return mStream->AsyncWait(aCallback, 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;
  }

  void
  MaybeStartReading();

  void
  MaybeCloseDestination();

private:
  ~DelayedStartInputStream() = default;

  IPCStreamDestination* mDestination;
  nsCOMPtr<nsIAsyncInputStream> mStream;

  // 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);
  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(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()
{
}

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

  if (aLength != -1) {
    nsCOMPtr<nsIInputStream> finalStream;
    finalStream = new InputStreamLengthWrapper(mReader.forget(), aLength);
    mReader = do_QueryInterface(finalStream);
    MOZ_ASSERT(mReader);
  }
}

already_AddRefed<nsIInputStream>
IPCStreamDestination::TakeReader()
{
  MOZ_ASSERT(mReader);
  MOZ_ASSERT(!mDelayedStartInputStream);

  if (mDelayedStart) {
    mDelayedStartInputStream =
      new DelayedStartInputStream(this, mReader.forget());
    RefPtr<nsIAsyncInputStream> inputStream = mDelayedStartInputStream;
    return inputStream.forget();
  }

  return mReader.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