Issue #1587 - Part 3: Hook FetchSignal up to the Fetch API

dom/fetch/Fetch.cpp

@@ -52,41 +52,137 @@ namespace dom {
 
 using namespace workers;
 
+// This class helps the proxying of FetchSignal changes cross threads.
+class FetchSignalProxy final : public FetchSignal::Follower
+{
+  // This is created and released on the main-thread.
+  RefPtr<FetchSignal> mSignalMainThread;
+
+  // This value is used only for the creation of FetchSignal on the
+  // main-thread. They are not updated.
+  const bool mAborted;
+
+  // This runnable propagates changes from the FetchSignal on workers to the
+  // FetchSignal on main-thread.
+  class FetchSignalProxyRunnable final : public Runnable
+  {
+    RefPtr<FetchSignalProxy> mProxy;
+
+  public:
+    explicit FetchSignalProxyRunnable(FetchSignalProxy* aProxy)
+      : mProxy(aProxy)
+    {}
+
+    NS_IMETHOD
+    Run() override
+    {
+      MOZ_ASSERT(NS_IsMainThread());
+      FetchSignal* signal = mProxy->GetOrCreateSignalForMainThread();
+      signal->Abort();
+      return NS_OK;
+    }
+  };
+
+public:
+  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(FetchSignalProxy)
+
+  explicit FetchSignalProxy(FetchSignal* aSignal)
+    : mAborted(aSignal->Aborted())
+  {
+    Follow(aSignal);
+  }
+
+  void
+  Aborted() override
+  {
+    RefPtr<FetchSignalProxyRunnable> runnable =
+      new FetchSignalProxyRunnable(this);
+    NS_DispatchToMainThread(runnable);
+  }
+
+  FetchSignal*
+  GetOrCreateSignalForMainThread()
+  {
+    MOZ_ASSERT(NS_IsMainThread());
+    if (!mSignalMainThread) {
+      mSignalMainThread = new FetchSignal(mAborted);
+    }
+    return mSignalMainThread;
+  }
+
+  void
+  Shutdown()
+  {
+    Unfollow();
+  }
+
+private:
+  ~FetchSignalProxy()
+  {
+    NS_ReleaseOnMainThread(mSignalMainThread.forget());
+  }
+};
+
 class WorkerFetchResolver final : public FetchDriverObserver
 {
   friend class MainThreadFetchRunnable;
+  friend class WorkerFetchResponseEndBase;
   friend class WorkerFetchResponseEndRunnable;
   friend class WorkerFetchResponseRunnable;
 
   RefPtr<PromiseWorkerProxy> mPromiseProxy;
+  RefPtr<FetchSignalProxy> mSignalProxy;
+
 public:
   // Returns null if worker is shutting down.
   static already_AddRefed<WorkerFetchResolver>
-  Create(workers::WorkerPrivate* aWorkerPrivate, Promise* aPromise)
+  Create(workers::WorkerPrivate* aWorkerPrivate, Promise* aPromise,
+         FetchSignal* aSignal)
   {
     MOZ_ASSERT(aWorkerPrivate);
     aWorkerPrivate->AssertIsOnWorkerThread();
-    RefPtr<PromiseWorkerProxy> proxy = PromiseWorkerProxy::Create(aWorkerPrivate, aPromise);
+    RefPtr<PromiseWorkerProxy> proxy =
+      PromiseWorkerProxy::Create(aWorkerPrivate, aPromise);
     if (!proxy) {
       return nullptr;
     }
 
-    RefPtr<WorkerFetchResolver> r = new WorkerFetchResolver(proxy);
+    RefPtr<FetchSignalProxy> signalProxy;
+    if (aSignal) {
+      signalProxy = new FetchSignalProxy(aSignal);
+    }
+
+    RefPtr<WorkerFetchResolver> r = new WorkerFetchResolver(proxy, signalProxy);
     return r.forget();
   }
 
+  FetchSignal*
+  GetFetchSignal()
+  {
+    MOZ_ASSERT(NS_IsMainThread());
+
+    if (!mSignalProxy) {
+      return nullptr;
+    }
+
+    return mSignalProxy->GetOrCreateSignalForMainThread();
+  }
+
   void
   OnResponseAvailableInternal(InternalResponse* aResponse) override;
 
   void
-  OnResponseEnd() override;
+  OnResponseEnd(FetchDriverObserver::EndReason eReason) override;
 
 private:
-  explicit WorkerFetchResolver(PromiseWorkerProxy* aProxy)
+   WorkerFetchResolver(PromiseWorkerProxy* aProxy,
+                       FetchSignalProxy* aSignalProxy)
     : mPromiseProxy(aProxy)
+    , mSignalProxy(aSignalProxy)
   {
     MOZ_ASSERT(!NS_IsMainThread());
     MOZ_ASSERT(mPromiseProxy);
+
   }
 
   ~WorkerFetchResolver()
@@ -115,8 +211,12 @@ public:
     mDocument = aDocument;
   }
 
-  virtual void OnResponseEnd() override
+  void OnResponseEnd(FetchDriverObserver::EndReason aReason) override
   {
+    if (aReason == eAborted) {
+      mPromise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
+    }
+
     FlushConsoleReport();
   }
 
@@ -170,9 +270,11 @@ public:
       fetch->SetWorkerScript(spec);
     }
 
