179 lines
6.7 KiB
Rust
179 lines
6.7 KiB
Rust
//! Process metadata, process lifetime, and attached children.
|
|
use crate::{Host, RpcResult};
|
|
use serde_json::{Value, json};
|
|
use tokio::sync::{mpsc, oneshot, watch};
|
|
#[derive(Clone, serde::Serialize)]
|
|
/// A snapshot of process metadata for inspection and the VFS.
|
|
pub struct ProcessInfo {
|
|
/// The plugin that supplied the process code.
|
|
pub plugin: String,
|
|
/// The SHA-256 of its component, or native for a native plugin.
|
|
pub version: String,
|
|
/// The method without its plugin prefix.
|
|
pub method: String,
|
|
/// The object supplied by the caller.
|
|
pub params: Value,
|
|
/// The attached parent, or no parent for a detached process.
|
|
pub parent: Option<String>,
|
|
/// Whether this process survives the end of its caller.
|
|
pub detached: bool,
|
|
/// The start time in milliseconds since the Unix epoch.
|
|
pub start_time: u128,
|
|
/// The current lifecycle state.
|
|
pub state: String,
|
|
}
|
|
pub(crate) struct Process {
|
|
pub(crate) info: ProcessInfo,
|
|
pub(crate) memory: sausage_vfs::Memory,
|
|
pub(crate) files: sausage_vfs::Process,
|
|
pub(crate) store_ended: watch::Sender<bool>,
|
|
pub(crate) input: Option<mpsc::UnboundedSender<Value>>,
|
|
pub(crate) abort: Option<tokio::task::AbortHandle>,
|
|
pub(crate) caller: String,
|
|
pub(crate) id: Option<Value>,
|
|
pub(crate) answer: Option<oneshot::Sender<RpcResult>>,
|
|
pub(crate) ended: watch::Sender<bool>,
|
|
pub(crate) stopped: bool,
|
|
pub(crate) superseded: bool,
|
|
}
|
|
impl Host {
|
|
/// Called only after the execution task ends. Wait for the store and file operations before releasing locks.
|
|
pub(crate) async fn finish(&self, pid: &str, status: &str) {
|
|
let (files, mut store_ended) = {
|
|
let mut state = self.0.state.lock().unwrap();
|
|
let Some(p) = state.processes.get_mut(pid) else {
|
|
return;
|
|
};
|
|
p.info.state = "stopping".into();
|
|
(p.files.clone(), p.store_ended.subscribe())
|
|
};
|
|
let _ = store_ended.wait_for(|ended| *ended).await;
|
|
files.idle().await;
|
|
let mut state = self.0.state.lock().unwrap();
|
|
let process = state.processes.remove(pid);
|
|
if let Some(p) = process {
|
|
self.0.journal.write(pid, &p.info.plugin, "core", status);
|
|
let logs = p
|
|
.answer
|
|
.as_ref()
|
|
.map(|_| self.0.journal.query_later(json!({"pid":pid,"limit":20})));
|
|
self.0.journal.forget(pid);
|
|
Self::release_locks(&mut state, pid);
|
|
self.service_ended(&mut state, pid, status, p.stopped, p.superseded);
|
|
let status = status.to_string();
|
|
tokio::spawn(async move {
|
|
if let Some(answer) = p.answer {
|
|
let lines = logs.unwrap().await.unwrap_or_else(|_| json!([]));
|
|
let failure = json!({
|
|
"code": -32002,
|
|
"message": "process ended without an answer",
|
|
"data": { "status": status, "lines": lines }
|
|
});
|
|
let _ = answer.send(Err(failure));
|
|
}
|
|
p.ended.send_replace(true);
|
|
});
|
|
let children: Vec<_> = state
|
|
.processes
|
|
.iter()
|
|
.filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
|
|
.map(|(id, _)| id.clone())
|
|
.collect();
|
|
drop(state);
|
|
for child in children {
|
|
self.begin_stop(&child);
|
|
}
|
|
}
|
|
}
|
|
/// Close stdin and start the same five-second deadline for each attached child.
|
|
pub(crate) fn begin_stop(&self, pid: &str) -> Option<watch::Receiver<bool>> {
|
|
self.stop_process(pid, false)
|
|
}
|
|
/// Grace expiry is retirement, not an operator stopping the service.
|
|
pub(crate) fn begin_retire(&self, pid: &str) -> Option<watch::Receiver<bool>> {
|
|
self.stop_process(pid, true)
|
|
}
|
|
fn stop_process(&self, pid: &str, retirement: bool) -> Option<watch::Receiver<bool>> {
|
|
let mut state = self.0.state.lock().unwrap();
|
|
let p = state.processes.get_mut(pid)?;
|
|
if !retirement {
|
|
p.superseded = false;
|
|
}
|
|
let ended = p.ended.subscribe();
|
|
if p.info.state == "stopping" {
|
|
return Some(ended);
|
|
}
|
|
p.info.state = "stopping".into();
|
|
p.stopped = true;
|
|
p.input.take();
|
|
self.0.journal.write(pid, &p.info.plugin, "core", "stop");
|
|
for lock in state.locks.values_mut() {
|
|
lock.queue.retain(|(queued, _)| queued != pid);
|
|
}
|
|
let children: Vec<_> = state
|
|
.processes
|
|
.iter()
|
|
.filter(|(_, p)| p.info.parent.as_deref() == Some(pid))
|
|
.map(|(id, _)| id.clone())
|
|
.collect();
|
|
drop(state);
|
|
for child in children {
|
|
self.begin_stop(&child);
|
|
}
|
|
let h = self.clone();
|
|
let pid = pid.to_string();
|
|
tokio::spawn(async move {
|
|
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
|
|
h.kill_process(&pid, retirement);
|
|
});
|
|
Some(ended)
|
|
}
|
|
/// Abort a process task and perform its ended-state cleanup.
|
|
pub(crate) fn kill(&self, pid: &str) {
|
|
self.kill_process(pid, false);
|
|
}
|
|
fn kill_process(&self, pid: &str, retirement: bool) {
|
|
let mut state = self.0.state.lock().unwrap();
|
|
for lock in state.locks.values_mut() {
|
|
lock.queue.retain(|(queued, _)| queued != pid);
|
|
}
|
|
if let Some(p) = state.processes.get_mut(pid) {
|
|
if !retirement {
|
|
p.superseded = false;
|
|
}
|
|
p.stopped = true;
|
|
p.info.state = "stopping".into();
|
|
p.input.take();
|
|
self.0.journal.write(pid, &p.info.plugin, "core", "kill");
|
|
if let Some(a) = &p.abort {
|
|
a.abort();
|
|
}
|
|
}
|
|
}
|
|
/// Stop every process in parallel and wait for journal writes to finish.
|
|
pub async fn stop(&self) {
|
|
{
|
|
let _state = self.0.state.lock().unwrap();
|
|
self.0
|
|
.shutdown
|
|
.store(true, std::sync::atomic::Ordering::SeqCst);
|
|
}
|
|
loop {
|
|
let ids: Vec<_> = self.processes().into_keys().collect();
|
|
if ids.is_empty() {
|
|
break;
|
|
}
|
|
let mut waits = Vec::new();
|
|
for pid in ids {
|
|
if let Some(n) = self.begin_stop(&pid) {
|
|
waits.push(n);
|
|
}
|
|
}
|
|
for mut n in waits {
|
|
let _ = n.wait_for(|ended| *ended).await;
|
|
}
|
|
}
|
|
self.0.journal.query(json!({"limit":0})).await;
|
|
}
|
|
}
|