crates/installer/nordisk-pairing/src/duty.rs
//! **A box's own background duty, supervised** (A28 M18; the rule: background work is a named
//! job, visible, its failures said, stopped at the box's stop).
//!
//! This crate is the lean, std-only half of the estate (no async runtime, no job ledger), so a
//! duty here is a named thread with the contract of a jera job, not a ledger row: a product that
//! keeps a jera ledger (holger-live `duty`) runs its own duties on it.
//!
//! * **Visible.** Every duty this process started is in [`running`], with its rounds and its
//! failures.
//! * **A failure is said.** A round that PANICS is an `ALERT <name>` line (through the duty's
//! [`Say`]) and is counted; the next round runs as due. A raw thread ended at the panic,
//! silently, and its duty with it.
//! * **It stops.** At [`Duty::stop`] and [`stop_all`] (the box's stop), and when a round answers
//! [`Round::Done`]. The sleep between rounds wakes at the stop.
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex, OnceLock};
use std::time::{Duration, Instant};
/// What a round answers: due again after its period, due again after a wait of its own (a
/// backoff), or the duty is done.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Round {
Again,
After(Duration),
Done,
}
/// Where a duty says its failures.
pub type Say = Arc<dyn Fn(&str) + Send + Sync>;
/// **One duty, started.**
pub struct Duty {
name: String,
stop: Mutex<bool>,
wake: Condvar,
rounds: AtomicU64,
failures: AtomicU64,
last_failure: Mutex<Option<String>>,
thread: Mutex<Option<std::thread::JoinHandle<()>>>,
}
impl Duty {
/// The duty's name (its thread's name).
pub fn name(&self) -> &str {
&self.name
}
/// The rounds run so far.
pub fn rounds(&self) -> u64 {
self.rounds.load(Ordering::SeqCst)
}
/// The rounds that panicked.
pub fn failures(&self) -> u64 {
self.failures.load(Ordering::SeqCst)
}
/// The last failure, in words.
pub fn last_failure(&self) -> Option<String> {
self.last_failure.lock().unwrap_or_else(|p| p.into_inner()).clone()
}
/// Its thread has ended (stopped, or done).
pub fn is_finished(&self) -> bool {
self.thread.lock().unwrap_or_else(|p| p.into_inner()).as_ref().is_none_or(|t| t.is_finished())
}
/// **Stop**, waiting up to `grace` for the round in flight. True when the thread ended.
pub fn stop(&self, grace: Duration) -> bool {
*self.stop.lock().unwrap_or_else(|p| p.into_inner()) = true;
self.wake.notify_all();
let until = Instant::now() + grace;
while !self.is_finished() {
if Instant::now() >= until {
return false;
}
std::thread::sleep(Duration::from_millis(5));
}
// Joined: the thread has ended, so this does not wait.
if let Some(t) = self.thread.lock().unwrap_or_else(|p| p.into_inner()).take() {
let _ = t.join();
}
true
}
fn stopped(&self) -> bool {
*self.stop.lock().unwrap_or_else(|p| p.into_inner())
}
/// Waits `every`, or less when stopped. True when the duty goes on.
fn pause(&self, every: Duration) -> bool {
let until = Instant::now() + every;
let mut stop = self.stop.lock().unwrap_or_else(|p| p.into_inner());
while !*stop {
let left = until.saturating_duration_since(Instant::now());
if left.is_zero() {
return true;
}
stop = self.wake.wait_timeout(stop, left).unwrap_or_else(|p| p.into_inner()).0;
}
false
}
/// One round, a panic caught: counted and said.
fn round(&self, round: &mut dyn FnMut() -> Round, say: &Say) -> Round {
self.rounds.fetch_add(1, Ordering::SeqCst);
match std::panic::catch_unwind(std::panic::AssertUnwindSafe(round)) {
Ok(r) => r,
Err(p) => {
let why = p.downcast_ref::<&str>().map(|s| s.to_string()).or_else(|| p.downcast_ref::<String>().cloned()).unwrap_or_else(|| "(no words)".into());
self.failures.fetch_add(1, Ordering::SeqCst);
say(&format!("ALERT {} round {} panicked: {why}; the next round runs as due", self.name, self.rounds()));
*self.last_failure.lock().unwrap_or_else(|p| p.into_inner()) = Some(why);
Round::Again
}
}
}
}
fn registry() -> &'static Mutex<Vec<Arc<Duty>>> {
static R: OnceLock<Mutex<Vec<Arc<Duty>>>> = OnceLock::new();
R.get_or_init(Mutex::default)
}
/// **Every duty of this process whose thread still runs.**
pub fn running() -> Vec<Arc<Duty>> {
registry().lock().unwrap_or_else(|p| p.into_inner()).iter().filter(|d| !d.is_finished()).cloned().collect()
}
/// The running duty named `name`, if one runs.
pub fn find(name: &str) -> Option<Arc<Duty>> {
running().into_iter().find(|d| d.name == name)
}
/// **Stop every duty** (the box's stop), each waited for up to `grace`. Answers how many ran.
pub fn stop_all(grace: Duration) -> usize {
let all = running();
// Every one is told first, so one that cannot end soon (a loop resting in a park) costs the
// others nothing; then each is waited for, all within the one `grace`.
for d in &all {
*d.stop.lock().unwrap_or_else(|p| p.into_inner()) = true;
d.wake.notify_all();
}
let until = Instant::now() + grace;
for d in &all {
d.stop(until.saturating_duration_since(Instant::now()));
}
all.len()
}
/// **Start a duty** on a thread named `name`: a round, then `every`, then the next. A thread
/// that cannot start is an answer.
pub fn start(name: &str, every: Duration, say: Say, round: impl FnMut() -> Round + Send + 'static) -> std::io::Result<Arc<Duty>> {
start_with(name, every, say, move || round)
}
/// **[`start`], its round built ON the duty's thread** by `init` (a round that holds what is
/// not `Send`: the pairing loop's tunnel driver). An `init` that panics is said, and ends the duty.
pub fn start_with<R, I>(name: &str, every: Duration, say: Say, init: I) -> std::io::Result<Arc<Duty>>
where
R: FnMut() -> Round + 'static,
I: FnOnce() -> R + Send + 'static,
{
let duty = Arc::new(Duty {
name: name.to_string(),
stop: Mutex::new(false),
wake: Condvar::new(),
rounds: AtomicU64::new(0),
failures: AtomicU64::new(0),
last_failure: Mutex::new(None),
thread: Mutex::new(None),
});
let d = Arc::clone(&duty);
// The handle is the duty's: kept, joined at the stop.
let mut slot = duty.thread.lock().unwrap_or_else(|p| p.into_inner());
*slot = Some(std::thread::Builder::new().name(name.to_string()).spawn(move || {
let mut round = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(init)) {
Ok(r) => r,
Err(_) => {
d.failures.fetch_add(1, Ordering::SeqCst);
say(&format!("ALERT {} could not begin (its start panicked): the duty does not run", d.name));
return;
}
};
while !d.stopped() {
let next = match d.round(&mut round, &say) {
Round::Done => return,
Round::Again => every,
Round::After(d) => d,
};
if !d.pause(next) {
return;
}
}
})?);
drop(slot);
let mut all = registry().lock().unwrap_or_else(|p| p.into_inner());
all.retain(|d| !d.is_finished());
all.push(Arc::clone(&duty));
drop(all);
Ok(duty)
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
/// [`stop_all`] stops every duty: one test at a time.
pub(crate) static ONE: Mutex<()> = Mutex::new(());
fn heard() -> (Say, Arc<Mutex<Vec<String>>>) {
let lines = Arc::new(Mutex::new(Vec::new()));
let l = Arc::clone(&lines);
(Arc::new(move |s: &str| l.lock().unwrap().push(s.to_string())), lines)
}
fn until(what: impl Fn() -> bool) -> bool {
let t = Instant::now();
while t.elapsed() < Duration::from_secs(10) {
if what() {
return true;
}
std::thread::sleep(Duration::from_millis(5));
}
false
}
/// ★ **A round that panics is said, counted, and the duty goes on** (A28 M18): the raw
/// thread ended at the panic, silently.
#[test]
fn a_round_that_panics_is_said_counted_and_the_duty_goes_on() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, lines) = heard();
let mut n = 0;
let d = start("test-panics", Duration::from_millis(5), say, move || {
n += 1;
if n == 1 {
panic!("the first round breaks");
}
Round::Again
})
.unwrap();
assert!(until(|| d.rounds() >= 3), "the rounds after the panic run (rounds {})", d.rounds());
assert_eq!(d.failures(), 1);
assert!(d.last_failure().unwrap_or_default().contains("the first round breaks"));
assert!(lines.lock().unwrap().iter().any(|l| l.contains("ALERT test-panics") && l.contains("panicked")), "said as an ALERT");
assert!(d.stop(Duration::from_secs(5)));
}
/// ★ **A stopped duty ends at once**, its sleep cut short: the raw thread slept and looped
/// for the life of the process.
#[test]
fn a_stopped_duty_ends_at_once() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let d = start("test-stops", Duration::from_secs(3600), say, || Round::Again).unwrap();
assert!(until(|| d.rounds() >= 1));
assert!(find("test-stops").is_some(), "a running duty is listed");
let t = Instant::now();
assert!(stop_all(Duration::from_millis(300)) >= 1);
assert!(until(|| d.is_finished()) && t.elapsed() < Duration::from_secs(2), "the hour's sleep is cut short");
assert!(find("test-stops").is_none(), "a stopped duty is not listed");
}
/// ★ **A round says when the next is due** (a backoff), the period is only the default.
#[test]
fn a_round_says_when_the_next_one_is_due() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let mut n = 0;
let d = start("test-after", Duration::from_secs(3600), say, move || {
n += 1;
if n >= 2 { Round::Done } else { Round::After(Duration::from_millis(5)) }
})
.unwrap();
assert!(until(|| d.is_finished()), "the second round ran after 5 ms, not after the hour");
assert_eq!(d.rounds(), 2);
}
/// ★ **A round that answers Done ends the duty.**
#[test]
fn a_done_round_ends_the_duty() {
let _one = ONE.lock().unwrap_or_else(|p| p.into_inner());
let (say, _) = heard();
let mut n = 0;
let d = start("test-done", Duration::from_millis(5), say, move || {
n += 1;
if n >= 2 { Round::Done } else { Round::Again }
})
.unwrap();
assert!(until(|| d.is_finished()));
assert_eq!(d.rounds(), 2);
}
}