+    RefPtr<FetchSignal> signal = mResolver->GetFetchSignal();
+
     // ...but release it before calling Fetch, because mResolver's callback can
     // be called synchronously and they want the mutex, too.
-    return fetch->Fetch(mResolver);
+    return fetch->Fetch(signal, mResolver);
   }
 };
 
@@ -210,6 +312,12 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
 
   RefPtr<InternalRequest> r = request->GetInternalRequest();
 
+  RefPtr<FetchSignal> signal;
+  if (aInit.mSignal.WasPassed()) {
+    signal = &aInit.mSignal.Value();
+    // Let's FetchDriver to deal with an already aborted signal.
+  }
+
   if (NS_IsMainThread()) {
     nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(aGlobal);
     nsCOMPtr<nsIDocument> doc;
@@ -240,7 +348,7 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
     RefPtr<FetchDriver> fetch = new FetchDriver(r, principal, loadGroup);
     fetch->SetDocument(doc);
     resolver->SetDocument(doc);
-    aRv = fetch->Fetch(resolver);
+    aRv = fetch->Fetch(signal, resolver);
     if (NS_WARN_IF(aRv.Failed())) {
       return nullptr;
     }
@@ -252,7 +360,7 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
       r->SetSkipServiceWorker();
     }
 
-    RefPtr<WorkerFetchResolver> resolver = WorkerFetchResolver::Create(worker, p);
+    RefPtr<WorkerFetchResolver> resolver = WorkerFetchResolver::Create(worker, p, signal);
     if (!resolver) {
       NS_WARNING("Could not add WorkerFetchResolver workerHolder to worker");
       aRv.Throw(NS_ERROR_DOM_ABORT_ERR);
@@ -306,6 +414,7 @@ public:
     , mResolver(aResolver)
     , mInternalResponse(aResponse)
   {
+     MOZ_ASSERT(mResolver);
   }
 
   bool
@@ -332,9 +441,13 @@ public:
 class WorkerFetchResponseEndBase
 {
   RefPtr<PromiseWorkerProxy> mPromiseProxy;
+  RefPtr<FetchSignalProxy> mSignalProxy;
+
 public:
-  explicit WorkerFetchResponseEndBase(PromiseWorkerProxy* aPromiseProxy)
+  WorkerFetchResponseEndBase(PromiseWorkerProxy* aPromiseProxy,
+                             FetchSignalProxy* aSignalProxy)
     : mPromiseProxy(aPromiseProxy)
+    , mSignalProxy(aSignalProxy)
   {
     MOZ_ASSERT(mPromiseProxy);
   }
@@ -344,7 +457,16 @@ public:
   {
     MOZ_ASSERT(aWorkerPrivate);
     aWorkerPrivate->AssertIsOnWorkerThread();
+
+    RefPtr<Promise> promise = mPromiseProxy->WorkerPromise();
+    promise->MaybeReject(NS_ERROR_DOM_ABORT_ERR);
+
     mPromiseProxy->CleanUp();
+
+    if (mSignalProxy) {
+      mSignalProxy->Shutdown();
+      mSignalProxy = nullptr;
+    }
   }
 };
 
@@ -352,9 +474,10 @@ class WorkerFetchResponseEndRunnable final : public MainThreadWorkerRunnable
                                            , public WorkerFetchResponseEndBase
 {
 public:
-  explicit WorkerFetchResponseEndRunnable(PromiseWorkerProxy* aPromiseProxy)
+  WorkerFetchResponseEndRunnable(PromiseWorkerProxy* aPromiseProxy,
+                                 FetchSignalProxy* aSignalProxy)
     : MainThreadWorkerRunnable(aPromiseProxy->GetWorkerPrivate())
-    , WorkerFetchResponseEndBase(aPromiseProxy)
+    , WorkerFetchResponseEndBase(aPromiseProxy, aSignalProxy)
   {
   }
 
@@ -379,9 +502,10 @@ class WorkerFetchResponseEndControlRunnable final : public MainThreadWorkerContr
                                                   , public WorkerFetchResponseEndBase
 {
 public:
-  explicit WorkerFetchResponseEndControlRunnable(PromiseWorkerProxy* aPromiseProxy)
+  WorkerFetchResponseEndControlRunnable(PromiseWorkerProxy* aPromiseProxy,
+                                        FetchSignalProxy* aSignalProxy)
     : MainThreadWorkerControlRunnable(aPromiseProxy->GetWorkerPrivate())
-    , WorkerFetchResponseEndBase(aPromiseProxy)
+    , WorkerFetchResponseEndBase(aPromiseProxy, aSignalProxy)
   {
   }
 
@@ -415,7 +539,7 @@ WorkerFetchResolver::OnResponseAvailableInternal(InternalResponse* aResponse)
 }
 
 void
-WorkerFetchResolver::OnResponseEnd()
+WorkerFetchResolver::OnResponseEnd(FetchDriverObserver::EndReason aReason)
 {
   AssertIsOnMainThread();
   MutexAutoLock lock(mPromiseProxy->Lock());
@@ -426,11 +550,11 @@ WorkerFetchResolver::OnResponseEnd()
   FlushConsoleReport();
 
   RefPtr<WorkerFetchResponseEndRunnable> r =
-    new WorkerFetchResponseEndRunnable(mPromiseProxy);
+    new WorkerFetchResponseEndRunnable(mPromiseProxy, mSignalProxy);
 
   if (!r->Dispatch()) {
     RefPtr<WorkerFetchResponseEndControlRunnable> cr =
-      new WorkerFetchResponseEndControlRunnable(mPromiseProxy);
+      new WorkerFetchResponseEndControlRunnable(mPromiseProxy, mSignalProxy);
     // This can fail if the worker thread is canceled or killed causing
     // the PromiseWorkerProxy to give up its WorkerHolder immediately,
     // allowing the worker thread to become Dead.

dom/fetch/FetchDriver.cpp

@@ -67,7 +67,7 @@ FetchDriver::~FetchDriver()
 }
 
 nsresult
-FetchDriver::Fetch(FetchDriverObserver* aObserver)
+FetchDriver::Fetch(FetchSignal* aSignal, FetchDriverObserver* aObserver)
 {
   workers::AssertIsOnMainThread();
 #ifdef DEBUG
@@ -90,6 +90,18 @@ FetchDriver::Fetch(FetchDriverObserver* aObserver)
   }
 
   mRequest->SetPrincipalInfo(Move(principalInfo));
+
+  // If the signal is aborted, it's time to inform the observer and terminate
+  // the operation.
+  if (aSignal) {
+    if (aSignal->Aborted()) {
+      Aborted();
+      return NS_OK;
+    }
+
+    Follow(aSignal);
+  }
+
   if (NS_FAILED(HttpFetch())) {
     FailWithNetworkError();
   }
@@ -114,11 +126,7 @@ FetchDriver::HttpFetch()
   nsAutoCString url;
   mRequest->GetURL(url);
   nsCOMPtr<nsIURI> uri;
-  rv = NS_NewURI(getter_AddRefs(uri),
-                          url,
-                          nullptr,
-                          nullptr,
-                          ios);
+  rv = NS_NewURI(getter_AddRefs(uri), url, nullptr, nullptr, ios);
   NS_ENSURE_SUCCESS(rv, rv);
 
   // Unsafe requests aren't allowed with when using no-core mode.
@@ -380,6 +388,8 @@ FetchDriver::HttpFetch()
   NS_ENSURE_SUCCESS(rv, rv);
 
   // Step 4 onwards of "HTTP Fetch" is handled internally by Necko.
+
+  mChannel = chan;
   return NS_OK;
 }
 already_AddRefed<InternalResponse>
@@ -433,9 +443,11 @@ FetchDriver::FailWithNetworkError()
 #ifdef DEBUG
     mResponseAvailableCalled = true;
 #endif
-    mObserver->OnResponseEnd();
+    mObserver->OnResponseEnd(FetchDriverObserver::eByNetworking);
     mObserver = nullptr;
   }
+  
+  mChannel = nullptr;
 }
 
 namespace {
@@ -777,10 +789,11 @@ FetchDriver::OnStopRequest(nsIRequest* aRequest,
       #endif
     }
 
-    mObserver->OnResponseEnd();
+    mObserver->OnResponseEnd(FetchDriverObserver::eByNetworking);
     mObserver = nullptr;
   }
 
