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 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415
//! 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
}
}