diff --git a/Cargo.lock b/Cargo.lock index 64fed08..5f1595e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -223,6 +223,7 @@ dependencies = [ "tracing", "tracing-human-layer", "tracing-subscriber", + "which", ] [[package]] @@ -851,6 +852,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "which" +version = "8.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81995fafaaaf6ae47a7d0cc83c67caf92aeb7e5331650ae6ff856f7c0c60c459" +dependencies = [ + "libc", +] + [[package]] name = "windows" version = "0.61.3" diff --git a/Cargo.toml b/Cargo.toml index 7408fb0..38a75c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,6 +47,7 @@ tap = "1.0.1" tracing = { version = "0.1.44", features = ["attributes"] } tracing-human-layer = "0.2.1" tracing-subscriber = { version = "0.3.22", default-features = false, features = ["std", "env-filter", "fmt", "ansi", "registry", "parking_lot"] } +which = "8.0.2" [profile.dev] opt-level = 1 @@ -63,3 +64,21 @@ opt-level = 1 [profile.release.package."*"] debug = true debug-assertions = true + +[lints.clippy] +#arithmetic_side_effects = "warn" +as_ptr_cast_mut = "warn" +assigning_clones = "warn" +borrow_as_ptr = "warn" +#cargo_common_metadata = "warn" +cast_lossless = "warn" +#cast_possible_truncation = "warn" +cast_possible_wrap = "warn" +cast_ptr_alignment = "warn" +cast_sign_loss = "warn" +clear_with_drain = "warn" +coerce_container_to_any = "warn" +derive_partial_eq_without_eq = "warn" +doc_broken_link = "warn" +doc_comment_double_space_linebreaks = "warn" +doc_markdown = "warn" diff --git a/package.nix b/package.nix index fa07871..8f828c9 100644 --- a/package.nix +++ b/package.nix @@ -58,6 +58,8 @@ in { inherit (self) strictDeps __structuredAttrs; inherit (self) doCheck doInstallCheck; + outputs = [ "out" "doc" ]; + src = lib.fileset.toSource { root = ./.; fileset = lib.fileset.unions [ diff --git a/rustfmt.toml b/rustfmt.toml index d662dc1..02c4dd5 100644 --- a/rustfmt.toml +++ b/rustfmt.toml @@ -4,3 +4,13 @@ match_block_trailing_comma = true merge_derives = false + +# Unstable options. +blank_lines_upper_bound = 3 +format_code_in_doc_comments = true +format_macro_matchers = true +# When structs, slices, arrays, and block/array-like macros are used as the last argument in an expression list, +# allow them to overflow (like blocks/closures) instead of being indented on a new line. +overflow_delimited_expr = true +# Put `type` and `const` items before methods. +reorder_impl_items = true diff --git a/shell.nix b/shell.nix index 39cd49f..a1b40ff 100644 --- a/shell.nix +++ b/shell.nix @@ -16,7 +16,21 @@ fenixLib ? let src = fetchTarball "https://github.com/nix-community/fenix/archive/main.tar.gz"; in import src { inherit pkgs; }, - fenixToolchain ? fenixLib.latest.toolchain, + fenixBaseToolchain ? fenixLib.stable.withComponents [ + "cargo" + "rustc" + "llvm-tools" + "rust-std" + "rust-docs" + "rust-src" + "rustc-dev" + "clippy" + ], + fenixToolchain ? fenixLib.combine [ + fenixBaseToolchain + # Rustfmt is very handy to have as nightly. + fenixLib.latest.rustfmt + ], }: let inherit (pkgs) lib; diff --git a/src/daemon.rs b/src/daemon.rs index 6543d3a..6fbbd86 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -1,6 +1,7 @@ use std::{ env, io, os::fd::{AsFd, BorrowedFd, IntoRawFd, OwnedFd, RawFd}, + process::{Command, Stdio}, sync::{ Arc, LazyLock, atomic::{AtomicUsize, Ordering}, @@ -12,7 +13,20 @@ use iddqd::{BiHashMap, IdOrdMap}; use mio::{Events, Interest, Poll, Token, event::Event, net::UnixListener, unix::SourceFd}; -use rustix::{buffer::spare_capacity, net::SocketFlags, process::Uid}; +use rustix::{ + buffer::spare_capacity, + net::SocketFlags, + process::{Pid, PidfdFlags, Uid, WaitId, WaitIdOptions}, +}; + +mod rustix { + pub use rustix::process::{getuid, pidfd_open, waitid}; + pub use rustix::*; +} + +//mod rustix_prelude { +// pub use rustix::process::{getuid, pidfd_open, waitid}; +//} use serde_json::StreamDeserializer; @@ -21,10 +35,7 @@ use crate::prelude::*; pub mod api; use api::DaemonCmd; -use crate::{ - SourceFile, SourceLine, - daemon_tokfd::{FdInfo, FdKind}, -}; +use crate::daemon_tokfd::{FdInfo, FdKind}; use crate::{OwnedFdWithFlags, TokenFd}; @@ -45,6 +56,14 @@ pub static TMPDIR: LazyLock<&'static Path> = LazyLock::new(|| { Box::leak(dir) }); +pub static NIXOS_REBUILD: LazyLock<&'static Path> = LazyLock::new(|| { + which::which("nixos-rebuild") + .inspect_err(|e| error!("couldn't find `nixos-rebuild` in PATH: {e}")) + .map(PathBuf::into_boxed_path) + .map(|boxed| &*Box::leak(boxed)) + .unwrap_or(Path::new("/run/current-system/sw/bin/nixos-rebuild")) +}); + const TIMEOUT_NEVER: Option = None; static NEXT_TOKEN_NUMBER: AtomicUsize = AtomicUsize::new(1); @@ -60,6 +79,42 @@ fn next_token() -> Token { Token(tok) } +trait EventExt { + type Display; + + fn display(&self) -> Self::Display; +} + +#[derive(Copy)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +struct EventDisplay { + token: Token, + error: bool, + writable: bool, + write_closed: bool, + readable: bool, + read_closed: bool, +} +impl EventExt for Event { + type Display = EventDisplay; + + fn display(&self) -> Self::Display { + EventDisplay { + token: self.token(), + error: self.is_error(), + writable: self.is_writable(), + write_closed: self.is_write_closed(), + readable: self.is_readable(), + read_closed: self.is_read_closed(), + } + } +} +impl Display for EventDisplay { + fn fmt(&self, f: &mut Formatter) -> FmtResult { + todo!() + } +} + #[derive(Debug)] pub struct Daemon { config_path: Arc, @@ -132,6 +187,32 @@ impl Daemon { token } + fn register_with_name(&mut self, fd: RawFd, kind: FdKind, name: Box) -> Token { + let token = next_token(); + + debug!( + "Registering new {} FdInfo for {fd} ({}) with token {token:?}", + name.to_string_lossy(), + kind.name_str(), + ); + + self.fd_info + .insert_unique(FdInfo::new_with_name(fd, kind, name)) + .unwrap(); + + self.tokfd + .insert_unique(TokenFd { token, fd }) + .unwrap_or_else(|e| todo!("{e}")); + + let mut source = SourceFd(&fd); + self.poller + .registry() + .register(&mut source, token, Interest::READABLE) + .unwrap_or_else(|e| unreachable!("registering {fd:?} with poller failed: {e}")); + + token + } + fn deregister(&mut self, fd: RawFd) { let info = self .fd_info @@ -279,6 +360,32 @@ const DAEMON: Token = Token(0); /// Private helpers. impl Daemon { + //fn proxy_stdio(&mut self, fd: &BorrowedFd) -> Result<(), IoError> { + // let info = self.fd_info.get(&fd.as_raw_fd()).unwrap(); + // let label = match info.kind { + // FdKind::ChildStdout => "stdout", + // FdKind::ChildStderr => "stderr", + // other => unreachable!("child stdio cannot have kind {other:?}"), + // }; + // // FIXME: don't use a new allocation every time. + // let mut buffer: Vec = Vec::with_capacity(1024); + // // FIXME: handle line buffering correctly. + // loop { + // let count = rustix::io::read(fd, spare_capacity(&mut buffer)) + // .inspect_err(|e| error!("read() on child stdio fd {fd:?} failed: {e}"))?; + // + // if count == 0 { + // break; + // } + // + // for line in buffer.lines() { + // debug!("[child {label}]: {}", line.as_bstr()) + // } + // } + // + // Ok(()) + //} + fn read_cmd(&mut self, fd: &BorrowedFd) -> Result<(), IoError> { // FIXME: don't use a new allocation every time. let mut cmd_buffer: Vec = Vec::with_capacity(1024); @@ -317,42 +424,63 @@ impl Daemon { } fn dispatch_cmd(&mut self, cmd: DaemonCmd) -> Result<(), IoError> { + // Write the new file... let (name, value) = match cmd { DaemonCmd::Append { name, value } => (name, value), }; - let mut opts = File::options(); - opts.read(true) - .write(true) - .create(false) - .custom_flags(libc::O_CLOEXEC); - let source_file = SourceFile::open_from(self.config_path.clone(), opts)?; + let source_file = crate::open_source_file(self.config_path.clone())?; let pri = crate::get_where(source_file.clone()).unwrap_or_else(|e| todo!("{e}")); let new_pri = pri - 1; - //let new_pri_line = - // crate::get_next_prio_line(source_file.clone(), Arc::from(name), Arc::from(value)); // Get next priority line. - let source_lines = source_file.lines()?; - let penultimate = source_lines.get(source_lines.len() - 2); - // FIXME: don't rely on whitespace lol - debug_assert_eq!(penultimate.map(SourceLine::text).as_deref(), Some(" ];")); - let penultimate = penultimate.unwrap(); - let new_generation = 0 - new_pri; - let new_line = SourceLine { - line: penultimate.line, - path: source_file.path(), - text: Arc::from(format!( - " {} = lib.mkOverride ({}) ({}); # DYNIX GENERATION {}", - name.to_nix_decl(), - new_pri, - value, - new_generation, - )), - }; - - drop(source_lines); + let opt_name = name.to_nix_decl(); + let new_line = crate::get_next_prio_line(source_file.clone(), &opt_name, new_pri, &value) + .unwrap_or_else(|e| panic!("someone is holding a reference to source.lines(): {e}")); crate::write_next_prio(source_file, new_line).unwrap_or_else(|e| todo!("{e}")); + // Rebuild and switch. + // FIXME: allow passing additional args. + let child = Command::new(*NIXOS_REBUILD) + .arg("switch") + .arg("--log-format") + .arg("raw-with-logs") + .arg("--no-reexec") + .arg("-v") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .inspect_err(|e| { + error!("failed to spawn `nixos-rebuild` command: {e}"); + })?; + + debug!("Spanwed child process {}", child.id()); + + let pid = Pid::from_child(&child); + + let stdout = child.stdout.unwrap_or_else(|| { + unreachable!("`child` is given `.stdout(Stdio::piped())`"); + }); + let stderr = child.stderr.unwrap_or_else(|| { + unreachable!("`child` is given `.stderr(Stdio::piped())`"); + }); + + let _token = self.register(stdout.into_raw_fd(), FdKind::ChildStdout); + let _token = self.register(stderr.into_raw_fd(), FdKind::ChildStderr); + + match rustix::process::pidfd_open(pid, PidfdFlags::NONBLOCK) { + Ok(pidfd) => { + debug!("Opened pidfd {pidfd:?}, for process {pid}"); + self.register(pidfd.into_raw_fd(), FdKind::Pid(pid)); + }, + Err(e) if e.kind() == IoErrorKind::NotFound => { + warn!("child {pid} not found; died before we could open it?"); + }, + Err(e) => { + error!("Error opening pidfd for child {pid}: {e}"); + return Err(e)?; + }, + } + Ok(()) } @@ -386,9 +514,9 @@ impl Daemon { loop { if tracing::enabled!(tracing::Level::DEBUG) { - trace!("Daemon loop iteration, with file descriptors: "); + debug!("Daemon loop iteration, with file descriptors: "); for info in &self.fd_info { - trace!("- {}", info.display()); + debug!("- {}", info.display()); } } @@ -441,7 +569,7 @@ impl Daemon { } fn handle_event(&mut self, event: &Event) -> Result<(), IoError> { - trace!("Handling event {event:?}"); + trace!("Handling event {event:#?}"); match event.token() { DAEMON => { @@ -487,17 +615,61 @@ impl Daemon { }, other_token => { // This must be a stream fd. - let stream_fd = self.fd_for_token(other_token).unwrap_or_else(|| { + let fd = self.fd_for_token(other_token).unwrap_or_else(|| { unreachable!("tried to get fd for non-existent token? {other_token:?}") }); + let Some(info) = self.fd_info.get(&fd) else { + panic!("Received an event on an unregistered fd {fd}; IO-safety violation?"); + }; if event.is_read_closed() { - self.deregister(stream_fd); - } else { - // SAFETY: oh boy. - let stream_fd = unsafe { BorrowedFd::borrow_raw(stream_fd) }; - self.read_cmd(&stream_fd).unwrap(); + self.deregister(fd); + return Ok(()); } + + match info.kind { + FdKind::Pid(pid) => { + debug!("Reaping child process {pid}"); + // SAFETY: `fd` cannot have been closed yet, since that's what we do here. + let pidfd = unsafe { BorrowedFd::borrow_raw(fd) }; + let status = rustix::waitid(WaitId::PidFd(pidfd), WaitIdOptions::EXITED) + .unwrap_or_else(|e| { + todo!("waitid() can fail? on pid {pid}: {e}"); + }) + .unwrap_or_else(|| { + todo!("waitid() returned None? for pid {pid}"); + }); + + debug!("waitid() for pid {pid} returned status: {status:?}"); + let is_dead = status.exited() || status.killed() || status.dumped(); + if !is_dead { + todo!("Handle process {pid} events that aren't death: {status:?}"); + } + let Some(exit_code) = status.exit_status() else { + unreachable!("Process {pid} died with no exit code at all? {status:?}"); + }; + debug!("Child process {pid} exited with code {exit_code}"); + + // Close the pidfd. + self.deregister(fd); + }, + //FdKind::ChildStdout => { + // warn!("got stdout"); + // todo!(); + //}, + //FdKind::ChildStderr => { + // warn!("got stderr"); + // // SAFETY: oh boy. + // let stderr = unsafe { BorrowedFd::borrow_raw(fd) }; + // self.proxy_stdio(&stderr).unwrap(); + //}, + FdKind::SockStream => { + // SAFETY: oh boy. + let stream_fd = unsafe { BorrowedFd::borrow_raw(fd) }; + self.read_cmd(&stream_fd).unwrap(); + }, + kind => todo!("{kind:?}"), + }; }, } diff --git a/src/daemon_tokfd.rs b/src/daemon_tokfd.rs index e00f5f2..cb436c6 100644 --- a/src/daemon_tokfd.rs +++ b/src/daemon_tokfd.rs @@ -3,6 +3,7 @@ use std::{os::fd::RawFd, sync::OnceLock}; use circular_buffer::CircularBuffer; use iddqd::{BiHashItem, IdOrdItem}; use mio::Token; +use rustix::process::Pid; use crate::prelude::*; @@ -76,11 +77,11 @@ impl FdInfo { impl IdOrdItem for FdInfo { type Key<'a> = &'a RawFd; + iddqd::id_upcast!(); + fn key(&self) -> &RawFd { &self.fd } - - iddqd::id_upcast!(); } #[derive(Debug)] @@ -105,13 +106,16 @@ impl<'a> Display for FdInfoDisplay<'a> { } #[derive(Copy)] -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Default)] +#[derive(Debug, Clone, PartialEq, Eq, Hash, Default)] #[non_exhaustive] pub enum FdKind { File, Socket, SockStream, Poller, + ChildStdout, + ChildStderr, + Pid(Pid), #[default] Unknown, } @@ -124,6 +128,9 @@ impl FdKind { Socket => "socket", SockStream => "socket stream", Poller => "poller", + ChildStdout => "child stdout", + ChildStderr => "child stderr", + Pid(_) => "pidfd", Unknown => "«unknown»", } } @@ -140,6 +147,8 @@ impl BiHashItem for TokenFd { type K1<'a> = Token; type K2<'a> = RawFd; + iddqd::bi_upcast!(); + fn key1(&self) -> Token { self.token } @@ -147,8 +156,6 @@ impl BiHashItem for TokenFd { fn key2(&self) -> RawFd { self.fd } - - iddqd::bi_upcast!(); } impl From for (Token, RawFd) { diff --git a/src/lib.rs b/src/lib.rs index b2b597a..9475e70 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -99,6 +99,22 @@ static MK_OVERRIDE_RE: LazyLock = LazyLock::new(|| { Regex::new(r"(?-u)\bmkOverride\s+\((?[\d-]+)\)").unwrap() }); +pub(crate) fn open_source_file(path: Arc) -> Result { + let mut opts = File::options(); + opts.read(true) + .write(true) + .create(false) + .custom_flags(libc::O_CLOEXEC); + + SourceFile::open_from(Arc::clone(&path), opts) + .tap_err(|e| error!("couldn't open source file at {}: {e}", path.display())) +} + +pub(crate) fn get_line_to_insert() -> SourceLine { + // + todo!(); +} + #[tracing::instrument(level = "debug")] pub fn do_append(args: Arc, append_args: AppendCmd) -> Result<(), BoxDynError> { let filepath = Path::new(&args.file); @@ -110,22 +126,16 @@ pub fn do_append(args: Arc, append_args: AppendCmd) -> Result<(), BoxDynEr filepath.to_path_buf() }; - let mut opts = File::options(); - opts.read(true) - .write(true) - .create(false) - .custom_flags(libc::O_CLOEXEC); - - let source_file = SourceFile::open_from(Arc::from(filepath), opts)?; + let source_file = open_source_file(Arc::from(filepath))?; let pri = get_where(source_file.clone())?; let new_pri = pri - 1; let new_pri_line = get_next_prio_line( source_file.clone(), - append_args.name, + &append_args.name, new_pri, - append_args.value, + &append_args.value, )?; debug!("new_pri_line={new_pri_line}"); @@ -210,9 +220,9 @@ pub fn get_where(dynamic_nix: SourceFile) -> Result { pub fn get_next_prio_line( source: SourceFile, - option_name: Arc, + option_name: &str, new_prio: i64, - new_value: Arc, + new_value: &str, ) -> Result { let source_lines = source.lines()?; let penultimate = source_lines.get(source_lines.len() - 2); diff --git a/src/source.rs b/src/source.rs index 287f72e..36cd70b 100644 --- a/src/source.rs +++ b/src/source.rs @@ -6,6 +6,7 @@ use std::{ cell::{Ref, RefCell}, hash::Hash, io::{BufRead, BufReader, BufWriter}, + mem::{self, MaybeUninit}, ops::Deref, ptr, sync::{Arc, Mutex, OnceLock}, @@ -103,6 +104,28 @@ pub struct SourceFile { lines: Arc>>>, } +#[derive(Debug)] +#[repr(transparent)] +pub struct OpaqueDerefSourceLines<'s>(Ref<'s, [SourceLine]>); +impl<'s> Deref for OpaqueDerefSourceLines<'s> { + type Target = [SourceLine]; + + fn deref(&self) -> &[SourceLine] { + &*self.0 + } +} + +#[derive(Debug)] +#[repr(transparent)] +pub struct OpaqueDerefSourceLine<'s>(Ref<'s, SourceLine>); +impl<'s> Deref for OpaqueDerefSourceLine<'s> { + type Target = SourceLine; + + fn deref(&self) -> &SourceLine { + &*self.0 + } +} + impl SourceFile { /// Panics if `path` is a directory path instead of a file path. pub fn open_from(path: Arc, options: OpenOptions) -> Result { @@ -159,15 +182,17 @@ impl SourceFile { Ok(self._lines_slice()) } - pub fn lines(&self) -> Result + '_, IoError> { - self._lines() + pub fn lines(&self) -> Result, IoError> { + let lines = self._lines()?; + + Ok(OpaqueDerefSourceLines(lines)) } - pub fn line(&self, line: Line) -> Result + '_, IoError> { + pub fn line(&self, line: Line) -> Result, IoError> { let lines_lock = self._lines()?; let line = Ref::map(lines_lock, |lines| &lines[line.index() as usize]); - Ok(line) + Ok(OpaqueDerefSourceLine(line)) } /// `lines` but already be initialized. diff --git a/tests/harmonia/test-script.py b/tests/harmonia/test-script.py index c40e130..39d0e5d 100644 --- a/tests/harmonia/test-script.py +++ b/tests/harmonia/test-script.py @@ -80,6 +80,7 @@ machine.wait_for_unit("install-dynix.service") dynix_out = machine.succeed("dynix --version") assert "dynix" in dynix_out, f"dynix not in {dynix_out=}" + # Config should have our initial values. config_toml = get_config_file() assert int(config_toml['workers']) == 4, f"{config_toml['workers']=} != 4"