Bug 1110814 P1 Implement Cache IPC actor for streaming data from child to parent. r=khuey

dom/cache/AutoUtils.cpp

@@ -7,6 +7,7 @@
 #include "mozilla/dom/cache/AutoUtils.h"
 
 #include "mozilla/unused.h"
+#include "mozilla/dom/cache/CachePushStreamChild.h"
 #include "mozilla/dom/cache/CacheStreamControlParent.h"
 #include "mozilla/dom/cache/ReadStream.h"
 #include "mozilla/dom/cache/SavedTypes.h"
@@ -19,6 +20,7 @@
 namespace {
 
 using mozilla::unused;
+using mozilla::dom::cache::CachePushStreamChild;
 using mozilla::dom::cache::PCacheReadStream;
 using mozilla::dom::cache::PCacheReadStreamOrVoid;
 using mozilla::ipc::FileDescriptor;
@@ -28,8 +30,8 @@ using mozilla::ipc::OptionalFileDescriptorSet;
 
 enum CleanupAction
 {
-  ForgetFds,
-  DeleteFds
+  Forget,
+  Delete
 };
 
 void
@@ -46,7 +48,7 @@ CleanupChildFds(PCacheReadStream& aReadStream, CleanupAction aAction)
     static_cast<FileDescriptorSetChild*>(aReadStream.fds().get_PFileDescriptorSetChild());
   MOZ_ASSERT(fdSetActor);
 
-  if (aAction == DeleteFds) {
+  if (aAction == Delete) {
     unused << fdSetActor->Send__delete__(fdSetActor);
   }
 
@@ -57,13 +59,39 @@ CleanupChildFds(PCacheReadStream& aReadStream, CleanupAction aAction)
 }
 
 void
-CleanupChildFds(PCacheReadStreamOrVoid& aReadStreamOrVoid, CleanupAction aAction)
+CleanupChildPushStream(PCacheReadStream& aReadStream, CleanupAction aAction)
+{
+  if (!aReadStream.pushStreamChild()) {
+    return;
+  }
+
+  auto pushStream =
+    static_cast<CachePushStreamChild*>(aReadStream.pushStreamChild());
+
+  if (aAction == Delete) {
+    pushStream->StartDestroy();
+    return;
+  }
+
+  // If we send the stream, then we need to start it before forgetting about it.
+  pushStream->Start();
+}
+
+void
+CleanupChild(PCacheReadStream& aReadStream, CleanupAction aAction)
+{
+  CleanupChildFds(aReadStream, aAction);
+  CleanupChildPushStream(aReadStream, aAction);
+}
+
+void
+CleanupChild(PCacheReadStreamOrVoid& aReadStreamOrVoid, CleanupAction aAction)
 {
   if (aReadStreamOrVoid.type() == PCacheReadStreamOrVoid::Tvoid_t) {
     return;
   }
 
-  CleanupChildFds(aReadStreamOrVoid.get_PCacheReadStream(), aAction);
+  CleanupChild(aReadStreamOrVoid.get_PCacheReadStream(), aAction);
 }
 
 void
@@ -80,7 +108,7 @@ CleanupParentFds(PCacheReadStream& aReadStream, CleanupAction aAction)
     static_cast<FileDescriptorSetParent*>(aReadStream.fds().get_PFileDescriptorSetParent());
   MOZ_ASSERT(fdSetActor);
 
-  if (aAction == DeleteFds) {
+  if (aAction == Delete) {
     unused << fdSetActor->Send__delete__(fdSetActor);
   }
 
@@ -133,8 +161,8 @@ AutoChildRequest::~AutoChildRequest()
     return;
   }
 
-  CleanupAction action = mSent ? ForgetFds : DeleteFds;
-  CleanupChildFds(mRequestOrVoid.get_PCacheRequest().body(), action);
+  CleanupAction action = mSent ? Forget : Delete;
+  CleanupChild(mRequestOrVoid.get_PCacheRequest().body(), action);
 }
 
 void
