surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Isolate watcher queues and selectively reload resolved service binaries

Browse Source
This commit is contained in:
surma
2026-11-06 20:00:29 +00:00
parent b39bbd4f18
commit 457a966068
10 changed files with 247 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
host/src/process/watcher.rs
+11 -10
View File
@@ -78,17 +78,18 @@ impl HostWatcher for State {
.root
.observer()
.ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?;
let paths: Vec<String> = paths
.iter()
.map(|p| {
sausage_vfs::resolve("/", p)
.trim_end_matches('/')
.to_owned()
})
.map(|p| if p.is_empty() { "/".into() } else { p })
.collect();
Ok(self.table.push(Watcher {
paths: paths
.iter()
.map(|p| {
sausage_vfs::resolve("/", p)
.trim_end_matches('/')
.to_owned()
})
.map(|p| if p.is_empty() { "/".into() } else { p })
.collect(),
events: observer.subscribe(),
events: observer.subscribe(paths.clone()),
paths,
pending: Vec::new(),
})?)
}
host/tests/init.rs
+71 -1
View File
@@ -154,6 +154,74 @@ fn lines(path: &Path) -> Vec<String> {
.map(str::to_owned)
.collect()
}
#[tokio::test]
async fn unrelated_bursts_and_watched_overflow_only_reload_changed_units() {
let _serial = SERIAL.lock().await;
let (temp, runtime, root) = world();
let supervisor = supervise(runtime.clone());
let path = temp.path().join("world");
wait(|| path.join("var/log/journal").exists()).await;
for name in ["changed", "unchanged"] {
write(
&root,
&format!("/run/systemd/transient/{name}.service"),
format!("[Service]\nExecStart=probe server /{name}.log\n").as_bytes(),
)
.await;
assert_eq!(
run(&runtime, &["/bin/systemctl.wasm", "start", name])
.await
.0,
0
);
wait(|| lines(&path.join(format!("{name}.log"))) == ["start"]).await;
}
tokio::task::unconstrained(async {
for _ in 0..4000 {
write(&root, "/run/unrelated", b"noise").await;
}
})
.await;
assert_eq!(
run(&runtime, &["/bin/systemctl.wasm", "start", "unchanged"])
.await
.0,
0
);
assert_eq!(lines(&path.join("changed.log")), ["start"]);
assert_eq!(lines(&path.join("unchanged.log")), ["start"]);
// This single-thread executor cannot schedule init during ready tmpfs writes.
// The watched queue therefore overflows before init can consume the edits.
tokio::task::unconstrained(async {
write(
&root,
"/run/systemd/transient/changed.service",
b"[Service]\nExecStart=probe server /changed.log\nEnvironment=REV=2\n",
)
.await;
for _ in 0..1000 {
write(&root, "/run/systemd/transient/noise", b"noise").await;
}
})
.await;
wait(|| lines(&path.join("changed.log")) == ["start", "stop", "start"]).await;
assert_eq!(
run(&runtime, &["/bin/systemctl.wasm", "start", "unchanged"])
.await
.0,
0
);
assert_eq!(lines(&path.join("unchanged.log")), ["start"]);
// Bare names also watch the resolved executable, not the literal name.
let mut bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap();
bytes.extend_from_slice(&[0, 2, 1, b'x']);
write(&root, "/bin/replacement.wasm", &bytes).await;
root.rename("/bin/replacement.wasm", "/bin/probe.wasm")
.await
.unwrap();
wait(|| lines(&path.join("unchanged.log")) == ["start", "stop", "start"]).await;
supervisor.shutdown().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn watcher_and_child_exit_pollable() {
let _serial = SERIAL.lock().await;
@@ -276,7 +344,9 @@ async fn live_unit_changes_and_binary_replacement() {
)
.await;
wait(|| lines(&path.join("live.log")) == ["start", "stop", "start"]).await;
let bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap();
let mut bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap();
// Add a valid custom section, so the replacement differs in content.
bytes.extend_from_slice(&[0, 2, 1, b'x']);
write(&root, "/bin/replacement.wasm", &bytes).await;
root.rename("/bin/replacement.wasm", "/bin/probe.wasm")
.await
userland/Cargo.lock
Generated
+1
View File
@@ -4074,6 +4074,7 @@ dependencies = [
name = "sausage-spawn"
version = "0.1.0"
dependencies = [
"wasi 0.14.7+wasi-0.2.4",
"wit-bindgen 0.46.0",
]
userland/init/README.md
+10 -2
View File
@@ -16,7 +16,14 @@ When init receives EOF, it stops dependents before prerequisites.
Enable records are files under `default.target.wants` or `multi-user.target.wants` in the persistent unit directory.
The `[Install]` section supplies `WantedBy` for `systemctl enable`.
Init observes VFS changes and applies unit edits without `daemon-reload`.
A replacement of a running unit's executable restarts that unit.
A change to a running unit's executable content restarts that unit.
Init resolves bare executable names through `PATH`, which defaults to `/usr/local/bin:/bin`.
Each directory tries the bare name before the `.wasm` suffix.
Init watches those candidate paths and compares the resolved executable's content hash.
After an overflow, init rescans definitions and executable hashes.
It restarts only units with a changed definition or executable.
Unrelated world writes do not enter these watcher queues.
Init and request clients block on pollables, with timers for lifecycle deadlines.
## Control protocol
@@ -71,5 +78,6 @@ The cap is 1 MiB. Init keeps one rotated file with the `.1` suffix.
A watcher matches each supplied path and all its descendants.
Its pollable remains ready while changes remain pending.
`changes()` returns and clears the pending changes.
The queue holds at most 256 changes. An `overflow` event requires a rescan.
Each watcher filters paths before its independent queue holds at most 256 changes.
An `overflow` event requires a rescan.
Observers report successful VFS mutations, not direct host filesystem writes outside the VFS.
userland/init/src/lib.rs
+5 -2
View File
@@ -131,6 +131,10 @@ pub fn request(action: &str, unit: &str) -> io::Result<Reply> {
})?,
)?;
let deadline = std::time::Instant::now() + Duration::from_secs(60);
let timeout = wasi::clocks::monotonic_clock::subscribe_duration(
Duration::from_secs(60).as_nanos() as u64,
);
let changed = watcher.subscribe();
let mut acknowledged = false;
loop {
if !acknowledged && let Ok(bytes) = fs::read(&reply) {
@@ -167,9 +171,8 @@ pub fn request(action: &str, unit: &str) -> io::Result<Reply> {
"init did not reply",
));
}
// The short timer also handles a missing or failed init.
sausage_spawn::poll(&[&changed, &timeout]);
let _ = watcher.changes();
std::thread::sleep(Duration::from_millis(20));
}
}
#[cfg(test)]
userland/init/src/main.rs
+94 -20
View File
@@ -5,9 +5,67 @@ use std::{
collections::{BTreeMap, BTreeSet, VecDeque},
fs,
io::{self, Write},
time::{Duration, Instant},
time::Instant,
};
fn program_paths(unit: &Unit) -> Vec<String> {
let program = shlex::split(unit.get("Service.ExecStart"))
.and_then(|args| args.first().cloned())
.unwrap_or_default();
if program.contains('/') {
return vec![program];
}
let path = shlex::split(unit.get("Service.Environment"))
.unwrap_or_default()
.into_iter()
.find_map(|v| v.strip_prefix("PATH=").map(str::to_owned))
.unwrap_or_else(|| "/usr/local/bin:/bin".into());
path.split(':')
.flat_map(|dir| [format!("{dir}/{program}"), format!("{dir}/{program}.wasm")])
.collect()
}
fn binary(unit: &Unit) -> Option<(String, u64)> {
use std::hash::{Hash, Hasher};
program_paths(unit).into_iter().find_map(|path| {
let bytes = fs::read(&path).ok()?;
let mut hash = std::collections::hash_map::DefaultHasher::new();
bytes.hash(&mut hash);
Some((path, hash.finish()))
})
}
fn wait_events(
services: &BTreeMap<String, Service>,
mut pollables: Vec<sausage_spawn::Pollable>,
tick_at: Instant,
) {
let now = Instant::now();
let mut deadline = None;
for service in services.values() {
if let Some(child) = &service.child {
if child.try_wait().is_none() {
pollables.push(child.subscribe());
}
for stream in [&child.stdout, &child.stderr].into_iter().flatten() {
pollables.push(stream.subscribe());
}
}
let time = service
.stopping
.or_else(|| (service.desired && service.child.is_none()).then_some(service.next));
if let Some(time) = time.filter(|time| *time > tick_at) {
deadline = Some(deadline.map_or(time, |old: Instant| old.min(time)));
}
}
if let Some(deadline) = deadline {
pollables.push(wasi::clocks::monotonic_clock::subscribe_duration(
deadline.saturating_duration_since(now).as_nanos() as u64,
));
}
if !pollables.is_empty() {
sausage_spawn::poll(&pollables.iter().collect::<Vec<_>>());
}
}
struct Service {
binary: Option<(String, u64)>,
unit: Unit,
child: Option<Child>,
state: String,
@@ -20,6 +78,7 @@ struct Service {
impl Service {
fn new(unit: Unit) -> Self {
Self {
binary: binary(&unit),
unit,
child: None,
state: "inactive".into(),
@@ -68,7 +127,12 @@ impl Service {
self.desired = false;
return;
};
let mut command = Command::new(&argv[0]);
let program = self
.binary
.as_ref()
.map(|(path, _)| path.as_str())
.unwrap_or(&argv[0]);
let mut command = Command::new(program);
command
.args(&argv[1..])
.new_group()
@@ -244,7 +308,9 @@ fn scan(services: &mut BTreeMap<String, Service>) {
let service = services
.entry(name.clone())
.or_insert_with(|| Service::new(unit.clone()));
if service.unit != unit {
let binary = binary(&unit);
if service.unit != unit || service.binary != binary {
service.binary = binary;
let desired = service.desired || service.state == "active";
service.stop();
service.unit = unit;
@@ -326,7 +392,8 @@ fn main() -> io::Result<()> {
] {
fs::create_dir_all(dir)?;
}
let watcher = sausage_spawn::Watcher::new(&["/".into()]);
let mut paths = vec![UNITS.into(), TRANSIENT.into(), format!("{RUN}/requests")];
let mut watcher = sausage_spawn::Watcher::new(&paths);
let stdin = wasi::cli::stdin::get_stdin();
let mut services = BTreeMap::new();
scan(&mut services);
@@ -336,23 +403,10 @@ fn main() -> io::Result<()> {
}
let changes = watcher.changes();
if !changes.is_empty() {
for service in services.values_mut() {
let program = shlex::split(service.unit.get("Service.ExecStart"))
.and_then(|a| a.first().cloned())
.unwrap_or_default();
if service.child.is_some()
&& changes.iter().any(|c| {
c.path == program || matches!(c.kind, sausage_spawn::ChangeKind::Overflow)
})
{
service.stop();
service.desired = true;
service.next = Instant::now();
}
}
scan(&mut services);
}
controls(&mut services);
let tick_at = Instant::now();
let names: Vec<_> = services.keys().cloned().collect();
for name in &names {
services.get_mut(name).unwrap().tick(name);
@@ -388,7 +442,24 @@ fn main() -> io::Result<()> {
for (name, service) in &services {
service.publish(name);
}
std::thread::sleep(Duration::from_millis(20));
let mut next_paths = vec![UNITS.into(), TRANSIENT.into(), format!("{RUN}/requests")];
for service in services.values() {
next_paths.extend(program_paths(&service.unit));
}
next_paths.sort();
next_paths.dedup();
if paths != next_paths {
paths = next_paths;
watcher = sausage_spawn::Watcher::new(&paths);
// Close the snapshot/subscription race before blocking.
scan(&mut services);
continue;
}
wait_events(
&services,
vec![watcher.subscribe(), stdin.subscribe()],
tick_at,
);
}
// Stop dependents before their prerequisites. A cycle stops together.
while services.values().any(|s| s.child.is_some()) {
@@ -409,13 +480,16 @@ fn main() -> io::Result<()> {
if leaves.is_empty() {
leaves = live;
}
let tick_at = Instant::now();
for name in leaves {
let service = services.get_mut(&name).unwrap();
service.stop();
service.tick(&name);
service.publish(&name);
}
std::thread::sleep(Duration::from_millis(20));
if services.values().any(|s| s.child.is_some()) {
wait_events(&services, Vec::new(), tick_at);
}
}
Ok(())
}
userland/init/src/systemd-run.rs
+5 -2
View File
@@ -1,6 +1,6 @@
//! Creates transient unit files and optionally waits for their final state.
use sausage_init::*;
use std::{fs, io, time::Duration};
use std::{fs, io};
fn run() -> io::Result<i32> {
let mut args = std::env::args().skip(1).peekable();
let mut unit = format!(
@@ -65,6 +65,8 @@ fn run() -> io::Result<i32> {
return Ok(reply.code);
}
if wait {
let watcher = sausage_spawn::Watcher::new(&[format!("{RUN}/state/{unit}")]);
let changed = watcher.subscribe();
loop {
if let Ok(bytes) = fs::read(format!("{RUN}/state/{unit}")) {
let status: Status = serde_json::from_slice(&bytes)?;
@@ -72,7 +74,8 @@ fn run() -> io::Result<i32> {
return Ok(code);
}
}
std::thread::sleep(Duration::from_millis(20));
changed.block();
let _ = watcher.changes();
}
}
Ok(0)
userland/spawn/Cargo.toml
+1
View File
@@ -5,3 +5,4 @@ edition = "2024"
[dependencies]
wit-bindgen = "=0.46.0"
wasi = "0.14"
userland/spawn/src/lib.rs
+2 -2
View File
@@ -3,12 +3,12 @@
//! Pipe callers must drain stdout and stderr before wait to avoid bounded-pipe deadlocks.
use std::{collections::BTreeMap, io, path::{Path, PathBuf}};
mod bindings {
wit_bindgen::generate!({ path: "../../host/wit", world: "program", generate_all });
wit_bindgen::generate!({ path: "../../host/wit", world: "program", generate_all, with: { "wasi:io/poll@0.2.12": ::wasi::io::poll } });
}
use bindings::sausage::process::host;
pub use bindings::wasi::io::streams::{InputStream, OutputStream};
pub use host::{Stdio, Watcher, Change, ChangeKind};
pub use bindings::wasi::io::poll::{Pollable, poll};
pub use wasi::io::poll::{Pollable, poll};
/// Hard-stop a process group by a live process ID.
pub fn kill(pid: u32) -> io::Result<()> { host::kill(pid).map_err(io::Error::other) }
vfs/src/observe.rs
+47 -5
View File
@@ -23,12 +23,16 @@ enum Held {
type TrackedPath = Mutex<Option<String>>;
type Path = Arc<TrackedPath>;
type Callback = dyn Fn(&str) + Send + Sync;
type Subscription = (
Vec<String>,
tokio::sync::broadcast::Sender<(String, ChangeKind)>,
);
/// Shared observation for mount namespaces. Paths supplied to the callback are virtual paths.
/// Callbacks run after filesystem locks have been released; they must not block.
#[derive(Clone)]
pub struct Observer {
changed: Arc<Callback>,
events: tokio::sync::broadcast::Sender<(String, ChangeKind)>,
events: Arc<Mutex<Vec<Subscription>>>,
paths: Arc<Mutex<Vec<Weak<TrackedPath>>>>,
// Only open-registration versus rename/remove needs ordering. Lock the affected
// subtrees, not all IO: a blocked open must not stall unrelated namespaces.
@@ -38,18 +42,31 @@ impl Observer {
pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self {
Self {
changed: Arc::new(changed),
events: tokio::sync::broadcast::channel(256).0,
events: Arc::default(),
paths: Arc::default(),
namespace: Arc::default(),
}
}
/// Subscribe before the first mutation of interest. Lag means callers must rescan.
pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> {
self.events.subscribe()
pub fn subscribe(
&self,
paths: Vec<String>,
) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> {
let (sender, receiver) = tokio::sync::broadcast::channel(256);
self.events.lock().unwrap().push((paths, sender));
receiver
}
fn notify(&self, path: &str, kind: ChangeKind) {
(self.changed)(path);
let _ = self.events.send((path.to_owned(), kind));
self.events.lock().unwrap().retain(|(paths, sender)| {
if sender.receiver_count() == 0 {
return false;
}
if paths.iter().any(|base| under(path, base)) {
let _ = sender.send((path.to_owned(), kind));
}
true
});
}
/// Wrap a layer at its canonical namespace prefix (e.g. `/run`).
pub fn layer(&self, layer: Arc<dyn Layer>, prefix: &str, process: Process) -> Arc<dyn Layer> {
@@ -334,6 +351,31 @@ impl File for Handle {
#[cfg(test)]
mod tests {
#[test]
fn queues_filter_before_the_bound_and_overflow_independently() {
let observer = super::Observer::new(|_| {});
let mut a = observer.subscribe(vec!["/watched/a".into()]);
let mut b = observer.subscribe(vec!["/watched/b".into()]);
for _ in 0..4000 {
observer.notify("/elsewhere", super::ChangeKind::Modified);
}
assert!(matches!(
a.try_recv(),
Err(tokio::sync::broadcast::error::TryRecvError::Empty)
));
for _ in 0..300 {
observer.notify("/watched/a", super::ChangeKind::Modified);
}
assert!(matches!(
a.try_recv(),
Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_))
));
assert!(matches!(
b.try_recv(),
Err(tokio::sync::broadcast::error::TryRecvError::Empty)
));
}
use super::*;
use tokio::sync::mpsc;
#[test]