OXYGENHUB
◆ v0.6.0 oxygen / crates / oxy-core / src / repo / checkpoint.rs
10.5 KiB · Rust · last changed Depth 0 takes no change, and a checkpoint is gathered in parallel (refs #137) 11 days ago source blame history raw
1 
//! Checkpoints (spec §11): a serialized state kept per bookmark so loading
2 
//! a state costs the changes since the checkpoint rather than all of them.
3 
//!
4 
//! The newest checkpoint of each bookmark is pointed at by
5 
//! `.oxy/checkpoints/<bookmark>`; older ones become unreachable and are
6 
//! garbage collected. A checkpoint applies to a requested head set when all
7 
//! of its heads lie in that set's closure, which the loader establishes while
8 
//! walking from the requested heads and stopping at checkpoint members.
9
10 
use super::*;
11 
use crate::objects::Checkpoint;
12
13 
impl Repo {
14 
    fn checkpoint_pointer(&self, bookmark: &str) -> PathBuf {
15 
        self.oxy.join("checkpoints").join(name_file(bookmark))
16 
    }
17
18 
    /// The newest checkpoint recorded for a bookmark.
19 
    pub fn latest_checkpoint(&self, bookmark: &str) -> Result<Option<(ObjectRef, Checkpoint)>> {
20 
        match fs::read_to_string(self.checkpoint_pointer(bookmark)) {
21 
            Ok(h) => {
22 
                let r = ObjectRef::from_hex(h.trim())
23 
                    .ok_or_else(|| RepoError::Msg("corrupt checkpoint pointer".into()))?;
24 
                match self.store.get_typed(&r, ObjectType::Checkpoint) {
25 
                    Ok(bytes) => Ok(Some((r, Checkpoint::decode(&bytes)?))),
26 
                    Err(StoreError::NotFound(_)) => Ok(None),
27 
                    Err(e) => Err(e.into()),
28 
                }
29 
            }
30 
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
31 
            Err(e) => Err(e.into()),
32 
        }
33 
    }
34
35 
    /// Every checkpoint pointer, for garbage collection roots.
36 
    pub(crate) fn checkpoint_refs(&self) -> Result<Vec<ObjectRef>> {
37 
        let dir = self.oxy.join("checkpoints");
38 
        let mut out = Vec::new();
39 
        if let Ok(rd) = fs::read_dir(&dir) {
40 
            for e in rd.flatten() {
41 
                if let Ok(h) = fs::read_to_string(e.path()) {
42 
                    if let Some(r) = ObjectRef::from_hex(h.trim()) {
43 
                        out.push(r);
44 
                    }
45 
                }
46 
            }
47 
        }
48 
        Ok(out)
49 
    }
50
51 
    /// Write a checkpoint of `state` (whose heads are `heads`) for `bookmark`.
52 
    pub(crate) fn write_checkpoint(
53 
        &self,
54 
        bookmark: &str,
55 
        state: &State,
56 
        heads: &[ChangeHash],
57 
    ) -> Result<(ObjectRef, usize)> {
58 
        let batch = self.store.begin_batch()?;
59 
        let (blob, payload_len) = self.store_checkpoint_payload(state)?;
60 
        let cp = Checkpoint {
61 
            heads: StateObject::new(heads.to_vec()).heads,
62 
            count: state.len() as u64,
63 
            payload: blob,
64 
        };
65 
        let r = self.store.put(ObjectType::Checkpoint, &cp.encode())?;
66 
        batch.finish()?;
67 
        fs::create_dir_all(self.oxy.join("checkpoints"))?;
68 
        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())?;
69 
        Ok((r, payload_len))
70 
    }
71
72 
    /// Store `state` as a version 2 payload (spec §11): the graphs of large
73 
    /// files as blobs of their own, shared with earlier checkpoints when
74 
    /// unchanged, and the index naming them. Returns the index blob and its size.
75 
    pub(crate) fn store_checkpoint_payload(&self, state: &State) -> Result<(ObjectRef, usize)> {
76 
        let graphs = state.checkpoint_file_graphs();
77 
        for (_, enc) in &graphs {
78 
            if enc.len() >= crate::state::INLINE_GRAPH_LIMIT {
79 
                self.store_blob(enc)?;
80 
            }
81 
        }
82 
        let index = state.checkpoint_index(&graphs, Repo::blob_ref);
83 
        Ok((self.store_blob(&index)?, index.len()))
84 
    }
85
86 
    /// The graph blobs a checkpoint's payload names (none for version 1).
87 
    pub(crate) fn checkpoint_graph_blobs(&self, cp: &Checkpoint) -> Result<Vec<ObjectRef>> {
88 
        let payload = self.read_blob(&cp.payload)?;
89 
        Ok(State::checkpoint_graph_refs(&payload)?)
90 
    }
91
92 
    /// Decode a checkpoint's payload, reading its graph blobs from the store.
93 
    pub(crate) fn state_from_checkpoint(&self, payload: &[u8]) -> Result<State> {
94 
        Ok(State::from_checkpoint(payload, |refs| {
95 
            self.read_blobs(refs).map_err(|e| {
96 
                crate::objects::ObjectError::Schema(format!("checkpoint graph blobs: {e}"))
97 
            })
98 
        })?)
99 
    }
100
101 
    /// Checkpoint the current bookmark's state now.
102 
    pub fn checkpoint(&self) -> Result<(ObjectRef, usize, usize)> {
103 
        let heads = self.bookmark_heads()?;
104 
        let state = self.load_state_full(&heads)?;
105 
        let (r, bytes) = self.write_checkpoint(&self.config.bookmark, &state, &heads)?;
106 
        Ok((r, state.len(), bytes))
107 
    }
108
109 
    /// Point `bookmark` at an already stored checkpoint (a shallow clone's base).
110 
    pub(crate) fn install_checkpoint(&self, bookmark: &str, r: &ObjectRef) -> Result<()> {
111 
        Checkpoint::decode(&self.store.get_typed(r, ObjectType::Checkpoint)?)?;
112 
        fs::create_dir_all(self.oxy.join("checkpoints"))?;
113 
        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())
114 
    }
