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