fix: workers should gracefully shutdown on close()

NativeScript/runtime/ConcurrentQueue.cpp

@@ -54,14 +54,19 @@ void ConcurrentQueue::SignalAndWakeUp() {
 void ConcurrentQueue::Terminate() {
     std::unique_lock<std::mutex> lock(initializationMutex_);
     terminated = true;
-    if (this->runLoop_) {
-        CFRunLoopStop(this->runLoop_);
+    CFRunLoopRef runLoop = this->runLoop_;
+    CFRunLoopSourceRef source = this->runLoopTasksSource_;
+    this->runLoopTasksSource_ = nullptr;
+    this->runLoop_ = nullptr;
+
+    if (runLoop) {
+        CFRunLoopStop(runLoop);
     }
 
-    if (this->runLoopTasksSource_) {
-        CFRunLoopRemoveSource(this->runLoop_, this->runLoopTasksSource_, kCFRunLoopCommonModes);
-        CFRunLoopSourceInvalidate(this->runLoopTasksSource_);
-        CFRelease(this->runLoopTasksSource_);
+    if (source) {
+        CFRunLoopRemoveSource(runLoop, source, kCFRunLoopCommonModes);
+        CFRunLoopSourceInvalidate(source);
+        CFRelease(source);
     }
 }
 

NativeScript/runtime/DataWrapper.h

@@ -690,11 +690,11 @@ class WorkerWrapper: public BaseDataWrapper {
 private:
     v8::Isolate* mainIsolate_;
     v8::Isolate* workerIsolate_;
-    bool isRunning_;
-    bool isClosing_;
+    std::atomic<bool> isRunning_;
+    std::atomic<bool> isClosing_;
     std::atomic<bool> isTerminating_;
-    bool isDisposed_;
-    bool isWeak_;
+    std::atomic<bool> isDisposed_;
+    std::atomic<bool> isWeak_;
     std::function<void (v8::Isolate*, v8::Local<v8::Object> thiz, std::shared_ptr<worker::Message>)> onMessage_;
     std::shared_ptr<v8::Persistent<v8::Value>> poWorker_;
     ConcurrentQueue queue_;

NativeScript/runtime/Runtime.mm

@@ -198,8 +198,6 @@ void DisposeIsolateWhenPossible(Isolate* isolate) {
   }
   this->isolate_->TerminateExecution();
 
-  // TODO: fix race condition on workers where a queue can leak (maybe calling Terminate before
-  // Initialize?)
   Caches::Workers->ForEach([currentIsolate](int& key, std::shared_ptr<Caches::WorkerState>& value) {
     auto childWorkerWrapper = static_cast<WorkerWrapper*>(value->UserData());
     if (childWorkerWrapper->GetMainIsolate() == currentIsolate) {

NativeScript/runtime/WorkerWrapper.mm

@@ -40,7 +40,7 @@
 const int WorkerWrapper::WorkerId() { return this->workerId_; }
 
 void WorkerWrapper::PostMessage(std::shared_ptr<worker::Message> message) {
-  if (!this->isTerminating_) {
+  if (!this->isTerminating_ && !this->isClosing_) {
     this->queue_.Push(message);
   }
 }
@@ -66,7 +66,7 @@
   Local<Object> global = context->Global();
 
   for (std::shared_ptr<worker::Message> message : messages) {
-    if (this->isTerminating_) {
+    if (this->isTerminating_ || this->isClosing_) {
       break;
     }
     TryCatch tc(this->workerIsolate_);
@@ -76,6 +76,14 @@
       this->CallOnErrorHandlers(tc);
     }
   }
+
+  if (this->isClosing_) {
+    bool wasTerminating = this->isTerminating_.exchange(true);
+    if (!wasTerminating) {
+      this->queue_.Terminate();
+      this->isRunning_ = false;
+    }
+  }
 }
 
 void WorkerWrapper::BackgroundLooper(std::function<Isolate*()> func) {
@@ -101,7 +109,18 @@
 
   this->isDisposed_ = true;
   Runtime* runtime = Runtime::GetCurrentRuntime();
-  delete runtime;
+  if (runtime != nullptr) {
+    delete runtime;
+  } else {
+    // Runtime was never created (worker terminated before initialization).
+    // The runtime destructor normally handles this cleanup, so do it here.
+    int workerId = this->workerId_;
+    bool found;
+    auto state = Caches::Workers->Get(workerId, found);
+    if (found) {
+      Caches::Workers->Remove(workerId);
+    }
+  }
 }
 
 void WorkerWrapper::Close() { this->isClosing_ = true; }

TestRunner/app/tests/shared/Workers/index.js

@@ -381,6 +381,28 @@ describe("TNS Workers", () => {
         }, DEFAULT_TIMEOUT_BEFORE_ASSERT);
     });
 
+    it("Worker should fully shut down after close() without needing terminate()", (done) => {
+        var worker = new Worker("./tests/shared/Workers/EvalWorker.js");
+
+        worker.postMessage({
+            eval: "close(); postMessage('closing');"
+        });
+
+        var responseCounter = 0;
+        worker.onmessage = (msg) => {
+            responseCounter++;
+        };
+
+        setTimeout(() => {
+            worker.postMessage({ eval: "postMessage('should not arrive');" });
+        }, 500);
+
+        setTimeout(() => {
+            expect(responseCounter).toBe(1);
+            done();
+        }, DEFAULT_TIMEOUT_BEFORE_ASSERT + 1000);
+    });
+
     it("Test onerror invoked for a script that has invalid syntax", (done) => {
         var worker = new Worker("./tests/shared/Workers/WorkerInvalidSyntax.js");