OXYGENHUB
◆ v0.5.3 oxygenflow / crates / flow-casfs / src / main.rs
14.7 KiB · Rust · last changed flow-casfs: at most 256 MB of files held in memory, oldest dropped first (refs #21) 1 month ago source blame history raw
f849c788 Andy McCall ↶ 1
//! flow-casfs: an Oxygen state mounted read-only (OxygenFlow spec 5.1).
2
//!
3
//! The tree comes from the repository's state index; an opaque file's bytes
4
//! are its stored blob; a text file's bytes are a rendering of its history,
5
//! made on first read or stat and cached on disk by graph key (equal keys
6
//! mean equal renderings), so every state in which a file has the same
7
//! history finds it at once.
8
//!
9
//!   flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]
10
//!
11
//! It mounts in the foreground and prints how long the mount took; unmount
12
//! with `fusermount3 -u MOUNTPOINT`.
13
14
use anyhow::{bail, Context, Result};
15
use fuser::{FileAttr, FileType, Filesystem, MountOption, ReplyAttr, ReplyData, ReplyDirectory, ReplyEntry, Request};
16
use oxy_core::ids::NodeId;
17
use oxy_core::objects::NodeKind;
18
use oxy_core::repo::Repo;
19
use oxy_core::state::State;
20
use std::collections::HashMap;
21
use std::ffi::{OsStr, OsString};
22
use std::path::{Path, PathBuf};
23
use std::sync::Arc;
24
use std::time::{Duration, Instant, SystemTime};
25
26
/// How long the kernel may keep attributes and names: the mount never
27
/// changes.
28
const TTL: Duration = Duration::from_secs(3600);
602ac9c8 Andy McCall ↶ 29
/// Bytes of files kept in memory; the kernel's page cache keeps the hot
30
/// ones too, and a file dropped here is cheap to read again.
31
const MEMORY_BYTES: usize = 256 << 20;
f849c788 Andy McCall ↶ 32
33
struct Inode {
34
    parent: u64,
35
    kind: NodeKind,
36
    node: NodeId,
37
    exec: bool,
38
    opaque: bool,
39
    /// For a directory: its children by name.
40
    children: Vec<(OsString, u64)>,
41
}
42
43
struct CasFs {
44
    repo: Repo,
45
    state: State,
46
    inodes: Vec<Inode>,
602ac9c8 Andy McCall ↶ 47
    /// Bytes already read, by inode, with the order they came in and their
48
    /// total: the oldest go first beyond `MEMORY_BYTES`.
f849c788 Andy McCall ↶ 49
    bytes: HashMap<u64, Arc<Vec<u8>>>,
602ac9c8 Andy McCall ↶ 50
    order: std::collections::VecDeque<u64>,
51
    held: usize,
f849c788 Andy McCall ↶ 52
    cache_dir: PathBuf,
53
    uid: u32,
54
    gid: u32,
55
    mounted: SystemTime,
56
}
57
58
impl CasFs {
59
    fn new(repo: Repo, state: State) -> CasFs {
60
        let view = state.tree().view();
61
        let root = Inode { parent: 1, kind: NodeKind::Dir, node: NodeId::ROOT, exec: false, opaque: false, children: Vec::new() };
62
        let mut inodes = vec![root];
63
        let mut by_path: HashMap<&str, u64> = HashMap::with_capacity(view.entries.len());
64
        by_path.insert("", 1);
65
        // In path order, so a directory comes before what it holds.
66
        for (path, e) in view.entries.iter() {
67
            let (dir, name) = path.rsplit_once('/').unwrap_or(("", path.as_str()));
68
            let Some(&parent) = by_path.get(dir) else { continue };
69
            let ino = inodes.len() as u64 + 1;
70
            inodes.push(Inode { parent, kind: e.kind, node: e.node, exec: e.mode.exec, opaque: e.mode.opaque, children: Vec::new() });
71
            inodes[parent as usize - 1].children.push((OsString::from(name), ino));
72
            if e.kind == NodeKind::Dir || e.kind == NodeKind::Subrepo {
73
                by_path.insert(path.as_str(), ino);
74
            }
75
        }
76
        for i in inodes.iter_mut() {
77
            i.children.sort();
78
        }
79
        let cache_dir = repo.root().join(".oxy").join("cache").join("render");
80
        CasFs {
81
            repo,
82
            state,
83
            inodes,
84
            bytes: HashMap::new(),
602ac9c8 Andy McCall ↶ 85
            order: std::collections::VecDeque::new(),
86
            held: 0,
f849c788 Andy McCall ↶ 87
            cache_dir,
88
            uid: unsafe { libc::getuid() },
89
            gid: unsafe { libc::getgid() },
90
            mounted: SystemTime::now(),
91
        }
92
    }
93
94
    fn inode(&self, ino: u64) -> Option<&Inode> {
95
        self.inodes.get(ino.checked_sub(1)? as usize)
96
    }
97
98
    /// A file's or symlink's bytes: its blob, or its rendering (cached on
99
    /// disk by graph key).
100
    fn content(&mut self, ino: u64) -> Result<Arc<Vec<u8>>> {
101
        if let Some(b) = self.bytes.get(&ino) {
102
            return Ok(b.clone());
103
        }
104
        let i = self.inode(ino).context("no such inode")?;
105
        let (node, kind, opaque, parent) = (i.node, i.kind, i.opaque, i.parent);
106
        // A file added once and never edited: read it with every such file
107
        // of its directory not read yet, in one batch (a directory is
108
        // usually read whole), on every core.
109
        if kind == NodeKind::File && !opaque && self.state.single_text(&node).is_some() {
110
            let siblings: Vec<(u64, oxy_core::objects::TextRef)> = self.inodes[parent as usize - 1]
111
                .children
112
                .iter()
113
                .map(|(_, c)| *c)
114
                .filter(|c| !self.bytes.contains_key(c))
115
                .filter_map(|c| {
116
                    let i = &self.inodes[c as usize - 1];
117
                    (i.kind == NodeKind::File && !i.opaque).then_some(())?;
118
                    Some((c, self.state.single_text(&i.node)?))
119
                })
120
                .collect();
121
            let texts: Vec<oxy_core::objects::TextRef> = siblings.iter().map(|(_, t)| *t).collect();
122
            for ((c, _), b) in siblings.iter().zip(self.repo.single_texts(&texts)?) {
602ac9c8 Andy McCall ↶ 123
                self.keep(*c, Arc::new(b));
f849c788 Andy McCall ↶ 124
            }
125
            if let Some(b) = self.bytes.get(&ino) {
126
                return Ok(b.clone());
127
            }
128
        }
129
        let bytes = if kind == NodeKind::File && !opaque {
130
            self.rendering(&node)?
131
        } else {
132
            match self.state.live_blobs(&node).first() {
133
                Some(b) => self.repo.read_blob(b)?,
134
                None => Vec::new(),
135
            }
136
        };
137
        let bytes = Arc::new(bytes);
602ac9c8 Andy McCall ↶ 138
        self.keep(ino, bytes.clone());
f849c788 Andy McCall ↶ 139
        Ok(bytes)
140
    }
141
602ac9c8 Andy McCall ↶ 142
    /// Keep a file's bytes, dropping the oldest beyond `MEMORY_BYTES`.
143
    fn keep(&mut self, ino: u64, bytes: Arc<Vec<u8>>) {
144
        self.held += bytes.len();
145
        if let Some(old) = self.bytes.insert(ino, bytes) {
146
            self.held -= old.len();
147
        } else {
148
            self.order.push_back(ino);
149
        }
150
        while self.held > MEMORY_BYTES && self.order.len() > 1 {
151
            let Some(oldest) = self.order.pop_front() else { break };
152
            if let Some(b) = self.bytes.remove(&oldest) {
153
                self.held -= b.len();
154
            }
155
        }
156
    }
157
f849c788 Andy McCall ↶ 158
    fn rendering(&self, node: &NodeId) -> Result<Vec<u8>> {
159
        // A file added once and never edited is its own text: no graph, and
160
        // nothing to cache.
161
        if let Some(t) = self.state.single_text(node) {
162
            return Ok(self.repo.single_texts(&[t])?.pop().unwrap_or_default());
163
        }
164
        let key = self.state.graph_key(node);
165
        let cached = key.map(|k| self.cache_dir.join(hex(&k)));
166
        if let Some(p) = &cached {
167
            if let Ok(b) = std::fs::read(p) {
168
                return Ok(b);
169
            }
170
        }
171
        let graphs = self.repo.build_graphs_detached(&self.state, &[*node])?;
172
        let bytes = match graphs.get(node) {
173
            Some(g) => self.state.render_graph(g).0,
174
            None => Vec::new(),
175
        };
176
        if let Some(p) = cached {
177
            // A cache: written whole or not at all, without waiting for
178
            // the disk.
179
            let _ = std::fs::create_dir_all(&self.cache_dir);
180
            let tmp = p.with_extension(format!("tmp-{}", std::process::id()));
181
            if std::fs::write(&tmp, &bytes).is_ok() {
182
                let _ = std::fs::rename(&tmp, &p);
183
            }
184
        }
185
        Ok(bytes)
186
    }
187
188
    fn attr(&mut self, ino: u64) -> Result<FileAttr> {
189
        let i = self.inode(ino).context("no such inode")?;
190
        let (kind, perm) = match i.kind {
191
            NodeKind::Dir | NodeKind::Subrepo => (FileType::Directory, 0o555),
192
            NodeKind::Symlink => (FileType::Symlink, 0o777),
193
            _ if i.exec => (FileType::RegularFile, 0o555),
194
            _ => (FileType::RegularFile, 0o444),
195
        };
196
        let nlink = if kind == FileType::Directory { 2 } else { 1 };
197
        let size = match kind {
198
            FileType::Directory => 0,
199
            _ => self.content(ino)?.len() as u64,
200
        };
201
        Ok(FileAttr {
202
            ino,
203
            size,
204
            blocks: size.div_ceil(512),
205
            atime: self.mounted,
206
            mtime: self.mounted,
207
            ctime: self.mounted,
208
            crtime: self.mounted,
209
            kind,
210
            perm,
211
            nlink,
212
            uid: self.uid,
213
            gid: self.gid,
214
            rdev: 0,
215
            blksize: 4096,
216
            flags: 0,
217
        })
218
    }
219
}
220
221
fn hex(b: &[u8]) -> String {
222
    b.iter().map(|x| format!("{x:02x}")).collect()
223
}
224
225
impl Filesystem for CasFs {
226
    fn lookup(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) {
227
        let found =
228
            self.inode(parent).and_then(|p| p.children.binary_search_by(|(n, _)| n.as_os_str().cmp(name)).ok().map(|i| p.children[i].1));
229
        match found.map(|ino| self.attr(ino)) {
230
            Some(Ok(a)) => reply.entry(&TTL, &a, 0),
231
            Some(Err(_)) => reply.error(libc::EIO),
232
            None => reply.error(libc::ENOENT),
233
        }
234
    }
235
236
    fn getattr(&mut self, _req: &Request<'_>, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {
237
        match self.attr(ino) {
238
            Ok(a) => reply.attr(&TTL, &a),
239
            Err(_) => reply.error(libc::ENOENT),
240
        }
241
    }
242
243
    fn readlink(&mut self, _req: &Request<'_>, ino: u64, reply: ReplyData) {
244
        match self.content(ino) {
245
            Ok(b) => reply.data(&b),
246
            Err(_) => reply.error(libc::EIO),
247
        }
248
    }
249
250
    #[allow(clippy::too_many_arguments)]
251
    fn read(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, size: u32, _flags: i32, _lock: Option<u64>, reply: ReplyData) {
252
        match self.content(ino) {
253
            Ok(b) => {
254
                let start = (offset.max(0) as usize).min(b.len());
255
                let end = (start + size as usize).min(b.len());
256
                reply.data(&b[start..end]);
257
            }
258
            Err(_) => reply.error(libc::EIO),
259
        }
260
    }
261
262
    fn readdir(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, mut reply: ReplyDirectory) {
263
        let Some(dir) = self.inode(ino) else {
264
            reply.error(libc::ENOENT);
265
            return;
266
        };
267
        let mut entries: Vec<(u64, FileType, OsString)> =
268
            vec![(ino, FileType::Directory, ".".into()), (dir.parent, FileType::Directory, "..".into())];
269
        for (name, child) in &dir.children {
270
            let kind = match self.inodes[*child as usize - 1].kind {
271
                NodeKind::Dir | NodeKind::Subrepo => FileType::Directory,
272
                NodeKind::Symlink => FileType::Symlink,
273
                _ => FileType::RegularFile,
274
            };
275
            entries.push((*child, kind, name.clone()));
276
        }
277
        for (i, (child, kind, name)) in entries.into_iter().enumerate().skip(offset.max(0) as usize) {
278
            if reply.add(child, (i + 1) as i64, kind, name) {
279
                break;
280
            }
281
        }
282
        reply.ok();
283
    }
284
}
285
286
fn main() -> Result<()> {
287
    let args: Vec<String> = std::env::args().collect();
288
    let usage = "usage: flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]";
289
    if args.len() < 4 || args[1] != "mount" {
290
        bail!("{usage}");
291
    }
292
    let (repo_dir, mountpoint) = (Path::new(&args[2]), Path::new(&args[3]));
293
    let bookmark = args.iter().position(|a| a == "--bookmark").and_then(|i| args.get(i + 1)).cloned();
294
    let started = Instant::now();
295
    let mut repo = Repo::open(repo_dir).with_context(|| format!("open {}", repo_dir.display()))?;
296
    if let Some(b) = &bookmark {
297
        repo.config.bookmark = b.clone();
298
    }
299
    let heads = repo.bookmark_heads()?;
300
    let state = repo.load_state_indexed(&heads)?;
301
    let loaded = started.elapsed();
302
    let fs = CasFs::new(repo, state);
303
    eprintln!(
304
        "flow-casfs: {} entries; state loaded in {:.1} ms, tree built in {:.1} ms",
305
        fs.inodes.len(),
306
        loaded.as_secs_f64() * 1000.0,
307
        (started.elapsed() - loaded).as_secs_f64() * 1000.0
308
    );
309
    let options = [MountOption::RO, MountOption::FSName("flow-casfs".into()), MountOption::DefaultPermissions];
310
    let session = fuser::spawn_mount2(fs, mountpoint, &options).context("mount")?;
311
    eprintln!("flow-casfs: mounted at {} in {:.1} ms", mountpoint.display(), started.elapsed().as_secs_f64() * 1000.0);
312
    // Until the mount goes away (fusermount3 -u) or the process is stopped.
313
    session.guard.join().ok();
314
    Ok(())
315
}
316
317
#[cfg(test)]
318
mod tests {
319
    use super::*;
320
321
    /// The inode of a path, walking from the root.
322
    fn ino_of(fs: &CasFs, path: &str) -> Option<u64> {
323
        let mut ino = 1u64;
324
        for part in path.split('/') {
325
            let dir = fs.inode(ino)?;
326
            let i = dir.children.binary_search_by(|(n, _)| n.as_os_str().cmp(OsStr::new(part))).ok()?;
327
            ino = dir.children[i].1;
328
        }
329
        Some(ino)
330
    }
331
332
    #[test]
333
    fn a_mounted_state_holds_the_committed_files() {
334
        let dir = tempfile::tempdir().unwrap();
335
        std::env::set_var("OXY_HOME", dir.path().join("home"));
336
        let root = dir.path().join("repo");
337
        let repo = Repo::init(&root).unwrap();
338
        let big: String = (0..3000).map(|i| format!("line {i}\n")).collect();
339
        let files = [("a.txt", "one\ntwo\n".to_string()), ("src/lib.rs", "pub fn f() {}\n".to_string()), ("src/deep/big.txt", big)];
340
        for (p, c) in &files {
341
            let full = root.join(p);
342
            std::fs::create_dir_all(full.parent().unwrap()).unwrap();
343
            std::fs::write(full, c).unwrap();
344
        }
345
        repo.commit("first", &[]).unwrap();
346
        // An edit, so one file's rendering is a history, not one text.
347
        std::fs::write(root.join("a.txt"), "one\n2\ntwo\n").unwrap();
348
        repo.commit("second", &[]).unwrap();
349
        let heads = repo.bookmark_heads().unwrap();
350
        let state = repo.load_state_indexed(&heads).unwrap();
351
        let mut fs = CasFs::new(Repo::open(&root).unwrap(), state);
352
        for (p, c) in [("a.txt", "one\n2\ntwo\n"), ("src/lib.rs", "pub fn f() {}\n")] {
353
            let ino = ino_of(&fs, p).unwrap();
354
            assert_eq!(fs.content(ino).unwrap().as_slice(), c.as_bytes(), "{p}");
355
        }
356
        let big = ino_of(&fs, "src/deep/big.txt").unwrap();
357
        assert_eq!(fs.content(big).unwrap().as_slice(), files[2].1.as_bytes());
358
        assert_eq!(fs.attr(big).unwrap().size, files[2].1.len() as u64);
359
        assert_eq!(fs.attr(ino_of(&fs, "src/deep").unwrap()).unwrap().kind, FileType::Directory);
360
        assert!(ino_of(&fs, "missing.txt").is_none());
361
        // Only the edited file needed rendering, and its rendering is cached
362
        // by graph key, so a new mount reads it back; the others are their
363
        // own text blobs.
364
        let cached = std::fs::read_dir(&fs.cache_dir).unwrap().count();
365
        assert_eq!(cached, 1, "renderings cached");
366
        let state = repo.load_state_indexed(&heads).unwrap();
367
        let mut again = CasFs::new(Repo::open(&root).unwrap(), state);
368
        let ino = ino_of(&again, "a.txt").unwrap();
369
        assert_eq!(again.content(ino).unwrap().as_slice(), b"one\n2\ntwo\n");
370
    }
371
}