start_command/
store_decode_cache.rs

1//! Decoded execution store, shared within one process and between processes
2//! (issue #243).
3//!
4//! Decoding `executions.lino` costs about 1-4 ms per record, and one
5//! `--status` used to decode it up to three times (`get`, then `get_all`
6//! twice for the execution chain). The file is still read on every call, so a
7//! write by any process is seen at once, but it is decoded again only when its
8//! content differs from the last decoded copy. The content itself is the key:
9//! unlike an mtime, it cannot miss two same-size writes within one clock tick.
10//!
11//! A new process (every `$ --status` poll) still had to decode once. The
12//! decoded records are therefore also kept as JSON next to the store, tagged
13//! with the SHA-256 of the content they were decoded from; parsing that copy
14//! is far faster than the lino decode. A copy whose digest does not match the
15//! current content is ignored and replaced.
16
17use crate::atomic_write::atomic_write;
18use crate::execution_store::ExecutionRecord;
19use serde_json::{json, Value};
20use sha2::{Digest, Sha256};
21use std::collections::HashMap;
22use std::fs;
23use std::path::{Path, PathBuf};
24use std::sync::{Mutex, OnceLock};
25
26/// Format tag of the JSON copy; a different tag is ignored.
27pub const DECODED_COPY_FORMAT: &str = "start-command-decoded-store-rust-1";
28
29/// Where the records of a read came from.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum DecodeSource {
32    /// This process already decoded the same content.
33    Memory,
34    /// The JSON copy next to the store matched the content.
35    Copy,
36    /// The content was decoded.
37    Decode,
38}
39
40struct DecodedStore {
41    content: String,
42    records: Vec<ExecutionRecord>,
43    decodes: usize,
44}
45
46fn decoded_stores() -> &'static Mutex<HashMap<PathBuf, DecodedStore>> {
47    static STORES: OnceLock<Mutex<HashMap<PathBuf, DecodedStore>>> = OnceLock::new();
48    STORES.get_or_init(|| Mutex::new(HashMap::new()))
49}
50
51fn digest_of(content: &str) -> String {
52    Sha256::digest(content.as_bytes())
53        .iter()
54        .map(|byte| format!("{:02x}", byte))
55        .collect()
56}
57
58/// Values of the JSON copy at `copy_path`, when it was decoded from content
59/// with this digest.
60fn read_decoded_copy(copy_path: &Path, digest: &str) -> Option<Vec<Value>> {
61    let copy: Value = serde_json::from_str(&fs::read_to_string(copy_path).ok()?).ok()?;
62    if copy.get("format")?.as_str()? != DECODED_COPY_FORMAT
63        || copy.get("sha256")?.as_str()? != digest
64    {
65        return None;
66    }
67    copy.get("records")?.as_array().cloned()
68}
69
70/// Keep `values` for the next process. Best effort: a read-only folder only
71/// costs the next decode.
72fn write_decoded_copy(copy_path: &Path, digest: &str, values: &[Value]) {
73    let copy = json!({
74        "format": DECODED_COPY_FORMAT,
75        "sha256": digest,
76        "records": values,
77    });
78    if let Ok(text) = serde_json::to_string(&copy) {
79        let _ = atomic_write(copy_path, text.as_bytes());
80    }
81}
82
83fn to_records(values: &[Value]) -> Vec<ExecutionRecord> {
84    values
85        .iter()
86        .filter_map(ExecutionRecord::from_json)
87        .collect()
88}
89
90/// Decode `content` of the store at `path` into JSON records, or reuse an
91/// earlier decode of the same content: this process's own, or the JSON copy
92/// at `copy_path`. Every call returns its own copy, so callers may change the
93/// records they get.
94pub fn decode_store_content<E>(
95    path: &Path,
96    content: &str,
97    decode: impl FnOnce(&str) -> Result<Vec<Value>, E>,
98    copy_path: Option<&Path>,
99) -> Result<(Vec<ExecutionRecord>, DecodeSource), E> {
100    let mut stores = decoded_stores()
101        .lock()
102        .unwrap_or_else(|poisoned| poisoned.into_inner());
103    if let Some(cached) = stores.get(path).filter(|cached| cached.content == content) {
104        return Ok((cached.records.clone(), DecodeSource::Memory));
105    }
106    let decodes = stores.get(path).map_or(0, |cached| cached.decodes);
107    let digest = copy_path.map(|_| digest_of(content));
108    let copied = copy_path
109        .zip(digest.as_deref())
110        .and_then(|(copy_path, digest)| read_decoded_copy(copy_path, digest));
111    let (records, source, decodes) = match copied {
112        Some(values) => (to_records(&values), DecodeSource::Copy, decodes),
113        None => {
114            let values = decode(content)?;
115            if let Some((copy_path, digest)) = copy_path.zip(digest.as_deref()) {
116                write_decoded_copy(copy_path, digest, &values);
117            }
118            (to_records(&values), DecodeSource::Decode, decodes + 1)
119        }
120    };
121    stores.insert(
122        path.to_path_buf(),
123        DecodedStore {
124            content: content.to_string(),
125            records: records.clone(),
126            decodes,
127        },
128    );
129    Ok((records, source))
130}
131
132/// How many times this process decoded the store at `path` since it was last
133/// forgotten.
134pub fn decode_count(path: &Path) -> usize {
135    decoded_stores()
136        .lock()
137        .unwrap_or_else(|poisoned| poisoned.into_inner())
138        .get(path)
139        .map_or(0, |cached| cached.decodes)
140}
141
142/// Drop the decoded copy of a store held by this process; all stores when
143/// `path` is `None`.
144pub fn forget_decoded_store(path: Option<&Path>) {
145    let mut stores = decoded_stores()
146        .lock()
147        .unwrap_or_else(|poisoned| poisoned.into_inner());
148    match path {
149        Some(path) => {
150            stores.remove(path);
151        }
152        None => stores.clear(),
153    }
154}
start_command/store_decode_cache.rs

start_command/
store_decode_cache.rs
