storage_resources/
storage_resources.rs

1//! Isolated reader measurement; generate inputs with tests/fixtures/file-storage/generate.py.
2use std::path::PathBuf;
3use std::sync::atomic::AtomicBool;
4use std::time::{Duration, Instant};
5
6use clap::{Parser, ValueEnum};
7use girt::{AlternateLimits, ObjectId, PackLimits, ReadLimits, Repository};
8
9fn handles() -> Option<usize> {
10    #[cfg(unix)]
11    {
12        let directory = if cfg!(target_os = "linux") {
13            "/proc/self/fd"
14        } else {
15            "/dev/fd"
16        };
17        std::fs::read_dir(directory)
18            .ok()
19            .map(|entries| entries.count())
20    }
21    #[cfg(not(unix))]
22    {
23        None
24    }
25}
26
27#[derive(Clone, Copy, PartialEq, Eq, ValueEnum)]
28enum Mode {
29    /// Measure cancellation of an in-flight store open instead of reads.
30    Cancel,
31    /// Keep the process alive for one second after measuring so it can be sampled.
32    Hold,
33}
34
35#[derive(Parser)]
36#[command(about, long_about = None)]
37struct Args {
38    /// Repository to open.
39    repository: PathBuf,
40    /// Object to read.
41    id: ObjectId,
42    /// Optional measurement mode.
43    #[arg(value_enum)]
44    mode: Option<Mode>,
45}
46
47fn main() -> Result<(), Box<dyn std::error::Error>> {
48    let args = Args::parse();
49    let repository = Repository::open(args.repository)?;
50    let id = args.id;
51
52    let baseline = handles();
53    let failed = repository.objects_with(PackLimits::default().with_max_index_bytes(0));
54    assert!(matches!(
55        failed,
56        Err(girt::ObjectReadError::Limit("pack index bytes"))
57    ));
58    assert_eq!(handles(), baseline);
59
60    if args.mode == Some(Mode::Cancel) {
61        let cancelled = AtomicBool::new(false);
62        let start = Instant::now();
63
64        let result = std::thread::scope(|scope| {
65            scope.spawn(|| {
66                std::thread::sleep(Duration::from_millis(10));
67                cancelled.store(true, std::sync::atomic::Ordering::Relaxed);
68            });
69            repository.objects_controlled(
70                PackLimits::default(),
71                AlternateLimits::default(),
72                &cancelled,
73            )
74        });
75
76        assert!(matches!(result, Err(girt::ObjectReadError::Cancelled)));
77        assert_eq!(handles(), baseline);
78
79        println!(
80            "cancel_us={} handles_before={baseline:?} handles_after={:?}",
81            start.elapsed().as_micros(),
82            handles()
83        );
84        return Ok(());
85    }
86
87    let cancelled = AtomicBool::new(false);
88    let start = Instant::now();
89    let mut objects = repository.objects_controlled(
90        PackLimits::default(),
91        AlternateLimits::default(),
92        &cancelled,
93    )?;
94
95    let opened = start.elapsed();
96    let retained = handles();
97    let clone = objects.clone();
98    let cloned = handles();
99
100    let failed = objects.refresh(
101        PackLimits::default().with_max_index_bytes(0),
102        AlternateLimits::default(),
103    );
104    assert!(matches!(
105        failed,
106        Err(girt::ObjectReadError::Limit("pack index bytes"))
107    ));
108    assert_eq!(handles(), cloned);
109
110    let start = Instant::now();
111    objects.refresh(PackLimits::default(), AlternateLimits::default())?;
112    let refreshed = start.elapsed();
113
114    let generations = handles();
115    assert_eq!(
116        generations,
117        baseline
118            .zip(retained)
119            .map(|(base, open)| base + 2 * (open - base))
120    );
121
122    println!(
123        "refresh_us={} handles_two_generations={generations:?}",
124        refreshed.as_micros()
125    );
126
127    let start = Instant::now();
128    let object = clone
129        .read_controlled(id, ReadLimits::default(), &cancelled)?
130        .ok_or("object missing")?;
131    let read = start.elapsed();
132
133    let start = Instant::now();
134    let second = clone
135        .read_with(id, ReadLimits::default())?
136        .ok_or("object missing")?;
137    let reread = start.elapsed();
138
139    assert_eq!(object, second);
140    println!(
141        "pid={} open_us={} first_read_us={} warm_read_us={} payload_bytes={} handles_before={baseline:?} handles_open={retained:?} handles_clone={cloned:?}",
142        std::process::id(),
143        opened.as_micros(),
144        read.as_micros(),
145        reread.as_micros(),
146        object.data().len()
147    );
148
149    drop(second);
150    drop(object);
151    drop(objects);
152    drop(clone);
153
154    println!("handles_after_drop={:?}", handles());
155    assert_eq!(handles(), baseline);
156
157    // Optional sampling interval is outside measured operations and retains no reader.
158    if args.mode == Some(Mode::Hold) {
159        std::thread::sleep(Duration::from_secs(1));
160    }
161    Ok(())
162}
storage_resources/storage_resources.rs

storage_resources/
storage_resources.rs
