OXYGENHUB

Coverage / crates/flow-casfs/src/main.rs 64.5%

171 of 265 lines run, 15 of 28 functions · run #203 on main · run missed

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);
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;
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>,
47
    /// Bytes already read, by inode, with the order they came in and their
48
    /// total: the oldest go first beyond `MEMORY_BYTES`.
49
    bytes: HashMap<u64, Arc<Vec<u8>>>,
50
    order: std::collections::VecDeque<u64>,
51
    held: usize,
52
    cache_dir: PathBuf,
53
    uid: u32,
54
    gid: u32,
55
    mounted: SystemTime,
56
}
57
58
impl CasFs {
592×
    fn new(repo: Repo, state: State) -> CasFs {
602×
        let view = state.tree().view();
612×
        let root = Inode { parent: 1, kind: NodeKind::Dir, node: NodeId::ROOT, exec: false, opaque: false, children: Vec::new() };
622×
        let mut inodes = vec![root];
632×
        let mut by_path: HashMap<&str, u64> = HashMap::with_capacity(view.entries.len());
642×
        by_path.insert("", 1);
65
        // In path order, so a directory comes before what it holds.
6610×
        for (path, e) in view.entries.iter() {
6710×
            let (dir, name) = path.rsplit_once('/').unwrap_or(("", path.as_str()));
6810×
            let Some(&parent) = by_path.get(dir) else { continue };
6910×
            let ino = inodes.len() as u64 + 1;
7010×
            inodes.push(Inode { parent, kind: e.kind, node: e.node, exec: e.mode.exec, opaque: e.mode.opaque, children: Vec::new() });
7110×
            inodes[parent as usize - 1].children.push((OsString::from(name), ino));
7210×
            if e.kind == NodeKind::Dir || e.kind == NodeKind::Subrepo {
734×
                by_path.insert(path.as_str(), ino);
746×
            }
75
        }
7612×
        for i in inodes.iter_mut() {
7712×
            i.children.sort();
7812×
        }
792×
        let cache_dir = repo.root().join(".oxy").join("cache").join("render");
802×
        CasFs {
812×
            repo,
822×
            state,
832×
            inodes,
842×
            bytes: HashMap::new(),
852×
            order: std::collections::VecDeque::new(),
862×
            held: 0,
872×
            cache_dir,
882×
            uid: unsafe { libc::getuid() },
892×
            gid: unsafe { libc::getgid() },
902×
            mounted: SystemTime::now(),
912×
        }
922×
    }
93
9416×
    fn inode(&self, ino: u64) -> Option<&Inode> {
9516×
        self.inodes.get(ino.checked_sub(1)? as usize)
9616×
    }
97
98
    /// A file's or symlink's bytes: its blob, or its rendering (cached on
99
    /// disk by graph key).
1005×
    fn content(&mut self, ino: u64) -> Result<Arc<Vec<u8>>> {
1015×
        if let Some(b) = self.bytes.get(&ino) {
1021×
            return Ok(b.clone());
1034×
        }
1044×
        let i = self.inode(ino).context("no such inode")?;
1054×
        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.
1094×
        if kind == NodeKind::File && !opaque && self.state.single_text(&node).is_some() {
1102×
            let siblings: Vec<(u64, oxy_core::objects::TextRef)> = self.inodes[parent as usize - 1]
1112×
                .children
1122×
                .iter()
1132×
                .map(|(_, c)| *c)
1143×
                .filter(|c| !self.bytes.contains_key(c))
1153×
                .filter_map(|c| {
1163×
                    let i = &self.inodes[c as usize - 1];
1173×
                    (i.kind == NodeKind::File && !i.opaque).then_some(())?;
1182×
                    Some((c, self.state.single_text(&i.node)?))
1193×
                })
1202×
                .collect();
1212×
            let texts: Vec<oxy_core::objects::TextRef> = siblings.iter().map(|(_, t)| *t).collect();
1222×
            for ((c, _), b) in siblings.iter().zip(self.repo.single_texts(&texts)?) {
1232×
                self.keep(*c, Arc::new(b));
1242×
            }
1252×
            if let Some(b) = self.bytes.get(&ino) {
1262×
                return Ok(b.clone());
1270×
            }
1282×
        }
1292×
        let bytes = if kind == NodeKind::File && !opaque {
1302×
            self.rendering(&node)?
131
        } else {
1320×
            match self.state.live_blobs(&node).first() {
1330×
                Some(b) => self.repo.read_blob(b)?,
1340×
                None => Vec::new(),
135
            }
136
        };
1372×
        let bytes = Arc::new(bytes);
1382×
        self.keep(ino, bytes.clone());
1392×
        Ok(bytes)
1405×
    }
141
142
    /// Keep a file's bytes, dropping the oldest beyond `MEMORY_BYTES`.
1434×
    fn keep(&mut self, ino: u64, bytes: Arc<Vec<u8>>) {
1444×
        self.held += bytes.len();
1454×
        if let Some(old) = self.bytes.insert(ino, bytes) {
1460×
            self.held -= old.len();
1474×
        } else {
1484×
            self.order.push_back(ino);
1494×
        }
1504×
        while self.held > MEMORY_BYTES && self.order.len() > 1 {
1510×
            let Some(oldest) = self.order.pop_front() else { break };
1520×
            if let Some(b) = self.bytes.remove(&oldest) {
1530×
                self.held -= b.len();
1540×
            }
155
        }
1564×
    }
157
1582×
    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.
1612×
        if let Some(t) = self.state.single_text(node) {
1620×
            return Ok(self.repo.single_texts(&[t])?.pop().unwrap_or_default());
1632×
        }
1642×
        let key = self.state.graph_key(node);
1652×
        let cached = key.map(|k| self.cache_dir.join(hex(&k)));
1662×
        if let Some(p) = &cached {
1672×
            if let Ok(b) = std::fs::read(p) {
1681×
                return Ok(b);
1691×
            }
1700×
        }
1711×
        let graphs = self.repo.build_graphs_detached(&self.state, &[*node])?;
1721×
        let bytes = match graphs.get(node) {
1731×
            Some(g) => self.state.render_graph(g).0,
1740×
            None => Vec::new(),
175
        };
1761×
        if let Some(p) = cached {
177
            // A cache: written whole or not at all, without waiting for
178
            // the disk.
1791×
            let _ = std::fs::create_dir_all(&self.cache_dir);
1801×
            let tmp = p.with_extension(format!("tmp-{}", std::process::id()));
1811×
            if std::fs::write(&tmp, &bytes).is_ok() {
1821×
                let _ = std::fs::rename(&tmp, &p);
1831×
            }
1840×
        }
1851×
        Ok(bytes)
1862×
    }