115
116 
    /// A bundle holding a checkpoint of exactly `heads` with its payload
117 
    /// (spec §16.1 `checkpoint`); the current bookmark checkpoint is reused
118 
    /// when it matches, otherwise one is made and, when `heads` are the
119 
    /// bookmark's, kept.
120 
    pub fn checkpoint_bundle_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {
121 
        let t_find = crate::timing::phase("checkpoint: find or make");
122 
        let wanted = StateObject::new(heads.to_vec()).heads;
123 
        let bookmark = self.config.bookmark.clone();
124 
        let current = self.bookmark_heads()?;
125 
        let r = match self.latest_checkpoint(&bookmark)? {
126 
            Some((r, cp)) if cp.heads == wanted => r,
127 
            _ => {
128 
                let state = self.load_state(&wanted)?;
129 
                if current == wanted {
130 
                    self.write_checkpoint(&bookmark, &state, &wanted)?.0
131 
                } else {
132 
                    let (blob, _) = self.store_checkpoint_payload(&state)?;
133 
                    let cp = Checkpoint {
134 
                        heads: wanted.clone(),
135 
                        count: state.len() as u64,
136 
                        payload: blob,
137 
                    };
138 
                    self.store.put(ObjectType::Checkpoint, &cp.encode())?
139 
                }
140 
            }
141 
        };
142 
        drop(t_find);
143 
        let _t = crate::timing::phase("checkpoint: gather its objects");
144 
        let cp = Checkpoint::decode(&self.store.get_typed(&r, ObjectType::Checkpoint)?)?;
145 
        let mut bundle = Bundle::default();
146 
        for c in Blob::decode(&self.store.get_typed(&cp.payload, ObjectType::Blob)?)?.chunks {
147 
            bundle.objects.push(self.store.get(&c)?);
148 
        }
149 
        bundle.objects.push(self.store.get(&cp.payload)?);
150 
        // The graph blobs and their chunks in two parallel reads (#137): a
151 
        // big state's are tens of thousands, and one by one they took six
152 
        // minutes for VS Code.
153 
        let graphs = self.checkpoint_graph_blobs(&cp)?;
154 
        let mut blobs = Vec::with_capacity(graphs.len());
155 
        for got in self.store.get_many(&graphs) {
156 
            blobs.push(got?);
157 
        }
158 
        let mut chunks = Vec::new();
159 
        for (ty, body) in &blobs {
160 
            if *ty != ObjectType::Blob {
161 
                return fail("a checkpoint graph is not a blob");
162 
            }
163 
            chunks.extend(Blob::decode(body)?.chunks);
164 
        }
165 
        for got in self.store.get_many(&chunks) {
166 
            bundle.objects.push(got?);
167 
        }
168 
        bundle.objects.extend(blobs);
169 
        bundle.objects.push((ObjectType::Checkpoint, cp.encode()));
170 
        Ok(bundle)
171 
    }
