Issue #1587 Part 11 (followup 1): Implement multithreaded signals for workers.

dom/cache/TypeUtils.cpp

@@ -253,7 +253,7 @@ TypeUtils::ToResponse(const CacheResponse& aIn)
 {
   if (aIn.type() == ResponseType::Error) {
     RefPtr<InternalResponse> error = InternalResponse::NetworkError();
-    RefPtr<Response> r = new Response(GetGlobalObject(), error);
+    RefPtr<Response> r = new Response(GetGlobalObject(), error, nullptr);
     return r.forget();
   }
 
@@ -302,7 +302,7 @@ TypeUtils::ToResponse(const CacheResponse& aIn)
   }
   MOZ_DIAGNOSTIC_ASSERT(ir);
 
-  RefPtr<Response> ref = new Response(GetGlobalObject(), ir);
+  RefPtr<Response> ref = new Response(GetGlobalObject(), ir, nullptr);
   return ref.forget();
 }
 already_AddRefed<InternalRequest>
@@ -345,7 +345,7 @@ already_AddRefed<Request>
 TypeUtils::ToRequest(const CacheRequest& aIn)
 {
   RefPtr<InternalRequest> internalRequest = ToInternalRequest(aIn);
-  RefPtr<Request> request = new Request(GetGlobalObject(), internalRequest);
+  RefPtr<Request> request = new Request(GetGlobalObject(), internalRequest, nullptr);
   return request.forget();
 }
 

dom/fetch/Fetch.cpp

@@ -111,6 +111,12 @@ public:
     return mSignalMainThread;
   }
 
+  AbortSignal*
+  GetSignalForTargetThread()
+  {
+    return mFollowingSignal;
+  }
+
   void
   Shutdown()
   {
@@ -161,7 +167,7 @@ public:
   }
 
   AbortSignal*
-  GetAbortSignal()
+  GetAbortSignalForMainThread()
   {
     MOZ_ASSERT(NS_IsMainThread());
 
@@ -172,6 +178,18 @@ public:
     return mSignalProxy->GetOrCreateSignalForMainThread();
   }
 
+  AbortSignal*
+  GetAbortSignalForTargetThread()
+  {
+    mPromiseProxy->GetWorkerPrivate()->AssertIsOnWorkerThread();
+
+    if (!mSignalProxy) {
+      return nullptr;
+    }
+
+    return mSignalProxy->GetSignalForTargetThread();
+  }
+
   void
   OnResponseAvailableInternal(InternalResponse* aResponse) override;
 
@@ -205,14 +223,16 @@ class MainThreadFetchResolver final : public FetchDriverObserver
   RefPtr<Promise> mPromise;
   RefPtr<Response> mResponse;
   RefPtr<FetchObserver> mFetchObserver;
+  RefPtr<AbortSignal> mSignal;
 
   nsCOMPtr<nsIDocument> mDocument;
 
   NS_DECL_OWNINGTHREAD
 public:
-  MainThreadFetchResolver(Promise* aPromise, FetchObserver* aObserver)
+  MainThreadFetchResolver(Promise* aPromise, FetchObserver* aObserver, AbortSignal* aSignal)
     : mPromise(aPromise)
     , mFetchObserver(aObserver)
+    , mSignal(aSignal)
   {}
 
   void
@@ -287,7 +307,7 @@ public:
       fetch->SetWorkerScript(spec);
     }
 
-    RefPtr<AbortSignal> signal = mResolver->GetAbortSignal();
+    RefPtr<AbortSignal> signal = mResolver->GetAbortSignalForMainThread();
 
     // ...but release it before calling Fetch, because mResolver's callback can
     // be called synchronously and they want the mutex, too.
@@ -329,10 +349,7 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
 
   RefPtr<InternalRequest> r = request->GetInternalRequest();
 