187
1882×
    fn attr(&mut self, ino: u64) -> Result<FileAttr> {
1892×
        let i = self.inode(ino).context("no such inode")?;
1902×
        let (kind, perm) = match i.kind {
1911×
            NodeKind::Dir | NodeKind::Subrepo => (FileType::Directory, 0o555),
1920×
            NodeKind::Symlink => (FileType::Symlink, 0o777),
1930×
            _ if i.exec => (FileType::RegularFile, 0o555),
1941×
            _ => (FileType::RegularFile, 0o444),
195
        };
1962×
        let nlink = if kind == FileType::Directory { 2 } else { 1 };
1972×
        let size = match kind {
1981×
            FileType::Directory => 0,
1991×
            _ => self.content(ino)?.len() as u64,
200
        };
2012×
        Ok(FileAttr {
2022×
            ino,
2032×
            size,
2042×
            blocks: size.div_ceil(512),
2052×
            atime: self.mounted,
2062×
            mtime: self.mounted,
2072×
            ctime: self.mounted,
2082×
            crtime: self.mounted,
2092×
            kind,
2102×
            perm,
2112×
            nlink,
2122×
            uid: self.uid,
2132×
            gid: self.gid,
2142×
            rdev: 0,
2152×
            blksize: 4096,
2162×
            flags: 0,
2172×
        })
2182×
    }
