github-mirrors/netris-nestri
1
0
Fork 0
mirror of https://github.com/nestriness/nestri.git synced 2026-09-19 17:25:19 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat(nesinit): PID 1 for a box — reaping, ordered shutdown, and one channel out (#319)

Browse Source 
PID 1 inside a box. A microVM has no init unless something is it, and
three of
the jobs belong to nothing else in the guest: reaping whatever the
workload
orphans, turning a signal into an ordered shutdown, and holding the
guest end
of the one channel out.

- `reap.rs` — reaping, plus the subreaper bit and taking init out of the
OOM
  killer's reach.
- `shutdown.rs` — the order, behind a trait so it can be asserted
without a VM:
workload first and alone, then everything else, then flush, then power
off.
- `session.rs` — the exchange. The guest speaks first with its protocol
version, is handed one boot descriptor, reports, and stops. Generic over
the
  byte stream, so the whole protocol is testable over an in-memory pipe.
- `workload.rs` — one trait the descriptor drops into, a real process
behind
  it, and a double.
- `nesprotocol::lifecycle` — the types, behind a feature that is off by
  default so the media components keep building without serde.

It reports and does not supervise: when the workload ends, the exit goes
up the
channel and the session is over. Nothing here restarts anything.

**Mounting is not implemented in this PR.** The descriptor's `mounts`
are
refused rather than ignored, and the next PR in the stack implements
them.

### On the OOM killer

`refuse_oom_kill()` writes `-1000` to this process's own
`oom_score_adj`, and
that part is init's job: everything else in the guest exiting is a
message up
the channel, whereas init exiting takes the channel with it, and the far
end
sees a box that stopped answering for no stated reason.

The rest is the image's job and cannot be done from here. Making the
workload
the *preferred* victim means scoring processes this component did not
start, so
the image has to leave the workload's score at or above the default and
must
not lower it for the guest's own services either.

## The tests, failing first

Reaping, with `reap_exited()` returning nothing:

```
running 3 tests
test a_child_that_exits_is_reaped_with_its_code ... FAILED
test a_child_that_is_killed_is_reaped_as_signalled ... FAILED
test an_orphan_is_reaped_by_whoever_inherits_it ... FAILED

failures:

---- a_child_that_exits_is_reaped_with_its_code stdout ----

thread 'a_child_that_exits_is_reaped_with_its_code' (304365) panicked at apps/nesinit/tests/reaping.rs:52:32:
the child was left a zombie
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

---- a_child_that_is_killed_is_reaped_as_signalled stdout ----

thread 'a_child_that_is_killed_is_reaped_as_signalled' (304366) panicked at apps/nesinit/tests/reaping.rs:66:32:
the child was left a zombie

---- an_orphan_is_reaped_by_whoever_inherits_it stdout ----

thread 'an_orphan_is_reaped_by_whoever_inherits_it' (304367) panicked at apps/nesinit/tests/reaping.rs:98:5:
the child was left a zombie

failures:
    a_child_that_exits_is_reaped_with_its_code
    a_child_that_is_killed_is_reaped_as_signalled
    an_orphan_is_reaped_by_whoever_inherits_it

test result: FAILED. 0 passed; 3 failed; 0 ignored; 0 measured; 0 filtered out; finished in 15.02s
```

Shutdown, with one signal to everything instead of an order:

```
running 3 tests
test shutdown::tests::a_workload_that_leaves_in_time_is_not_killed ... ok
test shutdown::tests::a_workload_that_overstays_its_grace_is_killed_and_shutdown_still_finishes ... FAILED
test shutdown::tests::the_workload_stops_before_anything_else_and_the_disks_flush_before_power ... FAILED

failures:

---- shutdown::tests::a_workload_that_overstays_its_grace_is_killed_and_shutdown_still_finishes stdout ----

thread 'shutdown::tests::a_workload_that_overstays_its_grace_is_killed_and_shutdown_still_finishes' (304568) panicked at apps/nesinit/src/shutdown.rs:105:79:
called `Option::unwrap()` on a `None` value
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

---- shutdown::tests::the_workload_stops_before_anything_else_and_the_disks_flush_before_power stdout ----

thread 'shutdown::tests::the_workload_stops_before_anything_else_and_the_disks_flush_before_power' (304569) panicked at apps/nesinit/src/shutdown.rs:80:9:
assertion `left == right` failed
  left: ["signal_rest", "kill_rest", "flush_disks", "power_off"]
 right: ["signal_workload", "await_workload", "signal_rest", "kill_rest", "flush_disks", "power_off"]

failures:
    shutdown::tests::a_workload_that_overstays_its_grace_is_killed_and_shutdown_still_finishes
    shutdown::tests::the_workload_stops_before_anything_else_and_the_disks_flush_before_power

test result: FAILED. 1 passed; 2 failed; 0 ignored; 0 measured; 8 filtered out; finished in 0.00s

error: test failed, to rerun pass `-p nesinit --lib`
```

The handshake, with the version left where it was before the bump:

```
running 1 test
test session::tests::the_guest_speaks_first_and_says_its_version ... FAILED

failures:

---- session::tests::the_guest_speaks_first_and_says_its_version stdout ----

thread 'session::tests::the_guest_speaks_first_and_says_its_version' (304772) panicked at apps/nesinit/src/session.rs:180:9:
assertion `left == right` failed
  left: Ready { protocol_version: 1 }
 right: Ready { protocol_version: 2 }
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    session::tests::the_guest_speaks_first_and_says_its_version

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 10 filtered out; finished in 0.00s

error: test failed, to rerun pass `-p nesinit --lib`
```

And the property most likely to be quietly regressed later — a
supervisor loop
that restarts what it started, instead of reporting:

```
running 1 test
test session::tests::an_exit_is_reported_and_the_workload_is_not_started_again ... FAILED

failures:

---- session::tests::an_exit_is_reported_and_the_workload_is_not_started_again stdout ----

thread 'session::tests::an_exit_is_reported_and_the_workload_is_not_started_again' (304968) panicked at apps/nesinit/src/session.rs:237:9:
assertion `left == right` failed: an exit is reported, never restarted
  left: 2
 right: 1
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    session::tests::an_exit_is_reported_and_the_workload_is_not_started_again

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 10 filtered out; finished in 0.00s

error: test failed, to rerun pass `-p nesinit --lib`
```

All green after: 11 unit tests, 3 against real forked children, plus 3
new in
`nesprotocol`.

## What this does not verify

- **Nothing has run in a VM, and nothing has run as PID 1.** Orphan
reparenting is exercised with `PR_SET_CHILD_SUBREAPER` in a test
process,
  which is the same kernel mechanism but not the same privilege.
- **No real vsock connection has been made.** The protocol is tested
over an
in-memory pipe; the dial itself — the address, the port, a listener that
is
not there — is unexercised, and the deliberate no-retry behaviour has
never
  met a refused connection.
- **The real shutdown is untested.** The order is asserted through a
double;
`kill(-1)`, `sync` and the power-off call themselves need a guest, and a
test
  process must not make them.
- **The OOM write is unchecked.** It warns and continues where there is
no
  procfs, and nothing here confirms the kernel honoured the score.
- **Dropping to `uid`/`gid` before exec is unexercised** — it needs
privileges
  a test does not have, so the `pre_exec` path has run in no test.
- **No number here is measured.** Nothing in this PR claims a timing, a
rate
  or a count from hardware.
- **A version mismatch is not refused by this end.** The guest announces
its
version in its first line and the far end compares; as the layer stands
there
  is nothing for the guest to compare against, so "both ends refuse on
  mismatch" is only half-implementable. Worth settling in the channel's
specification before either end grows a second version, and I would
rather
  raise it than invent a message for it here.

## Since review

`cb8f37a` — three fixes, all from the review, described in its message.
The one
worth naming here is that the reaper is now the only thing in the
component
that calls `wait`: it hands each exit to whoever asked for that pid, and
registering interest holds the same lock the delivery takes, so an exit
that
happens before its caller is registered is delivered rather than
dropped.
There is a test for exactly that, and it fails without the lock:

```
running 1 test
test an_exit_that_happens_before_the_caller_is_registered_is_not_lost ... FAILED

failures:

---- an_exit_that_happens_before_the_caller_is_registered_is_not_lost stdout ----

thread 'an_exit_that_happens_before_the_caller_is_registered_is_not_lost' (321463) panicked at apps/nesinit/tests/reaping.rs:171:9:
the exit was dropped on the way through
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    an_exit_that_happens_before_the_caller_is_registered_is_not_lost

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 4 filtered out; finished in 5.20s

error: test failed, to rerun pass `-p nesinit --test reaping`
```

Also, and still not verified: the ordered shutdown's real calls now
include
signalling and waiting for one pid rather than every child, and none of
that
has run in a guest either.

`071241f` — a pid stops being the workload's the moment it is reaped, so
a stop
or a kill during shutdown can no longer land on whatever the kernel gave
that
number to next. One window is left, between the reap and the delivery,
and
closing it needs a handle the kernel keeps rather than a number — noted
below
rather than papered over.

- **A pid is still a number here.** Between a reap and the exit being
handed
on, a freed pid is briefly treated as the workload's. Nothing has hit
that
window, and nothing can until a guest recycles pids under load; a
`pidfd`
  would remove the class of bug rather than narrow it.

<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR adds `nesinit`, a guest PID 1 implementation that reaps orphaned
children, exchanges lifecycle messages over vsock, launches one
workload, and performs ordered shutdown.
- Adds a centralized child reaper and waiter registry.
- Adds workload launch, identity changes, exit reporting, and explicit
rejection of unsupported mounts.
- Adds the lifecycle protocol behind an optional `nesprotocol` feature.
- Adds ordered workload-first shutdown and associated tests.
- Since the previous review, tracks whether the watched workload PID is
still live to reduce stale-PID signaling risk.

<h3>Confidence Score: 5/5</h3>

The PR appears safe to merge based on the accepted findings and the
resolved state of all previous review threads.

No new actionable finding remains after excluding the stale-PID race as
a duplicate of a manually resolved previous thread and confirming that
the stale state left by the shutdown-only wait path is not subsequently
used to signal a workload PID. Previous thread PRRC_kwDOLnCyk87q1Eew was
manually resolved without explanation.

<h3>Important Files Changed</h3>

| Filename | Overview |
|----------|----------|
| apps/nesinit/src/reap.rs | Adds centralized child reaping, waiter
registration, exit delivery, and shared workload-running state. |
| apps/nesinit/src/workload.rs | Implements workload launch and
signaling through a watched PID whose delivered exit marks it inactive.
|
| apps/nesinit/src/main.rs | Wires the reaper, vsock session, workload
handle, and real ordered-shutdown operations together. |
| apps/nesinit/src/session.rs | Implements the versioned one-descriptor
lifecycle exchange and reports one workload exit without restarting it.
|
| apps/nesinit/src/shutdown.rs | Encodes and tests workload-first
shutdown followed by guest services, disk flush, and power-off. |
| crates/nesprotocol/src/lifecycle.rs | Defines the feature-gated
lifecycle protocol types shared by the guest and host. |

<h3>Flowchart</h3>

```mermaid
%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[nesinit starts] --> B[Become subreaper and adjust OOM score]
    B --> C[Connect to host over vsock]
    C --> D[Send protocol version]
    D --> E[Receive boot descriptor]
    E --> F[Start workload and register PID]
    F --> G{Session event}
    G -->|Workload exits| H[Reaper delivers exit]
    H --> I[Report workload exit]
    G -->|Stop or channel closes| J[Signal workload]
    G -->|Shutdown or session ends| K[Ordered shutdown]
    I --> K
    J --> K
    K --> L[Stop workload]
    L --> M[Stop remaining processes]
    M --> N[Sync disks]
    N --> O[Power off]
```

