Repository: denoland/deno_core
Pull request #538 (2024-02-01 22:15:43)

fix(core): Ensure op driver is shut down unconditionally
We needed to revert the resource sanitizer in https://github.com/denoland/deno/pull/22125 because of a panic in third-party test code.

The problem was that in certain cases, if the OpDriver was kept alive either by Rc<ContextState> (in older code) or Rc<OpDriverImpl> (in newer code), the Tokio task that polls the underlying ops would potentially poll an op that expected something to be in OpState, but we had already removed all the items in the OpState by that point.




Patch:
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 45cd9f9..89d97bc 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -18,18 +18,13 @@ path = "main.rs"
 path = "lib.rs"
 
 [features]
-default = ["include_icu_data", "v8_use_custom_libcxx", "op_driver_futuresunordered"]
+default = ["include_icu_data", "v8_use_custom_libcxx"]
 include_icu_data = ["deno_core_icudata"]
 v8_use_custom_libcxx = ["v8/use_custom_libcxx"]
 include_js_files_for_snapshotting = []
 unsafe_runtime_options = []
 unsafe_use_unprotected_platform = []
 
-# Use the old, slower (but better tested) JoinSet driver
-op_driver_joinset = []
-# Use the new, faster FuturesUnordered driver
-op_driver_futuresunordered = []
-
 [dependencies]
 anyhow.workspace = true
 bit-set.workspace = true
diff --git a/core/runtime/jsrealm.rs b/core/runtime/jsrealm.rs
index 48dad9e..c237b50 100644
--- a/core/runtime/jsrealm.rs
+++ b/core/runtime/jsrealm.rs
@@ -44,15 +44,7 @@ impl Hasher for IdentityHasher {
   }
 }
 
