Bug 1770630 - Worker stream readers should contribute to busy count. r=dom-worker-reviewers,jstutte

Differential Revision: https://phabricator.services.mozilla.com/D149185

dom/base/BodyStream.cpp

@@ -115,8 +115,8 @@ void BodyStream::Create(JSContext* aCx, BodyStreamHolder* aStreamHolder,
     WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
     MOZ_ASSERT(workerPrivate);
 
-    RefPtr<WeakWorkerRef> workerRef =
-        WeakWorkerRef::Create(workerPrivate, [stream]() { stream->Close(); });
+    RefPtr<StrongWorkerRef> workerRef =
+        StrongWorkerRef::Create(workerPrivate, "BodyStream", [stream]() { stream->Close(); });
 
     if (NS_WARN_IF(!workerRef)) {
       aRv.Throw(NS_ERROR_DOM_INVALID_STATE_ERR);
@@ -215,6 +215,7 @@ already_AddRefed<Promise> BodyStream::PullCallback(
     ErrorPropagation(aCx, lock, stream, rv);
     return nullptr;
   }
+  mAsyncWaitWorkerRef = mWorkerRef;
 
   // All good.
   return resolvedWithUndefinedPromise.forget();
@@ -272,6 +273,7 @@ void BodyStream::WriteIntoReadRequestBuffer(JSContext* aCx,
     ErrorPropagation(aCx, lock, aStream, rv);
     return;
   }
+  mAsyncWaitWorkerRef = mWorkerRef;
 
   // All good.
 }
@@ -429,6 +431,7 @@ BodyStream::OnInputStreamReady(nsIAsyncInputStream* aStream)
     NO_THREAD_SAFETY_ANALYSIS {
   AssertIsOnOwningThread();
   MOZ_DIAGNOSTIC_ASSERT(aStream);
+  mAsyncWaitWorkerRef = nullptr;
 
   // Acquire |mMutex| in order to safely inspect |mState| and use |mGlobal|.
   Maybe<MutexSingleWriterAutoLock> lock;
@@ -581,7 +584,7 @@ void BodyStream::ReleaseObjects(const MutexSingleWriterAutoLock& aProofOfLock) {
     // Let's dispatch a WorkerControlRunnable if the owning thread is a worker.
     if (mWorkerRef) {
       RefPtr<WorkerShutdown> r =
-          new WorkerShutdown(mWorkerRef->GetUnsafePrivate(), this);
+          new WorkerShutdown(mWorkerRef->Private(), this);
       Unused << NS_WARN_IF(!r->Dispatch());
       return;
     }

dom/base/BodyStream.h

@@ -30,7 +30,7 @@ class ErrorResult;
 namespace dom {
 
 class BodyStream;
-class WeakWorkerRef;
+class StrongWorkerRef;
 class ReadableStream;
 class ReadableStreamController;
 
@@ -200,7 +200,8 @@ class BodyStream final : public nsIInputStreamCallback,
   nsCOMPtr<nsIInputStream> mOriginalInputStream;
   nsCOMPtr<nsIAsyncInputStream> mInputStream;
 
-  RefPtr<WeakWorkerRef> mWorkerRef;
+  RefPtr<StrongWorkerRef> mWorkerRef;
+  RefPtr<StrongWorkerRef> mAsyncWaitWorkerRef;
 };
 
 }  // namespace dom

dom/fetch/FetchStreamReader.cpp

@@ -76,16 +76,14 @@ nsresult FetchStreamReader::Create(JSContext* aCx, nsIGlobalObject* aGlobal,
     WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
     MOZ_ASSERT(workerPrivate);
 
-    RefPtr<WeakWorkerRef> workerRef =
-        WeakWorkerRef::Create(workerPrivate, [streamReader]() {
+    RefPtr<StrongWorkerRef> workerRef = StrongWorkerRef::Create(
+        workerPrivate, "FetchStreamReader", [streamReader]() {
           MOZ_ASSERT(streamReader);
           MOZ_ASSERT(streamReader->mWorkerRef);
 
-          WorkerPrivate* workerPrivate = streamReader->mWorkerRef->GetPrivate();
-          MOZ_ASSERT(workerPrivate);
-
-          streamReader->CloseAndRelease(workerPrivate->GetJSContext(),
-                                        NS_ERROR_DOM_INVALID_STATE_ERR);
+          streamReader->CloseAndRelease(
+              streamReader->mWorkerRef->Private()->GetJSContext(),
+              NS_ERROR_DOM_INVALID_STATE_ERR);
         });
 
     if (NS_WARN_IF(!workerRef)) {
@@ -194,6 +192,7 @@ void FetchStreamReader::StartConsuming(JSContext* aCx, ReadableStream* aStream,
   if (NS_WARN_IF(aRv.Failed())) {
     return;
   }
+  mAsyncWaitWorkerRef = mWorkerRef;
 }
 
 struct FetchReadRequest : public ReadRequest {
@@ -236,6 +235,7 @@ MOZ_CAN_RUN_SCRIPT_BOUNDARY
 NS_IMETHODIMP
 FetchStreamReader::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
   NS_ASSERT_OWNINGTHREAD(FetchStreamReader);
+  mAsyncWaitWorkerRef = nullptr;
   if (mStreamClosed) {
     return NS_OK;
   }

dom/fetch/FetchStreamReader.h

@@ -19,7 +19,7 @@ namespace mozilla::dom {
 
 class ReadableStream;
 class ReadableStreamDefaultReader;
-class WeakWorkerRef;
+class StrongWorkerRef;
 
 class FetchStreamReader final : public nsIOutputStreamCallback {
  public:
@@ -66,7 +66,8 @@ class FetchStreamReader final : public nsIOutputStreamCallback {
 
   nsCOMPtr<nsIAsyncOutputStream> mPipeOut;
 
-  RefPtr<WeakWorkerRef> mWorkerRef;
+  RefPtr<StrongWorkerRef> mWorkerRef;
+  RefPtr<StrongWorkerRef> mAsyncWaitWorkerRef;
 
   RefPtr<ReadableStreamDefaultReader> mReader;
 

dom/fetch/FetchUtil.cpp

@@ -295,14 +295,13 @@ class WorkerStreamOwner final {
     RefPtr<WorkerStreamOwner> self =
         new WorkerStreamOwner(aStream, std::move(target));
 
-    self->mWorkerRef = WeakWorkerRef::Create(aWorker, [self]() {
+    self->mWorkerRef = StrongWorkerRef::Create(aWorker, "JSStreamConsumer", [self]() {
       if (self->mStream) {
         // If this Close() calls JSStreamConsumer::OnInputStreamReady and drops
         // the last reference to the JSStreamConsumer, 'this' will not be
         // destroyed since ~JSStreamConsumer() only enqueues a release proxy.
         self->mStream->Close();
         self->mStream = nullptr;
-        self->mWorkerRef = nullptr;
       }
     });
 
@@ -326,7 +325,7 @@ class WorkerStreamOwner final {
   // Read from any thread but only set/cleared on the worker thread. The
   // lifecycle of WorkerStreamOwner prevents concurrent read/clear.
   nsCOMPtr<nsIAsyncInputStream> mStream;
-  RefPtr<WeakWorkerRef> mWorkerRef;
+  RefPtr<StrongWorkerRef> mWorkerRef;
   nsCOMPtr<nsIEventTarget> mOwningEventTarget;
 };
 

dom/file/FileReader.cpp

@@ -624,15 +624,15 @@ FileReader::Notify(nsITimer* aTimer) {
 // InputStreamCallback
 NS_IMETHODIMP
 FileReader::OnInputStreamReady(nsIAsyncInputStream* aStream) {
-  if (mReadyState != LOADING || aStream != mAsyncStream) {
-    return NS_OK;
-  }
-
   // We use this class to decrease the busy counter at the end of this method.
   // In theory we can do it immediatelly but, for debugging reasons, we want to
   // be 100% sure we have a workerRef when OnLoadEnd() is called.
   FileReaderDecreaseBusyCounter RAII(this);
 
+  if (mReadyState != LOADING || aStream != mAsyncStream) {
+    return NS_OK;
+  }
+
   uint64_t count;
   nsresult rv = aStream->Available(&count);
 
@@ -731,14 +731,7 @@ void FileReader::Abort() {
 
   MOZ_ASSERT(mReadyState == LOADING);
 
-  ClearProgressEventTimer();
-
-  if (mAsyncWaitRunnable) {
-    mAsyncWaitRunnable->Cancel();
-    mAsyncWaitRunnable = nullptr;
-  }
-
-  mReadyState = DONE;
+  Shutdown();
 
   // XXX The spec doesn't say this
   mError = DOMException::Create(NS_ERROR_DOM_ABORT_ERR);
@@ -747,30 +740,12 @@ void FileReader::Abort() {
   SetDOMStringToNull(mResult);
   mResultArrayBuffer = nullptr;
 
-  // If we have the stream and the busy-count is not 0, it means that we are
-  // waiting for an OnInputStreamReady() call. Let's abort the current
-  // AsyncWait() calling it again with a nullptr callback. See
-  // nsIAsyncInputStream.idl.
-  if (mAsyncStream && mBusyCount) {
-    mAsyncStream->AsyncWait(/* callback */ nullptr,
-                            /* aFlags*/ 0,
-                            /* aRequestedCount */ 0, mTarget);
-    DecreaseBusyCounter();
-    MOZ_ASSERT(mBusyCount == 0);
-
-    mAsyncStream->Close();
-  }
-
-  mAsyncStream = nullptr;
   mBlob = nullptr;
 
-  // Clean up memory buffer
-  FreeFileData();
-
   // Dispatch the events
   DispatchProgressEvent(nsLiteralString(ABORT_STR));
   DispatchProgressEvent(nsLiteralString(LOADEND_STR));
-}  // namespace dom
+}
 
 nsresult FileReader::IncreaseBusyCounter() {
   if (mWeakWorkerRef && mBusyCount++ == 0) {
@@ -817,10 +792,8 @@ void FileReader::Shutdown() {
   FreeFileData();
   mResultArrayBuffer = nullptr;
 
-  if (mWeakWorkerRef && mBusyCount != 0) {
-    mStrongWorkerRef = nullptr;
+  if (mWeakWorkerRef) {
     mWeakWorkerRef = nullptr;
-    mBusyCount = 0;
   }
 }