use agentos_kernel::process_table::{ DriverProcess, ProcessContext, ProcessExitCallback, ProcessResult, ProcessStatus, ProcessTable, ProcessWaitEvent, SigmaskHow, SignalSet, WaitPidFlags, SIGCHLD, SIGCONT, SIGHUP, SIGSTOP, SIGTERM, SIGTSTP,};use std::collections::BTreeMap;use std::fmt::Debug;use std::sync::{mpsc, Arc, Condvar, Mutex};use std::thread;use std::time::{Duration, Instant}; fn assert_error_code<T: Debug>(result: ProcessResult<T>, expected: &str) { let error = result.expect_err("operation should fail"); assert_eq!(error.code(), expected);} #[derive(Default)]struct MockProcessState { kills: Vec<i32>, exit_code: Option<i32>, on_exit: Option<ProcessExitCallback>, ignore_sigterm: bool,} #[derive(Default)]struct MockDriverProcess { state: Mutex<MockProcessState>, exited: Condvar,} impl MockDriverProcess { fn new() -> Arc<Self> { Arc::new(Self::default()) } fn stubborn() -> Arc<Self> { Arc::new(Self { state: Mutex::new(MockProcessState { ignore_sigterm: true, ..MockProcessState::default() }), exited: Condvar::new(), }) } fn schedule_exit(self: &Arc<Self>, delay: Duration, exit_code: i32) { let process = Arc::clone(self); thread::spawn(move || { thread::sleep(delay); process.exit(exit_code); }); } fn exit(&self, exit_code: i32) { let callback = { let mut state = self.state.lock().expect("mock process lock poisoned"); if state.exit_code.is_some() { return; } state.exit_code = Some(exit_code); self.exited.notify_all(); state.on_exit.clone() }; if let Some(callback) = callback { callback(exit_code); } } fn kills(&self) -> Vec<i32> { self.state .lock() .expect("mock process lock poisoned") .kills .clone() }} impl DriverProcess for MockDriverProcess { fn kill(&self, signal: i32) { let should_exit = { let mut state = self.state.lock().expect("mock process lock poisoned"); state.kills.push(signal); signal == 9 || (signal == 15 && !state.ignore_sigterm) }; if should_exit { self.exit(128 + signal); } } fn wait(&self, timeout: Duration) -> Option<i32> { let state = self.state.lock().expect("mock process lock poisoned"); if state.exit_code.is_some() { return state.exit_code; } let (state, _) = self .exited .wait_timeout(state, timeout) .expect("mock process wait lock poisoned"); state.exit_code } fn set_on_exit(&self, callback: ProcessExitCallback) { self.state .lock() .expect("mock process lock poisoned") .on_exit = Some(callback); }} fn create_context(ppid: u32) -> ProcessContext { ProcessContext { pid: 0, ppid, env: BTreeMap::new(), cwd: String::from("/"), ..ProcessContext::default() }} fn allocate_pid(table: &ProcessTable) -> u32 { table.allocate_pid().expect("allocate pid")} fn wait_for(predicate: impl Fn() -> bool, timeout: Duration) { let deadline = Instant::now() + timeout; while Instant::now() < deadline { if predicate() { return; } thread::sleep(Duration::from_millis(10)); } assert!(predicate(), "condition should become true before timeout");} #[test]fn register_allocates_expected_process_metadata_and_parent_groups() { let table = ProcessTable::new(); let parent = MockDriverProcess::new(); let child = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let child_pid = allocate_pid(&table); let parent_entry = table.register( parent_pid, "wasmvm", "grep", vec![String::from("-r"), String::from("foo")], create_context(0), parent, ); let child_entry = table.register( child_pid, "node", "node", vec![String::from("-e"), String::from("1+1")], create_context(parent_pid), child, ); assert_eq!(parent_entry.pid, 1); assert_eq!(child_entry.pid, 2); assert_eq!(parent_entry.pgid, 1); assert_eq!(parent_entry.sid, 1); assert_eq!(child_entry.pgid, 1); assert_eq!(child_entry.sid, 1); assert_eq!(child_entry.driver, "node");} #[test]fn waitpid_resolves_for_exiting_and_already_exited_processes() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let process = MockDriverProcess::new(); let pid = allocate_pid(&table); table.register( pid, "wasmvm", "echo", vec![String::from("hello")], create_context(0), process.clone(), ); process.schedule_exit(Duration::from_millis(10), 0); assert_eq!( table.waitpid(pid).expect("waitpid should resolve"), (pid, 0) ); assert_eq!(table.zombie_timer_count(), 0); assert!( table.get(pid).is_none(), "waitpid should reap exited processes" ); let exited_pid = allocate_pid(&table); table.register( exited_pid, "wasmvm", "true", Vec::new(), create_context(0), MockDriverProcess::new(), ); table.mark_exited(exited_pid, 42); assert_eq!( table .waitpid(exited_pid) .expect("waitpid should resolve immediately"), (exited_pid, 42) ); assert_eq!(table.zombie_timer_count(), 0); assert!( table.get(exited_pid).is_none(), "waitpid should reap already exited processes" );} #[test]fn long_lived_parent_retains_zombies_until_waited_under_pressure() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let parent = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let mut child_pids = Vec::new(); table.register( parent_pid, "wasmvm", "parent", Vec::new(), create_context(0), parent, ); for index in 0..100 { let child = MockDriverProcess::new(); let child_pid = allocate_pid(&table); table.register( child_pid, "wasmvm", format!("child-{index}"), Vec::new(), create_context(parent_pid), child.clone(), ); child.exit(index); child_pids.push((child_pid, index)); } for (child_pid, _) in &child_pids { assert_eq!( table .get(*child_pid) .expect("child zombie should be retained") .status, ProcessStatus::Exited ); } assert_eq!(table.zombie_timer_count(), child_pids.len()); for (child_pid, status) in child_pids { assert_eq!( table .waitpid_for(parent_pid, -1, WaitPidFlags::empty()) .expect("parent wait should succeed"), Some(agentos_kernel::process_table::ProcessWaitResult { pid: child_pid, status, event: ProcessWaitEvent::Exited, }) ); } assert_eq!(table.zombie_timer_count(), 0);} #[test]fn allocate_pid_wraps_without_reusing_live_or_zombie_entries() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let max_pid = i32::MAX as u32; let cursor_seed = MockDriverProcess::new(); let live_high = MockDriverProcess::new(); let zombie_high = MockDriverProcess::new(); let live_one = MockDriverProcess::new(); // Registering max_pid - 2 after the high PIDs moves the public allocation cursor back to max_pid - 1. table.register( max_pid - 1, "wasmvm", "live-high", Vec::new(), create_context(0), live_high, ); table.register( max_pid, "wasmvm", "zombie-high", Vec::new(), create_context(0), zombie_high.clone(), ); table.register( max_pid - 2, "wasmvm", "cursor-seed", Vec::new(), create_context(0), cursor_seed, ); table.register( 1, "wasmvm", "live-one", Vec::new(), create_context(0), live_one, ); zombie_high.exit(0); assert_eq!( table .get(max_pid) .expect("zombie high PID should remain allocated") .status, ProcessStatus::Exited ); assert_eq!(table.allocate_pid().expect("allocate wrapped pid"), 2); assert_eq!(table.allocate_pid().expect("allocate next pid"), 3);} #[test]fn waitpid_for_supports_wnohang_and_waiting_for_any_child() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let parent = MockDriverProcess::new(); let child_a = MockDriverProcess::new(); let child_b = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let child_a_pid = allocate_pid(&table); let child_b_pid = allocate_pid(&table); table.register( parent_pid, "wasmvm", "parent", Vec::new(), create_context(0), parent, ); table.register( child_a_pid, "wasmvm", "child-a", Vec::new(), create_context(parent_pid), child_a, ); table.register( child_b_pid, "wasmvm", "child-b", Vec::new(), create_context(parent_pid), child_b.clone(), ); assert_eq!( table .waitpid_for(parent_pid, -1, WaitPidFlags::WNOHANG) .expect("wnohang wait should succeed"), None ); child_b.exit(27); assert_eq!( table .waitpid_for(parent_pid, -1, WaitPidFlags::empty()) .expect("wait for any child should succeed"), Some(agentos_kernel::process_table::ProcessWaitResult { pid: child_b_pid, status: 27, event: ProcessWaitEvent::Exited, }) ); assert!( table.get(child_b_pid).is_none(), "waited child should be reaped" ); assert!( table.get(child_a_pid).is_some(), "other matching children should remain" );} #[test]fn on_process_exit_runs_before_waitpid_waiters_are_notified() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let process = MockDriverProcess::new(); let pid = allocate_pid(&table); table.register( pid, "wasmvm", "sleep", vec![String::from("1")], create_context(0), process.clone(), ); let (callback_entered_tx, callback_entered_rx) = mpsc::channel(); let callback_gate = Arc::new((Mutex::new(false), Condvar::new())); let callback_gate_for_exit = Arc::clone(&callback_gate); table.set_on_process_exit(Some(Arc::new(move |_| { callback_entered_tx .send(()) .expect("callback should report entry"); let (released, wake) = &*callback_gate_for_exit; let mut released = released.lock().expect("callback gate lock poisoned"); while !*released { released = wake .wait(released) .expect("callback gate wait lock poisoned"); } }))); let waiter_table = table.clone(); let (wait_result_tx, wait_result_rx) = mpsc::channel(); let waiter = thread::spawn(move || { let result = waiter_table.waitpid(pid).expect("waitpid should resolve"); wait_result_tx .send(result) .expect("waiter should report exit result"); }); thread::sleep(Duration::from_millis(10)); let process_for_exit = process.clone(); let exit_handle = thread::spawn(move || { process_for_exit.exit(0); }); callback_entered_rx .recv_timeout(Duration::from_millis(100)) .expect("exit callback should run"); assert!(wait_result_rx.try_recv().is_err()); let (released, wake) = &*callback_gate; *released.lock().expect("callback gate lock poisoned") = true; wake.notify_all(); assert_eq!( wait_result_rx .recv_timeout(Duration::from_millis(100)) .expect("waitpid should resolve after callback"), (pid, 0) ); exit_handle.join().expect("exit thread should finish"); waiter.join().expect("waiter thread should finish");} #[test]fn waitpid_for_reports_stopped_and_continued_children_once() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let parent = MockDriverProcess::new(); let child = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let child_pid = allocate_pid(&table); table.register( parent_pid, "wasmvm", "parent", Vec::new(), create_context(0), parent.clone(), ); table.register( child_pid, "wasmvm", "child", Vec::new(), create_context(parent_pid), child, ); table.mark_stopped(child_pid, SIGSTOP); assert_eq!( table .waitpid_for(parent_pid, child_pid as i32, WaitPidFlags::WNOHANG) .expect("stopped child lookup should succeed"), None ); assert_eq!( table .waitpid_for( parent_pid, child_pid as i32, WaitPidFlags::WNOHANG | WaitPidFlags::WUNTRACED,Lines 1–500 of 1453. Load more
use agentos_kernel::process_table::{ DriverProcess, ProcessContext, ProcessExitCallback, ProcessResult, ProcessStatus, ProcessTable, ProcessWaitEvent, SigmaskHow, SignalSet, WaitPidFlags, SIGCHLD, SIGCONT, SIGHUP, SIGSTOP, SIGTERM, SIGTSTP, }; use std::collections::BTreeMap; use std::fmt::Debug; use std::sync::{mpsc, Arc, Condvar, Mutex}; use std::thread; use std::time::{Duration, Instant}; fn assert_error_code<T: Debug>(result: ProcessResult<T>, expected: &str) { let error = result.expect_err("operation should fail"); assert_eq!(error.code(), expected); } #[derive(Default)] struct MockProcessState { kills: Vec<i32>, exit_code: Option<i32>, on_exit: Option<ProcessExitCallback>, ignore_sigterm: bool, } #[derive(Default)] struct MockDriverProcess { state: Mutex<MockProcessState>, exited: Condvar, } impl MockDriverProcess { fn new() -> Arc<Self> { Arc::new(Self::default()) } fn stubborn() -> Arc<Self> { Arc::new(Self { state: Mutex::new(MockProcessState { ignore_sigterm: true, ..MockProcessState::default() }), exited: Condvar::new(), }) } fn schedule_exit(self: &Arc<Self>, delay: Duration, exit_code: i32) { let process = Arc::clone(self); thread::spawn(move || { thread::sleep(delay); process.exit(exit_code); }); } fn exit(&self, exit_code: i32) { let callback = { let mut state = self.state.lock().expect("mock process lock poisoned"); if state.exit_code.is_some() { return; } state.exit_code = Some(exit_code); self.exited.notify_all(); state.on_exit.clone() }; if let Some(callback) = callback { callback(exit_code); } } fn kills(&self) -> Vec<i32> { self.state .lock() .expect("mock process lock poisoned") .kills .clone() } } impl DriverProcess for MockDriverProcess { fn kill(&self, signal: i32) { let should_exit = { let mut state = self.state.lock().expect("mock process lock poisoned"); state.kills.push(signal); signal == 9 || (signal == 15 && !state.ignore_sigterm) }; if should_exit { self.exit(128 + signal); } } fn wait(&self, timeout: Duration) -> Option<i32> { let state = self.state.lock().expect("mock process lock poisoned"); if state.exit_code.is_some() { return state.exit_code; } let (state, _) = self .exited .wait_timeout(state, timeout) .expect("mock process wait lock poisoned"); state.exit_code } fn set_on_exit(&self, callback: ProcessExitCallback) { self.state .lock() .expect("mock process lock poisoned") .on_exit = Some(callback); } } fn create_context(ppid: u32) -> ProcessContext { ProcessContext { pid: 0, ppid, env: BTreeMap::new(), cwd: String::from("/"), ..ProcessContext::default() } } fn allocate_pid(table: &ProcessTable) -> u32 { table.allocate_pid().expect("allocate pid") } fn wait_for(predicate: impl Fn() -> bool, timeout: Duration) { let deadline = Instant::now() + timeout; while Instant::now() < deadline { if predicate() { return; } thread::sleep(Duration::from_millis(10)); } assert!(predicate(), "condition should become true before timeout"); } #[test] fn register_allocates_expected_process_metadata_and_parent_groups() { let table = ProcessTable::new(); let parent = MockDriverProcess::new(); let child = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let child_pid = allocate_pid(&table); let parent_entry = table.register( parent_pid, "wasmvm", "grep", vec![String::from("-r"), String::from("foo")], create_context(0), parent, ); let child_entry = table.register( child_pid, "node", "node", vec![String::from("-e"), String::from("1+1")], create_context(parent_pid), child, ); assert_eq!(parent_entry.pid, 1); assert_eq!(child_entry.pid, 2); assert_eq!(parent_entry.pgid, 1); assert_eq!(parent_entry.sid, 1); assert_eq!(child_entry.pgid, 1); assert_eq!(child_entry.sid, 1); assert_eq!(child_entry.driver, "node"); } #[test] fn waitpid_resolves_for_exiting_and_already_exited_processes() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let process = MockDriverProcess::new(); let pid = allocate_pid(&table); table.register( pid, "wasmvm", "echo", vec![String::from("hello")], create_context(0), process.clone(), ); process.schedule_exit(Duration::from_millis(10), 0); assert_eq!( table.waitpid(pid).expect("waitpid should resolve"), (pid, 0) ); assert_eq!(table.zombie_timer_count(), 0); assert!( table.get(pid).is_none(), "waitpid should reap exited processes" ); let exited_pid = allocate_pid(&table); table.register( exited_pid, "wasmvm", "true", Vec::new(), create_context(0), MockDriverProcess::new(), ); table.mark_exited(exited_pid, 42); assert_eq!( table .waitpid(exited_pid) .expect("waitpid should resolve immediately"), (exited_pid, 42) ); assert_eq!(table.zombie_timer_count(), 0); assert!( table.get(exited_pid).is_none(), "waitpid should reap already exited processes" ); } #[test] fn long_lived_parent_retains_zombies_until_waited_under_pressure() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let parent = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let mut child_pids = Vec::new(); table.register( parent_pid, "wasmvm", "parent", Vec::new(), create_context(0), parent, ); for index in 0..100 { let child = MockDriverProcess::new(); let child_pid = allocate_pid(&table); table.register( child_pid, "wasmvm", format!("child-{index}"), Vec::new(), create_context(parent_pid), child.clone(), ); child.exit(index); child_pids.push((child_pid, index)); } for (child_pid, _) in &child_pids { assert_eq!( table .get(*child_pid) .expect("child zombie should be retained") .status, ProcessStatus::Exited ); } assert_eq!(table.zombie_timer_count(), child_pids.len()); for (child_pid, status) in child_pids { assert_eq!( table .waitpid_for(parent_pid, -1, WaitPidFlags::empty()) .expect("parent wait should succeed"), Some(agentos_kernel::process_table::ProcessWaitResult { pid: child_pid, status, event: ProcessWaitEvent::Exited, }) ); } assert_eq!(table.zombie_timer_count(), 0); } #[test] fn allocate_pid_wraps_without_reusing_live_or_zombie_entries() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let max_pid = i32::MAX as u32; let cursor_seed = MockDriverProcess::new(); let live_high = MockDriverProcess::new(); let zombie_high = MockDriverProcess::new(); let live_one = MockDriverProcess::new(); // Registering max_pid - 2 after the high PIDs moves the public allocation cursor back to max_pid - 1. table.register( max_pid - 1, "wasmvm", "live-high", Vec::new(), create_context(0), live_high, ); table.register( max_pid, "wasmvm", "zombie-high", Vec::new(), create_context(0), zombie_high.clone(), ); table.register( max_pid - 2, "wasmvm", "cursor-seed", Vec::new(), create_context(0), cursor_seed, ); table.register( 1, "wasmvm", "live-one", Vec::new(), create_context(0), live_one, ); zombie_high.exit(0); assert_eq!( table .get(max_pid) .expect("zombie high PID should remain allocated") .status, ProcessStatus::Exited ); assert_eq!(table.allocate_pid().expect("allocate wrapped pid"), 2); assert_eq!(table.allocate_pid().expect("allocate next pid"), 3); } #[test] fn waitpid_for_supports_wnohang_and_waiting_for_any_child() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let parent = MockDriverProcess::new(); let child_a = MockDriverProcess::new(); let child_b = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let child_a_pid = allocate_pid(&table); let child_b_pid = allocate_pid(&table); table.register( parent_pid, "wasmvm", "parent", Vec::new(), create_context(0), parent, ); table.register( child_a_pid, "wasmvm", "child-a", Vec::new(), create_context(parent_pid), child_a, ); table.register( child_b_pid, "wasmvm", "child-b", Vec::new(), create_context(parent_pid), child_b.clone(), ); assert_eq!( table .waitpid_for(parent_pid, -1, WaitPidFlags::WNOHANG) .expect("wnohang wait should succeed"), None ); child_b.exit(27); assert_eq!( table .waitpid_for(parent_pid, -1, WaitPidFlags::empty()) .expect("wait for any child should succeed"), Some(agentos_kernel::process_table::ProcessWaitResult { pid: child_b_pid, status: 27, event: ProcessWaitEvent::Exited, }) ); assert!( table.get(child_b_pid).is_none(), "waited child should be reaped" ); assert!( table.get(child_a_pid).is_some(), "other matching children should remain" ); } #[test] fn on_process_exit_runs_before_waitpid_waiters_are_notified() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let process = MockDriverProcess::new(); let pid = allocate_pid(&table); table.register( pid, "wasmvm", "sleep", vec![String::from("1")], create_context(0), process.clone(), ); let (callback_entered_tx, callback_entered_rx) = mpsc::channel(); let callback_gate = Arc::new((Mutex::new(false), Condvar::new())); let callback_gate_for_exit = Arc::clone(&callback_gate); table.set_on_process_exit(Some(Arc::new(move |_| { callback_entered_tx .send(()) .expect("callback should report entry"); let (released, wake) = &*callback_gate_for_exit; let mut released = released.lock().expect("callback gate lock poisoned"); while !*released { released = wake .wait(released) .expect("callback gate wait lock poisoned"); } }))); let waiter_table = table.clone(); let (wait_result_tx, wait_result_rx) = mpsc::channel(); let waiter = thread::spawn(move || { let result = waiter_table.waitpid(pid).expect("waitpid should resolve"); wait_result_tx .send(result) .expect("waiter should report exit result"); }); thread::sleep(Duration::from_millis(10)); let process_for_exit = process.clone(); let exit_handle = thread::spawn(move || { process_for_exit.exit(0); }); callback_entered_rx .recv_timeout(Duration::from_millis(100)) .expect("exit callback should run"); assert!(wait_result_rx.try_recv().is_err()); let (released, wake) = &*callback_gate; *released.lock().expect("callback gate lock poisoned") = true; wake.notify_all(); assert_eq!( wait_result_rx .recv_timeout(Duration::from_millis(100)) .expect("waitpid should resolve after callback"), (pid, 0) ); exit_handle.join().expect("exit thread should finish"); waiter.join().expect("waiter thread should finish"); } #[test] fn waitpid_for_reports_stopped_and_continued_children_once() { let table = ProcessTable::with_zombie_ttl(Duration::from_secs(3600)); let parent = MockDriverProcess::new(); let child = MockDriverProcess::new(); let parent_pid = allocate_pid(&table); let child_pid = allocate_pid(&table); table.register( parent_pid, "wasmvm", "parent", Vec::new(), create_context(0), parent.clone(), ); table.register( child_pid, "wasmvm", "child", Vec::new(), create_context(parent_pid), child, ); table.mark_stopped(child_pid, SIGSTOP); assert_eq!( table .waitpid_for(parent_pid, child_pid as i32, WaitPidFlags::WNOHANG) .expect("stopped child lookup should succeed"), None ); assert_eq!( table .waitpid_for( parent_pid, child_pid as i32, WaitPidFlags::WNOHANG | WaitPidFlags::WUNTRACED,