//! Synchronized types that allow access to a `Histogram` from multiple threads.
use crate::errors::*;
use crate::{Counter, Histogram};
use std::borrow::Borrow;
use std::borrow::BorrowMut;
use std::marker::PhantomData;
use std::ops::{AddAssign, Deref, DerefMut};
use std::sync::{atomic, Arc, Mutex};
use std::time;
/// A write-only handle to a [`SyncHistogram`].
///
/// This handle allows you to record samples from multiple threads, each with its own `Recorder`,
/// concurrently. Writes to a `Recorder` are wait-free and scalable except for when the
/// [`SyncHistogram`] initiates a _phase shift_. During a phase shift, the next write on each
/// associated `Recorder` merges its results into a shared [`Histogram`] that is then made
/// available to the [`SyncHistogram`] once the phase shift completes. Phase shifts should also be
/// relatively cheap for writers, as they mainly need to perform a channel send on an unbounded,
/// lock-free channel.
///
/// An idle `Recorder` will hold up a phase shift indefinitely, or until it times out (is using
/// [`SyncHistogram::refresh_timeout`]. If a `Recorder` will remain idle for extended periods of
/// time, it should call [`Recorder::idle`], which will tell the reader not to wait for this
/// particular writer.
///
/// When a `Recorder` is dropped, all samples are made visible to the next
/// [`SyncHistogram::refresh`].
#[derive(Debug)]
pub struct Recorder<C: Counter> {
local: Histogram<C>,
shared: Arc<Shared<C>>,
last_phase: usize,
}
// make it more ergonomic to record samples
impl<C: Counter> AddAssign<u64> for Recorder<C> {
fn add_assign(&mut self, value: u64) {
self.record(value).unwrap();
}
}
impl<C: Counter> Clone for Recorder<C> {
fn clone(&self) -> Self {
// reader will have to wait for one more recorder
{
let mut truth = self.shared.truth.lock().unwrap();
truth.recorders += 1;
}
// new recorder starts at the same phase as we do with an empty histogram
Recorder {
local: Histogram::new_from(&self.local),
shared: self.shared.clone(),
last_phase: self.last_phase,
}
}
}
impl<C: Counter> Drop for Recorder<C> {
fn drop(&mut self) {
// we'll need to decrement the # of recorders
let mut truth = self.shared.truth.lock().unwrap();
truth.recorders -= 1;
// we also want to communicate the remainder of our samples to the reader
// we do this under the lock so that if the reader reads .recorders after we update it
// above, it is guaranteed to see the samples from this recorder.
// we also _have_ to do it at some point during drop as the reader _may_ have read
// .recorders _before_ we decremented it above, in which case it's blocking on us!
// note that we cannot call self.update() here as it would drop the mutex guard
let h = Histogram::new_from(&self.local);
let h = std::mem::replace(&mut self.local, h);
let _ = self.shared.sender.send(h).is_ok(); // if this is err, the reader went away
// explicitly drop guard to ensure we don't accidentally drop it above
drop(truth);
}
}
#[derive(Debug)]
struct Critical {
recorders: usize,
}
#[derive(Debug)]
struct Shared<C: Counter> {
truth: Mutex<Critical>,
sender: crossbeam_channel::Sender<Histogram<C>>,
phase: atomic::AtomicUsize,
}
/// See [`IdleRecorder`]. This guard borrows the idle [`Recorder`].
pub type IdleRecorderGuard<'a, C> = IdleRecorder<&'a mut Recorder<C>, C>;
/// This guard denotes that a [`Recorder`] is currently idle, and should not be waited on by a
/// [`SyncHistogram`] phase-shift.
#[derive(Debug)]
pub struct IdleRecorder<T, C: Counter>
where
T: BorrowMut<Recorder<C>>,
{
recorder: Option<T>,
c: PhantomData<C>,
}
impl<T, C: Counter> IdleRecorder<T, C>
where
T: BorrowMut<Recorder<C>>,
{
fn reactivate(&mut self) {
let recorder = if let Some(ref mut r) = self.recorder {
r
} else {
// already reactivated
return;
};
let recorder = recorder.borrow_mut();
// the Recorder is no longer idle, so the reader has to wait for us again
// this basically means re-incrementing .recorders
let mut crit = recorder.shared.truth.lock().unwrap();
crit.recorders += 1;
// we need to figure out what phase we're joining
// the easiest way to do that is to adopt the current phase
//
// note that we have to load the phase while holding the lock.
// if we did not, the reader could come along, read our ++'d .recorders (and so wait for us
// to send), and bump the phase, all before we read it, which would lead us to believe that
// we were already synchronized when in reality we were not, which would stall the reader
// even if we issued more writes.
recorder.last_phase = recorder.shared.phase.load(atomic::Ordering::Acquire);
// explicitly drop guard to ensure we don't accidentally drop it above
drop(crit);
}
}
impl<C: Counter> IdleRecorder<Recorder<C>, C> {
/// Mark the wrapped [`Recorder`] as active again and return it.
pub fn activate(mut self) -> Recorder<C> {
self.reactivate();
self.recorder.take().unwrap()
}
/// Clone the wrapped [`Recorder`].
pub fn recorder(&self) -> Recorder<C> {
self.recorder.as_ref().unwrap().clone()
}
}
impl<T, C: Counter> Drop for IdleRecorder<T, C>
where
T: BorrowMut<Recorder<C>>,
{
fn drop(&mut self) {
self.reactivate()
}
}
impl<C: Counter> Recorder<C> {
fn with_hist<F, R>(&mut self, f: F) -> R
where
F: FnOnce(&mut Histogram<C>) -> R,
{
let r = f(&mut self.local);
let phase = self.shared.phase.load(atomic::Ordering::Acquire);
if phase != self.last_phase {
self.update();
self.last_phase = phase;
}
r
}
// return our current histogram and leave a cleared one in its place
fn shed(&mut self) -> Histogram<C> {
let h = Histogram::new_from(&self.local);
std::mem::replace(&mut self.local, h)
}
fn update(&mut self) {
let h = self.shed();
let _ = self.shared.sender.send(h).is_ok(); // if this is err, the reader went away
}
fn deactivate(&mut self) {
let phase;
{
// we're leaving rotation, so we need to decrement .recorders
let mut crit = self.shared.truth.lock().unwrap();
crit.recorders -= 1;
// make sure we don't hold up the current phase shift (if any)
phase = self.shared.phase.load(atomic::Ordering::Acquire);
if phase != self.last_phase {
// can't call self.update() due to borrow of self.shared above
let h = Histogram::new_from(&self.local);
let h = std::mem::replace(&mut self.local, h);
let _ = self.shared.sender.send(h).is_ok(); // if this is err, the reader went away
}
}
self.last_phase = phase;
}
/// Call this method if the Recorder will be idle for a while.
///
/// Until the returned guard is dropped, the associated [`SyncHistogram`] will not wait for
/// this recorder on a phase shift.
pub fn idle(&mut self) -> IdleRecorderGuard<C> {
self.deactivate();
IdleRecorder {
recorder: Some(self),
c: PhantomData,
}
}
/// Mark this `Recorder` as inactive.
///
/// Until the returned guard is consumed, either by calling [`IdleRecorder::activate`] or by
/// dropping it, the associated [`SyncHistogram`] will not wait for this recorder on a phase
/// shift.
pub fn into_idle(mut self) -> IdleRecorder<Self, C> {
self.deactivate();
IdleRecorder {
recorder: Some(self),
c: PhantomData,
}
}
/// See [`Histogram::add`].
pub fn add<B: Borrow<Histogram<C>>>(&mut self, source: B) -> Result<(), AdditionError> {
self.with_hist(move |h| h.add(source))
}
/// See [`Histogram::add_correct`].
pub fn add_correct<B: Borrow<Histogram<C>>>(
&mut self,
source: B,
interval: u64,
) -> Result<(), RecordError> {
self.with_hist(move |h| h.add_correct(source, interval))
}
/// See [`Histogram::subtract`].
pub fn subtract<B: Borrow<Histogram<C>>>(
&mut self,
subtrahend: B,
) -> Result<(), SubtractionError> {
self.with_hist(move |h| h.subtract(subtrahend))
}
/// See [`Histogram::record`].
pub fn record(&mut self, value: u64) -> Result<(), RecordError> {
self.with_hist(move |h| h.record(value))
}
/// See [`Histogram::saturating_record`].
pub fn saturating_record(&mut self, value: u64) {
self.with_hist(move |h| h.saturating_record(value))
}
/// See [`Histogram::record_n`].
pub fn record_n(&mut self, value: u64, count: C) -> Result<(), RecordError> {
self.with_hist(move |h| h.record_n(value, count))
}
/// See [`Histogram::saturating_record_n`].
pub fn saturating_record_n(&mut self, value: u64, count: C) {
self.with_hist(move |h| h.saturating_record_n(value, count))
}
/// See [`Histogram::record_correct`].
pub fn record_correct(&mut self, value: u64, interval: u64) -> Result<(), RecordError> {
self.with_hist(move |h| h.record_correct(value, interval))
}
/// See [`Histogram::record_n_correct`].
pub fn record_n_correct(
&mut self,
value: u64,
count: C,
interval: u64,
) -> Result<(), RecordError> {
self.with_hist(move |h| h.record_n_correct(value, count, interval))
}
}
/// A `Histogram` that can be written to by multiple threads concurrently.
///
/// Each writer thread should have a [`Recorder`], which allows it to record new samples without
/// synchronization. New recorded samples are made available through this histogram by calling
/// [`SyncHistogram::refresh`], which blocks until it has synchronized with every recorder.
#[derive(Debug)]
pub struct SyncHistogram<C: Counter> {
merged: Histogram<C>,
shared: Arc<Shared<C>>,
receiver: crossbeam_channel::Receiver<Histogram<C>>,
}
impl<C: Counter> SyncHistogram<C> {
fn refresh_inner(&mut self, timeout: Option<time::Duration>) {
let end = timeout.map(|dur| time::Instant::now() + dur);
// time to start a phase change
// we first want to drain any histograms left over by dropped recorders
// note that we do this _before_ incrementing the phase, so we know they're "old"
while let Ok(h) = self.receiver.try_recv() {
self.merged
.add(&h)
.expect("TODO: failed to merge histogram");
}
// make sure no recorders can join or leave in the middle of this
let recorders = self.shared.truth.lock().unwrap().recorders;
// then, we tell writers to phase
let _ = self.shared.phase.fetch_add(1, atomic::Ordering::AcqRel);
// we want to wait for writers to all have phased
let mut phased = 0;
// at this point, we expect to get at least truth.recorders histograms
while phased < recorders {
let h = if let Some(end) = end {
let now = time::Instant::now();
if now > end {
break;
}
match self.receiver.recv_timeout(end - now) {
Ok(h) => h,
Err(crossbeam_channel::RecvTimeoutError::Timeout) => break,
Err(crossbeam_channel::RecvTimeoutError::Disconnected) => unreachable!(),
}
} else {
self.receiver
.recv()
.expect("SyncHistogram has an Arc<Shared> with a Receiver")
};
self.merged
.add(&h)
.expect("TODO: failed to merge histogram");
phased += 1;
}
// we also gobble up extra histograms we may have been sent from more dropped writers
while let Ok(h) = self.receiver.try_recv() {
self.merged
.add(&h)
.expect("TODO: failed to merge histogram");
}
}
/// Block until writes from all [`Recorder`] instances for this histogram have been
/// incorporated.
pub fn refresh(&mut self) {
self.refresh_inner(None)
}
/// Block until writes from all [`Recorder`] instances for this histogram have been
/// incorporated, or until the given amount of time has passed.
pub fn refresh_timeout(&mut self, timeout: time::Duration) {
self.refresh_inner(Some(timeout))
}
/// Obtain another multi-threaded writer for this histogram.
///
/// Note that writes made to the `Recorder` will not be visible until the next call to
/// [`SyncHistogram::refresh`].
pub fn recorder(&self) -> Recorder<C> {
// we will have to wait for one more recorder
{
let mut truth = self.shared.truth.lock().unwrap();
truth.recorders += 1;
}
// new recorder starts at the current phase with an empty histogram
Recorder {
local: Histogram::new_from(&self.merged),
shared: self.shared.clone(),
last_phase: self.shared.phase.load(atomic::Ordering::Acquire),
}
}
}
impl<C: Counter> From<Histogram<C>> for SyncHistogram<C> {
fn from(h: Histogram<C>) -> Self {
let (tx, rx) = crossbeam_channel::unbounded();
SyncHistogram {
merged: h,
receiver: rx,
shared: Arc::new(Shared {
truth: Mutex::new(Critical { recorders: 0 }),
sender: tx,
phase: atomic::AtomicUsize::new(0),
}),
}
}
}
impl<C: Counter> Deref for SyncHistogram<C> {
type Target = Histogram<C>;
fn deref(&self) -> &Self::Target {
&self.merged
}
}
impl<C: Counter> DerefMut for SyncHistogram<C> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.merged
}
}
