Repository: denoland/deno_core
Pull request #320 (2023-11-09 16:08:28)

feat(core): export more metrics types + add merge
Required for https://github.com/denoland/deno/pull/21131


Patch:
diff --git a/core/benches/ops/sync.rs b/core/benches/ops/sync.rs
index 97b9e4b..f063f2b 100644
--- a/core/benches/ops/sync.rs
+++ b/core/benches/ops/sync.rs
@@ -162,7 +162,7 @@ fn bench_op(
     // Add metrics just for the metrics ops
     op_metrics_factory_fn: Some(Box::new(|_, _, op| {
       if op.name.ends_with("_metrics") {
-        Some(std::rc::Rc::new(|_, _| {}))
+        Some(std::rc::Rc::new(|_, _, _| {}))
       } else {
         None
       }
diff --git a/core/lib.rs b/core/lib.rs
index 41bdd0a..dadefd2 100644
--- a/core/lib.rs
+++ b/core/lib.rs
@@ -98,7 +98,6 @@ pub use crate::modules::NoopModuleLoader;
 pub use crate::modules::ResolutionKind;
 pub use crate::modules::StaticModuleLoader;
 pub use crate::normalize_path::normalize_path;
-pub use crate::ops::OpCall;
 pub use crate::ops::OpError;
 pub use crate::ops::OpId;
 pub use crate::ops::OpResult;
@@ -109,6 +108,11 @@ pub use crate::ops_builtin::op_print;
 pub use crate::ops_builtin::op_resources;
 pub use crate::ops_builtin::op_void_async;
 pub use crate::ops_builtin::op_void_sync;
+pub use crate::ops_metrics::merge_op_metrics;
+pub use crate::ops_metrics::OpMetricsEvent;
+pub use crate::ops_metrics::OpMetricsFactoryFn;
+pub use crate::ops_metrics::OpMetricsFn;
+pub use crate::ops_metrics::OpMetricsSource;
 pub use crate::ops_metrics::OpMetricsSummary;
 pub use crate::ops_metrics::OpMetricsSummaryTracker;
 pub use crate::path::strip_unc_prefix;
diff --git a/core/ops.rs b/core/ops.rs
index bc1649f..f9ccc5c 100644
--- a/core/ops.rs
+++ b/core/ops.rs
@@ -11,8 +11,6 @@ use crate::FeatureChecker;
 use crate::OpDecl;
 use anyhow::Error;
 use futures::task::AtomicWaker;
-use futures::Future;
-use pin_project::pin_project;
 use serde::Serialize;
 use std::cell::RefCell;
 use std::cell::UnsafeCell;
@@ -30,42 +28,6 @@ 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>> {
-  promise_id: PromiseId,
-  op_id: OpId,
-  /// Future is not necessarily Unpin, so we need to pin_project.
-  #[pin]
-  fut: F,
-}
-
-impl<F: Future<Output = OpResult>> OpCall<F> {
-  /// Wraps a future; the inner future is polled the usual way (lazily).
-  pub fn new(op_ctx: &OpCtx, promise_id: PromiseId, fut: F) -> Self {
-    Self {
-      op_id: op_ctx.id,
-      promise_id,
-      fut,
-    }
-  }
-}
-
-impl<F: Future<Output = OpResult>> Future for OpCall<F> {
-  type Output = PendingOp;
-
-  fn poll(
-    self: std::pin::Pin<&mut Self>,
-    cx: &mut std::task::Context<'_>,
-  ) -> std::task::Poll<Self::Output> {
-    let promise_id = self.promise_id;
-    let op_id = self.op_id;
-    let fut = self.project().fut;
-    fut
-      .poll(cx)
-      .map(move |res| PendingOp(promise_id, op_id, res, false))
-  }
-}
-
 #[allow(clippy::type_complexity)]
 pub enum OpResult {
   Ok(serde_v8::SerializablePkg),
diff --git a/core/ops_metrics.rs b/core/ops_metrics.rs
index 67de7c8..609b6d7 100644
--- a/core/ops_metrics.rs
+++ b/core/ops_metrics.rs
@@ -25,38 +25,88 @@ pub enum OpMetricsEvent {
   ErrorAsync,
 }
 
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+#[repr(u8)]
+pub enum OpMetricsSource {
+  Slow,
+  Fast,
+  Async,
+}
+
 /// A callback to receieve an [`OpMetricsEvent`].
-pub type OpMetricsFn = Rc<dyn Fn(&OpCtx, OpMetricsEvent)>;
+pub type OpMetricsFn = Rc<dyn Fn(&OpCtx, OpMetricsEvent, OpMetricsSource)>;
 
 // TODO(mmastrac): this would be better as a trait
 /// A callback to retrieve an optional [`OpMetricsFn`] for this op.
 pub type OpMetricsFactoryFn =
   Box<dyn Fn(OpId, usize, &OpDecl) -> Option<OpMetricsFn>>;
 
+/// Given two [`OpMetricsFactoryFn`] implementations, merges them so that op metric events are
+/// called on both.
+pub fn merge_op_metrics(
+  fn1: impl Fn(OpId, usize, &OpDecl) -> Option<OpMetricsFn> + 'static,
+  fn2: impl Fn(OpId, usize, &OpDecl) -> Option<OpMetricsFn> + 'static,
+) -> OpMetricsFactoryFn {
+  Box::new(move |op, count, decl| {
+    match (fn1(op, count, decl), fn2(op, count, decl)) {
+      (None, None) => None,
+      (Some(a), None) => Some(a),
+      (None, Some(b)) => Some(b),
+      (Some(a), Some(b)) => Some(Rc::new(move |ctx, event, source| {
+        a(ctx, event, source);
+        b(ctx, event, source);
+      })),
+    }
+  })
+}
+
 #[doc(hidden)]
 pub fn dispatch_metrics_fast(opctx: &OpCtx, metrics: OpMetricsEvent) {
   // SAFETY: this should only be called from ops where we know the function is Some
-  unsafe { (opctx.metrics_fn.as_ref().unwrap_unchecked())(opctx, metrics) }
+  unsafe {
+    (opctx.metrics_fn.as_ref().unwrap_unchecked())(
+      opctx,
+      metrics,
+      OpMetricsSource::Fast,
+    )
+  }
 }
 
 #[doc(hidden)]
 pub fn dispatch_metrics_slow(opctx: &OpCtx, metrics: OpMetricsEvent) {
   // SAFETY: this should only be called from ops where we know the function is Some
-  unsafe { (opctx.metrics_fn.as_ref().unwrap_unchecked())(opctx, metrics) }
+  unsafe {
+    (opctx.metrics_fn.as_ref().unwrap_unchecked())(
+      opctx,
+      metrics,
+      OpMetricsSource::Slow,
+    )
+  }
 }
 
 #[doc(hidden)]
 pub fn dispatch_metrics_async(opctx: &OpCtx, metrics: OpMetricsEvent) {
   // SAFETY: this should only be called from ops where we know the function is Some
-  unsafe { (opctx.metrics_fn.as_ref().unwrap_unchecked())(opctx, metrics) }
+  unsafe {
+    (opctx.metrics_fn.as_ref().unwrap_unchecked())(
+      opctx,
+      metrics,
+      OpMetricsSource::Async,
+    )
+  }
 }
 
 /// Used for both aggregate and per-op metrics.
 #[derive(Clone, Default, Debug, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct OpMetricsSummary {
+  // The number of ops dispatched synchronously
   pub ops_dispatched_sync: u64,
+  // The number of ops dispatched asynchronously
   pub ops_dispatched_async: u64,
+  // The number of sync ops dispatched fast
+  pub ops_dispatched_fast: u64,
+  // The number of asynchronously-dispatch ops completed
   pub ops_completed_async: u64,
 }
 
@@ -82,6 +132,7 @@ impl OpMetricsSummaryTracker {
 
     for metrics in self.ops.borrow().iter() {
       sum.ops_dispatched_sync += metrics.ops_dispatched_sync;
+      sum.ops_dispatched_fast += metrics.ops_dispatched_fast;
       sum.ops_dispatched_async += metrics.ops_dispatched_async;
       sum.ops_completed_async += metrics.ops_completed_async;
     }
@@ -96,12 +147,16 @@ impl OpMetricsSummaryTracker {
 
   /// Returns a [`OpMetricsFn`] for this tracker.
   fn op_metrics_fn(self: Rc<Self>) -> OpMetricsFn {
-    Rc::new(move |ctx, event| match event {
+    Rc::new(move |ctx, event, source| match event {
       OpMetricsEvent::Dispatched => {
+        let mut m = self.metrics_mut(ctx.id);
+        if source == OpMetricsSource::Fast {
+          m.ops_dispatched_fast += 1;
+        }
         if ctx.decl.is_async {
-          self.metrics_mut(ctx.id).ops_dispatched_async += 1;
+          m.ops_dispatched_async += 1;
         } else {
-          self.metrics_mut(ctx.id).ops_dispatched_sync += 1;
+          m.ops_dispatched_sync += 1;
         }
       }
       OpMetricsEvent::Completed
diff --git a/core/runtime/ops.rs b/core/runtime/ops.rs
index ec98a16..96e58f2 100644
--- a/core/runtime/ops.rs
+++ b/core/runtime/ops.rs
@@ -37,7 +37,6 @@ pub fn map_async_op_infallible<R: 'static>(
 ) -> Option<R> {
   let id = ctx.id;
   let metrics = ctx.metrics_enabled();
-
   if lazy {
     let mapper = move |r| {
       PendingOp(


Test patch:
diff --git a/core/runtime/tests/ops.rs b/core/runtime/tests/ops.rs
index e805166..6e4205c 100644
--- a/core/runtime/tests/ops.rs
+++ b/core/runtime/tests/ops.rs
@@ -643,7 +643,7 @@ pub async fn test_op_metrics() {
         return None;
       }
       let out_clone = out_clone.clone();
-      Some(Rc::new(move |_, metrics| {
+      Some(Rc::new(move |_, metrics, _| {
         let s = format!("{} {:?}", name, metrics);
         println!("{s}");
         out_clone.borrow_mut().push(s);
@@ -756,7 +756,8 @@ pub async fn test_op_metrics_summary_tracker() {
     OpMetricsSummary {
       ops_completed_async: 8,
       ops_dispatched_async: 8,
-      ops_dispatched_sync: 3
+      ops_dispatched_sync: 3,
+      ops_dispatched_fast: 0,
     }
   );
 }