219
}
220
2212×
fn hex(b: &[u8]) -> String {
22264×
    b.iter().map(|x| format!("{x:02x}")).collect()
2232×
}
224
225
impl Filesystem for CasFs {
2260×
    fn lookup(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) {
2270×
        let found =
2280×
            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));
2290×
        match found.map(|ino| self.attr(ino)) {
2300×
            Some(Ok(a)) => reply.entry(&TTL, &a, 0),
2310×
            Some(Err(_)) => reply.error(libc::EIO),
2320×
            None => reply.error(libc::ENOENT),
233
        }
2340×
    }
235
2360×
    fn getattr(&mut self, _req: &Request<'_>, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {
2370×
        match self.attr(ino) {
2380×
            Ok(a) => reply.attr(&TTL, &a),
2390×
            Err(_) => reply.error(libc::ENOENT),
240
        }
2410×
    }
242
2430×
    fn readlink(&mut self, _req: &Request<'_>, ino: u64, reply: ReplyData) {
2440×
        match self.content(ino) {
2450×
            Ok(b) => reply.data(&b),
2460×
            Err(_) => reply.error(libc::EIO),
247
        }
2480×
    }
249
250
    #[allow(clippy::too_many_arguments)]
2510×
    fn read(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, size: u32, _flags: i32, _lock: Option<u64>, reply: ReplyData) {
2520×
        match self.content(ino) {
2530×
            Ok(b) => {
2540×
                let start = (offset.max(0) as usize).min(b.len());
2550×
                let end = (start + size as usize).min(b.len());
2560×
                reply.data(&b[start..end]);
2570×
            }
2580×
            Err(_) => reply.error(libc::EIO),
259
        }
2600×
    }
261
2620×
    fn readdir(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, mut reply: ReplyDirectory) {
2630×
        let Some(dir) = self.inode(ino) else {
2640×
            reply.error(libc::ENOENT);
2650×
            return;
266
        };
2670×
        let mut entries: Vec<(u64, FileType, OsString)> =
2680×
            vec![(ino, FileType::Directory, ".".into()), (dir.parent, FileType::Directory, "..".into())];
2690×
        for (name, child) in &dir.children {
2700×
            let kind = match self.inodes[*child as usize - 1].kind {
2710×
                NodeKind::Dir | NodeKind::Subrepo => FileType::Directory,
2720×
                NodeKind::Symlink => FileType::Symlink,
2730×
                _ => FileType::RegularFile,
274
            };
2750×
            entries.push((*child, kind, name.clone()));
276
        }
2770×
        for (i, (child, kind, name)) in entries.into_iter().enumerate().skip(offset.max(0) as usize) {
2780×
            if reply.add(child, (i + 1) as i64, kind, name) {
2790×
                break;
2800×
            }
281
        }
2820×
        reply.ok();
2830×
    }
