This repository has no description
20 kB
611 lines
1//! The lower layer: all store paths of OUT/paths.json, read-only. Metadata
2//! comes from the caches' `.ls` listings, all fetched up front and kept in
3//! memory; the first open() of a file in a store path fetches, verifies
4//! (NarHash/NarSize) and unpacks that path's NAR. Opens of a path that is
5//! being fetched wait for that fetch.
6
7use std::collections::{BTreeMap, HashMap};
8use std::ffi::{CStr, CString};
9use std::fs::File;
10use std::io;
11use std::mem::ManuallyDrop;
12use std::os::fd::{AsRawFd, FromRawFd};
13use std::path::{Path, PathBuf};
14use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
15use std::sync::{Arc, Mutex, RwLock};
16use std::time::{Duration, Instant};
17
18use fuse_backend_rs::abi::fuse_abi::stat64;
19use fuse_backend_rs::api::filesystem::{
20 Context, DirEntry, Entry, FileSystem, GetxattrReply, ListxattrReply, OpenOptions,
21 ZeroCopyWriter,
22};
23
24use crate::closure::{par_map, PathInfo};
25use crate::http::{self, Http};
26use crate::nar::{self, CountingReader, Tree};
27use crate::stats::Stats;
28use crate::Backing;
29
30/// The store paths never change: let the kernel cache everything.
31pub const TTL: Duration = Duration::from_secs(86400);
32
33/// Set on the inode numbers (st_ino) of this layer, so they don't collide
34/// with the upper layer's.
35pub const INO_TAG: u64 = 1 << 62;
36
37enum Kind {
38 /// children sorted by name
39 Dir(Vec<(Box<str>, u32)>),
40 File {
41 size: u64,
42 exec: bool,
43 },
44 Symlink(Box<[u8]>),
45}
46
47struct Node {
48 kind: Kind,
49 /// index into `LazyFs::names`
50 path: u32,
51 /// path below the store path (empty for its root)
52 rel: Box<str>,
53}
54
55struct Handle {
56 file: File,
57 ino: u64,
58}
59
60pub struct LazyFs {
61 names: Vec<String>,
62 infos: Vec<PathInfo>,
63 /// inode `i` is `nodes[i - 1]`; inode 1 is the store dir
64 nodes: Vec<Node>,
65 cache: PathBuf,
66 http: Http,
67 pub stats: Arc<Stats>,
68 /// per store path: where it is unpacked, once fetched
69 fetched: Vec<Mutex<Option<PathBuf>>>,
70 handles: RwLock<HashMap<u64, Arc<Handle>>>,
71 next_handle: AtomicU64,
72 backing: Option<Arc<Backing>>,
73 passthrough: AtomicBool,
74 /// backing id and open count per inode: the kernel wants all
75 /// passthrough opens of an inode to use the same backing file
76 backing_ids: Mutex<HashMap<u64, (u32, usize)>>,
77 backing_failed: AtomicBool,
78}
79
80fn err(e: i32) -> io::Error {
81 io::Error::from_raw_os_error(e)
82}
83
84impl LazyFs {
85 pub fn new(outdir: &Path, backing: Option<Arc<Backing>>) -> io::Result<LazyFs> {
86 let paths: BTreeMap<String, PathInfo> =
87 serde_json::from_slice(&std::fs::read(outdir.join("paths.json"))?)?;
88 let (names, infos): (Vec<String>, Vec<PathInfo>) = paths.into_iter().unzip();
89 let cache = outdir.join("cache");
90 std::fs::create_dir_all(&cache)?;
91 let http = Http::new(infos.iter().map(|i| i.cache.as_str()));
92 let mut fs = LazyFs {
93 fetched: names.iter().map(|_| Mutex::new(None)).collect(),
94 names,
95 infos,
96 nodes: Vec::new(),
97 cache,
98 http,
99 stats: Arc::new(Stats::new(Some(outdir.join("stats.json")))),
100 handles: RwLock::new(HashMap::new()),
101 next_handle: AtomicU64::new(1),
102 backing,
103 passthrough: AtomicBool::new(false),
104 backing_ids: Mutex::new(HashMap::new()),
105 backing_failed: AtomicBool::new(false),
106 };
107 fs.load_listings()?;
108 fs.stats.save();
109 Ok(fs)
110 }
111
112 /// Fetch the `.ls` listing of every store path (falling back to the NAR
113 /// for paths without one) and build the inode table from them. Fails if
114 /// any path is left without a listing: all of them get registered as
115 /// valid in nix's database.
116 fn load_listings(&mut self) -> io::Result<()> {
117 let t = Instant::now();
118 let idx: Vec<usize> = (0..self.names.len()).collect();
119 let done = AtomicUsize::new(0);
120 let trees = par_map(&idx, 64, |&i| {
121 let r = self.listing(i);
122 let n = done.fetch_add(1, Ordering::Relaxed) + 1;
123 if n.is_multiple_of(500) {
124 eprintln!("listings: {n}/{}", idx.len());
125 }
126 r
127 });
128 let mut root = Vec::with_capacity(trees.len());
129 self.nodes.push(Node {
130 kind: Kind::Dir(Vec::new()),
131 path: u32::MAX,
132 rel: "".into(),
133 });
134 let mut failed = 0;
135 for (i, t) in trees.into_iter().enumerate() {
136 match t {
137 Ok(t) => {
138 let ino = self.add(i as u32, String::new(), t);
139 root.push((self.names[i].clone().into_boxed_str(), ino));
140 }
141 Err(e) => {
142 eprintln!("no listing for {}: {e}", self.names[i]);
143 failed += 1;
144 }
145 }
146 }
147 if failed > 0 {
148 return Err(io::Error::other(format!(
149 "{failed} of {} store paths without a listing",
150 self.names.len()
151 )));
152 }
153 root.sort();
154 self.nodes[0].kind = Kind::Dir(root);
155 eprintln!(
156 "listings: {} store paths, {} inodes in {:.1} s",
157 self.names.len(),
158 self.nodes.len(),
159 t.elapsed().as_secs_f64(),
160 );
161 Ok(())
162 }
163
164 fn listing(&self, i: usize) -> io::Result<Tree> {
165 let info = &self.infos[i];
166 let h = self.names[i].split('-').next().unwrap_or("");
167 match self.http.get(&format!("{}/{h}.ls", info.cache)) {
168 Ok((body, len)) => {
169 self.stats.add("ls", len as u64);
170 Tree::from_listing(&body)
171 }
172 // no listing: fall back to the NAR
173 Err(http::Error::NotFound) => Tree::from_path(&self.fetch(i)?),
174 Err(e) => Err(e.into()),
175 }
176 }
177
178 /// Add the nodes of `t` (at `rel` in store path `path`), return its inode.
179 fn add(&mut self, path: u32, rel: String, t: Tree) -> u32 {
180 let ino = self.nodes.len() as u32 + 1;
181 let (kind, entries) = match t {
182 Tree::File { size, exec } => (Kind::File { size, exec }, None),
183 Tree::Symlink(s) => (Kind::Symlink(s.into_bytes().into()), None),
184 Tree::Dir(e) => (Kind::Dir(Vec::new()), Some(e)),
185 };
186 self.nodes.push(Node {
187 kind,
188 path,
189 rel: rel.clone().into(),
190 });
191 if let Some(entries) = entries {
192 let mut children = Vec::with_capacity(entries.len());
193 for (name, c) in entries {
194 let r = if rel.is_empty() {
195 name.clone()
196 } else {
197 format!("{rel}/{name}")
198 };
199 children.push((name.into_boxed_str(), self.add(path, r, c)));
200 }
201 // BTreeMap order is sorted already
202 self.nodes[ino as usize - 1].kind = Kind::Dir(children);
203 }
204 ino
205 }
206
207 /// The listing of `ino` as built from the `.ls` (for `fetch`).
208 fn tree_of(&self, ino: u64) -> Tree {
209 match &self.nodes[ino as usize - 1].kind {
210 Kind::Dir(c) => Tree::Dir(
211 c.iter()
212 .map(|(n, i)| (n.to_string(), self.tree_of(*i as u64)))
213 .collect(),
214 ),
215 Kind::File { size, exec } => Tree::File {
216 size: *size,
217 exec: *exec,
218 },
219 Kind::Symlink(t) => Tree::Symlink(String::from_utf8_lossy(t).into_owned()),
220 }
221 }
222
223 /// Fetch store path `name` as an open() would and check that what got
224 /// unpacked matches its listing (`lazy-store fetch`, no FUSE needed).
225 pub fn fetch_and_check(&self, name: &str) -> io::Result<PathBuf> {
226 let c = CString::new(name).map_err(io::Error::other)?;
227 let ino = self.lookup_child(1, &c).ok_or_else(|| err(libc::ENOENT))?;
228 let p = self.fetch(self.nodes[ino as usize - 1].path as usize)?;
229 if Tree::from_path(&p)? != self.tree_of(ino) {
230 return Err(io::Error::other(format!(
231 "{name}: unpacked tree differs from its listing"
232 )));
233 }
234 Ok(p)
235 }
236
237 pub fn set_passthrough(&self, on: bool) {
238 self.passthrough
239 .store(on && self.backing.is_some(), Ordering::Relaxed);
240 }
241
242 fn node(&self, ino: u64) -> io::Result<&Node> {
243 (ino as usize)
244 .checked_sub(1)
245 .and_then(|i| self.nodes.get(i))
246 .ok_or_else(|| err(libc::ENOENT))
247 }
248
249 pub fn has_root_entry(&self, name: &CStr) -> bool {
250 self.lookup_child(1, name).is_some()
251 }
252
253 pub fn root_entries(&self) -> impl Iterator<Item = (&str, u64, u32)> {
254 let Kind::Dir(c) = &self.nodes[0].kind else {
255 unreachable!()
256 };
257 c.iter()
258 .map(|(n, i)| (&**n, *i as u64, self.dtype(*i as u64)))
259 }
260
261 fn lookup_child(&self, parent: u64, name: &CStr) -> Option<u64> {
262 let Kind::Dir(c) = &self.node(parent).ok()?.kind else {
263 return None;
264 };
265 let name = name.to_str().ok()?;
266 c.binary_search_by(|(n, _)| (**n).cmp(name))
267 .ok()
268 .map(|i| c[i].1 as u64)
269 }
270
271 fn dtype(&self, ino: u64) -> u32 {
272 match self.nodes[ino as usize - 1].kind {
273 Kind::Dir(_) => libc::DT_DIR as u32,
274 Kind::File { .. } => libc::DT_REG as u32,
275 Kind::Symlink(_) => libc::DT_LNK as u32,
276 }
277 }
278
279 fn stat(&self, ino: u64) -> io::Result<stat64> {
280 let n = self.node(ino)?;
281 // Safe: stat64 is plain old data.
282 let mut st: stat64 = unsafe { std::mem::zeroed() };
283 st.st_ino = ino | INO_TAG;
284 st.st_blksize = 4096;
285 st.st_atime = 1;
286 st.st_mtime = 1;
287 st.st_ctime = 1;
288 match &n.kind {
289 Kind::Dir(_) => {
290 st.st_mode = libc::S_IFDIR | 0o555;
291 // 1: "subdirectory count unknown", which keeps find(1)
292 // from skipping entries based on it
293 st.st_nlink = 1;
294 }
295 Kind::File { size, exec } => {
296 st.st_mode = libc::S_IFREG | if *exec { 0o555 } else { 0o444 };
297 st.st_nlink = 1;
298 st.st_size = *size as i64;
299 st.st_blocks = size.div_ceil(512) as i64;
300 }
301 Kind::Symlink(t) => {
302 st.st_mode = libc::S_IFLNK | 0o777;
303 st.st_nlink = 1;
304 st.st_size = t.len() as i64;
305 }
306 }
307 Ok(st)
308 }
309
310 /// The unpacked store path `i`, fetching it first if needed. Concurrent
311 /// callers for the same path wait for one fetch.
312 fn fetch(&self, i: usize) -> io::Result<PathBuf> {
313 let mut slot = self.fetched[i].lock().unwrap();
314 if let Some(p) = &*slot {
315 return Ok(p.clone());
316 }
317 let name = &self.names[i];
318 let dest = self.cache.join(name);
319 let mut last = None;
320 for attempt in 1..=3 {
321 match self.fetch_once(i, &dest) {
322 Ok(()) => {
323 *slot = Some(dest.clone());
324 return Ok(dest);
325 }
326 Err(e) => {
327 eprintln!("fetching {name} (attempt {attempt}/3): {e}");
328 last = Some(e);
329 }
330 }
331 }
332 Err(last.unwrap())
333 }
334
335 fn fetch_once(&self, i: usize, dest: &Path) -> io::Result<()> {
336 let (name, info) = (&self.names[i], &self.infos[i]);
337 let t = Instant::now();
338 let tmp = dest.with_extension("tmp");
339 nar::remove(&tmp)?;
340 nar::remove(dest)?;
341 let (body, _) = self.http.open(&format!("{}/{}", info.cache, info.url))?;
342 let mut counted = CountingReader { inner: body, n: 0 };
343 let (digest, size) = nar::unpack(http::decoder(&info.compression, &mut counted)?, &tmp)?;
344 if !nar::hash_matches(&info.nar_hash, &digest) || size != info.nar_size {
345 nar::remove(&tmp)?;
346 return Err(io::Error::other(format!(
347 "{name}: NAR hash mismatch, sha256:{} ({size} bytes) != {} ({} bytes)",
348 nar::nix32(&digest),
349 info.nar_hash,
350 info.nar_size
351 )));
352 }
353 std::fs::rename(&tmp, dest)?;
354 self.stats.nar(name, counted.n, size);
355 eprintln!(
356 "fetched {name}: {} bytes in {:.1} s",
357 counted.n,
358 t.elapsed().as_secs_f64()
359 );
360 Ok(())
361 }
362
363 fn entry(&self, ino: u64) -> io::Result<Entry> {
364 Ok(Entry {
365 inode: ino,
366 generation: 0,
367 attr: self.stat(ino)?,
368 attr_flags: 0,
369 attr_timeout: TTL,
370 entry_timeout: TTL,
371 })
372 }
373
374 fn backing_id(&self, ino: u64, file: &File) -> Option<u32> {
375 if !self.passthrough.load(Ordering::Relaxed) {
376 return None;
377 }
378 let backing = self.backing.as_ref()?;
379 let mut ids = self.backing_ids.lock().unwrap();
380 if let Some((id, n)) = ids.get_mut(&ino) {
381 *n += 1;
382 return Some(*id);
383 }
384 match backing.open_fd(file.as_raw_fd()) {
385 Ok(id) => {
386 ids.insert(ino, (id, 1));
387 Some(id)
388 }
389 Err(e) => {
390 if !self.backing_failed.swap(true, Ordering::Relaxed) {
391 eprintln!(
392 "lower: FUSE passthrough backing open failed, reading via the daemon: {e}"
393 );
394 }
395 None
396 }
397 }
398 }
399
400 fn release_backing(&self, ino: u64) {
401 let mut ids = self.backing_ids.lock().unwrap();
402 if let Some((id, n)) = ids.get_mut(&ino) {
403 *n -= 1;
404 if *n == 0 {
405 let id = *id;
406 ids.remove(&ino);
407 drop(ids);
408 if let Some(b) = &self.backing {
409 b.close_id(id);
410 }
411 }
412 }
413 }
414}
415
416impl FileSystem for LazyFs {
417 type Inode = u64;
418 type Handle = u64;
419
420 fn lookup(&self, _ctx: &Context, parent: u64, name: &CStr) -> io::Result<Entry> {
421 match self.lookup_child(parent, name) {
422 Some(ino) => self.entry(ino),
423 None => Err(err(libc::ENOENT)),
424 }
425 }
426
427 fn getattr(&self, _ctx: &Context, ino: u64, _h: Option<u64>) -> io::Result<(stat64, Duration)> {
428 Ok((self.stat(ino)?, TTL))
429 }
430
431 fn readlink(&self, _ctx: &Context, ino: u64) -> io::Result<Vec<u8>> {
432 match &self.node(ino)?.kind {
433 Kind::Symlink(t) => Ok(t.to_vec()),
434 _ => Err(err(libc::EINVAL)),
435 }
436 }
437
438 fn open(
439 &self,
440 _ctx: &Context,
441 ino: u64,
442 flags: u32,
443 _fuse_flags: u32,
444 ) -> io::Result<(Option<u64>, OpenOptions, Option<u32>)> {
445 let n = self.node(ino)?;
446 if !matches!(n.kind, Kind::File { .. }) {
447 return Err(err(libc::EISDIR));
448 }
449 let acc = flags as i32 & libc::O_ACCMODE;
450 if acc != libc::O_RDONLY || flags as i32 & libc::O_TRUNC != 0 {
451 return Err(err(libc::EROFS));
452 }
453 let root = self.fetch(n.path as usize).map_err(|e| {
454 eprintln!("open {}/{}: {e}", self.names[n.path as usize], n.rel);
455 err(libc::EIO)
456 })?;
457 let file = File::open(if n.rel.is_empty() {
458 root
459 } else {
460 root.join(&*n.rel)
461 })?;
462 let backing = self.backing_id(ino, &file);
463 let h = self.next_handle.fetch_add(1, Ordering::Relaxed);
464 self.handles
465 .write()
466 .unwrap()
467 .insert(h, Arc::new(Handle { file, ino }));
468 Ok(match backing {
469 Some(id) => (Some(h), OpenOptions::PASSTHROUGH, Some(id)),
470 None => (Some(h), OpenOptions::KEEP_CACHE, None),
471 })
472 }
473
474 fn read(
475 &self,
476 _ctx: &Context,
477 _ino: u64,
478 handle: u64,
479 w: &mut dyn ZeroCopyWriter,
480 size: u32,
481 offset: u64,
482 _lock_owner: Option<u64>,
483 _flags: u32,
484 ) -> io::Result<usize> {
485 let h = self
486 .handles
487 .read()
488 .unwrap()
489 .get(&handle)
490 .cloned()
491 .ok_or_else(|| err(libc::EBADF))?;
492 // Safe: borrows the fd of `h.file`, which outlives `f`, without closing it.
493 let mut f = ManuallyDrop::new(unsafe { File::from_raw_fd(h.file.as_raw_fd()) });
494 w.write_from(&mut *f, size as usize, offset)
495 }
496
497 fn lseek(
498 &self,
499 _ctx: &Context,
500 _ino: u64,
501 handle: u64,
502 offset: u64,
503 whence: u32,
504 ) -> io::Result<u64> {
505 let h = self
506 .handles
507 .read()
508 .unwrap()
509 .get(&handle)
510 .cloned()
511 .ok_or_else(|| err(libc::EBADF))?;
512 // Safe: plain syscall on an fd we own.
513 let r = unsafe { libc::lseek(h.file.as_raw_fd(), offset as i64, whence as i32) };
514 if r < 0 {
515 return Err(io::Error::last_os_error());
516 }
517 Ok(r as u64)
518 }
519
520 fn flush(&self, _ctx: &Context, _ino: u64, _handle: u64, _lock_owner: u64) -> io::Result<()> {
521 Ok(())
522 }
523
524 fn fsync(&self, _ctx: &Context, _ino: u64, _datasync: bool, _handle: u64) -> io::Result<()> {
525 Ok(())
526 }
527
528 fn release(
529 &self,
530 _ctx: &Context,
531 _ino: u64,
532 _flags: u32,
533 handle: u64,
534 _flush: bool,
535 _flock_release: bool,
536 _lock_owner: Option<u64>,
537 ) -> io::Result<()> {
538 if let Some(h) = self.handles.write().unwrap().remove(&handle) {
539 self.release_backing(h.ino);
540 }
541 Ok(())
542 }
543
544 fn opendir(
545 &self,
546 _ctx: &Context,
547 ino: u64,
548 _flags: u32,
549 ) -> io::Result<(Option<u64>, OpenOptions)> {
550 match self.node(ino)?.kind {
551 Kind::Dir(_) => Ok((None, OpenOptions::KEEP_CACHE | OpenOptions::CACHE_DIR)),
552 _ => Err(err(libc::ENOTDIR)),
553 }
554 }
555
556 fn readdir(
557 &self,
558 _ctx: &Context,
559 ino: u64,
560 _handle: u64,
561 _size: u32,
562 offset: u64,
563 add_entry: &mut dyn FnMut(DirEntry) -> io::Result<usize>,
564 ) -> io::Result<()> {
565 let Kind::Dir(c) = &self.node(ino)?.kind else {
566 return Err(err(libc::ENOTDIR));
567 };
568 for (i, (name, child)) in c.iter().enumerate().skip(offset as usize) {
569 let child = *child as u64;
570 let n = add_entry(DirEntry {
571 ino: child | INO_TAG,
572 offset: i as u64 + 1,
573 type_: self.dtype(child),
574 name: name.as_bytes(),
575 })?;
576 if n == 0 {
577 break;
578 }
579 }
580 Ok(())
581 }
582
583 fn releasedir(&self, _ctx: &Context, _ino: u64, _flags: u32, _handle: u64) -> io::Result<()> {
584 Ok(())
585 }
586
587 fn getxattr(
588 &self,
589 _ctx: &Context,
590 _ino: u64,
591 _name: &CStr,
592 _size: u32,
593 ) -> io::Result<GetxattrReply> {
594 Err(err(libc::ENODATA))
595 }
596
597 fn listxattr(&self, _ctx: &Context, _ino: u64, size: u32) -> io::Result<ListxattrReply> {
598 Ok(if size == 0 {
599 ListxattrReply::Count(0)
600 } else {
601 ListxattrReply::Names(Vec::new())
602 })
603 }
604
605 fn access(&self, _ctx: &Context, _ino: u64, mask: u32) -> io::Result<()> {
606 if mask as i32 & libc::W_OK != 0 {
607 return Err(err(libc::EROFS));
608 }
609 Ok(())
610 }
611}