-  RefPtr<AbortSignal> signal;
+  RefPtr<AbortSignal> signal = request->GetSignal();
-  if (aInit.mSignal.WasPassed()) {
-    signal = &aInit.mSignal.Value();
-  }
   
   if (signal && signal->Aborted()) {
     // An already aborted signal should reject immediately.
@@ -373,7 +390,7 @@ FetchRequest(nsIGlobalObject* aGlobal, const RequestOrUSVString& aInput,
     }
 
     RefPtr<MainThreadFetchResolver> resolver =
-      new MainThreadFetchResolver(p, observer);
+      new MainThreadFetchResolver(p, observer, signal);
     RefPtr<FetchDriver> fetch = new FetchDriver(r, principal, loadGroup);
     fetch->SetDocument(doc);
     resolver->SetDocument(doc);
@@ -416,7 +433,7 @@ MainThreadFetchResolver::OnResponseAvailableInternal(InternalResponse* aResponse
     }
 
     nsCOMPtr<nsIGlobalObject> go = mPromise->GetParentObject();
-    mResponse = new Response(go, aResponse);
+    mResponse = new Response(go, aResponse, mSignal);
     mPromise->MaybeResolve(mResponse);
   } else {
     if (mFetchObserver) {
@@ -479,7 +496,7 @@ public:
       }
 
       RefPtr<nsIGlobalObject> global = aWorkerPrivate->GlobalScope();
-      RefPtr<Response> response = new Response(global, mInternalResponse);
+      RefPtr<Response> response = new Response(global, mInternalResponse, mResolver->GetAbortSignalForTargetThread());
       promise->MaybeResolve(response);
     } else {
       if (mResolver->mFetchObserver) {
@@ -926,6 +943,12 @@ template <class Derived>
 already_AddRefed<Promise>
 FetchBody<Derived>::ConsumeBody(FetchConsumeType aType, ErrorResult& aRv)
 {
+  RefPtr<AbortSignal> signal = DerivedClass()->GetSignal();
+  if (signal && signal->Aborted()) {
+    aRv.Throw(NS_ERROR_DOM_ABORT_ERR);
+    return nullptr;
+  }
+
   if (BodyUsed()) {
     aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
     return nullptr;
@@ -935,7 +958,7 @@ FetchBody<Derived>::ConsumeBody(FetchConsumeType aType, ErrorResult& aRv)
 
   RefPtr<Promise> promise =
     FetchBodyConsumer<Derived>::Create(DerivedClass()->GetParentObject(),
-                                       this, aType, aRv);
+                                       this, signal, aType, aRv);
   if (NS_WARN_IF(aRv.Failed())) {
     return nullptr;
   }

dom/fetch/Fetch.h

@@ -162,6 +162,9 @@ public:
     return mMimeType;
   }
 
+  virtual AbortSignal*
+  GetSignal() const = 0;
+
 protected:
   FetchBody();
 

dom/fetch/FetchConsumer.cpp

@@ -284,6 +284,7 @@ template <class Derived>
 /* static */ already_AddRefed<Promise>
 FetchBodyConsumer<Derived>::Create(nsIGlobalObject* aGlobal,
                                    FetchBody<Derived>* aBody,
+                                   AbortSignal* aSignal,
                                    FetchConsumeType aType,
                                    ErrorResult& aRv)
 {
@@ -345,6 +346,10 @@ FetchBodyConsumer<Derived>::Create(nsIGlobalObject* aGlobal,
     return nullptr;
   }
 
+  if (aSignal) {
+    consumer->Follow(aSignal);
+  }
+
   return promise.forget();
 }
 
@@ -368,6 +373,8 @@ FetchBodyConsumer<Derived>::ReleaseObject()
 #ifdef DEBUG
   mBody = nullptr;
 #endif
+
+  Unfollow();
 }
 
 template <class Derived>
@@ -684,6 +691,14 @@ FetchBodyConsumer<Derived>::Observe(nsISupports* aSubject,
   return NS_OK;
 }
 
+template <class Derived> 
+void
+FetchBodyConsumer<Derived>::Aborted()
+{
+  AssertIsOnTargetThread();
+  ContinueConsumeBody(NS_ERROR_DOM_ABORT_ERR, 0, nullptr);
+}
+
 template <class Derived>
 NS_IMPL_ADDREF(FetchBodyConsumer<Derived>)
 

dom/fetch/FetchConsumer.h

@@ -11,6 +11,7 @@
 #include "nsIInputStream.h"
 #include "nsIObserver.h"
 #include "nsWeakReference.h"
+#include "mozilla/dom/AbortSignal.h"
 #include "mozilla/dom/MutableBlobStorage.h"
 
 class nsIThread;
@@ -34,6 +35,7 @@ template <class Derived> class FetchBody;
 template <class Derived>
 class FetchBodyConsumer final : public nsIObserver
                               , public nsSupportsWeakReference
+                              , public AbortSignal::Follower
 {
 public:
   NS_DECL_THREADSAFE_ISUPPORTS
@@ -42,6 +44,7 @@ public:
   static already_AddRefed<Promise>
   Create(nsIGlobalObject* aGlobal,
          FetchBody<Derived>* aBody,
+         AbortSignal* aSignal,
          FetchConsumeType aType,
          ErrorResult& aRv);
 
@@ -73,6 +76,9 @@ public:
     mConsumeBodyPump = nullptr;
   }
 
+  // Override AbortSignal::Follower::Aborted
+  void Aborted() override;
+
 private:
   FetchBodyConsumer(nsIGlobalObject* aGlobalObject,
                     workers::WorkerPrivate* aWorkerPrivate,

dom/fetch/Request.cpp

@@ -37,15 +37,18 @@ NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(Request)
   NS_INTERFACE_MAP_ENTRY(nsISupports)
 NS_INTERFACE_MAP_END
 
-Request::Request(nsIGlobalObject* aOwner, InternalRequest* aRequest)
+Request::Request(nsIGlobalObject* aOwner, InternalRequest* aRequest, AbortSignal* aSignal)
   : FetchBody<Request>()
   , mOwner(aOwner)
   , mRequest(aRequest)
+  , mSignal(aSignal)
 {
   MOZ_ASSERT(aRequest->Headers()->Guard() == HeadersGuardEnum::Immutable ||
              aRequest->Headers()->Guard() == HeadersGuardEnum::Request ||
              aRequest->Headers()->Guard() == HeadersGuardEnum::Request_no_cors);
   SetMimeType();
+  
+  // aSignal can be null.
 }
 
 Request::~Request()
@@ -287,6 +290,8 @@ Request::Constructor(const GlobalObject& aGlobal,
 
   nsCOMPtr<nsIGlobalObject> global = do_QueryInterface(aGlobal.GetAsSupports());
   
+  RefPtr<AbortSignal> signal;
+
   if (aInput.IsRequest()) {
     RefPtr<Request> inputReq = &aInput.GetAsRequest();
     nsCOMPtr<nsIInputStream> body;
@@ -300,6 +305,7 @@ Request::Constructor(const GlobalObject& aGlobal,
     }
 
     request = inputReq->GetInternalRequest();
+    signal = inputReq->GetOrCreateSignal();
   } else {
     // aInput is USVString.
     // We need to get url before we create a InternalRequest.
@@ -418,6 +424,10 @@ Request::Constructor(const GlobalObject& aGlobal,
     request->SetReferrerPolicy(aInit.mReferrerPolicy.Value());
   }
 
+  if (aInit.mSignal.WasPassed()) {
+    signal = &aInit.mSignal.Value();
+  }
+
   if (NS_IsMainThread()) {
     nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(global);
     if (window) {
@@ -579,7 +589,7 @@ Request::Constructor(const GlobalObject& aGlobal,
     }
   }
 
-  RefPtr<Request> domRequest = new Request(global, request);
+  RefPtr<Request> domRequest = new Request(global, request, signal);
   domRequest->SetMimeType();
 
   if (aInput.IsRequest()) {
@@ -595,7 +605,7 @@ Request::Constructor(const GlobalObject& aGlobal,
 }
 
 already_AddRefed<Request>
-Request::Clone(ErrorResult& aRv) const
+Request::Clone(ErrorResult& aRv)
 {
   if (BodyUsed()) {
     aRv.ThrowTypeError<MSG_FETCH_BODY_CONSUMED_ERROR>();
@@ -608,7 +618,7 @@ Request::Clone(ErrorResult& aRv) const
     return nullptr;
   }
 
-  RefPtr<Request> request = new Request(mOwner, ir);
+  RefPtr<Request> request = new Request(mOwner, ir, GetOrCreateSignal());
   return request.forget();
 }
 
@@ -622,5 +632,21 @@ Request::Headers_()
   return mHeaders;
 }
 
+AbortSignal*
+Request::GetOrCreateSignal()
+{
+  if (!mSignal) {
+    mSignal = new AbortSignal(false);
+  }
+
+  return mSignal;
+}
+
+AbortSignal*
+Request::GetSignal() const
+{
+  return mSignal;
+}
+
 } // namespace dom
 } // namespace mozilla

dom/fetch/Request.h

@@ -33,7 +33,7 @@ class Request final : public nsISupports
   NS_DECL_CYCLE_COLLECTION_SCRIPT_HOLDER_CLASS(Request)
 
 public:
-  Request(nsIGlobalObject* aOwner, InternalRequest* aRequest);
+  Request(nsIGlobalObject* aOwner, InternalRequest* aRequest, AbortSignal* aSignal);
 
   static bool
   RequestContextEnabled(JSContext* aCx, JSObject* aObj);
@@ -142,7 +142,7 @@ public:
   }
 
   already_AddRefed<Request>
-  Clone(ErrorResult& aRv) const;
+  Clone(ErrorResult& aRv);
 
   already_AddRefed<InternalRequest>
   GetInternalRequest();
@@ -153,13 +153,22 @@ public:
     return mRequest->GetPrincipalInfo();
   }
 
+  AbortSignal*
+  GetOrCreateSignal();
+
+  // This can return a null AbortSignal.
+  AbortSignal*
+  GetSignal() const override;
+
 private:
   ~Request();
 
   nsCOMPtr<nsIGlobalObject> mOwner;
   RefPtr<InternalRequest> mRequest;
+  
   // Lazily created.
   RefPtr<Headers> mHeaders;
+  RefPtr<AbortSignal> mSignal;
 };
 
 } // namespace dom