284
}
285
2860×
fn main() -> Result<()> {
2870×
    let args: Vec<String> = std::env::args().collect();
2880×
    let usage = "usage: flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]";
2890×
    if args.len() < 4 || args[1] != "mount" {
2900×
        bail!("{usage}");
2910×
    }
2920×
    let (repo_dir, mountpoint) = (Path::new(&args[2]), Path::new(&args[3]));
2930×
    let bookmark = args.iter().position(|a| a == "--bookmark").and_then(|i| args.get(i + 1)).cloned();
2940×
    let started = Instant::now();
2950×
    let mut repo = Repo::open(repo_dir).with_context(|| format!("open {}", repo_dir.display()))?;
2960×
    if let Some(b) = &bookmark {
2970×
        repo.config.bookmark = b.clone();
2980×
    }
2990×
    let heads = repo.bookmark_heads()?;
3000×
    let state = repo.load_state_indexed(&heads)?;
3010×
    let loaded = started.elapsed();
3020×
    let fs = CasFs::new(repo, state);
3030×
    eprintln!(
304
        "flow-casfs: {} entries; state loaded in {:.1} ms, tree built in {:.1} ms",
3050×
        fs.inodes.len(),
3060×
        loaded.as_secs_f64() * 1000.0,
3070×
        (started.elapsed() - loaded).as_secs_f64() * 1000.0
308
    );
3090×
    let options = [MountOption::RO, MountOption::FSName("flow-casfs".into()), MountOption::DefaultPermissions];
3100×
    let session = fuser::spawn_mount2(fs, mountpoint, &options).context("mount")?;
3110×
    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.
3130×
    session.guard.join().ok();
3140×
    Ok(())
3150×
}
316
317
#[cfg(test)]
318
mod tests {
319
    use super::*;
320
321
    /// The inode of a path, walking from the root.
3226×
    fn ino_of(fs: &CasFs, path: &str) -> Option<u64> {
3236×
        let mut ino = 1u64;
32410×
        for part in path.split('/') {
32510×
            let dir = fs.inode(ino)?;
32619×
            let i = dir.children.binary_search_by(|(n, _)| n.as_os_str().cmp(OsStr::new(part))).ok()?;
3279×
            ino = dir.children[i].1;
328
        }
3295×
        Some(ino)
3306×
    }
331
332
    #[test]
3331×
    fn a_mounted_state_holds_the_committed_files() {
3341×
        let dir = tempfile::tempdir().unwrap();
3351×
        std::env::set_var("OXY_HOME", dir.path().join("home"));
3361×
        let root = dir.path().join("repo");
3371×
        let repo = Repo::init(&root).unwrap();
3383000×
        let big: String = (0..3000).map(|i| format!("line {i}\n")).collect();
3391×
        let files = [("a.txt", "one\ntwo\n".to_string()), ("src/lib.rs", "pub fn f() {}\n".to_string()), ("src/deep/big.txt", big)];
3403×
        for (p, c) in &files {
3413×
            let full = root.join(p);
3423×
            std::fs::create_dir_all(full.parent().unwrap()).unwrap();
3433×
            std::fs::write(full, c).unwrap();
3443×
        }
3451×
        repo.commit("first", &[]).unwrap();
346
        // An edit, so one file's rendering is a history, not one text.
3471×
        std::fs::write(root.join("a.txt"), "one\n2\ntwo\n").unwrap();
3481×
        repo.commit("second", &[]).unwrap();
3491×
        let heads = repo.bookmark_heads().unwrap();
3501×
        let state = repo.load_state_indexed(&heads).unwrap();
3511×
        let mut fs = CasFs::new(Repo::open(&root).unwrap(), state);
3522×
        for (p, c) in [("a.txt", "one\n2\ntwo\n"), ("src/lib.rs", "pub fn f() {}\n")] {
3532×
            let ino = ino_of(&fs, p).unwrap();
3542×
            assert_eq!(fs.content(ino).unwrap().as_slice(), c.as_bytes(), "{p}");
355
        }
3561×
        let big = ino_of(&fs, "src/deep/big.txt").unwrap();
3571×
        assert_eq!(fs.content(big).unwrap().as_slice(), files[2].1.as_bytes());
3581×
        assert_eq!(fs.attr(big).unwrap().size, files[2].1.len() as u64);
3591×
        assert_eq!(fs.attr(ino_of(&fs, "src/deep").unwrap()).unwrap().kind, FileType::Directory);
3601×
        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.
3641×
        let cached = std::fs::read_dir(&fs.cache_dir).unwrap().count();
3651×
        assert_eq!(cached, 1, "renderings cached");
3661×
        let state = repo.load_state_indexed(&heads).unwrap();
3671×
        let mut again = CasFs::new(Repo::open(&root).unwrap(), state);
3681×
        let ino = ino_of(&again, "a.txt").unwrap();
3691×
        assert_eq!(again.content(ino).unwrap().as_slice(), b"one\n2\ntwo\n");
3701×
    }
371
}