Repository: denoland/deno_core
Pull request #414 (2024-01-04 20:41:54)

perf(core): use FuturesUnordered instead of JoinSet
Implements a new `OpDriver` based on `FuturesUnordered`.

As part of this work, we create a trait that allows us to test `OpDriver` without V8. This part of the code is a bit awkward, but we can likely remove it at a later date once the performance work slows down.

Other changes:

 - Add `OpScheduling` to simplify the lazy/deferred/eager traits.
 - Modify the batch size of dispatched async ops to 1024 to prevent stack overflow errors
 - Keeps errors in `anyhow::Error` until we're ready to dispatch them

Before:

```
test bench_op_async_void_deferred         ... bench:   1,034,629 ns/iter (+/- 74,835)
test bench_op_async_void_deferred_nofast  ... bench:   1,034,527 ns/iter (+/- 84,364)
test bench_op_async_void_deferred_return  ... bench:   1,031,853 ns/iter (+/- 66,991)
test bench_op_async_void_lazy             ... bench:     993,310 ns/iter (+/- 30,881)
test bench_op_async_void_lazy_nofast      ... bench:   1,011,262 ns/iter (+/- 37,490)
test bench_op_async_yield                 ... bench:   1,037,248 ns/iter (+/- 44,996)
test bench_op_async_yield_deferred        ... bench:   1,052,324 ns/iter (+/- 36,561)
test bench_op_async_yield_deferred_nofast ... bench:   1,048,557 ns/iter (+/- 55,217)
test bench_op_async_yield_lazy            ... bench:   1,546,403 ns/iter (+/- 83,743)
test bench_op_async_yield_lazy_nofast     ... bench:   1,541,818 ns/iter (+/- 51,707)
```

After:

```
test bench_op_async_void_deferred         ... bench:     799,585 ns/iter (+/- 21,081)
test bench_op_async_void_deferred_nofast  ... bench:     790,412 ns/iter (+/- 55,422)
test bench_op_async_void_deferred_return  ... bench:     793,884 ns/iter (+/- 24,304)
test bench_op_async_void_lazy             ... bench:     760,932 ns/iter (+/- 28,006)
test bench_op_async_void_lazy_nofast      ... bench:     759,421 ns/iter (+/- 17,968)
test bench_op_async_yield                 ... bench:     780,460 ns/iter (+/- 16,812)
test bench_op_async_yield_deferred        ... bench:     799,123 ns/iter (+/- 17,069)
test bench_op_async_yield_deferred_nofast ... bench:     791,680 ns/iter (+/- 39,701)
test bench_op_async_yield_lazy            ... bench:     994,607 ns/iter (+/- 19,175)
test bench_op_async_yield_lazy_nofast     ... bench:   1,000,546 ns/iter (+/- 28,675)
```


Patch:
diff --git a/core/Cargo.toml b/core/Cargo.toml
index e6fe328..50378cd 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -14,13 +14,16 @@ description = "A modern JavaScript/TypeScript runtime built with V8, Rust, and T
 path = "lib.rs"
 
 [features]
-default = ["include_icu_data", "v8_use_custom_libcxx", "op_driver_joinset"]
+default = ["include_icu_data", "v8_use_custom_libcxx", "op_driver_futuresunordered"]
 include_icu_data = []
 v8_use_custom_libcxx = ["v8/use_custom_libcxx"]
 include_js_files_for_snapshotting = []
 unsafe_runtime_options = []
 
+# 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
diff --git a/core/extensions.rs b/core/extensions.rs
index 535a4fd..fd07107 100644
--- a/core/extensions.rs
+++ b/core/extensions.rs
@@ -1,5 +1,5 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-use crate::modules::ModuleCode;
+use crate::modules::ModuleCodeString;
 use crate::OpState;
 use anyhow::Context as _;
 use anyhow::Error;
@@ -36,7 +36,7 @@ impl ExtensionFileSource {
     s.chars().filter(|c| !c.is_ascii()).collect::<String>()
   }
 
