graphforge_filesystem/
observation.rs

1//! Process-wide counters for an explicit diagnostic capture. Never reset them.
2use std::fs::File;
3use std::io;
4use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
5use std::time::Instant;
6
7static CAPTURES: AtomicUsize = AtomicUsize::new(0);
8static FSYNC_CALLS: AtomicU64 = AtomicU64::new(0);
9static FSYNC_NS: AtomicU64 = AtomicU64::new(0);
10
11/// Keeps instrumented process counters active, including on worker threads.
12#[derive(Debug)]
13pub struct Observation(());
14impl Observation {
15    /// Start observing without resetting counters used by another capture.
16    #[must_use]
17    pub fn start() -> Self {
18        CAPTURES.fetch_add(1, Ordering::Relaxed);
19        Self(())
20    }
21}
22impl Drop for Observation {
23    fn drop(&mut self) {
24        CAPTURES.fetch_sub(1, Ordering::Relaxed);
25    }
26}
27/// Whether an explicit capture is active anywhere in this process.
28#[must_use]
29pub fn active() -> bool {
30    CAPTURES.load(Ordering::Relaxed) != 0
31}
32/// Cumulative successful and failed barrier attempts, and their elapsed time.
33#[must_use]
34pub fn fsync_totals() -> (u64, u64) {
35    (
36        FSYNC_CALLS.load(Ordering::Relaxed),
37        FSYNC_NS.load(Ordering::Relaxed),
38    )
39}
40
41/// File barriers with the same I/O behavior and optional diagnostic accounting.
42pub trait ObservedSync {
43    /// Invoke `File::sync_all`, accounting for the attempt when capturing.
44    fn observed_sync_all(&self) -> io::Result<()>;
45    /// Invoke `File::sync_data`, accounting for the attempt when capturing.
46    fn observed_sync_data(&self) -> io::Result<()>;
47}
48impl ObservedSync for File {
49    fn observed_sync_all(&self) -> io::Result<()> {
50        observe(|| self.sync_all())
51    }
52    fn observed_sync_data(&self) -> io::Result<()> {
53        observe(|| self.sync_data())
54    }
55}
56fn observe(operation: impl FnOnce() -> io::Result<()>) -> io::Result<()> {
57    if !active() {
58        return operation();
59    }
60    let started = Instant::now();
61    let result = operation();
62    FSYNC_NS.fetch_add(
63        u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX),
64        Ordering::Relaxed,
65    );
66    FSYNC_CALLS.fetch_add(1, Ordering::Relaxed);
67    result
68}
graphforge_filesystem/observation.rs

graphforge_filesystem/
observation.rs
