persistant_blob/
speed.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! Stage 3 of backend benchmarking for the blob kind: **speed**, as data, with the environment
3//! recorded. The rules and the data types are the kind-agnostic ones in [`persistant_core::bench`]
4//! (only a backend that passes the contract leaf a workload depends on is timed; the rest are listed
5//! as excluded, never ranked; a figure carries its environment and the command that re-runs it).
6//! This module owns the blob workloads and how they are timed: wall-clock, one operation at a time,
7//! from the caller's side of the trait.
8
9use crate::conformance::Leaf;
10use crate::contract;
11use crate::{BlobBackend, BlobError};
12use persistant_core::choose::Eligibility;
13use persistant_core::{Config, Reason};
14use std::time::Instant;
15
16pub use persistant_core::bench::{
17    Comparison, Environment, Left, Outcome, Ranked, SpeedParseError as ParseError, SpeedRow,
18    SpeedTable, Stats, compare,
19};
20
21/// One thing worth timing, with the contract leaf it depends on.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
23#[non_exhaustive]
24pub enum Workload {
25    /// `write` of a 4 KiB value to a fresh key.
26    Write,
27    /// `read` of one present 4 KiB value.
28    Read,
29    /// `stat` of one present key.
30    Stat,
31    /// `list` pages of 50 entries over 100 keys.
32    List,
33    /// `create_exclusive` of a fresh key (the claim succeeds).
34    CreateExclusive,
35    /// `replace_atomic` of one key.
36    ReplaceAtomic,
37    /// `append` of 256 bytes to one key.
38    Append,
39}
40
41impl Workload {
42    /// Every workload.
43    pub const ALL: [Workload; 7] = [
44        Workload::Write,
45        Workload::Read,
46        Workload::Stat,
47        Workload::List,
48        Workload::CreateExclusive,
49        Workload::ReplaceAtomic,
50        Workload::Append,
51    ];
52
53    /// The id used in data files.
54    pub fn id(self) -> &'static str {
55        match self {
56            Workload::Write => "speed.blob.write",
57            Workload::Read => "speed.blob.read",
58            Workload::Stat => "speed.blob.stat",
59            Workload::List => "speed.blob.list",
60            Workload::CreateExclusive => "speed.blob.create_exclusive",
61            Workload::ReplaceAtomic => "speed.blob.replace_atomic",
62            Workload::Append => "speed.blob.append",
63        }
64    }
65
66    /// The contract leaf the workload needs: no pass, no number.
67    pub fn required(self) -> Leaf {
68        match self {
69            Workload::Write => Leaf::Write,
70            Workload::Read => Leaf::Read,
71            Workload::Stat => Leaf::Stat,
72            Workload::List => Leaf::List,
73            Workload::CreateExclusive => Leaf::CreateExclusive,
74            Workload::ReplaceAtomic => Leaf::ReplaceConcurrentRead,
75            Workload::Append => Leaf::Append,
76        }
77    }
78}
79
80/// Time `workloads` on `target`, `iterations` operations each. A workload whose leaf the target does
81/// not pass gets an excluded row and no timing.
82pub async fn run<B: BlobBackend>(
83    target: &B,
84    config: &Config,
85    env: &Environment,
86    workloads: &[Workload],
87    iterations: u32,
88) -> Vec<SpeedRow> {
89    let mut out = Vec::with_capacity(workloads.len());
90    for w in workloads.iter().copied() {
91        let leaf = w.required();
92        let outcome = match contract::check(target, &[leaf]).await {
93            Eligibility::Eligible => match time(target, w, iterations).await {
94                Ok(Some(s)) => Outcome::Measured(s),
95                Ok(None) => Outcome::Excluded {
96                    leaf: leaf.id().to_string(),
97                    reason: Reason::Untested,
98                    note: "no operations were timed".to_string(),
99                },
100                Err(e) => Outcome::Excluded {
101                    leaf: leaf.id().to_string(),
102                    reason: Reason::Missing,
103                    note: format!("the workload failed on a backend that passed the leaf: {e}"),
104                },
105            },
106            Eligibility::Excluded(ex) => {
107                let e = &ex[0];
108                Outcome::Excluded {
109                    leaf: e.leaf.clone(),
110                    reason: e.reason,
111                    note: e.note.clone(),
112                }
113            }
114        };
115        out.push(SpeedRow {
116            workload: w.id().to_string(),
117            backend: target.backend().to_string(),
118            version: target.version().to_string(),
119            config: config.clone(),
120            env: env.clone(),
121            outcome,
122        });
123    }
124    out
125}
126
127const VALUE_BYTES: usize = 4096;
128
129async fn time<B: BlobBackend>(
130    b: &B,
131    w: Workload,
132    iterations: u32,
133) -> Result<Option<Stats>, BlobError> {
134    let base = format!("persistant-speed/{}/", w.id());
135    let value = vec![0x5A_u8; VALUE_BYTES];
136    let mut us: Vec<u64> = Vec::with_capacity(iterations as usize);
137    let mut keys: Vec<String> = Vec::new();
138    macro_rules! timed {
139        ($e:expr) => {{
140            let t = Instant::now();
141            let r = $e;
142            us.push(u64::try_from(t.elapsed().as_nanos()).unwrap_or(u64::MAX));
143            r
144        }};
145    }
146    match w {
147        Workload::Write | Workload::CreateExclusive => {
148            for i in 0..iterations {
149                let k = format!("{base}{i:08}");
150                keys.push(k.clone());
151                if w == Workload::Write {
152                    timed!(b.write(&k, value.clone()).await?);
153                } else {
154                    let won = timed!(b.create_exclusive(&k, value.clone()).await?);
155                    if !won {
156                        return Err(BlobError::Backend(
157                            "create_exclusive on a fresh key did not create".to_string(),
158                        ));
159                    }
160                }
161            }
162        }
163        Workload::Read | Workload::Stat | Workload::ReplaceAtomic | Workload::Append => {
164            let k = format!("{base}k");
165            keys.push(k.clone());
166            b.write(&k, value.clone()).await?;
167            for _ in 0..iterations {
168                match w {
169                    Workload::Read => {
170                        timed!(b.read(&k).await?);
171                    }
172                    Workload::Stat => {
173                        timed!(b.stat(&k).await?);
174                    }
175                    Workload::ReplaceAtomic => timed!(b.replace_atomic(&k, value.clone()).await?),
176                    _ => timed!(b.append(&k, value[..256].to_vec()).await?),
177                }
178            }
179        }
180        Workload::List => {
181            for i in 0..100 {
182                let k = format!("{base}{i:04}");
183                keys.push(k.clone());
184                b.write(&k, value[..256].to_vec()).await?;
185            }
186            for _ in 0..iterations {
187                timed!(b.list(&base, None, 50).await?);
188            }
189        }
190    }
191    for k in &keys {
192        let _ = b.delete(k).await; // best effort: leave a shared target as found
193    }
194    Ok(Stats::from_samples(us))
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200    use crate::{Defect, MemoryBlob};
201
202    fn env() -> Environment {
203        Environment::detect("cargo test -p persistant-blob", None)
204    }
205
206    #[tokio::test]
207    async fn a_correct_backend_is_measured_for_every_workload_and_the_file_round_trips() {
208        let rows = run(
209            &MemoryBlob::new(),
210            &Config::new(),
211            &env(),
212            &Workload::ALL,
213            20,
214        )
215        .await;
216        assert_eq!(rows.len(), Workload::ALL.len());
217        for r in &rows {
218            let Outcome::Measured(s) = &r.outcome else {
219                panic!("{}: {:?}", r.workload, r.outcome)
220            };
221            assert_eq!(s.n, 20, "{}", r.workload);
222            assert!(s.p50_ns <= s.p95_ns && s.p95_ns <= s.p99_ns && s.p99_ns <= s.max_ns);
223        }
224        let t = SpeedTable::from_rows(rows);
225        assert_eq!(SpeedTable::parse(&t.to_tsv()).unwrap(), t);
226    }
227
228    #[tokio::test]
229    async fn a_backend_that_fails_the_contract_gets_no_number() {
230        let lossy = MemoryBlob::with_defect(Defect::WriteIsLost);
231        let rows = run(&lossy, &Config::new(), &env(), &[Workload::Write], 5).await;
232        assert!(
233            matches!(
234                &rows[0].outcome,
235                Outcome::Excluded {
236                    reason: Reason::Wrong,
237                    ..
238                }
239            ),
240            "{:?}",
241            rows[0].outcome
242        );
243        // Control: the same workload on a correct backend is measured.
244        let rows = run(
245            &MemoryBlob::new(),
246            &Config::new(),
247            &env(),
248            &[Workload::Write],
249            5,
250        )
251        .await;
252        assert!(matches!(rows[0].outcome, Outcome::Measured(_)));
253    }
254}
persistant_blob/speed.rs

persistant_blob/
speed.rs
