Compare commits
3 commits
420fac5f18
...
a9b63f4d58
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a9b63f4d58 | ||
|
|
2ecd987be6 | ||
|
|
d4b40e8cc1 |
12 changed files with 529 additions and 219 deletions
10
Cargo.lock
generated
10
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
19
Cargo.toml
19
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"
|
||||
|
|
|
|||
|
|
@ -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 [
|
||||
|
|
|
|||
10
rustfmt.toml
10
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
|
||||
|
|
|
|||
16
shell.nix
16
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;
|
||||
|
||||
|
|
|
|||
521
src/daemon.rs
521
src/daemon.rs
|
|
@ -1,7 +1,7 @@
|
|||
use std::{
|
||||
env, io,
|
||||
ops::Deref,
|
||||
os::fd::{AsFd, BorrowedFd, IntoRawFd, OwnedFd, RawFd},
|
||||
process::{Command, Stdio},
|
||||
sync::{
|
||||
Arc, LazyLock,
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
|
|
@ -11,18 +11,31 @@ use std::{
|
|||
|
||||
use iddqd::{BiHashMap, IdOrdMap};
|
||||
|
||||
use mio::{Events, Interest, Poll, Token, net::UnixListener, unix::SourceFd};
|
||||
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::{Deserialize, Serialize};
|
||||
use serde_json::StreamDeserializer;
|
||||
|
||||
use crate::{
|
||||
SourceFile, SourceLine,
|
||||
daemon_tokfd::{FdInfo, FdKind},
|
||||
prelude::*,
|
||||
};
|
||||
use crate::prelude::*;
|
||||
|
||||
pub mod api;
|
||||
use api::DaemonCmd;
|
||||
|
||||
use crate::daemon_tokfd::{FdInfo, FdKind};
|
||||
|
||||
use crate::{OwnedFdWithFlags, TokenFd};
|
||||
|
||||
|
|
@ -43,75 +56,13 @@ pub static TMPDIR: LazyLock<&'static Path> = LazyLock::new(|| {
|
|||
Box::leak(dir)
|
||||
});
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
#[derive(Deserialize, Serialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum ConvenientAttrPath {
|
||||
Dotted(Box<str>),
|
||||
Split(Box<[Box<str>]>),
|
||||
}
|
||||
|
||||
impl ConvenientAttrPath {
|
||||
/// Not currently used, but here for completeness.
|
||||
#[expect(dead_code)]
|
||||
pub fn clone_from_dotted(s: &str) -> Self {
|
||||
Self::Dotted(Box::from(s))
|
||||
}
|
||||
|
||||
/// Not currently used, but here for completeness.
|
||||
#[expect(dead_code)]
|
||||
pub fn clone_from_split(s: &[&str]) -> Self {
|
||||
Self::from_str_iter(s.iter().map(Deref::deref))
|
||||
}
|
||||
|
||||
pub fn from_str_iter<'i, I>(iter: I) -> Self
|
||||
where
|
||||
I: Iterator<Item = &'i str>,
|
||||
{
|
||||
let boxed = iter.map(Box::from);
|
||||
Self::Split(Box::from_iter(boxed))
|
||||
}
|
||||
|
||||
pub fn to_nix_decl(&self) -> Box<str> {
|
||||
use ConvenientAttrPath::*;
|
||||
match self {
|
||||
Dotted(s) => s.clone(),
|
||||
Split(path) => {
|
||||
// FIXME: quote if necessary
|
||||
path.join(".").into_boxed_str()
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<'i> FromIterator<&'i str> for ConvenientAttrPath {
|
||||
fn from_iter<I>(iter: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = &'i str>,
|
||||
{
|
||||
Self::from_str_iter(iter.into_iter())
|
||||
}
|
||||
}
|
||||
|
||||
impl FromIterator<Box<str>> for ConvenientAttrPath {
|
||||
fn from_iter<I>(iter: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = Box<str>>,
|
||||
{
|
||||
Self::Split(Box::from_iter(iter))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
#[derive(serde::Deserialize, serde::Serialize)]
|
||||
#[serde(tag = "action", content = "args", rename_all = "snake_case")]
|
||||
// FIXME: rename to not confuse with the clap argument type.
|
||||
pub enum DaemonCmd {
|
||||
Append {
|
||||
name: ConvenientAttrPath,
|
||||
value: Box<str>,
|
||||
},
|
||||
}
|
||||
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<Duration> = None;
|
||||
|
||||
|
|
@ -121,12 +72,49 @@ fn next_token() -> Token {
|
|||
|
||||
// If the increment wrapped to 0, then we just increment it again.
|
||||
if tok == 0 {
|
||||
return Token(NEXT_TOKEN_NUMBER.fetch_add(1, Ordering::SeqCst));
|
||||
warn!("File descriptor token wrapped. That's... a lot.");
|
||||
return next_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<Path>,
|
||||
|
|
@ -199,6 +187,32 @@ impl Daemon {
|
|||
token
|
||||
}
|
||||
|
||||
fn register_with_name<S>(&mut self, fd: RawFd, kind: FdKind, name: Box<OsStr>) -> 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
|
||||
|
|
@ -342,9 +356,36 @@ impl Daemon {
|
|||
self.fd.as_fd()
|
||||
}
|
||||
}
|
||||
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<u8> = 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<u8> = Vec::with_capacity(1024);
|
||||
|
|
@ -383,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(())
|
||||
}
|
||||
|
||||
|
|
@ -436,7 +498,6 @@ impl Daemon {
|
|||
);
|
||||
}
|
||||
let mut daemon_source = SourceFd(&raw_fd);
|
||||
const DAEMON: Token = Token(0);
|
||||
self.tokfd
|
||||
.insert_unique(TokenFd {
|
||||
token: DAEMON,
|
||||
|
|
@ -453,103 +514,167 @@ 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());
|
||||
}
|
||||
}
|
||||
|
||||
match self.poller.poll(&mut events, TIMEOUT_NEVER) {
|
||||
Ok(_) => {
|
||||
trace!(
|
||||
"mio::Poller::poll() got events: {:?}",
|
||||
events.iter().size_hint().0,
|
||||
);
|
||||
if events.is_empty() {
|
||||
unreachable!(
|
||||
"epoll_wait() with a \"forever\" timeout should never give empty events",
|
||||
);
|
||||
}
|
||||
|
||||
let _ = self.fd_error_pop(self.poller.as_raw_fd());
|
||||
},
|
||||
Err(e) if e.kind() == IoErrorKind::Interrupted => {
|
||||
// EINTR is silly.
|
||||
continue;
|
||||
},
|
||||
Err(e) => {
|
||||
if let Some(Errno::BADF) = e.raw_os_error().map(Errno::from_raw_os_error) {
|
||||
unreachable!("EBADF on poller fd; IO safety violation?");
|
||||
}
|
||||
warn!("mio Poll::poll() error: {e}");
|
||||
self.fd_error_push(self.poller.as_raw_fd(), e)
|
||||
.tap_err(|e| {
|
||||
error!("accumulated too many errors for mio::Poll::poll(): {e}")
|
||||
})?;
|
||||
},
|
||||
}
|
||||
|
||||
for event in &events {
|
||||
trace!("event is {event:?}");
|
||||
match event.token() {
|
||||
DAEMON => {
|
||||
let is_sock = self.main_fd_info().kind == FdKind::Socket;
|
||||
if !is_sock {
|
||||
// SAFETY: oh boy: disjoint borrows with extra steps.
|
||||
let file_fd = unsafe { BorrowedFd::borrow_raw(self.fd.as_raw_fd()) };
|
||||
self.read_cmd(&file_fd).unwrap();
|
||||
continue;
|
||||
}
|
||||
|
||||
// Accept, first.
|
||||
let flags = SocketFlags::NONBLOCK | SocketFlags::CLOEXEC;
|
||||
let stream_fd = match rustix::net::accept_with(&self.fd, flags) {
|
||||
Ok(stream) => {
|
||||
debug!(
|
||||
"Accepted connection from socket {:?} as stream {:?}",
|
||||
self.fd, stream,
|
||||
);
|
||||
stream
|
||||
},
|
||||
Err(e) => {
|
||||
error!("accept4 on daemon socket failed: {e}");
|
||||
self.fd_error_push(self.fd.as_raw_fd(), e.into())
|
||||
.tap_err(|e| {
|
||||
error!(
|
||||
"Accumulated too many errors for daemon fd {:?}: {e}",
|
||||
self.fd
|
||||
)
|
||||
})?;
|
||||
continue;
|
||||
},
|
||||
};
|
||||
|
||||
// Add this stream to our poll interest list.
|
||||
// NOTE: `stream_fd` is now effectively `ManuallyDrop`.
|
||||
let stream_fd = stream_fd.into_raw_fd();
|
||||
let _token = self.register(stream_fd, FdKind::SockStream);
|
||||
|
||||
// Wait for the next poll to handle.
|
||||
},
|
||||
other_token => {
|
||||
// This must be a stream fd.
|
||||
let stream_fd = self.fd_for_token(other_token).unwrap_or_else(|| {
|
||||
unreachable!("tried to get fd for no-existent token? {other_token:?}")
|
||||
});
|
||||
|
||||
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();
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
let poll_result = self.poller.poll(&mut events, TIMEOUT_NEVER);
|
||||
self.handle_poll(poll_result, &events)?;
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_poll(
|
||||
&mut self,
|
||||
poll_result: Result<(), IoError>,
|
||||
events: &Events,
|
||||
) -> Result<(), IoError> {
|
||||
match poll_result {
|
||||
Ok(()) => {
|
||||
trace!(
|
||||
"mio::Poller::poll() got events: {:?}",
|
||||
events.iter().size_hint().0,
|
||||
);
|
||||
if events.is_empty() {
|
||||
unreachable!(
|
||||
"epoll_wait() with a \"forever\" timeout should never give empty events",
|
||||
);
|
||||
}
|
||||
|
||||
let _ = self.fd_error_pop(self.poller.as_raw_fd());
|
||||
},
|
||||
Err(e) if e.kind() == IoErrorKind::Interrupted => {
|
||||
// EINTR is silly.
|
||||
// Return early, and poll() again.
|
||||
return Ok(());
|
||||
},
|
||||
Err(e) => {
|
||||
if let Some(Errno::BADF) = e.raw_os_error().map(Errno::from_raw_os_error) {
|
||||
panic!("EBADF on poller fd; IO safety violation?");
|
||||
}
|
||||
warn!("mio Poll::poll() error: {e}");
|
||||
self.fd_error_push(self.poller.as_raw_fd(), e)
|
||||
.tap_err(|e| {
|
||||
error!("accumulated too many errors for mio::Poll::poll(): {e}")
|
||||
})?;
|
||||
},
|
||||
}
|
||||
|
||||
for event in events {
|
||||
self.handle_event(event)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn handle_event(&mut self, event: &Event) -> Result<(), IoError> {
|
||||
trace!("Handling event {event:#?}");
|
||||
|
||||
match event.token() {
|
||||
DAEMON => {
|
||||
let is_sock = self.main_fd_info().kind == FdKind::Socket;
|
||||
if !is_sock {
|
||||
// SAFETY: oh boy: disjoint borrows with extra steps.
|
||||
let file_fd = unsafe { BorrowedFd::borrow_raw(self.fd.as_raw_fd()) };
|
||||
self.read_cmd(&file_fd).unwrap();
|
||||
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Accept, first.
|
||||
let flags = SocketFlags::NONBLOCK | SocketFlags::CLOEXEC;
|
||||
let stream_fd = match rustix::net::accept_with(&self.fd, flags) {
|
||||
Ok(stream) => {
|
||||
debug!(
|
||||
"Accepted connection from socket {:?} as stream {:?}",
|
||||
self.fd, stream,
|
||||
);
|
||||
stream
|
||||
},
|
||||
Err(e) => {
|
||||
error!("accept4 on daemon socket failed: {e}");
|
||||
self.fd_error_push(self.fd.as_raw_fd(), e.into())
|
||||
.tap_err(|e| {
|
||||
error!(
|
||||
"Accumulated too many errors for daemon fd {:?}: {e}",
|
||||
self.fd
|
||||
)
|
||||
})?;
|
||||
|
||||
return Ok(());
|
||||
},
|
||||
};
|
||||
|
||||
// Add this stream to our poll interest list.
|
||||
// NOTE: `stream_fd` is now effectively `ManuallyDrop`.
|
||||
let stream_fd = stream_fd.into_raw_fd();
|
||||
let _token = self.register(stream_fd, FdKind::SockStream);
|
||||
|
||||
// Wait for the next poll to handle.
|
||||
},
|
||||
other_token => {
|
||||
// This must be a stream fd.
|
||||
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(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:?}"),
|
||||
};
|
||||
},
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Daemon {
|
||||
|
|
|
|||
59
src/daemon/api.rs
Normal file
59
src/daemon/api.rs
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
use std::ops::Deref;
|
||||
|
||||
use crate::prelude::*;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
mod impls;
|
||||
|
||||
/// Unfortunately, empty-string identifiers are valid Nix...
|
||||
///
|
||||
/// This type does not provide a [`Default`] impl, however.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
#[derive(Deserialize, Serialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum ConvenientAttrPath {
|
||||
Dotted(Box<str>),
|
||||
Split(Box<[Box<str>]>),
|
||||
}
|
||||
|
||||
impl ConvenientAttrPath {
|
||||
pub fn clone_from_dotted(s: &str) -> Self {
|
||||
Self::Dotted(Box::from(s))
|
||||
}
|
||||
|
||||
pub fn clone_from_split(s: &[&str]) -> Self {
|
||||
Self::from_str_iter(s.iter().map(Deref::deref))
|
||||
}
|
||||
|
||||
pub fn from_str_iter<'i, I>(iter: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = &'i str>,
|
||||
{
|
||||
let iter = iter.into_iter();
|
||||
let boxed = iter.map(Box::from);
|
||||
Self::Split(Box::from_iter(boxed))
|
||||
}
|
||||
|
||||
pub fn to_nix_decl(&self) -> Box<str> {
|
||||
use ConvenientAttrPath::*;
|
||||
match self {
|
||||
Dotted(s) => s.clone(),
|
||||
Split(path) => {
|
||||
// FIXME: quote if necessary
|
||||
path.join(".").into_boxed_str()
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
#[derive(serde::Deserialize, serde::Serialize)]
|
||||
#[serde(tag = "action", content = "args", rename_all = "snake_case")]
|
||||
// FIXME: rename to not confuse with the clap argument type.
|
||||
pub enum DaemonCmd {
|
||||
Append {
|
||||
name: ConvenientAttrPath,
|
||||
value: Box<str>,
|
||||
},
|
||||
}
|
||||
27
src/daemon/api/impls.rs
Normal file
27
src/daemon/api/impls.rs
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
use super::ConvenientAttrPath;
|
||||
|
||||
impl<'i> FromIterator<&'i str> for ConvenientAttrPath {
|
||||
fn from_iter<I>(iter: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = &'i str>,
|
||||
{
|
||||
Self::from_str_iter(iter)
|
||||
}
|
||||
}
|
||||
impl FromIterator<Box<str>> for ConvenientAttrPath {
|
||||
fn from_iter<I>(iter: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = Box<str>>,
|
||||
{
|
||||
Self::Split(Box::from_iter(iter))
|
||||
}
|
||||
}
|
||||
impl FromIterator<String> for ConvenientAttrPath {
|
||||
fn from_iter<I>(iter: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = String>,
|
||||
{
|
||||
let iter = iter.into_iter();
|
||||
Self::Split(Box::from_iter(iter.map(String::into_boxed_str)))
|
||||
}
|
||||
}
|
||||
|
|
@ -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<TokenFd> for (Token, RawFd) {
|
||||
|
|
|
|||
33
src/lib.rs
33
src/lib.rs
|
|
@ -55,6 +55,7 @@ mod color;
|
|||
pub use color::{_CLI_ENABLE_COLOR, SHOULD_COLOR};
|
||||
mod daemon;
|
||||
pub use daemon::Daemon;
|
||||
pub use daemon::api as daemon_api;
|
||||
mod daemon_io;
|
||||
pub use daemon_io::OwnedFdWithFlags;
|
||||
mod daemon_tokfd;
|
||||
|
|
@ -98,6 +99,22 @@ static MK_OVERRIDE_RE: LazyLock<Regex> = LazyLock::new(|| {
|
|||
Regex::new(r"(?-u)\bmkOverride\s+\((?<priority>[\d-]+)\)").unwrap()
|
||||
});
|
||||
|
||||
pub(crate) fn open_source_file(path: Arc<Path>) -> Result<SourceFile, IoError> {
|
||||
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<Args>, append_args: AppendCmd) -> Result<(), BoxDynError> {
|
||||
let filepath = Path::new(&args.file);
|
||||
|
|
@ -109,22 +126,16 @@ pub fn do_append(args: Arc<Args>, 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}");
|
||||
|
|
@ -209,9 +220,9 @@ pub fn get_where(dynamic_nix: SourceFile) -> Result<i64, BoxDynError> {
|
|||
|
||||
pub fn get_next_prio_line(
|
||||
source: SourceFile,
|
||||
option_name: Arc<str>,
|
||||
option_name: &str,
|
||||
new_prio: i64,
|
||||
new_value: Arc<str>,
|
||||
new_value: &str,
|
||||
) -> Result<SourceLine, BoxDynError> {
|
||||
let source_lines = source.lines()?;
|
||||
let penultimate = source_lines.get(source_lines.len() - 2);
|
||||
|
|
|
|||
|
|
@ -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<OnceLock<RefCell<Vec<SourceLine>>>>,
|
||||
}
|
||||
|
||||
#[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<Path>, options: OpenOptions) -> Result<Self, IoError> {
|
||||
|
|
@ -159,15 +182,17 @@ impl SourceFile {
|
|||
Ok(self._lines_slice())
|
||||
}
|
||||
|
||||
pub fn lines(&self) -> Result<impl Deref<Target = [SourceLine]> + '_, IoError> {
|
||||
self._lines()
|
||||
pub fn lines(&self) -> Result<OpaqueDerefSourceLines<'_>, IoError> {
|
||||
let lines = self._lines()?;
|
||||
|
||||
Ok(OpaqueDerefSourceLines(lines))
|
||||
}
|
||||
|
||||
pub fn line(&self, line: Line) -> Result<impl Deref<Target = SourceLine> + '_, IoError> {
|
||||
pub fn line(&self, line: Line) -> Result<OpaqueDerefSourceLine<'_>, 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.
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue