feat: add AsyncUnref ops (denoland#3721)

This is in order to support features like signal handlers, which
shouldn't prevent the program from exiting.

cli/ops/dispatch_json.rs

@@ -14,6 +14,9 @@ pub type AsyncJsonOp =
 pub enum JsonOp {
  Sync(Value),
  Async(AsyncJsonOp),
+ /// AsyncUnref is the variation of Async, which doesn't block the program
+ /// exiting.
+ AsyncUnref(AsyncJsonOp),
 }
 
 fn json_err(err: ErrBox) -> Value {
@@ -77,6 +80,13 @@ where
  });
  CoreOp::Async(fut2.boxed())
  }
+ Ok(JsonOp::AsyncUnref(fut)) => {
+ assert!(promise_id.is_some());
+ let fut2 = fut.then(move |result| {
+ futures::future::ok(serialize_result(promise_id, result))
+ });
+ CoreOp::AsyncUnref(fut2.boxed())
+ }
  Err(sync_err) => {
  let buf = serialize_result(promise_id, Err(sync_err));
  if is_sync {

cli/state.rs

@@ -110,6 +110,14 @@ impl ThreadSafeState {
  });
  Op::Async(result_fut.boxed())
  }
+ Op::AsyncUnref(fut) => {
+ let state = state.clone();
+ let result_fut = fut.map_ok(move |buf: Buf| {
+ state.metrics_op_completed(buf.len());
+ buf
+ });
+ Op::AsyncUnref(result_fut.boxed())
+ }
  }
  }
  }

core/isolate.rs

@@ -16,6 +16,7 @@ use crate::shared_queue::SharedQueue;
 use crate::shared_queue::RECOMMENDED_SIZE;
 use futures::future::FutureExt;
 use futures::future::TryFutureExt;
+use futures::stream::select;
 use futures::stream::FuturesUnordered;
 use futures::stream::StreamExt;
 use futures::task::AtomicWaker;
@@ -178,6 +179,7 @@ pub struct Isolate {
  needs_init: bool,
  pub(crate) shared: SharedQueue,
  pending_ops: FuturesUnordered<PendingOpFuture>,
+ pending_unref_ops: FuturesUnordered<PendingOpFuture>,
  have_unpolled_ops: bool,
  startup_script: Option<OwnedScript>,
  pub op_registry: Arc<OpRegistry>,
@@ -340,6 +342,7 @@ impl Isolate {
  shared,
  needs_init,
  pending_ops: FuturesUnordered::new(),
+ pending_unref_ops: FuturesUnordered::new(),
  have_unpolled_ops: false,
  startup_script,
  op_registry: Arc::new(OpRegistry::new()),
@@ -519,6 +522,12 @@ impl Isolate {
  self.have_unpolled_ops = true;
  None
  }
+ Op::AsyncUnref(fut) => {
+ let fut2 = fut.map_ok(move |buf| (op_id, buf));
+ self.pending_unref_ops.push(fut2.boxed());
+ self.have_unpolled_ops = true;
+ None
+ }
  }
  }
 
@@ -713,7 +722,9 @@ impl Future for Isolate {
  // Now handle actual ops.
  inner.have_unpolled_ops = false;
  #[allow(clippy::match_wild_err_arm)]
- match inner.pending_ops.poll_next_unpin(cx) {
+ match select(&mut inner.pending_ops, &mut inner.pending_unref_ops)
+ .poll_next_unpin(cx)
+ {
  Poll::Ready(Some(Err(_))) => panic!("unexpected op error"),
  Poll::Ready(None) => break,
  Poll::Pending => break,
@@ -816,6 +827,7 @@ pub mod tests {
 
  pub enum Mode {
  Async,
+ AsyncUnref,
  OverflowReqSync,
  OverflowResSync,
  OverflowReqAsync,
@@ -838,6 +850,17 @@ pub mod tests {
  let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
  Op::Async(futures::future::ok(buf).boxed())
  }
+ Mode::AsyncUnref => {
+ assert_eq!(control.len(), 1);
+ assert_eq!(control[0], 42);
+ let fut = async {
+ // This future never finish.
+ futures::future::pending::<()>().await;
+ let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
+ Ok(buf)
+ };
+ Op::AsyncUnref(fut.boxed())
+ }
  Mode::OverflowReqSync => {
  assert_eq!(control.len(), 100 * 1024 * 1024);
  let buf = vec![43u8, 0, 0, 0].into_boxed_slice();
@@ -955,6 +978,31 @@ pub mod tests {
  });
  }
 
+ #[test]
+ fn test_poll_async_optional_ops() {
+ run_in_task(|cx| {
+ let (mut isolate, dispatch_count) = setup(Mode::AsyncUnref);
+ js_check(isolate.execute(
+ "check1.js",
+ r#"
+ Deno.core.setAsyncHandler(1, (buf) => {
+ // This handler will never be called
+ assert(false);
+ });
+ let control = new Uint8Array([42]);
+ Deno.core.send(1, control);
+ "#,
+ ));
+ assert_eq!(dispatch_count.load(Ordering::Relaxed), 1);
+ // The above op never finish, but isolate can finish
+ // because the op is an unreffed async op.
+ assert!(match isolate.poll_unpin(cx) {
+ Poll::Ready(Ok(_)) => true,
+ _ => false,
+ });
+ })
+ }
+
  #[test]
  fn terminate_execution() {
  let (tx, rx) = std::sync::mpsc::channel::<bool>();

core/ops.rs

@@ -21,6 +21,9 @@ pub type OpResult<E> = Result<Op<E>, E>;
 pub enum Op<E> {
  Sync(Buf),
  Async(OpAsyncFuture<E>),
+ /// AsyncUnref is the variation of Async, which doesn't block the program
+ /// exiting.
+ AsyncUnref(OpAsyncFuture<E>),
 }
 
 pub type CoreError = ();