1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
// Copyright 2017-2021 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Polkadot is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Metered variant of mpsc channels to be able to extract metrics.
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
};
use derive_more::Display;
mod bounded;
pub mod oneshot;
mod unbounded;
pub use self::{bounded::*, unbounded::*};
pub use coarsetime::Duration as CoarseDuration;
use coarsetime::Instant as CoarseInstant;
#[cfg(test)]
mod tests;
/// A peek into the inner state of a meter.
#[derive(Debug, Clone)]
pub struct Meter {
// Number of sends on this channel.
sent: Arc<AtomicUsize>,
// Number of receives on this channel.
received: Arc<AtomicUsize>,
// Number of times senders blocked while sending messages to a subsystem.
blocked: Arc<AtomicUsize>,
// Atomic ringbuffer of the last 50 time of flight values
tof: Arc<crossbeam_queue::ArrayQueue<CoarseDuration>>,
}
impl std::default::Default for Meter {
fn default() -> Self {
Self {
sent: Arc::new(AtomicUsize::new(0)),
received: Arc::new(AtomicUsize::new(0)),
blocked: Arc::new(AtomicUsize::new(0)),
tof: Arc::new(crossbeam_queue::ArrayQueue::new(100)),
}
}
}
/// A readout of sizes from the meter. Note that it is possible, due to asynchrony, for received
/// to be slightly higher than sent.
#[derive(Debug, Display, Clone, Default, PartialEq)]
#[display(fmt = "(sent={} received={})", sent, received)]
pub struct Readout {
/// The amount of messages sent on the channel, in aggregate.
pub sent: usize,
/// The amount of messages received on the channel, in aggregate.
pub received: usize,
/// How many times the caller blocked when sending messages.
pub blocked: usize,
/// Time of flight in micro seconds (us)
pub tof: Vec<CoarseDuration>,
}
impl Meter {
/// Count the number of items queued up inside the channel.
pub fn read(&self) -> Readout {
// when obtaining we don't care much about off by one
// accuracy
Readout {
sent: self.sent.load(Ordering::Relaxed),
received: self.received.load(Ordering::Relaxed),
blocked: self.blocked.load(Ordering::Relaxed),
tof: {
let mut acc = Vec::with_capacity(self.tof.len());
while let Some(value) = self.tof.pop() {
acc.push(value)
}
acc
},
}
}
fn note_sent(&self) -> usize {
self.sent.fetch_add(1, Ordering::Relaxed)
}
fn retract_sent(&self) {
self.sent.fetch_sub(1, Ordering::Relaxed);
}
fn note_received(&self) {
self.received.fetch_add(1, Ordering::Relaxed);
}
fn note_blocked(&self) {
self.blocked.fetch_add(1, Ordering::Relaxed);
}
fn note_time_of_flight(&self, tof: CoarseDuration) {
let _ = self.tof.force_push(tof);
}
}
/// Determine if this instance shall be measured
#[inline(always)]
fn measure_tof_check(nth: usize) -> bool {
if cfg!(test) {
// for tests, be deterministic and pick every second
nth & 0x01 == 0
} else {
use nanorand::Rng;
let mut rng = nanorand::WyRand::new_seed(nth as u64);
let pick = rng.generate_range(1_usize..=1000);
// measure 5.3%
pick <= 53
}
}
/// Measure the time of flight between insertion and removal
/// of a single type `T`
#[derive(Debug)]
pub enum MaybeTimeOfFlight<T> {
Bare(T),
WithTimeOfFlight(T, CoarseInstant),
}
impl<T> From<T> for MaybeTimeOfFlight<T> {
fn from(value: T) -> Self {
Self::Bare(value)
}
}
// Has some unexplicable conflict with a wildcard impl of std
impl<T> MaybeTimeOfFlight<T> {
/// Extract the inner `T` value.
pub fn into(self) -> T {
match self {
Self::Bare(value) => value,
Self::WithTimeOfFlight(value, _tof_start) => value,
}
}
}
impl<T> std::ops::Deref for MaybeTimeOfFlight<T> {
type Target = T;
fn deref(&self) -> &Self::Target {
match self {
Self::Bare(ref value) => value,
Self::WithTimeOfFlight(ref value, _tof_start) => value,
}
}
}