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