Adding upstream version 124.0.1.upstream/124.0.1

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 620 insertions, 0 deletions

diff --git a/dom/cache/ReadStream.cpp b/dom/cache/ReadStream.cpp
new file mode 100644
index 0000000000..366dd5ab13
--- /dev/null
+++ b/dom/cache/ReadStream.cpp
@@ -0,0 +1,620 @@
+/* -*- 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 "mozilla/dom/cache/ReadStream.h"
+
+#include "mozilla/Unused.h"
+#include "mozilla/dom/cache/CacheStreamControlChild.h"
+#include "mozilla/dom/cache/CacheStreamControlParent.h"
+#include "mozilla/dom/cache/CacheTypes.h"
+#include "mozilla/ipc/IPCStreamUtils.h"
+#include "mozilla/SnappyUncompressInputStream.h"
+#include "nsIAsyncInputStream.h"
+#include "nsIThread.h"
+#include "nsStringStream.h"
+#include "nsTArray.h"
+
+namespace mozilla::dom::cache {
+
+using mozilla::Unused;
+
+// ----------------------------------------------------------------------------
+
+// The inner stream class.  This is where all of the real work is done.  As
+// an invariant Inner::Close() must be called before ~Inner().  This is
+// guaranteed by our outer ReadStream class.
+class ReadStream::Inner final : public ReadStream::Controllable {
+ public:
+  Inner(StreamControl* aControl, const nsID& aId, nsIInputStream* aStream);
+
+  void Serialize(Maybe<CacheReadStream>* aReadStreamOut, ErrorResult& aRv);
+
+  void Serialize(CacheReadStream* aReadStreamOut, ErrorResult& aRv);
+
+  // ReadStream::Controllable methods
+  virtual void CloseStream() override;
+
+  virtual void CloseStreamWithoutReporting() override;
+
+  virtual bool HasEverBeenRead() const override;
+
+  // Simulate nsIInputStream methods, but we don't actually inherit from it
+  nsresult Close();
+
+  nsresult Available(uint64_t* aNumAvailableOut);
+
+  nsresult StreamStatus();
+
+  nsresult Read(char* aBuf, uint32_t aCount, uint32_t* aNumReadOut);
+
+  nsresult ReadSegments(nsWriteSegmentFun aWriter, void* aClosure,
+                        uint32_t aCount, uint32_t* aNumReadOut);
+
+  nsresult IsNonBlocking(bool* aNonBlockingOut);
+
+  NS_DECL_OWNINGTHREAD;
+
+  ~Inner();
+
+ private:
+  class NoteClosedRunnable;
+  class ForgetRunnable;
+
+  void NoteClosed();
+
+  void Forget();
+
+  void NoteClosedOnOwningThread();
+
+  void ForgetOnOwningThread();
+
+  nsIInputStream* EnsureStream();
+
+  void AsyncOpenStreamOnOwningThread();
+
+  void MaybeAbortAsyncOpenStream();
+
+  void OpenStreamFailed();
+
+  inline SafeRefPtr<Inner> SafeRefPtrFromThis() {
+    return Controllable::SafeRefPtrFromThis().downcast<Inner>();
+  }
+
+  // Weak ref to the stream control actor.  The actor will always call either
+  // CloseStream() or CloseStreamWithoutReporting() before it's destroyed.  The
+  // weak ref is cleared in the resulting NoteClosedOnOwningThread() or
+  // ForgetOnOwningThread() method call.
+  StreamControl* mControl;
+
+  const nsID mId;
+  nsCOMPtr<nsISerialEventTarget> mOwningEventTarget;
+
+  enum State { Open, Closed, NumStates };
+  Atomic<State> mState;
+  Atomic<bool> mHasEverBeenRead;
+  bool mAsyncOpenStarted;
+
+  // The wrapped stream objects may not be threadsafe.  We need to be able
+  // to close a stream on our owning thread while an IO thread is simultaneously
+  // reading the same stream.  Therefore, protect all access to these stream
+  // objects with a mutex.
+  Mutex mMutex MOZ_UNANNOTATED;
+  CondVar mCondVar;
+  nsCOMPtr<nsIInputStream> mStream;
+  nsCOMPtr<nsIInputStream> mSnappyStream;
+};
+
+// ----------------------------------------------------------------------------
+
+// Runnable to notify actors that the ReadStream has closed.  This must
+// be done on the thread associated with the PBackground actor.  Must be
+// cancelable to execute on Worker threads (which can occur when the
+// ReadStream is constructed on a child process Worker thread).
+class ReadStream::Inner::NoteClosedRunnable final : public CancelableRunnable {
+ public:
+  explicit NoteClosedRunnable(SafeRefPtr<ReadStream::Inner> aStream)
+      : CancelableRunnable("dom::cache::ReadStream::Inner::NoteClosedRunnable"),
+        mStream(std::move(aStream)) {}
+
+  NS_IMETHOD Run() override {
+    mStream->NoteClosedOnOwningThread();
+    return NS_OK;
+  }
+
+  // Note, we must proceed with the Run() method since our actor will not
+  // clean itself up until we note that the stream is closed.
+  nsresult Cancel() override {
+    Run();
+    return NS_OK;
+  }
+
+ private:
+  ~NoteClosedRunnable() = default;
+
+  const SafeRefPtr<ReadStream::Inner> mStream;
+};
+
+// ----------------------------------------------------------------------------
+
+// Runnable to clear actors without reporting that the ReadStream has
+// closed.  Since this can trigger actor destruction, we need to do
+// it on the thread associated with the PBackground actor.  Must be
+// cancelable to execute on Worker threads (which can occur when the
+// ReadStream is constructed on a child process Worker thread).
+class ReadStream::Inner::ForgetRunnable final : public CancelableRunnable {
+ public:
+  explicit ForgetRunnable(SafeRefPtr<ReadStream::Inner> aStream)
+      : CancelableRunnable("dom::cache::ReadStream::Inner::ForgetRunnable"),
+        mStream(std::move(aStream)) {}
+
+  NS_IMETHOD Run() override {
+    mStream->ForgetOnOwningThread();
+    return NS_OK;
+  }
+
+  // Note, we must proceed with the Run() method so that we properly
+  // call RemoveListener on the actor.
+  nsresult Cancel() override {
+    Run();
+    return NS_OK;
+  }
+
+ private:
+  ~ForgetRunnable() = default;
+
+  const SafeRefPtr<ReadStream::Inner> mStream;
+};
+
+// ----------------------------------------------------------------------------
+
+ReadStream::Inner::Inner(StreamControl* aControl, const nsID& aId,
+                         nsIInputStream* aStream)
+    : mControl(aControl),
+      mId(aId),
+      mOwningEventTarget(GetCurrentSerialEventTarget()),
+      mState(Open),
+      mHasEverBeenRead(false),
+      mAsyncOpenStarted(false),
+      mMutex("dom::cache::ReadStream"),
+      mCondVar(mMutex, "dom::cache::ReadStream"),
+      mStream(aStream),
+      mSnappyStream(aStream ? new SnappyUncompressInputStream(aStream)
+                            : nullptr) {
+  MOZ_DIAGNOSTIC_ASSERT(mControl);
+  mControl->AddReadStream(SafeRefPtrFromThis());
+}
+
+void ReadStream::Inner::Serialize(Maybe<CacheReadStream>* aReadStreamOut,
+                                  ErrorResult& aRv) {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+  MOZ_DIAGNOSTIC_ASSERT(aReadStreamOut);
+  aReadStreamOut->emplace(CacheReadStream());
+  Serialize(&aReadStreamOut->ref(), aRv);
+}
+
+void ReadStream::Inner::Serialize(CacheReadStream* aReadStreamOut,
+                                  ErrorResult& aRv) {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+  MOZ_DIAGNOSTIC_ASSERT(aReadStreamOut);
+
+  if (mState != Open) {
+    aRv.ThrowTypeError(
+        "Response body is a cache file stream that has already been closed.");
+    return;
+  }
+
+  MOZ_DIAGNOSTIC_ASSERT(mControl);
+
+  aReadStreamOut->id() = mId;
+  mControl->SerializeControl(aReadStreamOut);
+
+  {
+    MutexAutoLock lock(mMutex);
+    mControl->SerializeStream(aReadStreamOut, mStream);
+  }
+
+  MOZ_DIAGNOSTIC_ASSERT(aReadStreamOut->stream().isNothing() ||
+                        aReadStreamOut->stream().ref().stream().type() !=
+                            mozilla::ipc::InputStreamParams::T__None);
+
+  // We're passing ownership across the IPC barrier with the control, so
+  // do not signal that the stream is closed here.
+  Forget();
+}
+
+void ReadStream::Inner::CloseStream() {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+  MOZ_ALWAYS_SUCCEEDS(Close());
+}
+
+void ReadStream::Inner::CloseStreamWithoutReporting() {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+  Forget();
+}
+
+bool ReadStream::Inner::HasEverBeenRead() const {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+  return mHasEverBeenRead;
+}
+
+nsresult ReadStream::Inner::Close() {
+  // stream ops can happen on any thread
+  nsresult rv = NS_OK;
+  {
+    MutexAutoLock lock(mMutex);
+    if (mSnappyStream) {
+      rv = mSnappyStream->Close();
+    }
+  }
+  NoteClosed();
+  return rv;
+}
+
+nsresult ReadStream::Inner::Available(uint64_t* aNumAvailableOut) {
+  // stream ops can happen on any thread
+  nsresult rv = NS_OK;
+  {
+    MutexAutoLock lock(mMutex);
+    rv = EnsureStream()->Available(aNumAvailableOut);
+  }
+
+  if (NS_FAILED(rv)) {
+    Close();
+  }
+
+  return rv;
+}
+
+nsresult ReadStream::Inner::StreamStatus() {
+  // stream ops can happen on any thread
+  nsresult rv = NS_OK;
+  {
+    MutexAutoLock lock(mMutex);
+    rv = EnsureStream()->StreamStatus();
+  }
+
+  if (NS_FAILED(rv)) {
+    Close();
+  }
+
+  return rv;
+}
+
+nsresult ReadStream::Inner::Read(char* aBuf, uint32_t aCount,
+                                 uint32_t* aNumReadOut) {
+  // stream ops can happen on any thread
+  MOZ_DIAGNOSTIC_ASSERT(aNumReadOut);
+
+  nsresult rv = NS_OK;
+  {
+    MutexAutoLock lock(mMutex);
+    rv = EnsureStream()->Read(aBuf, aCount, aNumReadOut);
+  }
+
+  if ((NS_FAILED(rv) && rv != NS_BASE_STREAM_WOULD_BLOCK) ||
+      *aNumReadOut == 0) {
+    Close();
+  }
+
+  mHasEverBeenRead = true;
+
+  return rv;
+}
+
+nsresult ReadStream::Inner::ReadSegments(nsWriteSegmentFun aWriter,
+                                         void* aClosure, uint32_t aCount,
+                                         uint32_t* aNumReadOut) {
+  // stream ops can happen on any thread
+  MOZ_DIAGNOSTIC_ASSERT(aNumReadOut);
+
+  if (aCount) {
+    mHasEverBeenRead = true;
+  }
+
+  nsresult rv = NS_OK;
+  {
+    MutexAutoLock lock(mMutex);
+    rv = EnsureStream()->ReadSegments(aWriter, aClosure, aCount, aNumReadOut);
+  }
+
+  if ((NS_FAILED(rv) && rv != NS_BASE_STREAM_WOULD_BLOCK &&
+       rv != NS_ERROR_NOT_IMPLEMENTED) ||
+      *aNumReadOut == 0) {
+    Close();
+  }
+
+  // Verify bytes were actually read before marking as being ever read.  For
+  // example, code can test if the stream supports ReadSegments() by calling
+  // this method with a dummy callback which doesn't read anything.  We don't
+  // want to trigger on that.
+  if (*aNumReadOut) {
+    mHasEverBeenRead = true;
+  }
+
+  return rv;
+}
+
+nsresult ReadStream::Inner::IsNonBlocking(bool* aNonBlockingOut) {
+  // stream ops can happen on any thread
+  MutexAutoLock lock(mMutex);
+  if (mSnappyStream) {
+    return mSnappyStream->IsNonBlocking(aNonBlockingOut);
+  }
+  *aNonBlockingOut = false;
+  return NS_OK;
+}
+
+ReadStream::Inner::~Inner() {
+  // Any thread
+  MOZ_DIAGNOSTIC_ASSERT(mState == Closed);
+  MOZ_DIAGNOSTIC_ASSERT(!mControl);
+}
+
+void ReadStream::Inner::NoteClosed() {
+  // Any thread
+  if (mState == Closed) {
+    return;
+  }
+
+  if (mOwningEventTarget->IsOnCurrentThread()) {
+    NoteClosedOnOwningThread();
+    return;
+  }
+
+  nsCOMPtr<nsIRunnable> runnable = new NoteClosedRunnable(SafeRefPtrFromThis());
+  MOZ_ALWAYS_SUCCEEDS(mOwningEventTarget->Dispatch(runnable.forget(),
+                                                   nsIThread::DISPATCH_NORMAL));
+}
+
+void ReadStream::Inner::Forget() {
+  // Any thread
+  if (mState == Closed) {
+    return;
+  }
+
+  if (mOwningEventTarget->IsOnCurrentThread()) {
+    ForgetOnOwningThread();
+    return;
+  }
+
+  nsCOMPtr<nsIRunnable> runnable = new ForgetRunnable(SafeRefPtrFromThis());
+  MOZ_ALWAYS_SUCCEEDS(mOwningEventTarget->Dispatch(runnable.forget(),
+                                                   nsIThread::DISPATCH_NORMAL));
+}
+
+void ReadStream::Inner::NoteClosedOnOwningThread() {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+
+  // Mark closed and do nothing if we were already closed
+  if (!mState.compareExchange(Open, Closed)) {
+    return;
+  }
+
+  MaybeAbortAsyncOpenStream();
+
+  MOZ_DIAGNOSTIC_ASSERT(mControl);
+  mControl->NoteClosed(SafeRefPtrFromThis(), mId);
+  mControl = nullptr;
+}
+
+void ReadStream::Inner::ForgetOnOwningThread() {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+
+  // Mark closed and do nothing if we were already closed
+  if (!mState.compareExchange(Open, Closed)) {
+    return;
+  }
+
+  MaybeAbortAsyncOpenStream();
+
+  MOZ_DIAGNOSTIC_ASSERT(mControl);
+  mControl->ForgetReadStream(SafeRefPtrFromThis());
+  mControl = nullptr;
+}
+
+nsIInputStream* ReadStream::Inner::EnsureStream() {
+  mMutex.AssertCurrentThreadOwns();
+
+  // We need to block the current thread while we open the stream.  We
+  // cannot do this safely from the main owning thread since it would
+  // trigger deadlock.  This should be ok, though, since a blocking
+  // stream like this should never be read on the owning thread anyway.
+  if (mOwningEventTarget->IsOnCurrentThread()) {
+    MOZ_CRASH("Blocking read on the js/ipc owning thread!");
+  }
+
+  if (mSnappyStream) {
+    return mSnappyStream;
+  }
+
+  nsCOMPtr<nsIRunnable> r = NewCancelableRunnableMethod(
+      "ReadStream::Inner::AsyncOpenStreamOnOwningThread", this,
+      &ReadStream::Inner::AsyncOpenStreamOnOwningThread);
+  nsresult rv =
+      mOwningEventTarget->Dispatch(r.forget(), nsIThread::DISPATCH_NORMAL);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    OpenStreamFailed();
+    return mSnappyStream;
+  }
+
+  mCondVar.Wait();
+  MOZ_DIAGNOSTIC_ASSERT(mSnappyStream);
+
+  return mSnappyStream;
+}
+
+void ReadStream::Inner::AsyncOpenStreamOnOwningThread() {
+  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
+
+  if (mSnappyStream) {
+    // Different threads might request opening the stream at the same time. If
+    // the earlier request succeeded, then use the result.
+    mCondVar.NotifyAll();
+    return;
+  }
+
+  if (!mControl || mState == Closed) {
+    MutexAutoLock lock(mMutex);
+    OpenStreamFailed();
+    mCondVar.NotifyAll();
+    return;
+  }
+
+  if (mAsyncOpenStarted) {
+    return;
+  }
+  mAsyncOpenStarted = true;
+
+  RefPtr<ReadStream::Inner> self = this;
+  mControl->OpenStream(mId, [self](nsCOMPtr<nsIInputStream>&& aStream) {
+    MutexAutoLock lock(self->mMutex);
+    self->mAsyncOpenStarted = false;
+    if (!self->mStream) {
+      if (!aStream) {
+        self->OpenStreamFailed();
+      } else {
+        self->mStream = std::move(aStream);
+        self->mSnappyStream = new SnappyUncompressInputStream(self->mStream);
+      }
+    }
+    self->mCondVar.NotifyAll();
+  });
+}
+
+void ReadStream::Inner::MaybeAbortAsyncOpenStream() {
+  if (!mAsyncOpenStarted) {
+    return;
+  }
+
+  MutexAutoLock lock(mMutex);
+  OpenStreamFailed();
+  mCondVar.NotifyAll();
+}
+
+void ReadStream::Inner::OpenStreamFailed() {
+  MOZ_DIAGNOSTIC_ASSERT(!mStream);
+  MOZ_DIAGNOSTIC_ASSERT(!mSnappyStream);
+  mMutex.AssertCurrentThreadOwns();
+  Unused << NS_NewCStringInputStream(getter_AddRefs(mStream), ""_ns);
+  mSnappyStream = mStream;
+  mStream->Close();
+  NoteClosed();
+}
+
+// ----------------------------------------------------------------------------
+
+NS_IMPL_ISUPPORTS(cache::ReadStream, nsIInputStream, ReadStream);
+
+// static
+already_AddRefed<ReadStream> ReadStream::Create(
+    const Maybe<CacheReadStream>& aMaybeReadStream) {
+  if (aMaybeReadStream.isNothing()) {
+    return nullptr;
+  }
+
+  return Create(aMaybeReadStream.ref());
+}
+
+// static
+already_AddRefed<ReadStream> ReadStream::Create(
+    const CacheReadStream& aReadStream) {
+  // The parameter may or may not be for a Cache created stream.  The way we
+  // tell is by looking at the stream control actor.  If the actor exists,
+  // then we know the Cache created it.
+  if (!aReadStream.control()) {
+    return nullptr;
+  }
+
+  MOZ_DIAGNOSTIC_ASSERT(aReadStream.stream().isNothing() ||
+                        aReadStream.stream().ref().stream().type() !=
+                            mozilla::ipc::InputStreamParams::T__None);
+
+  // Control is guaranteed to survive this method as ActorDestroy() cannot
+  // run on this thread until we complete.
+  StreamControl* control;
+  if (aReadStream.control().IsChild()) {
+    auto actor =
+        static_cast<CacheStreamControlChild*>(aReadStream.control().AsChild());
+    control = actor;
+  } else {
+    auto actor = static_cast<CacheStreamControlParent*>(
+        aReadStream.control().AsParent());
+    control = actor;
+  }
+  MOZ_DIAGNOSTIC_ASSERT(control);
+
+  nsCOMPtr<nsIInputStream> stream = DeserializeIPCStream(aReadStream.stream());
+
+  // Currently we expect all cache read streams to be blocking file streams.
+#if defined(MOZ_DIAGNOSTIC_ASSERT_ENABLED)
+  if (stream) {
+    nsCOMPtr<nsIAsyncInputStream> asyncStream = do_QueryInterface(stream);
+    MOZ_DIAGNOSTIC_ASSERT(!asyncStream);
+  }
+#endif
+
+  return MakeAndAddRef<ReadStream>(MakeSafeRefPtr<ReadStream::Inner>(
+      std::move(control), aReadStream.id(), stream));
+}
+
+// static
+already_AddRefed<ReadStream> ReadStream::Create(
+    PCacheStreamControlParent* aControl, const nsID& aId,
+    nsIInputStream* aStream) {
+  MOZ_DIAGNOSTIC_ASSERT(aControl);
+
+  return MakeAndAddRef<ReadStream>(MakeSafeRefPtr<ReadStream::Inner>(
+      static_cast<CacheStreamControlParent*>(aControl), aId, aStream));
+}
+
+void ReadStream::Serialize(Maybe<CacheReadStream>* aReadStreamOut,
+                           ErrorResult& aRv) {
+  mInner->Serialize(aReadStreamOut, aRv);
+}
+
+void ReadStream::Serialize(CacheReadStream* aReadStreamOut, ErrorResult& aRv) {
+  mInner->Serialize(aReadStreamOut, aRv);
+}
+
+ReadStream::ReadStream(SafeRefPtr<ReadStream::Inner> aInner)
+    : mInner(std::move(aInner)) {
+  MOZ_DIAGNOSTIC_ASSERT(mInner);
+}
+
+ReadStream::~ReadStream() {
+  // Explicitly close the inner stream so that it does not have to
+  // deal with implicitly closing at destruction time.
+  mInner->Close();
+}
+
+NS_IMETHODIMP
+ReadStream::Close() { return mInner->Close(); }
+
+NS_IMETHODIMP
+ReadStream::Available(uint64_t* aNumAvailableOut) {
+  return mInner->Available(aNumAvailableOut);
+}
+
+NS_IMETHODIMP
+ReadStream::StreamStatus() { return mInner->StreamStatus(); }
+
+NS_IMETHODIMP
+ReadStream::Read(char* aBuf, uint32_t aCount, uint32_t* aNumReadOut) {
+  return mInner->Read(aBuf, aCount, aNumReadOut);
+}
+
+NS_IMETHODIMP
+ReadStream::ReadSegments(nsWriteSegmentFun aWriter, void* aClosure,
+                         uint32_t aCount, uint32_t* aNumReadOut) {
+  return mInner->ReadSegments(aWriter, aClosure, aCount, aNumReadOut);
+}
+
+NS_IMETHODIMP
+ReadStream::IsNonBlocking(bool* aNonBlockingOut) {
+  return mInner->IsNonBlocking(aNonBlockingOut);
+}
+
+}  // namespace mozilla::dom::cache