dom/fetch/Response.cpp

@@ -34,10 +34,11 @@ NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(Response)
   NS_INTERFACE_MAP_ENTRY(nsISupports)
 NS_INTERFACE_MAP_END
 
-Response::Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse)
+Response::Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse, AbortSignal* aSignal)
   : FetchBody<Response>()
   , mOwner(aGlobal)
   , mInternalResponse(aInternalResponse)
+  , mSignal(aSignal)
 {
   MOZ_ASSERT(aInternalResponse->Headers()->Guard() == HeadersGuardEnum::Immutable ||
              aInternalResponse->Headers()->Guard() == HeadersGuardEnum::Response);
@@ -53,7 +54,7 @@ Response::Error(const GlobalObject& aGlobal)
 {
   nsCOMPtr<nsIGlobalObject> global = do_QueryInterface(aGlobal.GetAsSupports());
   RefPtr<InternalResponse> error = InternalResponse::NetworkError();
-  RefPtr<Response> r = new Response(global, error);
+  RefPtr<Response> r = new Response(global, error, nullptr);
   return r.forget();
 }
 
@@ -173,7 +174,7 @@ Response::Constructor(const GlobalObject& aGlobal,
     internalResponse->InitChannelInfo(worker->GetChannelInfo());
   }
 
-  RefPtr<Response> r = new Response(global, internalResponse);
+  RefPtr<Response> r = new Response(global, internalResponse, nullptr);
 
   if (aInit.mHeaders.WasPassed()) {
     internalResponse->Headers()->Clear();
@@ -236,7 +237,7 @@ Response::Clone(ErrorResult& aRv) const
   }
 
   RefPtr<InternalResponse> ir = mInternalResponse->Clone();
-  RefPtr<Response> response = new Response(mOwner, ir);
+  RefPtr<Response> response = new Response(mOwner, ir, mSignal);
   return response.forget();
 }
 
