fix(runtime/signal): use op_async_unref for op_signal_poll (#11097)

cli/tests/raw_mode_cbreak.ts

@@ -5,11 +5,13 @@ const signal = Deno.signals.interrupt();
 
 Deno.stdout.writeSync(new TextEncoder().encode("S"));
 
-await signal;
+signal.then(() => {
+  Deno.stdout.writeSync(new TextEncoder().encode("A"));
 
-Deno.stdout.writeSync(new TextEncoder().encode("A"));
+  signal.dispose();
 
-signal.dispose();
+  Deno.setRaw(0, false); // restores old mode.
+  Deno.setRaw(0, false); // Can be safely called multiple times
+});
 
-Deno.setRaw(0, false); // restores old mode.
+setTimeout(() => {}, 10000); // Keep the program running
-Deno.setRaw(0, false); // Can be safely called multiple times

cli/tests/unit/signal_test.ts

@@ -129,6 +129,24 @@ unitTest(
   },
 );
 
+// This tests that pending op_signal_poll doesn't block the runtime from exiting the process.
+unitTest(
+  { ignore: Deno.build.os === "windows", perms: { run: true, read: true } },
+  async function signalStreamExitTest(): Promise<void> {
+    const p = Deno.run({
+      cmd: [
+        Deno.execPath(),
+        "eval",
+        "--unstable",
+        "(async () => { for await (const _ of Deno.signals.io()) {} })()",
+      ],
+    });
+    const res = await p.status();
+    assertEquals(res.code, 0);
+    p.close();
+  },
+);
+
 unitTest(
   { ignore: Deno.build.os === "windows", perms: { run: true } },
   async function signalPromiseTest(): Promise<void> {

core/lib.rs

@@ -73,6 +73,7 @@ pub use crate::ops_builtin::op_close;
 pub use crate::ops_builtin::op_print;
 pub use crate::ops_builtin::op_resources;
 pub use crate::ops_json::op_async;
+pub use crate::ops_json::op_async_unref;
 pub use crate::ops_json::op_sync;
 pub use crate::resources::Resource;
 pub use crate::resources::ResourceId;

core/ops_json.rs

@@ -51,6 +51,9 @@ where
 
 /// Creates an op that passes data asynchronously using JSON.
 ///
+/// When this op is dispatched, the runtime doesn't exit while processing it.
+/// Use op_async_unref instead if you want to make the runtime exit while processing it.
+///
 /// The provided function `op_fn` has the following parameters:
 /// * `Rc<RefCell<OpState>`: the op state, can be used to read/write resources in the runtime from an op.
 /// * `V`: the deserializable value that is passed to the Rust function.
@@ -99,6 +102,37 @@ where
   })
 }
 
+/// Creates an op that passes data asynchronously using JSON.
+///
+/// When this op is dispatched, the runtime still can exit while processing it.
+///
+/// The other usages are the same as `op_async`.
+pub fn op_async_unref<F, A, B, R, RV>(op_fn: F) -> Box<OpFn>
+where
+  F: Fn(Rc<RefCell<OpState>>, A, B) -> R + 'static,
+  A: DeserializeOwned,
+  B: DeserializeOwned,
+  R: Future<Output = Result<RV, AnyError>> + 'static,
+  RV: Serialize + 'static,
+{
+  Box::new(move |state, payload| -> Op {
+    let pid = payload.promise_id;
+    // Deserialize args, sync error on failure
+    let args = match payload.deserialize() {
+      Ok(args) => args,
+      Err(err) => {
+        return Op::Sync(serialize_op_result(Err::<(), AnyError>(err), state))
+      }
+    };
+    let (a, b) = args;
+
+    use crate::futures::FutureExt;
+    let fut = op_fn(state.clone(), a, b)
+      .map(move |result| (pid, serialize_op_result(result, state)));
+    Op::AsyncUnref(Box::pin(fut))
+  })
+}
+
 #[cfg(test)]
 mod tests {
   use super::*;

runtime/ops/signal.rs

@@ -1,6 +1,6 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
 use deno_core::error::AnyError;
-use deno_core::op_async;
+use deno_core::op_async_unref;
 use deno_core::op_sync;
 use deno_core::Extension;
 use deno_core::OpState;
@@ -31,7 +31,7 @@ pub fn init() -> Extension {
     .ops(vec![
       ("op_signal_bind", op_sync(op_signal_bind)),
       ("op_signal_unbind", op_sync(op_signal_unbind)),
-      ("op_signal_poll", op_async(op_signal_poll)),
+      ("op_signal_poll", op_async_unref(op_signal_poll)),
     ])
     .build()
 }