fix(ext/node): allow automatic worker_thread termination (#22647)

Co-authored-by: Matt Mastracci <matthew@mastracci.com>

cli/worker.rs

@@ -610,6 +610,7 @@ impl CliMainWorkerFactory {
         disable_deprecated_api_warning: shared.disable_deprecated_api_warning,
         verbose_deprecated_api_warning: shared.verbose_deprecated_api_warning,
         future: shared.enable_future_features,
+        close_on_idle: true,
       },
       extensions: custom_extensions,
       startup_snapshot: crate::js::deno_isolate_init(),
@@ -814,6 +815,7 @@ fn create_web_worker_callback(
         disable_deprecated_api_warning: shared.disable_deprecated_api_warning,
         verbose_deprecated_api_warning: shared.verbose_deprecated_api_warning,
         future: false,
+        close_on_idle: args.close_on_idle,
       },
       extensions: vec![],
       startup_snapshot: crate::js::deno_isolate_init(),
@@ -841,6 +843,8 @@ fn create_web_worker_callback(
       stdio: stdio.clone(),
       cache_storage_dir,
       feature_checker,
+      strace_ops: shared.options.strace_ops.clone(),
+      close_on_idle: args.close_on_idle,
       maybe_worker_metadata: args.maybe_worker_metadata,
     };
 

ext/node/polyfills/worker_threads.ts

@@ -211,6 +211,7 @@ class NodeWorker extends EventEmitter {
         permissions: null,
         name: this.#name,
         workerType: "module",
+        closeOnIdle: true,
       },
       serializedWorkerMetadata,
     );