@@ -250,7 +251,7 @@ Response::CloneUnfiltered(ErrorResult& aRv) const
 
   RefPtr<InternalResponse> clone = mInternalResponse->Clone();
   RefPtr<InternalResponse> ir = clone->Unfiltered();
-  RefPtr<Response> ref = new Response(mOwner, ir);
+  RefPtr<Response> ref = new Response(mOwner, ir, mSignal);
   return ref.forget();
 }
 

dom/fetch/Response.h

@@ -33,7 +33,7 @@ class Response final : public nsISupports
   NS_DECL_CYCLE_COLLECTION_SCRIPT_HOLDER_CLASS(Response)
 
 public:
-  Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse);
+  Response(nsIGlobalObject* aGlobal, InternalResponse* aInternalResponse, AbortSignal* aSignal);
 
   Response(const Response& aOther) = delete;
 
@@ -134,13 +134,21 @@ public:
   already_AddRefed<InternalResponse>
   GetInternalResponse() const;
 
+  AbortSignal*
+  GetSignal() const override
+  {
+    return mSignal;
+  }
+
 private:
   ~Response();
 
   nsCOMPtr<nsIGlobalObject> mOwner;
   RefPtr<InternalResponse> mInternalResponse;
+
   // Lazily created
   RefPtr<Headers> mHeaders;
