refactor(metrics): move to core (#12386)

Avoids overhead of wrapping ops (and allocs when inspecting async-op futures)
This commit is contained in:
Aaron O'Mullan 2021-10-10 17:20:30 +02:00 committed by GitHub
parent f2ac7ff23a
commit 5a8a989b78
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 162 additions and 270 deletions

View File

@ -1792,7 +1792,7 @@ declare namespace Deno {
* Requires `allow-write` permission. */
export function truncate(name: string, len?: number): Promise<void>;
export interface Metrics {
export interface OpMetrics {
opsDispatched: number;
opsDispatchedSync: number;
opsDispatchedAsync: number;
@ -1806,6 +1806,10 @@ declare namespace Deno {
bytesReceived: number;
}
export interface Metrics extends OpMetrics {
ops: Record<string, OpMetrics>;
}
/** Receive metrics from the privileged side of Deno. This is primarily used
* in the development of Deno. 'Ops', also called 'bindings', are the go-between
* between Deno JavaScript and Deno Rust.

View File

@ -804,24 +804,6 @@ declare namespace Deno {
*/
export function sleepSync(millis: number): void;
export interface Metrics extends OpMetrics {
ops: Record<string, OpMetrics>;
}
export interface OpMetrics {
opsDispatched: number;
opsDispatchedSync: number;
opsDispatchedAsync: number;
opsDispatchedAsyncUnref: number;
opsCompleted: number;
opsCompletedSync: number;
opsCompletedAsync: number;
opsCompletedAsyncUnref: number;
bytesSentControl: number;
bytesSentData: number;
bytesReceived: number;
}
/** **UNSTABLE**: New option, yet to be vetted. */
export interface TestDefinition {
/** Specifies the permissions that should be used to run the test.

View File

@ -12,8 +12,10 @@
Map,
Array,
ArrayPrototypeFill,
ArrayPrototypeMap,
ErrorCaptureStackTrace,
Promise,
ObjectEntries,
ObjectFreeze,
ObjectFromEntries,
MapPrototypeGet,
@ -152,6 +154,15 @@
opSync("op_print", str, isErr);
}
function metrics() {
const [aggregate, perOps] = opSync("op_metrics");
aggregate.ops = ObjectFromEntries(ArrayPrototypeMap(
ObjectEntries(opsCache),
([opName, opId]) => [opName, perOps[opId]],
));
return aggregate;
}
// Some "extensions" rely on "BadResource" and "Interrupted" errors in the
// JS code (eg. "deno_net") so they are provided in "Deno.core" but later
// reexported on "Deno.errors"
@ -178,6 +189,7 @@
tryClose,
print,
resources,
metrics,
registerErrorBuilder,
registerErrorClass,
opresolve,

View File

@ -315,7 +315,7 @@ fn opcall_sync<'s>(
mut rv: v8::ReturnValue,
) {
let state_rc = JsRuntime::state(scope);
let state = state_rc.borrow();
let state = state_rc.borrow_mut();
let op_id = match v8::Local::<v8::Integer>::try_from(args.get(0))
.map(|l| l.value() as OpId)
@ -344,11 +344,13 @@ fn opcall_sync<'s>(
scope,
a,
b,
op_id,
promise_id: 0,
};
let op = OpTable::route_op(op_id, state.op_state.clone(), payload);
match op {
Op::Sync(result) => {
state.op_state.borrow_mut().tracker.track_sync(op_id);
rv.set(result.to_v8(scope).unwrap());
}
Op::NotFound => {
@ -405,6 +407,7 @@ fn opcall_async<'s>(
scope,
a,
b,
op_id,
promise_id,
};
let op = OpTable::route_op(op_id, state.op_state.clone(), payload);
@ -417,10 +420,12 @@ fn opcall_async<'s>(
OpResult::Err(_) => rv.set(result.to_v8(scope).unwrap()),
},
Op::Async(fut) => {
state.op_state.borrow_mut().tracker.track_async(op_id);
state.pending_ops.push(fut);
state.have_unpolled_ops = true;
}
Op::AsyncUnref(fut) => {
state.op_state.borrow_mut().tracker.track_unref(op_id);
state.pending_unref_ops.push(fut);
state.have_unpolled_ops = true;
}

View File

@ -13,6 +13,7 @@ mod normalize_path;
mod ops;
mod ops_builtin;
mod ops_json;
mod ops_metrics;
mod resources;
mod runtime;

View File

@ -1008,7 +1008,7 @@ mod tests {
dispatch_count_.fetch_add(1, Ordering::Relaxed);
let (control, _): (u8, ()) = payload.deserialize().unwrap();
assert_eq!(control, 42);
let resp = (0, serialize_op_result(Ok(43), state));
let resp = (0, 1, serialize_op_result(Ok(43), state));
Op::Async(Box::pin(futures::future::ready(resp)))
};

View File

@ -3,6 +3,7 @@
use crate::error::type_error;
use crate::error::AnyError;
use crate::gotham_state::GothamState;
use crate::ops_metrics::OpsTracker;
use crate::resources::ResourceTable;
use crate::runtime::GetErrorClassFn;
use futures::Future;
@ -18,7 +19,8 @@ use std::pin::Pin;
use std::rc::Rc;
pub type PromiseId = u64;
pub type OpAsyncFuture = Pin<Box<dyn Future<Output = (PromiseId, OpResult)>>>;
pub type OpAsyncFuture =
Pin<Box<dyn Future<Output = (PromiseId, OpId, OpResult)>>>;
pub type OpFn = dyn Fn(Rc<RefCell<OpState>>, OpPayload) -> Op + 'static;
pub type OpId = usize;
@ -26,6 +28,7 @@ pub struct OpPayload<'a, 'b, 'c> {
pub(crate) scope: &'a mut v8::HandleScope<'b>,
pub(crate) a: v8::Local<'c, v8::Value>,
pub(crate) b: v8::Local<'c, v8::Value>,
pub(crate) op_id: OpId,
pub(crate) promise_id: PromiseId,
}
@ -96,6 +99,7 @@ pub struct OpState {
pub resource_table: ResourceTable,
pub op_table: OpTable,
pub get_error_class_fn: GetErrorClassFn,
pub(crate) tracker: OpsTracker,
gotham_state: GothamState,
}
@ -105,6 +109,9 @@ impl OpState {
resource_table: Default::default(),
op_table: OpTable::default(),
get_error_class_fn: &|_| "Error",
tracker: OpsTracker {
ops: Vec::with_capacity(256),
},
gotham_state: Default::default(),
}
}

View File

@ -2,6 +2,7 @@ use crate::error::type_error;
use crate::error::AnyError;
use crate::include_js_files;
use crate::op_sync;
use crate::ops_metrics::OpMetrics;
use crate::resources::ResourceId;
use crate::void_op_async;
use crate::void_op_sync;
@ -32,6 +33,7 @@ pub(crate) fn init_builtins() -> Extension {
"op_wasm_streaming_set_url",
op_sync(op_wasm_streaming_set_url),
),
("op_metrics", op_sync(op_metrics)),
("op_void_sync", void_op_sync()),
("op_void_async", void_op_async()),
])
@ -158,3 +160,13 @@ pub fn op_wasm_streaming_set_url(
Ok(())
}
pub fn op_metrics(
state: &mut OpState,
_: (),
_: (),
) -> Result<(OpMetrics, Vec<OpMetrics>), AnyError> {
let aggregate = state.tracker.aggregate();
let per_op = state.tracker.per_op();
Ok((aggregate, per_op))
}

View File

@ -32,9 +32,10 @@ pub fn void_op_async() -> Box<OpFn> {
// to deserialize to the unit type instead of failing with `ExpectedNull`
// op_async(|_, _: (), _: ()| futures::future::ok(()))
Box::new(move |state, payload| -> Op {
let op_id = payload.op_id;
let pid = payload.promise_id;
let op_result = serialize_op_result(Ok(()), state);
Op::Async(Box::pin(futures::future::ready((pid, op_result))))
Op::Async(Box::pin(futures::future::ready((pid, op_id, op_result))))
})
}
@ -112,6 +113,7 @@ where
RV: Serialize + 'static,
{
Box::new(move |state, payload| -> Op {
let op_id = payload.op_id;
let pid = payload.promise_id;
// Deserialize args, sync error on failure
let args = match payload.deserialize() {
@ -124,7 +126,7 @@ where
use crate::futures::FutureExt;
let fut = op_fn(state.clone(), a, b)
.map(move |result| (pid, serialize_op_result(result, state)));
.map(move |result| (pid, op_id, serialize_op_result(result, state)));
Op::Async(Box::pin(fut))
})
}
@ -143,6 +145,7 @@ where
RV: Serialize + 'static,
{
Box::new(move |state, payload| -> Op {
let op_id = payload.op_id;
let pid = payload.promise_id;
// Deserialize args, sync error on failure
let args = match payload.deserialize() {
@ -155,7 +158,7 @@ where
use crate::futures::FutureExt;
let fut = op_fn(state.clone(), a, b)
.map(move |result| (pid, serialize_op_result(result, state)));
.map(move |result| (pid, op_id, serialize_op_result(result, state)));
Op::AsyncUnref(Box::pin(fut))
})
}

96
core/ops_metrics.rs Normal file
View File

@ -0,0 +1,96 @@
// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
use crate::serde::Serialize;
use crate::OpId;
// TODO(@AaronO): split into AggregateMetrics & PerOpMetrics
#[derive(Clone, Default, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct OpMetrics {
pub ops_dispatched: u64,
pub ops_dispatched_sync: u64,
pub ops_dispatched_async: u64,
pub ops_dispatched_async_unref: u64,
pub ops_completed: u64,
pub ops_completed_sync: u64,
pub ops_completed_async: u64,
pub ops_completed_async_unref: u64,
pub bytes_sent_control: u64,
pub bytes_sent_data: u64,
pub bytes_received: u64,
}
// TODO(@AaronO): track errors
#[derive(Default, Debug)]
pub struct OpsTracker {
pub ops: Vec<OpMetrics>,
}
impl OpsTracker {
pub fn per_op(&self) -> Vec<OpMetrics> {
self.ops.clone()
}
pub fn aggregate(&self) -> OpMetrics {
let mut sum = OpMetrics::default();
for metrics in self.ops.iter() {
sum.ops_dispatched += metrics.ops_dispatched;
sum.ops_dispatched_sync += metrics.ops_dispatched_sync;
sum.ops_dispatched_async += metrics.ops_dispatched_async;
sum.ops_dispatched_async_unref += metrics.ops_dispatched_async_unref;
sum.ops_completed += metrics.ops_completed;
sum.ops_completed_sync += metrics.ops_completed_sync;
sum.ops_completed_async += metrics.ops_completed_async;
sum.ops_completed_async_unref += metrics.ops_completed_async_unref;
sum.bytes_sent_control += metrics.bytes_sent_control;
sum.bytes_sent_data += metrics.bytes_sent_data;
sum.bytes_received += metrics.bytes_received;
}
sum
}
fn ensure_capacity(&mut self, op_id: OpId) {
if op_id >= self.ops.len() {
let delta_len = 1 + op_id - self.ops.len();
self.ops.extend(vec![OpMetrics::default(); delta_len])
}
}
fn metrics_mut(&mut self, id: OpId) -> &mut OpMetrics {
self.ensure_capacity(id);
self.ops.get_mut(id).unwrap()
}
pub fn track_sync(&mut self, id: OpId) {
let metrics = self.metrics_mut(id);
metrics.ops_dispatched += 1;
metrics.ops_completed += 1;
metrics.ops_dispatched_sync += 1;
metrics.ops_completed_sync += 1;
}
pub fn track_async(&mut self, id: OpId) {
let metrics = self.metrics_mut(id);
metrics.ops_dispatched += 1;
metrics.ops_dispatched_async += 1;
}
pub fn track_async_completed(&mut self, id: OpId) {
let metrics = self.metrics_mut(id);
metrics.ops_completed += 1;
metrics.ops_completed_async += 1;
}
pub fn track_unref(&mut self, id: OpId) {
let metrics = self.metrics_mut(id);
metrics.ops_dispatched += 1;
metrics.ops_dispatched_async_unref += 1;
}
pub fn track_unref_completed(&mut self, id: OpId) {
let metrics = self.metrics_mut(id);
metrics.ops_completed += 1;
metrics.ops_completed_async_unref += 1;
}
}

View File

@ -44,7 +44,8 @@ use std::sync::Once;
use std::task::Context;
use std::task::Poll;
type PendingOpFuture = Pin<Box<dyn Future<Output = (PromiseId, OpResult)>>>;
type PendingOpFuture =
Pin<Box<dyn Future<Output = (PromiseId, OpId, OpResult)>>>;
pub enum Snapshot {
Static(&'static [u8]),
@ -1477,7 +1478,9 @@ impl JsRuntime {
match pending_r {
Poll::Ready(None) => break,
Poll::Pending => break,
Poll::Ready(Some((promise_id, resp))) => {
Poll::Ready(Some((promise_id, op_id, resp))) => {
let tracker = &mut state.op_state.borrow_mut().tracker;
tracker.track_async_completed(op_id);
async_responses.push((promise_id, resp));
}
};
@ -1488,7 +1491,9 @@ impl JsRuntime {
match unref_r {
Poll::Ready(None) => break,
Poll::Pending => break,
Poll::Ready(Some((promise_id, resp))) => {
Poll::Ready(Some((promise_id, op_id, resp))) => {
let tracker = &mut state.op_state.borrow_mut().tracker;
tracker.track_unref_completed(op_id);
async_responses.push((promise_id, resp));
}
};
@ -1639,7 +1644,7 @@ pub mod tests {
match test_state.mode {
Mode::Async => {
assert_eq!(control, 42);
let resp = (0, serialize_op_result(Ok(43), rc_op_state));
let resp = (0, 1, serialize_op_result(Ok(43), rc_op_state));
Op::Async(Box::pin(futures::future::ready(resp)))
}
Mode::AsyncZeroCopy(has_buffer) => {
@ -1649,7 +1654,7 @@ pub mod tests {
}
let resp = serialize_op_result(Ok(43), rc_op_state);
Op::Async(Box::pin(futures::future::ready((0, resp))))
Op::Async(Box::pin(futures::future::ready((0, 1, resp))))
}
}
}

View File

@ -1,18 +0,0 @@
// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
"use strict";
((window) => {
const core = window.Deno.core;
function metrics() {
const { combined, ops } = core.opSync("op_metrics");
if (ops) {
combined.ops = ops;
}
return combined;
}
window.__bootstrap.metrics = {
metrics,
};
})(this);

View File

@ -6,7 +6,7 @@
const { parsePermissions } = window.__bootstrap.worker;
const { setExitHandler } = window.__bootstrap.os;
const { Console, inspectArgs } = window.__bootstrap.console;
const { metrics } = window.__bootstrap.metrics;
const { metrics } = core;
const { assert } = window.__bootstrap.util;
const {
ArrayPrototypeFilter,

View File

@ -2,10 +2,11 @@
"use strict";
((window) => {
const core = window.Deno.core;
const __bootstrap = window.__bootstrap;
__bootstrap.denoNs = {
metrics: core.metrics,
test: __bootstrap.testing.test,
metrics: __bootstrap.metrics.metrics,
Process: __bootstrap.process.Process,
run: __bootstrap.process.run,
isatty: __bootstrap.tty.isatty,

View File

@ -21,7 +21,6 @@ pub mod errors;
pub mod fs_util;
pub mod inspector_server;
pub mod js;
pub mod metrics;
pub mod ops;
pub mod permissions;
pub mod tokio_util;

View File

@ -1,210 +0,0 @@
// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
use crate::ops::UnstableChecker;
use deno_core::error::AnyError;
use deno_core::op_sync;
use deno_core::serde::Serialize;
use deno_core::serde_json::json;
use deno_core::serde_json::Value;
use deno_core::Extension;
use deno_core::OpState;
pub fn init() -> Extension {
Extension::builder()
.ops(vec![("op_metrics", op_sync(op_metrics))])
.state(|state| {
state.put(RuntimeMetrics::default());
Ok(())
})
.middleware(metrics_op)
.build()
}
#[derive(serde::Serialize)]
struct MetricsReturn {
combined: OpMetrics,
ops: Value,
}
fn op_metrics(
state: &mut OpState,
_: (),
_: (),
) -> Result<MetricsReturn, AnyError> {
let m = state.borrow::<RuntimeMetrics>();
let combined = m.combined_metrics();
let unstable_checker = state.borrow::<UnstableChecker>();
let maybe_ops = if unstable_checker.unstable {
Some(&m.ops)
} else {
None
};
Ok(MetricsReturn {
combined,
ops: json!(maybe_ops),
})
}
#[derive(Default, Debug)]
pub struct RuntimeMetrics {
pub ops: HashMap<&'static str, OpMetrics>,
}
impl RuntimeMetrics {
pub fn combined_metrics(&self) -> OpMetrics {
let mut total = OpMetrics::default();
for metrics in self.ops.values() {
total.ops_dispatched += metrics.ops_dispatched;
total.ops_dispatched_sync += metrics.ops_dispatched_sync;
total.ops_dispatched_async += metrics.ops_dispatched_async;
total.ops_dispatched_async_unref += metrics.ops_dispatched_async_unref;
total.ops_completed += metrics.ops_completed;
total.ops_completed_sync += metrics.ops_completed_sync;
total.ops_completed_async += metrics.ops_completed_async;
total.ops_completed_async_unref += metrics.ops_completed_async_unref;
total.bytes_sent_control += metrics.bytes_sent_control;
total.bytes_sent_data += metrics.bytes_sent_data;
total.bytes_received += metrics.bytes_received;
}
total
}
}
#[derive(Default, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct OpMetrics {
pub ops_dispatched: u64,
pub ops_dispatched_sync: u64,
pub ops_dispatched_async: u64,
pub ops_dispatched_async_unref: u64,
pub ops_completed: u64,
pub ops_completed_sync: u64,
pub ops_completed_async: u64,
pub ops_completed_async_unref: u64,
pub bytes_sent_control: u64,
pub bytes_sent_data: u64,
pub bytes_received: u64,
}
impl OpMetrics {
fn op_dispatched(
&mut self,
bytes_sent_control: usize,
bytes_sent_data: usize,
) {
self.ops_dispatched += 1;
self.bytes_sent_control += bytes_sent_control as u64;
self.bytes_sent_data += bytes_sent_data as u64;
}
fn op_completed(&mut self, bytes_received: usize) {
self.ops_completed += 1;
self.bytes_received += bytes_received as u64;
}
pub fn op_sync(
&mut self,
bytes_sent_control: usize,
bytes_sent_data: usize,
bytes_received: usize,
) {
self.ops_dispatched_sync += 1;
self.op_dispatched(bytes_sent_control, bytes_sent_data);
self.ops_completed_sync += 1;
self.op_completed(bytes_received);
}
pub fn op_dispatched_async(
&mut self,
bytes_sent_control: usize,
bytes_sent_data: usize,
) {
self.ops_dispatched_async += 1;
self.op_dispatched(bytes_sent_control, bytes_sent_data)
}
pub fn op_dispatched_async_unref(
&mut self,
bytes_sent_control: usize,
bytes_sent_data: usize,
) {
self.ops_dispatched_async_unref += 1;
self.op_dispatched(bytes_sent_control, bytes_sent_data)
}
pub fn op_completed_async(&mut self, bytes_received: usize) {
self.ops_completed_async += 1;
self.op_completed(bytes_received);
}
pub fn op_completed_async_unref(&mut self, bytes_received: usize) {
self.ops_completed_async_unref += 1;
self.op_completed(bytes_received);
}
}
use deno_core::Op;
use deno_core::OpFn;
use std::collections::HashMap;
pub fn metrics_op(name: &'static str, op_fn: Box<OpFn>) -> Box<OpFn> {
Box::new(move |op_state, payload| -> Op {
// TODOs:
// * The 'bytes' metrics seem pretty useless, especially now that the
// distinction between 'control' and 'data' buffers has become blurry.
// * Tracking completion of async ops currently makes us put the boxed
// future into _another_ box. Keeping some counters may not be expensive
// in itself, but adding a heap allocation for every metric seems bad.
// TODO: remove this, doesn't make a ton of sense
let bytes_sent_control = 0;
let bytes_sent_data = 0;
let op = (op_fn)(op_state.clone(), payload);
let op_state_ = op_state.clone();
let mut s = op_state.borrow_mut();
let runtime_metrics = s.borrow_mut::<RuntimeMetrics>();
let metrics = if let Some(metrics) = runtime_metrics.ops.get_mut(name) {
metrics
} else {
runtime_metrics.ops.insert(name, OpMetrics::default());
runtime_metrics.ops.get_mut(name).unwrap()
};
use deno_core::futures::future::FutureExt;
match op {
Op::Sync(result) => {
metrics.op_sync(bytes_sent_control, bytes_sent_data, 0);
Op::Sync(result)
}
Op::Async(fut) => {
metrics.op_dispatched_async(bytes_sent_control, bytes_sent_data);
let fut = fut
.inspect(move |_resp| {
let mut s = op_state_.borrow_mut();
let runtime_metrics = s.borrow_mut::<RuntimeMetrics>();
let metrics = runtime_metrics.ops.get_mut(name).unwrap();
metrics.op_completed_async(0);
})
.boxed_local();
Op::Async(fut)
}
Op::AsyncUnref(fut) => {
metrics.op_dispatched_async_unref(bytes_sent_control, bytes_sent_data);
let fut = fut
.inspect(move |_resp| {
let mut s = op_state_.borrow_mut();
let runtime_metrics = s.borrow_mut::<RuntimeMetrics>();
let metrics = runtime_metrics.ops.get_mut(name).unwrap();
metrics.op_completed_async_unref(0);
})
.boxed_local();
Op::AsyncUnref(fut)
}
other => other,
}
})
}

View File

@ -14,7 +14,6 @@ mod utils;
pub mod web_worker;
pub mod worker_host;
use crate::metrics::metrics_op;
use deno_core::error::AnyError;
use deno_core::op_async;
use deno_core::op_sync;
@ -37,7 +36,7 @@ pub fn reg_async<F, A, B, R, RV>(
R: Future<Output = Result<RV, AnyError>> + 'static,
RV: Serialize + 'static,
{
rt.register_op(name, metrics_op(name, op_async(op_fn)));
rt.register_op(name, op_async(op_fn));
}
pub fn reg_sync<F, A, B, R>(rt: &mut JsRuntime, name: &'static str, op_fn: F)
@ -47,7 +46,7 @@ where
B: DeserializeOwned,
R: Serialize + 'static,
{
rt.register_op(name, metrics_op(name, op_sync(op_fn)));
rt.register_op(name, op_sync(op_fn));
}
/// `UnstableChecker` is a struct so it can be placed inside `GothamState`;

View File

@ -2,7 +2,6 @@
use crate::colors;
use crate::inspector_server::InspectorServer;
use crate::js;
use crate::metrics;
use crate::ops;
use crate::permissions::Permissions;
use crate::tokio_util::create_basic_runtime;
@ -337,8 +336,6 @@ impl WebWorker {
deno_timers::init::<Permissions>(),
// ffi
deno_ffi::init::<Permissions>(unstable),
// Metrics
metrics::init(),
// Permissions ext (worker specific state)
perm_ext,
];

View File

@ -2,7 +2,6 @@
use crate::inspector_server::InspectorServer;
use crate::js;
use crate::metrics;
use crate::ops;
use crate::permissions::Permissions;
use crate::BootstrapOptions;
@ -122,8 +121,6 @@ impl MainWorker {
deno_timers::init::<Permissions>(),
// ffi
deno_ffi::init::<Permissions>(unstable),
// Metrics
metrics::init(),
// Runtime ops
ops::runtime::init(main_module.clone()),
ops::worker_host::init(options.create_web_worker_cb.clone()),