guest: hand layered init control to the managed runtime

nvx-rs/src/guest/init.rs

@@ -1,5 +1,5 @@
-//! `nvx-init TIME_DAEMON [ANNOUNCER]`: NVX's PID 1 for layered roots, which runs ahead of the
-//! workloads.
+//! `nvx-init TIME_DAEMON [ANNOUNCER]`: NVX's PID 1, which initializes a layered root and serves
+//! its managed control session through the same dispatcher as blockless VMs.
 //!
 //! A layered root's early boot, which the shell `/init` runs, mounts the kernel's file systems,
 //! runs `nvx-time boot`, whose child becomes the time daemon, configures the network, and, when
@@ -30,8 +30,8 @@
 //!   nothing from it;
 //! - ends the VM itself, without another guest program: it writes every file system's modified
 //!   data to its storage, with `sync`, and then stops the VM through OpenVMM's control port with a
-//!   status: 0 after `SIGTERM`, or 125, as `nvx-init-agent`'s failures do, after a failure, which
-//!   an `NVX-INIT-ERROR:` line on the console names first.
+//!   status: 0 after a host `STOP` or `SIGTERM`, or 125, as `nvx-init-agent`'s failures do, after
+//!   a failure, which an `NVX-INIT-ERROR:` line on the console names first.
 //!
 //! Outside PID 1, `nvx-init` changes nothing: it fails with status 2 and a message.
 
@@ -64,8 +64,6 @@ const NICE: i32 = -10;
 /// The signals that PID 1 reads from its signal descriptor: the end of a child, and the request
 /// to end the VM.
 const SIGNALS: [Signal; 2] = [Signal::SIGCHLD, Signal::SIGTERM];
-/// Pause before PID 1 polls its signal descriptor again after `poll` failed.
-const POLL_RETRY: Duration = Duration::from_secs(1);
 
 /// The helpers that the shell `/init` hands over: the only processes whose end ends the VM.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -769,8 +767,6 @@ trait System {
     /// Collects the status of one child that ended, without waiting, or returns `None` when none
     /// did.
     fn reap(&mut self) -> Option<(ProcessId, ChildStatus, i32)>;
-    /// Waits for one of [`SIGNALS`], and returns it.
-    fn wait_signal(&mut self) -> Option<Signal>;
     /// Writes the diagnostic `line` to the console.
     fn report(&mut self, line: &[u8]);
     /// Writes every file system's modified data to its storage, and then stops the VM with
@@ -811,19 +807,6 @@ impl System for RealSystem {
         sys::reap_any_nonblocking().map(|child| (child.process, child.status, child.wait_status))
     }
 
-    fn wait_signal(&mut self) -> Option<Signal> {
-        let signals = self.signals.as_ref()?;
-        loop {
-            if let Some(event) = signals.read() {
-                return event.signal;
-            }
-            match sys::poll([Pollable::Signals(signals)], None) {
-                Ok(_) | Err(Errno::EINTR) => {}
-                Err(_) => std::thread::sleep(POLL_RETRY),
-            }
-        }
-    }
-
     fn report(&mut self, line: &[u8]) {
         write_console(kernel::CONSOLE_DEVICE, line);
     }
@@ -853,10 +836,6 @@ impl Runtime {
     }
 
     /// Poll alongside control input so a blocked peer cannot hide a helper's end.
-    #[cfg_attr(
-        not(test),
-        expect(dead_code, reason = "SL7.6b polls the init runtime with control input")
-    )]
     pub(crate) fn wake(&self) -> Pollable<'_> {
         Pollable::Signals(self.system.signals.as_ref().expect("listening"))
     }
@@ -907,12 +886,13 @@ fn serve_layered(runtime: &mut Runtime) -> Result<(), String> {
         )
         .as_bytes(),
     );
-    loop {
-        if runtime.service(|_, _| {})? {
-            return Ok(());
-        }
-        runtime.terminating |= runtime.system.wait_signal() == Some(Signal::SIGTERM);
-    }
+    super::managed_agent::AgentConfig::layered(&boot.command_line, boot.user, boot.home)
+        .map_err(|error| format!("invalid layered identity: {}", error.desc()))?
+        .serve_layered(
+            runtime,
+            boot.control,
+            (boot.table, boot.mapper, boot.export),
+        )
 }
 
 /// Keeps diagnostics, sync and direct power-off with the original PID-1 owner.

nvx-rs/src/guest/init/tests.rs

@@ -23,12 +23,12 @@ use crate::guest::sys::{ForkResult, wait};
 
 /// Drives the unchanged helper/signal scripts through production preparation, collection and
 /// shutdown; layered boot itself is exercised by the boot tests.