+  mChannel = nullptr;
   return NS_OK;
 }
 
@@ -921,5 +934,21 @@ FetchDriver::SetRequestHeaders(nsIHttpChannel* aChannel) const
   }
 }
 
+void FetchDriver::Aborted()
+{
+  if (mObserver) {
+#ifdef DEBUG
+    mResponseAvailableCalled = true;
+#endif
+    mObserver->OnResponseEnd(FetchDriverObserver::eAborted);
+    mObserver = nullptr;
+  }
+
+  if (mChannel) {
+    mChannel->Cancel(NS_BINDING_ABORTED);
+    mChannel = nullptr;
+  }
+}
+
 } // namespace dom
 } // namespace mozilla

dom/fetch/FetchDriver.h

@@ -12,6 +12,7 @@
 #include "nsIStreamListener.h"
 #include "nsIThreadRetargetableStreamListener.h"
 #include "mozilla/ConsoleReportCollector.h"
+#include "mozilla/dom/FetchSignal.h"
 #include "mozilla/dom/SRIMetadata.h"
 #include "mozilla/RefPtr.h"
 
@@ -49,7 +50,14 @@ public:
     mGotResponseAvailable = true;
     OnResponseAvailableInternal(aResponse);
   }
-  virtual void OnResponseEnd()
+
+  enum EndReason
+  {
+    eAborted,
+    eByNetworking,
+  };
+
+  virtual void OnResponseEnd(EndReason aReason)
   { };
 
   nsIConsoleReportCollector* GetReporter() const
