This repository has no description
8.0 kB
243 lines
1//! Lazy nix store for CI's machine builds (.github/lazy-store.sh).
2//!
3//! closure OUT collect the closure of the store paths on stdin
4//! (one per line) from the caches by fetching only
5//! narinfos. Writes OUT/paths.json (per path:
6//! cache, NAR url, compression, hash), OUT/db.dump
7//! (input for `nix-store --load-db`), OUT/narinfo
8//! (their narinfos as a binary cache, for `nix
9//! store copy-sigs`) and OUT/closure.json.
10//! fetch OUT NAME... fetch store paths of OUT/paths.json like the
11//! first open() would, and check them against
12//! their listing (testing without FUSE)
13//! mount OUT UPPER MNT mount the store dir UPPER, merged with the paths
14//! of OUT/paths.json, at MNT (see storefs.rs).
15//! Paths without a listing, and those referring
16//! to them, are left out of OUT/paths.json and
17//! OUT/db.dump (counted in OUT/no-listing.json).
18//! Stats go to OUT/stats.json.
19//! [--workers N] FUSE worker threads (default 64)
20//! [--no-passthrough] don't use kernel FUSE passthrough
21
22mod closure;
23mod http;
24mod lazy;
25mod nar;
26mod stats;
27mod storefs;
28
29use std::fs::File;
30use std::io;
31use std::os::fd::{AsRawFd, BorrowedFd, RawFd};
32use std::path::{Path, PathBuf};
33use std::sync::{Arc, OnceLock};
34
35use fuse_backend_rs::api::server::Server;
36use fuse_backend_rs::passthrough::{self, BackingFiles, PassthroughFs};
37use fuse_backend_rs::transport::{
38 fuse_backing_close, fuse_backing_open, FuseServing, FuseSession, SyncFuseServing,
39 SyncServingConfig,
40};
41
42pub const STORE: &str = "/nix/store/";
43
44/// Registers backing files for kernel FUSE passthrough on the session's
45/// /dev/fuse fd, which only exists once mounted.
46#[derive(Default)]
47pub struct Backing(OnceLock<File>);
48
49impl Backing {
50 pub fn open_fd(&self, fd: RawFd) -> io::Result<u32> {
51 let f = self
52 .0
53 .get()
54 .ok_or_else(|| io::Error::from_raw_os_error(libc::ENODEV))?;
55 fuse_backing_open(f, fd)
56 }
57
58 pub fn close_id(&self, id: u32) {
59 if let Some(f) = self.0.get() {
60 if let Err(e) = fuse_backing_close(f, id) {
61 eprintln!("closing backing id {id}: {e}");
62 }
63 }
64 }
65}
66
67struct SharedBacking(Arc<Backing>);
68
69impl BackingFiles for SharedBacking {
70 fn open(&self, fd: BorrowedFd<'_>) -> io::Result<u32> {
71 self.0.open_fd(fd.as_raw_fd())
72 }
73
74 fn close(&self, id: u32) -> io::Result<()> {
75 let f = self
76 .0
77 .0
78 .get()
79 .ok_or_else(|| io::Error::from_raw_os_error(libc::ENODEV))?;
80 fuse_backing_close(f, id)
81 }
82}
83
84/// fuse-backend-rs logs through `log`: warnings and errors to stderr.
85struct Logger;
86
87impl log::Log for Logger {
88 fn enabled(&self, m: &log::Metadata) -> bool {
89 m.level() <= log::Level::Warn
90 }
91
92 fn log(&self, r: &log::Record) {
93 if self.enabled(r.metadata()) {
94 eprintln!("{} {}: {}", r.level(), r.target(), r.args());
95 }
96 }
97
98 fn flush(&self) {}
99}
100
101/// PassthroughFs keeps an O_PATH fd per inode the kernel looked up and
102/// hasn't forgotten, so the default soft limit of 1024 runs out within the
103/// first build. Raise the limit as far as the kernel allows (we run as
104/// root, so the hard limit too, up to fs.nr_open).
105fn raise_nofile() {
106 let nr_open = std::fs::read_to_string("/proc/sys/fs/nr_open")
107 .ok()
108 .and_then(|s| s.trim().parse::<libc::rlim_t>().ok())
109 .unwrap_or(1 << 20);
110 let mut rl = libc::rlimit {
111 rlim_cur: 0,
112 rlim_max: 0,
113 };
114 unsafe {
115 libc::getrlimit(libc::RLIMIT_NOFILE, &mut rl);
116 let want = libc::rlimit {
117 rlim_cur: nr_open,
118 rlim_max: nr_open.max(rl.rlim_max),
119 };
120 if libc::setrlimit(libc::RLIMIT_NOFILE, &want) != 0 {
121 let soft = libc::rlimit {
122 rlim_cur: rl.rlim_max,
123 rlim_max: rl.rlim_max,
124 };
125 libc::setrlimit(libc::RLIMIT_NOFILE, &soft);
126 }
127 libc::getrlimit(libc::RLIMIT_NOFILE, &mut rl);
128 }
129 eprintln!("RLIMIT_NOFILE: {} (hard {})", rl.rlim_cur, rl.rlim_max);
130}
131
132fn mount(
133 rt: &tokio::runtime::Runtime,
134 outdir: &Path,
135 upper: &Path,
136 mnt: &Path,
137 workers: usize,
138 use_passthrough: bool,
139) -> io::Result<()> {
140 raise_nofile();
141 let backing = use_passthrough.then(|| Arc::new(Backing::default()));
142 let lower = rt.block_on(lazy::LazyFs::new(outdir, backing.clone()))?;
143
144 let upper_fs = {
145 let cfg = passthrough::Config {
146 root_dir: upper
147 .to_str()
148 .ok_or_else(|| io::Error::other("non-UTF-8 upper dir"))?
149 .into(),
150 do_import: true,
151 xattr: false,
152 ..Default::default()
153 };
154 let mut fs = PassthroughFs::<()>::new(cfg)?;
155 if let Some(b) = &backing {
156 fs.set_backing_files(Arc::new(SharedBacking(b.clone())));
157 }
158 fs
159 };
160 let stats = lower.stats.clone();
161 let fs = storefs::StoreFs::new(upper_fs, upper.to_path_buf(), lower);
162
163 let mut se = FuseSession::new(mnt, "lazystore", "", false).map_err(io::Error::other)?;
164 se.mount().map_err(io::Error::other)?;
165 if let Some(b) = &backing {
166 let f = se
167 .get_fuse_file()
168 .ok_or_else(|| io::Error::other("no fuse fd"))?
169 .try_clone()?;
170 let _ = b.0.set(f);
171 }
172 let server = Arc::new(Server::new(fs));
173 let serving = SyncFuseServing::new(
174 se,
175 server,
176 SyncServingConfig {
177 workers,
178 blocking: true,
179 },
180 )
181 .map_err(io::Error::other)?;
182 eprintln!(
183 "mounted {} on {} ({workers} workers)",
184 upper.display(),
185 mnt.display()
186 );
187 serving.wait();
188 stats.save();
189 eprintln!("FUSE main loop exited");
190 Ok(())
191}
192
193fn usage() -> ! {
194 eprintln!(
195 "usage: lazy-store closure OUT <PATHS | fetch OUT NAME... | mount OUT UPPER MNT [--workers N] [--no-passthrough]"
196 );
197 std::process::exit(2)
198}
199
200fn main() {
201 let _ = log::set_logger(&Logger).map(|_| log::set_max_level(log::LevelFilter::Warn));
202 let args: Vec<String> = std::env::args().skip(1).collect();
203 let rt = tokio::runtime::Runtime::new().expect("starting the tokio runtime");
204 let r = match args.first().map(String::as_str) {
205 Some("closure") if args.len() == 2 => rt.block_on(closure::closure(Path::new(&args[1]))),
206 Some("fetch") if args.len() >= 3 => rt.block_on(async {
207 let fs = lazy::LazyFs::new(Path::new(&args[1]), None).await?;
208 for name in &args[2..] {
209 println!("{}", fs.fetch_and_check(name).await?.display());
210 }
211 Ok(())
212 }),
213 Some("mount") if args.len() >= 4 => {
214 let (mut workers, mut pt) = (64, true);
215 let mut rest = args[4..].iter();
216 while let Some(a) = rest.next() {
217 match a.as_str() {
218 "--workers" => {
219 workers = rest
220 .next()
221 .and_then(|n| n.parse().ok())
222 .unwrap_or_else(|| usage())
223 }
224 "--no-passthrough" => pt = false,
225 _ => usage(),
226 }
227 }
228 mount(
229 &rt,
230 &PathBuf::from(&args[1]),
231 Path::new(&args[2]),
232 Path::new(&args[3]),
233 workers,
234 pt,
235 )
236 }
237 _ => usage(),
238 };
239 if let Err(e) = r {
240 eprintln!("error: {e}");
241 std::process::exit(1);
242 }
243}