OXYGENHUB
# d2077a57b1f9 oxygenhub / crates / hub / src / repos / cache.rs
10.9 KiB · Rust · last changed State cache: weigh a loaded state by what it really costs 20 days ago source blame history raw
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;
10
use std::sync::{Arc, Mutex};
11
12
/// A state and its materialised snapshot.
13
pub struct Loaded {
14
    pub heads: Vec<ChangeHash>,
15
    pub state: State,
16
    pub snapshot: Snapshot,
17
}
18
19
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
20
struct Key {
21
    repo: [u8; 16],
22
    heads: Vec<ChangeHash>,
23
}
24
25
struct Entry {
26
    loaded: Arc<Loaded>,
27
    last_used: u64,
28
    /// Its estimated size in bytes (`weight`).
29
    weight: usize,
30
}
31
32
/// A loaded state's estimated size (oxygenhub#246): five bytes per byte of
33
/// text and 384 per line (each file's line graph), and a kilobyte per change.
34
/// Measured on 2026-10-17 against what loading a state really adds to the
35
/// process (`measures_a_real_state`): oxygen's states ~85 MiB, oxygenhub's
36
/// ~70 MiB, which this puts at 88 and 75. The older "text three times over"
37
/// said 18 and 28, so a cache that thought it held 2.5 GB held 6-8 GB.
38
fn weight(l: &Loaded) -> usize {
39
    let (mut text, mut lines) = (0usize, 0usize);
40
    for f in l.snapshot.files.values() {
41
        match &f.content {
42
            oxy_core::state::FileContent::Text { bytes, .. } => {
43
                text += bytes.len();
44
                lines += bytes.iter().filter(|b| **b == b'\n').count() + 1;
45
            }
46
            _ => text += 64,
47
        }
48
    }
49
    text.saturating_mul(5).saturating_add(lines.saturating_mul(384)).saturating_add(l.state.len().saturating_mul(1024))
50
}
51
52
/// What the state cache may hold, in bytes: OXYHUB_STATE_CACHE_MB, else
53
/// 3 GiB. A count alone let a few large repositories' states grow the hub
54
/// to 17 GB.
55
fn budget() -> usize {
56
    std::env::var("OXYHUB_STATE_CACHE_MB").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(3 * 1024).saturating_mul(1 << 20)
57
}
58
59
/// What the bookmark lists and a change's "in bookmarks" need of a state:
60
/// small, so many are kept, and computed without filling the state cache.
61
/// A repository with more bookmarks than the state cache holds otherwise
62
/// cycled through it on every such page, so nothing stayed cached
63
/// (metalgear's 128 bookmarks: 7.7 s per change page, and every page of
64
/// every repository slower after it).
65
pub struct Summary {
66
    pub changes: std::collections::HashSet<ChangeHash>,
67
    pub latest_ms: u64,
68
    pub conflicted: bool,
69
}
70
71
/// Summaries kept; each is the state's change set, a few hundred kilobytes
72
/// for a long history.
73
const SUMMARIES: usize = 2048;
74
75
/// Summaries by heads, each with when it was last used, and the clock.
76
type Summaries = (HashMap<Key, (Arc<Summary>, u64)>, u64);
77
78
/// A bounded least-recently-used cache of loaded states.
79
pub struct StateCache {
80
    capacity: usize,
81
    /// Bytes of estimated state it may hold (`budget`).
82
    budget: usize,
83
    inner: Mutex<(HashMap<Key, Entry>, u64)>,
84
    summaries: Mutex<Summaries>,
85
}
86
87
impl StateCache {
88
    pub fn new(capacity: usize) -> StateCache {
89
        StateCache::with_budget(capacity, budget())
90
    }
91
92
    pub fn with_budget(capacity: usize, budget: usize) -> StateCache {
93
        StateCache { capacity: capacity.max(1), budget, inner: Mutex::new((HashMap::new(), 0)), summaries: Mutex::new((HashMap::new(), 0)) }
94
    }
95
96
    /// The summary of the state with these heads: from the state cache when
97
    /// it holds the state, else loaded and dropped, the state cache left as
98
    /// it was.
99
    pub fn summary(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Summary>, oxy_core::repo::RepoError> {
100
        let mut sorted = heads.to_vec();
101
        sorted.sort();
102
        sorted.dedup();
103
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
104
        {
105
            let mut guard = self.summaries.lock().unwrap();
106
            let (map, tick) = &mut *guard;
107
            *tick += 1;
108
            if let Some(e) = map.get_mut(&key) {
109
                e.1 = *tick;
110
                return Ok(e.0.clone());
111
            }
112
        }
113
        let held = self.inner.lock().unwrap().0.get(&key).map(|e| e.loaded.clone());
114
        let summary = Arc::new(match held {
115
            Some(l) => summarise(&l.state, &l.snapshot),
116
            None => {
117
                let state = repo.load_state(&sorted)?;
118
                let snapshot = state.materialize();
119
                summarise(&state, &snapshot)
120
            }
121
        });
122
        let mut guard = self.summaries.lock().unwrap();
123
        let (map, tick) = &mut *guard;
124
        if map.len() >= SUMMARIES {
125
            if let Some(oldest) = map.iter().min_by_key(|(_, e)| e.1).map(|(k, _)| k.clone()) {
126
                map.remove(&oldest);
127
            }
128
        }
129
        map.insert(key, (summary.clone(), *tick));
130
        Ok(summary)
131
    }
132
133
    /// The state with these heads, loaded once per hub process.
134
    pub fn load(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Loaded>, oxy_core::repo::RepoError> {
135
        let mut sorted = heads.to_vec();
136
        sorted.sort();
137
        sorted.dedup();
138
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
139
        {
140
            let mut guard = self.inner.lock().unwrap();
141
            let (map, tick) = &mut *guard;
142
            *tick += 1;
143
            if let Some(e) = map.get_mut(&key) {
144
                e.last_used = *tick;
145
                return Ok(e.loaded.clone());
146
            }
147
        }
148
        let state = repo.load_state(&sorted)?;
149
        let snapshot = state.materialize();
150
        let loaded = Arc::new(Loaded { heads: sorted, state, snapshot });
151
        let w = weight(&loaded);
152
        let mut guard = self.inner.lock().unwrap();
153
        let (map, tick) = &mut *guard;
154
        // The oldest go until it fits by count and by size; the new one
155
        // stays even when it alone is over the budget.
156
        let budget = self.budget;
157
        let mut total: usize = map.values().map(|e| e.weight).sum::<usize>() + w;
158
        while !map.is_empty() && (map.len() >= self.capacity || total > budget) {
159
            let Some(oldest) = map.iter().min_by_key(|(_, e)| e.last_used).map(|(k, _)| k.clone()) else {
160
                break;
161
            };
162
            if let Some(e) = map.remove(&oldest) {
163
                total -= e.weight;
164
            }
165
        }
166
        map.insert(key, Entry { loaded: loaded.clone(), last_used: *tick, weight: w });
167
        Ok(loaded)
168
    }
169
170
    pub fn len(&self) -> usize {
171
        self.inner.lock().unwrap().0.len()
172
    }
173
174
    pub fn is_empty(&self) -> bool {
175
        self.len() == 0
176
    }
177
}
178
179
fn summarise(state: &State, snapshot: &Snapshot) -> Summary {
180
    Summary {
181
        changes: state.changes().map(|(h, _)| *h).collect(),
182
        latest_ms: state.changes().map(|(_, m)| m.hlc.wall_ms).max().unwrap_or(0),
183
        conflicted: snapshot.is_conflicted(),
184
    }
185
}
186
187
#[cfg(test)]
188
mod tests {
189
    use super::*;
190
191
    /// What a loaded state really costs against `weight` (oxygenhub#246):
192
    /// `OXYHUB_MEASURE_REPO=path cargo test -p oxygenhub --lib
193
    /// measures_a_real_state -- --ignored --nocapture`.
194
    #[test]
195
    #[ignore]
196
    fn measures_a_real_state() {
197
        fn rss_kib() -> usize {
198
            let s = std::fs::read_to_string("/proc/self/status").unwrap();
199
            s.lines().find(|l| l.starts_with("VmRSS:")).and_then(|l| l.split_whitespace().nth(1)).unwrap().parse().unwrap()
200
        }
201
        let path = std::env::var("OXYHUB_MEASURE_REPO").expect("OXYHUB_MEASURE_REPO");
202
        let repo = Repo::open(std::path::Path::new(&path)).unwrap();
203
        let view = repo.view().unwrap();
204
        // Kept alive, so each load's memory is new, not reused.
205
        let cache = StateCache::with_budget(64, usize::MAX);
206
        let mut held = Vec::new();
207
        for bookmark in view.bookmarks.keys().take(6).cloned().collect::<Vec<_>>() {
208
            let heads = repo.heads_of(&view, &bookmark).unwrap();
209
            let before = rss_kib();
210
            let loaded = cache.load(&repo, &heads).unwrap();
211
            let after = rss_kib();
212
            let files = loaded.snapshot.files.len();
213
            let (mut text, mut lines) = (0usize, 0usize);
214
            for f in loaded.snapshot.files.values() {
215
                if let oxy_core::state::FileContent::Text { bytes, .. } = &f.content {
216
                    text += bytes.len();
217
                    lines += bytes.iter().filter(|b| **b == b'\n').count();
218
                }
219
            }
220
            println!(
221
                "{bookmark}: {} changes, {files} files, {} KiB text, {lines} lines: weight {} MiB, resident +{} MiB",
222
                loaded.state.len(),
223
                text >> 10,
224
                weight(&loaded) >> 20,
225
                (after.saturating_sub(before)) >> 10
226
            );
227
            held.push(loaded);
228
        }
229
    }
230
231
    #[test]
232
    fn caches_by_heads_and_evicts_the_oldest() {
233
        let dir = tempfile::tempdir().unwrap();
234
        let repo = Repo::init_bare(dir.path()).unwrap();
235
        let cache = StateCache::new(2);
236
        let a = cache.load(&repo, &[]).unwrap();
237
        let b = cache.load(&repo, &[]).unwrap();
238
        assert!(Arc::ptr_eq(&a, &b), "same heads, same entry");
239
        assert_eq!(cache.len(), 1);
240
        let h1 = oxy_core::hash::ObjectRef([1; 32]);
241
        let h2 = oxy_core::hash::ObjectRef([2; 32]);
242
        // Unknown heads fail to load and are not cached.
243
        assert!(cache.load(&repo, &[h1]).is_err());
244
        assert_eq!(cache.len(), 1);
245
        let _ = (h2, dir);
246
    }
247
248
    #[test]
249
    fn summaries_leave_the_state_cache_alone() {
250
        let dir = tempfile::tempdir().unwrap();
251
        let repo = Repo::init_bare(dir.path()).unwrap();
252
        let cache = StateCache::new(2);
253
        let a = cache.summary(&repo, &[]).unwrap();
254
        assert_eq!(cache.len(), 0, "a summary does not fill the state cache");
255
        assert!(Arc::ptr_eq(&a, &cache.summary(&repo, &[]).unwrap()), "kept");
256
        assert!(a.changes.is_empty() && !a.conflicted);
257
        assert!(cache.summary(&repo, &[oxy_core::hash::ObjectRef([3; 32])]).is_err());
258
    }
259
260
    #[test]
261
    fn a_size_budget_evicts_the_oldest_and_keeps_the_newest() {
262
        let dir = tempfile::tempdir().unwrap();
263
        let repo = Repo::init(dir.path()).unwrap();
264
        std::fs::write(dir.path().join("a.txt"), "a\n".repeat(1000)).unwrap();
265
        let one = repo.commit("one", &[]).unwrap();
266
        std::fs::write(dir.path().join("b.txt"), "b\n".repeat(1000)).unwrap();
267
        let two = repo.commit("two", &[]).unwrap();
268
        // Room for nothing: each load stays, the one before goes.
269
        let cache = StateCache::with_budget(64, 0);
270
        cache.load(&repo, &[one]).unwrap();
271
        assert_eq!(cache.len(), 1, "kept though over the budget");
272
        cache.load(&repo, &[two]).unwrap();
273
        assert_eq!(cache.len(), 1, "the older one went");
274
        // Room for plenty: both stay.
275
        let roomy = StateCache::with_budget(64, 1 << 30);
276
        roomy.load(&repo, &[one]).unwrap();
277
        roomy.load(&repo, &[two]).unwrap();
278
        assert_eq!(roomy.len(), 2);
279
    }
280
}