tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-24 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

metrics: use MetricAtomic* for task counters (#6624)

Browse Source
This commit is contained in:
Alice Ryhl
2024-06-17 08:33:08 +00:00
committed by GitHub
parent 3bf4f93854
commit 9a75d6f7f7
8 changed files with 331 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/runtime/metrics/runtime.rs
+283 -283
View File
@@ -102,30 +102,6 @@ impl RuntimeMetrics {
self.handle.inner.active_tasks_count()
}
/// Returns the number of tasks spawned in this runtime since it was created.
///
/// This count starts at zero when the runtime is created and increases by one each time a task is spawned.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.spawned_tasks_count();
/// println!("Runtime has had {} tasks spawned", n);
/// }
/// ```
pub fn spawned_tasks_count(&self) -> u64 {
self.handle.inner.spawned_tasks_count()
}
/// Returns the number of idle threads, which have spawned by the runtime
/// for `spawn_blocking` calls.
///
@@ -151,271 +127,295 @@ impl RuntimeMetrics {
}
cfg_64bit_metrics! {
/// Returns the number of tasks scheduled from **outside** of the runtime.
///
/// The remote schedule count starts at zero when the runtime is created and
/// increases by one each time a task is woken from **outside** of the
/// runtime. This usually means that a task is spawned or notified from a
/// non-runtime thread and must be queued using the Runtime's injection
/// queue, which tends to be slower.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.remote_schedule_count();
/// println!("{} tasks were scheduled from outside the runtime", n);
/// }
/// ```
pub fn remote_schedule_count(&self) -> u64 {
self.handle
.inner
.scheduler_metrics()
.remote_schedule_count
.load(Relaxed)
}
/// Returns the number of tasks spawned in this runtime since it was created.
///
/// This count starts at zero when the runtime is created and increases by one each time a task is spawned.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.spawned_tasks_count();
/// println!("Runtime has had {} tasks spawned", n);
/// }
/// ```
pub fn spawned_tasks_count(&self) -> u64 {
self.handle.inner.spawned_tasks_count()
}
/// Returns the number of times that tasks have been forced to yield back to the scheduler
/// after exhausting their task budgets.
///
/// This count starts at zero when the runtime is created and increases by one each time a task yields due to exhausting its budget.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
pub fn budget_forced_yield_count(&self) -> u64 {
self.handle
.inner
.scheduler_metrics()
.budget_forced_yield_count
.load(Relaxed)
}
/// Returns the number of tasks scheduled from **outside** of the runtime.
///
/// The remote schedule count starts at zero when the runtime is created and
/// increases by one each time a task is woken from **outside** of the
/// runtime. This usually means that a task is spawned or notified from a
/// non-runtime thread and must be queued using the Runtime's injection
/// queue, which tends to be slower.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.remote_schedule_count();
/// println!("{} tasks were scheduled from outside the runtime", n);
/// }
/// ```
pub fn remote_schedule_count(&self) -> u64 {
self.handle
.inner
.scheduler_metrics()
.remote_schedule_count
.load(Relaxed)
}
/// Returns the total number of times the given worker thread has parked.
///
/// The worker park count starts at zero when the runtime is created and
/// increases by one each time the worker parks the thread waiting for new
/// inbound events to process. This usually means the worker has processed
/// all pending work and is currently idle.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_park_count(0);
/// println!("worker 0 parked {} times", n);
/// }
/// ```
pub fn worker_park_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.park_count
.load(Relaxed)
}
/// Returns the number of times that tasks have been forced to yield back to the scheduler
/// after exhausting their task budgets.
///
/// This count starts at zero when the runtime is created and increases by one each time a task yields due to exhausting its budget.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
pub fn budget_forced_yield_count(&self) -> u64 {
self.handle
.inner
.scheduler_metrics()
.budget_forced_yield_count
.load(Relaxed)
}
/// Returns the number of times the given worker thread unparked but
/// performed no work before parking again.
///
/// The worker no-op count starts at zero when the runtime is created and
/// increases by one each time the worker unparks the thread but finds no
/// new work and goes back to sleep. This indicates a false-positive wake up.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_noop_count(0);
/// println!("worker 0 had {} no-op unparks", n);
/// }
/// ```
pub fn worker_noop_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.noop_count
.load(Relaxed)
}
/// Returns the total number of times the given worker thread has parked.
///
/// The worker park count starts at zero when the runtime is created and
/// increases by one each time the worker parks the thread waiting for new
/// inbound events to process. This usually means the worker has processed
/// all pending work and is currently idle.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_park_count(0);
/// println!("worker 0 parked {} times", n);
/// }
/// ```
pub fn worker_park_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.park_count
.load(Relaxed)
}
/// Returns the number of tasks the given worker thread stole from
/// another worker thread.
///
/// This metric only applies to the **multi-threaded** runtime and will
/// always return `0` when using the current thread runtime.
///
/// The worker steal count starts at zero when the runtime is created and
/// increases by `N` each time the worker has processed its scheduled queue
/// and successfully steals `N` more pending tasks from another worker.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_steal_count(0);
/// println!("worker 0 has stolen {} tasks", n);
/// }
/// ```
pub fn worker_steal_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.steal_count
.load(Relaxed)
}
/// Returns the number of times the given worker thread unparked but
/// performed no work before parking again.
///
/// The worker no-op count starts at zero when the runtime is created and
/// increases by one each time the worker unparks the thread but finds no
/// new work and goes back to sleep. This indicates a false-positive wake up.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_noop_count(0);
/// println!("worker 0 had {} no-op unparks", n);
/// }
/// ```
pub fn worker_noop_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.noop_count
.load(Relaxed)
}
/// Returns the number of times the given worker thread stole tasks from
/// another worker thread.
///
/// This metric only applies to the **multi-threaded** runtime and will
/// always return `0` when using the current thread runtime.
///
/// The worker steal count starts at zero when the runtime is created and
/// increases by one each time the worker has processed its scheduled queue
/// and successfully steals more pending tasks from another worker.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_steal_operations(0);
/// println!("worker 0 has stolen tasks {} times", n);
/// }
/// ```
pub fn worker_steal_operations(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.steal_operations
.load(Relaxed)
}
/// Returns the number of tasks the given worker thread stole from
/// another worker thread.
///
/// This metric only applies to the **multi-threaded** runtime and will
/// always return `0` when using the current thread runtime.
///
/// The worker steal count starts at zero when the runtime is created and
/// increases by `N` each time the worker has processed its scheduled queue
/// and successfully steals `N` more pending tasks from another worker.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_steal_count(0);
/// println!("worker 0 has stolen {} tasks", n);
/// }
/// ```
pub fn worker_steal_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.steal_count
.load(Relaxed)
}
/// Returns the number of tasks the given worker thread has polled.
///
/// The worker poll count starts at zero when the runtime is created and
/// increases by one each time the worker polls a scheduled task.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_poll_count(0);
/// println!("worker 0 has polled {} tasks", n);
/// }
/// ```
pub fn worker_poll_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.poll_count
.load(Relaxed)
}
/// Returns the number of times the given worker thread stole tasks from
/// another worker thread.
///
/// This metric only applies to the **multi-threaded** runtime and will
/// always return `0` when using the current thread runtime.
///
/// The worker steal count starts at zero when the runtime is created and
/// increases by one each time the worker has processed its scheduled queue
/// and successfully steals more pending tasks from another worker.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_steal_operations(0);
/// println!("worker 0 has stolen tasks {} times", n);
/// }
/// ```
pub fn worker_steal_operations(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.steal_operations
.load(Relaxed)
}
/// Returns the number of tasks the given worker thread has polled.
///
/// The worker poll count starts at zero when the runtime is created and
/// increases by one each time the worker polls a scheduled task.
///
/// The counter is monotonically increasing. It is never decremented or
/// reset to zero.
///
/// # Arguments
///
/// `worker` is the index of the worker being queried. The given value must
/// be between 0 and `num_workers()`. The index uniquely identifies a single
/// worker and will continue to identify the worker throughout the lifetime
/// of the runtime instance.
///
/// # Panics
///
/// The method panics when `worker` represents an invalid worker, i.e. is
/// greater than or equal to `num_workers()`.
///
/// # Examples
///
/// ```
/// use tokio::runtime::Handle;
///
/// #[tokio::main]
/// async fn main() {
/// let metrics = Handle::current().metrics();
///
/// let n = metrics.worker_poll_count(0);
/// println!("worker 0 has polled {} tasks", n);
/// }
/// ```
pub fn worker_poll_count(&self, worker: usize) -> u64 {
self.handle
.inner
.worker_metrics(worker)
.poll_count
.load(Relaxed)
}
/// Returns the amount of time the given worker thread has been busy.
///
tokio/src/runtime/scheduler/current_thread/mod.rs
+4 -2
View File
@@ -537,8 +537,10 @@ cfg_unstable_metrics! {
self.shared.owned.active_tasks_count()
}
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.shared.owned.spawned_tasks_count()
cfg_64bit_metrics! {
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.shared.owned.spawned_tasks_count()
}
}
}
}
tokio/src/runtime/scheduler/mod.rs
+6 -4
View File
@@ -179,6 +179,12 @@ cfg_rt! {
use crate::runtime::{SchedulerMetrics, WorkerMetrics};
impl Handle {
cfg_64bit_metrics! {
pub(crate) fn spawned_tasks_count(&self) -> u64 {
match_flavor!(self, Handle(handle) => handle.spawned_tasks_count())
}
}
pub(crate) fn num_blocking_threads(&self) -> usize {
match_flavor!(self, Handle(handle) => handle.num_blocking_threads())
}
@@ -191,10 +197,6 @@ cfg_rt! {
match_flavor!(self, Handle(handle) => handle.active_tasks_count())
}
pub(crate) fn spawned_tasks_count(&self) -> u64 {
match_flavor!(self, Handle(handle) => handle.spawned_tasks_count())
}
pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
match_flavor!(self, Handle(handle) => handle.scheduler_metrics())
}
tokio/src/runtime/scheduler/multi_thread/handle/metrics.rs
+6 -4
View File
@@ -10,6 +10,12 @@ impl Handle {
}
cfg_unstable_metrics! {
cfg_64bit_metrics! {
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.shared.owned.spawned_tasks_count()
}
}
pub(crate) fn num_blocking_threads(&self) -> usize {
// workers are currently spawned using spawn_blocking
self.blocking_spawner
@@ -25,10 +31,6 @@ impl Handle {
self.shared.owned.active_tasks_count()
}
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.shared.owned.spawned_tasks_count()
}
pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
&self.shared.scheduler_metrics
}
tokio/src/runtime/scheduler/multi_thread_alt/handle/metrics.rs
+4 -2
View File
@@ -22,8 +22,10 @@ impl Handle {
self.shared.owned.active_tasks_count()
}
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.shared.owned.spawned_tasks_count()
cfg_64bit_metrics! {
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.shared.owned.spawned_tasks_count()
}
}
pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
tokio/src/runtime/task/list.rs
+4 -2
View File
@@ -170,8 +170,10 @@ impl<S: 'static> OwnedTasks<S> {
self.list.len()
}
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.list.added()
cfg_64bit_metrics! {
pub(crate) fn spawned_tasks_count(&self) -> u64 {
self.list.added()
}
}
pub(crate) fn remove(&self, task: &Task<S>) -> Option<Task<S>> {
tokio/src/util/metric_atomics.rs
+8 -2
View File
@@ -1,4 +1,4 @@
use std::sync::atomic::Ordering;
use std::sync::atomic::{AtomicUsize, Ordering};
cfg_64bit_metrics! {
use std::sync::atomic::AtomicU64;
@@ -52,11 +52,17 @@ impl MetricAtomicU64 {
/// This exposes simplified APIs for use in metrics & uses `std::sync` instead of Loom to avoid polluting loom logs with metric information.
#[derive(Debug, Default)]
pub(crate) struct MetricAtomicUsize {
value: std::sync::atomic::AtomicUsize,
value: AtomicUsize,
}
#[cfg_attr(not(all(tokio_unstable, feature = "rt")), allow(dead_code))]
impl MetricAtomicUsize {
pub(crate) fn new(value: usize) -> Self {
Self {
value: AtomicUsize::new(value),
}
}
pub(crate) fn load(&self, ordering: Ordering) -> usize {
self.value.load(ordering)
}
tokio/src/util/sharded_list.rs
+16 -15
View File
@@ -1,9 +1,8 @@
use std::ptr::NonNull;
use std::sync::atomic::Ordering;
use crate::loom::sync::atomic::AtomicU64;
use crate::loom::sync::{Mutex, MutexGuard};
use std::sync::atomic::AtomicUsize;
use crate::util::metric_atomics::{MetricAtomicU64, MetricAtomicUsize};
use super::linked_list::{Link, LinkedList};
@@ -15,8 +14,8 @@ use super::linked_list::{Link, LinkedList};
/// Note: Due to its inner sharded design, the order of nodes cannot be guaranteed.
pub(crate) struct ShardedList<L, T> {
lists: Box<[Mutex<LinkedList<L, T>>]>,
added: AtomicU64,
count: AtomicUsize,
added: MetricAtomicU64,
count: MetricAtomicUsize,
shard_mask: usize,
}
@@ -44,8 +43,8 @@ impl<L, T> ShardedList<L, T> {
}
Self {
lists: lists.into_boxed_slice(),
added: AtomicU64::new(0),
count: AtomicUsize::new(0),
added: MetricAtomicU64::new(0),
count: MetricAtomicUsize::new(0),
shard_mask,
}
}
@@ -54,8 +53,8 @@ impl<L, T> ShardedList<L, T> {
/// Used to get the lock of shard.
pub(crate) struct ShardGuard<'a, L, T> {
lock: MutexGuard<'a, LinkedList<L, T>>,
added: &'a AtomicU64,
count: &'a AtomicUsize,
added: &'a MetricAtomicU64,
count: &'a MetricAtomicUsize,
id: usize,
}
@@ -66,7 +65,7 @@ impl<L: ShardedListItem> ShardedList<L, L::Target> {
let mut lock = self.shard_inner(shard_id);
let node = lock.pop_back();
if node.is_some() {
self.count.fetch_sub(1, Ordering::Relaxed);
self.count.decrement();
}
node
}
@@ -86,7 +85,7 @@ impl<L: ShardedListItem> ShardedList<L, L::Target> {
// to be in any other list of the same sharded list.
let node = unsafe { lock.remove(node) };
if node.is_some() {
self.count.fetch_sub(1, Ordering::Relaxed);
self.count.decrement();
}
node
}
@@ -107,9 +106,11 @@ impl<L: ShardedListItem> ShardedList<L, L::Target> {
self.count.load(Ordering::Relaxed)
}
/// Gets the total number of elements added to this list.
pub(crate) fn added(&self) -> u64 {
self.added.load(Ordering::Relaxed)
cfg_64bit_metrics! {
/// Gets the total number of elements added to this list.
pub(crate) fn added(&self) -> u64 {
self.added.load(Ordering::Relaxed)
}
}
/// Returns whether the linked list does not contain any node.
@@ -137,8 +138,8 @@ impl<'a, L: ShardedListItem> ShardGuard<'a, L, L::Target> {
let id = unsafe { L::get_shard_id(L::as_raw(&val)) };
assert_eq!(id, self.id);
self.lock.push_front(val);
self.added.fetch_add(1, Ordering::Relaxed);
self.count.fetch_add(1, Ordering::Relaxed);
self.added.add(1, Ordering::Relaxed);
self.count.increment();
}
}