-fn init(system: &mut impl System, arguments: &[CString]) {
+fn init(system: &mut impl Waiting, arguments: &[CString]) {
     let result = serve(system, arguments);
     finish(system, result);
 }
 
-fn serve(system: &mut impl System, arguments: &[CString]) -> Result<(), String> {
+fn serve(system: &mut impl Waiting, arguments: &[CString]) -> Result<(), String> {
     let helpers = prepare(system, arguments)?;
     let mut terminating = false;
     loop {
@@ -40,6 +40,25 @@ fn serve(system: &mut impl System, arguments: &[CString]) -> Result<(), String>
     }
 }
 
+trait Waiting: System {
+    fn wait_signal(&mut self) -> Option<Signal>;
+}
+
+impl Waiting for RealSystem {
+    fn wait_signal(&mut self) -> Option<Signal> {
+        let signals = self.signals.as_ref()?;
+        loop {
+            if let Some(event) = signals.read() {
+                return event.signal;
+            }
+            match sys::poll([Pollable::Signals(signals)], None) {
+                Ok(_) | Err(Errno::EINTR) => {}
+                Err(_) => std::thread::sleep(Duration::from_secs(1)),
+            }
+        }
+    }
+}
+
 /// One operation that PID 1 asked its system for.
 #[derive(Clone, Debug, PartialEq, Eq)]
 enum Call {
@@ -51,7 +70,7 @@ enum Call {
     Adopt(i32),
     /// [`System::reap`].
     Reap,
-    /// [`System::wait_signal`].
+    /// [`Waiting::wait_signal`].
     Wait,
     /// [`System::report`] of this line.
     Report(String),
@@ -73,7 +92,7 @@ struct Fake {
     /// The answers of [`System::reap`], in order: a child that ended, or `None` when none did.
     /// Once they run out, no child ended.
     reaped: VecDeque<Option<(i32, ChildStatus)>>,
-    /// The signals that [`System::wait_signal`] returns, in order. The check fails if PID 1 waits
+    /// The signals that [`Waiting::wait_signal`] returns, in order. The check fails if PID 1 waits
     /// for another.
     signals: VecDeque<Signal>,
 }
@@ -112,15 +131,6 @@ impl System for Fake {
         Some((pid(process), status, raw))
     }
 
-    fn wait_signal(&mut self) -> Option<Signal> {
-        self.calls.push(Call::Wait);
-        Some(
-            self.signals
-                .pop_front()
-                .expect("PID 1 waits for no signal that the check sends"),
-        )
-    }
-
     fn report(&mut self, line: &[u8]) {
         self.calls
             .push(Call::Report(String::from_utf8(line.to_vec()).unwrap()));
@@ -131,6 +141,17 @@ impl System for Fake {
     }
 }
 
+impl Waiting for Fake {
+    fn wait_signal(&mut self) -> Option<Signal> {
+        self.calls.push(Call::Wait);
+        Some(
+            self.signals
+                .pop_front()
+                .expect("PID 1 waits for no signal that the check sends"),
+        )
+    }
+}
+
 #[test]
 fn the_shared_collector_forwards_workload_status_and_keeps_helper_failure_fatal() {
     let helpers = Helpers {
@@ -261,6 +282,16 @@ impl System for Subreaper {
         Some(child)
     }
 
+    fn report(&mut self, line: &[u8]) {
+        self.reports.push_str(std::str::from_utf8(line).unwrap());
+    }
+
+    fn shut_down(&mut self, status: u8) {
+        self.status = Some(status);
+    }
+}
+
+impl Waiting for Subreaper {
     fn wait_signal(&mut self) -> Option<Signal> {
         if let Some(before_wait) = self.before_wait.take() {
             before_wait();
@@ -269,14 +300,6 @@ impl System for Subreaper {
         self.signals.extend(signal);
         signal
     }
-
-    fn report(&mut self, line: &[u8]) {
-        self.reports.push_str(std::str::from_utf8(line).unwrap());
-    }
-
-    fn shut_down(&mut self, status: u8) {
-        self.status = Some(status);
-    }
 }
 
 /// A system whose priorities, children, and statuses are the real ones, which records the
@@ -312,6 +335,16 @@ impl System for Observer {
         self.real.reap()
     }
 
+    fn report(&mut self, line: &[u8]) {
+        self.reports.push_str(std::str::from_utf8(line).unwrap());
+    }
+
+    fn shut_down(&mut self, status: u8) {
+        self.status = Some(status);
+    }
+}
+
+impl Waiting for Observer {
     fn wait_signal(&mut self) -> Option<Signal> {
         let processes = std::iter::once(sys::current_process()).chain(self.helpers);
         self.observed = processes
@@ -323,14 +356,6 @@ impl System for Observer {
             .collect();
         Some(Signal::SIGTERM)
     }
-
-    fn report(&mut self, line: &[u8]) {
-        self.reports.push_str(std::str::from_utf8(line).unwrap());
-    }
-
-    fn shut_down(&mut self, status: u8) {
-        self.status = Some(status);
-    }
 }
 
 /// Returns the process ID `raw`.

nvx-rs/src/guest/managed_agent/dispatch.rs

@@ -1,19 +1,76 @@
-use crate::guest::sys::Errno;
+use std::ffi::CString;
+
+use crate::guest::init::Runtime;
+use crate::guest::mappings::{Mapper, MappingTable, SourceRoot};
+use crate::guest::sys::{self, Errno, FileDescriptor};
 use crate::protocol::{
     self, APP_CANCEL, APP_EXEC, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, APP_STOP, APP_STOPPED,
     GuestFeatures, INVALID_REQUEST, MAPPINGS_INCOMPLETE, UNSUPPORTED_OPERATION,
 };
 
 use super::diagnostics;
 use super::mappings::{HostMappings, MappingError};
-use super::seam::System;
+use super::seam::{Paths, RealSystem, System};
 use super::session::{Event, ReceiveFailure, Session};
-use super::{AgentConfig, exec, tty};
+use super::{AgentConfig, exec, launch, tty};
+
+impl AgentConfig {
+    /// Takes a layered cold boot's CLI identity and its already-resolved account defaults.
+    pub(crate) fn layered(
+        command_line: &[u8],
+        user: CString,
+        home: CString,
+    ) -> Result<Self, Errno> {
+        let mut identities = crate::protocol::cmdline::init_tokens(command_line)
+            .filter(|token| token.name == b"nvx_identity")
+            .filter_map(|token| token.value);
+        let identity = identities.next().ok_or(Errno::EINVAL)?;
+        let mut ids = identity.split(|&byte| byte == b':');
+        let (Some(uid), Some(gid), None) = (ids.next(), ids.next(), ids.next()) else {
+            return Err(Errno::EINVAL);
+        };
+        if identities.next().is_some() {
+            return Err(Errno::EINVAL);
+        }
+        launch::parse_id(uid)?;
+        launch::parse_id(gid)?;
+        Ok(Self {
+            rootfs: c"-".to_owned(),
+            hostname: c"nvx-managed".to_owned(),
+            uid: sys::c_string(uid)?,
+            gid: sys::c_string(gid)?,
+            user,
+            home,
+            direct: true,
+            entry: None,
+        })
+    }
+
+    /// Serves an initialized layered root, returning shutdown to its init owner.
+    pub(crate) fn serve_layered(
+        &self,
+        runtime: &mut Runtime,
+        control: FileDescriptor,
+        (table, mapper, export): (MappingTable, Mapper, Option<SourceRoot>),
+    ) -> Result<(), String> {
+        let mut system = RealSystem::new(Paths::default());
+        let mut session = Session::with_runtime(control, runtime)?;
+        let mut mappings = HostMappings::layered(table, mapper, export)
+            .map_err(|error| format!("invalid layered mappings: {}", error.desc()))?;
+        run(&mut system, &mut session, self, &mut mappings)
+            .map(|Outcome::Stopped| ())
+            .map_err(|error| {
+                session
+                    .runtime_failure()
+                    .unwrap_or_else(|| format!("control session failed: {}", error.desc()))
+            })
+    }
+}
 
 /// Terminal outcome of the long-lived dispatch loop.
 #[derive(Debug)]
 pub(crate) enum Outcome {
-    /// The host requested and acknowledged an orderly stop.
+    /// The host requested STOP, or init accepted an orderly termination.
     Stopped,
 }
 
@@ -24,16 +81,30 @@ pub(crate) fn run(
     config: &AgentConfig,
     mappings: &mut HostMappings,
 ) -> Result<Outcome, Errno> {
-    session.send_attach()?;
+    if session.stopping() {
+        return Ok(Outcome::Stopped);
+    }
+    if let Err(error) = session.send_attach() {
+        return if session.stopping() {
+            Ok(Outcome::Stopped)
+        } else {
+            Err(error)
+        };
+    }
     loop {
         let payload = match session.next_event() {
             Ok(Event::Reset) => {
-                session.acknowledge().inspect_err(|error| {
-                    diagnostics::error(system, "reset-ack", *error as i32);
-                })?;
+                if let Err(error) = session.acknowledge() {
+                    if session.stopping() {
+                        return Ok(Outcome::Stopped);
+                    }
+                    diagnostics::error(system, "reset-ack", error as i32);
+                    return Err(error);
+                }
                 continue;
             }
             Ok(Event::Data(payload)) => payload,
+            Err(_) if session.stopping() => return Ok(Outcome::Stopped),
             Err(failure) => return Err(report_failure(system, failure)),
         };
         // As in the C agent, a DATA record that fails also reports the `data` stage, after any
@@ -43,6 +114,7 @@ pub(crate) fn run(
         match handle_data(system, session, config, mappings, &payload) {
             Ok(Some(outcome)) => return Ok(outcome),
             Ok(None) => {}
+            Err(_) if session.stopping() => return Ok(Outcome::Stopped),
             Err(error) => {
                 diagnostics::error(system, "data", error as i32);
                 return Err(error);