-/// We will be experimenting with different driver types in the future. This allows us to
-/// swap the driver out for experimentation.
-#[cfg(all(
-  feature = "op_driver_joinset",
-  not(feature = "op_driver_futuresunordered")
-))]
-pub(crate) type OpDriverImpl = super::op_driver::JoinSetDriver;
-
-#[cfg(feature = "op_driver_futuresunordered")]
+/// We may wish to experiment with alternative drivers in the future.
 pub(crate) type OpDriverImpl = super::op_driver::FuturesUnorderedDriver;
 
 pub(crate) struct ContextState {
diff --git a/core/runtime/jsruntime.rs b/core/runtime/jsruntime.rs
index 6b594d4..c9c7d12 100644
--- a/core/runtime/jsruntime.rs
+++ b/core/runtime/jsruntime.rs
@@ -137,6 +137,9 @@ impl InnerIsolateState {
   /// after we've torn down the contexts. If the inspector is not correctly torn down, random crashes
   /// happen in tests (and possibly for users using the inspector).
   pub fn prepare_for_cleanup(&mut self) {
+    // Explicitly shut down the op driver here, just in case there are other references to it
+    // that prevent it from dropping after we invalidate the state.
+    self.main_realm.0.context_state.pending_ops.shutdown();
     let inspector = self.state.inspector.take();
     self.state.op_state.borrow_mut().clear();
     if let Some(inspector) = inspector {
@@ -1217,9 +1220,8 @@ impl JsRuntime {
 
   /// Returns the runtime's op names, ordered by OpId.
   pub fn op_names(&self) -> Vec<&'static str> {
-    let main_realm = self.inner.main_realm.clone();
-    let state_rc = main_realm.0.state();
-    state_rc.op_ctxs.iter().map(|o| o.decl.name).collect()
+    let state = &self.inner.main_realm.0.context_state;
+    state.op_ctxs.iter().map(|o| o.decl.name).collect()
   }
 
   /// Executes traditional JavaScript code (traditional = not ES modules).
diff --git a/core/runtime/op_driver/futures_unordered_driver.rs b/core/runtime/op_driver/futures_unordered_driver.rs
index a85ed56..96a50d4 100644
--- a/core/runtime/op_driver/futures_unordered_driver.rs
+++ b/core/runtime/op_driver/futures_unordered_driver.rs
@@ -66,9 +66,7 @@ pub struct FuturesUnorderedDriver<
 
 impl<C: OpMappingContext + 'static> Drop for FuturesUnorderedDriver<C> {
   fn drop(&mut self) {
-    if let MaybeTask::Handle(h) = self.task.take() {
-      h.abort()
-    }
+    self.shutdown()
   }
 }
 
@@ -230,6 +228,14 @@ impl<C: OpMappingContext> OpDriver<C> for FuturesUnorderedDriver<C> {
     self.len.get()
   }
 
+  fn shutdown(&self) {
+    if let MaybeTask::Handle(h) = self.task.take() {
+      h.abort()
+    }
+    self.completed_ops.borrow_mut().clear();
+    self.queue.queue.queue.borrow_mut().clear();
+  }
+
   fn stats(&self, op_exclusions: &BitSet) -> OpInflightStats {
     let q = self.queue.queue.queue.borrow();
     let mut v: Vec<PendingOpInfo> = Vec::with_capacity(self.len.get());
diff --git a/core/runtime/op_driver/joinset_driver.rs b/core/runtime/op_driver/joinset_driver.rs
deleted file mode 100644
index 93ba335..0000000
--- a/core/runtime/op_driver/joinset_driver.rs
+++ /dev/null
@@ -1,162 +0,0 @@
-// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-use super::op_results::*;
-use super::OpDriver;
-use super::OpInflightStats;
-use crate::OpId;
-use crate::PromiseId;
-use anyhow::Error;
-use bit_set::BitSet;
-use deno_unsync::JoinSet;
-use futures::task::noop_waker_ref;
-use futures::FutureExt;
-use std::cell::RefCell;
-use std::future::ready;
-use std::future::Future;
-use std::task::ready;
-use std::task::Context;
-use std::task::Poll;
-
-/// [`OpDriver`] implementation built on a tokio [`JoinSet`].
-pub struct JoinSetDriver<C: OpMappingContext = V8OpMappingContext> {
-  pending_ops: RefCell<JoinSet<PendingOp<C>>>,
-}
-
-impl<C: OpMappingContext> Default for JoinSetDriver<C> {
-  fn default() -> Self {
-    Self {
-      pending_ops: Default::default(),
-    }
-  }
-}
-
-impl<C: OpMappingContext> JoinSetDriver<C> {
-  /// Spawn an unpolled task, along with a function that can map it to a [`PendingOp`].
-  #[inline(always)]
-  fn spawn_unpolled<R>(
-    &self,
-    task: impl Future<Output = R> + 'static,
-    map: impl FnOnce(R) -> PendingOp<C> + 'static,
-  ) {
-    self.pending_ops.borrow_mut().spawn(task.map(map));
-  }
-
-  /// Spawn a ready task that already has a [`PendingOp`].
-  #[inline(always)]
-  fn spawn_ready(&self, ready_op: PendingOp<C>) {
-    self.pending_ops.borrow_mut().spawn(ready(ready_op));
-  }
-
-  /// Spawn a polled task inside a [`FutureAllocation`], along with a function that can map it to a [`PendingOp`].
-  #[inline(always)]
-  fn spawn_polled<R>(
-    &self,
-    task: impl Future<Output = R> + 'static,
-    map: impl FnOnce(R) -> PendingOp<C> + 'static,
-  ) {
-    self.pending_ops.borrow_mut().spawn(task.map(map));
-  }
-}
-
-impl<C: OpMappingContext> OpDriver<C> for JoinSetDriver<C> {
-  fn submit_op_fallible<
-    R: 'static,
-    E: Into<Error> + 'static,
-    const LAZY: bool,
-    const DEFERRED: bool,
-  >(
-    &self,
-    op_id: OpId,
-    promise_id: i32,
-    op: impl Future<Output = Result<R, E>> + 'static,
-    rv_map: C::MappingFn<R>,
-  ) -> Option<Result<R, E>> {
-    {
-      let info = PendingOpInfo(promise_id, op_id);
-
-      if LAZY {
-        self.spawn_unpolled(op, move |r| PendingOp::new(info, rv_map, r));
-        return None;
-      }
-
-      // We poll every future here because it's much faster to return a result than
-      // spin the event loop to get it.
-      let mut pinned = op.boxed_local();
-      match pinned.poll_unpin(&mut Context::from_waker(noop_waker_ref())) {
-        Poll::Pending => {
-          self.spawn_polled(pinned, move |r| PendingOp::new(info, rv_map, r))
-        }
-        Poll::Ready(res) => {
-          if DEFERRED {
-            self.spawn_ready(PendingOp::new(info, rv_map, res))
-          } else {
-            return Some(res);
-          }
-        }
-      };
-      None
-    }
-  }
-
-  fn submit_op_infallible<
-    R: 'static,
-    const LAZY: bool,
-    const DEFERRED: bool,
-  >(
-    &self,
-    op_id: OpId,
-    promise_id: i32,
-    op: impl Future<Output = R> + 'static,
-    rv_map: C::MappingFn<R>,
-  ) -> Option<R> {
-    {
-      let info = PendingOpInfo(promise_id, op_id);
-      if LAZY {
-        self.spawn_unpolled(op, move |r| PendingOp::ok(info, rv_map, r));
-        return None;
-      }
-
-      // We poll every future here because it's much faster to return a result than
-      // spin the event loop to get it.
-      let mut pinned = op.boxed_local();
-      match pinned.poll_unpin(&mut Context::from_waker(noop_waker_ref())) {
-        Poll::Pending => {
-          self.spawn_polled(pinned, move |res| PendingOp::ok(info, rv_map, res))
-        }
-        Poll::Ready(res) => {
-          if DEFERRED {
-            self.spawn_ready(PendingOp::ok(info, rv_map, res))
-          } else {
-            return Some(res);
-          }
-        }
-      };
-
-      None
-    }
-  }
-
-  #[inline(always)]
-  fn poll_ready<'s>(
-    &self,
-    cx: &mut Context,
-  ) -> Poll<(PromiseId, OpId, OpResult<C>)> {
-    let item = ready!(self.pending_ops.borrow_mut().poll_join_next(cx));
-    let PendingOp(PendingOpInfo(promise_id, op_id), resp) = match item {
-      Ok(x) => x,
-      Err(_) => {
-        // If this task is really errored, things could be pretty bad
-        panic!("Unrecoverable error: op panicked");
-      }
-    };
-    Poll::Ready((promise_id, op_id, resp))
-  }
-
-  #[inline(always)]
-  fn len(&self) -> usize {
-    self.pending_ops.borrow().len()
-  }
-
-  fn stats(&self, _op_exclusions: &BitSet) -> OpInflightStats {
-    OpInflightStats::default()
-  }
-}
diff --git a/core/runtime/op_driver/mod.rs b/core/runtime/op_driver/mod.rs
index f20b730..1b3062b 100644
--- a/core/runtime/op_driver/mod.rs
+++ b/core/runtime/op_driver/mod.rs
@@ -10,13 +10,10 @@ use std::task::Poll;
 mod erased_future;
 mod future_arena;
 mod futures_unordered_driver;
-mod joinset_driver;
 mod op_results;
 
 #[allow(unused)]
 pub use futures_unordered_driver::FuturesUnorderedDriver;
-#[allow(unused)]
-pub use joinset_driver::JoinSetDriver;
 
 pub use self::op_results::OpMappingContext;
 pub use self::op_results::OpResult;
@@ -131,6 +128,10 @@ pub(crate) trait OpDriver<C: OpMappingContext = V8OpMappingContext>:
   /// picked up in `poll_ready`.
   fn len(&self) -> usize;
 
+  /// Shuts down this driver, preventing any tasks from being polled beyond this point. It is legal
+  /// to call this shutdown method multiple times, and further calls have no effect.
+  fn shutdown(&self);
+
   /// Capture the statistics of in-flight ops, for op sanitizer purposes. Note that this
   /// may not be a cheap operation and calling it large number of times (for example, in an
   /// event loop) may cause slowdowns.
@@ -242,7 +243,6 @@ mod tests {
   }
 
   #[rstest]
-  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
   #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
   fn test_driver<D: OpDriver<TestMappingContext>>(
     #[case] driver: D,
@@ -273,7 +273,6 @@ mod tests {
   }
 
   #[rstest]
-  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
   #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
   fn test_driver_yield<D: OpDriver<TestMappingContext>>(
     #[case] driver: D,
@@ -311,7 +310,6 @@ mod tests {
   }
 
   #[rstest]
-  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
   #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
   fn test_driver_large<D: OpDriver<TestMappingContext>>(
     #[case] driver: D,
@@ -351,7 +349,6 @@ mod tests {
 
   #[cfg(not(miri))]
   #[rstest]
-  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
   #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
   fn test_driver_io<D: OpDriver<TestMappingContext>>(
     #[case] driver: D,


Test patch:
diff --git a/testing/checkin/runner/mod.rs b/testing/checkin/runner/mod.rs
index 1105329..68b12e0 100644
--- a/testing/checkin/runner/mod.rs
+++ b/testing/checkin/runner/mod.rs
@@ -6,10 +6,14 @@ use deno_core::CrossIsolateStore;
 use deno_core::JsRuntime;
 use deno_core::PollEventLoopOptions;
 use deno_core::RuntimeOptions;
+use futures::Future;
 use pretty_assertions::assert_eq;
 use std::path::Path;
 use std::path::PathBuf;
 use std::rc::Rc;
+use std::sync::mpsc::channel;
+use std::sync::mpsc::RecvTimeoutError;
+use std::time::Duration;
 use testing::Output;
 use tokio::fs::File;
 use tokio::io::AsyncReadExt;
@@ -44,6 +48,7 @@ deno_core::extension!(
     ops_async::op_async_yield,
     ops_async::op_async_barrier_create,
     ops_async::op_async_barrier_await,
+    ops_async::op_async_spin_on_state,
     ops_error::op_async_throw_error_eager,
     ops_error::op_async_throw_error_lazy,
     ops_error::op_async_throw_error_deferred,
@@ -112,17 +117,38 @@ fn create_runtime(
   (runtime, worker_host_side)
 }
 
-/// Run a integration test within the `checkin` runtime. This executes a single file, imports and all,
-/// and compares its output with the `.out` file in the same directory.
-pub fn run_integration_test(test: &str) {
-  let (runtime, _) = create_runtime(Output::default(), None);
+fn run_async(f: impl Future<Output = Result<(), Error>>) {
   let tokio = tokio::runtime::Builder::new_current_thread()
     .enable_all()
     .build()
     .expect("Failed to build a runtime");
-  tokio
-    .block_on(run_integration_test_task(runtime, test.to_owned()))
-    .expect("Failed to complete test");
+  tokio.block_on(f).expect("Failed to run the given task");
+
+  // We don't have a good way to wait for tokio to go idle here, but we'd like tokio
+  // to poll any remaining tasks to shake out any errors.
+  let handle = tokio.spawn(async {
+    tokio::task::yield_now().await;
+  });
+  _ = tokio.block_on(handle);
+
+  let (tx, rx) = channel::<()>();
+  let timeout = std::thread::spawn(move || {
+    if rx.recv_timeout(Duration::from_secs(10))
+      == Err(RecvTimeoutError::Timeout)
+    {
+      panic!("Failed to shut down the runtime in time");
+    }
+  });
+  drop(tokio);
+  drop(tx);
+  _ = timeout.join();
+}
+
+/// Run a integration test within the `checkin` runtime. This executes a single file, imports and all,
+/// and compares its output with the `.out` file in the same directory.
+pub fn run_integration_test(test: &str) {
+  let (runtime, _) = create_runtime(Output::default(), None);
+  run_async(run_integration_test_task(runtime, test.to_owned()));
 }
 
 async fn run_integration_test_task(
@@ -162,13 +188,7 @@ async fn run_integration_test_task(
 /// then each test is run individually and failures are printed.
 pub fn run_unit_test(test: &str) {
   let (runtime, _) = create_runtime(Output::default(), None);
-  let tokio = tokio::runtime::Builder::new_current_thread()
-    .enable_all()
-    .build()
-    .expect("Failed to build a runtime");
-  tokio
-    .block_on(run_unit_test_task(runtime, test.to_owned()))
-    .expect("Failed to complete test");
+  run_async(run_unit_test_task(runtime, test.to_owned()));
 }
 
 async fn run_unit_test_task(
diff --git a/testing/checkin/runner/ops_async.rs b/testing/checkin/runner/ops_async.rs
index 0e45834..fe7f926 100644
--- a/testing/checkin/runner/ops_async.rs
+++ b/testing/checkin/runner/ops_async.rs
@@ -1,8 +1,12 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 use deno_core::op2;
+use deno_core::OpState;
+use futures::future::poll_fn;
+use std::cell::RefCell;
 use std::future::Future;
 use std::rc::Rc;
 
+use super::testing::Output;
 use super::testing::TestData;
 
 #[op2(async)]
@@ -31,3 +35,14 @@ pub fn op_async_barrier_await(
     barrier.wait().await;
   }
 }
+
+#[op2(async)]
+pub async fn op_async_spin_on_state(state: Rc<RefCell<OpState>>) {
+  poll_fn(|cx| {
+    // Ensure that we never get polled when the state has been emptied
+    state.borrow().borrow::<Output>();
+    cx.waker().wake_by_ref();
+    std::task::Poll::Pending
+  })
+  .await
+}
diff --git a/testing/checkin/runner/ops_worker.rs b/testing/checkin/runner/ops_worker.rs
index 7c3fa47..b5e947b 100644
--- a/testing/checkin/runner/ops_worker.rs
+++ b/testing/checkin/runner/ops_worker.rs
@@ -23,6 +23,7 @@ use tokio::sync::watch;
 use tokio::sync::Mutex;
 
 use super::create_runtime;
+use super::run_async;
 use super::testing::Output;
 
 /// Our cppgc object.
@@ -110,13 +111,7 @@ pub fn op_worker_spawn<'s>(
       })
       .map_err(|_| unreachable!())
       .unwrap();
-    let tokio = tokio::runtime::Builder::new_current_thread()
-      .enable_all()
-      .build()
-      .expect("Failed to build a runtime");
-    tokio
-      .block_on(run_worker_task(runtime, base_url, main_script, shutdown_rx))
-      .expect("Failed to complete test");
+    run_async(run_worker_task(runtime, base_url, main_script, shutdown_rx));
   });
 
   // This is technically a blocking call
diff --git a/testing/checkin/runtime/async.ts b/testing/checkin/runtime/async.ts
index a4d9fd2..de5b533 100644
--- a/testing/checkin/runtime/async.ts
+++ b/testing/checkin/runtime/async.ts
@@ -3,6 +3,7 @@ const {
   op_async_barrier_create,
   op_async_barrier_await,
   op_async_yield,
+  op_async_spin_on_state,
   op_stats_capture,
   op_stats_diff,
   op_stats_dump,
@@ -23,6 +24,11 @@ export async function asyncYield() {
   await op_async_yield();
 }
 
+// This function never returns.
+export async function asyncSpin() {
+  await op_async_spin_on_state();
+}
+
 let nextStats = 0;
 
 export class Stats {
diff --git a/testing/integration/worker_terminate_op/worker.ts b/testing/integration/worker_terminate_op/worker.ts
new file mode 100644
index 0000000..83eb0d1
--- /dev/null
+++ b/testing/integration/worker_terminate_op/worker.ts
@@ -0,0 +1,8 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { asyncSpin, asyncYield } from "checkin:async";
+import { Worker } from "checkin:worker";
+const p = asyncSpin();
+await asyncYield();
+Worker.parent.sendMessage("hello from client");
+await p;
+console.log("worker shouldn't get here!");
diff --git a/testing/integration/worker_terminate_op/worker_terminate_op.out b/testing/integration/worker_terminate_op/worker_terminate_op.out
new file mode 100644
index 0000000..c48897a
--- /dev/null
+++ b/testing/integration/worker_terminate_op/worker_terminate_op.out
@@ -0,0 +1,5 @@
+main started
+main worker booted
+main got from worker "hello from client"
+[ERR] Uncaught Error: execution terminated
+main exiting
diff --git a/testing/integration/worker_terminate_op/worker_terminate_op.ts b/testing/integration/worker_terminate_op/worker_terminate_op.ts
new file mode 100644
index 0000000..4b3869b
--- /dev/null
+++ b/testing/integration/worker_terminate_op/worker_terminate_op.ts
@@ -0,0 +1,12 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { setTimeout } from "checkin:timers";
+import { Worker } from "checkin:worker";
+console.log("main started");
+const worker = new Worker(import.meta.url, "./worker.ts");
+console.log("main worker booted");
+const message = await worker.receiveMessage();
+console.log(`main got from worker "${message}"`);
+worker.terminate();
+await worker.closed;
+console.log("main exiting");
+await new Promise((r) => setTimeout(r, 100));
diff --git a/testing/lib.rs b/testing/lib.rs
index 09605aa..f0c95f4 100644
--- a/testing/lib.rs
+++ b/testing/lib.rs
@@ -62,4 +62,5 @@ integration_test!(
   timer_many,
   worker_spawn,
   worker_terminate,
+  worker_terminate_op,
 );