498 lines
16 KiB
Rust
498 lines
16 KiB
Rust
//! Runs the world's unit state machine, journal drains, and file-based control requests.
|
|
use sausage_init::*;
|
|
use sausage_spawn::{Child, Command, Stdio};
|
|
use std::{
|
|
collections::{BTreeMap, BTreeSet, VecDeque},
|
|
fs,
|
|
io::{self, Write},
|
|
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,
|
|
code: Option<i32>,
|
|
desired: bool,
|
|
starts: VecDeque<Instant>,
|
|
next: Instant,
|
|
stopping: Option<Instant>,
|
|
}
|
|
impl Service {
|
|
fn new(unit: Unit) -> Self {
|
|
Self {
|
|
binary: binary(&unit),
|
|
unit,
|
|
child: None,
|
|
state: "inactive".into(),
|
|
code: None,
|
|
desired: false,
|
|
starts: VecDeque::new(),
|
|
next: Instant::now(),
|
|
stopping: None,
|
|
}
|
|
}
|
|
fn stop(&mut self) {
|
|
self.desired = false;
|
|
if let Some(child) = &mut self.child {
|
|
child.stdin.take();
|
|
self.stopping
|
|
.get_or_insert(Instant::now() + self.unit.seconds("Service.TimeoutStopSec", 10.));
|
|
self.state = "deactivating".into();
|
|
} else {
|
|
self.state = "inactive".into();
|
|
}
|
|
}
|
|
fn start(&mut self) {
|
|
let now = Instant::now();
|
|
let interval = self.unit.seconds("Service.StartLimitIntervalSec", 10.);
|
|
while self
|
|
.starts
|
|
.front()
|
|
.is_some_and(|t| now.duration_since(*t) >= interval)
|
|
{
|
|
self.starts.pop_front();
|
|
}
|
|
let burst = self
|
|
.unit
|
|
.get("Service.StartLimitBurst")
|
|
.parse()
|
|
.unwrap_or(5);
|
|
if burst != 0 && self.starts.len() >= burst {
|
|
self.state = "failed".into();
|
|
self.desired = false;
|
|
return;
|
|
}
|
|
self.starts.push_back(now);
|
|
let Some(argv) = shlex::split(self.unit.get("Service.ExecStart")).filter(|a| !a.is_empty())
|
|
else {
|
|
self.state = "failed".into();
|
|
self.desired = false;
|
|
return;
|
|
};
|
|
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()
|
|
.stdin(Stdio::Pipe)
|
|
.stdout(Stdio::Pipe)
|
|
.stderr(Stdio::Pipe);
|
|
let cwd = self.unit.get("Service.WorkingDirectory");
|
|
command.current_dir(if cwd.is_empty() { "/" } else { cwd });
|
|
for assignment in shlex::split(self.unit.get("Service.Environment")).unwrap_or_default() {
|
|
if let Some((key, value)) = assignment.split_once('=') {
|
|
command.env(key, value);
|
|
}
|
|
}
|
|
match command.spawn() {
|
|
Ok(child) => {
|
|
self.child = Some(child);
|
|
self.state = if self.unit.get("Service.Type") == "oneshot" {
|
|
"activating"
|
|
} else {
|
|
"active"
|
|
}
|
|
.into();
|
|
self.code = None;
|
|
}
|
|
Err(_) => {
|
|
self.code = Some(127);
|
|
self.exited(127);
|
|
}
|
|
}
|
|
}
|
|
fn exited(&mut self, code: i32) {
|
|
self.code = Some(code);
|
|
if self.stopping.take().is_some() {
|
|
self.state = if self.desired {
|
|
"activating"
|
|
} else {
|
|
"inactive"
|
|
}
|
|
.into();
|
|
if self.desired {
|
|
self.code = None;
|
|
}
|
|
return;
|
|
}
|
|
self.state = if code != 0 {
|
|
"failed"
|
|
} else if matches!(
|
|
self.unit.get("Service.RemainAfterExit"),
|
|
"yes" | "true" | "1"
|
|
) {
|
|
"active"
|
|
} else {
|
|
"inactive"
|
|
}
|
|
.into();
|
|
let restart = self.unit.get("Service.Restart");
|
|
if restart == "always" || restart == "on-failure" && code != 0 {
|
|
self.next = Instant::now() + self.unit.seconds("Service.RestartSec", 0.1);
|
|
} else {
|
|
self.desired = false;
|
|
}
|
|
}
|
|
fn tick(&mut self, name: &str) {
|
|
if let Some(child) = &mut self.child {
|
|
for slot in [&mut child.stdout, &mut child.stderr] {
|
|
if let Some(stream) = slot {
|
|
loop {
|
|
match stream.read(8192) {
|
|
Ok(bytes) if !bytes.is_empty() => journal(name, &bytes),
|
|
Ok(_) => break,
|
|
Err(_) => {
|
|
*slot = None;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if child.try_wait().is_none() && self.stopping.is_some_and(|t| Instant::now() >= t) {
|
|
let _ = sausage_spawn::kill(child.id());
|
|
}
|
|
if let Some(code) = child.try_wait() {
|
|
if child.stdout.is_none() && child.stderr.is_none() {
|
|
self.child = None;
|
|
self.exited(code);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
fn publish(&self, name: &str) {
|
|
let status = Status {
|
|
state: self.state.clone(),
|
|
pid: self.child.as_ref().map(Child::id),
|
|
code: self.code,
|
|
};
|
|
let bytes = serde_json::to_vec(&status).unwrap();
|
|
let path = format!("{RUN}/state/{name}");
|
|
if fs::read(&path).ok().as_deref() != Some(bytes.as_slice()) {
|
|
let _ = atomic(&path, &bytes);
|
|
}
|
|
}
|
|
}
|
|
fn journal(name: &str, bytes: &[u8]) {
|
|
let path = format!("/var/log/journal/{name}.log");
|
|
if fs::metadata(&path).is_ok_and(|m| m.len() + bytes.len() as u64 > 1024 * 1024) {
|
|
let _ = fs::rename(&path, format!("{path}.1"));
|
|
}
|
|
if let Ok(mut f) = fs::OpenOptions::new().create(true).append(true).open(path) {
|
|
let _ = f.write_all(bytes);
|
|
}
|
|
}
|
|
fn activate(name: &str, services: &mut BTreeMap<String, Service>, seen: &mut BTreeSet<String>) {
|
|
if !seen.insert(name.into()) {
|
|
return;
|
|
}
|
|
let Some(service) = services.get(name) else {
|
|
return;
|
|
};
|
|
let deps = [
|
|
service.unit.words("Unit.Requires"),
|
|
service.unit.words("Unit.Wants"),
|
|
]
|
|
.concat();
|
|
for dep in deps {
|
|
activate(&dep, services, seen);
|
|
}
|
|
let service = services.get_mut(name).unwrap();
|
|
service.desired = true;
|
|
if service.child.is_none() {
|
|
service.state = "activating".into();
|
|
service.code = None;
|
|
}
|
|
service.next = Instant::now();
|
|
}
|
|
fn predecessors(name: &str, services: &BTreeMap<String, Service>) -> Vec<String> {
|
|
let mut deps = services[name].unit.words("Unit.After");
|
|
deps.extend(services[name].unit.words("Unit.Requires"));
|
|
for (other, service) in services {
|
|
if service.unit.words("Unit.Before").iter().any(|s| s == name) {
|
|
deps.push(other.clone());
|
|
}
|
|
}
|
|
deps
|
|
}
|
|
fn scan(services: &mut BTreeMap<String, Service>) {
|
|
let mut found = BTreeMap::new();
|
|
let mut present = BTreeSet::new();
|
|
for dir in [TRANSIENT, UNITS] {
|
|
for entry in fs::read_dir(dir).into_iter().flatten().flatten() {
|
|
let name = entry.file_name().to_string_lossy().into_owned();
|
|
if !name.ends_with(".service") {
|
|
continue;
|
|
}
|
|
present.insert(name.clone());
|
|
if let Ok(source) = fs::read_to_string(entry.path()) {
|
|
if let Ok(unit) = Unit::parse(source) {
|
|
found.insert(name, unit);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for (name, service) in services.iter_mut() {
|
|
if !present.contains(name) {
|
|
service.stop();
|
|
}
|
|
}
|
|
services.retain(|name, service| present.contains(name) || service.child.is_some());
|
|
let mut new_enabled = Vec::new();
|
|
for (name, unit) in found {
|
|
if !services.contains_key(&name) && enabled(&name) {
|
|
new_enabled.push(name.clone());
|
|
}
|
|
let service = services
|
|
.entry(name.clone())
|
|
.or_insert_with(|| Service::new(unit.clone()));
|
|
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;
|
|
service.desired = desired;
|
|
service.starts.clear();
|
|
service.next = Instant::now();
|
|
}
|
|
}
|
|
for name in new_enabled {
|
|
activate(&name, services, &mut BTreeSet::new());
|
|
}
|
|
}
|
|
fn controls(services: &mut BTreeMap<String, Service>) {
|
|
for entry in fs::read_dir(format!("{RUN}/requests"))
|
|
.into_iter()
|
|
.flatten()
|
|
.flatten()
|
|
{
|
|
let filename = entry.file_name().to_string_lossy().into_owned();
|
|
if !filename.ends_with(".json") {
|
|
continue;
|
|
}
|
|
let Ok(bytes) = fs::read(entry.path()) else {
|
|
continue;
|
|
};
|
|
let Ok(request) = serde_json::from_slice::<Request>(&bytes) else {
|
|
continue;
|
|
};
|
|
// A unit edit can arrive after the event loop's scan but before this request.
|
|
scan(services);
|
|
let mut reply = Reply {
|
|
code: 0,
|
|
text: String::new(),
|
|
};
|
|
if !services.contains_key(&request.unit) {
|
|
reply.code = 5;
|
|
reply.text = "unit not found".into();
|
|
} else {
|
|
match request.action.as_str() {
|
|
"start" => {
|
|
services.get_mut(&request.unit).unwrap().starts.clear();
|
|
activate(&request.unit, services, &mut BTreeSet::new());
|
|
}
|
|
"stop" => services.get_mut(&request.unit).unwrap().stop(),
|
|
"restart" => {
|
|
let service = services.get_mut(&request.unit).unwrap();
|
|
service.stop();
|
|
service.desired = true;
|
|
if service.child.is_none() {
|
|
service.state = "activating".into();
|
|
service.code = None;
|
|
}
|
|
service.starts.clear();
|
|
service.next = Instant::now();
|
|
}
|
|
_ => {
|
|
reply.code = 1;
|
|
reply.text = "unknown action".into();
|
|
}
|
|
}
|
|
}
|
|
if let Some(service) = services.get(&request.unit) {
|
|
service.publish(&request.unit);
|
|
}
|
|
let id = filename.trim_end_matches(".json");
|
|
let _ = atomic(
|
|
&format!("{RUN}/replies/{id}"),
|
|
&serde_json::to_vec(&reply).unwrap(),
|
|
);
|
|
let _ = fs::remove_file(entry.path());
|
|
}
|
|
}
|
|
fn main() -> io::Result<()> {
|
|
for dir in [
|
|
UNITS,
|
|
TRANSIENT,
|
|
"/run/systemd/requests",
|
|
"/run/systemd/replies",
|
|
"/run/systemd/state",
|
|
"/var/log/journal",
|
|
] {
|
|
fs::create_dir_all(dir)?;
|
|
}
|
|
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);
|
|
loop {
|
|
if stdin.read(1).is_err() {
|
|
break;
|
|
}
|
|
let changes = watcher.changes();
|
|
if !changes.is_empty() {
|
|
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);
|
|
}
|
|
for name in &names {
|
|
let service = &services[name];
|
|
if !service.desired
|
|
|| service.child.is_some()
|
|
|| Instant::now() < service.next
|
|
|| service.stopping.is_some()
|
|
{
|
|
continue;
|
|
}
|
|
let required = service.unit.words("Unit.Requires");
|
|
if required
|
|
.iter()
|
|
.any(|dep| !services.contains_key(dep) || services[dep].state == "failed")
|
|
{
|
|
let service = services.get_mut(name).unwrap();
|
|
service.state = "failed".into();
|
|
service.desired = false;
|
|
continue;
|
|
}
|
|
if predecessors(name, &services).iter().any(|dep| {
|
|
services.get(dep).is_some_and(|s| {
|
|
s.state == "activating" || s.desired && s.child.is_none() && s.code.is_none()
|
|
})
|
|
}) {
|
|
continue;
|
|
}
|
|
services.get_mut(name).unwrap().start();
|
|
}
|
|
for (name, service) in &services {
|
|
service.publish(name);
|
|
}
|
|
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()) {
|
|
let live: Vec<_> = services
|
|
.iter()
|
|
.filter(|(_, s)| s.child.is_some())
|
|
.map(|(n, _)| n.clone())
|
|
.collect();
|
|
let mut leaves: Vec<_> = live
|
|
.iter()
|
|
.filter(|name| {
|
|
!live
|
|
.iter()
|
|
.any(|other| other != *name && predecessors(other, &services).contains(name))
|
|
})
|
|
.cloned()
|
|
.collect();
|
|
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);
|
|
}
|
|
if services.values().any(|s| s.child.is_some()) {
|
|
wait_events(&services, Vec::new(), tick_at);
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|