diff --git a/Cargo.lock b/Cargo.lock index 821ffc66..c7bbf979 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2318,6 +2318,15 @@ dependencies = [ "libc", ] +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "minimal-lexical" version = "0.2.1" @@ -2490,9 +2499,26 @@ dependencies = [ "uuid", ] +[[package]] +name = "nesinit" +version = "0.1.0" +dependencies = [ + "anyhow", + "libc", + "nesprotocol", + "tokio", + "tokio-vsock", + "tracing", + "tracing-subscriber", +] + [[package]] name = "nesprotocol" version = "0.1.0" +dependencies = [ + "serde", + "serde_json", +] [[package]] name = "neswire" @@ -2687,6 +2713,7 @@ dependencies = [ "cfg-if", "cfg_aliases", "libc", + "memoffset", ] [[package]] @@ -4265,6 +4292,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-vsock" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b319ef9394889dab2e1b4f0085b45ba11d0c79dc9d1a9d1afc057d009d0f1c7" +dependencies = [ + "bytes", + "futures", + "libc", + "tokio", + "vsock", +] + [[package]] name = "tokio-websockets" version = "0.13.3" @@ -4640,6 +4680,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vsock" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba782755fc073877e567c2253c0be48e4aa9a254c232d36d3985dfae0bd5205" +dependencies = [ + "libc", + "nix", +] + [[package]] name = "walkdir" version = "2.5.0" diff --git a/Cargo.toml b/Cargo.toml index 7d4491db..90b8be6f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ members = [ "apps/nescope", "apps/nesdoctor", "apps/neshub", + "apps/nesinit", "apps/neswire", "crates/nesprotocol", ] @@ -31,6 +32,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2" tokio = { version = "1", features = ["full"] } +tokio-vsock = "0.7" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/apps/nesinit/Cargo.toml b/apps/nesinit/Cargo.toml new file mode 100644 index 00000000..615743e8 --- /dev/null +++ b/apps/nesinit/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "nesinit" +version = "0.1.0" +description = "PID 1 inside a box: reaps, shuts down in order, and runs the workload it is handed" +edition.workspace = true +license.workspace = true +repository.workspace = true + +[lib] +name = "nesinit" +path = "src/lib.rs" + +[[bin]] +name = "nesinit" +path = "src/main.rs" + +[dependencies] +anyhow.workspace = true +libc.workspace = true +tokio.workspace = true +tokio-vsock.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true + +nesprotocol = { path = "../../crates/nesprotocol", features = ["lifecycle"] } + +[dev-dependencies] +tokio = { workspace = true } diff --git a/apps/nesinit/README.md b/apps/nesinit/README.md new file mode 100644 index 00000000..0dcc43f3 --- /dev/null +++ b/apps/nesinit/README.md @@ -0,0 +1,71 @@ +## nesinit + +PID 1 inside a box. + +A microVM has no init unless something is it. Three of the jobs are nobody +else's, and this is all of them: + +- **Reaping.** A process whose parent dies is reparented to PID 1. Without a + reaper, every orphan the workload leaves behind holds a pid and a slot in the + process table until the guest is gone. +- **Ordered shutdown.** The workload stops first and alone, then everything + else, then the disks are flushed and the machine is powered off. An init that + returns leaves a guest running with nothing in it. +- **The guest end of the control channel.** One vsock connection out, carrying + what to run in and what happened back. + +It does not know what it is running. It is handed a command line, a set of +shares and what an exit means; there is no code path here that branches on +which workload it started, and there is not meant to be. + +### The channel + +The guest dials out on a fixed vsock port and speaks first: + +``` +guest → { "type": "ready", "protocol_version": 2 } +guest ← { "type": "boot", "exec": {...}, "mounts": [...], "geometry": {...}, "on_exit": {...} } +guest → { "type": "workload_exited", "exit_code": 0 } +``` + +Newline-delimited JSON. Dialling out rather than being connected to is worth +keeping for two reasons: the listener is up before the VM starts, so nothing +races a booting kernel and nothing has to retry, and the connection +establishing is itself the liveness signal — without it the far end needs a +timeout to tell a slow boot from a dead one. + +The version goes out before anything is read, so a peer that cannot talk to +this build refuses it before handing over a descriptor rather than failing +later on a field that turned out to be missing. + +The types are in [`nesprotocol::lifecycle`](../../crates/nesprotocol/src/lifecycle.rs), +behind the `lifecycle` feature, so both ends of the channel read one definition +and neither can drift from it silently. + +### It reports; it does not supervise + +When the workload ends, the exit goes up the channel and the session is over. +`on_exit` says what that exit *means* — whether it ends the session — and +nothing here restarts anything. Starting something again is a decision for the +end that can see whether restarting is repair or a loop. + +A signalled workload is reported as signalled, with no exit code. Reporting +`0` for a killed process would make a kill look like a clean run. + +### What is not here yet + +Mounting shares. The descriptor's `mounts` are refused rather than ignored — a +workload started without the shares it was promised fails later, somewhere +else, for a reason nobody can see from the guest. + +### Testing + +``` +cargo test -p nesinit +``` + +No VM required, and that is the point of the two seams. Reaping is tested +against real forked children — `PR_SET_CHILD_SUBREAPER` makes a test process +inherit orphans the same way PID 1 does — and the channel is tested over an +in-memory pipe, because the transport contributes nothing to the protocol +beyond ordering and framing. diff --git a/apps/nesinit/src/lib.rs b/apps/nesinit/src/lib.rs new file mode 100644 index 00000000..2ff5eecb --- /dev/null +++ b/apps/nesinit/src/lib.rs @@ -0,0 +1,14 @@ +// PID 1 inside a box. +// +// A microVM has no init unless something is it, and three of the jobs are +// nobody else's: reaping whatever the workload orphans, turning a signal into +// an ordered shutdown, and being the guest end of the one channel out. +// +// It does not know what it is running. It is handed a command, a set of shares +// and what an exit means, and it carries that out; a field that only makes +// sense for one kind of workload cannot reach it. ref(d-0033) + +pub mod reap; +pub mod session; +pub mod shutdown; +pub mod workload; diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs new file mode 100644 index 00000000..9040150f --- /dev/null +++ b/apps/nesinit/src/main.rs @@ -0,0 +1,190 @@ +// nesinit: PID 1 inside a box. +// +// Three jobs, in the order they matter: reap what the workload orphans, run +// the workload the channel describes, and turn the end of either into an +// ordered shutdown. + +use std::time::Duration; + +use nesinit::reap::{self, Waiters}; +use nesinit::session::{self, Outcome}; +use nesinit::shutdown::{self, Machine}; +use nesinit::workload::{Process, Workload}; +use nesprotocol::lifecycle::CONTROL_PORT; +use tokio::signal::unix::{SignalKind, signal}; +use tokio_vsock::{VMADDR_CID_HOST, VsockAddr, VsockStream}; + +/// How long a process gets between being asked to stop and being made to. +const GRACE: Duration = Duration::from_secs(10); + +fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()), + ) + .init(); + + // Both before anything is started, so nothing can be orphaned or scored + // in the window where neither is true yet. + if let Err(error) = reap::become_subreaper() { + tracing::warn!(%error, "orphans may not be reaped by this process"); + } + if let Err(error) = reap::refuse_oom_kill() { + // Not fatal: outside a guest there may be no procfs to write to, and + // refusing to boot over it would be worse than the risk. + tracing::warn!(%error, "init is eligible for the OOM killer"); + } + + let pid = std::process::id(); + if pid != 1 { + tracing::warn!(pid, "not PID 1: the kernel will reparent orphans elsewhere"); + } + + let waiters = Waiters::new(); + let mut workload = Process::new(waiters.clone()); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()?; + let outcome = runtime.block_on(guest(&waiters, &mut workload)); + match &outcome { + Ok(outcome) => tracing::info!(?outcome, "the session ended"), + Err(error) => tracing::error!(%error, "the session failed"), + } + + // Before anything below waits on a pid: the reaper runs on this runtime's + // threads, and two things calling `wait` is what the registry exists to + // prevent. + drop(runtime); + + // Reached however the session ended, including an error: an init that + // returns leaves the guest running with nothing in it. + let mut machine = Guest { workload }; + shutdown::ordered(&mut machine, GRACE); + unreachable!("power_off does not return"); +} + +async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result { + // Reaping runs for as long as the guest does. A workload that leaks + // orphans leaks them while it is running, not when it stops. + tokio::spawn(reaper(waiters.clone())); + + let address = VsockAddr::new(VMADDR_CID_HOST, CONTROL_PORT); + // Dialled once, with no retry: the far end is listening before this + // machine exists, so a refused connection means something is wrong that + // waiting will not fix. + let channel = VsockStream::connect(address).await?; + + let outcome = tokio::select! { + outcome = session::run(channel, workload) => outcome?, + signal = asked_to_stop() => { + signal?; + tracing::info!("asked to stop"); + Outcome::Shutdown + } + }; + Ok(outcome) +} + +/// Drain exited children whenever the kernel says there are some, and hand +/// each exit to whoever is waiting for it. +async fn reaper(waiters: Waiters) { + let mut children = match signal(SignalKind::child()) { + Ok(children) => children, + Err(error) => { + tracing::error!(%error, "orphans will not be reaped"); + return; + } + }; + loop { + children.recv().await; + for (pid, exit) in reap::reap_exited() { + if !waiters.deliver(pid, exit) { + // An orphan nothing asked about, which is most of them. + tracing::debug!(pid, ?exit, "reaped"); + } + } + } +} + +/// A signal from outside the channel. In a guest this is the hypervisor's +/// shutdown request. +async fn asked_to_stop() -> std::io::Result<()> { + let mut term = signal(SignalKind::terminate())?; + let mut int = signal(SignalKind::interrupt())?; + tokio::select! { + _ = term.recv() => Ok(()), + _ = int.recv() => Ok(()), + } +} + +/// The machine, for real. +struct Guest { + workload: Process, +} + +impl Machine for Guest { + fn signal_workload(&mut self) { + self.workload.signal_stop(); + } + + fn await_workload(&mut self, grace: Duration) -> bool { + // The session already reported the exit if there was one; this is the + // window for a workload that was asked to stop on the way down. It + // waits for that pid and no other, or the first service to leave would + // look like the workload leaving. + self.workload.await_exit(grace) + } + + fn kill_workload(&mut self) { + // The workload alone. Everything else in the guest is still expected + // to get the ordered stop below, and a kill to every process here + // would take the services with it. + self.workload.signal(libc::SIGKILL); + self.workload.await_exit(Duration::from_secs(1)); + } + + fn signal_rest(&mut self, grace: Duration) { + // -1 is every process this one may signal, which as PID 1 is all of + // them but itself. The workload has already stopped by here. + unsafe { libc::kill(-1, libc::SIGTERM) }; + wait_for_quiet(grace); + } + + fn kill_rest(&mut self) { + unsafe { libc::kill(-1, libc::SIGKILL) }; + wait_for_quiet(Duration::from_secs(1)); + } + + fn flush_disks(&mut self) { + unsafe { libc::sync() }; + } + + fn power_off(&mut self) { + // SAFETY: reboot is the only way out of a guest whose init is done. + unsafe { libc::reboot(libc::RB_POWER_OFF) }; + // Reached only if the guest refused to power off, which no caller can + // be told about — the channel is gone by now. + std::process::exit(0); + } +} + +/// Reap until nothing is left or the deadline passes. +fn wait_for_quiet(grace: Duration) -> bool { + let deadline = std::time::Instant::now() + grace; + loop { + let mut status: libc::c_int = 0; + // Blocking on purpose: this runs after the runtime has stopped, so + // there is nothing left to keep responsive. + let pid = unsafe { libc::waitpid(-1, &mut status, libc::WNOHANG) }; + if pid == -1 { + return true; // ECHILD: nothing left to wait for + } + if std::time::Instant::now() >= deadline { + return false; + } + if pid == 0 { + std::thread::sleep(Duration::from_millis(50)); + } + } +} diff --git a/apps/nesinit/src/reap.rs b/apps/nesinit/src/reap.rs new file mode 100644 index 00000000..c85fb8e3 --- /dev/null +++ b/apps/nesinit/src/reap.rs @@ -0,0 +1,180 @@ +// Reaping. The part of being PID 1 that no other component can do. +// +// A process whose parent dies is reparented to PID 1, so every orphan in the +// guest becomes this process's child and stays a zombie until it is waited +// for. Zombies hold a pid and a slot in the process table; a workload that +// leaks them in a long session eventually cannot fork. + +use std::collections::HashMap; +use std::io; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use nesprotocol::lifecycle::Exit; +use tokio::sync::oneshot; + +/// Reap every child that has already exited, without blocking on any that +/// have not. +/// +/// Returns what it collected, which is what makes this testable: the caller +/// decides whether an exit is interesting, and the same call site both frees +/// the process table and answers "did the process I care about end". +pub fn reap_exited() -> Vec<(i32, Exit)> { + let mut reaped = Vec::new(); + loop { + let mut status: libc::c_int = 0; + // -1 is "any child"; WNOHANG makes this a poll rather than a wait, so + // one call drains the queue and never blocks the caller. + let pid = unsafe { libc::waitpid(-1, &mut status, libc::WNOHANG) }; + match pid { + 0 => return reaped, + -1 => return reaped, // no children left, or a signal interrupted the poll + pid => reaped.push((pid, exit_of(status))), + } + } +} + +/// Read a `wait` status as an exit. +/// +/// A signalled process has no exit code. Reporting `0` for one would make a +/// kill indistinguishable from a clean run, which is the difference a caller +/// most needs from this. +pub fn exit_of(status: libc::c_int) -> Exit { + if libc::WIFSIGNALED(status) { + Exit::signal(libc::WTERMSIG(status)) + } else { + Exit::code(libc::WEXITSTATUS(status)) + } +} + +/// Ask the kernel to reparent orphans to this process even when it is not +/// PID 1. +/// +/// In the guest this is redundant — PID 1 already collects them. It is called +/// anyway because it is what makes the reaper testable off a VM, and because a +/// nesinit that is accidentally not PID 1 should still reap rather than leak. +pub fn become_subreaper() -> io::Result<()> { + // SAFETY: prctl with this option takes one integer argument and returns + // -1/errno on failure; nothing here is borrowed by the kernel. + if unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// Take this process out of the reach of the OOM killer. +/// +/// Under memory pressure the kernel picks a victim by score, and init being +/// eligible is the one loss the guest cannot report: everything else exiting +/// is a message up the channel, whereas init exiting takes the channel with +/// it, and the caller sees a box that stopped answering for no stated reason. +/// +/// This covers init only. Making the workload the preferred victim is the +/// image's job — this process cannot score other processes it did not start. +pub fn refuse_oom_kill() -> io::Result<()> { + std::fs::write("/proc/self/oom_score_adj", "-1000\n") +} + +/// The children whose exit somebody is waiting for. +/// +/// Reaping and waiting cannot be two mechanisms. `waitpid(-1, ...)` collects +/// any child, so a reaper running beside a caller that waits on its own child +/// will sometimes collect that child first and the caller's wait then fails +/// with no status — the exit is gone, and an exit is the one thing this +/// component exists to report. So the reaper is the only waiter, and this is +/// how it hands an exit to whoever asked for one. +#[derive(Clone, Default)] +pub struct Waiters(Arc>>); + +/// The registry's half of one watched child. +struct Watch { + exit: oneshot::Sender, + running: Arc, +} + +/// A child something is waiting for: its pid, its exit, and whether the pid +/// still means what it meant. +/// +/// The flag is the point. A pid is only a name for a process until that +/// process is reaped, after which the kernel may hand the same number to +/// something else — so a caller that kept a pid and signals it later can hit a +/// process it has never heard of. Anything that signals asks here first. +pub struct Watched { + pub pid: i32, + running: Arc, + exit: Option>, +} + +impl Watched { + /// The exit, which only one caller may hold — whoever takes it is the one + /// that reports it. What is left behind is the pid and whether it still + /// means this child, which is what a signaller needs. + pub fn take_exit(&mut self) -> Option> { + self.exit.take() + } + + /// Whether the pid is still this child's. + /// + /// False from the moment its exit is delivered. There is a window between + /// the reap and the delivery in which this still says true; closing it + /// entirely needs a handle the kernel keeps for us rather than a number, + /// and the number is what the rest of this component has. + pub fn running(&self) -> bool { + self.running.load(Ordering::SeqCst) + } +} + +impl Waiters { + pub fn new() -> Self { + Self::default() + } + + /// Start something and be waiting for it before it can be reaped. + /// + /// The lock is held across the spawn, and that is the whole point: a child + /// that exits immediately is reaped by a thread that has to take the same + /// lock to deliver the exit, so it waits until the slot exists rather than + /// finding none and dropping it. + pub fn watch(&self, spawn: F) -> io::Result + where + F: FnOnce() -> io::Result, + { + let mut waiting = self.lock(); + let pid = spawn()?; + let (sender, receiver) = oneshot::channel(); + let running = Arc::new(AtomicBool::new(true)); + waiting.insert( + pid, + Watch { + exit: sender, + running: running.clone(), + }, + ); + Ok(Watched { + pid, + running, + exit: Some(receiver), + }) + } + + /// Hand an exit to whoever is waiting for that pid. `false` if nobody is, + /// which is the common case: most of what an init reaps is an orphan + /// nothing asked about. + pub fn deliver(&self, pid: i32, exit: Exit) -> bool { + let Some(watch) = self.lock().remove(&pid) else { + return false; + }; + // Before the exit goes anywhere: the pid has been reaped, so the + // number is free for the kernel to give to something else and nothing + // may signal it from here on. + watch.running.store(false, Ordering::SeqCst); + // An error here means the caller stopped waiting, which is its right. + watch.exit.send(exit).is_ok() + } + + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.0 + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs new file mode 100644 index 00000000..e0e21af1 --- /dev/null +++ b/apps/nesinit/src/session.rs @@ -0,0 +1,408 @@ +// The guest end of the control channel. +// +// The shape of the exchange, and none of it is negotiable from this side: the +// guest speaks first with its version, is handed one boot descriptor, and from +// then on reports. It is not a supervisor: when the workload ends, the exit +// goes up the channel and this returns. Starting something again is the +// caller's decision, because the caller is the only end that can see whether +// restarting is repair or a loop. ref(d-0033) + +use nesprotocol::lifecycle::{ + BootDescriptor, CONTROL_VERSION, Exit, GuestToHost, HostToGuest, from_line, to_line, +}; +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; + +use crate::workload::{Failure, Workload}; + +/// How a session ended. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Outcome { + /// The workload ended and the exit was reported. + WorkloadExited(Exit), + /// The caller asked for a shutdown. + Shutdown, + /// The channel closed under us. Not an error by itself — a caller that has + /// stopped listening has also stopped being able to tell us to stop. + ChannelClosed, + /// The descriptor could not be carried out. The reason is the operating + /// system's, verbatim. + Refused(Failure), +} + +/// Run one session over an already-connected channel. +/// +/// Generic over the channel so the exchange can be driven from a test without +/// a VM: the transport contributes nothing to the protocol beyond ordering and +/// framing, which any byte stream has. +pub async fn run(channel: C, workload: &mut W) -> std::io::Result +where + C: AsyncRead + AsyncWrite, + W: Workload, +{ + let (reader, mut writer) = tokio::io::split(channel); + let mut lines = BufReader::new(reader).lines(); + + // First line on the connection, before anything is read. The version is + // here rather than in a round trip because the caller has to be able to + // refuse a guest it cannot talk to before it hands over a descriptor. + send( + &mut writer, + &GuestToHost::Ready { + protocol_version: CONTROL_VERSION, + }, + ) + .await?; + + let mut running: Option = None; + + loop { + let line = match running.as_mut() { + Some(exited) => tokio::select! { + ended = exited => { + let exit = ended?; + send(&mut writer, &GuestToHost::WorkloadExited { exit }).await?; + return Ok(Outcome::WorkloadExited(exit)); + } + line = lines.next_line() => line?, + }, + None => lines.next_line().await?, + }; + + let Some(line) = line else { + // The far end is gone. Stop the workload rather than leave it + // running with nobody to report to. + workload.signal_stop(); + return Ok(Outcome::ChannelClosed); + }; + + let message: HostToGuest = match from_line(&line) { + Ok(message) => message, + Err(error) => { + // Skipped rather than fatal: a line this build does not + // understand is not a reason to end a running session, and the + // version handshake is what catches a peer we cannot talk to. + tracing::warn!(%error, "ignoring an unreadable line"); + continue; + } + }; + + match message { + HostToGuest::Boot { descriptor } => { + if running.is_some() { + tracing::warn!("ignoring a second descriptor: one is read per connection"); + continue; + } + match begin(&descriptor, workload) { + Ok(exited) => running = Some(exited), + Err(failure) => { + tracing::error!(reason = %failure.reason, "the descriptor was refused"); + return Ok(Outcome::Refused(failure)); + } + } + } + HostToGuest::Stop => workload.signal_stop(), + HostToGuest::Shutdown => return Ok(Outcome::Shutdown), + } + } +} + +/// Carry out a descriptor: shares first, then the command. +/// +/// The two stay distinguishable on the way out because they want different +/// things looked at — a share that did not mount and a command that did not +/// start are not the same incident. +fn begin( + descriptor: &BootDescriptor, + workload: &mut W, +) -> Result { + workload.mount(&descriptor.mounts)?; + workload.start(&descriptor.exec) +} + +async fn send(writer: &mut W, message: &GuestToHost) -> std::io::Result<()> +where + W: AsyncWrite + Unpin, +{ + let line = to_line(message).map_err(std::io::Error::other)?; + writer.write_all(line.as_bytes()).await?; + writer.flush().await +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::workload::double::Double; + use nesprotocol::lifecycle::{Exec, Geometry, Mount, OnExit}; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream}; + + fn descriptor() -> BootDescriptor { + BootDescriptor { + exec: Exec { + argv: vec!["/usr/bin/workload".into(), "--windowed".into()], + env: Default::default(), + cwd: None, + uid: 1000, + gid: 1000, + }, + mounts: vec![Mount { + tag: "user".into(), + at: "/mnt/user".into(), + ro: false, + }], + geometry: Geometry { + width: 1920, + height: 1080, + fps: 60, + hdr: false, + }, + on_exit: OnExit { terminal: true }, + } + } + + /// The other end of the channel, as a caller would drive it. + struct Caller { + lines: tokio::io::Lines>, + } + + impl Caller { + fn new(stream: DuplexStream) -> Self { + Self { + lines: BufReader::new(stream).lines(), + } + } + + async fn expect(&mut self) -> GuestToHost { + let line = self + .lines + .next_line() + .await + .unwrap() + .expect("the guest said nothing"); + from_line(&line).unwrap() + } + + async fn say(&mut self, message: &HostToGuest) { + let line = to_line(message).unwrap(); + self.lines + .get_mut() + .write_all(line.as_bytes()) + .await + .unwrap(); + } + } + + #[tokio::test] + async fn the_guest_speaks_first_and_says_its_version() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_when_stopped(Exit::code(0)); + let outcome = run(guest, &mut workload).await.unwrap(); + (outcome, workload) + }); + + // Nothing has been sent to the guest, so this can only be unprompted. + assert_eq!( + caller.expect().await, + GuestToHost::Ready { + protocol_version: 2 + } + ); + + caller.say(&HostToGuest::Shutdown).await; + let (outcome, _) = session.await.unwrap(); + assert_eq!(outcome, Outcome::Shutdown); + } + + #[tokio::test] + async fn the_descriptor_mounts_and_starts_what_it_names() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_when_stopped(Exit::code(0)); + let outcome = run(guest, &mut workload).await.unwrap(); + (outcome, workload) + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + caller.say(&HostToGuest::Stop).await; + + let (outcome, workload) = session.await.unwrap(); + assert_eq!(outcome, Outcome::WorkloadExited(Exit::code(0))); + assert_eq!(workload.mounted, vec![descriptor().mounts]); + assert_eq!(workload.started, vec![descriptor().exec]); + } + + #[tokio::test] + async fn an_exit_is_reported_and_the_workload_is_not_started_again() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_at_once(Exit::code(3)); + let outcome = run(guest, &mut workload).await.unwrap(); + (outcome, workload) + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + + assert_eq!( + caller.expect().await, + GuestToHost::WorkloadExited { + exit: Exit::code(3) + }, + ); + + let (outcome, workload) = session.await.unwrap(); + assert_eq!(outcome, Outcome::WorkloadExited(Exit::code(3))); + assert_eq!( + workload.started.len(), + 1, + "an exit is reported, never restarted" + ); + } + + #[tokio::test] + async fn a_signalled_workload_is_reported_as_signalled() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_at_once(Exit::signal(9)); + run(guest, &mut workload).await.unwrap() + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + + assert_eq!( + caller.expect().await, + GuestToHost::WorkloadExited { + exit: Exit::signal(9) + }, + ); + assert_eq!( + session.await.unwrap(), + Outcome::WorkloadExited(Exit::signal(9)) + ); + } + + #[tokio::test] + async fn a_stop_is_idempotent_and_does_not_end_the_session() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_when_stopped(Exit::code(0)); + let outcome = run(guest, &mut workload).await.unwrap(); + (outcome, workload) + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + // No descriptor yet, so there is nothing to stop and the session has + // to survive being told to anyway. + caller.say(&HostToGuest::Stop).await; + caller.say(&HostToGuest::Stop).await; + caller.say(&HostToGuest::Shutdown).await; + + let (outcome, workload) = session.await.unwrap(); + assert_eq!(outcome, Outcome::Shutdown); + assert_eq!(workload.stops, 2); + assert!(workload.started.is_empty()); + } + + #[tokio::test] + async fn a_share_that_will_not_mount_is_refused_before_anything_starts() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_at_once(Exit::code(0)); + workload.mount_failure = Some(Failure::new("EACCES: /mnt/user")); + let outcome = run(guest, &mut workload).await.unwrap(); + (outcome, workload) + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + + let (outcome, workload) = session.await.unwrap(); + assert_eq!(outcome, Outcome::Refused(Failure::new("EACCES: /mnt/user"))); + assert!( + workload.started.is_empty(), + "a workload without its shares is not started" + ); + } + + #[tokio::test] + async fn an_unreadable_line_does_not_end_a_session() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_when_stopped(Exit::code(0)); + run(guest, &mut workload).await.unwrap() + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .lines + .get_mut() + .write_all(b"{\"type\":\"from_a_later_version\"}\n") + .await + .unwrap(); + caller.say(&HostToGuest::Shutdown).await; + + assert_eq!(session.await.unwrap(), Outcome::Shutdown); + } + + #[tokio::test] + async fn a_closed_channel_stops_the_workload() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let mut workload = Double::exits_when_stopped(Exit::code(0)); + let outcome = run(guest, &mut workload).await.unwrap(); + (outcome, workload) + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + drop(caller); + + let (outcome, workload) = session.await.unwrap(); + assert!( + matches!(outcome, Outcome::ChannelClosed | Outcome::WorkloadExited(_)), + "unexpected outcome: {outcome:?}", + ); + assert!( + workload.stops >= 1, + "the workload was left running with nobody listening" + ); + } +} diff --git a/apps/nesinit/src/shutdown.rs b/apps/nesinit/src/shutdown.rs new file mode 100644 index 00000000..d5507783 --- /dev/null +++ b/apps/nesinit/src/shutdown.rs @@ -0,0 +1,135 @@ +// Ordered shutdown: the second job that is nobody else's. +// +// The order is the whole content of this module. The workload goes first and +// alone, because it is the only process whose exit anyone is waiting to hear +// about; everything else goes after, so a service is never killed while the +// workload still needs it. Then the disks are flushed and the machine is +// powered off, because a guest whose init returns is a guest that hangs. + +use std::time::Duration; + +/// What an ordered shutdown does to the machine, behind a trait so the order +/// can be asserted without a VM and without root. +pub trait Machine { + /// Ask the workload to stop. + fn signal_workload(&mut self); + /// Wait up to `grace` for the workload to leave. `true` if it did. + fn await_workload(&mut self, grace: Duration) -> bool; + /// Stop waiting. + fn kill_workload(&mut self); + /// Ask every remaining process to stop, then wait up to `grace`. + fn signal_rest(&mut self, grace: Duration); + /// Stop waiting for the rest. + fn kill_rest(&mut self); + fn flush_disks(&mut self); + fn power_off(&mut self); +} + +/// Run the shutdown, in order, and do not return. +/// +/// A workload that leaves inside its grace period is never killed: an exit +/// code that says "asked to stop" is worth more to whoever reads the report +/// than one that says "killed", and some workloads only save on the way out. +pub fn ordered(machine: &mut M, grace: Duration) { + machine.signal_workload(); + if !machine.await_workload(grace) { + machine.kill_workload(); + } + machine.signal_rest(grace); + machine.kill_rest(); + machine.flush_disks(); + machine.power_off(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Default)] + struct Recorder { + steps: Vec<&'static str>, + workload_leaves: bool, + } + + impl Machine for Recorder { + fn signal_workload(&mut self) { + self.steps.push("signal_workload"); + } + fn await_workload(&mut self, _grace: Duration) -> bool { + self.steps.push("await_workload"); + self.workload_leaves + } + fn kill_workload(&mut self) { + self.steps.push("kill_workload"); + } + fn signal_rest(&mut self, _grace: Duration) { + self.steps.push("signal_rest"); + } + fn kill_rest(&mut self) { + self.steps.push("kill_rest"); + } + fn flush_disks(&mut self) { + self.steps.push("flush_disks"); + } + fn power_off(&mut self) { + self.steps.push("power_off"); + } + } + + #[test] + fn the_workload_stops_before_anything_else_and_the_disks_flush_before_power() { + let mut machine = Recorder { + workload_leaves: true, + ..Default::default() + }; + ordered(&mut machine, Duration::from_secs(5)); + + assert_eq!( + machine.steps, + vec![ + "signal_workload", + "await_workload", + "signal_rest", + "kill_rest", + "flush_disks", + "power_off", + ], + ); + } + + #[test] + fn a_workload_that_leaves_in_time_is_not_killed() { + let mut machine = Recorder { + workload_leaves: true, + ..Default::default() + }; + ordered(&mut machine, Duration::from_secs(5)); + assert!(!machine.steps.contains(&"kill_workload")); + } + + #[test] + fn a_workload_that_overstays_its_grace_is_killed_and_shutdown_still_finishes() { + let mut machine = Recorder { + workload_leaves: false, + ..Default::default() + }; + ordered(&mut machine, Duration::from_secs(5)); + + let killed = machine + .steps + .iter() + .position(|s| *s == "kill_workload") + .unwrap(); + let rest = machine + .steps + .iter() + .position(|s| *s == "signal_rest") + .unwrap(); + assert!( + killed < rest, + "the rest of the guest outlives the workload: {:?}", + machine.steps + ); + assert_eq!(machine.steps.last(), Some(&"power_off")); + } +} diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs new file mode 100644 index 00000000..0ef5433f --- /dev/null +++ b/apps/nesinit/src/workload.rs @@ -0,0 +1,267 @@ +// The one thing a descriptor turns into: mount what it names, run what it +// names, report how that ended. +// +// It is a trait because the two halves of it arrive at different times and +// because the interesting behaviour — one start and never a second, an exit +// reported rather than acted on — is behaviour of the caller, which a double +// can test without a VM, a share or a workload. + +use std::future::Future; +use std::io; +use std::pin::Pin; + +use nesprotocol::lifecycle::{Exec, Exit, Mount}; + +use std::os::unix::process::CommandExt; + +use crate::reap::{Waiters, Watched}; + +/// Why something could not be done, in the words the operating system used. +/// +/// The reason is passed through verbatim on purpose: `EACCES` and a path can +/// be acted on, where "could not start the workload" cannot. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Failure { + pub reason: String, +} + +impl Failure { + pub fn new(reason: impl Into) -> Self { + Self { + reason: reason.into(), + } + } +} + +/// Resolves when the workload ends. +pub type Exited = Pin> + Send>>; + +pub trait Workload { + /// Make the shares the descriptor names, where it says to put them. + fn mount(&mut self, mounts: &[Mount]) -> Result<(), Failure>; + + /// Start the command the descriptor names. + /// + /// Returning the exit as a future, rather than a `wait` method, is what + /// keeps the caller free to read the channel while the workload runs — a + /// stop has to arrive during the workload's life or it is not a stop. + fn start(&mut self, exec: &Exec) -> Result; + + /// Ask the workload to stop. Idempotent, and never ends the session by + /// itself. + fn signal_stop(&mut self); +} + +/// The workload as a local process. +pub struct Process { + waiters: Waiters, + running: Option, +} + +impl Process { + /// Started through the reaper's registry, because the reaper is the only + /// thing in this component that may call `wait`. + pub fn new(waiters: Waiters) -> Self { + Self { + waiters, + running: None, + } + } + + /// Send a signal to the workload and nothing else. + /// + /// Nothing here ever signals the process group or every process: the order + /// a shutdown promises is only true if the workload can be stopped alone. + pub fn signal(&self, signal: libc::c_int) { + let Some(watched) = &self.running else { return }; + // Nothing is signalled once the exit has been delivered. The pid was + // freed by the reap that produced it, and the kernel is entitled to + // give that number to something else — a stop or a kill aimed at it + // then lands on a process nobody meant. + if !watched.running() { + return; + } + // SAFETY: two integers, and it cannot touch this process's memory. A + // pid that has already gone fails with ESRCH, which is exactly the + // idempotence the callers of this rely on. + unsafe { libc::kill(watched.pid, signal) }; + } + + /// Wait up to `grace` for the workload to leave, without a runtime. + /// + /// Used on the way down, after the reaper has stopped: it polls for the + /// one pid rather than for any child, so a slow service cannot be mistaken + /// for the workload still running. + pub fn await_exit(&self, grace: std::time::Duration) -> bool { + let Some(watched) = &self.running else { + return true; + }; + if !watched.running() { + return true; // already reaped, and its exit already reported + } + let pid = watched.pid; + let deadline = std::time::Instant::now() + grace; + loop { + let mut status: libc::c_int = 0; + // SAFETY: waitpid writes only into `status`. + let seen = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + // -1 is ECHILD: already reaped, which is also gone. + if seen == pid || seen == -1 { + return true; + } + if std::time::Instant::now() >= deadline { + return false; + } + std::thread::sleep(std::time::Duration::from_millis(50)); + } + } +} + +impl Workload for Process { + fn mount(&mut self, mounts: &[Mount]) -> Result<(), Failure> { + if mounts.is_empty() { + return Ok(()); + } + // Refused rather than ignored: a workload started without the shares + // it was promised fails later, somewhere else, for a reason nobody can + // see from here. + Err(Failure::new(format!( + "this build mounts nothing; {} share(s) were requested", + mounts.len() + ))) + } + + fn start(&mut self, exec: &Exec) -> Result { + let Some((program, args)) = exec.argv.split_first() else { + return Err(Failure::new("the command is empty")); + }; + + // The standard library's process rather than the runtime's: the + // runtime reaps the children it spawns, and in this component reaping + // belongs to one place. See `reap::Waiters`. + let mut command = std::process::Command::new(program); + command.args(args); + // Cleared rather than inherited: init's environment is the kernel's + // and says nothing a workload should read. + command.env_clear(); + command.envs(&exec.env); + if let Some(cwd) = &exec.cwd { + command.current_dir(cwd); + } + + let (uid, gid) = (exec.uid, exec.gid); + // SAFETY: the closure runs between fork and exec in the child, where + // only async-signal-safe calls are allowed. These two are, and it + // allocates nothing. + unsafe { + command.pre_exec(move || { + // gid first: dropping the uid first would lose the privilege + // needed to set the gid at all. + if libc::setgid(gid) != 0 { + return Err(io::Error::last_os_error()); + } + if libc::setuid(uid) != 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) + }); + } + + let mut watched = self + .waiters + .watch(|| Ok(command.spawn()?.id() as i32)) + .map_err(|error| Failure::new(error.to_string()))?; + + // The caller gets the exit and reports it; this handle keeps the pid + // and whether that pid is still this child's. + let exit = watched.take_exit().expect("a new watch has its exit"); + self.running = Some(watched); + + Ok(Box::pin(async move { + exit.await + .map_err(|_| io::Error::other("the workload's exit was not delivered")) + })) + } + + fn signal_stop(&mut self) { + self.signal(libc::SIGTERM); + } +} + +#[cfg(test)] +pub mod double { + use super::*; + use tokio::sync::oneshot; + + /// A workload that starts nothing, so what the caller does with it is the + /// only thing under test. + pub struct Double { + pub mounted: Vec>, + pub started: Vec, + pub stops: usize, + pub mount_failure: Option, + pub start_failure: Option, + exit: Exit, + on_stop: Option>, + holds_until_stopped: bool, + } + + impl Double { + /// Its workload has already ended by the time it is started. + pub fn exits_at_once(exit: Exit) -> Self { + Self::new(exit, false) + } + + /// Its workload runs until it is asked to stop. + pub fn exits_when_stopped(exit: Exit) -> Self { + Self::new(exit, true) + } + + fn new(exit: Exit, holds_until_stopped: bool) -> Self { + Self { + mounted: Vec::new(), + started: Vec::new(), + stops: 0, + mount_failure: None, + start_failure: None, + exit, + on_stop: None, + holds_until_stopped, + } + } + } + + impl Workload for Double { + fn mount(&mut self, mounts: &[Mount]) -> Result<(), Failure> { + self.mounted.push(mounts.to_vec()); + match &self.mount_failure { + Some(failure) => Err(failure.clone()), + None => Ok(()), + } + } + + fn start(&mut self, exec: &Exec) -> Result { + self.started.push(exec.clone()); + if let Some(failure) = &self.start_failure { + return Err(failure.clone()); + } + let exit = self.exit; + if !self.holds_until_stopped { + return Ok(Box::pin(async move { Ok(exit) })); + } + let (tx, rx) = oneshot::channel(); + self.on_stop = Some(tx); + Ok(Box::pin(async move { + rx.await + .map_err(|_| io::Error::other("the workload was dropped")) + })) + } + + fn signal_stop(&mut self) { + self.stops += 1; + if let Some(tx) = self.on_stop.take() { + let _ = tx.send(self.exit); + } + } + } +} diff --git a/apps/nesinit/tests/reaping.rs b/apps/nesinit/tests/reaping.rs new file mode 100644 index 00000000..66f0a30b --- /dev/null +++ b/apps/nesinit/tests/reaping.rs @@ -0,0 +1,217 @@ +// Reaping, against real processes. +// +// Its own test binary on purpose: the reaper waits on any child, so it would +// collect processes another test in the same binary was waiting for. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Duration, Instant}; + +use nesinit::reap::{Waiters, become_subreaper, reap_exited}; + +/// Reaping is process-wide: `waitpid(-1, ...)` collects any child, so two of +/// these tests running at once would each reap the other's. One at a time. +static ONE_REAPER: Mutex<()> = Mutex::new(()); + +fn alone() -> MutexGuard<'static, ()> { + ONE_REAPER + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// Fork a child, run `body` in it, and never return from the child. +/// +/// Raw fork rather than `Command` because the point is a child nothing else +/// holds a handle to — the standard library reaps the children it spawns, +/// which is precisely the work under test. +fn fork_child(body: impl FnOnce()) -> i32 { + let pid = unsafe { libc::fork() }; + assert!(pid >= 0, "fork failed: {}", std::io::Error::last_os_error()); + if pid == 0 { + body(); + unsafe { libc::_exit(0) }; + } + pid +} + +/// Reap until `pid` turns up, or give up. +fn reap_until(pid: i32) -> Option { + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + for (reaped, exit) in reap_exited() { + if reaped == pid { + return Some(exit); + } + } + std::thread::sleep(Duration::from_millis(10)); + } + None +} + +#[test] +fn a_child_that_exits_is_reaped_with_its_code() { + let _alone = alone(); + let pid = fork_child(|| unsafe { libc::_exit(7) }); + let exit = reap_until(pid).expect("the child was left a zombie"); + assert_eq!(exit.exit_code, Some(7)); + assert_eq!(exit.signal, None); +} + +#[test] +fn a_child_that_is_killed_is_reaped_as_signalled() { + let _alone = alone(); + let pid = fork_child(|| { + // Sleep long enough to be killed rather than to exit on its own. + unsafe { libc::pause() }; + }); + assert_eq!(unsafe { libc::kill(pid, libc::SIGKILL) }, 0); + + let exit = reap_until(pid).expect("the child was left a zombie"); + assert_eq!(exit.signal, Some(libc::SIGKILL)); + assert_eq!(exit.exit_code, None, "a killed process has no exit code"); +} + +#[test] +fn an_orphan_is_reaped_by_whoever_inherits_it() { + let _alone = alone(); + become_subreaper().expect("PR_SET_CHILD_SUBREAPER"); + + // A grandchild that outlives its parent. In a guest the kernel hands it to + // PID 1; here the same reparenting is arranged with the subreaper bit, so + // the reaper is exercised rather than the privilege. + let (read_fd, write_fd) = pipe(); + let child = fork_child(|| { + let grandchild = unsafe { libc::fork() }; + if grandchild == 0 { + // Outlive the parent, then exit with a code the test can pick out. + std::thread::sleep(Duration::from_millis(200)); + unsafe { libc::_exit(11) }; + } + let pid = grandchild.to_le_bytes(); + unsafe { libc::write(write_fd, pid.as_ptr().cast(), pid.len()) }; + unsafe { libc::_exit(0) }; + }); + + let mut buf = [0u8; 4]; + let read = unsafe { libc::read(read_fd, buf.as_mut_ptr().cast(), buf.len()) }; + assert_eq!(read, 4, "the child never reported its own child"); + let grandchild = i32::from_le_bytes(buf); + + // The parent goes first; the orphan is what this test is about. + assert!(reap_until(child).is_some(), "the child was left a zombie"); + + let exit = reap_until(grandchild).expect("the orphan was left a zombie"); + assert_eq!(exit.exit_code, Some(11)); +} + +fn pipe() -> (i32, i32) { + let mut fds = [0i32; 2]; + assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0); + (fds[0], fds[1]) +} + +#[test] +fn an_exit_reaches_whoever_asked_for_it() { + let _alone = alone(); + let waiters = Waiters::new(); + + let mut watched = waiters + .watch(|| Ok(fork_child(|| unsafe { libc::_exit(9) }))) + .expect("the child never started"); + let mut exited = watched.take_exit().unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + for (pid, exit) in reap_exited() { + waiters.deliver(pid, exit); + } + if let Ok(exit) = exited.try_recv() { + assert_eq!(exit.exit_code, Some(9)); + return; + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!("the exit was reaped by the reaper and never handed on"); +} + +#[test] +fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() { + let _alone = alone(); + let waiters = Waiters::new(); + + // A reaper already running, as it is in the guest: it will collect this + // child before `watch` has finished registering interest in it. + let stop = Arc::new(AtomicBool::new(false)); + let reaper = std::thread::spawn({ + let waiters = waiters.clone(); + let stop = stop.clone(); + move || { + while !stop.load(Ordering::Relaxed) { + for (pid, exit) in reap_exited() { + waiters.deliver(pid, exit); + } + std::thread::sleep(Duration::from_millis(1)); + } + } + }); + + let mut watched = waiters + .watch(|| { + let pid = fork_child(|| unsafe { libc::_exit(5) }); + // Long enough for the child to exit and the reaper to reach it. + std::thread::sleep(Duration::from_millis(200)); + Ok(pid) + }) + .expect("the child never started"); + let mut exited = watched.take_exit().unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + let exit = loop { + if let Ok(exit) = exited.try_recv() { + break exit; + } + assert!( + Instant::now() < deadline, + "the exit was dropped on the way through" + ); + std::thread::sleep(Duration::from_millis(10)); + }; + assert_eq!(exit.exit_code, Some(5)); + + stop.store(true, Ordering::Relaxed); + reaper.join().unwrap(); +} + +#[test] +fn a_pid_stops_being_the_workload_the_moment_it_is_reaped() { + let _alone = alone(); + let waiters = Waiters::new(); + + let mut watched = waiters + .watch(|| Ok(fork_child(|| unsafe { libc::_exit(0) }))) + .expect("the child never started"); + let mut exited = watched.take_exit().unwrap(); + assert!( + watched.running(), + "a child that has not been reaped is still itself" + ); + + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + for (pid, exit) in reap_exited() { + waiters.deliver(pid, exit); + } + if exited.try_recv().is_ok() { + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + + // The pid was freed by the reap that produced that exit, and the kernel is + // entitled to hand the number to something else. Signalling it after this + // point is signalling a stranger. + assert!( + !watched.running(), + "a reaped pid is still being treated as the workload's", + ); +} diff --git a/crates/nesprotocol/Cargo.toml b/crates/nesprotocol/Cargo.toml index 057a8f0b..4d036fa6 100644 --- a/crates/nesprotocol/Cargo.toml +++ b/crates/nesprotocol/Cargo.toml @@ -5,3 +5,13 @@ description = "Wire types shared by the capture, audio and compositor components edition.workspace = true license.workspace = true repository.workspace = true + +[features] +# The lifecycle layer of the guest's control channel. Off by default: the media +# components have no use for it and keeping it optional keeps their build free +# of serde. +lifecycle = ["dep:serde", "dep:serde_json"] + +[dependencies] +serde = { workspace = true, optional = true } +serde_json = { workspace = true, optional = true } diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs index 799d14f3..f4bfd1ec 100644 --- a/crates/nesprotocol/src/lib.rs +++ b/crates/nesprotocol/src/lib.rs @@ -4,6 +4,8 @@ pub mod datagram; pub mod input; +#[cfg(feature = "lifecycle")] +pub mod lifecycle; pub mod reliable; pub mod stats; diff --git a/crates/nesprotocol/src/lifecycle.rs b/crates/nesprotocol/src/lifecycle.rs new file mode 100644 index 00000000..f08448ea --- /dev/null +++ b/crates/nesprotocol/src/lifecycle.rs @@ -0,0 +1,267 @@ +// The lifecycle layer of the control channel between a box and whatever runs +// it: the boot descriptor the guest is handed, and what the guest says back +// about carrying it out. +// +// It lives beside the media types for the same reason they live here — one +// definition, so the two ends cannot drift from each other silently. +// +// Nothing in this module describes *what* the guest runs. A command line, a +// set of share tags, an output geometry, and what an exit means: that is the +// whole vocabulary, and a field that only makes sense for one kind of workload +// does not belong in it. ref(d-0033) +// +// The channel also carries a second layer, which the guest relays as opaque +// bytes and never parses. Those types land with the relay that needs them. + +use std::collections::BTreeMap; + +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +/// The vsock port the guest dials. +/// +/// The guest dials out rather than being connected to, which is worth keeping +/// for two reasons: the listener is up before the VM starts, so nothing races a +/// booting kernel and nothing has to retry; and the connection establishing is +/// itself the liveness signal, without which a caller needs a timeout to tell a +/// slow boot from a dead one. +pub const CONTROL_PORT: u32 = 7000; + +/// Version of this layer. Both ends compare it during the handshake and refuse +/// on mismatch, so a guest built against one version meeting a caller built +/// against another fails immediately and legibly, rather than later on a field +/// that turned out to be missing. +/// +/// Adding a variant or a field does not need a bump; removing or renaming one +/// does. +pub const CONTROL_VERSION: u32 = 2; + +/// The command to run, and who runs it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Exec { + /// The program and its arguments. Never a shell string: a guest that splits + /// words is a guest that can split them differently than the caller meant. + pub argv: Vec, + /// Environment for the process. Sorted, so two descriptors that say the + /// same thing serialize identically. + #[serde(default)] + pub env: BTreeMap, + /// Working directory. `None` means the root of the guest filesystem. + #[serde(default)] + pub cwd: Option, + /// The uid and gid to drop to before exec. + /// + /// These are load-bearing rather than hygiene. Whoever writes this + /// descriptor is also whoever exported the writable share, so the ids have + /// to agree; when they do not, the share refuses the first write and the + /// failure surfaces here as `EACCES` with a path, instead of as a workload + /// that misbehaves much later for no visible reason. + pub uid: u32, + pub gid: u32, +} + +/// One share to mount, named by tag. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Mount { + /// The share's tag. Never a path on the other side of the channel — the + /// guest learns nothing about the filesystem it is being handed a piece of. + pub tag: String, + /// Where it lands inside the guest. + /// + /// The caller names this, not the guest: choosing a mount point means + /// knowing what the workload expects to find there, which is exactly the + /// knowledge a workload-independent init does not have. ref(d-0033) + pub at: String, + #[serde(default)] + pub ro: bool, +} + +/// The output the compositor should produce. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct Geometry { + pub width: u32, + pub height: u32, + pub fps: u32, + #[serde(default)] + pub hdr: bool, +} + +/// What the workload exiting means for the session. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnExit { + /// Whether the exit ends the session. + /// + /// This says what an exit *means*; it is not a restart policy. The guest + /// reports the exit and stops, and starting something again is a new + /// command from the caller — the only end that can see whether restarting + /// is repair or a loop. ref(d-0033) + pub terminal: bool, +} + +/// Everything the guest is told at boot, in one document. +/// +/// Sent once, immediately after the handshake, and read once. Deliberately not +/// a conversation: boot configuration is a document, and a document cannot +/// half-arrive. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct BootDescriptor { + pub exec: Exec, + #[serde(default)] + pub mounts: Vec, + pub geometry: Geometry, + pub on_exit: OnExit, +} + +/// How a workload ended. +/// +/// Exactly one of these is set: a process that was signalled has no exit code, +/// and reporting `0` for one would make a kill look like a clean run. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct Exit { + #[serde(skip_serializing_if = "Option::is_none", default)] + pub exit_code: Option, + #[serde(skip_serializing_if = "Option::is_none", default)] + pub signal: Option, +} + +impl Exit { + pub fn code(code: i32) -> Self { + Self { + exit_code: Some(code), + signal: None, + } + } + + pub fn signal(signal: i32) -> Self { + Self { + exit_code: None, + signal: Some(signal), + } + } +} + +/// What the guest says. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum GuestToHost { + /// First line on the connection, before anything else is read or written. + Ready { protocol_version: u32 }, + /// The workload the descriptor named has ended. Terminal or not is the + /// descriptor's answer, not this message's. + WorkloadExited { + #[serde(flatten)] + exit: Exit, + }, +} + +/// What the guest is told. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum HostToGuest { + /// The boot descriptor. One per connection. + Boot { + #[serde(flatten)] + descriptor: Box, + }, + /// Stop the workload. Idempotent, and does not end the session. + Stop, + /// Shut the guest down. + Shutdown, +} + +/// Encode one message as a line, framing included. +/// +/// Newline-delimited JSON: the channel is a byte stream, so it needs a frame, +/// and a frame a person can read in a log of the channel itself is worth more +/// here than a compact one. +pub fn to_line(message: &T) -> Result { + let mut line = serde_json::to_string(message)?; + line.push('\n'); + Ok(line) +} + +/// Decode one line. The trailing newline is optional, so a caller may pass what +/// a line-oriented reader handed it either way. +pub fn from_line(line: &str) -> Result { + serde_json::from_str(line.trim_end_matches(['\n', '\r'])) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn descriptor() -> BootDescriptor { + BootDescriptor { + exec: Exec { + argv: vec!["/usr/bin/true".into()], + env: BTreeMap::from([("HOME".to_string(), "/mnt/user".to_string())]), + cwd: Some("/mnt/user".into()), + uid: 1000, + gid: 1000, + }, + mounts: vec![Mount { + tag: "install".into(), + at: "/mnt/install".into(), + ro: true, + }], + geometry: Geometry { + width: 1920, + height: 1080, + fps: 60, + hdr: false, + }, + on_exit: OnExit { terminal: true }, + } + } + + #[test] + fn a_line_round_trips() { + let line = to_line(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .unwrap(); + assert!(line.ends_with('\n'), "a line has to carry its own frame"); + assert!(!line.trim_end().contains('\n'), "one message is one line"); + + let back: HostToGuest = from_line(&line).unwrap(); + assert_eq!( + back, + HostToGuest::Boot { + descriptor: Box::new(descriptor()) + } + ); + } + + #[test] + fn a_signalled_exit_is_not_a_zero_exit() { + let signalled = to_line(&GuestToHost::WorkloadExited { + exit: Exit::signal(9), + }) + .unwrap(); + assert!( + !signalled.contains("exit_code"), + "a signalled workload has no exit code: {signalled}" + ); + + let clean = to_line(&GuestToHost::WorkloadExited { + exit: Exit::code(0), + }) + .unwrap(); + assert!( + !clean.contains("signal"), + "a clean exit was not signalled: {clean}" + ); + } + + #[test] + fn defaults_cover_what_a_caller_may_leave_out() { + let json = r#"{"exec":{"argv":["/bin/sh"],"uid":1000,"gid":1000}, + "geometry":{"width":1280,"height":720,"fps":30}, + "on_exit":{"terminal":false}}"#; + let parsed: BootDescriptor = from_line(json).unwrap(); + assert!(parsed.mounts.is_empty()); + assert!(parsed.exec.env.is_empty()); + assert_eq!(parsed.exec.cwd, None); + assert!(!parsed.geometry.hdr); + } +}