@@ -58,6 +66,7 @@ public:
   }
 
   virtual void FlushConsoleReport() = 0;
+
 protected:
   virtual ~FetchDriverObserver()
   { };
@@ -72,7 +81,8 @@ private:
 class FetchDriver final : public nsIStreamListener,
                           public nsIChannelEventSink,
                           public nsIInterfaceRequestor,
-                          public nsIThreadRetargetableStreamListener
+                          public nsIThreadRetargetableStreamListener,
+                          public FetchSignal::Follower
 {
 public:
   NS_DECL_ISUPPORTS
@@ -82,9 +92,12 @@ public:
   NS_DECL_NSIINTERFACEREQUESTOR
   NS_DECL_NSITHREADRETARGETABLESTREAMLISTENER
 
-  explicit FetchDriver(InternalRequest* aRequest, nsIPrincipal* aPrincipal,
-                       nsILoadGroup* aLoadGroup);
-  NS_IMETHOD Fetch(FetchDriverObserver* aObserver);
+  FetchDriver(InternalRequest* aRequest,
+              nsIPrincipal* aPrincipal,
+              nsILoadGroup* aLoadGroup);
+
+  nsresult Fetch(FetchSignal* aSignal,
+                 FetchDriverObserver* aObserver);
 
   void
   SetDocument(nsIDocument* aDocument);
@@ -96,6 +109,11 @@ public:
     mWorkerScript = aWorkerScirpt;
   }
 
+  // FetchSignal::Follower
+
+  void
+  Aborted() override;
+
 private:
   nsCOMPtr<nsIPrincipal> mPrincipal;
   nsCOMPtr<nsILoadGroup> mLoadGroup;
@@ -104,6 +122,7 @@ private:
   nsCOMPtr<nsIOutputStream> mPipeOutputStream;
   RefPtr<FetchDriverObserver> mObserver;
   nsCOMPtr<nsIDocument> mDocument;
+  nsCOMPtr<nsIChannel> mChannel;
   nsAutoPtr<SRICheckDataVerifier> mSRIDataVerifier;
   SRIMetadata mSRIMetadata;
   nsCString mWorkerScript;

dom/fetch/FetchSignal.cpp

@@ -37,6 +37,10 @@ FetchSignal::FetchSignal(FetchController* aController,
   , mAborted(aAborted)
 {}
 
+FetchSignal::FetchSignal(bool aAborted)
+  : mAborted(aAborted)
+{}
+
 JSObject*
 FetchSignal::WrapObject(JSContext* aCx, JS::Handle<JSObject*> aGivenProto)
 {
@@ -95,6 +99,10 @@ FetchSignal::CanAcceptFollower(FetchSignal::Follower* aFollower) const
 {
   MOZ_DIAGNOSTIC_ASSERT(aFollower);
 
+  if (!mController) {
+    return true;
+  }
+
   if (aFollower == mController) {
     return false;
   }

dom/fetch/FetchSignal.h

@@ -40,6 +40,7 @@ public:
   NS_DECL_CYCLE_COLLECTION_CLASS_INHERITED(FetchSignal, DOMEventTargetHelper)
 
   FetchSignal(FetchController* aController, bool aAborted);
+  explicit FetchSignal(bool aAborted);
 
   JSObject*
   WrapObject(JSContext* aCx, JS::Handle<JSObject*> aGivenProto) override;

dom/tests/mochitest/fetch/file_fetch_controller.html

@@ -84,13 +84,94 @@ function testAbortEvent() {
   fc.abort();
 }
  
+function testAbortedFetch() {
+  var fc = new FetchController();
+  fc.abort();
+
+  fetch('data:,foo', { signal: fc.signal }).then(() => {
+    ok(false, "Fetch should not return a resolved promise");
+  }, e => {
+    is(e.name, "AbortError", "We have an abort error");
+  }).then(next);
+}
+
+function testFetchAndAbort() {
+  var fc = new FetchController();
+
+  var p = fetch('data:,foo', { signal: fc.signal });
+  fc.abort();
+
+  p.then(() => {
+    ok(false, "Fetch should not return a resolved promise");
+  }, e => {
+    is(e.name, "AbortError", "We have an abort error");
+  }).then(next);
+}
+
+function testWorkerAbortedFetch() {
+  function worker() {
+    var fc = new FetchController();
+    fc.abort();
+
+    fetch('data:,foo', { signal: fc.signal }).then(() => {
+      postMessage(false);
+    }, e => {
+      postMessage(e.name == "AbortError");
+    });
+  }
+
+  var str = worker.toString();
+  var content = str.substring(0, str.length - 1).split('\n').splice(1).join(' ');
+  var url = URL.createObjectURL(new Blob([content], { type: "application/javascript" }));
+  var w = new Worker(url);
+  w.onmessage = function(e) {
+    ok(e.data, "Abort + Fetch works in workers");
+    next();
+  }
+}
+
+function testWorkerFetchAndAbort() {
+  function worker() {
+    var fc = new FetchController();
+
+    var p = fetch('data:,foo', { signal: fc.signal });
+    fc.abort();
+
+    p.then(() => {
+      postMessage(false);
+    }, e => {
+      postMessage(e.name == "AbortError");
+    });
+  }
+
+  var str = worker.toString();
+  var content = str.substring(0, str.length - 1).split('\n').splice(1).join(' ');
+  var url = URL.createObjectURL(new Blob([content], { type: "application/javascript" }));
+  var w = new Worker(url);
+  w.onmessage = function(e) {
+    ok(e.data, "Abort + Fetch works in workers");
+    next();
+  }
+}
+
 var steps = [
+  // Simple stuff
   testWebIDL,
   testUpdateData,
+
+  // Following algorithm
   testFollowingOurself,
   testFollowingOther,
   testFollowingLoop,
+
+  // Event propagation
   testAbortEvent,
+
+  // fetch + signaling
+  testAbortedFetch,
+  testFetchAndAbort,
+  testWorkerAbortedFetch,
+  testWorkerFetchAndAbort,
 ];
 
 function next() {