use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::process::ChildStdin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;
use std::time::Instant;
use serde::Deserialize;
#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Config {
pub name: String,
pub program: PathBuf,
pub args: Vec<String>,
pub after: Option<String>,
pub ready: String,
pub log: PathBuf,
pub store: Option<String>,
pub give_up_after: u64,
#[serde(skip)]
pub(super) given_up: Arc<AtomicBool>,
}
pub(super) fn check(configs: &[Config], stores: &BTreeMap<String, PathBuf>) -> Result<(), String> {
let mut names = BTreeSet::new();
let mut served = BTreeSet::new();
for config in configs {
let invalid = |field, why| format!("door_child_{field}_invalid: {}: {why}", config.name);
if config.name.is_empty()
|| !config
.name
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
|| names.contains(&config.name)
{
return Err(invalid(
"name",
"names must be unique lowercase letters, digits or -",
));
}
if config
.after
.as_ref()
.is_some_and(|after| !names.contains(after))
{
return Err(invalid("after", "after must name an earlier child"));
}
if let Some(store) = &config.store
&& (!stores.contains_key(store) || !served.insert(store))
{
return Err(invalid(
"store",
"store must name a configured store served by only one child",
));
}
if !config.program.is_absolute() {
return Err(invalid("program", "program must be absolute"));
}
if !config.log.is_absolute() {
return Err(invalid("log", "log must be absolute"));
}
if config.ready.is_empty() {
return Err(invalid("ready", "ready prefix must not be empty"));
}
if config.give_up_after == 0 {
return Err(format!(
"door_child_give_up_after_zero: {}: a child needs at least one start",
config.name
));
}
names.insert(&config.name);
}
Ok(())
}
#[path = "children_io.rs"]
mod io;
#[path = "children_run.rs"]
mod run;
#[derive(Clone, Copy)]
struct Launch {
generation: u64,
started: Instant,
}
#[derive(Default, Eq, PartialEq)]
enum Phase {
#[default]
Waiting,
NotReady,
Ready,
Exited,
GivenUp,
}
#[derive(Default)]
struct Entry {
#[cfg(test)]
starts: u64,
#[cfg(test)]
finished: bool,
phase: Phase,
seen_ready: bool,
pending_ready: usize,
generation: u64,
log_failed: Arc<AtomicBool>,
stdin: Option<ChildStdin>,
}
struct State {
stopped: bool,
entries: Vec<Entry>,
}
struct Shared {
state: Mutex<State>,
changed: Condvar,
}
impl Shared {
fn given_up(&self, index: usize, config: &Config, why: &str) {
eprintln!("door_child_given_up: {} {why}", config.name);
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => {
eprintln!("door_children_state_failed: {poisoned}");
poisoned.into_inner()
}
};
config.given_up.store(true, Ordering::Release);
state.entries[index].phase = Phase::GivenUp;
drop(state.entries[index].stdin.take());
drop(state);
self.changed.notify_all();
}
fn stop(&self) {
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => {
eprintln!("door_children_state_failed: {poisoned}");
poisoned.into_inner()
}
};
state.stopped = true;
for entry in &mut state.entries {
drop(entry.stdin.take());
}
drop(state);
self.changed.notify_all();
}
}
pub(super) struct Children {
shared: Arc<Shared>,
workers: Vec<JoinHandle<()>>,
}
impl Children {
pub(super) fn start(configs: &[Config]) -> Result<Self, String> {
let shared = Arc::new(Shared {
state: Mutex::new(State {
stopped: false,
entries: configs.iter().map(|_| Entry::default()).collect(),
}),
changed: Condvar::new(),
});
let mut children = Self {
shared: Arc::clone(&shared),
workers: Vec::new(),
};
for (index, config) in configs.iter().enumerate() {
let predecessor = config
.after
.as_ref()
.and_then(|after| configs.iter().position(|c| c.name == *after));
let mut failures = 0;
loop {
let owned = Arc::clone(&shared);
let task_config = config.clone();
let worker = std::thread::Builder::new()
.name(format!("door-child-{}", config.name))
.spawn(move || {
if let Err(error) = run::run_child(&owned, index, predecessor, &task_config)
{
eprintln!("{error}");
owned.given_up(
index,
&task_config,
&format!("supervisor failed: {error}"),
);
}
#[cfg(test)]
match owned.state.lock() {
Ok(mut state) => {
state.entries[index].finished = true;
drop(state);
owned.changed.notify_all();
}
Err(error) => eprintln!("door_children_state_failed: {error}"),
}
});
match worker {
Ok(worker) => {
children.workers.push(worker);
break;
}
Err(error) => {
eprintln!(
"door_child_start_failed: {}: waiter thread: {error}",
config.name
);
failures += 1;
if failures == config.give_up_after {
shared.given_up(index, config, &format!("after {failures} exits before ready; last status waiter thread: {error}"));
break;
}
}
}
}
}
let mut state = shared
.state
.lock()
.map_err(|e| format!("door_children_state_failed: {e}"))?;
while !state.entries.iter().all(|entry| {
entry.pending_ready == 0 && (entry.seen_ready || entry.phase == Phase::GivenUp)
}) {
state = shared
.changed
.wait(state)
.map_err(|e| format!("door_children_state_failed: {e}"))?;
}
drop(state);
Ok(children)
}
}
#[cfg(test)]
impl Children {
fn wait_restarted(&self, index: usize) -> Result<u64, String> {
let mut state = self
.shared
.state
.lock()
.map_err(|e| format!("door_children_state_failed: {e}"))?;
while !(state.entries[index].finished
|| state.entries[index].starts >= 2
&& state.entries[index].phase == Phase::Ready
&& state.entries[index].pending_ready == 0)
{
state = self
.shared
.changed
.wait(state)
.map_err(|e| format!("door_children_state_failed: {e}"))?;
}
Ok(state.entries[index].starts)
}
}
#[cfg(test)]
#[path = "children_wait_tests.rs"]
mod wait_tests;
#[cfg(test)]
#[path = "children_ready_tests.rs"]
mod ready_tests;
impl Drop for Children {
fn drop(&mut self) {
self.shared.stop();
for worker in self.workers.drain(..) {
if worker.join().is_err() {
eprintln!("door_child_thread_failed: child waiter panicked");
}
}
}
}