use crate::user::ProcessIdentity;use std::collections::{BTreeMap, BTreeSet, VecDeque};use std::error::Error;use std::fmt;use std::ops::{BitOr, BitOrAssign};use std::sync::{Arc, Condvar, Mutex, MutexGuard};use std::time::Duration;use web_time::{Instant, SystemTime, UNIX_EPOCH}; const ZOMBIE_TTL: Duration = Duration::from_secs(60);const INIT_PID: u32 = 1;const MAX_ALLOCATED_PID: u32 = i32::MAX as u32;pub const DEFAULT_PROCESS_UMASK: u32 = 0o022;pub const SIGHUP: i32 = 1;pub const SIGCHLD: i32 = 17;pub const SIGCONT: i32 = 18;pub const SIGSTOP: i32 = 19;pub const SIGTSTP: i32 = 20;pub const SIGTERM: i32 = 15;pub const SIGKILL: i32 = 9;pub const SIGPIPE: i32 = 13;pub const SIGWINCH: i32 = 28;const MAX_SIGNAL: i32 = 64; pub type ProcessResult<T> = Result<T, ProcessTableError>;pub type ProcessExitCallback = Arc<dyn Fn(i32) + Send + Sync + 'static>; pub trait DriverProcess: Send + Sync { fn kill(&self, signal: i32); fn wait(&self, timeout: Duration) -> Option<i32>; fn set_on_exit(&self, callback: ProcessExitCallback);} #[derive(Debug, Clone, PartialEq, Eq)]pub struct ProcessTableError { code: &'static str, message: String,} impl ProcessTableError { pub fn code(&self) -> &'static str { self.code } fn invalid_signal(signal: i32) -> Self { Self { code: "EINVAL", message: format!("invalid signal {signal}"), } } fn no_such_process(pid: u32) -> Self { Self { code: "ESRCH", message: format!("no such process {pid}"), } } fn no_such_process_group(pgid: u32) -> Self { Self { code: "ESRCH", message: format!("no such process group {pgid}"), } } fn no_matching_child(waiter_pid: u32, pid: i32) -> Self { Self { code: "ECHILD", message: format!("process {waiter_pid} has no matching child for waitpid({pid})"), } } fn pid_space_exhausted() -> Self { Self { code: "EAGAIN", message: String::from("process id space exhausted"), } } fn permission_denied(message: impl Into<String>) -> Self { Self { code: "EPERM", message: message.into(), } }} impl fmt::Display for ProcessTableError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}: {}", self.code, self.message) }} impl Error for ProcessTableError {} #[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum ProcessStatus { Running, Stopped, Exited,} #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]pub struct SignalSet { bits: u64,} impl SignalSet { pub const fn empty() -> Self { Self { bits: 0 } } pub const fn is_empty(self) -> bool { self.bits == 0 } pub fn from_signal(signal: i32) -> ProcessResult<Self> { Ok(Self { bits: signal_bit(signal)?, }) } pub fn from_signals(signals: impl IntoIterator<Item = i32>) -> ProcessResult<Self> { let mut set = Self::empty(); for signal in signals { set.insert(signal)?; } Ok(set) } pub fn contains(self, signal: i32) -> bool { signal_bit(signal) .map(|bit| self.bits & bit != 0) .unwrap_or(false) } pub fn insert(&mut self, signal: i32) -> ProcessResult<()> { self.bits |= signal_bit(signal)?; Ok(()) } pub fn remove(&mut self, signal: i32) -> ProcessResult<()> { self.bits &= !signal_bit(signal)?; Ok(()) } pub fn union(self, other: Self) -> Self { Self { bits: self.bits | other.bits, } } pub fn difference(self, other: Self) -> Self { Self { bits: self.bits & !other.bits, } } pub fn signals(self) -> Vec<i32> { let mut signals = Vec::new(); for signal in 1..=MAX_SIGNAL { if self.contains(signal) { signals.push(signal); } } signals }} #[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum SigmaskHow { Block, Unblock, SetMask,} #[derive(Debug, Clone, Copy, PartialEq, Eq)]pub struct WaitPidFlags { bits: u32,} impl WaitPidFlags { pub const WNOHANG: Self = Self { bits: 1 << 0 }; pub const WUNTRACED: Self = Self { bits: 1 << 1 }; pub const WCONTINUED: Self = Self { bits: 1 << 2 }; pub const fn empty() -> Self { Self { bits: 0 } } pub const fn contains(self, other: Self) -> bool { (self.bits & other.bits) == other.bits }} impl Default for WaitPidFlags { fn default() -> Self { Self::empty() }} impl BitOr for WaitPidFlags { type Output = Self; fn bitor(self, rhs: Self) -> Self::Output { Self { bits: self.bits | rhs.bits, } }} impl BitOrAssign for WaitPidFlags { fn bitor_assign(&mut self, rhs: Self) { self.bits |= rhs.bits; }} #[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum ProcessWaitEvent { Exited, Stopped, Continued,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct ProcessWaitResult { pub pid: u32, pub status: i32, pub event: ProcessWaitEvent,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct ProcessFileDescriptors { pub stdin: u32, pub stdout: u32, pub stderr: u32,} impl Default for ProcessFileDescriptors { fn default() -> Self { Self { stdin: 0, stdout: 1, stderr: 2, } }} #[derive(Debug, Clone, PartialEq, Eq)]pub struct ProcessContext { pub pid: u32, pub ppid: u32, pub env: BTreeMap<String, String>, pub cwd: String, pub umask: u32, pub fds: ProcessFileDescriptors, pub identity: ProcessIdentity, pub blocked_signals: SignalSet, pub pending_signals: SignalSet,} impl Default for ProcessContext { fn default() -> Self { Self { pid: 0, ppid: 0, env: BTreeMap::new(), cwd: String::from("/"), umask: DEFAULT_PROCESS_UMASK, fds: ProcessFileDescriptors::default(), identity: ProcessIdentity::default(), blocked_signals: SignalSet::empty(), pending_signals: SignalSet::empty(), } }} #[derive(Debug, Clone, PartialEq, Eq)]pub struct ProcessEntry { pub pid: u32, pub ppid: u32, pub pgid: u32, pub sid: u32, pub driver: String, pub command: String, pub args: Vec<String>, pub status: ProcessStatus, pub exit_code: Option<i32>, pub exit_time_ms: Option<u64>, pub env: BTreeMap<String, String>, pub cwd: String, pub umask: u32, pub identity: ProcessIdentity,} #[derive(Debug, Clone, PartialEq, Eq)]pub struct ProcessInfo { pub pid: u32, pub ppid: u32, pub pgid: u32, pub sid: u32, pub driver: String, pub command: String, pub status: ProcessStatus, pub exit_code: Option<i32>, pub identity: ProcessIdentity,} #[derive(Clone)]pub struct ProcessTable { inner: Arc<ProcessTableInner>,} struct ProcessTableInner { state: Mutex<ProcessTableState>, waiters: Condvar, reaper: Arc<ZombieReaper>,} struct ProcessRecord { entry: ProcessEntry, driver_process: Arc<dyn DriverProcess>, pending_wait_events: VecDeque<PendingWaitEvent>, blocked_signals: SignalSet, pending_signals: SignalSet,} struct ScheduledSignalDelivery { pid: u32, signal: i32, status: ProcessStatus, driver_process: Arc<dyn DriverProcess>,} #[derive(Debug, Clone, Copy, PartialEq, Eq)]struct PendingWaitEvent { status: i32, event: ProcessWaitEvent,} #[derive(Debug, Clone, Copy, PartialEq, Eq)]enum WaitSelector { AnyChild, ChildPid(u32), ProcessGroup(u32),} struct ZombieReaper { state: Mutex<ZombieReaperState>,} #[derive(Default)]struct ZombieReaperState { deadlines: BTreeMap<u32, Instant>,} struct ProcessTableState { entries: BTreeMap<u32, ProcessRecord>, next_pid: u32, zombie_ttl: Duration, on_process_exit: Option<Arc<dyn Fn(u32) + Send + Sync + 'static>>, terminating_all: bool,} impl Default for ProcessTableState { fn default() -> Self { Self { entries: BTreeMap::new(), next_pid: 1, zombie_ttl: ZOMBIE_TTL, on_process_exit: None, terminating_all: false, } }} impl Default for ProcessTable { fn default() -> Self { let reaper = Arc::new(ZombieReaper::default()); Self { inner: Arc::new(ProcessTableInner { state: Mutex::new(ProcessTableState::default()), waiters: Condvar::new(), reaper, }), } }} impl ProcessTable { pub fn new() -> Self { Self::default() } pub fn with_zombie_ttl(zombie_ttl: Duration) -> Self { let table = Self::new(); table.inner.lock_state().zombie_ttl = zombie_ttl; table } pub fn allocate_pid(&self) -> ProcessResult<u32> { let mut state = self.inner.lock_state(); let start = normalize_next_pid(state.next_pid); let mut pid = start; loop { if !state.entries.contains_key(&pid) { state.next_pid = next_allocated_pid_after(pid); return Ok(pid); } pid = next_allocated_pid_after(pid); if pid == start { return Err(ProcessTableError::pid_space_exhausted()); } } } pub fn set_on_process_exit(&self, callback: Option<Arc<dyn Fn(u32) + Send + Sync + 'static>>) { self.inner.lock_state().on_process_exit = callback; } pub fn register( &self, pid: u32, driver: impl Into<String>, command: impl Into<String>, args: Vec<String>, ctx: ProcessContext, driver_process: Arc<dyn DriverProcess>, ) -> ProcessEntry { self.register_with_process_group(pid, driver, command, args, ctx, driver_process, None) .expect("inheriting a process group cannot fail") } // Registration keeps the process image, context, driver, and requested // group explicit so ownership validation happens at one boundary. #[allow(clippy::too_many_arguments)] pub fn register_with_process_group( &self, pid: u32, driver: impl Into<String>, command: impl Into<String>, args: Vec<String>, ctx: ProcessContext, driver_process: Arc<dyn DriverProcess>, requested_pgid: Option<u32>, ) -> ProcessResult<ProcessEntry> { let driver = driver.into(); let command = command.into(); let mut state = self.inner.lock_state(); let (inherited_pgid, sid) = match state.entries.get(&ctx.ppid) { Some(parent) => (parent.entry.pgid, parent.entry.sid), None => (pid, pid), }; let pgid = requested_pgid.map_or(inherited_pgid, |pgid| if pgid == 0 { pid } else { pgid }); if requested_pgid.is_some() && pgid != pid { let mut group_exists = false; for record in state.entries.values() { if record.entry.pgid != pgid || record.entry.status == ProcessStatus::Exited { continue; } if record.entry.sid != sid { return Err(ProcessTableError::permission_denied( "cannot join process group in different session", )); } group_exists = true; break; } if !group_exists { return Err(ProcessTableError::permission_denied(format!( "no such process group {pgid}" ))); } } let entry = ProcessEntry { pid, ppid: ctx.ppid, pgid, sid, driver, command, args, status: ProcessStatus::Running, exit_code: None, exit_time_ms: None, env: ctx.env, cwd: ctx.cwd, umask: ctx.umask & 0o777, identity: ctx.identity, }; state.next_pid = next_pid_after_registered(state.next_pid, pid); state.entries.insert( pid, ProcessRecord { entry: entry.clone(), driver_process: driver_process.clone(),Lines 1–500 of 1724. Load more
use crate::user::ProcessIdentity; use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::error::Error; use std::fmt; use std::ops::{BitOr, BitOrAssign}; use std::sync::{Arc, Condvar, Mutex, MutexGuard}; use std::time::Duration; use web_time::{Instant, SystemTime, UNIX_EPOCH}; const ZOMBIE_TTL: Duration = Duration::from_secs(60); const INIT_PID: u32 = 1; const MAX_ALLOCATED_PID: u32 = i32::MAX as u32; pub const DEFAULT_PROCESS_UMASK: u32 = 0o022; pub const SIGHUP: i32 = 1; pub const SIGCHLD: i32 = 17; pub const SIGCONT: i32 = 18; pub const SIGSTOP: i32 = 19; pub const SIGTSTP: i32 = 20; pub const SIGTERM: i32 = 15; pub const SIGKILL: i32 = 9; pub const SIGPIPE: i32 = 13; pub const SIGWINCH: i32 = 28; const MAX_SIGNAL: i32 = 64; pub type ProcessResult<T> = Result<T, ProcessTableError>; pub type ProcessExitCallback = Arc<dyn Fn(i32) + Send + Sync + 'static>; pub trait DriverProcess: Send + Sync { fn kill(&self, signal: i32); fn wait(&self, timeout: Duration) -> Option<i32>; fn set_on_exit(&self, callback: ProcessExitCallback); } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProcessTableError { code: &'static str, message: String, } impl ProcessTableError { pub fn code(&self) -> &'static str { self.code } fn invalid_signal(signal: i32) -> Self { Self { code: "EINVAL", message: format!("invalid signal {signal}"), } } fn no_such_process(pid: u32) -> Self { Self { code: "ESRCH", message: format!("no such process {pid}"), } } fn no_such_process_group(pgid: u32) -> Self { Self { code: "ESRCH", message: format!("no such process group {pgid}"), } } fn no_matching_child(waiter_pid: u32, pid: i32) -> Self { Self { code: "ECHILD", message: format!("process {waiter_pid} has no matching child for waitpid({pid})"), } } fn pid_space_exhausted() -> Self { Self { code: "EAGAIN", message: String::from("process id space exhausted"), } } fn permission_denied(message: impl Into<String>) -> Self { Self { code: "EPERM", message: message.into(), } } } impl fmt::Display for ProcessTableError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}: {}", self.code, self.message) } } impl Error for ProcessTableError {} #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ProcessStatus { Running, Stopped, Exited, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct SignalSet { bits: u64, } impl SignalSet { pub const fn empty() -> Self { Self { bits: 0 } } pub const fn is_empty(self) -> bool { self.bits == 0 } pub fn from_signal(signal: i32) -> ProcessResult<Self> { Ok(Self { bits: signal_bit(signal)?, }) } pub fn from_signals(signals: impl IntoIterator<Item = i32>) -> ProcessResult<Self> { let mut set = Self::empty(); for signal in signals { set.insert(signal)?; } Ok(set) } pub fn contains(self, signal: i32) -> bool { signal_bit(signal) .map(|bit| self.bits & bit != 0) .unwrap_or(false) } pub fn insert(&mut self, signal: i32) -> ProcessResult<()> { self.bits |= signal_bit(signal)?; Ok(()) } pub fn remove(&mut self, signal: i32) -> ProcessResult<()> { self.bits &= !signal_bit(signal)?; Ok(()) } pub fn union(self, other: Self) -> Self { Self { bits: self.bits | other.bits, } } pub fn difference(self, other: Self) -> Self { Self { bits: self.bits & !other.bits, } } pub fn signals(self) -> Vec<i32> { let mut signals = Vec::new(); for signal in 1..=MAX_SIGNAL { if self.contains(signal) { signals.push(signal); } } signals } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SigmaskHow { Block, Unblock, SetMask, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct WaitPidFlags { bits: u32, } impl WaitPidFlags { pub const WNOHANG: Self = Self { bits: 1 << 0 }; pub const WUNTRACED: Self = Self { bits: 1 << 1 }; pub const WCONTINUED: Self = Self { bits: 1 << 2 }; pub const fn empty() -> Self { Self { bits: 0 } } pub const fn contains(self, other: Self) -> bool { (self.bits & other.bits) == other.bits } } impl Default for WaitPidFlags { fn default() -> Self { Self::empty() } } impl BitOr for WaitPidFlags { type Output = Self; fn bitor(self, rhs: Self) -> Self::Output { Self { bits: self.bits | rhs.bits, } } } impl BitOrAssign for WaitPidFlags { fn bitor_assign(&mut self, rhs: Self) { self.bits |= rhs.bits; } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ProcessWaitEvent { Exited, Stopped, Continued, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProcessWaitResult { pub pid: u32, pub status: i32, pub event: ProcessWaitEvent, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProcessFileDescriptors { pub stdin: u32, pub stdout: u32, pub stderr: u32, } impl Default for ProcessFileDescriptors { fn default() -> Self { Self { stdin: 0, stdout: 1, stderr: 2, } } } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProcessContext { pub pid: u32, pub ppid: u32, pub env: BTreeMap<String, String>, pub cwd: String, pub umask: u32, pub fds: ProcessFileDescriptors, pub identity: ProcessIdentity, pub blocked_signals: SignalSet, pub pending_signals: SignalSet, } impl Default for ProcessContext { fn default() -> Self { Self { pid: 0, ppid: 0, env: BTreeMap::new(), cwd: String::from("/"), umask: DEFAULT_PROCESS_UMASK, fds: ProcessFileDescriptors::default(), identity: ProcessIdentity::default(), blocked_signals: SignalSet::empty(), pending_signals: SignalSet::empty(), } } } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProcessEntry { pub pid: u32, pub ppid: u32, pub pgid: u32, pub sid: u32, pub driver: String, pub command: String, pub args: Vec<String>, pub status: ProcessStatus, pub exit_code: Option<i32>, pub exit_time_ms: Option<u64>, pub env: BTreeMap<String, String>, pub cwd: String, pub umask: u32, pub identity: ProcessIdentity, } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProcessInfo { pub pid: u32, pub ppid: u32, pub pgid: u32, pub sid: u32, pub driver: String, pub command: String, pub status: ProcessStatus, pub exit_code: Option<i32>, pub identity: ProcessIdentity, } #[derive(Clone)] pub struct ProcessTable { inner: Arc<ProcessTableInner>, } struct ProcessTableInner { state: Mutex<ProcessTableState>, waiters: Condvar, reaper: Arc<ZombieReaper>, } struct ProcessRecord { entry: ProcessEntry, driver_process: Arc<dyn DriverProcess>, pending_wait_events: VecDeque<PendingWaitEvent>, blocked_signals: SignalSet, pending_signals: SignalSet, } struct ScheduledSignalDelivery { pid: u32, signal: i32, status: ProcessStatus, driver_process: Arc<dyn DriverProcess>, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] struct PendingWaitEvent { status: i32, event: ProcessWaitEvent, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum WaitSelector { AnyChild, ChildPid(u32), ProcessGroup(u32), } struct ZombieReaper { state: Mutex<ZombieReaperState>, } #[derive(Default)] struct ZombieReaperState { deadlines: BTreeMap<u32, Instant>, } struct ProcessTableState { entries: BTreeMap<u32, ProcessRecord>, next_pid: u32, zombie_ttl: Duration, on_process_exit: Option<Arc<dyn Fn(u32) + Send + Sync + 'static>>, terminating_all: bool, } impl Default for ProcessTableState { fn default() -> Self { Self { entries: BTreeMap::new(), next_pid: 1, zombie_ttl: ZOMBIE_TTL, on_process_exit: None, terminating_all: false, } } } impl Default for ProcessTable { fn default() -> Self { let reaper = Arc::new(ZombieReaper::default()); Self { inner: Arc::new(ProcessTableInner { state: Mutex::new(ProcessTableState::default()), waiters: Condvar::new(), reaper, }), } } } impl ProcessTable { pub fn new() -> Self { Self::default() } pub fn with_zombie_ttl(zombie_ttl: Duration) -> Self { let table = Self::new(); table.inner.lock_state().zombie_ttl = zombie_ttl; table } pub fn allocate_pid(&self) -> ProcessResult<u32> { let mut state = self.inner.lock_state(); let start = normalize_next_pid(state.next_pid); let mut pid = start; loop { if !state.entries.contains_key(&pid) { state.next_pid = next_allocated_pid_after(pid); return Ok(pid); } pid = next_allocated_pid_after(pid); if pid == start { return Err(ProcessTableError::pid_space_exhausted()); } } } pub fn set_on_process_exit(&self, callback: Option<Arc<dyn Fn(u32) + Send + Sync + 'static>>) { self.inner.lock_state().on_process_exit = callback; } pub fn register( &self, pid: u32, driver: impl Into<String>, command: impl Into<String>, args: Vec<String>, ctx: ProcessContext, driver_process: Arc<dyn DriverProcess>, ) -> ProcessEntry { self.register_with_process_group(pid, driver, command, args, ctx, driver_process, None) .expect("inheriting a process group cannot fail") } // Registration keeps the process image, context, driver, and requested // group explicit so ownership validation happens at one boundary. #[allow(clippy::too_many_arguments)] pub fn register_with_process_group( &self, pid: u32, driver: impl Into<String>, command: impl Into<String>, args: Vec<String>, ctx: ProcessContext, driver_process: Arc<dyn DriverProcess>, requested_pgid: Option<u32>, ) -> ProcessResult<ProcessEntry> { let driver = driver.into(); let command = command.into(); let mut state = self.inner.lock_state(); let (inherited_pgid, sid) = match state.entries.get(&ctx.ppid) { Some(parent) => (parent.entry.pgid, parent.entry.sid), None => (pid, pid), }; let pgid = requested_pgid.map_or(inherited_pgid, |pgid| if pgid == 0 { pid } else { pgid }); if requested_pgid.is_some() && pgid != pid { let mut group_exists = false; for record in state.entries.values() { if record.entry.pgid != pgid || record.entry.status == ProcessStatus::Exited { continue; } if record.entry.sid != sid { return Err(ProcessTableError::permission_denied( "cannot join process group in different session", )); } group_exists = true; break; } if !group_exists { return Err(ProcessTableError::permission_denied(format!( "no such process group {pgid}" ))); } } let entry = ProcessEntry { pid, ppid: ctx.ppid, pgid, sid, driver, command, args, status: ProcessStatus::Running, exit_code: None, exit_time_ms: None, env: ctx.env, cwd: ctx.cwd, umask: ctx.umask & 0o777, identity: ctx.identity, }; state.next_pid = next_pid_after_registered(state.next_pid, pid); state.entries.insert( pid, ProcessRecord { entry: entry.clone(), driver_process: driver_process.clone(),