From ceebee7183bb8372cd6bc68b9f39bad6c7c249a6 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 07:28:56 +0300 Subject: [PATCH] fix(nesinit): a launch's group survives a failed join, a re-add and a busy rmdir; a dead compositor ends the box --- apps/nesinit/src/session.rs | 51 ++++++++ apps/nesinit/src/workload.rs | 225 ++++++++++++++++++++++++++++------- 2 files changed, 236 insertions(+), 40 deletions(-) diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 8ae86a45..4edff5a2 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -158,6 +158,8 @@ where // resolves immediately and forever, so an unguarded branch on one turns // this loop into a spin that still looks like it is waiting. let mut services_open = true; + // The box's compositor exit, once there is a compositor. + let mut compositor_exit: Option = None; loop { // The exits are guarded like the channels below: an empty set resolves @@ -177,6 +179,9 @@ where up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up), found = addresses.recv(), if carrier_open => Event::Address(found), gone = services.deaths().recv(), if services_open => Event::ServiceDied(gone), + gone = async { compositor_exit.as_mut().expect("guarded").await }, if compositor_exit.is_some() => { + Event::CompositorDied(gone) + }, }; let line = match event { @@ -281,6 +286,18 @@ where send(&mut writer, &GuestToHost::ServiceDied { name, exit }).await?; continue; } + Event::CompositorDied(gone) => { + // Every launch draws on it, so with it gone nothing in this box + // can show a window: said as a service death, then the box + // ends rather than carrying on looking healthy. + let exit = gone.unwrap_or_else(|_| Exit::code(1)); + tracing::error!(?exit, "the compositor has exited: no window in this box can draw"); + send(&mut writer, &GuestToHost::ServiceDied { name: "nescope".into(), exit }).await?; + workload.signal_stop(); + return Ok(Outcome::Refused(Failure::new(format!( + "the compositor exited ({exit:?}), so the box can no longer draw" + )))); + } Event::ServiceDied(None) => { // The supervisor is gone, which on the way down is ordinary. // Nothing to report and nothing to end: a box with no service @@ -411,6 +428,7 @@ where Ok(up) }) { Ok(up) => { + compositor_exit = workload.compositor_exit(); tracing::info!(services = up.len(), "the box is ready to be launched into"); send(&mut writer, &GuestToHost::Initialized { services: up }).await? } @@ -651,6 +669,7 @@ enum Event { FromWorkload(Option), Address(Option), ServiceDied(Option), + CompositorDied(std::io::Result), } /// Hand an envelope to the relay, and treat a relay that is not there as the @@ -971,6 +990,38 @@ mod tests { assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}"); } + /// A compositor that exits ends the box, and the host is told which + /// service it was. + #[tokio::test] + async fn a_compositor_that_exits_is_reported_and_ends_the_box() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (tx, rx) = tokio::sync::oneshot::channel(); + let mut double = Double::exits_when_stopped(Exit::code(0)); + double.compositor_exit = Some(rx); + let session = spawn(guest, Given::new(double)); + + let mut given = descriptor(); + given.compositor = Some(nesprotocol::lifecycle::Compositor { + width: 1280, + height: 720, + fps: 30, + uid: 1001, + gid: 1001, + env: Default::default(), + }); + caller.expect_ready().await; + caller.say(&HostToGuest::Boot { descriptor: Box::new(given) }).await; + caller.expect_booted().await; + tx.send(Exit::signal(11)).unwrap(); + assert_eq!( + caller.expect().await, + GuestToHost::ServiceDied { name: "nescope".into(), exit: Exit::signal(11) } + ); + let (outcome, _, _) = session.await.unwrap(); + assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}"); + } + #[tokio::test] async fn the_descriptors_video_limits_reach_the_services() { // The ceiling is useless if it stops at the descriptor. `neshub` is the diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index dc19299d..6684bc7d 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -91,6 +91,14 @@ pub trait Workload { /// launch's stop may reach it. ref(d-0085) fn start_compositor(&mut self, compositor: &Compositor) -> Result<(), Failure>; + /// The compositor's exit, once, after [`Workload::start_compositor`] has + /// returned. The session reports it and ends the box: a box that cannot + /// draw is no use to anyone in it. `None` when there is no compositor or it + /// was already taken. + fn compositor_exit(&mut self) -> Option { + None + } + /// Ask one launch to stop: its whole process group, not its top process. /// Idempotent, and never ends the session by itself. A launch that is not /// running is not signalled: its pid may already belong to something else. @@ -118,6 +126,9 @@ pub struct Process { /// The box's compositor, if it has one. Not a launch: nothing in /// `running` is it, and a launch's stop cannot reach it. compositor: Option, + /// The compositor's exit, until the session takes it. See + /// [`Workload::compositor_exit`]. + compositor_exit: Option, } /// How long a launch's group has to leave after being asked before it is @@ -194,8 +205,14 @@ pub(crate) struct Group { /// The top process, which is also the session id. sid: i32, cgroup: Option, + /// Set when the same id is started again: this group is no longer the + /// launch's, and nothing waiting on it may act for the launch. + superseded: std::sync::atomic::AtomicBool, } +/// Numbers each group, so two launches of one id never share a cgroup. +static GENERATION: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + /// An id as a directory name: only what is safe in a path. fn sanitise(id: &str) -> String { id.chars() @@ -203,18 +220,14 @@ fn sanitise(id: &str) -> String { .collect() } -/// Make the cgroup for launch `id` under `base`, or `None` when it cannot be -/// (no cgroup v2, no permission, or a populated one of that name already). -fn make_cgroup(base: &str, id: &str) -> Option { - let path = std::path::Path::new(base).join(format!("nestri-launch-{}", sanitise(id))); +/// Make the cgroup for launch `id`, generation `generation`, under `base`, or +/// `None` when it cannot be (no cgroup v2, no permission, or one of that name +/// already). The generation is in the name so a restart never shares a +/// directory with what is left of the launch before it. +fn make_cgroup(base: &str, id: &str, generation: u64) -> Option { + let path = std::path::Path::new(base).join(format!("nestri-launch-{}-{generation}", sanitise(id))); match std::fs::create_dir(&path) { Ok(()) => Some(path), - Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { - // A leftover from a launch with this id is reused only when empty: - // sharing one would make a stop of one launch the stop of another. - let procs = std::fs::read_to_string(path.join("cgroup.procs")).ok()?; - procs.trim().is_empty().then_some(path) - } Err(error) => { tracing::warn!(%id, base, "no cgroup for this launch ({error}); its members are found by environment"); None @@ -222,6 +235,14 @@ fn make_cgroup(base: &str, id: &str) -> Option { } } +/// Whether `pid` is listed in the cgroup at `path`: the proof the launch's +/// child joined it, which its own unchecked write could not give. +fn joined(path: &std::path::Path, pid: i32) -> bool { + std::fs::read_to_string(path.join("cgroup.procs")) + .map(|procs| procs.lines().any(|line| line.trim().parse::().ok() == Some(pid))) + .unwrap_or(false) +} + /// Whether `pid` is not a zombie. fn is_live(pid: i32) -> bool { std::fs::read_to_string(format!("/proc/{pid}/stat")) @@ -239,18 +260,28 @@ fn environ_names(environ: &[u8], id: &str) -> bool { impl Group { /// The group of a launch whose top process is `sid`. fn new(id: &LaunchId, sid: i32, cgroup: Option) -> Self { - Self { id: id.as_str().to_string(), sid, cgroup } + Self { id: id.as_str().to_string(), sid, cgroup, superseded: false.into() } + } + + /// Whether the launch has been started again since this group was made. + fn is_superseded(&self) -> bool { + self.superseded.load(std::sync::atomic::Ordering::SeqCst) } /// Every live process of the launch. pub(crate) fn members(&self) -> Vec { if let Some(path) = &self.cgroup { if let Ok(procs) = std::fs::read_to_string(path.join("cgroup.procs")) { - return procs + let found: Vec = procs .lines() .filter_map(|line| line.trim().parse::().ok()) .filter(|pid| is_live(*pid)) .collect(); + // An empty cgroup with the top process alive is a cgroup the + // launch is not in: the scan below is what finds it. + if !found.is_empty() || !in_session(self.sid, self.sid) { + return found; + } } } let mut members = session_members(self.sid); @@ -297,22 +328,35 @@ impl Group { self.signal(libc::SIGKILL); } - /// Remove the launch's cgroup, once it is empty. - fn remove(&self) { - if let Some(path) = &self.cgroup { - let _ = std::fs::remove_dir(path); + /// Remove the launch's cgroup, once it is empty. A zombie still counts as + /// a member to the kernel until it is reaped, so `EBUSY` is retried for a + /// moment before it is given up on. + async fn remove(&self) { + let Some(path) = &self.cgroup else { return }; + let mut error = None; + for _ in 0..40 { + match std::fs::remove_dir(path) { + Ok(()) => return, + Err(e) if e.kind() == io::ErrorKind::NotFound => return, + Err(e) => error = Some(e), + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + if let Some(error) = error { + tracing::warn!(path = %path.display(), "a launch's cgroup was not removed: {error}"); } } } /// Wait for a launch's group to be gone, ending it if it will not go alone. /// -/// Reached once the top process has been reaped. A launch counts as ended only -/// when nothing of it is left: a window owned by a process still running is a -/// launch still running, whatever became of the process that started it. +/// Reached once the top process has been reaped. When it leaves, whatever it +/// started is ended with it (TERM, then KILL after the grace): a launch is not +/// reported ended while something of it is still running, and what it left +/// behind is not kept alive. async fn group_gone(group: std::sync::Arc) { if group.members().is_empty() { - group.remove(); + group.remove().await; return; } tracing::info!(sid = group.sid, "a launch's top process has gone and its group has not: ending the group"); @@ -325,12 +369,13 @@ async fn group_gone(group: std::sync::Arc) { let give_up = deadline + STOP_GRACE * 2; loop { if group.members().is_empty() { - group.remove(); + group.remove().await; return; } let now = std::time::Instant::now(); if now >= give_up { tracing::warn!(sid = group.sid, "a launch's group would not die; reporting the launch ended anyway"); + group.remove().await; return; } if now >= deadline && !killed { @@ -349,12 +394,17 @@ fn stop_group(group: std::sync::Arc) { std::thread::spawn(move || { let deadline = std::time::Instant::now() + STOP_GRACE; while std::time::Instant::now() < deadline { - if group.members().is_empty() { + // The launch was started again under this id: what is in the + // environment-named fallback is now the new launch's, and a kill + // here would be its death. + if group.is_superseded() || group.members().is_empty() { return; } std::thread::sleep(std::time::Duration::from_millis(100)); } - group.kill(); + if !group.is_superseded() { + group.kill(); + } }); } @@ -368,6 +418,7 @@ impl Process { groups: std::collections::BTreeMap::new(), cgroup_root: CGROUP_ROOT.to_string(), compositor: None, + compositor_exit: None, } } @@ -574,7 +625,8 @@ impl Workload for Process { // The launch's cgroup, made here and joined before exec: what a child // that makes its own session still cannot leave. The path is built // now because the closure below may not allocate. - let cgroup = make_cgroup(&self.cgroup_root, id.as_str()); + let generation = GENERATION.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let cgroup = make_cgroup(&self.cgroup_root, id.as_str(), generation); let procs_file = cgroup .as_ref() .and_then(|path| std::ffi::CString::new(path.join("cgroup.procs").into_os_string().into_vec()).ok()); @@ -592,8 +644,9 @@ impl Workload for Process { return Err(io::Error::last_os_error()); } // Into the launch's cgroup, as root, before the uid drops. - // A failure is not one: the launch is still found by its - // environment. + // A failure is not one here, where nothing may be said: the + // parent checks the child is in the cgroup after the spawn and + // falls back to the environment when it is not. if let Some(file) = &procs_file { let fd = libc::open(file.as_ptr(), libc::O_WRONLY | libc::O_CLOEXEC); if fd >= 0 { @@ -643,10 +696,27 @@ impl Workload for Process { // The session id is the top pid (`setsid` above), and stays that // number for as long as anything is in it. let sid = self.running.get(id).map(|w| w.pid).expect("just inserted"); + // The join was unchecked in the child. A cgroup the child is not in + // would read as an empty launch, and a stop of it would do nothing, so + // it is dropped for the session-and-environment fallback. + let cgroup = match cgroup { + Some(path) if !joined(&path, sid) => { + tracing::warn!( + %id, + path = %path.display(), + "the launch did not join its cgroup; its members are found by environment" + ); + let _ = std::fs::remove_dir(&path); + None + } + other => other, + }; // Groups with nothing left in them are dropped here, as `running` is. self.groups.retain(|_, group| !group.members().is_empty()); let group = std::sync::Arc::new(Group::new(id, sid, cgroup)); - self.groups.insert(id.clone(), group.clone()); + if let Some(old) = self.groups.insert(id.clone(), group.clone()) { + old.superseded.store(true, std::sync::atomic::Ordering::SeqCst); + } Ok(Box::pin(async move { let exit = exit .await @@ -706,29 +776,38 @@ impl Workload for Process { Failure::new(format!("{} as {uid}:{gid}: {error}", exec.argv[0])) })?; let exit = watched.take_exit().expect("a new watch has its exit"); - // Said when it goes: with it gone no launch has anywhere to draw, and - // nothing else in the box would say why. - if let Ok(handle) = tokio::runtime::Handle::try_current() { - handle.spawn(async move { - if let Ok(exit) = exit.await { - tracing::error!(?exit, "the compositor has exited: no window in this box can draw"); - } - }); - } let pid = watched.pid; let flag = &watched; - crate::services::await_path( + if let Err(failure) = crate::services::await_path( "the compositor", &socket, before, &|| flag.running(), COMPOSITOR_READY, - )?; + ) { + // One that never served is not left running behind a box that + // reports it could not be made. + if watched.running() { + // SAFETY: two integers; the pid is still this child's. + unsafe { libc::kill(-pid, libc::SIGKILL) }; + unsafe { libc::kill(pid, libc::SIGKILL) }; + } + return Err(failure); + } tracing::info!(pid, %socket, "the compositor is serving"); + // Handed to the session, which reports it and ends the box: with it + // gone no launch has anywhere to draw. + self.compositor_exit = Some(Box::pin(async move { + exit.await.map_err(|_| io::Error::other("the compositor's exit was not delivered")) + })); self.compositor = Some(watched); Ok(()) } + fn compositor_exit(&mut self) -> Option { + self.compositor_exit.take() + } + fn stop_launch(&mut self, id: &LaunchId) { if let Some(group) = self.groups.get(id) { // Whether or not the top process is still there: its children may @@ -1465,6 +1544,22 @@ mod tests { }); } + /// A process whose cgroups are made in a directory of the test's own, so + /// no test touches the real `/sys/fs/cgroup`. It is a plain directory with + /// no `cgroup.procs`, so a join into it fails, which is the fallback path. + fn process_in_tmp(waiters: crate::reap::Waiters) -> (Process, std::path::PathBuf) { + static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + let root = std::env::temp_dir().join(format!( + "nesinit-cg-{}-{}", + std::process::id(), + NEXT.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + )); + std::fs::create_dir_all(&root).unwrap(); + let mut process = Process::new(waiters); + process.cgroup_root = root.to_string_lossy().into_owned(); + (process, root) + } + fn me(script: &str) -> Exec { // SAFETY: no arguments. let (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) }; @@ -1479,7 +1574,7 @@ mod tests { async fn started(script: &str) -> (Process, LaunchId, i32, Exited) { let waiters = crate::reap::Waiters::new(); - let mut process = Process::new(waiters.clone()); + let (mut process, _root) = process_in_tmp(waiters.clone()); static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); // Unique per test: members are found by the id in their environment. let id = LaunchId::new(format!( @@ -1541,11 +1636,51 @@ mod tests { assert!(group.members().is_empty()); } + /// A cgroup the child could not join is dropped, not trusted: it would + /// read as an empty launch and a stop would do nothing. + #[tokio::test] + async fn a_launch_that_cannot_join_its_cgroup_falls_back_and_is_still_stopped() { + let waiters = crate::reap::Waiters::new(); + let (mut process, root) = process_in_tmp(waiters.clone()); + let id = LaunchId::new(format!("join-{}", std::process::id())); + let exited = process.start(&id, &me("sleep 300 & wait")).unwrap(); + reap_for(waiters, process.running[&id].pid); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let group = process.groups[&id].clone(); + assert!(group.cgroup.is_none(), "the cgroup was kept"); + assert_eq!(std::fs::read_dir(&root).unwrap().count(), 0, "its directory was left"); + assert!(group.members().len() >= 2, "{:?}", group.members()); + process.stop_launch(&id); + tokio::time::timeout(std::time::Duration::from_secs(10), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert!(group.members().is_empty()); + } + + /// A stop that is still waiting out its grace must not kill a launch that + /// has taken the same id since. + #[tokio::test] + async fn a_pending_stop_does_not_kill_the_launch_that_replaced_it() { + let (mut process, id, _top, _exited) = started("sleep 300 & wait").await; + let old = process.groups[&id].clone(); + process.stop_launch(&id); + let waiters = process.waiters.clone(); + let _again = process.start(&id, &me("sleep 300 & wait")).unwrap(); + reap_for(waiters, process.running[&id].pid); + assert!(old.is_superseded()); + tokio::time::sleep(STOP_GRACE + std::time::Duration::from_secs(1)).await; + let new = process.groups[&id].clone(); + let alive = new.members(); + new.kill(); + assert!(alive.len() >= 2, "the new launch was killed by the old stop: {alive:?}"); + } + /// Two launches are two groups: stopping one leaves the other running. #[tokio::test] async fn stopping_one_launch_leaves_another_running() { let waiters = crate::reap::Waiters::new(); - let mut process = Process::new(waiters.clone()); + let (mut process, _root) = process_in_tmp(waiters.clone()); let (a, b) = (LaunchId::new("iso-a"), LaunchId::new("iso-b")); let exited_a = process.start(&a, &me("sleep 300 & wait")).unwrap(); let _exited_b = process.start(&b, &me("sleep 300 & wait")).unwrap(); @@ -1945,6 +2080,8 @@ pub mod double { /// Every compositor asked for. pub compositors: Vec, pub compositor_failure: Option, + /// When the compositor "exits": the session takes this after it starts. + pub compositor_exit: Option>, pub mount_failure: Option, pub image_failure: Option, /// Like `image_failure`, but only for a call that names an image, so a @@ -1984,6 +2121,7 @@ pub mod double { stops: 0, compositors: Vec::new(), compositor_failure: None, + compositor_exit: None, mount_failure: None, image_failure: None, named_image_failure: None, @@ -2082,6 +2220,13 @@ pub mod double { } } + fn compositor_exit(&mut self) -> Option { + let rx = self.compositor_exit.take()?; + Some(Box::pin(async move { + rx.await.map_err(|_| io::Error::other("the compositor was dropped")) + })) + } + fn stop_launch(&mut self, id: &LaunchId) { self.stops += 1; self.stopped.push(Some(id.clone()));