-  pub fn load(&self) -> Result<ModuleCode, Error> {
+  pub fn load(&self) -> Result<ModuleCodeString, Error> {
     match &self.code {
       ExtensionFileSourceCode::IncludedInBinary(code) => {
         debug_assert!(
@@ -45,7 +45,7 @@ impl ExtensionFileSource {
           self.specifier,
           Self::find_non_ascii(code)
         );
-        Ok(ModuleCode::from_static(code))
+        Ok(ModuleCodeString::from_static(code))
       }
       ExtensionFileSourceCode::LoadedFromFsDuringSnapshot(path) => {
         let msg = || format!("Failed to read \"{}\"", path);
@@ -65,7 +65,7 @@ impl ExtensionFileSource {
           self.specifier,
           Self::find_non_ascii(code)
         );
-        Ok(ModuleCode::Arc(code.clone()))
+        Ok(ModuleCodeString::Arc(code.clone()))
       }
     }
   }
diff --git a/core/lib.rs b/core/lib.rs
index cda4f05..924d119 100644
--- a/core/lib.rs
+++ b/core/lib.rs
@@ -92,7 +92,8 @@ pub use crate::module_specifier::ModuleResolutionError;
 pub use crate::module_specifier::ModuleSpecifier;
 pub use crate::modules::ExtModuleLoaderCb;
 pub use crate::modules::FsModuleLoader;
-pub use crate::modules::ModuleCode;
+pub use crate::modules::ModuleCodeBytes;
+pub use crate::modules::ModuleCodeString;
 pub use crate::modules::ModuleId;
 pub use crate::modules::ModuleLoader;
 pub use crate::modules::ModuleSource;
diff --git a/core/modules/loaders.rs b/core/modules/loaders.rs
index 916cb45..a74385e 100644
--- a/core/modules/loaders.rs
+++ b/core/modules/loaders.rs
@@ -3,7 +3,7 @@ use crate::error::generic_error;
 use crate::error::AnyError;
 use crate::extensions::ExtensionFileSource;
 use crate::module_specifier::ModuleSpecifier;
-use crate::modules::ModuleCode;
+use crate::modules::ModuleCodeString;
 use crate::modules::ModuleSource;
 use crate::modules::ModuleSourceFuture;
 use crate::modules::ModuleType;
@@ -106,7 +106,7 @@ impl ModuleLoader for NoopModuleLoader {
 /// Function that can be passed to the `ExtModuleLoader` that allows to
 /// transpile sources before passing to V8.
 pub type ExtModuleLoaderCb =
-  Box<dyn Fn(&ExtensionFileSource) -> Result<ModuleCode, Error>>;
+  Box<dyn Fn(&ExtensionFileSource) -> Result<ModuleCodeString, Error>>;
 
 pub(crate) struct ExtModuleLoader {
   sources: RefCell<HashMap<String, ExtensionFileSource>>,
@@ -308,7 +308,7 @@ impl ModuleLoader for FsModuleLoader {
       })?;
       let module = ModuleSource::new(
         module_type,
-        ModuleSourceCode::Bytes(code),
+        ModuleSourceCode::Bytes(code.into_boxed_slice().into()),
         module_specifier,
       );
       Ok(module)
@@ -321,13 +321,13 @@ impl ModuleLoader for FsModuleLoader {
 /// A module loader that you can pre-load a number of modules into and resolve from. Useful for testing and
 /// embedding situations where the filesystem and snapshot systems are not usable or a good fit.
 pub struct StaticModuleLoader {
-  map: HashMap<ModuleSpecifier, ModuleCode>,
+  map: HashMap<ModuleSpecifier, ModuleCodeString>,
 }
 
 impl StaticModuleLoader {
   /// Create a new [`StaticModuleLoader`] from an `Iterator` of specifiers and code.
   pub fn new(
-    from: impl IntoIterator<Item = (ModuleSpecifier, ModuleCode)>,
+    from: impl IntoIterator<Item = (ModuleSpecifier, ModuleCodeString)>,
   ) -> Self {
     Self {
       map: HashMap::from_iter(
@@ -339,7 +339,7 @@ impl StaticModuleLoader {
   }
 
   /// Create a new [`StaticModuleLoader`] from a single code item.
-  pub fn with(specifier: ModuleSpecifier, code: ModuleCode) -> Self {
+  pub fn with(specifier: ModuleSpecifier, code: ModuleCodeString) -> Self {
     Self::new([(specifier, code)])
   }
 }
diff --git a/core/modules/map.rs b/core/modules/map.rs
index 74d0a54..842d74c 100644
--- a/core/modules/map.rs
+++ b/core/modules/map.rs
@@ -9,7 +9,7 @@ use crate::modules::get_requested_module_type_from_attributes;
 use crate::modules::parse_import_attributes;
 use crate::modules::recursive_load::RecursiveModuleLoad;
 use crate::modules::ImportAttributesKind;
-use crate::modules::ModuleCode;
+use crate::modules::ModuleCodeString;
 use crate::modules::ModuleError;
 use crate::modules::ModuleId;
 use crate::modules::ModuleLoadId;
@@ -20,7 +20,6 @@ use crate::modules::ModuleType;
 use crate::modules::ResolutionKind;
 use crate::runtime::exception_state::ExceptionState;
 use crate::runtime::JsRealm;
-use crate::runtime::SnapshottedData;
 use crate::ExtensionFileSource;
 use crate::JsRuntime;
 use crate::ModuleSource;
@@ -48,6 +47,7 @@ use std::task::Poll;
 use tokio::sync::oneshot;
 
 use super::module_map_data::ModuleMapData;
+use super::module_map_data::ModuleMapSnapshottedData;
 use super::LazyEsmModuleLoader;
 use super::RequestedModuleType;
 
@@ -177,7 +177,7 @@ impl ModuleMap {
     exception_state: Rc<ExceptionState>,
     import_meta_resolve_cb: ImportMetaResolveCallback,
     scope: &mut v8::HandleScope,
-    data: SnapshottedData,
+    data: ModuleMapSnapshottedData,
   ) -> Self {
     let new = Self::new(loader, exception_state, import_meta_resolve_cb);
     new
@@ -232,7 +232,7 @@ impl ModuleMap {
   pub(crate) fn serialize_for_snapshotting(
     &self,
     scope: &mut v8::HandleScope,
-  ) -> SnapshottedData {
+  ) -> ModuleMapSnapshottedData {
     self.data.borrow().serialize_for_snapshotting(scope)
   }
 
@@ -347,7 +347,7 @@ impl ModuleMap {
     scope: &mut v8::HandleScope,
     main: bool,
     name: ModuleName,
-    source: ModuleCode,
+    source: ModuleCodeString,
     is_dynamic_import: bool,
   ) -> Result<ModuleId, ModuleError> {
     let name_str = name.v8(scope);
@@ -450,7 +450,7 @@ impl ModuleMap {
     &self,
     scope: &mut v8::HandleScope,
     name: ModuleName,
-    code: ModuleCode,
+    code: ModuleCodeString,
   ) -> Result<ModuleId, ModuleError> {
     let source_str = v8::String::new_from_utf8(
       scope,
@@ -1303,7 +1303,7 @@ impl ModuleMap {
     &self,
     scope: &mut v8::HandleScope,
     module_specifier: &str,
-    source_code: ModuleCode,
+    source_code: ModuleCodeString,
   ) -> Result<v8::Global<v8::Value>, Error> {
     let specifier = ModuleSpecifier::parse(module_specifier)?;
     let mod_id = self
diff --git a/core/modules/mod.rs b/core/modules/mod.rs
index 7903af1..2e0124c 100644
--- a/core/modules/mod.rs
+++ b/core/modules/mod.rs
@@ -10,6 +10,7 @@ use serde::Serialize;
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::future::Future;
+use std::sync::Arc;
 
 mod loaders;
 mod map;
@@ -32,6 +33,7 @@ pub use loaders::ModuleLoader;
 pub use loaders::NoopModuleLoader;
 pub use loaders::StaticModuleLoader;
 pub(crate) use map::ModuleMap;
+pub(crate) use module_map_data::ModuleMapSnapshottedData;
 
 pub type ModuleId = usize;
 pub(crate) type ModuleLoadId = i32;
@@ -41,15 +43,62 @@ pub(crate) type ModuleLoadId = i32;
 /// converted to a string or not.
 #[derive(Debug)]
 pub enum ModuleSourceCode {
-  String(ModuleCode),
-  Bytes(Vec<u8>),
+  String(ModuleCodeString),
+  Bytes(ModuleCodeBytes),
 }
 
-// TODO(bartlomieju): maybe remove it? It might be confusing between `ModuleSourceCode`
-// and `ModuleCode`.
-pub type ModuleCode = FastString;
+pub type ModuleCodeString = FastString;
 pub type ModuleName = FastString;
 
+#[derive(Debug)]
+pub enum ModuleCodeBytes {
+  /// Created from static data.
+  Static(&'static [u8]),
+
+  /// An owned chunk of data. Note that we use `Box` rather than `Vec` to avoid
+  /// the storage overhead.
+  Boxed(Box<[u8]>),
+
+  /// Code loaded from the `deno_graph` infrastructure.
+  Arc(Arc<[u8]>),
+}
+
+impl ModuleCodeBytes {
+  pub fn as_bytes(&self) -> &[u8] {
+    match self {
+      ModuleCodeBytes::Static(s) => s,
+      ModuleCodeBytes::Boxed(s) => s,
+      ModuleCodeBytes::Arc(s) => s,
+    }
+  }
+
+  pub fn to_vec(&self) -> Vec<u8> {
+    match self {
+      ModuleCodeBytes::Static(s) => s.to_vec(),
+      ModuleCodeBytes::Boxed(s) => s.to_vec(),
+      ModuleCodeBytes::Arc(s) => s.to_vec(),
+    }
+  }
+}
+
+impl From<Arc<[u8]>> for ModuleCodeBytes {
+  fn from(value: Arc<[u8]>) -> Self {
+    Self::Arc(value)
+  }
+}
+
+impl From<Box<[u8]>> for ModuleCodeBytes {
+  fn from(value: Box<[u8]>) -> Self {
+    Self::Boxed(value)
+  }
+}
+
+impl From<&'static [u8]> for ModuleCodeBytes {
+  fn from(value: &'static [u8]) -> Self {
+    Self::Static(value)
+  }
+}
+
 /// Callback to customize value of `import.meta.resolve("./foo.ts")`.
 pub type ImportMetaResolveCallback = Box<
   dyn Fn(&dyn ModuleLoader, String, String) -> Result<ModuleSpecifier, Error>,
@@ -274,14 +323,14 @@ impl ModuleSource {
   pub fn get_string_source(
     specifier: &str,
     code: ModuleSourceCode,
-  ) -> Result<ModuleCode, AnyError> {
+  ) -> Result<ModuleCodeString, AnyError> {
     match code {
       ModuleSourceCode::String(code) => Ok(code),
       ModuleSourceCode::Bytes(bytes) => {
-        let str_ = String::from_utf8(bytes).with_context(|| {
+        let str_ = String::from_utf8(bytes.to_vec()).with_context(|| {
           format!("Can't convert source code to string for {}", specifier)
         })?;
-        Ok(ModuleCode::from(str_))
+        Ok(ModuleCodeString::from(str_))
       }
     }
   }
diff --git a/core/modules/module_map_data.rs b/core/modules/module_map_data.rs
index 4482d5c..df18ab6 100644
--- a/core/modules/module_map_data.rs
+++ b/core/modules/module_map_data.rs
@@ -7,7 +7,6 @@ use crate::modules::ModuleLoadId;
 use crate::modules::ModuleName;
 use crate::modules::ModuleRequest;
 use crate::modules::ModuleType;
-use crate::runtime::SnapshottedData;
 use crate::ExtensionFileSource;
 use std::cell::RefCell;
 use std::collections::HashMap;
@@ -113,6 +112,11 @@ impl<T> ModuleNameTypeMap<T> {
   }
 }
 
+pub(crate) struct ModuleMapSnapshottedData {
+  pub module_map_data: v8::Global<v8::Array>,
+  pub module_handles: Vec<v8::Global<v8::Module>>,
+}
+
 #[derive(Default)]
 pub(crate) struct ModuleMapData {
   /// Inverted index from module to index in `info`.
@@ -262,7 +266,7 @@ impl ModuleMapData {
   pub fn serialize_for_snapshotting(
     &self,
     scope: &mut v8::HandleScope,
-  ) -> SnapshottedData {
+  ) -> ModuleMapSnapshottedData {
     let data = self;
     let array = v8::Array::new(scope, 3);
 
@@ -340,7 +344,7 @@ impl ModuleMapData {
     let array_global = v8::Global::new(scope, array);
 
     let handles = data.handles.clone();
-    SnapshottedData {
+    ModuleMapSnapshottedData {
       module_map_data: array_global,
       module_handles: handles,
     }
@@ -349,7 +353,7 @@ impl ModuleMapData {
   pub fn update_with_snapshotted_data(
     &mut self,
     scope: &mut v8::HandleScope,
-    snapshotted_data: SnapshottedData,
+    snapshotted_data: ModuleMapSnapshottedData,
   ) {
     let local_data: v8::Local<v8::Array> =
       v8::Local::new(scope, snapshotted_data.module_map_data);
diff --git a/core/runtime/jsrealm.rs b/core/runtime/jsrealm.rs
index dcf3f3e..eaa85d5 100644
--- a/core/runtime/jsrealm.rs
+++ b/core/runtime/jsrealm.rs
@@ -4,13 +4,14 @@ use super::exception_state::ExceptionState;
 use super::op_driver::OpDriver;
 use crate::error::exception_to_err_result;
 use crate::module_specifier::ModuleSpecifier;
-use crate::modules::ModuleCode;
+use crate::modules::ModuleCodeString;
 use crate::modules::ModuleError;
 use crate::modules::ModuleId;
 use crate::modules::ModuleMap;
 use crate::ops::OpCtx;
 use crate::tasks::V8TaskSpawnerFactory;
 use crate::web_timeout::WebTimers;
+use crate::GetErrorClassFn;
 use anyhow::Error;
 use futures::stream::StreamExt;
 use std::cell::Cell;
@@ -49,7 +50,12 @@ impl Hasher for IdentityHasher {
 #[cfg(feature = "op_driver_joinset")]
 type DefaultOpDriver = super::op_driver::JoinSetDriver;
 
-#[derive(Default)]
+#[cfg(all(
+  feature = "op_driver_futuresunordered",
+  not(feature = "op_driver_joinset")
+))]
+type DefaultOpDriver = super::op_driver::FuturesUnorderedDriver;
+
 pub(crate) struct ContextState<OpDriverImpl: OpDriver = DefaultOpDriver> {
   pub(crate) task_spawner_factory: Arc<V8TaskSpawnerFactory>,
   pub(crate) timers: WebTimers<(v8::Global<v8::Function>, u32)>,
@@ -66,6 +72,31 @@ pub(crate) struct ContextState<OpDriverImpl: OpDriver = DefaultOpDriver> {
   pub(crate) isolate: Option<*mut v8::OwnedIsolate>,
   pub(crate) exception_state: Rc<ExceptionState>,
   pub(crate) has_next_tick_scheduled: Cell<bool>,
+  pub(crate) get_error_class_fn: GetErrorClassFn,
+  pub(crate) ops_with_metrics: Vec<bool>,
+}
+
+impl<O: OpDriver> ContextState<O> {
+  pub(crate) fn new(
+    isolate_ptr: *mut v8::OwnedIsolate,
+    get_error_class_fn: GetErrorClassFn,
+    ops_with_metrics: Vec<bool>,
+  ) -> Self {
+    Self {
+      isolate: Some(isolate_ptr),
+      get_error_class_fn,
+      ops_with_metrics,
+      exception_state: Default::default(),
+      has_next_tick_scheduled: Default::default(),
+      js_event_loop_tick_cb: Default::default(),
+      js_wasm_streaming_cb: Default::default(),
+      op_ctxs: Default::default(),
+      pending_ops: Default::default(),
+      task_spawner_factory: Default::default(),
+      timers: Default::default(),
+      unrefed_ops: Default::default(),
+    }
+  }
 }
 
 /// A representation of a JavaScript realm tied to a [`JsRuntime`], that allows
@@ -227,7 +258,7 @@ impl JsRealm {
 
   fn string_from_code<'a>(
     scope: &mut HandleScope<'a>,
-    code: &ModuleCode,
+    code: &ModuleCodeString,
   ) -> Option<Local<'a, v8::String>> {
     if let Some(code) = code.try_static_ascii() {
       v8::String::new_external_onebyte_static(scope, code)
@@ -261,7 +292,11 @@ impl JsRealm {
     name: &'static str,
     source_code: &'static str,
   ) -> Result<v8::Global<v8::Value>, Error> {
-    self.execute_script(isolate, name, ModuleCode::from_static(source_code))
+    self.execute_script(
+      isolate,
+      name,
+      ModuleCodeString::from_static(source_code),
+    )
   }
 
   /// Executes traditional JavaScript code (traditional = not ES modules) in the
@@ -282,7 +317,7 @@ impl JsRealm {
     &self,
     isolate: &mut v8::Isolate,
     name: &'static str,
-    source_code: ModuleCode,
+    source_code: ModuleCodeString,
   ) -> Result<v8::Global<v8::Value>, Error> {
     let scope = &mut self.0.handle_scope(isolate);
 
@@ -358,7 +393,7 @@ impl JsRealm {
     &self,
     isolate: &mut v8::Isolate,
     specifier: &ModuleSpecifier,
-    code: Option<ModuleCode>,
+    code: Option<ModuleCodeString>,
   ) -> Result<ModuleId, Error> {
     let module_map_rc = self.0.module_map();
     if let Some(code) = code {
@@ -415,7 +450,7 @@ impl JsRealm {
     &self,
     isolate: &mut v8::Isolate,
     specifier: &ModuleSpecifier,
-    code: Option<ModuleCode>,
+    code: Option<ModuleCodeString>,
   ) -> Result<ModuleId, Error> {
     let module_map_rc = self.0.module_map();
     if let Some(code) = code {
@@ -472,7 +507,7 @@ impl JsRealm {
     &self,
     isolate: &mut v8::Isolate,
     module_specifier: &str,
-    code: ModuleCode,
+    code: ModuleCodeString,
   ) -> Result<v8::Global<v8::Value>, Error> {
     let module_map_rc = self.0.module_map();
     let scope = &mut self.handle_scope(isolate);
diff --git a/core/runtime/jsruntime.rs b/core/runtime/jsruntime.rs
index 166f291..c0ea547 100644
--- a/core/runtime/jsruntime.rs
+++ b/core/runtime/jsruntime.rs
@@ -5,6 +5,7 @@ use super::exception_state::ExceptionState;
 use super::jsrealm::JsRealmInner;
 use super::op_driver::OpDriver;
 use super::snapshot_util;
+use super::SnapshottedData;
 use crate::error::exception_to_err_result;
 use crate::error::AnyError;
 use crate::error::GetErrorClassFn;
@@ -18,10 +19,11 @@ use crate::module_specifier::ModuleSpecifier;
 use crate::modules::default_import_meta_resolve_cb;
 use crate::modules::ExtModuleLoader;
 use crate::modules::ImportMetaResolveCallback;
-use crate::modules::ModuleCode;
+use crate::modules::ModuleCodeString;
 use crate::modules::ModuleId;
 use crate::modules::ModuleLoader;
 use crate::modules::ModuleMap;
+use crate::modules::ModuleMapSnapshottedData;
 use crate::modules::RequestedModuleType;
 use crate::modules::ValidateImportAttributesCb;
 use crate::ops::*;
@@ -670,10 +672,26 @@ impl JsRuntime {
       waker,
     });
 
-    let context_state = Rc::new(ContextState {
-      isolate: Some(isolate_ptr),
-      ..Default::default()
-    });
+    let count = ops.len();
+    let op_metrics_fns = ops
+      .iter()
+      .enumerate()
+      .map(|(id, decl)| {
+        options
+          .op_metrics_factory_fn
+          .as_ref()
+          .and_then(|f| (f)(id as _, count, decl))
+      })
+      .collect::<Vec<_>>();
+    let ops_with_metrics = op_metrics_fns
+      .iter()
+      .map(|o| o.is_some())
+      .collect::<Vec<_>>();
+    let context_state = Rc::new(ContextState::new(
+      isolate_ptr,
+      options.get_error_class_fn.unwrap_or(&|_| "Error"),
+      ops_with_metrics,
+    ));
 
     // Add the task spawners to the OpState
     let spawner = context_state
@@ -687,15 +705,11 @@ impl JsRuntime {
       .new_cross_thread_spawner();
     op_state.borrow_mut().put(spawner);
 
-    let count = ops.len();
     let mut op_ctxs = ops
       .into_iter()
       .enumerate()
-      .map(|(id, decl)| {
-        let metrics_fn = options
-          .op_metrics_factory_fn
-          .as_ref()
-          .and_then(|f| (f)(id as _, count, &decl));
+      .zip(op_metrics_fns)
+      .map(|((id, decl), metrics_fn)| {
         OpCtx::new(
           id as _,
           std::ptr::null_mut(),
@@ -812,12 +826,18 @@ impl JsRuntime {
 
     let exception_state = context_state.exception_state.clone();
     let module_map = if let Some(snapshotted_data) = snapshotted_data {
+      *exception_state.js_handled_promise_rejection_cb.borrow_mut() =
+        snapshotted_data.js_handled_promise_rejection_cb;
+      let module_map_snapshotted_data = ModuleMapSnapshottedData {
+        module_map_data: snapshotted_data.module_map_data,
+        module_handles: snapshotted_data.module_handles,
+      };
       Rc::new(ModuleMap::new_from_snapshotted_data(
         loader,
         exception_state,
         import_meta_resolve_cb,
         scope,
-        snapshotted_data,
+        module_map_snapshotted_data,
       ))
     } else {
       Rc::new(ModuleMap::new(
@@ -1247,7 +1267,7 @@ impl JsRuntime {
   pub fn execute_script(
     &mut self,
     name: &'static str,
-    source_code: ModuleCode,
+    source_code: ModuleCodeString,
   ) -> Result<v8::Global<v8::Value>, Error> {
     let isolate = &mut self.inner.v8_isolate;
     self
@@ -1279,7 +1299,7 @@ impl JsRuntime {
     self.inner.main_realm.execute_script(
       isolate,
       name,
-      ModuleCode::from_static(source_code),
+      ModuleCodeString::from_static(source_code),
     )
   }
 
@@ -1851,12 +1871,26 @@ impl JsRuntimeForSnapshot {
 
     // Serialize the module map and store its data in the snapshot.
     {
-      let snapshotted_data = {
+      let module_map_snapshotted_data = {
         let module_map = realm.0.module_map();
         module_map.serialize_for_snapshotting(
           &mut realm.handle_scope(self.v8_isolate()),
         )
       };
+      let maybe_js_handled_promise_rejection_cb = {
+        let context_state = &realm.0.context_state;
+        let exception_state = &context_state.exception_state;
+        exception_state
+          .js_handled_promise_rejection_cb
+          .borrow()
+          .clone()
+      };
+
+      let snapshotted_data = SnapshottedData {
+        module_map_data: module_map_snapshotted_data.module_map_data,
+        module_handles: module_map_snapshotted_data.module_handles,
+        js_handled_promise_rejection_cb: maybe_js_handled_promise_rejection_cb,
+      };
 
       let mut scope = realm.handle_scope(self.v8_isolate());
       snapshot_util::set_snapshotted_data(
@@ -2037,7 +2071,7 @@ impl JsRuntime {
   pub async fn load_main_module(
     &mut self,
     specifier: &ModuleSpecifier,
-    code: Option<ModuleCode>,
+    code: Option<ModuleCodeString>,
   ) -> Result<ModuleId, Error> {
     let isolate = &mut self.inner.v8_isolate;
     self
@@ -2057,7 +2091,7 @@ impl JsRuntime {
   pub async fn load_side_module(
     &mut self,
     specifier: &ModuleSpecifier,
-    code: Option<ModuleCode>,
+    code: Option<ModuleCodeString>,
   ) -> Result<ModuleId, Error> {
     let isolate = &mut self.inner.v8_isolate;
     self
@@ -2077,7 +2111,7 @@ impl JsRuntime {
   pub fn lazy_load_es_module_from_code(
     &mut self,
     specifier: &str,
-    code: ModuleCode,
+    code: ModuleCodeString,
   ) -> Result<v8::Global<v8::Value>, Error> {
     let isolate = &mut self.inner.v8_isolate;
     self
@@ -2105,27 +2139,36 @@ impl JsRuntime {
       }
     }
 
-    // We return async responses to JS in unbounded batches (may change),
+    // We return async responses to JS in bounded batches. Note that because
+    // we're passing these to JS as arguments, it is possible to overflow the
+    // JS stack by just passing too many.
+    const MAX_VEC_SIZE_FOR_OPS: usize = 1024;
+
     // each batch is a flat vector of tuples:
     // `[promise_id1, op_result1, promise_id2, op_result2, ...]`
     // promise_id is a simple integer, op_result is an ops::OpResult
     // which contains a value OR an error, encoded as a tuple.
     // This batch is received in JS via the special `arguments` variable
     // and then each tuple is used to resolve or reject promises
-    //
-    // This can handle 15 promises futures in a single batch without heap
-    // allocations.
     let mut args: SmallVec<[v8::Local<v8::Value>; 32]> =
       SmallVec::with_capacity(32);
 
     loop {
-      let Poll::Ready((promise_id, op_id, metrics_event, res)) =
-        context_state.pending_ops.poll_ready(cx, scope)
+      if args.len() >= MAX_VEC_SIZE_FOR_OPS {
+        // We have too many, bail for now but re-wake the waker
+        cx.waker().wake_by_ref();
+        break;
+      }
+
+      let Poll::Ready((promise_id, op_id, res)) =
+        context_state.pending_ops.poll_ready(cx)
       else {
         break;
       };
 
-      if metrics_event {
+      let res = res.unwrap(scope, context_state.get_error_class_fn);
+
+      if context_state.ops_with_metrics[op_id as usize] {
         if res.is_ok() {
           dispatch_metrics_async(
             &context_state.op_ctxs.borrow()[op_id as usize],
diff --git a/core/runtime/op_driver/futures_unordered_driver.rs b/core/runtime/op_driver/futures_unordered_driver.rs
new file mode 100644
index 0000000..00f1206
--- /dev/null
+++ b/core/runtime/op_driver/futures_unordered_driver.rs
@@ -0,0 +1,316 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+use super::future_arena::FutureAllocation;
+use super::future_arena::FutureArena;
+use super::op_results::*;
+use super::OpDriver;
+use crate::OpId;
+use crate::PromiseId;
+use anyhow::Error;
+use deno_unsync::spawn;
+use deno_unsync::JoinHandle;
+use deno_unsync::UnsyncWaker;
+use futures::future::poll_fn;
+use futures::stream::FuturesUnordered;
+use futures::task::noop_waker_ref;
+use futures::FutureExt;
+use futures::Stream;
+use std::cell::Cell;
+use std::cell::RefCell;
+use std::collections::VecDeque;
+use std::future::ready;
+use std::future::Future;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::task::ready;
+use std::task::Context;
+use std::task::Poll;
+
+async fn poll_task<C: OpMappingContext>(
+  mut results: SubmissionQueueResults<
+    FuturesUnordered<FutureAllocation<PendingOp<C>>>,
+  >,
+  tx: Rc<RefCell<VecDeque<PendingOp<C>>>>,
+  tx_waker: Rc<UnsyncWaker>,
+) {
+  loop {
+    let ready = poll_fn(|cx| results.poll_next_unpin(cx)).await;
+    tx.borrow_mut().push_back(ready);
+    tx_waker.wake_by_ref();
+  }
+}
+
+#[derive(Default)]
+enum MaybeTask {
+  #[default]
+  Empty,
+  Task(Pin<Box<dyn Future<Output = ()>>>),
+  Handle(JoinHandle<()>),
+}
+
+/// [`OpDriver`] implementation built on a tokio [`JoinSet`].
+pub struct FuturesUnorderedDriver<
+  C: OpMappingContext + 'static = V8OpMappingContext,
+> {
+  len: Cell<usize>,
+  task: Cell<MaybeTask>,
+  task_set: Cell<bool>,
+  queue: SubmissionQueue<FuturesUnordered<FutureAllocation<PendingOp<C>>>>,
+  completed_ops: Rc<RefCell<VecDeque<PendingOp<C>>>>,
+  completed_waker: Rc<UnsyncWaker>,
+  arena: FutureArena,
+}
+
+impl<C: OpMappingContext + 'static> Drop for FuturesUnorderedDriver<C> {
+  fn drop(&mut self) {
+    if let MaybeTask::Handle(h) = self.task.take() {
+      h.abort()
+    }
+  }
+}
+
+impl<C: OpMappingContext> Default for FuturesUnorderedDriver<C> {
+  fn default() -> Self {
+    let (queue, results) = new_submission_queue();
+    let completed_ops = Rc::new(RefCell::new(VecDeque::with_capacity(128)));
+    let completed_waker = Rc::new(UnsyncWaker::default());
+    let task = MaybeTask::Task(Box::pin(poll_task(
+      results,
+      completed_ops.clone(),
+      completed_waker.clone(),
+    )))
+    .into();
+
+    Self {
+      len: Default::default(),
+      task,
+      task_set: Default::default(),
+      completed_ops,
+      queue,
+      completed_waker,
+      arena: Default::default(),
+    }
+  }
+}
+
+impl<C: OpMappingContext> FuturesUnorderedDriver<C> {
+  #[inline(always)]
+  fn ensure_task(&self) {
+    if !self.task_set.get() {
+      self.spawn_task();
+    }
+  }
+
+  #[inline(never)]
+  #[cold]
+  fn spawn_task(&self) {
+    let MaybeTask::Task(task) = self.task.replace(Default::default()) else {
+      unreachable!()
+    };
+    self.task.set(MaybeTask::Handle(spawn(task)));
+    self.task_set.set(true);
+  }
+
+  /// 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.ensure_task();
+    self.len.set(self.len.get() + 1);
+    self.queue.spawn(self.arena.allocate(task.map(map)));
+  }
+
+  /// Spawn a ready task that already has a [`PendingOp`].
+  #[inline(always)]
+  fn spawn_ready(&self, ready_op: PendingOp<C>) {
+    self.ensure_task();
+    self.len.set(self.len.get() + 1);
+    self.queue.spawn(self.arena.allocate(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: FutureAllocation<R>,
+    map: impl FnOnce(R) -> PendingOp<C> + 'static,
+  ) {
+    self.ensure_task();
+    self.len.set(self.len.get() + 1);
+    self.queue.spawn(self.arena.allocate(task.map(map)));
+  }
+}
+
+impl<C: OpMappingContext> OpDriver<C> for FuturesUnorderedDriver<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 = self.arena.allocate(op);
+      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 = self.arena.allocate(op);
+      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(
+    &self,
+    cx: &mut Context,
+  ) -> Poll<(PromiseId, OpId, OpResult<C>)> {
+    let mut ops = self.completed_ops.borrow_mut();
+    if ops.is_empty() {
+      self.completed_waker.register(cx.waker());
+      return Poll::Pending;
+    }
+    let item = ops.pop_front().unwrap();
+    let PendingOp(PendingOpInfo(promise_id, op_id), resp) = item;
+    self.len.set(self.len.get() - 1);
+    Poll::Ready((promise_id, op_id, resp))
+  }
+
+  #[inline(always)]
+  fn len(&self) -> usize {
+    self.len.get()
+  }
+}
+
+impl<F: Future<Output = R>, R> SubmissionQueueFutures for FuturesUnordered<F> {
+  type Future = F;
+  type Output = F::Output;
+
+  fn len(&self) -> usize {
+    self.len()
+  }
+
+  fn spawn(&mut self, f: Self::Future) {
+    self.push(f)
+  }
+
+  fn poll_next_unpin(&mut self, cx: &mut Context) -> Poll<Self::Output> {
+    Poll::Ready(ready!(Pin::new(self).poll_next(cx)).unwrap())
+  }
+}
+
+#[derive(Default)]
+struct Queue<F: SubmissionQueueFutures> {
+  queue: RefCell<F>,
+  item_waker: UnsyncWaker,
+}
+
+pub trait SubmissionQueueFutures: Default {
+  type Future: Future<Output = Self::Output>;
+  type Output;
+
+  fn len(&self) -> usize;
+  fn spawn(&mut self, f: Self::Future);
+  fn poll_next_unpin(&mut self, cx: &mut Context) -> Poll<Self::Output>;
+}
+
+pub struct SubmissionQueueResults<F: SubmissionQueueFutures> {
+  queue: Rc<Queue<F>>,
+}
+
+impl<F: SubmissionQueueFutures> SubmissionQueueResults<F> {
+  pub fn poll_next_unpin(&mut self, cx: &mut Context) -> Poll<F::Output> {
+    let mut queue = self.queue.queue.borrow_mut();
+    self.queue.item_waker.register(cx.waker());
+    if queue.len() == 0 {
+      return Poll::Pending;
+    }
+    queue.poll_next_unpin(cx)
+  }
+}
+
+pub struct SubmissionQueue<F: SubmissionQueueFutures> {
+  queue: Rc<Queue<F>>,
+}
+
+impl<F: SubmissionQueueFutures> SubmissionQueue<F> {
+  pub fn spawn(&self, f: F::Future) {
+    self.queue.queue.borrow_mut().spawn(f);
+    self.queue.item_waker.wake_by_ref();
+  }
+}
+
+/// Create a [`SubmissionQueue`] and [`SubmissionQueueResults`] that allow for submission of tasks
+/// and reception of task results. We may add work to the [`SubmissionQueue`] from any task, and the
+/// [`SubmissionQueueResults`] will be polled from a single location.
+pub fn new_submission_queue<F: SubmissionQueueFutures>(
+) -> (SubmissionQueue<F>, SubmissionQueueResults<F>) {
+  let queue: Rc<Queue<F>> = Default::default();
+  (
+    SubmissionQueue {
+      queue: queue.clone(),
+    },
+    SubmissionQueueResults { queue },
+  )
+}
diff --git a/core/runtime/op_driver/joinset_driver.rs b/core/runtime/op_driver/joinset_driver.rs
index 5a63588..e08ca18 100644
--- a/core/runtime/op_driver/joinset_driver.rs
+++ b/core/runtime/op_driver/joinset_driver.rs
@@ -3,9 +3,6 @@ use super::future_arena::FutureAllocation;
 use super::future_arena::FutureArena;
 use super::op_results::*;
 use super::OpDriver;
-use super::RetValMapper;
-use crate::ops::OpCtx;
-use crate::GetErrorClassFn;
 use crate::OpId;
 use crate::PromiseId;
 use anyhow::Error;
@@ -20,26 +17,34 @@ use std::task::Context;
 use std::task::Poll;
 
 /// [`OpDriver`] implementation built on a tokio [`JoinSet`].
-#[derive(Default)]
-pub struct JoinSetDriver {
-  pending_ops: RefCell<JoinSet<PendingOp>>,
+pub struct JoinSetDriver<C: OpMappingContext = V8OpMappingContext> {
+  pending_ops: RefCell<JoinSet<PendingOp<C>>>,
   arena: FutureArena,
 }
 
-impl JoinSetDriver {
+impl<C: OpMappingContext> Default for JoinSetDriver<C> {
+  fn default() -> Self {
+    Self {
+      pending_ops: Default::default(),
+      arena: 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 + '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) {
+  fn spawn_ready(&self, ready_op: PendingOp<C>) {
     self.pending_ops.borrow_mut().spawn(ready(ready_op));
   }
 
@@ -48,44 +53,13 @@ impl JoinSetDriver {
   fn spawn_polled<R>(
     &self,
     task: FutureAllocation<R>,
-    map: impl FnOnce(R) -> PendingOp + 'static,
+    map: impl FnOnce(R) -> PendingOp<C> + 'static,
   ) {
     self.pending_ops.borrow_mut().spawn(task.map(map));
   }
-
-  #[inline(always)]
-  fn pending_op_success<R: 'static>(
-    info: PendingOpInfo,
-    rv_map: RetValMapper<R>,
-    v: R,
-  ) -> PendingOp {
-    PendingOp(info, OpResult::new_value(v, rv_map))
-  }
-
-  #[inline(always)]
-  fn pending_op_failure<E: Into<Error> + 'static>(
-    info: PendingOpInfo,
-    get_class: GetErrorClassFn,
-    err: E,
-  ) -> PendingOp {
-    PendingOp(info, OpResult::Err(OpError::new(get_class, err.into())))
-  }
-
-  #[inline(always)]
-  fn pending_op_result<R: 'static, E: Into<Error> + 'static>(
-    info: PendingOpInfo,
-    rv_map: RetValMapper<R>,
-    get_class: GetErrorClassFn,
-    result: Result<R, E>,
-  ) -> PendingOp {
-    match result {
-      Ok(r) => Self::pending_op_success(info, rv_map, r),
-      Err(err) => Self::pending_op_failure(info, get_class, err),
-    }
-  }
 }
 
-impl OpDriver for JoinSetDriver {
+impl<C: OpMappingContext> OpDriver<C> for JoinSetDriver<C> {
   fn submit_op_fallible<
     R: 'static,
     E: Into<Error> + 'static,
@@ -93,19 +67,16 @@ impl OpDriver for JoinSetDriver {
     const DEFERRED: bool,
   >(
     &self,
-    ctx: &OpCtx,
+    op_id: OpId,
     promise_id: i32,
     op: impl Future<Output = Result<R, E>> + 'static,
-    rv_map: RetValMapper<R>,
+    rv_map: C::MappingFn<R>,
   ) -> Option<Result<R, E>> {
     {
-      let info = PendingOpInfo(promise_id, ctx.id, ctx.metrics_enabled());
-      let get_class = ctx.get_error_class_fn;
+      let info = PendingOpInfo(promise_id, op_id);
 
       if LAZY {
-        self.spawn_unpolled(op, move |r| {
-          Self::pending_op_result(info, rv_map, get_class, r)
-        });
+        self.spawn_unpolled(op, move |r| PendingOp::new(info, rv_map, r));
         return None;
       }
 
@@ -113,14 +84,12 @@ impl OpDriver for JoinSetDriver {
       // spin the event loop to get it.
       let mut pinned = self.arena.allocate(op);
       match pinned.poll_unpin(&mut Context::from_waker(noop_waker_ref())) {
-        Poll::Pending => self.spawn_polled(pinned, move |r| {
-          Self::pending_op_result(info, rv_map, get_class, r)
-        }),
+        Poll::Pending => {
+          self.spawn_polled(pinned, move |r| PendingOp::new(info, rv_map, r))
+        }
         Poll::Ready(res) => {
           if DEFERRED {
-            self.spawn_ready(Self::pending_op_result(
-              info, rv_map, get_class, res,
-            ))
+            self.spawn_ready(PendingOp::new(info, rv_map, res))
           } else {
             return Some(res);
           }
@@ -136,17 +105,15 @@ impl OpDriver for JoinSetDriver {
     const DEFERRED: bool,
   >(
     &self,
-    ctx: &OpCtx,
+    op_id: OpId,
     promise_id: i32,
     op: impl Future<Output = R> + 'static,
-    rv_map: RetValMapper<R>,
+    rv_map: C::MappingFn<R>,
   ) -> Option<R> {
     {
-      let info = PendingOpInfo(promise_id, ctx.id, ctx.metrics_enabled());
+      let info = PendingOpInfo(promise_id, op_id);
       if LAZY {
-        self.spawn_unpolled(op, move |r| {
-          Self::pending_op_success(info, rv_map, r)
-        });
+        self.spawn_unpolled(op, move |r| PendingOp::ok(info, rv_map, r));
         return None;
       }
 
@@ -154,12 +121,12 @@ impl OpDriver for JoinSetDriver {
       // spin the event loop to get it.
       let mut pinned = self.arena.allocate(op);
       match pinned.poll_unpin(&mut Context::from_waker(noop_waker_ref())) {
-        Poll::Pending => self.spawn_polled(pinned, move |res| {
-          Self::pending_op_success(info, rv_map, res)
-        }),
+        Poll::Pending => {
+          self.spawn_polled(pinned, move |res| PendingOp::ok(info, rv_map, res))
+        }
         Poll::Ready(res) => {
           if DEFERRED {
-            self.spawn_ready(Self::pending_op_success(info, rv_map, res))
+            self.spawn_ready(PendingOp::ok(info, rv_map, res))
           } else {
             return Some(res);
           }
@@ -174,39 +141,16 @@ impl OpDriver for JoinSetDriver {
   fn poll_ready<'s>(
     &self,
     cx: &mut Context,
-    scope: &mut v8::HandleScope<'s>,
-  ) -> Poll<(
-    PromiseId,
-    OpId,
-    bool,
-    Result<v8::Local<'s, v8::Value>, v8::Local<'s, v8::Value>>,
-  )> {
+  ) -> Poll<(PromiseId, OpId, OpResult<C>)> {
     let item = ready!(self.pending_ops.borrow_mut().poll_join_next(cx));
-    let PendingOp(PendingOpInfo(promise_id, op_id, metrics_event), resp) =
-      match item {
-        Ok(x) => x,
-        Err(_) => {
-          // If this task is really errored, things could be pretty bad
-          panic!("Unrecoverable error: op panicked");
-        }
-      };
-
-    let was_error = matches!(resp, OpResult::Err(_));
-    let res = match resp.into_v8(scope) {
-      Ok(v) => {
-        if was_error {
-          Err(v)
-        } else {
-          Ok(v)
-        }
+    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");
       }
-      Err(e) => Err(
-        OpResult::Err(OpError::new(&|_| "TypeError", e.into()))
-          .into_v8(scope)
-          .unwrap(),
-      ),
     };
-    Poll::Ready((promise_id, op_id, metrics_event, res))
+    Poll::Ready((promise_id, op_id, resp))
   }
 
   #[inline(always)]
diff --git a/core/runtime/op_driver/mod.rs b/core/runtime/op_driver/mod.rs
index 4826f38..1db4ed1 100644
--- a/core/runtime/op_driver/mod.rs
+++ b/core/runtime/op_driver/mod.rs
@@ -1,5 +1,4 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-use crate::ops::OpCtx;
 use crate::OpId;
 use crate::PromiseId;
 use anyhow::Error;
@@ -9,31 +8,70 @@ 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 type RetValMapper<R> =
-  for<'r> fn(
-    &mut v8::HandleScope<'r>,
-    R,
-  ) -> Result<v8::Local<'r, v8::Value>, serde_v8::Error>;
+pub use self::op_results::OpMappingContext;
+pub use self::op_results::OpResult;
+pub use self::op_results::V8OpMappingContext;
+pub use self::op_results::V8RetValMapper;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum OpScheduling {
+  Eager,
+  Lazy,
+  Deferred,
+}
 
 /// `OpDriver` encapsulates the interface for handling operations within Deno's runtime.
 ///
 /// This trait defines methods for submitting ops and polling readiness inside of the
 /// event loop.
-pub(crate) trait OpDriver: Default {
+///
+/// Ops are always submitted with a mapping function that can convert the output of the
+/// op to the underlying [`OpMappingContext`] output type. In the case of V8, this is a
+/// function that creates [`v8::Local`] values.
+///
+/// The driver takes an optional [`OpMappingContext`] implementation, which defaults to
+/// one compatible with v8. This is used solely for testing purposes.
+pub(crate) trait OpDriver<C: OpMappingContext = V8OpMappingContext>:
+  Default
+{
   /// Submits an operation that is expected to complete successfully without errors.
   fn submit_op_infallible<R: 'static, const LAZY: bool, const DEFERRED: bool>(
     &self,
-    ctx: &OpCtx,
+    op_id: OpId,
     promise_id: i32,
     op: impl Future<Output = R> + 'static,
-    rv_map: RetValMapper<R>,
+    rv_map: C::MappingFn<R>,
   ) -> Option<R>;
 
+  /// Submits an operation that is expected to complete successfully without errors.
+  #[inline(always)]
+  fn submit_op_infallible_scheduling<R: 'static>(
+    &self,
+    scheduling: OpScheduling,
+    op_id: OpId,
+    promise_id: i32,
+    op: impl Future<Output = R> + 'static,
+    rv_map: C::MappingFn<R>,
+  ) -> Option<R> {
+    match scheduling {
+      OpScheduling::Eager => self
+        .submit_op_infallible::<R, false, false>(op_id, promise_id, op, rv_map),
+      OpScheduling::Lazy => self
+        .submit_op_infallible::<R, true, false>(op_id, promise_id, op, rv_map),
+      OpScheduling::Deferred => self
+        .submit_op_infallible::<R, false, true>(op_id, promise_id, op, rv_map),
+    }
+  }
+
   /// Submits an operation that may produce errors during execution.
   ///
   /// This method is similar to `submit_op_infallible` but is used when the op
@@ -45,24 +83,283 @@ pub(crate) trait OpDriver: Default {
     const DEFERRED: bool,
   >(
     &self,
-    ctx: &OpCtx,
+    op_id: OpId,
     promise_id: i32,
     op: impl Future<Output = Result<R, E>> + 'static,
-    rv_map: RetValMapper<R>,
+    rv_map: C::MappingFn<R>,
   ) -> Option<Result<R, E>>;
 
+  /// Submits an operation that is expected to complete successfully without errors.
+  #[inline(always)]
+  #[allow(clippy::too_many_arguments)]
+  fn submit_op_fallible_scheduling<R: 'static, E: Into<Error> + 'static>(
+    &self,
+    scheduling: OpScheduling,
+    op_id: OpId,
+    promise_id: i32,
+    op: impl Future<Output = Result<R, E>> + 'static,
+    rv_map: C::MappingFn<R>,
+  ) -> Option<Result<R, E>> {
+    match scheduling {
+      OpScheduling::Eager => self.submit_op_fallible::<R, E, false, false>(
+        op_id, promise_id, op, rv_map,
+      ),
+      OpScheduling::Lazy => self
+        .submit_op_fallible::<R, E, true, false>(op_id, promise_id, op, rv_map),
+      OpScheduling::Deferred => self
+        .submit_op_fallible::<R, E, false, true>(op_id, promise_id, op, rv_map),
+    }
+  }
+
   #[allow(clippy::type_complexity)]
   /// Polls the readiness of the op driver.
-  fn poll_ready<'s>(
+  fn poll_ready(
     &self,
     cx: &mut Context,
-    scope: &mut v8::HandleScope<'s>,
-  ) -> Poll<(
-    PromiseId,
-    OpId,
-    bool,
-    Result<v8::Local<'s, v8::Value>, v8::Local<'s, v8::Value>>,
-  )>;
+  ) -> Poll<(PromiseId, OpId, OpResult<C>)>;
 
+  /// Return the number of futures currently being polled.
   fn len(&self) -> usize;
 }
+
+#[cfg(test)]
+mod tests {
+  use crate::GetErrorClassFn;
+
+  use super::op_results::*;
+  use super::*;
+  use bit_set::BitSet;
+  use rstest::rstest;
+  use std::future::poll_fn;
+
+  struct TestMappingContext {}
+  impl<'s> OpMappingContextLifetime<'s> for TestMappingContext {
+    type Context = ();
+    type Result = String;
+    type MappingError = anyhow::Error;
+
+    fn map_error(
+      _context: &mut Self::Context,
+      err: Error,
+      _get_error_class_fn: GetErrorClassFn,
+    ) -> UnmappedResult<'s, Self> {
+      Ok(format!("{err:?}"))
+    }
+
+    fn map_mapping_error(
+      _context: &mut Self::Context,
+      err: Self::MappingError,
+    ) -> Self::Result {
+      format!("{err:?}")
+    }
+  }
+  impl OpMappingContext for TestMappingContext {
+    type MappingFn<R: 'static> = for<'s> fn(R) -> Result<String, anyhow::Error>;
+    fn erase_mapping_fn<R: 'static>(f: Self::MappingFn<R>) -> *const fn() {
+      f as _
+    }
+
+    fn unerase_mapping_fn<'s, R: 'static>(
+      f: *const fn(),
+      _context: &mut <Self as OpMappingContextLifetime<'s>>::Context,
+      r: R,
+    ) -> UnmappedResult<'s, Self> {
+      let f: Self::MappingFn<R> = unsafe { std::mem::transmute(f) };
+      f(r)
+    }
+  }
+
+  fn submit_task(
+    driver: &impl OpDriver<TestMappingContext>,
+    scheduling: OpScheduling,
+    id: usize,
+    op: impl Future<Output = i32> + 'static,
+  ) {
+    assert_eq!(
+      None,
+      driver.submit_op_infallible_scheduling(
+        scheduling,
+        1234,
+        id as _,
+        op,
+        |r| { Ok(format!("{r}")) }
+      )
+    );
+  }
+
+  fn submit_task_eager_ready(
+    driver: &impl OpDriver<TestMappingContext>,
+    id: usize,
+    op: impl Future<Output = i32> + 'static,
+    result: i32,
+  ) {
+    assert_eq!(
+      Some(result),
+      driver.submit_op_infallible_scheduling(
+        OpScheduling::Eager,
+        1234,
+        id as _,
+        op,
+        |r| { Ok(format!("{r}")) }
+      )
+    );
+  }
+
+  async fn reap_task(
+    driver: &impl OpDriver<TestMappingContext>,
+    bitset: &mut BitSet,
+    expected: &str,
+  ) {
+    let (promise_id, op_id, result) = poll_fn(|cx| driver.poll_ready(cx)).await;
+    assert!(bitset.insert(promise_id as usize));
+    assert_eq!(1234, op_id);
+    assert_eq!(expected, &(result.unwrap(&mut (), &|_| "Error").unwrap()));
+  }
+
+  #[rstest]
+  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
+  #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
+  fn test_driver<D: OpDriver<TestMappingContext>>(
+    #[case] driver: D,
+    #[values(2, 16)] count: usize,
+    #[values(OpScheduling::Eager, OpScheduling::Lazy, OpScheduling::Deferred)]
+    scheduling: OpScheduling,
+  ) {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+      .build()
+      .unwrap();
+    runtime.block_on(async {
+      for i in 0..count {
+        if scheduling == OpScheduling::Eager {
+          submit_task_eager_ready(&driver, i, async { 1 }, 1);
+        } else {
+          submit_task(&driver, scheduling, i, async { 1 });
+        }
+      }
+      if scheduling != OpScheduling::Eager {
+        let mut bitset = BitSet::default();
+        for i in 0..count {
+          assert_eq!(driver.len(), count - i);
+          reap_task(&driver, &mut bitset, "1").await;
+        }
+        assert_eq!(bitset.len(), count);
+      }
+    });
+  }
+
+  #[rstest]
+  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
+  #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
+  fn test_driver_yield<D: OpDriver<TestMappingContext>>(
+    #[case] driver: D,
+    #[values(2, 16)] count: usize,
+    #[values(1, 5)] outer: usize,
+    #[values(OpScheduling::Eager, OpScheduling::Lazy, OpScheduling::Deferred)]
+    scheduling: OpScheduling,
+  ) {
+    async fn task() -> i32 {
+      let v = [0_u8, 1, 2, 3];
+      for i in &v {
+        for _ in 0..*i {
+          tokio::task::yield_now().await;
+        }
+      }
+      v.len() as _
+    }
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+      .build()
+      .unwrap();
+    runtime.block_on(async {
+      for _ in 0..outer {
+        for i in 0..count {
+          submit_task(&driver, scheduling, i, task());
+        }
+        let mut bitset = BitSet::default();
+        for i in 0..count {
+          assert_eq!(driver.len(), count - i);
+          reap_task(&driver, &mut bitset, "4").await;
+        }
+        assert_eq!(bitset.len(), count);
+      }
+    });
+  }
+
+  #[rstest]
+  #[case::joinset(JoinSetDriver::<TestMappingContext>::default())]
+  #[case::futures_unordered(FuturesUnorderedDriver::<TestMappingContext>::default())]
+  fn test_driver_large<D: OpDriver<TestMappingContext>>(
+    #[case] driver: D,
+    #[values(2, 16)] count: usize,
+    #[values(OpScheduling::Eager, OpScheduling::Lazy, OpScheduling::Deferred)]
+    scheduling: OpScheduling,
+  ) {
+    async fn task() -> i32 {
+      let mut v = [0; 10 * 1024];
+      #[allow(clippy::needless_range_loop)]
+      for i in 0..10 {
+        tokio::task::yield_now().await;
+        v[i] = 1;
+      }
+      let mut s = 0;
+      for i in v {
+        s += i;
+      }
+      s
+    }
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+      .build()
+      .unwrap();
+    runtime.block_on(async {
+      for i in 0..count {
+        submit_task(&driver, scheduling, i, task());
+      }
+      let mut bitset = BitSet::default();
+      for i in 0..count {
+        assert_eq!(driver.len(), count - i);
+        reap_task(&driver, &mut bitset, "10").await;
+      }
+      assert_eq!(bitset.len(), count);
+    });
+  }
+
+  #[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,
+    #[values(2, 16)] count: usize,
+    #[values(OpScheduling::Eager, OpScheduling::Lazy, OpScheduling::Deferred)]
+    scheduling: OpScheduling,
+  ) {
+    async fn task() -> i32 {
+      use tokio::net::TcpSocket;
+      let socket = TcpSocket::new_v4().unwrap();
+      socket.bind("127.0.0.1:0".parse().unwrap()).unwrap();
+      let listen = socket.listen(1).unwrap();
+      let connect = TcpSocket::new_v4().unwrap();
+      let f = tokio::spawn(connect.connect(listen.local_addr().unwrap()));
+      listen.accept().await.unwrap();
+      f.await.unwrap().unwrap();
+      1
+    }
+
+    let runtime = tokio::runtime::Builder::new_current_thread()
+      .enable_io()
+      .build()
+      .unwrap();
+    runtime.block_on(async {
+      for i in 0..count {
+        submit_task(&driver, scheduling, i, task());
+      }
+      let mut bitset = BitSet::default();
+      for i in 0..count {
+        assert_eq!(driver.len(), count - i);
+        reap_task(&driver, &mut bitset, "1").await;
+      }
+      assert_eq!(bitset.len(), count);
+    });
+  }
+}
diff --git a/core/runtime/op_driver/op_results.rs b/core/runtime/op_driver/op_results.rs
index dc21b5f..1a7e2db 100644
--- a/core/runtime/op_driver/op_results.rs
+++ b/core/runtime/op_driver/op_results.rs
@@ -1,6 +1,5 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 use super::erased_future::TypeErased;
-use super::RetValMapper;
 use crate::GetErrorClassFn;
 use crate::OpId;
 use crate::PromiseId;
@@ -9,67 +8,208 @@ use serde::Serialize;
 
 const MAX_RESULT_SIZE: usize = 32;
 
-pub struct PendingOp(pub PendingOpInfo, pub OpResult);
+pub type MappedResult<'s, C> = Result<
+  <C as OpMappingContextLifetime<'s>>::Result,
+  <C as OpMappingContextLifetime<'s>>::Result,
+>;
+pub type UnmappedResult<'s, C> = Result<
+  <C as OpMappingContextLifetime<'s>>::Result,
+  <C as OpMappingContextLifetime<'s>>::MappingError,
+>;
 
-pub struct PendingOpInfo(pub PromiseId, pub OpId, pub bool);
+pub trait OpMappingContextLifetime<'s> {
+  type Context: 's;
+  type Result: 's;
+  type MappingError: 's;
+
+  fn map_error(
+    context: &mut Self::Context,
+    err: Error,
+    get_error_class_fn: GetErrorClassFn,
+  ) -> UnmappedResult<'s, Self>;
+  fn map_mapping_error(
+    context: &mut Self::Context,
+    err: Self::MappingError,
+  ) -> Self::Result;
+}
+
+/// A type that allows for arbitrary mapping systems for op output with lifetime
+/// control. We add this extra layer of generics because we want to test our drivers
+/// without requiring V8.
+pub trait OpMappingContext:
+  for<'s> OpMappingContextLifetime<'s> + 'static
+{
+  type MappingFn<R: 'static>;
+
+  fn erase_mapping_fn<R: 'static>(f: Self::MappingFn<R>) -> *const fn();
+  fn unerase_mapping_fn<'s, R: 'static>(
+    f: *const fn(),
+    scope: &mut <Self as OpMappingContextLifetime<'s>>::Context,
+    r: R,
+  ) -> UnmappedResult<'s, Self>;
+}
+
+#[derive(Default)]
+pub struct V8OpMappingContext {}
+
+pub type V8RetValMapper<R> =
+  for<'r> fn(
+    &mut v8::HandleScope<'r>,
+    R,
+  ) -> Result<v8::Local<'r, v8::Value>, serde_v8::Error>;
+
+impl<'s> OpMappingContextLifetime<'s> for V8OpMappingContext {
+  type Context = v8::HandleScope<'s>;
+  type Result = v8::Local<'s, v8::Value>;
+  type MappingError = serde_v8::Error;
+
+  #[inline(always)]
+  fn map_error(
+    scope: &mut v8::HandleScope<'s>,
+    err: Error,
+    get_error_class_fn: GetErrorClassFn,
+  ) -> UnmappedResult<'s, Self> {
+    serde_v8::to_v8(scope, OpError::new(get_error_class_fn, err))
+  }
+
+  fn map_mapping_error(
+    scope: &mut v8::HandleScope<'s>,
+    err: Self::MappingError,
+  ) -> v8::Local<'s, v8::Value> {
+    serde_v8::to_v8(scope, OpError::new(&|_| "TypeError", err.into())).unwrap()
+  }
+}
+
+impl OpMappingContext for V8OpMappingContext {
+  type MappingFn<R: 'static> = V8RetValMapper<R>;
+
+  #[inline(always)]
+  fn erase_mapping_fn<R: 'static>(f: Self::MappingFn<R>) -> *const fn() {
+    f as _
+  }
+  #[inline(always)]
+  fn unerase_mapping_fn<'s, R: 'static>(
+    f: *const fn(),
+    scope: &mut <Self as OpMappingContextLifetime<'s>>::Context,
+    r: R,
+  ) -> Result<v8::Local<'s, v8::Value>, serde_v8::Error> {
+    let f: Self::MappingFn<R> = unsafe { std::mem::transmute(f) };
+    f(scope, r)
+  }
+}
+
+/// [`PendingOp`] holds the metadata and result (success or failure) of a submitted
+/// and completed op.
+pub struct PendingOp<C: OpMappingContext>(pub PendingOpInfo, pub OpResult<C>);
+
+impl<C: OpMappingContext> PendingOp<C> {
+  #[inline(always)]
+  pub fn new<R: 'static, E: Into<Error> + 'static>(
+    info: PendingOpInfo,
+    rv_map: C::MappingFn<R>,
+    result: Result<R, E>,
+  ) -> Self {
+    match result {
+      Ok(r) => PendingOp(info, OpResult::new_value(r, rv_map)),
+      Err(err) => PendingOp(info, OpResult::Err(err.into())),
+    }
+  }
+
+  #[inline(always)]
+  pub fn ok<R: 'static>(
+    info: PendingOpInfo,
+    rv_map: C::MappingFn<R>,
+    r: R,
+  ) -> Self {
+    PendingOp(info, OpResult::new_value(r, rv_map))
+  }
+}
+
+pub struct PendingOpInfo(pub PromiseId, pub OpId);
+
+type MapRawFn<C> = for<'a> fn(
+  _lifetime: &'a (),
+  scope: &mut <C as OpMappingContextLifetime<'a>>::Context,
+  rv_map: *const fn(),
+  value: TypeErased<MAX_RESULT_SIZE>,
+) -> UnmappedResult<'a, C>;
 
 #[allow(clippy::type_complexity)]
-pub struct OpValue {
+pub struct OpValue<C: OpMappingContext> {
   value: TypeErased<MAX_RESULT_SIZE>,
   rv_map: *const fn(),
-  map_fn: for<'a> fn(
-    scope: &mut v8::HandleScope<'a>,
-    rv_map: *const fn(),
-    value: TypeErased<MAX_RESULT_SIZE>,
-  ) -> Result<v8::Local<'a, v8::Value>, serde_v8::Error>,
+  map_fn: MapRawFn<C>,
 }
 
-impl OpValue {
-  pub fn new<R: 'static>(rv_map: RetValMapper<R>, v: R) -> Self {
+impl<C: OpMappingContext> OpValue<C> {
+  pub fn new<R: 'static>(rv_map: C::MappingFn<R>, v: R) -> Self {
     Self {
       value: TypeErased::new(v),
-      rv_map: rv_map as _,
-      map_fn: |scope, rv_map, erased| unsafe {
+      rv_map: C::erase_mapping_fn(rv_map),
+      map_fn: |_, scope, rv_map, erased| unsafe {
         let r = erased.take();
-        let rv_map: RetValMapper<R> = std::mem::transmute(rv_map);
-        rv_map(scope, r)
+        C::unerase_mapping_fn::<R>(rv_map, scope, r)
       },
     }
   }
 }
 
-pub type ValueLargeFn =
-  dyn for<'a> FnOnce(
-    &mut v8::HandleScope<'a>,
-  ) -> Result<v8::Local<'a, v8::Value>, serde_v8::Error>;
+pub trait ValueLargeFn<C: OpMappingContext> {
+  fn unwrap<'a>(
+    self: Box<Self>,
+    ctx: &mut <C as OpMappingContextLifetime<'a>>::Context,
+  ) -> UnmappedResult<'a, C>;
+}
+
+struct ValueLarge<C: OpMappingContext, R: 'static> {
+  v: R,
+  rv_map: C::MappingFn<R>,
+}
+
+impl<C: OpMappingContext, R> ValueLargeFn<C> for ValueLarge<C, R> {
+  fn unwrap<'a>(
+    self: Box<Self>,
+    ctx: &mut <C as OpMappingContextLifetime<'a>>::Context,
+  ) -> UnmappedResult<'a, C> {
+    C::unerase_mapping_fn(C::erase_mapping_fn(self.rv_map), ctx, self.v)
+  }
+}
 
-pub enum OpResult {
+pub enum OpResult<C: OpMappingContext> {
   /// Errors.
-  Err(OpError),
+  Err(Error),
   /// For small ops, we include them in an erased type container.
-  Value(OpValue),
+  Value(OpValue<C>),
   /// For ops that return "large" results (> MAX_RESULT_SIZE bytes) we just box a function
   /// that can turn it into a v8 value.
-  ValueLarge(Box<ValueLargeFn>),
+  ValueLarge(Box<dyn ValueLargeFn<C>>),
 }
 
-impl OpResult {
-  pub fn new_value<R: 'static>(v: R, rv_map: RetValMapper<R>) -> Self {
+impl<C: OpMappingContext> OpResult<C> {
+  pub fn new_value<R: 'static>(v: R, rv_map: C::MappingFn<R>) -> Self {
     if std::mem::size_of::<R>() > MAX_RESULT_SIZE {
-      OpResult::ValueLarge(Box::new(move |scope| rv_map(scope, v)))
+      OpResult::ValueLarge(Box::new(ValueLarge { v, rv_map }))
     } else {
       OpResult::Value(OpValue::new(rv_map, v))
     }
   }
 
-  pub fn into_v8<'a>(
+  pub fn unwrap<'a>(
     self,
-    scope: &mut v8::HandleScope<'a>,
-  ) -> Result<v8::Local<'a, v8::Value>, serde_v8::Error> {
-    match self {
-      Self::Err(err) => serde_v8::to_v8(scope, err),
-      Self::Value(f) => (f.map_fn)(scope, f.rv_map, f.value),
-      Self::ValueLarge(f) => f(scope),
+    context: &mut <C as OpMappingContextLifetime<'a>>::Context,
+    get_error_class_fn: GetErrorClassFn,
+  ) -> MappedResult<'a, C> {
+    let (success, res) = match self {
+      Self::Err(err) => (false, C::map_error(context, err, get_error_class_fn)),
+      Self::Value(f) => (true, (f.map_fn)(&(), context, f.rv_map, f.value)),
+      Self::ValueLarge(f) => (true, f.unwrap(context)),
+    };
+    match (success, res) {
+      (true, Ok(x)) => Ok(x),
+      (false, Ok(x)) => Err(x),
+      (_, Err(err)) => Err(
+        <C as OpMappingContextLifetime<'a>>::map_mapping_error(context, err),
+      ),
     }
   }
 }
diff --git a/core/runtime/ops.rs b/core/runtime/ops.rs
index 550ed7a..1a892c1 100644
--- a/core/runtime/ops.rs
+++ b/core/runtime/ops.rs
@@ -13,13 +13,24 @@ use std::ptr::NonNull;
 use v8::WriteOptions;
 
 use super::op_driver::OpDriver;
-use super::op_driver::RetValMapper;
+use super::op_driver::OpScheduling;
+use super::op_driver::V8RetValMapper;
 
 /// The default string buffer size on the stack that prevents mallocs in some
 /// string functions. Keep in mind that Windows only offers 1MB stacks by default,
 /// so this is a limited resource!
 pub const STRING_STACK_BUFFER_SIZE: usize = 1024 * 8;
 
+fn op_scheduling(lazy: bool, deferred: bool) -> OpScheduling {
+  if lazy {
+    OpScheduling::Lazy
+  } else if deferred {
+    OpScheduling::Deferred
+  } else {
+    OpScheduling::Eager
+  }
+}
+
 #[inline(always)]
 pub fn map_async_op_infallible<R: 'static>(
   ctx: &OpCtx,
@@ -27,19 +38,16 @@ pub fn map_async_op_infallible<R: 'static>(
   deferred: bool,
   promise_id: i32,
   op: impl Future<Output = R> + 'static,
-  rv_map: RetValMapper<R>,
+  rv_map: V8RetValMapper<R>,
 ) -> Option<R> {
   let pending_ops = &ctx.context_state().pending_ops;
-  if lazy {
-    pending_ops
-      .submit_op_infallible::<_, true, false>(ctx, promise_id, op, rv_map)
-  } else if deferred {
-    pending_ops
-      .submit_op_infallible::<_, false, true>(ctx, promise_id, op, rv_map)
-  } else {
-    pending_ops
-      .submit_op_infallible::<_, false, false>(ctx, promise_id, op, rv_map)
-  }
+  pending_ops.submit_op_infallible_scheduling(
+    op_scheduling(lazy, deferred),
+    ctx.id,
+    promise_id,
+    op,
+    rv_map,
+  )
 }
 
 #[inline(always)]
@@ -49,19 +57,16 @@ pub fn map_async_op_fallible<R: 'static, E: Into<Error> + 'static>(
   deferred: bool,
   promise_id: i32,
   op: impl Future<Output = Result<R, E>> + 'static,
-  rv_map: RetValMapper<R>,
+  rv_map: V8RetValMapper<R>,
 ) -> Option<Result<R, E>> {
   let pending_ops = &ctx.context_state().pending_ops;
-  if lazy {
-    pending_ops
-      .submit_op_fallible::<_, _, true, false>(ctx, promise_id, op, rv_map)
-  } else if deferred {
-    pending_ops
-      .submit_op_fallible::<_, _, false, true>(ctx, promise_id, op, rv_map)
-  } else {
-    pending_ops
-      .submit_op_fallible::<_, _, false, false>(ctx, promise_id, op, rv_map)
-  }
+  pending_ops.submit_op_fallible_scheduling(
+    op_scheduling(lazy, deferred),
+    ctx.id,
+    promise_id,
+    op,
+    rv_map,
+  )
 }
 
 macro_rules! try_number_some {
diff --git a/core/runtime/snapshot_util.rs b/core/runtime/snapshot_util.rs
index ca0ff45..06e312f 100644
--- a/core/runtime/snapshot_util.rs
+++ b/core/runtime/snapshot_util.rs
@@ -171,9 +171,11 @@ fn data_error_to_panic(err: v8::DataError) -> ! {
 pub(crate) struct SnapshottedData {
   pub module_map_data: v8::Global<v8::Array>,
   pub module_handles: Vec<v8::Global<v8::Module>>,
+  pub js_handled_promise_rejection_cb: Option<v8::Global<v8::Function>>,
 }
 
 static MODULE_MAP_CONTEXT_DATA_INDEX: usize = 0;
+static JS_HANDLED_PROMISE_REJECTION_CB_DATA_INDEX: usize = 1;
 
 pub(crate) fn get_snapshotted_data(
   scope: &mut v8::HandleScope<()>,
@@ -199,9 +201,24 @@ pub(crate) fn get_snapshotted_data(
     info_data.length()
   };
 
+  let js_handled_promise_rejection_cb = match scope
+    .get_context_data_from_snapshot_once::<v8::Value>(
+      JS_HANDLED_PROMISE_REJECTION_CB_DATA_INDEX,
+    ) {
+    Ok(val) => {
+      if val.is_undefined() {
+        None
+      } else {
+        let fn_: v8::Local<v8::Function> = val.try_into().unwrap();
+        Some(v8::Global::new(&mut scope, fn_))
+      }
+    }
+    Err(err) => data_error_to_panic(err),
+  };
+
   // Over allocate so executing a few scripts doesn't have to resize this vec.
   let mut module_handles = Vec::with_capacity(next_module_id as usize + 16);
-  for i in 1..=next_module_id {
+  for i in 2..=next_module_id + 1 {
     match scope.get_context_data_from_snapshot_once::<v8::Module>(i as usize) {
       Ok(val) => {
         let module_global = v8::Global::new(&mut scope, val);
@@ -214,6 +231,7 @@ pub(crate) fn get_snapshotted_data(
   SnapshottedData {
     module_map_data: v8::Global::new(&mut scope, val),
     module_handles,
+    js_handled_promise_rejection_cb,
   }
 }
 
@@ -227,11 +245,23 @@ pub(crate) fn set_snapshotted_data(
   let offset = scope.add_context_data(local_context, local_data);
   assert_eq!(offset, MODULE_MAP_CONTEXT_DATA_INDEX);
 
+  let local_handled_promise_rejection_cb: v8::Local<v8::Value> =
+    if let Some(handled_promise_rejection_cb) =
+      snapshotted_data.js_handled_promise_rejection_cb
+    {
+      v8::Local::new(scope, handled_promise_rejection_cb).into()
+    } else {
+      v8::undefined(scope).into()
+    };
+  let offset =
+    scope.add_context_data(local_context, local_handled_promise_rejection_cb);
+  assert_eq!(offset, JS_HANDLED_PROMISE_REJECTION_CB_DATA_INDEX);
+
   for (index, handle) in snapshotted_data.module_handles.into_iter().enumerate()
   {
     let module_handle = v8::Local::new(scope, handle);
     let offset = scope.add_context_data(local_context, module_handle);
-    assert_eq!(offset, index + 1);
+    assert_eq!(offset, index + 2);
   }
 }
 


Test patch:
diff --git a/core/modules/tests.rs b/core/modules/tests.rs
index 9418f51..9674563 100644
--- a/core/modules/tests.rs
+++ b/core/modules/tests.rs
@@ -13,12 +13,11 @@ use crate::resolve_url;
 use crate::runtime::JsRuntime;
 use crate::runtime::JsRuntimeForSnapshot;
 use crate::FastString;
-use crate::ModuleCode;
+use crate::ModuleCodeString;
 use crate::ModuleSource;
 use crate::ModuleSourceFuture;
 use crate::ModuleSpecifier;
 use crate::ModuleType;
-use crate::OpState;
 use crate::ResolutionKind;
 use crate::RuntimeOptions;
 use crate::Snapshot;
@@ -39,7 +38,6 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::task::Context;
 use std::task::Poll;
-use tokio::sync::Barrier;
 use tokio::task::LocalSet;
 use url::Url;
 
@@ -659,54 +657,6 @@ fn test_validate_import_attributes_default() {
   }
 }
 
-#[tokio::test]
-async fn dyn_import_op() {
-  #[op2(async)]
-  async fn op_test() {
-    tokio::task::yield_now().await;
-  }
-
-  #[op2(async)]
-  async fn op_wait(state: Rc<RefCell<OpState>>) {
-    eprintln!("op_wait!");
-    let barrier: Rc<Barrier> = state.borrow().borrow::<Rc<Barrier>>().clone();
-    barrier.wait().await;
-  }
-
-  deno_core::extension!(test_ext,
-    ops = [op_test, op_wait],
-    options = { count: Barrier },
-    state = |state, options| {
-      state.put(Rc::new(options.count));
-    },
-  );
-
-  let loader = Rc::new(TestingModuleLoader::new(StaticModuleLoader::new([
-    (Url::parse("file:///main.js").unwrap(), ascii_str!("const { op_wait } = Deno.core.ensureFastOps(); (async () => { await import('./dynamic.js'); await op_wait(); })();")),
-    (Url::parse("file:///dynamic.js").unwrap(), ascii_str!("const { op_test } = Deno.core.ensureFastOps(); await op_test();")),
-  ])));
-  let barrier = Barrier::new(2);
-  let mut runtime = JsRuntime::new(RuntimeOptions {
-    module_loader: Some(loader.clone()),
-    extensions: vec![test_ext::init_ops(barrier)],
-    ..Default::default()
-  });
-  let id = runtime
-    .load_main_module(
-      &Url::parse("file:///root.js").unwrap(),
-      Some(ascii_str!(
-        "import 'file:///main.js'; const { op_wait } = Deno.core.ensureFastOps(); await op_wait();"
-      )),
-    )
-    .await
-    .unwrap();
-  let f = runtime.mod_evaluate(id);
-  poll_fn(|cx| runtime.poll_event_loop(cx, Default::default()))
-    .await
-    .unwrap();
-  _ = f.await;
-}
-
 #[tokio::test]
 async fn dyn_import_err() {
   let loader = Rc::new(TestingModuleLoader::new(StaticModuleLoader::new([])));
@@ -1216,7 +1166,7 @@ fn dynamic_imports_snapshot() {
 #[test]
 fn import_meta_snapshot() {
   let snapshot = {
-    const MAIN_WITH_CODE_SRC: ModuleCode = ascii_str!(
+    const MAIN_WITH_CODE_SRC: ModuleCodeString = ascii_str!(
       r#"
   if (import.meta.url != 'file:///main_with_code.js') throw Error();
   globalThis.meta = import.meta;
@@ -1344,7 +1294,7 @@ async fn no_duplicate_loads() {
           .unwrap()
       };
       let module_source = ModuleSource {
-        code: ModuleSourceCode::String(ModuleCode::from(source_code)),
+        code: ModuleSourceCode::String(ModuleCodeString::from(source_code)),
         module_type: ModuleType::JavaScript,
         module_url_specified: module_specifier.clone().into(),
         module_url_found: found_specifier.map(|s| s.into()),
diff --git a/core/runtime/tests/ops.rs b/core/runtime/tests/ops.rs
index 0f7db50..384bdeb 100644
--- a/core/runtime/tests/ops.rs
+++ b/core/runtime/tests/ops.rs
@@ -15,7 +15,6 @@ use std::cell::RefCell;
 use std::rc::Rc;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
-use tokio::sync::Barrier;
 use url::Url;
 
 #[tokio::test]
@@ -354,41 +353,6 @@ fn ops_in_js_have_proper_names() {
   runtime.execute_script_static("test", src).unwrap();
 }
 
-/// Dipatch more promises than the ring can hold, but wait for them all to queue up first.
-#[tokio::test]
-async fn test_dispatch_many_ops() {
-  #[op2(async)]
-  async fn op_wait(state: Rc<RefCell<OpState>>) {
-    let barrier: Rc<Barrier> = state.borrow().borrow::<Rc<Barrier>>().clone();
-    barrier.wait().await;
-  }
-
-  deno_core::extension!(test_ext,
-    ops = [op_wait],
-    options = { count: Barrier },
-    state = |state, options| {
-      state.put(Rc::new(options.count));
-    }
-  );
-
-  let mut runtime = JsRuntime::new(RuntimeOptions {
-    extensions: vec![test_ext::init_ops(Barrier::new(5000))],
-    ..Default::default()
-  });
-
-  let src = r#"
-  const { op_wait } = Deno.core.ensureFastOps();
-  const promises = [];
-  for (let i = 0; i < 5000; i++) {
-    promises.push(op_wait());
-  }
-  Promise.all(promises);
-  "#;
-  let promise = runtime.execute_script_static("test", src).unwrap();
-  #[allow(deprecated)]
-  runtime.resolve_value(promise).await.unwrap();
-}
-
 #[tokio::test]
 async fn test_ref_unref_ops() {
   let (mut runtime, _dispatch_count) = setup(Mode::AsyncDeferred);
diff --git a/testing/checkin/runner/mod.rs b/testing/checkin/runner/mod.rs
index 0fb83c8..673c8fc 100644
--- a/testing/checkin/runner/mod.rs
+++ b/testing/checkin/runner/mod.rs
@@ -19,6 +19,7 @@ use crate::checkin::runner::testing::TestData;
 use self::testing::TestFunctions;
 
 mod ops;
+mod ops_async;
 mod ops_buffer;
 mod testing;
 mod ts_module_loader;
@@ -29,6 +30,9 @@ deno_core::extension!(
     ops::op_log_debug,
     ops::op_log_info,
     ops::op_test_register,
+    ops_async::op_async_yield,
+    ops_async::op_async_barrier_create,
+    ops_async::op_async_barrier_await,
     ops_buffer::op_v8slice_store,
     ops_buffer::op_v8slice_clone,
   ],
@@ -36,6 +40,7 @@ deno_core::extension!(
   esm = [
     dir "checkin/runtime",
     "__init.js",
+    "async.ts" with_specifier "checkin:async",
     "console.ts" with_specifier "checkin:console",
     "testing.ts" with_specifier "checkin:testing",
     "timers.ts" with_specifier "checkin:timers",
diff --git a/testing/checkin/runner/ops_async.rs b/testing/checkin/runner/ops_async.rs
new file mode 100644
index 0000000..0e45834
--- /dev/null
+++ b/testing/checkin/runner/ops_async.rs
@@ -0,0 +1,33 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+use deno_core::op2;
+use std::future::Future;
+use std::rc::Rc;
+
+use super::testing::TestData;
+
+#[op2(async)]
+pub async fn op_async_yield() {
+  tokio::task::yield_now().await
+}
+
+#[op2(fast)]
+pub fn op_async_barrier_create(
+  #[state] test_data: &mut TestData,
+  #[string] name: String,
+  count: u32,
+) {
+  let barrier = Rc::new(tokio::sync::Barrier::new(count as _));
+  test_data.insert(name, barrier);
+}
+
+#[op2(async)]
+pub fn op_async_barrier_await(
+  #[state] test_data: &TestData,
+  #[string] name: String,
+) -> impl Future<Output = ()> {
+  let barrier: &Rc<tokio::sync::Barrier> = test_data.get(name);
+  let barrier = barrier.clone();
+  async move {
+    barrier.wait().await;
+  }
+}
diff --git a/testing/checkin/runtime/__init.js b/testing/checkin/runtime/__init.js
index b6b6685..20ae054 100644
--- a/testing/checkin/runtime/__init.js
+++ b/testing/checkin/runtime/__init.js
@@ -1,9 +1,10 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import * as async from "checkin:async";
 import * as testing from "checkin:testing";
 import * as console from "checkin:console";
 import * as timers from "checkin:timers";
-
 testing;
+async;
 
 globalThis.console = console.console;
 globalThis.setTimeout = timers.setTimeout;
diff --git a/testing/checkin/runtime/async.ts b/testing/checkin/runtime/async.ts
new file mode 100644
index 0000000..e3d43c3
--- /dev/null
+++ b/testing/checkin/runtime/async.ts
@@ -0,0 +1,16 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+const { op_async_barrier_create, op_async_barrier_await, op_async_yield } = Deno
+  .core
+  .ensureFastOps();
+
+export function barrierCreate(name: string, count: number) {
+  op_async_barrier_create(name, count);
+}
+
+export async function barrierAwait(name: string) {
+  await op_async_barrier_await(name);
+}
+
+export async function asyncYield() {
+  await op_async_yield();
+}
diff --git a/testing/checkin/runtime/console.ts b/testing/checkin/runtime/console.ts
index e93f346..0d60c49 100644
--- a/testing/checkin/runtime/console.ts
+++ b/testing/checkin/runtime/console.ts
@@ -1,5 +1,4 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-
 const {
   op_log_debug,
   op_log_info,
diff --git a/testing/integration/dyn_import_op/dyn_import_op.js b/testing/integration/dyn_import_op/dyn_import_op.js
new file mode 100644
index 0000000..69009e9
--- /dev/null
+++ b/testing/integration/dyn_import_op/dyn_import_op.js
@@ -0,0 +1,5 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import "./main.js";
+import { barrierAwait } from "checkin:async";
+await barrierAwait("barrier");
+console.log("done");
diff --git a/testing/integration/dyn_import_op/dyn_import_op.out b/testing/integration/dyn_import_op/dyn_import_op.out
new file mode 100644
index 0000000..19f86f4
--- /dev/null
+++ b/testing/integration/dyn_import_op/dyn_import_op.out
@@ -0,0 +1,1 @@
+done
diff --git a/testing/integration/dyn_import_op/dynamic.js b/testing/integration/dyn_import_op/dynamic.js
new file mode 100644
index 0000000..e91041e
--- /dev/null
+++ b/testing/integration/dyn_import_op/dynamic.js
@@ -0,0 +1,4 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { asyncYield } from "checkin:async";
+
+await asyncYield();
diff --git a/testing/integration/dyn_import_op/main.js b/testing/integration/dyn_import_op/main.js
new file mode 100644
index 0000000..3cee03d
--- /dev/null
+++ b/testing/integration/dyn_import_op/main.js
@@ -0,0 +1,8 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { barrierAwait, barrierCreate } from "checkin:async";
+
+barrierCreate("barrier", 2);
+(async () => {
+  await import("./dynamic.js");
+  await barrierAwait("barrier");
+})();
diff --git a/testing/lib.rs b/testing/lib.rs
index 93e0279..ff2bd5e 100644
--- a/testing/lib.rs
+++ b/testing/lib.rs
@@ -35,6 +35,7 @@ macro_rules! integration_test {
 unit_test!(
   encode_decode_test,
   microtask_test,
+  ops_async_test,
   ops_buffer_test,
   serialize_deserialize_test,
   tc39_test,
@@ -46,6 +47,7 @@ unit_test!(
 // These files are loaded from the system/ dir.
 integration_test!(
   dyn_import_circular,
+  dyn_import_op,
   error_async_stack,
   error_rejection_catch,
   error_rejection_order,
diff --git a/testing/tsconfig.json b/testing/tsconfig.json
index d05433c..6b80ae4 100644
--- a/testing/tsconfig.json
+++ b/testing/tsconfig.json
@@ -3,6 +3,7 @@
     "target": "ES2022",
     "baseUrl": ".",
     "paths": {
+      "checkin:async": ["checkin/runtime/async.ts"],
       "checkin:console": ["checkin/runtime/console.ts"],
       "checkin:testing": ["checkin/runtime/testing.ts"]
     },
diff --git a/testing/unit/ops_async_test.ts b/testing/unit/ops_async_test.ts
new file mode 100644
index 0000000..0b5acc8
--- /dev/null
+++ b/testing/unit/ops_async_test.ts
@@ -0,0 +1,19 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { test } from "checkin:testing";
+import { asyncYield, barrierAwait, barrierCreate } from "checkin:async";
+
+test(async function testAsyncOp() {
+  await asyncYield();
+});
+
+// Test a large number of async ops resolving at the same time. This stress-tests both
+// large-batch op dispatch and the JS-side promise-tracking implementation.
+test(async function testAsyncBarrier() {
+  const count = 1e6;
+  barrierCreate("barrier", count);
+  const promises = [];
+  for (let i = 0; i < count; i++) {
+    promises.push(barrierAwait("barrier"));
+  }
+  await Promise.all(promises);
+});