feat(ops): zero-cost-if-unused, pluggable metrics for op2 (denoland#244)

Introduces a basic metrics API for op2 that is zero cost when unused, pluggable-per-op and a flexible system that we can add more to in the future.

As we remove op1 from the system, we will likely be able to simplify the async portion of this metrics system. Right now we're juggling the old and new op systems.

Author: mmastrac

.gitattributes

@@ -1,3 +1,4 @@
 # Use Unix line endings in all text files.
 * text=auto eol=lf
 *.png -text
+*.out linguist-generated=true

Cargo.lock

@@ -49,6 +49,7 @@ path = "examples/http_bench_json_ops/main.rs"
 cooked-waker.workspace = true
 deno_ast.workspace = true
 bencher.workspace = true
+pretty_assertions.workspace = true
 
 [[bench]]
 name = "ops_sync"

core/Cargo.toml

@@ -99,8 +99,14 @@ pub struct OpDecl {
   pub is_unstable: bool,
   pub is_v8: bool,
   pub arg_count: u8,
-  pub(crate) v8_fn_ptr: OpFnRef,
+  /// The slow dispatch call. If metrics are disabled, the `v8::Function` is created with this callback.
+  pub(crate) slow_fn: OpFnRef,
+  /// The slow dispatch call with metrics enabled. If metrics are enabled, the `v8::Function` is created with this callback.
+  pub(crate) slow_fn_with_metrics: OpFnRef,
+  /// The fast dispatch call. If metrics are disabled, the `v8::Function`'s fastcall is created with this callback.
   pub(crate) fast_fn: Option<FastFunction>,
+  /// The fast dispatch call with metrics enabled. If metrics are enabled, the `v8::Function`'s fastcall is created with this callback.
+  pub(crate) fast_fn_with_metrics: Option<FastFunction>,
 }
 
 impl OpDecl {
@@ -112,7 +118,7 @@ impl OpDecl {
     is_unstable: bool,
     is_v8: bool,
     arg_count: u8,
-    v8_fn_ptr: OpFnRef,
+    slow_fn: OpFnRef,
     fast_fn: Option<FastFunction>,
   ) -> Self {
     Self {
@@ -122,8 +128,35 @@ impl OpDecl {
       is_unstable,
       is_v8,
       arg_count,
-      v8_fn_ptr,
+      slow_fn,
+      slow_fn_with_metrics: slow_fn,
       fast_fn,
+      fast_fn_with_metrics: fast_fn,
+    }
+  }
+
+  /// For use by internal op implementation only.
+  #[doc(hidden)]
+  pub const fn new_internal_op2(
+    name: &'static str,
+    is_async: bool,
+    arg_count: u8,
+    slow_fn: OpFnRef,
+    slow_fn_with_metrics: OpFnRef,
+    fast_fn: Option<FastFunction>,
+    fast_fn_with_metrics: Option<FastFunction>,
+  ) -> Self {
+    Self {
+      name,
+      enabled: true,
+      is_async,
+      is_unstable: false,
+      is_v8: false,
+      arg_count,
+      slow_fn,
+      slow_fn_with_metrics,
+      fast_fn,
+      fast_fn_with_metrics,
     }
   }
 
@@ -141,8 +174,10 @@ impl OpDecl {
   /// `OpDecl`.
   pub const fn with_implementation_from(self, from: &Self) -> Self {
     Self {
-      v8_fn_ptr: from.v8_fn_ptr,
+      slow_fn: from.slow_fn,
+      slow_fn_with_metrics: from.slow_fn_with_metrics,
       fast_fn: from.fast_fn,
+      fast_fn_with_metrics: from.fast_fn_with_metrics,
       ..self
     }
   }

core/extensions.rs

@@ -140,6 +140,7 @@ pub mod _ops {
   pub use super::extensions::OpDecl;
   pub use super::ops::to_op_result;
   pub use super::ops::OpCtx;
+  pub use super::ops::OpMetricsEvent;
   pub use super::ops::OpResult;
   pub use super::runtime::ops::*;
   pub use super::runtime::ops_rust_to_v8::*;

core/lib.rs

@@ -28,6 +28,7 @@ use v8::Isolate;
 
 pub type PromiseId = i32;
 pub type OpId = u16;
+pub struct PendingOp(pub PromiseId, pub OpId, pub OpResult, pub bool);
 
 #[pin_project]
 pub struct OpCall<F: Future<Output = OpResult>> {
@@ -50,7 +51,7 @@ impl<F: Future<Output = OpResult>> OpCall<F> {
 }
 
 impl<F: Future<Output = OpResult>> Future for OpCall<F> {
-  type Output = (PromiseId, OpId, OpResult);
+  type Output = PendingOp;
 
   fn poll(
     self: std::pin::Pin<&mut Self>,
@@ -59,7 +60,9 @@ impl<F: Future<Output = OpResult>> Future for OpCall<F> {
     let promise_id = self.promise_id;
     let op_id = self.op_id;
     let fut = self.project().fut;
-    fut.poll(cx).map(move |res| (promise_id, op_id, res))
+    fut
+      .poll(cx)
+      .map(move |res| PendingOp(promise_id, op_id, res, false))
   }
 }
 
@@ -120,6 +123,22 @@ pub fn to_op_result<R: Serialize + 'static>(
   }
 }
 
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub enum OpMetricsEvent {
+  /// Entered an op.
+  Dispatched,
+  /// Left an op synchronously.
+  Completed,
+  /// Left an op asynchronously.
+  CompletedAsync,
+  /// Left an op synchronously with an exception.
+  Error,
+  /// Left an op asynchronously with an exception.
+  ErrorAsync,
+}
+
+pub type OpMetricsFn = Rc<dyn Fn(&OpDecl, OpMetricsEvent)>;
+
 /// Per-op context.
 ///
 // Note: We don't worry too much about the size of this struct because it's allocated once per realm, and is
@@ -133,12 +152,14 @@ pub struct OpCtx {
   pub decl: Rc<OpDecl>,
   pub fast_fn_c_info: Option<NonNull<v8::fast_api::CFunctionInfo>>,
   pub runtime_state: Weak<RefCell<JsRuntimeState>>,
+  pub(crate) metrics_fn: Option<OpMetricsFn>,
   pub(crate) context_state: Rc<RefCell<ContextState>>,
   /// If the last fast op failed, stores the error to be picked up by the slow op.
   pub(crate) last_fast_error: UnsafeCell<Option<AnyError>>,
 }
 
 impl OpCtx {
+  #[allow(clippy::too_many_arguments)]
   pub(crate) fn new(
     id: OpId,
     isolate: *mut Isolate,
@@ -147,10 +168,21 @@ impl OpCtx {
     state: Rc<RefCell<OpState>>,
     runtime_state: Weak<RefCell<JsRuntimeState>>,
     get_error_class_fn: GetErrorClassFn,
+    metrics_fn: Option<OpMetricsFn>,
   ) -> Self {
     let mut fast_fn_c_info = None;
 
-    if let Some(fast_fn) = &decl.fast_fn {
+    // If we want metrics for this function, create the fastcall `CFunctionInfo` from the metrics
+    // `FastFunction`. For some extremely fast ops, the parameter list may change for the metrics
+    // version and require a slightly different set of arguments (for example, it may need the fastcall
+    // callback information to get the `OpCtx`).
+    let fast_fn = if metrics_fn.is_some() {
+      &decl.fast_fn
+    } else {
+      &decl.fast_fn_with_metrics
+    };
+
+    if let Some(fast_fn) = fast_fn {
       let args = CTypeInfo::new_from_slice(fast_fn.args);
       let ret = CTypeInfo::new(fast_fn.return_type);
 
@@ -169,7 +201,7 @@ impl OpCtx {
       fast_fn_c_info = Some(c_fn);
     }
 
-    OpCtx {
+    Self {
       id,
       state,
       get_error_class_fn,
@@ -179,9 +211,14 @@ impl OpCtx {
       fast_fn_c_info,
       last_fast_error: UnsafeCell::new(None),
       isolate,
+      metrics_fn,
     }
   }
 
+  pub fn metrics_enabled(&self) -> bool {
+    self.metrics_fn.is_some()
+  }
+
   /// This takes the last error from an [`OpCtx`], assuming that no other code anywhere
   /// can hold a `&mut` to the last_fast_error field.
   ///

core/ops.rs

@@ -43,18 +43,34 @@ pub(crate) fn external_references(
   });
 
   for ctx in ops {
-    let ctx_ptr = ctx as *const OpCtx as _;
-    references.push(v8::ExternalReference { pointer: ctx_ptr });
-    references.push(v8::ExternalReference {
-      function: ctx.decl.v8_fn_ptr,
-    });
-    if let Some(fast_fn) = &ctx.decl.fast_fn {
+    if ctx.metrics_enabled() {
+      let ctx_ptr = ctx as *const OpCtx as _;
+      references.push(v8::ExternalReference { pointer: ctx_ptr });
       references.push(v8::ExternalReference {
-        pointer: fast_fn.function as _,
+        function: ctx.decl.slow_fn_with_metrics,
       });
+      if let Some(fast_fn) = &ctx.decl.fast_fn_with_metrics {
+        references.push(v8::ExternalReference {
+          pointer: fast_fn.function as _,
+        });
+        references.push(v8::ExternalReference {
+          pointer: ctx.fast_fn_c_info.unwrap().as_ptr() as _,
+        });
+      }
+    } else {
+      let ctx_ptr = ctx as *const OpCtx as _;
+      references.push(v8::ExternalReference { pointer: ctx_ptr });
       references.push(v8::ExternalReference {
-        pointer: ctx.fast_fn_c_info.unwrap().as_ptr() as _,
+        function: ctx.decl.slow_fn,
       });
+      if let Some(fast_fn) = &ctx.decl.fast_fn {
+        references.push(v8::ExternalReference {
+          pointer: fast_fn.function as _,
+        });
+        references.push(v8::ExternalReference {
+          pointer: ctx.fast_fn_c_info.unwrap().as_ptr() as _,
+        });
+      }
     }
   }
 
@@ -190,12 +206,22 @@ fn op_ctx_function<'s>(
   let v8name =
     v8::String::new_external_onebyte_static(scope, op_ctx.decl.name.as_bytes())
       .unwrap();
+
+  let (slow_fn, fast_fn) = if op_ctx.metrics_enabled() {
+    (
+      op_ctx.decl.slow_fn_with_metrics,
+      op_ctx.decl.fast_fn_with_metrics,
+    )
+  } else {
+    (op_ctx.decl.slow_fn, op_ctx.decl.fast_fn)
+  };
+
   let builder: v8::FunctionBuilder<v8::FunctionTemplate> =
-    v8::FunctionTemplate::builder_raw(op_ctx.decl.v8_fn_ptr)
+    v8::FunctionTemplate::builder_raw(slow_fn)
       .data(external.into())
       .length(op_ctx.decl.arg_count as i32);
 
-  let template = if let Some(fast_function) = &op_ctx.decl.fast_fn {
+  let template = if let Some(fast_function) = &fast_fn {
     builder.build_fast(
       scope,
       fast_function,

core/runtime/bindings.rs

@@ -10,11 +10,9 @@ use crate::modules::ModuleId;
 use crate::modules::ModuleLoadId;
 use crate::modules::ModuleMap;
 use crate::ops::OpCtx;
+use crate::ops::PendingOp;
 use crate::runtime::JsRuntimeState;
 use crate::JsRuntime;
-use crate::OpId;
-use crate::OpResult;
-use crate::PromiseId;
 use anyhow::Error;
 use deno_unsync::JoinSet;
 use futures::channel::oneshot;
@@ -76,7 +74,7 @@ pub(crate) struct ContextState {
   pending_dyn_mod_evaluate: Vec<DynImportModEvaluate>,
   pub(crate) pending_mod_evaluate: Option<ModEvaluate>,
   pub(crate) unrefed_ops: HashSet<i32, BuildHasherDefault<IdentityHasher>>,
-  pub(crate) pending_ops: JoinSet<(PromiseId, OpId, OpResult)>,
+  pub(crate) pending_ops: JoinSet<PendingOp>,
   // We don't explicitly re-read this prop but need the slice to live alongside
   // the context
   pub(crate) op_ctxs: Box<[OpCtx]>,

core/runtime/jsrealm.rs

@@ -2,6 +2,7 @@
 
 use super::bindings;
 use super::jsrealm::JsRealmInner;
+use super::ops::dispatch_metrics_async;
 use super::snapshot_util;
 use crate::error::exception_to_err_result;
 use crate::error::generic_error;
@@ -361,6 +362,8 @@ fn v8_init(
   v8::V8::initialize();
 }
 
+pub type OpMetricsFactoryFn = Box<dyn Fn(&OpDecl) -> Option<OpMetricsFn>>;
+
 #[derive(Default)]
 pub struct RuntimeOptions {
   /// Source map reference for errors.
@@ -377,6 +380,10 @@ pub struct RuntimeOptions {
   /// executed tries to load modules.
   pub module_loader: Option<Rc<dyn ModuleLoader>>,
 
+  /// Provide a function that may optionally provide a metrics collector
+  /// for a given op.
+  pub op_metrics_fn: Option<OpMetricsFactoryFn>,
+
   /// JsRuntime extensions, not to be confused with ES modules.
   /// Only ops registered by extensions will be initialized. If you need
   /// to execute JS code from extensions, pass source files in `js` or `esm`
@@ -594,6 +601,8 @@ impl JsRuntime {
       .into_iter()
       .enumerate()
       .map(|(id, decl)| {
+        let metrics_fn =
+          options.op_metrics_fn.as_ref().and_then(|f| (f)(&decl));
         OpCtx::new(
           id as u16,
           std::ptr::null_mut(),
@@ -602,6 +611,7 @@ impl JsRuntime {
           op_state.clone(),
           weak.clone(),
           options.get_error_class_fn.unwrap_or(&|_| "Error"),
+          metrics_fn,
         )
       })
       .collect::<Vec<_>>()
@@ -832,6 +842,7 @@ impl JsRuntime {
             op_ctx.state.clone(),
             op_ctx.runtime_state.clone(),
             op_ctx.get_error_class_fn,
+            op_ctx.metrics_fn.clone(),
           )
         })
         .collect();
@@ -1965,7 +1976,7 @@ impl JsRuntime {
           break;
         };
         // TODO(mmastrac): If this task is really errored, things could be pretty bad
-        let (promise_id, op_id, resp) = item.unwrap();
+        let PendingOp(promise_id, op_id, resp, metrics_event) = item.unwrap();
         state
           .borrow()
           .op_state
@@ -1975,7 +1986,22 @@ impl JsRuntime {
         context_state.unrefed_ops.remove(&promise_id);
         dispatched_ops |= true;
         args.push(v8::Integer::new(scope, promise_id).into());
-        args.push(match resp.to_v8(scope) {
+        let was_error = matches!(resp, OpResult::Err(_));
+        let res = resp.to_v8(scope);
+        if metrics_event {
+          if res.is_ok() && !was_error {
+            dispatch_metrics_async(
+              &context_state.op_ctxs[op_id as usize],
+              OpMetricsEvent::CompletedAsync,
+            );
+          } else {
+            dispatch_metrics_async(
+              &context_state.op_ctxs[op_id as usize],
+              OpMetricsEvent::ErrorAsync,
+            );
+          }
+        }
+        args.push(match res {
           Ok(v) => v,
           Err(e) => OpResult::Err(OpError::new(&|_| "TypeError", e.into()))
             .to_v8(scope)