how_far_along/
ext.rs

1//! Report adapters. Importing this trait is optional.
2
3use crate::Report;
4use core::num::NonZeroU64;
5
6/// Adapters for any [`Report`] sink.
7pub trait ReportExt: Report {
8    /// Buffer one worker's reports, publishing them in batches of at least
9    /// `threshold` units, and the remainder on [`Batch::flush`] or drop.
10    ///
11    /// Batching reduces contention on a counter shared by many workers. It
12    /// never delays cancellation, because a batch only counts. Flush or drop
13    /// every batch before finishing the phase.
14    fn batched(self, threshold: NonZeroU64) -> Batch<Self>
15    where
16        Self: Sized,
17    {
18        Batch {
19            report: self,
20            threshold,
21            pending: 0,
22        }
23    }
24}
25impl<T: Report + ?Sized> ReportExt for T {}
26
27/// One worker's report buffer, from [`ReportExt::batched`].
28///
29/// It is single-owner by design: `advance` takes `&mut self`. Give each worker
30/// its own batch.
31#[derive(Debug)]
32#[must_use = "report through the batch; dropping it only flushes pending counts"]
33pub struct Batch<R: Report> {
34    report: R,
35    threshold: NonZeroU64,
36    pending: u64,
37}
38impl<R: Report> Batch<R> {
39    /// Buffer `completed` finished units, publishing once the threshold is met.
40    /// A sum that would overflow is published first, so counts never wrap.
41    #[track_caller]
42    pub fn advance(&mut self, completed: u64) {
43        if let Some(sum) = self.pending.checked_add(completed) {
44            self.pending = sum;
45        } else {
46            self.flush();
47            self.pending = completed;
48        }
49        if self.pending >= self.threshold.get() {
50            self.flush();
51        }
52    }
53    /// Publish every buffered unit. Nothing is reported when none are pending.
54    #[track_caller]
55    pub fn flush(&mut self) {
56        if self.pending != 0 {
57            self.report.advance(core::mem::take(&mut self.pending));
58        }
59    }
60    /// Units buffered but not yet visible to observers.
61    pub fn pending(&self) -> u64 {
62        self.pending
63    }
64}
65impl<R: Report> Drop for Batch<R> {
66    fn drop(&mut self) {
67        self.flush();
68    }
69}
how_far_along/ext.rs

how_far_along/
ext.rs