172
173 
    /// Load the state of `heads` using a checkpoint when one applies.
174 
    pub(crate) fn load_state_checkpointed(&self, heads: &[ChangeHash]) -> Result<Option<State>> {
175 
        self.load_state_checkpointed_as(heads, false)
176 
    }
177
178 
    /// As [`Repo::load_state_checkpointed`]; `history` builds no file graph.
179 
    pub(crate) fn load_state_checkpointed_as(
180 
        &self,
181 
        heads: &[ChangeHash],
182 
        history: bool,
183 
    ) -> Result<Option<State>> {
184 
        let mut bookmarks: Vec<String> = vec![self.config.bookmark.clone()];
185 
        if let Ok(rd) = fs::read_dir(self.oxy.join("checkpoints")) {
186 
            for e in rd.flatten() {
187 
                let name = file_name_to_name(&e.file_name().to_string_lossy());
188 
                if !bookmarks.contains(&name) {
189 
                    bookmarks.push(name);
190 
                }
191 
            }
192 
        }
193 
        for b in bookmarks {
194 
            let Some((_, cp)) = self.latest_checkpoint(&b)? else {
195 
                continue;
196 
            };
197 
            let payload = {
198 
                let _t = crate::timing::phase("load: read checkpoint");
199 
                self.read_blob(&cp.payload)?
200 
            };
201 
            let base = {
202 
                let _t = crate::timing::phase("load: decode checkpoint");
203 
                if history {
204 
                    State::from_checkpoint_history(&payload)?
205 
                } else {
206 
                    self.state_from_checkpoint(&payload)?
207 
                }
208 
            };
209 
            let cp_heads: BTreeSet<ChangeHash> = cp.heads.iter().copied().collect();
210 
            let mut found: BTreeSet<ChangeHash> = BTreeSet::new();
211 
            let mut delta: Vec<(ChangeHash, Change, Vec<crate::objects::TextRef>)> = Vec::new();
212 
            let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();
213 
            let mut stack: Vec<ChangeHash> = heads.to_vec();
214 
            while let Some(h) = stack.pop() {
215 
                if !seen.insert(h) {
216 
                    continue;
217 
                }
218 
                if base.contains(&h) {
219 
                    if cp_heads.contains(&h) {
220 
                        found.insert(h);
221 
                    }
222 
                    continue;
223 
                }
224 
                // For history, the skeleton: text is read only for the
225 
                // graphs built later, and a state holding none can be
226 
                // indexed (oxygen#93).
227 
                let (c, texts) = if history {
228 
                    self.store.get_change_lazy(&h)?
229 
                } else {
230 
                    (self.store.get_change(&h)?, Vec::new())
231 
                };
232 
                stack.extend(c.deps.iter().copied());
233 
                delta.push((h, c, texts));
234 
            }
235 
            if found.len() != cp_heads.len() {
236 
                continue;
237 
            }
238 
            let _t = crate::timing::phase("load: apply delta and validate");
239 
            let mut state = base;
240 
            for (h, c, texts) in delta {
241 
                if history {
242 
                    state.add_known(h, c, texts);
243 
                } else {
244 
                    state.add_unchecked(c);
245 
                }
246 
            }
247 
            state.validate()?;
248 
            return Ok(Some(state));
249 
        }
250 
        Ok(None)
251 
    }
252 
}