OXYGENHUB
◆ v0.9.0 oxygenhub / crates / hub / src / repos / cache.rs
17.8 KiB · Rust · last changed State cache: weigh a loaded state by what it really costs 19 days ago source blame history raw
9ff27da5 Andy McCall ↶ 1
//! Loaded states cached by their heads (hub spec §16.3). A state is a pure
2
//! function of its heads, so entries never go stale; the cache is only
3
//! bounded. Shared across threads because `oxy_core::state::State` is
4
//! `Send + Sync`.
5
6
use oxy_core::hash::ChangeHash;
7
use oxy_core::repo::Repo;
8
use oxy_core::state::{Snapshot, State};
9
use std::collections::HashMap;
39442b00 Andy McCall ↶ 10
use std::path::{Path, PathBuf};
9ff27da5 Andy McCall ↶ 11
use std::sync::{Arc, Mutex};
12
13
/// A state and its materialised snapshot.
14
pub struct Loaded {
15
    pub heads: Vec<ChangeHash>,
16
    pub state: State,
17
    pub snapshot: Snapshot,
18
}
19
20
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
21
struct Key {
22
    repo: [u8; 16],
23
    heads: Vec<ChangeHash>,
24
}
25
26
struct Entry {
27
    loaded: Arc<Loaded>,
28
    last_used: u64,
736ae87e Andy McCall ↶ 29
    /// Its estimated size in bytes (`weight`).
30
    weight: usize,
9ff27da5 Andy McCall ↶ 31
}
32
d2077a57 Andy McCall ↶ 33
/// A loaded state's estimated size (oxygenhub#246): five bytes per byte of
34
/// text and 384 per line (each file's line graph), and a kilobyte per change.
35
/// Measured on 2026-10-17 against what loading a state really adds to the
36
/// process (`measures_a_real_state`): oxygen's states ~85 MiB, oxygenhub's
37
/// ~70 MiB, which this puts at 88 and 75. The older "text three times over"
38
/// said 18 and 28, so a cache that thought it held 2.5 GB held 6-8 GB.
736ae87e Andy McCall ↶ 39
fn weight(l: &Loaded) -> usize {
d2077a57 Andy McCall ↶ 40
    let (mut text, mut lines) = (0usize, 0usize);
41
    for f in l.snapshot.files.values() {
42
        match &f.content {
43
            oxy_core::state::FileContent::Text { bytes, .. } => {
44
                text += bytes.len();
45
                lines += bytes.iter().filter(|b| **b == b'\n').count() + 1;
46
            }
47
            _ => text += 64,
48
        }
49
    }
50
    text.saturating_mul(5).saturating_add(lines.saturating_mul(384)).saturating_add(l.state.len().saturating_mul(1024))
736ae87e Andy McCall ↶ 51
}
52
53
/// What the state cache may hold, in bytes: OXYHUB_STATE_CACHE_MB, else
54
/// 3 GiB. A count alone let a few large repositories' states grow the hub
55
/// to 17 GB.
b2107275 Andy McCall ↶ 56
pub fn budget() -> usize {
736ae87e Andy McCall ↶ 57
    std::env::var("OXYHUB_STATE_CACHE_MB").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(3 * 1024).saturating_mul(1 << 20)
58
}
59
2e640ffa Andy McCall ↶ 60
/// What the bookmark lists and a change's "in bookmarks" need of a state:
61
/// small, so many are kept, and computed without filling the state cache.
62
/// A repository with more bookmarks than the state cache holds otherwise
63
/// cycled through it on every such page, so nothing stayed cached
64
/// (metalgear's 128 bookmarks: 7.7 s per change page, and every page of
65
/// every repository slower after it).
66
pub struct Summary {
67
    pub changes: std::collections::HashSet<ChangeHash>,
68
    pub latest_ms: u64,
69
    pub conflicted: bool,
70
}
71
39442b00 Andy McCall ↶ 72
impl Summary {
73
    pub fn brief(&self) -> Brief {
74
        Brief { changes: self.changes.len(), latest_ms: self.latest_ms, conflicted: self.conflicted }
75
    }
76
}
77
78
/// What a bookmark list and the overview show of a state (#145): how many
79
/// changes, the latest, and whether it is conflicted. A pure function of
80
/// the heads, so it is kept on disk as well and a restart costs nothing:
81
/// metalgear's 162 bookmarks took ~31 s to summarise again after one.
82
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
83
pub struct Brief {
84
    pub changes: usize,
85
    pub latest_ms: u64,
86
    pub conflicted: bool,
87
}
88
2e640ffa Andy McCall ↶ 89
/// Summaries kept; each is the state's change set, a few hundred kilobytes
90
/// for a long history.
91
const SUMMARIES: usize = 2048;
39442b00 Andy McCall ↶ 92
/// Most summaries computed at once when a page needs many.
93
const PARALLEL: usize = 4;
2e640ffa Andy McCall ↶ 94
95
/// Summaries by heads, each with when it was last used, and the clock.
96
type Summaries = (HashMap<Key, (Arc<Summary>, u64)>, u64);
97
9ff27da5 Andy McCall ↶ 98
/// A bounded least-recently-used cache of loaded states.
99
pub struct StateCache {
100
    capacity: usize,
736ae87e Andy McCall ↶ 101
    /// Bytes of estimated state it may hold (`budget`).
102
    budget: usize,
9ff27da5 Andy McCall ↶ 103
    inner: Mutex<(HashMap<Key, Entry>, u64)>,
2e640ffa Andy McCall ↶ 104
    summaries: Mutex<Summaries>,
39442b00 Andy McCall ↶ 105
    /// Where briefs are kept on disk, when anywhere.
106
    briefs: Option<PathBuf>,
9ff27da5 Andy McCall ↶ 107
}
108
109
impl StateCache {
110
    pub fn new(capacity: usize) -> StateCache {
736ae87e Andy McCall ↶ 111
        StateCache::with_budget(capacity, budget())
112
    }
113
114
    pub fn with_budget(capacity: usize, budget: usize) -> StateCache {
39442b00 Andy McCall ↶ 115
        StateCache {
116
            capacity: capacity.max(1),
117
            budget,
118
            inner: Mutex::new((HashMap::new(), 0)),
119
            summaries: Mutex::new((HashMap::new(), 0)),
120
            briefs: None,
121
        }
122
    }
123
124
    /// Keep briefs on disk under `dir`, one small file per state.
125
    pub fn with_briefs(mut self, dir: PathBuf) -> StateCache {
126
        self.briefs = Some(dir);
127
        self
128
    }
129
130
    fn brief_path(&self, key: &Key) -> Option<PathBuf> {
131
        let dir = self.briefs.as_ref()?;
132
        let mut h = blake3::Hasher::new();
133
        for c in &key.heads {
134
            h.update(&c.0);
135
        }
136
        Some(dir.join(hex::encode(key.repo)).join(h.finalize().to_hex().as_str()))
137
    }
138
139
    fn known_brief(&self, key: &Key) -> Option<Brief> {
140
        if let Some((s, _)) = self.summaries.lock().unwrap().0.get(key) {
141
            return Some(s.brief());
142
        }
143
        read_brief(&self.brief_path(key)?)
144
    }
145
146
    /// The brief of the state with these heads: from memory, from disk, or
147
    /// worked out and kept.
148
    pub fn brief(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Brief, oxy_core::repo::RepoError> {
149
        Ok(self.briefs(repo, &[heads.to_vec()])?[0])
150
    }
151
152
    /// The briefs of many states, those not yet known worked out a few at
153
    /// a time in parallel.
154
    pub fn briefs(&self, repo: &Repo, all: &[Vec<ChangeHash>]) -> Result<Vec<Brief>, oxy_core::repo::RepoError> {
155
        let id = repo.id()?;
156
        let mut out: Vec<Option<Brief>> = all.iter().map(|h| self.known_brief(&Key { repo: id, heads: sorted(h) })).collect();
157
        let missing: Vec<usize> = (0..all.len()).filter(|&i| out[i].is_none()).collect();
158
        if missing.is_empty() {
159
            return Ok(out.into_iter().map(|b| b.expect("known")).collect());
160
        }
161
        let next = Mutex::new(missing.iter());
162
        let found: Mutex<Vec<(usize, Result<Brief, oxy_core::repo::RepoError>)>> = Mutex::new(Vec::new());
163
        let threads = PARALLEL.min(missing.len()).min(std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));
164
        // A `Repo` stays on its thread, so each worker opens its own.
165
        let root = repo.root().to_path_buf();
166
        std::thread::scope(|s| {
167
            for _ in 0..threads {
168
                s.spawn(|| {
169
                    let mine = Repo::open(&root);
170
                    loop {
171
                        let Some(&i) = next.lock().unwrap().next() else { break };
172
                        let b = match &mine {
173
                            Ok(r) => self.summary(r, &all[i]).map(|s| s.brief()),
174
                            Err(e) => Err(oxy_core::repo::RepoError::Msg(format!("open {}: {e}", root.display()))),
175
                        };
176
                        found.lock().unwrap().push((i, b));
177
                    }
178
                });
179
            }
180
        });
181
        for (i, b) in found.into_inner().unwrap() {
182
            out[i] = Some(b?);
183
        }
184
        Ok(out.into_iter().map(|b| b.expect("worked out")).collect())
2e640ffa Andy McCall ↶ 185
    }
186
187
    /// The summary of the state with these heads: from the state cache when
188
    /// it holds the state, else loaded and dropped, the state cache left as
189
    /// it was.
190
    pub fn summary(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Summary>, oxy_core::repo::RepoError> {
39442b00 Andy McCall ↶ 191
        let sorted = sorted(heads);
2e640ffa Andy McCall ↶ 192
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
193
        {
194
            let mut guard = self.summaries.lock().unwrap();
195
            let (map, tick) = &mut *guard;
196
            *tick += 1;
197
            if let Some(e) = map.get_mut(&key) {
198
                e.1 = *tick;
199
                return Ok(e.0.clone());
200
            }
201
        }
202
        let held = self.inner.lock().unwrap().0.get(&key).map(|e| e.loaded.clone());
203
        let summary = Arc::new(match held {
204
            Some(l) => summarise(&l.state, &l.snapshot),
205
            None => {
206
                let state = repo.load_state(&sorted)?;
207
                let snapshot = state.materialize();
208
                summarise(&state, &snapshot)
209
            }
210
        });
39442b00 Andy McCall ↶ 211
        if let Some(path) = self.brief_path(&key) {
212
            write_brief(&path, summary.brief());
213
        }
2e640ffa Andy McCall ↶ 214
        let mut guard = self.summaries.lock().unwrap();
215
        let (map, tick) = &mut *guard;
216
        if map.len() >= SUMMARIES {
217
            if let Some(oldest) = map.iter().min_by_key(|(_, e)| e.1).map(|(k, _)| k.clone()) {
218
                map.remove(&oldest);
219
            }
220
        }
221
        map.insert(key, (summary.clone(), *tick));
222
        Ok(summary)
9ff27da5 Andy McCall ↶ 223
    }
224
225
    /// The state with these heads, loaded once per hub process.
226
    pub fn load(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Loaded>, oxy_core::repo::RepoError> {
227
        let mut sorted = heads.to_vec();
228
        sorted.sort();
229
        sorted.dedup();
230
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
231
        {
232
            let mut guard = self.inner.lock().unwrap();
233
            let (map, tick) = &mut *guard;
234
            *tick += 1;
235
            if let Some(e) = map.get_mut(&key) {
236
                e.last_used = *tick;
237
                return Ok(e.loaded.clone());
238
            }
239
        }
240
        let state = repo.load_state(&sorted)?;
241
        let snapshot = state.materialize();
242
        let loaded = Arc::new(Loaded { heads: sorted, state, snapshot });
736ae87e Andy McCall ↶ 243
        let w = weight(&loaded);
9ff27da5 Andy McCall ↶ 244
        let mut guard = self.inner.lock().unwrap();
245
        let (map, tick) = &mut *guard;
736ae87e Andy McCall ↶ 246
        // The oldest go until it fits by count and by size; the new one
247
        // stays even when it alone is over the budget.
248
        let budget = self.budget;
249
        let mut total: usize = map.values().map(|e| e.weight).sum::<usize>() + w;
250
        while !map.is_empty() && (map.len() >= self.capacity || total > budget) {
251
            let Some(oldest) = map.iter().min_by_key(|(_, e)| e.last_used).map(|(k, _)| k.clone()) else {
252
                break;
253
            };
254
            if let Some(e) = map.remove(&oldest) {
255
                total -= e.weight;
9ff27da5 Andy McCall ↶ 256
            }
257
        }
736ae87e Andy McCall ↶ 258
        map.insert(key, Entry { loaded: loaded.clone(), last_used: *tick, weight: w });
9ff27da5 Andy McCall ↶ 259
        Ok(loaded)
260
    }
261
b2107275 Andy McCall ↶ 262
    /// What it holds: loaded states, their estimated bytes, and summaries.
263
    pub fn stats(&self) -> (usize, usize, usize) {
264
        let (entries, weight) = {
265
            let guard = self.inner.lock().unwrap();
266
            (guard.0.len(), guard.0.values().map(|e| e.weight).sum())
267
        };
268
        (entries, weight, self.summaries.lock().unwrap().0.len())
269
    }
270
9ff27da5 Andy McCall ↶ 271
    pub fn len(&self) -> usize {
272
        self.inner.lock().unwrap().0.len()
273
    }
274
275
    pub fn is_empty(&self) -> bool {
276
        self.len() == 0
39442b00 Andy McCall ↶ 277
    }
278
}
279
280
fn sorted(heads: &[ChangeHash]) -> Vec<ChangeHash> {
281
    let mut v = heads.to_vec();
282
    v.sort();
283
    v.dedup();
284
    v
285
}
286
287
/// A brief on disk: `changes latest_ms conflicted`.
288
fn read_brief(path: &Path) -> Option<Brief> {
289
    let text = std::fs::read_to_string(path).ok()?;
290
    let mut f = text.split_whitespace();
291
    Some(Brief { changes: f.next()?.parse().ok()?, latest_ms: f.next()?.parse().ok()?, conflicted: f.next()? == "1" })
292
}
293
294
/// Kept when it can be; a brief that is not kept is only worked out again.
295
fn write_brief(path: &Path, b: Brief) {
296
    let Some(dir) = path.parent() else { return };
297
    if std::fs::create_dir_all(dir).is_err() {
298
        return;
299
    }
300
    let tmp = path.with_extension(format!("tmp{}", std::process::id()));
301
    if std::fs::write(&tmp, format!("{} {} {}", b.changes, b.latest_ms, u8::from(b.conflicted))).is_ok() {
302
        std::fs::rename(&tmp, path).ok();
9ff27da5 Andy McCall ↶ 303
    }
304
}
305
2e640ffa Andy McCall ↶ 306
fn summarise(state: &State, snapshot: &Snapshot) -> Summary {
307
    Summary {
308
        changes: state.changes().map(|(h, _)| *h).collect(),
309
        latest_ms: state.changes().map(|(_, m)| m.hlc.wall_ms).max().unwrap_or(0),
310
        conflicted: snapshot.is_conflicted(),
311
    }
312
}
313
9ff27da5 Andy McCall ↶ 314
#[cfg(test)]
315
mod tests {
316
    use super::*;
d2077a57 Andy McCall ↶ 317
318
    /// What a loaded state really costs against `weight` (oxygenhub#246):
319
    /// `OXYHUB_MEASURE_REPO=path cargo test -p oxygenhub --lib
320
    /// measures_a_real_state -- --ignored --nocapture`.
321
    #[test]
322
    #[ignore]
323
    fn measures_a_real_state() {
324
        fn rss_kib() -> usize {
325
            let s = std::fs::read_to_string("/proc/self/status").unwrap();
326
            s.lines().find(|l| l.starts_with("VmRSS:")).and_then(|l| l.split_whitespace().nth(1)).unwrap().parse().unwrap()
327
        }
328
        let path = std::env::var("OXYHUB_MEASURE_REPO").expect("OXYHUB_MEASURE_REPO");
329
        let repo = Repo::open(std::path::Path::new(&path)).unwrap();
330
        let view = repo.view().unwrap();
331
        // Kept alive, so each load's memory is new, not reused.
332
        let cache = StateCache::with_budget(64, usize::MAX);
333
        let mut held = Vec::new();
334
        for bookmark in view.bookmarks.keys().take(6).cloned().collect::<Vec<_>>() {
335
            let heads = repo.heads_of(&view, &bookmark).unwrap();
336
            let before = rss_kib();
337
            let loaded = cache.load(&repo, &heads).unwrap();
338
            let after = rss_kib();
339
            let files = loaded.snapshot.files.len();
340
            let (mut text, mut lines) = (0usize, 0usize);
341
            for f in loaded.snapshot.files.values() {
342
                if let oxy_core::state::FileContent::Text { bytes, .. } = &f.content {
343
                    text += bytes.len();
344
                    lines += bytes.iter().filter(|b| **b == b'\n').count();
345
                }
346
            }
347
            println!(
348
                "{bookmark}: {} changes, {files} files, {} KiB text, {lines} lines: weight {} MiB, resident +{} MiB",
349
                loaded.state.len(),
350
                text >> 10,
351
                weight(&loaded) >> 20,
352
                (after.saturating_sub(before)) >> 10
353
            );
354
            held.push(loaded);
355
        }
356
    }
9ff27da5 Andy McCall ↶ 357
358
    #[test]
359
    fn caches_by_heads_and_evicts_the_oldest() {
360
        let dir = tempfile::tempdir().unwrap();
361
        let repo = Repo::init_bare(dir.path()).unwrap();
362
        let cache = StateCache::new(2);
363
        let a = cache.load(&repo, &[]).unwrap();
364
        let b = cache.load(&repo, &[]).unwrap();
365
        assert!(Arc::ptr_eq(&a, &b), "same heads, same entry");
366
        assert_eq!(cache.len(), 1);
367
        let h1 = oxy_core::hash::ObjectRef([1; 32]);
368
        let h2 = oxy_core::hash::ObjectRef([2; 32]);
369
        // Unknown heads fail to load and are not cached.
370
        assert!(cache.load(&repo, &[h1]).is_err());
371
        assert_eq!(cache.len(), 1);
372
        let _ = (h2, dir);
2e640ffa Andy McCall ↶ 373
    }
374
375
    #[test]
376
    fn summaries_leave_the_state_cache_alone() {
377
        let dir = tempfile::tempdir().unwrap();
378
        let repo = Repo::init_bare(dir.path()).unwrap();
379
        let cache = StateCache::new(2);
380
        let a = cache.summary(&repo, &[]).unwrap();
381
        assert_eq!(cache.len(), 0, "a summary does not fill the state cache");
382
        assert!(Arc::ptr_eq(&a, &cache.summary(&repo, &[]).unwrap()), "kept");
383
        assert!(a.changes.is_empty() && !a.conflicted);
384
        assert!(cache.summary(&repo, &[oxy_core::hash::ObjectRef([3; 32])]).is_err());
736ae87e Andy McCall ↶ 385
    }
386
387
    #[test]
388
    fn a_size_budget_evicts_the_oldest_and_keeps_the_newest() {
389
        let dir = tempfile::tempdir().unwrap();
390
        let repo = Repo::init(dir.path()).unwrap();
391
        std::fs::write(dir.path().join("a.txt"), "a\n".repeat(1000)).unwrap();
392
        let one = repo.commit("one", &[]).unwrap();
393
        std::fs::write(dir.path().join("b.txt"), "b\n".repeat(1000)).unwrap();
394
        let two = repo.commit("two", &[]).unwrap();
395
        // Room for nothing: each load stays, the one before goes.
396
        let cache = StateCache::with_budget(64, 0);
397
        cache.load(&repo, &[one]).unwrap();
398
        assert_eq!(cache.len(), 1, "kept though over the budget");
399
        cache.load(&repo, &[two]).unwrap();
400
        assert_eq!(cache.len(), 1, "the older one went");
401
        // Room for plenty: both stay.
402
        let roomy = StateCache::with_budget(64, 1 << 30);
403
        roomy.load(&repo, &[one]).unwrap();
404
        roomy.load(&repo, &[two]).unwrap();
405
        assert_eq!(roomy.len(), 2);
39442b00 Andy McCall ↶ 406
    }
407
408
    #[test]
409
    fn briefs_are_worked_out_in_parallel_and_survive_a_restart() {
410
        let dir = tempfile::tempdir().unwrap();
411
        let repo = Repo::init(dir.path()).unwrap();
412
        std::fs::write(dir.path().join("a.txt"), "a\n").unwrap();
413
        let one = repo.commit("one", &[]).unwrap();
414
        std::fs::write(dir.path().join("b.txt"), "b\n").unwrap();
415
        let two = repo.commit("two", &[]).unwrap();
416
        let kept = tempfile::tempdir().unwrap();
417
        let cache = StateCache::new(8).with_briefs(kept.path().to_path_buf());
418
        let got = cache.briefs(&repo, &[vec![one], vec![one, two], vec![]]).unwrap();
419
        assert_eq!(got.iter().map(|b| b.changes).collect::<Vec<_>>(), vec![1, 2, 0]);
420
        assert!(got[1].latest_ms >= got[0].latest_ms && !got[1].conflicted);
421
        // A new cache (a restarted hub) reads them back without loading a
422
        // state: a repository that no longer has the changes still answers.
423
        let gone = tempfile::tempdir().unwrap();
424
        let empty = Repo::init(gone.path()).unwrap();
425
        let empty_id = empty.id().unwrap();
426
        for f in std::fs::read_dir(kept.path()).unwrap() {
427
            let f = f.unwrap().path();
428
            std::fs::rename(&f, kept.path().join(hex::encode(empty_id))).unwrap();
429
        }
430
        let restarted = StateCache::new(8).with_briefs(kept.path().to_path_buf());
431
        assert_eq!(restarted.brief(&empty, &[two, one]).unwrap(), got[1], "from disk, whatever the heads' order");
432
        assert!(StateCache::new(8).brief(&empty, &[one, two]).is_err(), "without the disk it would have to load");
433
        // Unknown heads are an error, not a guess.
434
        assert!(cache.briefs(&repo, &[vec![one], vec![oxy_core::hash::ObjectRef([9; 32])]]).is_err());
435
        // A bare repository, as the hub keeps them, works across threads too.
436
        let bare_dir = tempfile::tempdir().unwrap();
437
        let bare = Repo::init_bare(bare_dir.path()).unwrap();
438
        assert_eq!(cache.briefs(&bare, &[vec![], vec![]]).unwrap().len(), 2);
9ff27da5 Andy McCall ↶ 439
    }
440
}