<sub>Reviews (3): Last reviewed commit: ["fix(nesinit): a pid stops
being the
work..."](071241f944)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=60585100)</sub>

<!-- /greptile_comment -->
This commit is contained in:
KAAL1 (Bingus)
2026-09-05 10:07:32 +03:00
committed by GitHub
parent 4eff67a11a 071241f944
commit 736c0013e9
14 changed files with 1841 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
Cargo.lock generated
View File

@@ -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"

2
Cargo.toml
View File

@@ -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"] }

28
apps/nesinit/Cargo.toml Normal file
View File

@@ -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 }

71
apps/nesinit/README.md Normal file
View File

@@ -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.

14
apps/nesinit/src/lib.rs Normal file
View File

@@ -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;

190
apps/nesinit/src/main.rs Normal file
View File

@@ -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<Outcome> {
// 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));
}
}
}

180
apps/nesinit/src/reap.rs Normal file
View File

@@ -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<Mutex<HashMap<i32, Watch>>>);
/// The registry's half of one watched child.
struct Watch {
exit: oneshot::Sender<Exit>,
running: Arc<AtomicBool>,
}
/// 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<AtomicBool>,
exit: Option<oneshot::Receiver<Exit>>,
}
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<oneshot::Receiver<Exit>> {
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<F>(&self, spawn: F) -> io::Result<Watched>
where
F: FnOnce() -> io::Result<i32>,
{
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<i32, Watch>> {
self.0
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
}

408
apps/nesinit/src/session.rs Normal file
View File

@@ -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<C, W>(channel: C, workload: &mut W) -> std::io::Result<Outcome>
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<crate::workload::Exited> = 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<W: Workload>(
descriptor: &BootDescriptor,
workload: &mut W,
) -> Result<crate::workload::Exited, Failure> {
workload.mount(&descriptor.mounts)?;
workload.start(&descriptor.exec)
}
async fn send<W>(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<BufReader<DuplexStream>>,
}
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"
);
}
}

135
apps/nesinit/src/shutdown.rs Normal file
View File

@@ -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<M: Machine>(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"));
}
}

267
apps/nesinit/src/workload.rs Normal file
View File

@@ -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<String>) -> Self {
Self {
reason: reason.into(),
}
}
}
/// Resolves when the workload ends.
pub type Exited = Pin<Box<dyn Future<Output = io::Result<Exit>> + 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<Exited, Failure>;
/// 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<Watched>,
}
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<Exited, Failure> {
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<Vec<Mount>>,
pub started: Vec<Exec>,
pub stops: usize,
pub mount_failure: Option<Failure>,
pub start_failure: Option<Failure>,
exit: Exit,
on_stop: Option<oneshot::Sender<Exit>>,
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<Exited, Failure> {
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);
}
}
}
}

217
apps/nesinit/tests/reaping.rs Normal file
View File

@@ -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<nesprotocol::lifecycle::Exit> {
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",
);
}

10
crates/nesprotocol/Cargo.toml
View File

@@ -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 }

2
crates/nesprotocol/src/lib.rs
View File

@@ -4,6 +4,8 @@
pub mod datagram;
pub mod input;
#[cfg(feature = "lifecycle")]
pub mod lifecycle;
pub mod reliable;
pub mod stats;

267
crates/nesprotocol/src/lifecycle.rs Normal file
View File

@@ -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<String>,
/// Environment for the process. Sorted, so two descriptors that say the
/// same thing serialize identically.
#[serde(default)]
pub env: BTreeMap<String, String>,
/// Working directory. `None` means the root of the guest filesystem.
#[serde(default)]
pub cwd: Option<String>,
/// 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<Mount>,
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<i32>,
#[serde(skip_serializing_if = "Option::is_none", default)]
pub signal: Option<i32>,
}
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<BootDescriptor>,
},
/// 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<T: Serialize>(message: &T) -> Result<String, serde_json::Error> {
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<T: DeserializeOwned>(line: &str) -> Result<T, serde_json::Error> {
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);
}
}