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::Mutex;
5use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
6use std::time::Instant;
7
8use crate::FileIdentity;
9
10static CAPTURES: AtomicUsize = AtomicUsize::new(0);
11static FSYNC_CALLS: AtomicU64 = AtomicU64::new(0);
12static FSYNC_NS: AtomicU64 = AtomicU64::new(0);
13static FSYNC_DIRECTORY_CALLS: AtomicU64 = AtomicU64::new(0);
14
15/// Keeps instrumented process counters active, including on worker threads.
16#[derive(Debug)]
17pub struct Observation(());
18impl Observation {
19    /// Start observing without resetting counters used by another capture.
20    #[must_use]
21    pub fn start() -> Self {
22        CAPTURES.fetch_add(1, Ordering::Relaxed);
23        Self(())
24    }
25}
26impl Drop for Observation {
27    fn drop(&mut self) {
28        CAPTURES.fetch_sub(1, Ordering::Relaxed);
29    }
30}
31/// Mirrors `AuditState::holders` so the barrier path can skip the lock when no
32/// audit runs. Only ever written under the lock.
33static AUDITS: AtomicUsize = AtomicUsize::new(0);
34static AUDIT: Mutex<AuditState> = Mutex::new(AuditState {
35    holders: 0,
36    log: Vec::new(),
37});
38
39struct AuditState {
40    /// Audits alive. The log is kept, and grows, only while there are some.
41    holders: usize,
42    log: Vec<BarrierEntry>,
43}
44
45/// What the log keeps for one barrier attempt: the record, or why none could be
46/// made. A barrier that cannot be attributed is not dropped from the audit.
47#[derive(Clone, Debug)]
48enum BarrierEntry {
49    Record(BarrierRecord),
50    Unattributed(String),
51}
52
53fn audit() -> std::sync::MutexGuard<'static, AuditState> {
54    AUDIT
55        .lock()
56        .unwrap_or_else(std::sync::PoisonError::into_inner)
57}
58
59/// One barrier attempt: the inode it reached and whether that is a directory.
60/// A filesystem reuses the inode number of a deleted file, and its creation time
61/// repeats within a coarse timer tick, so neither says which file a barrier
62/// reached. The inode's generation number, which the filesystem changes every
63/// time it reuses the inode, does.
64#[derive(Clone, Copy, Debug, PartialEq, Eq)]
65pub struct BarrierRecord {
66    /// The inode the barrier reached.
67    pub identity: FileIdentity,
68    /// The generation number of that inode ([`crate::inode_generation`]).
69    pub generation: u32,
70    /// Whether it is a directory.
71    pub directory: bool,
72}
73
74/// Everything an audit recorded.
75#[derive(Clone, Debug, Default, PartialEq, Eq)]
76pub struct BarrierLog {
77    /// The barriers attempted, in order.
78    pub records: Vec<BarrierRecord>,
79    /// One message for each barrier that reached a file the audit could not
80    /// name (an identity or generation number it could not read). A correct
81    /// audit has none, and a test asserts so.
82    pub unattributed: Vec<String>,
83}
84
85/// Records which file or directory each barrier attempt reached, for as long
86/// as it lives: an exact inventory of barriers per inode. The counters above say
87/// how many barriers ran; this says on what. Barriers on every thread of the
88/// process are recorded, so it is meant for single-purpose processes and tests
89/// that serialize their measurements. The log grows by one entry per barrier.
90#[derive(Debug)]
91pub struct BarrierAudit {
92    start: usize,
93}
94impl BarrierAudit {
95    /// Begin recording barrier attempts.
96    #[must_use]
97    pub fn start() -> Self {
98        let mut state = audit();
99        state.holders += 1;
100        AUDITS.store(state.holders, Ordering::SeqCst);
101        Self {
102            start: state.log.len(),
103        }
104    }
105
106    /// The barriers attempted since [`BarrierAudit::start`], in order.
107    #[must_use]
108    pub fn finish(self) -> BarrierLog {
109        let state = audit();
110        let mut log = BarrierLog::default();
111        // The log only shrinks when no audit is alive, so `start` is in range.
112        for entry in &state.log[self.start.min(state.log.len())..] {
113            match entry {
114                BarrierEntry::Record(record) => log.records.push(*record),
115                BarrierEntry::Unattributed(reason) => log.unattributed.push(reason.clone()),
116            }
117        }
118        log
119    }
120}
121impl Drop for BarrierAudit {
122    fn drop(&mut self) {
123        let mut state = audit();
124        state.holders -= 1;
125        AUDITS.store(state.holders, Ordering::SeqCst);
126        if state.holders == 0 {
127            state.log.clear();
128        }
129    }
130}
131
132/// Whether an explicit capture is active anywhere in this process.
133#[must_use]
134pub fn active() -> bool {
135    CAPTURES.load(Ordering::Relaxed) != 0 || AUDITS.load(Ordering::Relaxed) != 0
136}
137/// Cumulative successful and failed barrier attempts, and their elapsed time.
138#[must_use]
139pub fn fsync_totals() -> (u64, u64) {
140    (
141        FSYNC_CALLS.load(Ordering::Relaxed),
142        FSYNC_NS.load(Ordering::Relaxed),
143    )
144}
145
146/// Cumulative barrier attempts on directories, a subset of the calls
147/// [`fsync_totals`] counts. The rest are barriers on files.
148#[must_use]
149pub fn fsync_directory_calls() -> u64 {
150    FSYNC_DIRECTORY_CALLS.load(Ordering::Relaxed)
151}
152
153/// File barriers with the same I/O behavior and optional diagnostic accounting.
154pub trait ObservedSync {
155    /// Invoke `File::sync_all`, accounting for the attempt when capturing.
156    fn observed_sync_all(&self) -> io::Result<()>;
157    /// Invoke `File::sync_data`, accounting for the attempt when capturing.
158    fn observed_sync_data(&self) -> io::Result<()>;
159}
160impl ObservedSync for File {
161    fn observed_sync_all(&self) -> io::Result<()> {
162        observe(self, || self.sync_all())
163    }
164    fn observed_sync_data(&self) -> io::Result<()> {
165        observe(self, || self.sync_data())
166    }
167}
168fn observe(file: &File, operation: impl FnOnce() -> io::Result<()>) -> io::Result<()> {
169    if !active() {
170        return operation();
171    }
172    let metadata = file.metadata();
173    let directory = metadata.as_ref().is_ok_and(std::fs::Metadata::is_dir);
174    if directory {
175        FSYNC_DIRECTORY_CALLS.fetch_add(1, Ordering::Relaxed);
176    }
177    if AUDITS.load(Ordering::SeqCst) != 0 {
178        let entry = match (
179            &metadata,
180            crate::file_identity(file),
181            crate::inode_generation(file),
182        ) {
183            (Ok(_), Ok(identity), Ok(generation)) => BarrierEntry::Record(BarrierRecord {
184                identity,
185                generation,
186                directory,
187            }),
188            (metadata, identity, generation) => BarrierEntry::Unattributed(format!(
189                "metadata: {:?}; identity: {:?}; generation: {:?}",
190                metadata.as_ref().err(),
191                identity.err(),
192                generation.err()
193            )),
194        };
195        let mut state = audit();
196        if state.holders != 0 {
197            state.log.push(entry);
198        }
199    }
200    let started = Instant::now();
201    let result = operation();
202    FSYNC_NS.fetch_add(
203        u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX),
204        Ordering::Relaxed,
205    );
206    FSYNC_CALLS.fetch_add(1, Ordering::Relaxed);
207    result
208}
209
210#[cfg(test)]
211mod tests {
212    use super::*;
213
214    /// The audit is process-wide, and `cargo test` runs a crate's tests on
215    /// threads of one process: the tests that start audits take turns, as each
216    /// asserts on the state when no other audit is alive.
217    static TURNS: Mutex<()> = Mutex::new(());
218
219    fn take_turn() -> std::sync::MutexGuard<'static, ()> {
220        TURNS
221            .lock()
222            .unwrap_or_else(std::sync::PoisonError::into_inner)
223    }
224
225    fn barrier_on(file: &File) {
226        file.observed_sync_all().unwrap();
227    }
228
229    /// How many barriers of `identity` the log holds. Where the platform names
230    /// inodes (Linux) these are the records of that inode; elsewhere no barrier
231    /// can be attributed, so the log's entries, all unattributed, are counted:
232    /// the lifecycle properties below do not depend on attribution, and a test
233    /// there is the only user of the audit in its process.
234    fn mine(log: &BarrierLog, identity: FileIdentity) -> usize {
235        if cfg!(target_os = "linux") {
236            assert!(log.unattributed.is_empty(), "{:?}", log.unattributed);
237            log.records
238                .iter()
239                .filter(|record| record.identity == identity)
240                .count()
241        } else {
242            // Never a weaker name: without a generation number nothing is
243            // recorded as attributed.
244            assert!(log.records.is_empty(), "{:?}", log.records);
245            log.unattributed.len()
246        }
247    }
248
249    /// The log holds at least `expected` barriers of the file: exactly where
250    /// barriers are attributed, and the barriers of any other thread's audit
251    /// too where they are not.
252    fn assert_mine(log: &BarrierLog, identity: FileIdentity, expected: usize, what: &str) {
253        let held = mine(log, identity);
254        if cfg!(target_os = "linux") {
255            assert_eq!(held, expected, "{what}");
256        } else {
257            assert!(held >= expected, "{what}: {held} < {expected}");
258        }
259    }
260
261    /// Audits nest and overlap: one ending clears nothing another still needs,
262    /// and the log is only cleared with the last, under the same lock that
263    /// starts them. Nothing here depends on attributing a barrier to a file.
264    #[test]
265    fn overlapping_audits_keep_their_logs_until_the_last_ends() {
266        let _turn = take_turn();
267        let directory = tempfile::tempdir().unwrap();
268        let file = File::create(directory.path().join("file")).unwrap();
269        let identity = crate::file_identity(&file).unwrap();
270        let first = BarrierAudit::start();
271        barrier_on(&file);
272        let second = BarrierAudit::start();
273        barrier_on(&file);
274        // The first audit, ending, leaves the second's records intact.
275        assert_mine(&first.finish(), identity, 2, "first");
276        barrier_on(&file);
277        let second = second.finish();
278        assert_mine(&second, identity, 2, "second");
279        if cfg!(target_os = "linux") {
280            assert!(
281                second
282                    .records
283                    .iter()
284                    .filter(|record| record.identity == identity)
285                    .all(|record| !record.directory)
286            );
287        }
288        // Nothing is recorded, or kept, once no audit is alive.
289        barrier_on(&file);
290        assert!(audit().log.is_empty());
291        // A later audit starts from an empty log.
292        assert_mine(&BarrierAudit::start().finish(), identity, 0, "later");
293    }
294
295    /// Many threads start, use and end audits that overlap one another, some
296    /// ending by `finish` and some by drop. None may panic, and each audit's
297    /// slice holds the barriers its own thread made while it lived (exactly
298    /// them, where barriers are attributed to files).
299    ///
300    /// Start and drop used to adjust the live-audit count before taking the lock
301    /// that guards the log, so a start could read a log length that the last
302    /// drop then cleared, and `finish` sliced past the end.
303    #[test]
304    fn audits_started_and_ended_concurrently_keep_their_own_slices() {
305        const THREADS: usize = 16;
306        const ROUNDS: usize = 60_000;
307        let _turn = take_turn();
308        let directory = tempfile::tempdir().unwrap();
309        std::thread::scope(|scope| {
310            for thread in 0..THREADS {
311                let directory = directory.path();
312                scope.spawn(move || {
313                    let file = File::create(directory.join(format!("file-{thread}"))).unwrap();
314                    let identity = crate::file_identity(&file).unwrap();
315                    for round in 0..ROUNDS {
316                        let audit = BarrierAudit::start();
317                        let barriers = 1 + (round % 3);
318                        for _ in 0..barriers {
319                            barrier_on(&file);
320                        }
321                        if round % 4 == 3 {
322                            drop(audit);
323                        } else {
324                            assert_mine(&audit.finish(), identity, barriers, "own slice");
325                        }
326                    }
327                });
328            }
329        });
330        assert!(audit().log.is_empty());
331        assert_eq!(audit().holders, 0);
332    }
333
334    /// A barrier that reaches a file the audit cannot name is reported, not
335    /// dropped: procfs has no inode generation, and syncing it is refused.
336    #[cfg(target_os = "linux")]
337    #[test]
338    fn a_barrier_the_audit_cannot_name_is_reported() {
339        let _turn = take_turn();
340        let file = File::open("/proc/self/status").unwrap();
341        let audit = BarrierAudit::start();
342        // The sync itself fails on procfs; the attempt is what is recorded.
343        let _ = file.observed_sync_all();
344        let log = audit.finish();
345        assert!(log.records.is_empty(), "{:?}", log.records);
346        assert_eq!(log.unattributed.len(), 1, "{:?}", log.unattributed);
347        assert!(log.unattributed[0].contains("generation"));
348    }
349}
graphforge_filesystem/observation.rs

graphforge_filesystem/
observation.rs