@@ -413,7 +414,7 @@ internals.__initWorkerThreads = (
     >();
 
     parentPort = self as ParentPort;
-    if (typeof maybeWorkerMetadata !== "undefined") {
+    if (maybeWorkerMetadata) {
       const { 0: metadata, 1: _ } = maybeWorkerMetadata;
       workerData = metadata.workerData;
       environmentData = metadata.environmentData;

runtime/js/11_workers.js

@@ -46,6 +46,7 @@ function createWorker(
   permissions,
   name,
   workerType,
+  closeOnIdle,
 ) {
   return op_create_worker({
     hasSourceCode,
@@ -54,6 +55,7 @@ function createWorker(
     sourceCode,
     specifier,
     workerType,
+    closeOnIdle,
   });
 }
 
@@ -75,14 +77,6 @@ function hostRecvMessage(id) {
 
 const privateWorkerRef = Symbol();
 
-function refWorker(worker) {
-  worker[privateWorkerRef](true);
-}
-
-function unrefWorker(worker) {
-  worker[privateWorkerRef](false);
-}
-
 class Worker extends EventTarget {
   #id = 0;
   #name = "";
@@ -134,8 +128,9 @@ class Worker extends EventTarget {
       hasSourceCode,
       sourceCode,
       deno?.permissions,
-      name,
+      this.#name,
       workerType,
+      false,
     );
     this.#id = id;
     this.#pollControl();
@@ -325,4 +320,4 @@ webidl.converters["WorkerType"] = webidl.createEnumConverter("WorkerType", [
   "module",
 ]);
 
-export { refWorker, unrefWorker, Worker };
+export { Worker };

runtime/js/99_main.js

@@ -279,6 +279,7 @@ function postMessage(message, transferOrOptions = {}) {
 
 let isClosing = false;
 let globalDispatchEvent;
+let closeOnIdle;
 
 async function pollForMessages() {
   if (!globalDispatchEvent) {
@@ -288,7 +289,14 @@ async function pollForMessages() {
     );
   }
   while (!isClosing) {
-    const data = await op_worker_recv_message();
+    const op = op_worker_recv_message();
+    // In a Node.js worker, unref() the op promise to prevent it from
+    // keeping the event loop alive. This avoids the need to explicitly
+    // call self.close() or worker.terminate().
+    if (closeOnIdle) {
+      core.unrefOpPromise(op);
+    }
+    const data = await op;
     if (data === null) break;
     const v = messagePort.deserializeJsMessageData(data);
     const message = v[0];
@@ -803,6 +811,8 @@ function bootstrapWorkerRuntime(
     6: argv0,
     7: shouldDisableDeprecatedApiWarning,
     8: shouldUseVerboseDeprecatedApiWarning,
+    9: _future,
+    10: closeOnIdle_,
   } = runtimeOptions;
 
   deprecatedApiWarningDisabled = shouldDisableDeprecatedApiWarning;
@@ -864,6 +874,7 @@ function bootstrapWorkerRuntime(
 
   location.setLocationHref(location_);
 
+  closeOnIdle = closeOnIdle_;
   globalThis.pollForMessages = pollForMessages;
 
   // TODO(bartlomieju): deprecate --unstable

runtime/ops/worker_host.rs

@@ -35,6 +35,7 @@ pub struct CreateWebWorkerArgs {
   pub permissions: PermissionsContainer,
   pub main_module: ModuleSpecifier,
   pub worker_type: WebWorkerType,
+  pub close_on_idle: bool,
   pub maybe_worker_metadata: Option<JsMessageData>,
 }
 
@@ -114,6 +115,7 @@ pub struct CreateWorkerArgs {
   source_code: String,
   specifier: String,
   worker_type: WebWorkerType,
+  close_on_idle: bool,
 }
 
 /// Create worker as the host
@@ -191,6 +193,7 @@ fn op_create_worker(
         permissions: worker_permissions,
         main_module: module_specifier.clone(),
         worker_type,
+        close_on_idle: args.close_on_idle,
         maybe_worker_metadata,
       });
 

runtime/web_worker.rs

@@ -5,6 +5,7 @@ use crate::permissions::PermissionsContainer;
 use crate::shared::maybe_transpile_source;
 use crate::shared::runtime;
 use crate::tokio_util::create_and_run_current_thread;
+use crate::worker::create_op_metrics;
 use crate::worker::import_meta_resolve_callback;
 use crate::worker::validate_import_attributes_callback;
 use crate::worker::FormatJsErrorFn;
@@ -34,7 +35,6 @@ use deno_core::ModuleCodeString;
 use deno_core::ModuleId;
 use deno_core::ModuleLoader;
 use deno_core::ModuleSpecifier;
-use deno_core::OpMetricsSummaryTracker;
 use deno_core::PollEventLoopOptions;
 use deno_core::RuntimeOptions;
 use deno_core::SharedArrayBufferStore;
@@ -327,6 +327,7 @@ pub struct WebWorker {
   id: WorkerId,
   pub js_runtime: JsRuntime,
   pub name: String,
+  close_on_idle: bool,
   internal_handle: WebWorkerInternalHandle,
   pub worker_type: WebWorkerType,
   pub main_module: ModuleSpecifier,
@@ -359,6 +360,8 @@ pub struct WebWorkerOptions {
   pub cache_storage_dir: Option<std::path::PathBuf>,
   pub stdio: Stdio,
   pub feature_checker: Arc<FeatureChecker>,
+  pub strace_ops: Option<Vec<String>>,
+  pub close_on_idle: bool,
   pub maybe_worker_metadata: Option<JsMessageData>,
 }
 
@@ -511,17 +514,11 @@ impl WebWorker {
     #[cfg(feature = "only_snapshotted_js_sources")]
     options.startup_snapshot.as_ref().expect("A user snapshot was not provided, even though 'only_snapshotted_js_sources' is used.");
 
-    // Hook up the summary metrics if the user or subcommand requested them
-    let (op_summary_metrics, op_metrics_factory_fn) =
-      if options.bootstrap.enable_op_summary_metrics {
-        let op_summary_metrics = Rc::new(OpMetricsSummaryTracker::default());
-        (
-          Some(op_summary_metrics.clone()),
-          Some(op_summary_metrics.op_metrics_factory_fn(|_| true)),
-        )
-      } else {
-        (None, None)
-      };
+    // Get our op metrics
+    let (op_summary_metrics, op_metrics_factory_fn) = create_op_metrics(
+      options.bootstrap.enable_op_summary_metrics,
+      options.strace_ops,
+    );
 
     let mut js_runtime = JsRuntime::new(RuntimeOptions {
       module_loader: Some(options.module_loader.clone()),
@@ -606,6 +603,7 @@ impl WebWorker {
         main_module,
         poll_for_messages_fn: None,
         bootstrap_fn_global: Some(bootstrap_fn_global),
+        close_on_idle: options.close_on_idle,
         maybe_worker_metadata: options.maybe_worker_metadata,
       },
       external_handle,
@@ -759,6 +757,10 @@ impl WebWorker {
           return Poll::Ready(Err(e));
         }
 
+        if self.close_on_idle {
+          return Poll::Ready(Ok(()));
+        }
+
         // TODO(mmastrac): we don't want to test this w/classic workers because
         // WPT triggers a failure here. This is only exposed via --enable-testing-features-do-not-use.
         if self.worker_type == WebWorkerType::Module {

runtime/worker.rs

@@ -227,7 +227,7 @@ impl Default for WorkerOptions {
   }
 }
 
-fn create_op_metrics(
+pub fn create_op_metrics(
   enable_op_summary_metrics: bool,
   strace_ops: Option<Vec<String>>,
 ) -> (

runtime/worker_bootstrap.rs

@@ -63,6 +63,7 @@ pub struct BootstrapOptions {
   pub disable_deprecated_api_warning: bool,
   pub verbose_deprecated_api_warning: bool,
   pub future: bool,
+  pub close_on_idle: bool,
 }
 
 impl Default for BootstrapOptions {
@@ -94,6 +95,7 @@ impl Default for BootstrapOptions {
       disable_deprecated_api_warning: false,
       verbose_deprecated_api_warning: false,
       future: false,
+      close_on_idle: false,
     }
   }
 }
@@ -129,6 +131,8 @@ struct BootstrapV8<'a>(
   bool,
   // future
   bool,
+  // close_on_idle
+  bool,
 );
 
 impl BootstrapOptions {
@@ -151,6 +155,7 @@ impl BootstrapOptions {
       self.disable_deprecated_api_warning,
       self.verbose_deprecated_api_warning,
       self.future,
+      self.close_on_idle,
     );
 
     bootstrap.serialize(ser).unwrap()

tests/integration/worker_tests.rs

@@ -111,3 +111,10 @@ itest!(worker_doest_stall_event_loop {
   output: "workers/worker_doest_stall_event_loop.ts.out",
   exit_code: 0,
 });
+
+// Test for https://github.com/denoland/deno/issues/22629
+itest!(node_worker_auto_exits {
+  args: "run --quiet --allow-read workers/node_worker_auto_exits.mjs",
+  output: "workers/node_worker_auto_exits.mjs.out",
+  exit_code: 0,
+});

tests/testdata/workers/node_worker_auto_exits.mjs

@@ -0,0 +1,9 @@
+import { isMainThread, Worker } from "node:worker_threads";
+
+if (isMainThread) {
+  // This re-loads the current file inside a Worker instance.
+  const w = new Worker(import.meta.filename);
+} else {
+  console.log("Inside Worker!");
+  console.log(isMainThread); // Prints 'false'.
+}

tests/testdata/workers/node_worker_auto_exits.mjs.out

@@ -0,0 +1,2 @@
+Inside Worker!
+false