+  RefPtr<AbortSignal> mSignal;
 };
 
 } // namespace dom

dom/webidl/Request.webidl

@@ -27,6 +27,10 @@ interface Request {
   readonly attribute RequestRedirect redirect;
   readonly attribute DOMString integrity;
 
+  [Func="AbortController::IsEnabled",
+   BinaryName="getOrCreateSignal"]
+  readonly attribute AbortSignal signal;
+
   [Throws,
    NewObject] Request clone();
 

dom/workers/ScriptLoader.cpp

@@ -694,7 +694,7 @@ private:
     ir->SetPrincipalInfo(Move(principalInfo));
 
     RefPtr<mozilla::dom::Response> response =
-      new mozilla::dom::Response(mCacheCreator->Global(), ir);
+      new mozilla::dom::Response(mCacheCreator->Global(), ir, nullptr);
 
     mozilla::dom::RequestOrUSVString request;
 

dom/workers/ServiceWorkerPrivate.cpp

@@ -1510,7 +1510,7 @@ private:
     if (NS_WARN_IF(!global)) {
       return false;
     }
-    RefPtr<Request> request = new Request(global, internalReq);
+    RefPtr<Request> request = new Request(global, internalReq, nullptr);
 
     MOZ_ASSERT_IF(internalReq->IsNavigationRequest(),
                   request->Redirect() == RequestRedirect::Manual);

dom/workers/ServiceWorkerScriptCache.cpp

@@ -554,7 +554,7 @@ private:
       ir->SetPrincipalInfo(Move(mPrincipalInfo));
     }
 
-    RefPtr<Response> response = new Response(aCache->GetGlobalObject(), ir);
+    RefPtr<Response> response = new Response(aCache->GetGlobalObject(), ir, nullptr);
 
     RequestOrUSVString request;
     request.SetAsUSVString().Rebind(URL().Data(), URL().Length());