fix(worker): isolate and wrapper lifetime fixes around startup and teardown (#478)

* fix(worker): isolate and wrapper lifetime fixes around startup and teardown

A throwing onclose run from the entry script read the worker isolate before it was published and crashed. Terminate() used the worker isolate with no synchronization against the worker thread deleting its runtime. EndWrapperLifetime kept using the wrapper after dispatching nsworkerended, whose listeners can shut the runtime down and delete it.

* fix(worker): serialize the message queue's run loop source with its teardown

Push() and Signal() read the run loop source and run loop with no lock while Terminate(), on another thread, invalidates and releases the source. They now go through the mutex Terminate() holds, and the worker thread terminates its queue before it leaves, since the queue borrows that thread's run loop. Also removes the unused GetWorkerIsolate() getter, which handed out the isolate without the mutex that now guards it.

* refactor(worker): rename the queue's run loop mutex and scope the isolate lock

initializationMutex_ guards the run loop source and run loop for the queue's whole life now that Signal() takes it, so it is named runLoopMutex_. Terminate() holds workerIsolateMutex_ in a scoped block, like the inspectorMutex_ block next to it, instead of a unique_lock released by hand. The comment in Push() said Terminate() flips the terminated flag under the queue mutex; it sets it under the run loop mutex, before emptying the queue.

---------

Co-authored-by: Eduardo Speroni <edusperoni@gmail.com>

Author: adrian-niculescu

NativeScript/runtime/ConcurrentQueue.cpp

@@ -4,33 +4,29 @@
 namespace tns {
 
 void ConcurrentQueue::Initialize(CFRunLoopRef runLoop, void (*performWork)(void*), void* info) {
-    std::unique_lock<std::mutex> lock(initializationMutex_);
-    if (terminated) {
-        return;
-    }
+  std::unique_lock<std::mutex> lock(runLoopMutex_);
+  if (terminated) {
+    return;
+  }
     this->runLoop_ = runLoop;
     CFRunLoopSourceContext sourceContext = { 0, info, 0, 0, 0, 0, 0, 0, 0, performWork };
     this->runLoopTasksSource_ = CFRunLoopSourceCreate(kCFAllocatorDefault, 0, &sourceContext);
     CFRunLoopAddSource(this->runLoop_, this->runLoopTasksSource_, kCFRunLoopCommonModes);
 }
 
 void ConcurrentQueue::Push(std::shared_ptr<worker::Message> message) {
-    if (this->runLoopTasksSource_ != nullptr && !CFRunLoopSourceIsValid(this->runLoopTasksSource_)) {
-        return;
-    }
-
     {
-      // Checked under the queue mutex, where Terminate() also flips it while
-      // emptying the queue: a push that loses the race is dropped rather than
-      // landing in a queue nothing will ever pop again.
+      // Terminate() sets the flag before it empties the queue under this
+      // mutex, so a push either sees it here or lands ahead of that sweep:
+      // never in a queue nothing will pop again.
       std::unique_lock<std::mutex> mlock(this->mutex_);
       if (this->terminated) {
         return;
       }
         this->messagesQueue_.push(message);
     }
 
-    this->SignalAndWakeUp();
+    this->Signal();
 }
 
 std::vector<std::shared_ptr<worker::Message>> ConcurrentQueue::PopAll() {
@@ -52,20 +48,14 @@ bool ConcurrentQueue::IsEmpty() {
 }
 
 void ConcurrentQueue::Signal() {
-  // Mirrors Push()'s validity handling instead of SignalAndWakeUp()'s
-  // assert: a retry racing Terminate() must be a silent no-op.
-  if (this->runLoopTasksSource_ == nullptr ||
-      !CFRunLoopSourceIsValid(this->runLoopTasksSource_)) {
-    return;
+  // Serializes signaling and waking with Initialize() and Terminate().
+  // Terminate() clears both pointers and invalidates and releases the source;
+  // the run loop is borrowed from the worker thread, which terminates the
+  // queue before it leaves.
+  std::unique_lock<std::mutex> lock(runLoopMutex_);
+  if (this->runLoopTasksSource_ != nullptr) {
+    CFRunLoopSourceSignal(this->runLoopTasksSource_);
   }
-  this->SignalAndWakeUp();
-}
-
-void ConcurrentQueue::SignalAndWakeUp() {
-    if (this->runLoopTasksSource_ != nullptr) {
-        tns::Assert(CFRunLoopSourceIsValid(this->runLoopTasksSource_));
-        CFRunLoopSourceSignal(this->runLoopTasksSource_);
-    }
 
     if (this->runLoop_ != nullptr) {
         CFRunLoopWakeUp(this->runLoop_);
@@ -78,7 +68,7 @@ void ConcurrentQueue::Terminate() {
   // its sibling group's lock and posts to the sibling's loop.
   std::queue<std::shared_ptr<worker::Message>> dropped;
   {
-    std::unique_lock<std::mutex> lock(initializationMutex_);
+    std::unique_lock<std::mutex> lock(runLoopMutex_);
     terminated = true;
     CFRunLoopRef runLoop = this->runLoop_;
     CFRunLoopSourceRef source = this->runLoopTasksSource_;

NativeScript/runtime/ConcurrentQueue.h

@@ -31,8 +31,7 @@ struct ConcurrentQueue {
     CFRunLoopRef runLoop_ = nullptr;
     std::atomic<bool> terminated{false};
     std::mutex mutex_;
-    std::mutex initializationMutex_;
-    void SignalAndWakeUp();
+    std::mutex runLoopMutex_;
 };
 
 }

NativeScript/runtime/DataWrapper.h

@@ -553,12 +553,16 @@ class WorkerWrapper : public BaseDataWrapper {
   void Start(std::shared_ptr<v8::Persistent<v8::Value>> poWorker,
              std::function<v8::Isolate*()> func,
              std::optional<int> qualityOfService = std::nullopt);
-  void CallOnErrorHandlers(v8::TryCatch& tc);
+  // Both reporters take the isolate from their caller, which is running on
+  // it: they are reachable while the entry script is still evaluating, before
+  // workerIsolate_ is published.
+  void CallOnErrorHandlers(v8::Isolate* isolate, v8::TryCatch& tc);
   // Reports a rejected entry-evaluation promise. A rejection carries a reason
   // rather than a TryCatch, so it cannot go through CallOnErrorHandlers, but it
   // follows the same web order: the worker scope's `onerror` first, then — only
   // if that did not handle it — the parent's Worker error event.
-  void ReportEntryEvaluationRejection(v8::Local<v8::Context> context,
+  void ReportEntryEvaluationRejection(v8::Isolate* isolate,
+                                      v8::Local<v8::Context> context,
                                       v8::Local<v8::Value> reason);
   void PassUncaughtExceptionFromWorkerToMain(v8::Local<v8::Context> context,
                                              v8::TryCatch& tc,
@@ -622,7 +626,6 @@ class WorkerWrapper : public BaseDataWrapper {
   const inline v8::Isolate* GetMainIsolate() { return mainIsolate_; }
   // The only route from the worker thread to the parent: see mainLoop_.
   std::weak_ptr<EventLoop> MainLoop() const { return mainLoop_; }
-  const inline v8::Isolate* GetWorkerIsolate() { return workerIsolate_; }
 
   // Deletion is decided by who still holds the wrapper. The parent's Worker
   // object holds it from construction; Start() adds the worker thread. Each
@@ -647,7 +650,11 @@ class WorkerWrapper : public BaseDataWrapper {
   enum class Holders : uint8_t { Parent, Both, WorkerThread };
 
   v8::Isolate* mainIsolate_;
+  // Written by the worker thread only: published once the worker's startup
+  // function returns, withdrawn before the worker's runtime is deleted. Any
+  // other thread reads and uses it under workerIsolateMutex_.
   v8::Isolate* workerIsolate_;
+  std::mutex workerIsolateMutex_;
   std::atomic<bool> isRunning_;
   std::atomic<bool> isClosing_;
   std::atomic<bool> isTerminating_;

NativeScript/runtime/Worker.mm

@@ -564,7 +564,7 @@ throw NativeScriptException(
                                       ? info[0]
                                       : Local<Value>(v8::Exception::Error(tns::ToV8String(
                                             iso, "Worker entry module evaluation rejected")));
-            w->ReportEntryEvaluationRejection(ctx, reason);
+            w->ReportEntryEvaluationRejection(iso, ctx, reason);
           };
           Local<v8::Function> onFulfilled;
           Local<v8::Function> onRejected;
@@ -798,7 +798,7 @@ throw NativeScriptException(
     TryCatch tc(isolate);
     success = onCloseFunc->Call(context, v8::Undefined(isolate), 0, args).ToLocal(&result);
     if (!success && tc.HasCaught()) {
-      worker->CallOnErrorHandlers(tc);
+      worker->CallOnErrorHandlers(isolate, tc);
     }
   }
 }

NativeScript/runtime/WorkerWrapper.mm

@@ -122,17 +122,26 @@ static void PostToLoop(const std::shared_ptr<EventLoop>& loop, std::function<voi
 }
 
 void WorkerWrapper::EndWrapperLifetime() {
-  Local<Value> worker =
-      this->poWorker_ != nullptr ? this->poWorker_->Get(this->mainIsolate_) : Local<Value>();
+  // The dispatch below runs listeners, and a listener may shut the runtime
+  // down, whose teardown deletes this wrapper. Everything the dispatch needs is
+  // read first, and the liveness token says afterwards whether `this` is still
+  // there to unroot.
+  Isolate* isolate = this->mainIsolate_;
+  std::shared_ptr<std::atomic<WorkerWrapper*>> selfRef = this->selfRef_;
+  Local<Value> worker = this->poWorker_ != nullptr ? this->poWorker_->Get(isolate) : Local<Value>();
   if (!worker.IsEmpty() && worker->IsObject()) {
-    TryCatch tc(this->mainIsolate_);
-    Worker::EmitEnded(this->mainIsolate_, worker.As<Object>());
+    TryCatch tc(isolate);
+    Worker::EmitEnded(isolate, worker.As<Object>());
     if (tc.HasCaught()) {
       Local<Value> error = tc.Exception();
-      Log(@"%s", tns::ToString(this->mainIsolate_, error).c_str());
-      this->mainIsolate_->ThrowException(error);
+      Log(@"%s", tns::ToString(isolate, error).c_str());
+      isolate->ThrowException(error);
     }
   }
+  if (selfRef->load(std::memory_order_acquire) == nullptr) {
+    // Deleted during the dispatch; that teardown released the Worker object.
+    return;
+  }
   this->UnrootWorkerObject();
 }
 
@@ -183,7 +192,7 @@ static void PostToLoop(const std::shared_ptr<EventLoop>& loop, std::function<voi
     this->onMessage_(this->workerIsolate_, globalTarget, message);
 
     if (tc.HasCaught()) {
-      this->CallOnErrorHandlers(tc);
+      this->CallOnErrorHandlers(this->workerIsolate_, tc);
     }
   }
 
@@ -234,7 +243,11 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
         },
         this);
 
-    this->workerIsolate_ = func();
+    Isolate* workerIsolate = func();
+    {
+      std::lock_guard<std::mutex> lock(this->workerIsolateMutex_);
+      this->workerIsolate_ = workerIsolate;
+    }
 
     this->DrainPendingTasks();
 
@@ -244,6 +257,20 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
     }
   }
 
+  // The queue borrows this thread's run loop, so it lets go of it before the
+  // thread leaves: a terminate() that made this thread skip the loop above
+  // only reaches its own queue_.Terminate() later, possibly after the thread
+  // and its run loop are gone. A second Terminate() finds nothing set.
+  this->queue_.Terminate();
+
+  // Withdrawn before the runtime and its isolate go away below. Terminate()
+  // uses the isolate under this mutex, so a terminate that already read it has
+  // finished with it by the time this returns, and a later one finds null.
+  {
+    std::lock_guard<std::mutex> lock(this->workerIsolateMutex_);
+    this->workerIsolate_ = nullptr;
+  }
+
   // The inspector must be gone before the Runtime (and with it the isolate)
   // is deleted below.
   this->DestroyInspector();
@@ -315,20 +342,25 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
   // set terminating to true atomically
   bool wasTerminating = this->isTerminating_.exchange(true);
   if (!wasTerminating) {
-    if (this->workerIsolate_ != nullptr) {
-      // Flagged before the request so a pump that is between iterations sees
-      // it on its next check, rather than only once V8 has some JS to
-      // interrupt — which a parked graph never provides.
-      //
-      // NOTE: `workerIsolate_` is assigned only after the worker's ENTRY has
-      // finished evaluating, so a worker still parked in its entry is not
-      // reachable from here at all and terminate() does nothing for it. That
-      // is a pre-existing worker-lifecycle gap, not something this flag can
-      // close — see the follow-up filed for it.
-      if (Runtime* workerRuntime = Runtime::GetRuntime(this->workerIsolate_)) {
-        workerRuntime->RequestTermination();
+    {
+      // Held across the use, not just the read: the worker thread withdraws
+      // the isolate under the same mutex before deleting its runtime.
+      std::lock_guard<std::mutex> lock(this->workerIsolateMutex_);
+      if (this->workerIsolate_ != nullptr) {
+        // Flagged before the request so a pump that is between iterations sees
+        // it on its next check, rather than only once V8 has some JS to
+        // interrupt — which a parked graph never provides.
+        //
+        // NOTE: `workerIsolate_` is assigned only after the worker's ENTRY has
+        // finished evaluating, so a worker still parked in its entry is not
+        // reachable from here at all and terminate() does nothing for it. That
+        // is a pre-existing worker-lifecycle gap, not something this flag can
+        // close — see the follow-up filed for it.
+        if (Runtime* workerRuntime = Runtime::GetRuntime(this->workerIsolate_)) {
+          workerRuntime->RequestTermination();
+        }
+        this->workerIsolate_->TerminateExecution();
       }
-      this->workerIsolate_->TerminateExecution();
     }
     {
       // A worker paused at a breakpoint sits in the inspector's nested pause
@@ -437,11 +469,10 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
   delete client;
 }
 
-void WorkerWrapper::CallOnErrorHandlers(TryCatch& tc) {
+void WorkerWrapper::CallOnErrorHandlers(Isolate* isolate, TryCatch& tc) {
   if (this->isTerminating_) {
     return;
   }
-  Isolate* isolate = this->workerIsolate_;
   Local<Context> context = Caches::Get(isolate)->GetContext();
   Local<Object> global = context->Global();
 
@@ -470,11 +501,11 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
   this->PassUncaughtExceptionFromWorkerToMain(context, tc);
 }
 
-void WorkerWrapper::ReportEntryEvaluationRejection(Local<Context> context, Local<Value> reason) {
+void WorkerWrapper::ReportEntryEvaluationRejection(Isolate* isolate, Local<Context> context,
+                                                   Local<Value> reason) {
   if (this->isTerminating_) {
     return;
   }
-  Isolate* isolate = this->workerIsolate_;
   Local<Object> global = context->Global();
 
   Local<Value> onErrorVal;

TestRunner/app/tests/MessagingTests.js

@@ -252,6 +252,16 @@ describe("Messaging runtime edges", function () {
                 }
             };
         });
+
+        it("reports an error onclose threw while the entry script was still running", function (done) {
+            var worker = new Worker("./messaging/throwingOncloseWorker.js");
+            worker.onerror = function (event) {
+                event.preventDefault();
+                expect(event.message).toContain("boom from onclose");
+                worker.terminate();
+                done();
+            };
+        });
     });
 
     describe("AbortSignal handler attribute accounting", function () {

TestRunner/app/tests/messaging/throwingOncloseWorker.js

@@ -0,0 +1,6 @@
+// Closes from inside the entry script, so onclose runs before the entry has
+// finished evaluating.
+onclose = function () {
+    throw new Error("boom from onclose");
+};
+close();