@@ -173,9 +201,9 @@ AutoChildRequestList::AutoChildRequestList(TypeUtils* aTypeUtils,
 
 AutoChildRequestList::~AutoChildRequestList()
 {
-  CleanupAction action = mSent ? ForgetFds : DeleteFds;
+  CleanupAction action = mSent ? Forget : Delete;
   for (uint32_t i = 0; i < mRequestList.Length(); ++i) {
-    CleanupChildFds(mRequestList[i].body(), action);
+    CleanupChild(mRequestList[i].body(), action);
   }
 }
 
@@ -223,9 +251,9 @@ AutoChildRequestResponse::AutoChildRequestResponse(TypeUtils* aTypeUtils)
 
 AutoChildRequestResponse::~AutoChildRequestResponse()
 {
-  CleanupAction action = mSent ? ForgetFds : DeleteFds;
-  CleanupChildFds(mRequestResponse.request().body(), action);
-  CleanupChildFds(mRequestResponse.response().body(), action);
+  CleanupAction action = mSent ? Forget : Delete;
+  CleanupChild(mRequestResponse.request().body(), action);
+  CleanupChild(mRequestResponse.response().body(), action);
 }
 
 void
@@ -311,7 +339,7 @@ AutoParentRequestList::AutoParentRequestList(PBackgroundParent* aManager,
 
 AutoParentRequestList::~AutoParentRequestList()
 {
-  CleanupAction action = mSent ? ForgetFds : DeleteFds;
+  CleanupAction action = mSent ? Forget : Delete;
   for (uint32_t i = 0; i < mRequestList.Length(); ++i) {
     CleanupParentFds(mRequestList[i].body(), action);
   }
@@ -355,7 +383,7 @@ AutoParentResponseList::AutoParentResponseList(PBackgroundParent* aManager,
 
 AutoParentResponseList::~AutoParentResponseList()
 {
-  CleanupAction action = mSent ? ForgetFds : DeleteFds;
+  CleanupAction action = mSent ? Forget : Delete;
   for (uint32_t i = 0; i < mResponseList.Length(); ++i) {
     CleanupParentFds(mResponseList[i].body(), action);
   }
@@ -402,7 +430,7 @@ AutoParentResponseOrVoid::~AutoParentResponseOrVoid()
     return;
   }
 
-  CleanupAction action = mSent ? ForgetFds : DeleteFds;
+  CleanupAction action = mSent ? Forget : Delete;
   CleanupParentFds(mResponseOrVoid.get_PCacheResponse().body(), action);
 }
 

dom/cache/Cache.cpp

@@ -14,6 +14,7 @@
 #include "mozilla/dom/CacheBinding.h"
 #include "mozilla/dom/cache/AutoUtils.h"
 #include "mozilla/dom/cache/CacheChild.h"
+#include "mozilla/dom/cache/CachePushStreamChild.h"
 #include "mozilla/dom/cache/ReadStream.h"
 #include "mozilla/dom/cache/TypeUtils.h"
 #include "mozilla/ErrorResult.h"
@@ -526,6 +527,18 @@ Cache::AssertOwningThread() const
 }
 #endif
 
+CachePushStreamChild*
+Cache::CreatePushStream(nsIAsyncInputStream* aStream)
+{
+  NS_ASSERT_OWNINGTHREAD(Cache);
+  MOZ_ASSERT(mActor);
+  MOZ_ASSERT(aStream);
+  auto actor = mActor->SendPCachePushStreamConstructor(
+    new CachePushStreamChild(mActor->GetFeature(), aStream));
+  MOZ_ASSERT(actor);
+  return static_cast<CachePushStreamChild*>(actor);
+}
+
 void
 Cache::ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue)
 {

dom/cache/Cache.h

@@ -39,8 +39,8 @@ class PCacheResponse;
 class PCacheResponseOrVoid;
 
 class Cache final : public PromiseNativeHandler
-                      , public nsWrapperCache
-                      , public TypeUtils
+                  , public nsWrapperCache
+                  , public TypeUtils
 {
 public:
   Cache(nsIGlobalObject* aGlobal, CacheChild* aActor);
@@ -97,6 +97,9 @@ public:
   virtual void AssertOwningThread() const override;
 #endif
 
+  virtual CachePushStreamChild*
+  CreatePushStream(nsIAsyncInputStream* aStream) override;
+
   // PromiseNativeHandler methods
   virtual void
   ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue) override;

dom/cache/CacheChild.cpp

@@ -9,6 +9,7 @@
 #include "mozilla/unused.h"
 #include "mozilla/dom/cache/ActorUtils.h"
 #include "mozilla/dom/cache/Cache.h"
+#include "mozilla/dom/cache/PCachePushStreamChild.h"
 #include "mozilla/dom/cache/StreamUtils.h"
 
 namespace mozilla {
@@ -94,6 +95,20 @@ CacheChild::ActorDestroy(ActorDestroyReason aReason)
   RemoveFeature();
 }
 
+PCachePushStreamChild*
+CacheChild::AllocPCachePushStreamChild()
+{
+  MOZ_CRASH("CachePushStreamChild should be manually constructed.");
+  return nullptr;
+}
+
+bool
+CacheChild::DeallocPCachePushStreamChild(PCachePushStreamChild* aActor)
+{
+  delete aActor;
+  return true;
+}
+
 bool
 CacheChild::RecvMatchResponse(const RequestId& requestId, const nsresult& aRv,
                               const PCacheResponseOrVoid& aResponse)

dom/cache/CacheChild.h

@@ -17,7 +17,7 @@ namespace cache {
 class Cache;
 
 class CacheChild final : public PCacheChild
-                           , public ActorChild
+                       , public ActorChild
 {
 public:
   CacheChild();
@@ -41,6 +41,12 @@ private:
   virtual void
   ActorDestroy(ActorDestroyReason aReason) override;
 
+  virtual PCachePushStreamChild*
+  AllocPCachePushStreamChild() override;
+
+  virtual bool
+  DeallocPCachePushStreamChild(PCachePushStreamChild* aActor) override;
+
   virtual bool
   RecvMatchResponse(const RequestId& requestId, const nsresult& aRv,
                     const PCacheResponseOrVoid& aResponse) override;

dom/cache/CacheParent.cpp

@@ -8,6 +8,7 @@
 
 #include "mozilla/DebugOnly.h"
 #include "mozilla/dom/cache/AutoUtils.h"
+#include "mozilla/dom/cache/CachePushStreamParent.h"
 #include "mozilla/dom/cache/CacheStreamControlParent.h"
 #include "mozilla/dom/cache/ReadStream.h"
 #include "mozilla/dom/cache/SavedTypes.h"
@@ -61,6 +62,19 @@ CacheParent::ActorDestroy(ActorDestroyReason aReason)
   mManager = nullptr;
 }
 
+PCachePushStreamParent*
+CacheParent::AllocPCachePushStreamParent()
+{
+  return CachePushStreamParent::Create();
+}
+
+bool
+CacheParent::DeallocPCachePushStreamParent(PCachePushStreamParent* aActor)
+{
+  delete aActor;
+  return true;
+}
+
 bool
 CacheParent::RecvTeardown()
 {
@@ -259,13 +273,27 @@ CacheParent::DeserializeCacheStream(const PCacheReadStreamOrVoid& aStreamOrVoid)
     return nullptr;
   }
 
+  nsCOMPtr<nsIInputStream> stream;
   const PCacheReadStream& readStream = aStreamOrVoid.get_PCacheReadStream();
 
-  nsCOMPtr<nsIInputStream> stream = ReadStream::Create(readStream);
+  // Option 1: A push stream actor was sent for nsPipe data
+  if (readStream.pushStreamParent()) {
+    MOZ_ASSERT(!readStream.controlParent());
+    CachePushStreamParent* pushStream =
+      static_cast<CachePushStreamParent*>(readStream.pushStreamParent());
+    stream = pushStream->TakeReader();
+    MOZ_ASSERT(stream);
+    return stream.forget();
+  }
+
+  // Option 2: One of our own ReadStreams was passed back to us with a stream
+  //           control actor.
+  stream = ReadStream::Create(readStream);
   if (stream) {
     return stream.forget();
   }
 
+  // Option 3: A stream was serialized using normal methods.
   nsAutoTArray<FileDescriptor, 4> fds;
   if (readStream.fds().type() ==
       OptionalFileDescriptorSet::TPFileDescriptorSetChild) {

dom/cache/CacheParent.h

@@ -22,8 +22,8 @@ namespace cache {
 struct SavedResponse;
 
 class CacheParent final : public PCacheParent
-                            , public Manager::Listener
-                            , public FetchPut::Listener
+                        , public Manager::Listener
+                        , public FetchPut::Listener
 {
 public:
   CacheParent(cache::Manager* aManager, CacheId aCacheId);
@@ -32,6 +32,8 @@ public:
 private:
   // PCacheParent method
   virtual void ActorDestroy(ActorDestroyReason aReason) override;
+  virtual PCachePushStreamParent* AllocPCachePushStreamParent();
+  virtual bool DeallocPCachePushStreamParent(PCachePushStreamParent* aActor);
   virtual bool RecvTeardown() override;
   virtual bool
   RecvMatch(const RequestId& aRequestId, const PCacheRequest& aRequest,

dom/cache/CachePushStreamChild.cpp

@@ -0,0 +1,259 @@
+/* -*- 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/CachePushStreamChild.h"
+
+#include "mozilla/unused.h"
+#include "nsIAsyncInputStream.h"
+#include "nsICancelableRunnable.h"
+#include "nsIThread.h"
+#include "nsStreamUtils.h"
+
+namespace mozilla {
+namespace dom {
+namespace cache {
+
+class CachePushStreamChild::Callback final : public nsIInputStreamCallback
+                                           , public nsICancelableRunnable
+{
+public:
+  explicit Callback(CachePushStreamChild* aActor)
+    : mActor(aActor)
+    , mOwningThread(NS_GetCurrentThread())
+  {
+    MOZ_ASSERT(mActor);
+  }
+
+  NS_IMETHOD
+  OnInputStreamReady(nsIAsyncInputStream* aStream) override
+  {
+    // any thread
+    if (mOwningThread == NS_GetCurrentThread()) {
+      return Run();
+    }
+
+    // If this fails, then it means the owning thread is a Worker that has
+    // been shutdown.  Its ok to lose the event in this case because the
+    // CachePushStreamChild listens for this event through the Feature.
+    nsresult rv = mOwningThread->Dispatch(this, nsIThread::DISPATCH_NORMAL);
+    if (NS_FAILED(rv)) {
+      NS_WARNING("Failed to dispatch stream readable event to owning thread");
+    }
+
+    return NS_OK;
+  }
+
+  NS_IMETHOD
+  Run() override
+  {
+    MOZ_ASSERT(mOwningThread == NS_GetCurrentThread());
+    if (mActor) {
+      mActor->OnStreamReady(this);
+    }
+    return NS_OK;
+  }
+
+  NS_IMETHOD
+  Cancel() override
+  {
+    // Cancel() gets called when the Worker thread is being shutdown.  We have
+    // nothing to do here because CachePushStreamChild handles this case via
+    // the Feature.
+    return NS_OK;
+  }
+
+  void
+  ClearActor()
+  {
+    MOZ_ASSERT(mOwningThread == NS_GetCurrentThread());
+    MOZ_ASSERT(mActor);
+    mActor = nullptr;
+  }
+
+private:
+  ~Callback()
+  {
+    // called on any thread
+
+    // ClearActor() should be called before the Callback is destroyed
+    MOZ_ASSERT(!mActor);
+  }
+
+  CachePushStreamChild* mActor;
+  nsCOMPtr<nsIThread> mOwningThread;
+
+  NS_DECL_THREADSAFE_ISUPPORTS
+};
+
+NS_IMPL_ISUPPORTS(CachePushStreamChild::Callback, nsIInputStreamCallback,
+                                                  nsIRunnable,
+                                                  nsICancelableRunnable);
+
+CachePushStreamChild::CachePushStreamChild(Feature* aFeature,
+                                           nsIAsyncInputStream* aStream)
+  : mStream(aStream)
+  , mClosed(false)
+{
+  MOZ_ASSERT(mStream);
+  MOZ_ASSERT_IF(!NS_IsMainThread(), aFeature);
+  SetFeature(aFeature);
+}
+
+CachePushStreamChild::~CachePushStreamChild()
+{
+  NS_ASSERT_OWNINGTHREAD(CachePushStreamChild);
+  MOZ_ASSERT(mClosed);
+  MOZ_ASSERT(!mCallback);
+}
+
+void
+CachePushStreamChild::Start()
+{
+  DoRead();
+}
+
+void
+CachePushStreamChild::StartDestroy()
+{
+  // called if we are running on a Worker and the thread gets shutdown
+  OnEnd(NS_ERROR_ABORT);
+}
+
+void
+CachePushStreamChild::ActorDestroy(ActorDestroyReason aReason)
+{
+  NS_ASSERT_OWNINGTHREAD(CachePushStreamChild);
+
+  // If the parent side runs into a problem then the actor will be destroyed.
+  // In this case we have not run OnEnd(), so still need to close the input
+  // stream.
+  if (!mClosed) {
+    mStream->CloseWithStatus(NS_ERROR_ABORT);
+    mClosed = true;
+  }
+
+  if (mCallback) {
+    mCallback->ClearActor();
+    mCallback = nullptr;
+  }
+
+  RemoveFeature();
+}
+
+void
+CachePushStreamChild::DoRead()
+{
+  NS_ASSERT_OWNINGTHREAD(CachePushStreamChild);
+  MOZ_ASSERT(!mClosed);
+  MOZ_ASSERT(!mCallback);
+
+  // The input stream (likely a pipe) probably uses a segment size of
+  // 4kb.  If there is data already buffered it would be nice to aggregate
+  // multiple segments into a single IPC call.  Conversely, don't send too
+  // too large of a buffer in a single call to avoid spiking memory.
+  static const uint64_t kMaxBytesPerMessage = 32 * 1024;
+  static_assert(kMaxBytesPerMessage <= static_cast<uint64_t>(UINT32_MAX),
+                "kMaxBytesPerMessage must cleanly cast to uint32_t");
+
+  while (!mClosed) {
+    // Use non-auto here as we're unlikely to hit stack storage with the
+    // sizes we are sending.  Also, it would be nice to avoid another copy
+    // to the IPC layer which we avoid if we use COW strings.  Unfortunately
+    // IPC does not seem to support passing dependent storage types.
+    nsCString buffer;
+
+    uint64_t available = 0;
+    nsresult rv = mStream->Available(&available);
+    if (NS_FAILED(rv)) {
+      OnEnd(rv);
+      return;
+    }
+
+    if (available == 0) {
+      Wait();
+      return;
+    }
+
+    uint32_t expectedBytes =
+      static_cast<uint32_t>(std::min(available, kMaxBytesPerMessage));
+
+    buffer.SetLength(expectedBytes);
+
+    uint32_t bytesRead = 0;
+    rv = mStream->Read(buffer.BeginWriting(), buffer.Length(), &bytesRead);
+    buffer.SetLength(bytesRead);
+
+    // If we read any data from the stream, send it across.
+    if (!buffer.IsEmpty()) {
+      unused << SendBuffer(buffer);
+    }
+
+    if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
+      Wait();
+      return;
+    }
+
+    // Any other error or zero-byte read indicates end-of-stream
+    if (NS_FAILED(rv) || buffer.IsEmpty()) {
+      OnEnd(rv);
+      return;
+    }
+  }
+}
+
+void
+CachePushStreamChild::Wait()
+{
+  NS_ASSERT_OWNINGTHREAD(CachePushStreamChild);
+  MOZ_ASSERT(!mClosed);
+  MOZ_ASSERT(!mCallback);
+
+  // Set mCallback immediately instead of waiting for success.  Its possible
+  // AsyncWait() will callback synchronously.
+  mCallback = new Callback(this);
+  nsresult rv = mStream->AsyncWait(mCallback, 0, 0, nullptr);
+  if (NS_FAILED(rv)) {
+    OnEnd(rv);
+    return;
+  }
+}
+
+void
+CachePushStreamChild::OnStreamReady(Callback* aCallback)
+{
+  NS_ASSERT_OWNINGTHREAD(CachePushStreamChild);
+  MOZ_ASSERT(mCallback);
+  MOZ_ASSERT(aCallback == mCallback);
+  mCallback->ClearActor();
+  mCallback = nullptr;
+  DoRead();
+}
+
+void
+CachePushStreamChild::OnEnd(nsresult aRv)
+{
+  NS_ASSERT_OWNINGTHREAD(CachePushStreamChild);
+  MOZ_ASSERT(aRv != NS_BASE_STREAM_WOULD_BLOCK);
+
+  if (mClosed) {
+    return;
+  }
+
+  mClosed = true;
+
+  mStream->CloseWithStatus(aRv);
+
+  if (aRv == NS_BASE_STREAM_CLOSED) {
+    aRv = NS_OK;
+  }
+
+  // This will trigger an ActorDestroy() from the parent side
+  unused << SendClose(aRv);
+}
+
+} // namespace cache
+} // namespace dom
+} // namespace mozilla

dom/cache/CachePushStreamChild.h

@@ -0,0 +1,57 @@
+/* -*- 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/. */
+
+#ifndef mozilla_dom_cache_CachePushStreamChild_h
+#define mozilla_dom_cache_CachePushStreamChild_h
+
+#include "mozilla/dom/cache/ActorChild.h"
+#include "mozilla/dom/cache/PCachePushStreamChild.h"
+#include "nsCOMPtr.h"
+
+class nsIAsyncInputStream;
+
+namespace mozilla {
+namespace dom {
+namespace cache {
+
+class CachePushStreamChild final : public PCachePushStreamChild
+                                 , public ActorChild
+{
+public:
+  CachePushStreamChild(Feature* aFeature, nsIAsyncInputStream* aStream);
+  ~CachePushStreamChild();
+
+  virtual void StartDestroy() override;
+
+  void Start();
+
+private:
+  class Callback;
+
+  // PCachePushStreamChild methods
+  virtual void
+  ActorDestroy(ActorDestroyReason aReason) override;
+
+  void DoRead();
+
+  void Wait();
+
+  void OnStreamReady(Callback* aCallback);
+
+  void OnEnd(nsresult aRv);
+
+  nsCOMPtr<nsIAsyncInputStream> mStream;
+  nsRefPtr<Callback> mCallback;
+  bool mClosed;
+
+  NS_DECL_OWNINGTHREAD
+};
+
+} // namespace cache
+} // namespace dom
+} // namespace mozilla
+
+#endif // mozilla_dom_cache_CachePushStreamChild_h

dom/cache/CachePushStreamParent.cpp

@@ -0,0 +1,97 @@
+/* -*- 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/CachePushStreamParent.h"
+
+#include "mozilla/unused.h"
+#include "nsIAsyncInputStream.h"
+#include "nsIAsyncOutputStream.h"
+#include "nsIPipe.h"
+
+namespace mozilla {
+namespace dom {
+namespace cache {
+
+// static
+CachePushStreamParent*
+CachePushStreamParent::Create()
+{
+  // 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.
+  nsCOMPtr<nsIAsyncInputStream> reader;
+  nsCOMPtr<nsIAsyncOutputStream> writer;
+
+  // 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(reader),
+                            getter_AddRefs(writer),
+                            true, true,   // non-blocking
+                            0,            // segment size
+                            UINT32_MAX);  // "infinite" pipe
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    return nullptr;
+  }
+
+  return new CachePushStreamParent(reader, writer);
+}
+
+CachePushStreamParent::~CachePushStreamParent()
+{
+}
+
+already_AddRefed<nsIInputStream>
+CachePushStreamParent::TakeReader()
+{
+  MOZ_ASSERT(mReader);
+  return mReader.forget();
+}
+
+void
+CachePushStreamParent::ActorDestroy(ActorDestroyReason aReason)
+{
+  // 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);
+}
+
+bool
+CachePushStreamParent::RecvBuffer(const nsCString& aBuffer)
+{
+  uint32_t numWritten = 0;
+
+  // This should only fail if we hit an OOM condition.
+  nsresult rv = mWriter->Write(aBuffer.get(), aBuffer.Length(), &numWritten);
+  if (NS_WARN_IF(NS_FAILED(rv))) {
+    RecvClose(rv);
+  }
+
+  return true;
+}
+
+bool
+CachePushStreamParent::RecvClose(const nsresult& aRv)
+{
+  mWriter->CloseWithStatus(aRv);
+  unused << Send__delete__(this);
+  return true;
+}
+
+CachePushStreamParent::CachePushStreamParent(nsIAsyncInputStream* aReader,
+                                             nsIAsyncOutputStream* aWriter)
+  : mReader(aReader)
+  , mWriter(aWriter)
+{
+  MOZ_ASSERT(mReader);
+  MOZ_ASSERT(mWriter);
+}
+
+} // namespace cache
+} // namespace dom
+} // namespace mozilla

dom/cache/CachePushStreamParent.h

@@ -0,0 +1,55 @@
+/* -*- 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/. */
+
+#ifndef mozilla_dom_cache_CachePushStreamParent_h
+#define mozilla_dom_cache_CachePushStreamParent_h
+
+#include "mozilla/dom/cache/PCachePushStreamParent.h"
+
+class nsIAsyncInputStream;
+class nsIAsyncOutputStream;
+class nsIInputStream;
+
+namespace mozilla {
+namespace dom {
+namespace cache {
+
+class CachePushStreamParent final : public PCachePushStreamParent
+{
+public:
+  static CachePushStreamParent*
+  Create();
+
+  ~CachePushStreamParent();
+
+  already_AddRefed<nsIInputStream>
+  TakeReader();
+
+private:
+  CachePushStreamParent(nsIAsyncInputStream* aReader,
+                        nsIAsyncOutputStream* aWriter);
+
+  // PCachePushStreamParent methods
+  virtual void
+  ActorDestroy(ActorDestroyReason aReason) override;
+
+  virtual bool
+  RecvBuffer(const nsCString& aBuffer) override;
+
+  virtual bool
+  RecvClose(const nsresult& aRv) override;
+
+  nsCOMPtr<nsIAsyncInputStream> mReader;
+  nsCOMPtr<nsIAsyncOutputStream> mWriter;
+
+  NS_DECL_OWNINGTHREAD
+};
+
+} // namespace cache
+} // namespace dom
+} // namespace mozilla
+
+#endif // mozilla_dom_cache_CachePushStreamParent_h

dom/cache/CacheStorage.cpp

@@ -516,6 +516,13 @@ CacheStorage::AssertOwningThread() const
 }
 #endif
 
+CachePushStreamChild*
+CacheStorage::CreatePushStream(nsIAsyncInputStream* aStream)
+{
+  // This is true because CacheStorage always uses IgnoreBody for requests.
+  MOZ_CRASH("CacheStorage should never create a push stream.");
+}
+
 void
 CacheStorage::ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue)
 {

dom/cache/CacheStorage.h

@@ -44,9 +44,9 @@ class Feature;
 class PCacheResponseOrVoid;
 
 class CacheStorage final : public nsIIPCBackgroundChildCreateCallback
-                             , public nsWrapperCache
-                             , public TypeUtils
-                             , public PromiseNativeHandler
+                         , public nsWrapperCache
+                         , public TypeUtils
+                         , public PromiseNativeHandler
 {
   typedef mozilla::ipc::PBackgroundChild PBackgroundChild;
 
@@ -97,6 +97,9 @@ public:
   virtual void AssertOwningThread() const override;
 #endif
 
+  virtual CachePushStreamChild*
+  CreatePushStream(nsIAsyncInputStream* aStream) override;
+
   // PromiseNativeHandler methods
   virtual void
   ResolvedCallback(JSContext* aCx, JS::Handle<JS::Value> aValue) override;

dom/cache/CacheStorageChild.h

@@ -20,7 +20,7 @@ class PCacheChild;
 class Feature;
 
 class CacheStorageChild final : public PCacheStorageChild
-                                  , public ActorChild
+                              , public ActorChild
 {
 public:
   CacheStorageChild(CacheStorage* aListener, Feature* aFeature);
@@ -41,6 +41,7 @@ public:
 private:
   // PCacheStorageChild methods
   virtual void ActorDestroy(ActorDestroyReason aReason) override;
+
   virtual bool RecvMatchResponse(const RequestId& aRequestId,
                                  const nsresult& aRv,
                                  const PCacheResponseOrVoid& response) override;

dom/cache/CacheStorageParent.h

@@ -23,8 +23,8 @@ class CacheStreamControlParent;
 class ManagerId;
 
 class CacheStorageParent final : public PCacheStorageParent
-                                   , public PrincipalVerifier::Listener
-                                   , public Manager::Listener
+                               , public PrincipalVerifier::Listener
+                               , public Manager::Listener
 {
 public:
   CacheStorageParent(PBackgroundParent* aManagingActor, Namespace aNamespace,

dom/cache/CacheStreamControlChild.h

@@ -19,8 +19,8 @@ namespace cache {
 class ReadStream;
 
 class CacheStreamControlChild final : public PCacheStreamControlChild
-                                        , public StreamControl
-                                        , public ActorChild
+                                    , public StreamControl
+                                    , public ActorChild
 {
 public:
   CacheStreamControlChild();

dom/cache/CacheStreamControlParent.h

@@ -18,8 +18,8 @@ namespace cache {
 class ReadStream;
 class StreamList;
 
-class CacheStreamControlParent : public PCacheStreamControlParent
-                               , public StreamControl
+class CacheStreamControlParent final : public PCacheStreamControlParent
+                                     , public StreamControl
 {
 public:
   CacheStreamControlParent();

dom/cache/Context.cpp

@@ -320,7 +320,7 @@ Context::QuotaInitRunnable::Run()
 // runnable executes the Action on the appropriate threads while the Context
 // is initialized.
 class Context::ActionRunnable final : public nsIRunnable
-                                        , public Action::Resolver
+                                    , public Action::Resolver
 {
 public:
   ActionRunnable(Context* aContext, nsIEventTarget* aTarget, Action* aAction,

dom/cache/FetchPut.cpp

@@ -458,6 +458,12 @@ FetchPut::AssertOwningThread() const
 }
 #endif
 
+CachePushStreamChild*
+FetchPut::CreatePushStream(nsIAsyncInputStream* aStream)
+{
+  MOZ_CRASH("FetchPut should never create a push stream!");
+}
+
 } // namespace cache
 } // namespace dom
 } // namespace mozilla

dom/cache/FetchPut.h

@@ -30,7 +30,7 @@ class Response;
 namespace cache {
 
 class FetchPut final : public Manager::Listener
-                         , public TypeUtils
+                     , public TypeUtils
 {
 public:
   typedef std::pair<nsRefPtr<Request>, nsRefPtr<Response>> PutPair;
@@ -94,6 +94,9 @@ private:
   virtual void AssertOwningThread() const override;
 #endif
 
+  virtual CachePushStreamChild*
+  CreatePushStream(nsIAsyncInputStream* aStream) override;
+
   Listener* mListener;
   nsRefPtr<Manager> mManager;
   const RequestId mRequestId;

dom/cache/PCache.ipdl

@@ -3,6 +3,7 @@
  * You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 include protocol PBackground;
+include protocol PCachePushStream;
 include PCacheTypes;
 include protocol PFileDescriptorSet;
 
@@ -19,8 +20,10 @@ namespace cache {
 protocol PCache
 {
   manager PBackground;
+  manages PCachePushStream;
 
 parent:
+  PCachePushStream();
   Teardown();
   Match(RequestId requestId, PCacheRequest request, PCacheQueryParams params);
   MatchAll(RequestId requestId, PCacheRequestOrVoid request, PCacheQueryParams params);

dom/cache/PCachePushStream.ipdl

@@ -0,0 +1,28 @@
+/* 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 protocol PCache;
+
+namespace mozilla {
+namespace dom {
+namespace cache {
+
+protocol PCachePushStream
+{
+  manager PCache;
+
+parent:
+  Buffer(nsCString aBuffer);
+  Close(nsresult aRv);
+
+child:
+  // Stream is always destroyed from the parent side.  This occurs if the
+  // parent encounters an error while writing to its pipe or if the child
+  // signals the stream should close by SendClose().
+  __delete__();
+};
+
+} // namespace cache
+} // namespace dom
+} // namespace mozilla

dom/cache/PCacheTypes.ipdlh

@@ -2,6 +2,7 @@
  * 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 protocol PCachePushStream;
 include protocol PCacheStreamControl;
 include PHeaders;
 include InputStreamParams;
@@ -32,6 +33,7 @@ struct PCacheReadStream
   OptionalInputStreamParams params;
   OptionalFileDescriptorSet fds;
   nullable PCacheStreamControl control;
+  nullable PCachePushStream pushStream;
 };
 
 union PCacheReadStreamOrVoid

dom/cache/ReadStream.cpp

@@ -210,6 +210,11 @@ ReadStream::Inner::Serialize(PCacheReadStream* aReadStreamOut)
   MOZ_ASSERT(mState == Open);
   MOZ_ASSERT(mControl);
 
+  // If we are sending a ReadStream, then we never want to set the
+  // pushStream actors at the same time.
+  aReadStreamOut->pushStreamChild() = nullptr;
+  aReadStreamOut->pushStreamParent() = nullptr;
+
   aReadStreamOut->id() = mId;
   mControl->SerializeControl(aReadStreamOut);
 
@@ -406,6 +411,9 @@ ReadStream::Create(const PCacheReadStream& aReadStream)
     return nullptr;
   }
 
+  MOZ_ASSERT(!aReadStream.pushStreamChild());
+  MOZ_ASSERT(!aReadStream.pushStreamParent());
+
   // Control is guaranteed to survive this method as ActorDestroy() cannot
   // run on this thread until we complete.
   StreamControl* control;

dom/cache/StreamList.h

@@ -21,7 +21,7 @@ class CacheStreamControlParent;
 class Context;
 class Manager;
 
-class StreamList
+class StreamList final
 {
 public:
   StreamList(Manager* aManager, Context* aContext);

dom/cache/TypeUtils.cpp

@@ -11,6 +11,7 @@
 #include "mozilla/dom/InternalRequest.h"
 #include "mozilla/dom/Request.h"
 #include "mozilla/dom/Response.h"
+#include "mozilla/dom/cache/CachePushStreamChild.h"
 #include "mozilla/dom/cache/PCacheTypes.h"
 #include "mozilla/dom/cache/ReadStream.h"
 #include "mozilla/ipc/BackgroundChild.h"
@@ -29,6 +30,13 @@
 namespace {
 
 using mozilla::ErrorResult;
+using mozilla::unused;
+using mozilla::void_t;
+using mozilla::dom::cache::PCacheReadStream;
+using mozilla::ipc::BackgroundChild;
+using mozilla::ipc::FileDescriptor;
+using mozilla::ipc::PBackgroundChild;
+using mozilla::ipc::PFileDescriptorSetChild;
 
 // Utility function to remove the fragment from a URL, check its scheme, and optionally
 // provide a URL without the query.  We're not using nsIURL or URL to do this because
@@ -96,6 +104,31 @@ ProcessURL(nsAString& aUrl, bool* aSchemeValidOut,
   *aUrlWithoutQueryOut = Substring(aUrl, 0, queryPos - 1);
 }
 
+void
+SerializeNormalStream(nsIInputStream* aStream, PCacheReadStream& aReadStreamOut)
+{
+  nsAutoTArray<FileDescriptor, 4> fds;
+  SerializeInputStream(aStream, aReadStreamOut.params(), fds);
+
+  PFileDescriptorSetChild* fdSet = nullptr;
+  if (!fds.IsEmpty()) {
+    // We should not be serializing until we have an actor ready
+    PBackgroundChild* manager = BackgroundChild::GetForCurrentThread();
+    MOZ_ASSERT(manager);
+
+    fdSet = manager->SendPFileDescriptorSetConstructor(fds[0]);
+    for (uint32_t i = 1; i < fds.Length(); ++i) {
+      unused << fdSet->SendAddFileDescriptor(fds[i]);
+    }
+  }
+
+  if (fdSet) {
+    aReadStreamOut.fds() = fdSet;
+  } else {
+    aReadStreamOut.fds() = void_t();
+  }
+}
+
 } // anonymous namespace
 
 namespace mozilla {
@@ -413,64 +446,61 @@ TypeUtils::SerializeCacheStream(nsIInputStream* aStream,
     return;
   }
 
+  // Option 1: Send a cache-specific ReadStream if we can.
   nsRefPtr<ReadStream> controlled = do_QueryObject(aStream);
   if (controlled) {
     controlled->Serialize(aStreamOut);
     return;
   }
 
-  // TODO: implement CrossProcessPipe if we cannot directly serialize (bug 1110814)
-  nsCOMPtr<nsIIPCSerializableInputStream> serial = do_QueryInterface(aStream);
-  if (!serial) {
-    aRv.Throw(NS_ERROR_FAILURE);
-    return;
-  }
-
   PCacheReadStream readStream;
   readStream.controlChild() = nullptr;
   readStream.controlParent() = nullptr;
+  readStream.pushStreamChild() = nullptr;
+  readStream.pushStreamParent() = nullptr;
 
-  nsAutoTArray<FileDescriptor, 4> fds;
-  SerializeInputStream(aStream, readStream.params(), fds);
+  // Option 2: Do normal stream serialization if its supported.
+  nsCOMPtr<nsIIPCSerializableInputStream> serial = do_QueryInterface(aStream);
+  if (serial) {
+    SerializeNormalStream(aStream, readStream);
 
-  PFileDescriptorSetChild* fdSet = nullptr;
-  if (!fds.IsEmpty()) {
-    // We should not be serializing until we have an actor ready
-    PBackgroundChild* manager = BackgroundChild::GetForCurrentThread();
-    MOZ_ASSERT(manager);
-
-    fdSet = manager->SendPFileDescriptorSetConstructor(fds[0]);
-    for (uint32_t i = 1; i < fds.Length(); ++i) {
-      unused << fdSet->SendAddFileDescriptor(fds[i]);
-    }
-  }
-
-  if (fdSet) {
-    readStream.fds() = fdSet;
+  // Option 3: As a last resort push data across manually.  Should only be
+  //           needed for nsPipe input stream.  Only works for async,
+  //           non-blocking streams.
   } else {
-    readStream.fds() = void_t();
+    SerializePushStream(aStream, readStream, aRv);
+    if (NS_WARN_IF(aRv.Failed())) { return; }
   }
 
   *aStreamOut = readStream;
 }
 
-nsIThread*
-TypeUtils::GetStreamThread()
+void
+TypeUtils::SerializePushStream(nsIInputStream* aStream,
+                               PCacheReadStream& aReadStreamOut,
+                               ErrorResult& aRv)
 {
-  AssertOwningThread();
-
-  if (!mStreamThread) {
-    // Named threads only allow 16 bytes for their names.  Try to make
-    // it meaningful...
-    // TODO: use a thread pool or singleton thread here (bug 1119864)
-    nsresult rv = NS_NewNamedThread("DOMCacheTypeU",
-                                    getter_AddRefs(mStreamThread));
-    if (NS_FAILED(rv) || !mStreamThread) {
-      MOZ_CRASH("Failed to create DOM Cache serialization thread.");
-    }
+  nsCOMPtr<nsIAsyncInputStream> asyncStream = do_QueryInterface(aStream);
+  if (NS_WARN_IF(!asyncStream)) {
+    aRv = NS_ERROR_FAILURE;
+    return;
   }
 
-  return mStreamThread;
+  bool nonBlocking = false;
+  aRv = asyncStream->IsNonBlocking(&nonBlocking);
+  if (NS_WARN_IF(aRv.Failed())) { return; }
+  if (NS_WARN_IF(!nonBlocking)) {
+    aRv = NS_ERROR_FAILURE;
+    return;
+  }
+
+  aReadStreamOut.pushStreamChild() = CreatePushStream(asyncStream);
+  MOZ_ASSERT(aReadStreamOut.pushStreamChild());
+  aReadStreamOut.params() = void_t();
+  aReadStreamOut.fds() = void_t();
+
+  // CachePushStreamChild::Start() must be called after sending the stream
+  // across to the parent side.
 }
 
 } // namespace cache

dom/cache/TypeUtils.h

@@ -9,10 +9,10 @@
 
 #include "mozilla/Attributes.h"
 #include "mozilla/dom/BindingUtils.h"
-#include "nsCOMPtr.h"
 #include "nsError.h"
 
 class nsIGlobalObject;
+class nsIAsyncInputStream;
 class nsIInputStream;
 
 namespace mozilla {
@@ -28,7 +28,9 @@ class Response;
 
 namespace cache {
 
+class CachePushStreamChild;
 class PCacheQueryParams;
+class PCacheReadStream;
 class PCacheReadStreamOrVoid;
 class PCacheRequest;
 class PCacheResponse;
@@ -63,6 +65,9 @@ public:
   inline void AssertOwningThread() const { }
 #endif
 
+  virtual CachePushStreamChild*
+  CreatePushStream(nsIAsyncInputStream* aStream) = 0;
+
   already_AddRefed<InternalRequest>
   ToInternalRequest(const RequestOrUSVString& aIn, BodyAction aBodyAction,
                     ErrorResult& aRv);
@@ -107,9 +112,9 @@ private:
   SerializeCacheStream(nsIInputStream* aStream, PCacheReadStreamOrVoid* aStreamOut,
                        ErrorResult& aRv);
 
-  nsIThread* GetStreamThread();
-
-  nsCOMPtr<nsIThread> mStreamThread;
+  void
+  SerializePushStream(nsIInputStream* aStream, PCacheReadStream& aReadStreamOut,
+                      ErrorResult& aRv);
 };
 
 } // namespace cache

dom/cache/moz.build

@@ -12,6 +12,8 @@ EXPORTS.mozilla.dom.cache += [
     'Cache.h',
     'CacheChild.h',
     'CacheParent.h',
+    'CachePushStreamChild.h',
+    'CachePushStreamParent.h',
     'CacheStorage.h',
     'CacheStorageChild.h',
     'CacheStorageParent.h',
@@ -44,6 +46,8 @@ UNIFIED_SOURCES += [
     'Cache.cpp',
     'CacheChild.cpp',
     'CacheParent.cpp',
+    'CachePushStreamChild.cpp',
+    'CachePushStreamParent.cpp',
     'CacheStorage.cpp',
     'CacheStorageChild.cpp',
     'CacheStorageParent.cpp',
@@ -69,6 +73,7 @@ UNIFIED_SOURCES += [
 IPDL_SOURCES += [
     'CacheInitData.ipdlh',
     'PCache.ipdl',
+    'PCachePushStream.ipdl',
     'PCacheStorage.ipdl',
     'PCacheStreamControl.ipdl',
     'PCacheTypes.ipdlh',

dom/cache/test/mochitest/mochitest.ini

@@ -16,6 +16,7 @@ support-files =
   vary.sjs
   test_caches.js
   test_cache_keys.js
+  test_cache_put.js
 
 [test_cache.html]
 [test_cache_add.html]
@@ -25,3 +26,4 @@ support-files =
 [test_cache_match_vary.html]
 [test_caches.html]
 [test_cache_keys.html]
+[test_cache_put.html]

dom/cache/test/mochitest/test_cache_put.html

@@ -0,0 +1,20 @@
+<!-- Any copyright is dedicated to the Public Domain.
+   - http://creativecommons.org/publicdomain/zero/1.0/ -->
+<!DOCTYPE HTML>
+<html>
+<head>
+  <title>Validate Interfaces Exposed to Workers</title>
+  <script type="text/javascript" src="/tests/SimpleTest/SimpleTest.js"></script>
+  <link rel="stylesheet" type="text/css" href="/tests/SimpleTest/test.css" />
+  <script type="text/javascript" src="driver.js"></script>
+</head>
+<body>
+<iframe id="frame"></iframe>
+<script class="testbody" type="text/javascript">
+  runTests("test_cache_put.js")
+    .then(function() {
+      SimpleTest.finish();
+    });
+</script>
+</body>
+</html>

dom/cache/test/mochitest/test_cache_put.js

@@ -0,0 +1,25 @@
+var url = 'test_cache.js';
+var cache;
+var fetchResponse;
+Promise.all([fetch(url),
+             caches.open('putter' + context)]).then(function(results) {
+  fetchResponse = results[0];
+  cache = results[1];
+  return cache.put(url, fetchResponse.clone());
+}).then(function(result) {
+  is(undefined, result, 'Successful put() should resolve undefined');
+  return cache.match(url);
+}).then(function(response) {
+  ok(response, 'match() should find resppnse that was previously put()');
+  ok(response.url.endsWith(url), 'matched response should match original url');
+  return Promise.all([fetchResponse.text(),
+                      response.text()]);
+}).then(function(results) {
+  // suppress large assert spam unless its relevent
+  if (results[0] !== results[1]) {
+    is(results[0], results[1], 'stored response body should match original');
+  }
+  return cache.delete('putter' + context);
+}).then(function() {
+  testDone();
+});