1use crate::Result;
2#[cfg(unix)]
3use crate::settings::settings;
4#[cfg(windows)]
5use crate::shell::HideConsoleWindow;
6use miette::IntoDiagnostic;
7use once_cell::sync::Lazy;
8use std::collections::HashMap;
9#[cfg(target_os = "linux")]
10use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
11use std::sync::Mutex;
12use std::time::{Duration, Instant};
13use sysinfo::ProcessesToUpdate;
14#[cfg(windows)]
15use windows_sys::Win32::Foundation::{CloseHandle, FILETIME, HANDLE};
16#[cfg(windows)]
17use windows_sys::Win32::System::Threading::{
18 GetProcessTimes, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
19};
20
21type ParentToChildren = HashMap<u32, Vec<u32>>;
23
24type ProcessNames = HashMap<u32, (String, Option<String>)>;
26
27const FULL_REFRESH_INTERVAL: Duration = Duration::from_secs(5);
33
34pub struct Procs {
35 system: Mutex<sysinfo::System>,
36 last_full_refresh: Mutex<Option<Instant>>,
39}
40
41pub static PROCS: Lazy<Procs> = Lazy::new(Procs::new);
42
43impl Default for Procs {
44 fn default() -> Self {
45 Self::new()
46 }
47}
48
49impl Procs {
50 pub fn new() -> Self {
51 Self {
64 system: Mutex::new(sysinfo::System::new()),
65 last_full_refresh: Mutex::new(None),
66 }
67 }
68
69 fn lock_system(&self) -> std::sync::MutexGuard<'_, sysinfo::System> {
70 self.system.lock().unwrap_or_else(|poisoned| {
71 warn!("System mutex was poisoned, recovering");
72 poisoned.into_inner()
73 })
74 }
75
76 pub fn title(&self, pid: u32) -> Option<String> {
77 self.lock_system()
78 .process(sysinfo::Pid::from_u32(pid))
79 .map(|p| p.name().to_string_lossy().to_string())
80 }
81
82 pub fn boot_time(&self) -> u64 {
89 sysinfo::System::boot_time()
90 }
91
92 pub fn start_time(&self, pid: u32) -> Option<u64> {
98 process_start_token(pid)
99 }
100
101 #[cfg(not(windows))]
102 fn start_time_matches(&self, pid: u32, expected: u64) -> bool {
103 self.start_time(pid) == Some(expected)
104 }
105
106 pub fn is_running(&self, pid: u32) -> bool {
107 #[cfg(unix)]
113 {
114 unsafe {
115 if libc::kill(pid as i32, 0) == 0 {
116 return true;
117 }
118 std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH)
119 }
120 }
121 #[cfg(not(unix))]
122 {
123 self.refresh_pids(&[pid]);
124 self.lock_system()
125 .process(sysinfo::Pid::from_u32(pid))
126 .is_some()
127 }
128 }
129
130 #[allow(dead_code)]
133 pub fn all_children(&self, pid: u32) -> Vec<u32> {
134 let system = self.lock_system();
135 let all = system.processes();
136 let mut children = vec![];
137 for (child_pid, process) in all {
138 let mut process = process;
139 while let Some(parent) = process.parent() {
140 if parent == sysinfo::Pid::from_u32(pid) {
141 children.push(child_pid.as_u32());
142 break;
143 }
144 match system.process(parent) {
145 Some(p) => process = p,
146 None => break,
147 }
148 }
149 }
150 children
151 }
152 pub fn collect_process_tree_info(&self) -> (ParentToChildren, ProcessNames) {
157 let system = self.lock_system();
158 let all = system.processes();
159 let mut parent_to_children: ParentToChildren = HashMap::new();
160 let mut process_info: ProcessNames = HashMap::new();
161
162 for (pid, proc) in all {
163 let pid_u32 = pid.as_u32();
164 process_info.insert(
165 pid_u32,
166 (
167 proc.name().to_string_lossy().to_string(),
168 proc.exe().map(|e| e.to_string_lossy().to_string()),
169 ),
170 );
171
172 if let Some(ppid) = proc.parent() {
173 parent_to_children
174 .entry(ppid.as_u32())
175 .or_default()
176 .push(pid_u32);
177 }
178 }
179
180 (parent_to_children, process_info)
181 }
182 pub async fn kill_process_group_async(
183 &self,
184 pid: u32,
185 stop_signal: i32,
186 stop_timeout: Option<std::time::Duration>,
187 ) -> Result<bool> {
188 tokio::task::spawn_blocking(move || {
189 PROCS.kill_process_group(pid, stop_signal, stop_timeout, None)
190 })
191 .await
192 .into_diagnostic()?
193 }
194
195 pub async fn kill_process_group_if_start_time_matches_async(
205 &self,
206 pid: u32,
207 expected_start_time: Option<u64>,
208 stop_signal: i32,
209 stop_timeout: Option<std::time::Duration>,
210 ) -> Result<bool> {
211 let Some(expected_start_time) = expected_start_time else {
212 warn!(
213 "no recorded start time for pid {pid}; refusing to signal it (identity cannot be bound to a process generation)"
214 );
215 return Ok(false);
216 };
217 tokio::task::spawn_blocking(move || {
218 PROCS.kill_process_group(pid, stop_signal, stop_timeout, Some(expected_start_time))
219 })
220 .await
221 .into_diagnostic()?
222 }
223
224 pub async fn kill_if_start_time_matches_async(
244 &self,
245 pid: u32,
246 expected_start_time: Option<u64>,
247 stop_signal: i32,
248 stop_timeout: Option<std::time::Duration>,
249 ) -> Result<bool> {
250 let Some(expected_start_time) = expected_start_time else {
251 warn!(
252 "no recorded start time for pid {pid}; refusing to signal it (identity cannot be bound to a process generation)"
253 );
254 return Ok(false);
255 };
256 tokio::task::spawn_blocking(move || {
257 PROCS.kill_if_start_time_matches(pid, expected_start_time, stop_signal, stop_timeout)
258 })
259 .await
260 .into_diagnostic()?
261 }
262
263 #[cfg(target_os = "linux")]
264 fn kill_if_start_time_matches(
265 &self,
266 pid: u32,
267 expected_start_time: u64,
268 stop_signal: i32,
269 stop_timeout: Option<std::time::Duration>,
270 ) -> Result<bool> {
271 let pidfd = match open_pidfd(pid) {
272 Ok(pidfd) => pidfd,
273 Err(err) if err.raw_os_error() == Some(libc::ESRCH) => {
274 debug!("process {pid} no longer exists");
275 return Ok(false);
276 }
277 Err(err) => {
278 return Err(miette::miette!(
279 "cannot securely identify process {pid}: {err}"
280 ));
281 }
282 };
283 if !self.verify_start_time_before_signal(pid, expected_start_time)? {
288 return Ok(false);
289 }
290 let target = [(pid, pidfd)];
291 let signal_name = signal_name(stop_signal);
292 debug!("sending {signal_name} to pinned process {pid}");
293 signal_pidfds(&target, stop_signal, signal_name)?;
294
295 let stop_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
298 let fast_ms = 10u64;
299 let slow_ms = 50u64;
300 let total_ms = stop_timeout.as_millis().max(1) as u64;
301 let fast_count = ((total_ms / fast_ms) as usize).min(10);
302 let fast_total_ms = fast_ms * fast_count as u64;
303 let slow_count = (total_ms.saturating_sub(fast_total_ms) / slow_ms) as usize;
304 for i in 0..fast_count {
305 std::thread::sleep(std::time::Duration::from_millis(fast_ms));
306 if !pidfd_is_running(&target[0].1) {
307 debug!(
308 "process {pid} terminated after {signal_name} ({} ms)",
309 (i + 1) as u64 * fast_ms
310 );
311 return Ok(true);
312 }
313 }
314 for i in 0..slow_count {
315 std::thread::sleep(std::time::Duration::from_millis(slow_ms));
316 if !pidfd_is_running(&target[0].1) {
317 debug!(
318 "process {pid} terminated after {signal_name} ({} ms)",
319 fast_total_ms + (i + 1) as u64 * slow_ms
320 );
321 return Ok(true);
322 }
323 }
324
325 warn!(
326 "process {pid} did not respond to {signal_name} after {}ms, sending SIGKILL",
327 stop_timeout.as_millis()
328 );
329 signal_pidfds(&target, libc::SIGKILL, "SIGKILL")?;
330 std::thread::sleep(std::time::Duration::from_millis(100));
332 Ok(true)
333 }
334
335 #[cfg(not(target_os = "linux"))]
336 fn kill_if_start_time_matches(
337 &self,
338 pid: u32,
339 expected_start_time: u64,
340 stop_signal: i32,
341 stop_timeout: Option<std::time::Duration>,
342 ) -> Result<bool> {
343 #[cfg(windows)]
347 let _pin = open_process_handle(pid).ok();
348
349 if !self.verify_start_time_before_signal(pid, expected_start_time)? {
350 return Ok(false);
351 }
352 self.kill(pid, stop_signal, stop_timeout, Some(expected_start_time))
353 }
354
355 fn verify_start_time_before_signal(&self, pid: u32, expected_start_time: u64) -> Result<bool> {
367 match self.start_time(pid) {
368 Some(current) if current == expected_start_time => Ok(true),
369 Some(current) => {
370 warn!(
371 "pid {pid} is not the recorded process (start time {current}, expected {expected_start_time}); not signalling it"
372 );
373 Ok(false)
374 }
375 None if self.is_running(pid) => Err(miette::miette!(
376 "cannot verify the identity of pid {pid}: its start time is unreadable; not signalling it"
377 )),
378 None => {
379 debug!("process {pid} no longer exists");
380 Ok(false)
381 }
382 }
383 }
384
385 #[cfg(unix)]
402 fn kill_process_group(
403 &self,
404 pid: u32,
405 stop_signal: i32,
406 stop_timeout: Option<std::time::Duration>,
407 expected_start_time: Option<u64>,
408 ) -> Result<bool> {
409 let pgid = pid as i32;
410 let signal_name = signal_name(stop_signal);
411
412 #[cfg(target_os = "linux")]
413 if let Some(expected) = expected_start_time {
414 return self.kill_process_group_with_pidfds(pid, expected, stop_signal, stop_timeout);
415 }
416
417 #[cfg(not(target_os = "linux"))]
429 if let Some(expected) = expected_start_time
430 && !self.start_time_matches(pid, expected)
431 {
432 debug!("process {pid} identity changed before killpg; refusing to signal it");
433 return Ok(false);
434 }
435
436 debug!("killing process group {pgid} with {signal_name}");
437
438 let ret = unsafe { libc::killpg(pgid, stop_signal) };
443 if ret == -1 {
444 let err = std::io::Error::last_os_error();
445 if err.raw_os_error() == Some(libc::ESRCH) {
446 debug!("process group {pgid} no longer exists");
447 return Ok(false);
448 }
449 if err.raw_os_error() == Some(libc::EPERM) {
450 return Err(miette::miette!(
451 "failed to send {signal_name} to process group {pgid}: permission denied"
452 ));
453 }
454 warn!("failed to send {signal_name} to process group {pgid}: {err}");
455 }
456
457 let stop_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
467 let fast_ms = 10u64;
468 let slow_ms = 50u64;
469 let total_ms = stop_timeout.as_millis().max(1) as u64;
470 let fast_count = ((total_ms / fast_ms) as usize).min(10);
471 let fast_total_ms = fast_ms * fast_count as u64;
472 let remaining_ms = total_ms.saturating_sub(fast_total_ms);
473 let slow_count = (remaining_ms / slow_ms) as usize;
474
475 let fast_checks =
476 std::iter::repeat_n(std::time::Duration::from_millis(fast_ms), fast_count);
477 let slow_checks =
478 std::iter::repeat_n(std::time::Duration::from_millis(slow_ms), slow_count);
479 let mut elapsed_ms = 0u64;
480
481 for sleep_duration in fast_checks.chain(slow_checks) {
482 std::thread::sleep(sleep_duration);
483 elapsed_ms += sleep_duration.as_millis() as u64;
484 if process_group_terminated(pgid) {
485 debug!("process group {pgid} terminated after {signal_name} ({elapsed_ms} ms)",);
486 return Ok(true);
487 }
488 }
489
490 warn!(
492 "process group {pgid} did not respond to {signal_name} after {}ms, sending SIGKILL",
493 stop_timeout.as_millis()
494 );
495 let ret = unsafe { libc::killpg(pgid, libc::SIGKILL) };
496 if ret == -1 {
497 let err = std::io::Error::last_os_error();
498 if err.raw_os_error() != Some(libc::ESRCH) {
499 warn!("failed to send SIGKILL to process group {pgid}: {err}");
500 }
501 }
502
503 for _ in 0..40 {
507 std::thread::sleep(std::time::Duration::from_millis(50));
508 if process_group_terminated(pgid) {
509 return Ok(true);
510 }
511 }
512 Err(miette::miette!(
515 "process group {pgid} still has members after SIGKILL \
516 (possibly stuck in uninterruptible sleep)"
517 ))
518 }
519
520 pub fn process_group_alive(&self, pid: u32) -> bool {
523 #[cfg(unix)]
524 {
525 !process_group_terminated(pid as i32)
526 }
527 #[cfg(not(unix))]
528 {
529 self.is_running(pid)
530 }
531 }
532
533 #[cfg(target_os = "linux")]
534 fn kill_process_group_with_pidfds(
535 &self,
536 pid: u32,
537 expected_start_time: u64,
538 _stop_signal: i32,
539 stop_timeout: Option<std::time::Duration>,
540 ) -> Result<bool> {
541 let leader = match open_pidfd(pid) {
542 Ok(pidfd) => pidfd,
543 Err(err) => {
544 warn!("cannot securely identify process group {pid}: {err}");
545 return Ok(false);
546 }
547 };
548 if !self.start_time_matches(pid, expected_start_time) {
549 debug!("process group {pid} leader identity changed before signaling");
550 return Ok(false);
551 }
552
553 let mut members = vec![(pid, leader)];
554 if let Err(err) = stop_pidfds(&members) {
555 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
556 return Err(err);
557 }
558 if !pidfd_is_running(&members[0].1) {
559 debug!("process group {pid} leader exited before it could be frozen");
560 return Ok(false);
561 }
562
563 loop {
567 let known_members = members.len();
568 let added = match extend_process_group_pidfds(pid as i32, &mut members) {
569 Ok(added) => added,
570 Err(err) => {
571 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
572 return Err(miette::miette!(
573 "failed to scan pinned process group {pid}: {err}"
574 ));
575 }
576 };
577 if added == 0 {
578 break;
579 }
580 if let Err(err) = stop_pidfds(&members[known_members..]) {
581 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
582 return Err(err);
583 }
584 }
585
586 warn!(
587 "force-terminating {} pinned orphan process(es) in group {pid}",
588 members.len()
589 );
590 if let Err(err) = signal_pidfds(&members, libc::SIGKILL, "SIGKILL") {
591 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
592 return Err(err);
593 }
594
595 let exit_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
596 let checks = exit_timeout.as_millis().max(1).div_ceil(50) as usize;
597 for _ in 0..checks {
598 if members.iter().all(|(_, pidfd)| !pidfd_is_running(pidfd)) {
599 return Ok(true);
600 }
601 std::thread::sleep(std::time::Duration::from_millis(50));
602 }
603 if members.iter().all(|(_, pidfd)| !pidfd_is_running(pidfd)) {
604 return Ok(true);
605 }
606
607 warn!("one or more pinned processes in orphan group {pid} remained alive after SIGKILL");
608 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
609 Ok(false)
610 }
611
612 #[cfg(not(unix))]
613 fn kill_process_group(
614 &self,
615 pid: u32,
616 _stop_signal: i32,
617 _stop_timeout: Option<std::time::Duration>,
618 expected_start_time: Option<u64>,
619 ) -> Result<bool> {
620 #[cfg(windows)]
623 let _identity_handle = if let Some(expected) = expected_start_time {
624 let handle = match open_process_handle(pid) {
625 Ok(handle) => handle,
626 Err(err) => {
627 warn!("cannot securely identify process {pid}: {err}");
628 return Ok(false);
629 }
630 };
631 if process_start_token_from_handle(handle.0) != Some(expected) {
632 debug!("process {pid} identity changed before taskkill");
633 return Ok(false);
634 }
635 Some(handle)
636 } else {
637 None
638 };
639
640 #[cfg(not(windows))]
641 if let Some(expected) = expected_start_time
642 && !self.start_time_matches(pid, expected)
643 {
644 debug!("process {pid} identity changed before termination");
645 return Ok(false);
646 }
647
648 self.kill(pid, 0, None, expected_start_time)
649 }
650
651 #[cfg(not(target_os = "linux"))]
669 fn kill(
670 &self,
671 pid: u32,
672 stop_signal: i32,
673 stop_timeout: Option<std::time::Duration>,
674 expected_start_time: Option<u64>,
675 ) -> Result<bool> {
676 debug!("killing process {pid}");
677
678 #[cfg(windows)]
679 {
680 let _ = (stop_signal, stop_timeout, expected_start_time);
683 let output = std::process::Command::new("taskkill")
688 .args(["/F", "/T", "/PID"])
689 .arg(pid.to_string())
690 .hide_console_window()
691 .output();
692 let taskkill_succeeded = match output {
693 Ok(o) if o.status.success() => {
694 debug!("taskkill /F /T /PID {pid} succeeded");
695 true
696 }
697 Ok(o) => {
698 debug!(
699 "taskkill /F /T /PID {pid} exited with status {}: {}",
700 o.status,
701 String::from_utf8_lossy(&o.stderr).trim()
702 );
703 false
704 }
705 Err(e) => {
706 debug!("failed to spawn taskkill for pid {pid}: {e}");
707 false
708 }
709 };
710 std::thread::sleep(std::time::Duration::from_millis(200));
714 if !taskkill_succeeded && self.is_running(pid) {
715 return Err(miette::miette!(
716 "taskkill failed and process {pid} is still running"
717 ));
718 }
719 Ok(true)
720 }
721
722 #[cfg(unix)]
723 {
724 let sysinfo_pid = sysinfo::Pid::from_u32(pid);
725 let signal_name = signal_name(stop_signal);
726 if let Some(expected) = expected_start_time
731 && !self.verify_start_time_before_signal(pid, expected)?
732 {
733 return Ok(false);
734 }
735 debug!("sending {signal_name} to process {pid}");
739 let ret = unsafe { libc::kill(pid as i32, stop_signal) };
740 if ret == -1 {
741 let err = std::io::Error::last_os_error();
742 if err.raw_os_error() == Some(libc::ESRCH) {
743 debug!("process {pid} no longer exists");
744 return Ok(false);
745 }
746 if err.raw_os_error() == Some(libc::EPERM) {
747 return Err(miette::miette!(
748 "failed to send {signal_name} to process {pid}: permission denied"
749 ));
750 }
751 return Err(miette::miette!(
752 "failed to send {signal_name} to process {pid}: {err}"
753 ));
754 }
755
756 let stop_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
759 let fast_ms = 10u64;
760 let slow_ms = 50u64;
761 let total_ms = stop_timeout.as_millis().max(1) as u64;
762 let fast_count = ((total_ms / fast_ms) as usize).min(10);
763 let fast_total_ms = fast_ms * fast_count as u64;
764 let remaining_ms = total_ms.saturating_sub(fast_total_ms);
765 let slow_count = (remaining_ms / slow_ms) as usize;
766
767 for i in 0..fast_count {
768 std::thread::sleep(std::time::Duration::from_millis(fast_ms));
769 self.refresh_pids(&[pid]);
770 if self.is_terminated_or_zombie(sysinfo_pid) {
771 debug!(
772 "process {pid} terminated after {signal_name} ({} ms)",
773 (i + 1) * fast_ms as usize
774 );
775 return Ok(true);
776 }
777 }
778
779 for i in 0..slow_count {
781 std::thread::sleep(std::time::Duration::from_millis(slow_ms));
782 self.refresh_pids(&[pid]);
783 if self.is_terminated_or_zombie(sysinfo_pid) {
784 debug!(
785 "process {pid} terminated after {signal_name} ({} ms)",
786 fast_total_ms + (i + 1) as u64 * slow_ms
787 );
788 return Ok(true);
789 }
790 }
791
792 if let Some(expected) = expected_start_time
797 && !self.start_time_matches(pid, expected)
798 {
799 debug!(
800 "process {pid} exited during the stop timeout and its PID was recycled; not sending SIGKILL"
801 );
802 return Ok(true);
803 }
804 warn!(
805 "process {pid} did not respond to {signal_name} after {}ms, sending SIGKILL",
806 stop_timeout.as_millis()
807 );
808 let ret = unsafe { libc::kill(pid as i32, libc::SIGKILL) };
809 if ret == -1 {
810 let err = std::io::Error::last_os_error();
811 if err.raw_os_error() != Some(libc::ESRCH) {
812 warn!("failed to send SIGKILL to process {pid}: {err}");
813 }
814 }
815
816 std::thread::sleep(std::time::Duration::from_millis(100));
818 Ok(true)
819 }
820 }
821
822 #[cfg(all(unix, not(target_os = "linux")))]
826 fn is_terminated_or_zombie(&self, sysinfo_pid: sysinfo::Pid) -> bool {
827 let system = self.lock_system();
828 match system.process(sysinfo_pid) {
829 None => true,
830 Some(process) => {
831 matches!(process.status(), sysinfo::ProcessStatus::Zombie)
832 }
833 }
834 }
835
836 pub(crate) fn refresh_processes(&self) {
837 let mut system = self.lock_system();
838 system.refresh_processes(ProcessesToUpdate::All, true);
839 #[cfg(windows)]
844 system.refresh_cpu_usage();
845 }
846
847 pub(crate) fn refresh_pids(&self, pids: &[u32]) {
850 let sysinfo_pids: Vec<sysinfo::Pid> =
851 pids.iter().map(|p| sysinfo::Pid::from_u32(*p)).collect();
852 self.lock_system()
853 .refresh_processes(ProcessesToUpdate::Some(&sysinfo_pids), true);
854 }
855
856 pub(crate) fn refresh_if_stale(&self) {
873 let mut last = self
874 .last_full_refresh
875 .lock()
876 .unwrap_or_else(|poisoned| poisoned.into_inner());
877 if last.is_none_or(|t| t.elapsed() >= FULL_REFRESH_INTERVAL) {
878 self.refresh_processes();
879 *last = Some(Instant::now());
880 }
881 }
882
883 pub fn get_batch_group_stats(&self, pids: &[u32]) -> Vec<(u32, Option<ProcessStats>)> {
889 if pids.is_empty() {
890 return Vec::new();
891 }
892
893 let system = self.lock_system();
894 let processes = system.processes();
895
896 let now = std::time::SystemTime::now()
897 .duration_since(std::time::UNIX_EPOCH)
898 .map(|d| d.as_secs())
899 .unwrap_or(0);
900
901 let mut children_map: std::collections::HashMap<sysinfo::Pid, Vec<sysinfo::Pid>> =
903 std::collections::HashMap::new();
904 for (child_pid, child) in processes {
905 if child.thread_kind().is_some() {
908 continue;
909 }
910 if let Some(ppid) = child.parent() {
911 children_map.entry(ppid).or_default().push(*child_pid);
912 }
913 }
914
915 pids.iter()
916 .map(|&pid| {
917 let root_pid = sysinfo::Pid::from_u32(pid);
918 let Some(root) = processes.get(&root_pid) else {
919 return (pid, None);
920 };
921
922 let root_disk = root.disk_usage();
923 let mut stats = ProcessStats {
924 cpu_percent: root.cpu_usage(),
925 memory_bytes: root.memory(),
926 uptime_secs: now.saturating_sub(root.start_time()),
927 disk_read_bytes: root_disk.read_bytes,
928 disk_write_bytes: root_disk.written_bytes,
929 };
930
931 let mut queue = std::collections::VecDeque::new();
933 if let Some(direct_children) = children_map.get(&root_pid) {
934 queue.extend(direct_children);
935 }
936 while let Some(child_pid) = queue.pop_front() {
937 if let Some(child) = processes.get(&child_pid) {
938 let disk = child.disk_usage();
939 stats.cpu_percent += child.cpu_usage();
940 stats.memory_bytes += child.memory();
941 stats.disk_read_bytes += disk.read_bytes;
942 stats.disk_write_bytes += disk.written_bytes;
943 }
944 if let Some(grandchildren) = children_map.get(&child_pid) {
945 queue.extend(grandchildren);
946 }
947 }
948
949 (pid, Some(stats))
950 })
951 .collect()
952 }
953 pub fn refresh_and_get_batch_stats(&self, pids: &[u32]) -> HashMap<u32, ProcessStats> {
963 self.refresh_processes();
964 self.get_batch_group_stats(pids)
965 .into_iter()
966 .filter_map(|(pid, stats)| stats.map(|s| (pid, s)))
967 .collect()
968 }
969
970 pub fn refresh_and_get_batch_stats_if_stale(&self, pids: &[u32]) -> HashMap<u32, ProcessStats> {
977 self.refresh_if_stale();
978 self.get_batch_group_stats(pids)
979 .into_iter()
980 .filter_map(|(pid, stats)| stats.map(|s| (pid, s)))
981 .collect()
982 }
983
984 pub fn get_batch_tree_stats_map(&self, pids: &[u32]) -> HashMap<u32, ProcessStats> {
986 self.get_batch_group_stats(pids)
987 .into_iter()
988 .filter_map(|(pid, stats)| stats.map(|stats| (pid, stats)))
989 .collect()
990 }
991
992 pub fn get_stats(&self, pid: u32) -> Option<ProcessStats> {
994 self.get_batch_group_stats(&[pid])
995 .into_iter()
996 .next()
997 .and_then(|(_, stats)| stats)
998 }
999
1000 pub fn get_extended_stats(&self, pid: u32) -> Option<ExtendedProcessStats> {
1002 let system = self.lock_system();
1003 let processes = system.processes();
1004 let root_pid = sysinfo::Pid::from_u32(pid);
1005 let p = processes.get(&root_pid)?;
1006
1007 let now = std::time::SystemTime::now()
1008 .duration_since(std::time::UNIX_EPOCH)
1009 .map(|d| d.as_secs())
1010 .unwrap_or(0);
1011
1012 let root_disk = p.disk_usage();
1013 let mut aggregate_stats = ProcessStats {
1014 cpu_percent: p.cpu_usage(),
1015 memory_bytes: p.memory(),
1016 uptime_secs: now.saturating_sub(p.start_time()),
1017 disk_read_bytes: root_disk.read_bytes,
1018 disk_write_bytes: root_disk.written_bytes,
1019 };
1020
1021 let mut children_map: HashMap<sysinfo::Pid, Vec<sysinfo::Pid>> = HashMap::new();
1022 for (child_pid, child) in processes {
1023 if let Some(ppid) = child.parent() {
1024 children_map.entry(ppid).or_default().push(*child_pid);
1025 }
1026 }
1027
1028 let mut queue = std::collections::VecDeque::new();
1029 if let Some(direct_children) = children_map.get(&root_pid) {
1030 queue.extend(direct_children);
1031 }
1032 while let Some(child_pid) = queue.pop_front() {
1033 if let Some(child) = processes.get(&child_pid) {
1034 let disk = child.disk_usage();
1035 aggregate_stats.cpu_percent += child.cpu_usage();
1036 aggregate_stats.memory_bytes += child.memory();
1037 aggregate_stats.disk_read_bytes += disk.read_bytes;
1038 aggregate_stats.disk_write_bytes += disk.written_bytes;
1039 }
1040 if let Some(grandchildren) = children_map.get(&child_pid) {
1041 queue.extend(grandchildren);
1042 }
1043 }
1044
1045 Some(ExtendedProcessStats {
1046 name: p.name().to_string_lossy().to_string(),
1047 status: format!("{:?}", p.status()),
1048 cpu_percent: aggregate_stats.cpu_percent,
1049 memory_bytes: aggregate_stats.memory_bytes,
1050 virtual_memory_bytes: p.virtual_memory(),
1051 uptime_secs: aggregate_stats.uptime_secs,
1052 thread_count: p.tasks().map(|t| t.len()).unwrap_or(0),
1053 })
1054 }
1055}
1056
1057#[cfg(target_os = "linux")]
1058fn process_start_token(pid: u32) -> Option<u64> {
1059 let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
1060 let command_end = stat.rfind(')')?;
1061 stat.get(command_end + 1..)?
1063 .split_whitespace()
1064 .nth(19)?
1065 .parse()
1066 .ok()
1067}
1068
1069#[cfg(target_os = "macos")]
1070fn process_start_token(pid: u32) -> Option<u64> {
1071 let mut info = std::mem::MaybeUninit::<libc::proc_bsdinfo>::zeroed();
1072 let size = std::mem::size_of::<libc::proc_bsdinfo>() as i32;
1073 let read = unsafe {
1074 libc::proc_pidinfo(
1075 pid as i32,
1076 libc::PROC_PIDTBSDINFO,
1077 0,
1078 info.as_mut_ptr().cast(),
1079 size,
1080 )
1081 };
1082 if read != size {
1083 return None;
1084 }
1085 let info = unsafe { info.assume_init() };
1086 info.pbi_start_tvsec
1087 .checked_mul(1_000_000)?
1088 .checked_add(info.pbi_start_tvusec)
1089}
1090
1091#[cfg(windows)]
1092fn process_start_token(pid: u32) -> Option<u64> {
1093 let handle = open_process_handle(pid).ok()?;
1094 process_start_token_from_handle(handle.0)
1095}
1096
1097#[cfg(windows)]
1098fn process_start_token_from_handle(handle: HANDLE) -> Option<u64> {
1099 let mut creation = FILETIME {
1100 dwLowDateTime: 0,
1101 dwHighDateTime: 0,
1102 };
1103 let mut exit = creation;
1104 let mut kernel = creation;
1105 let mut user = creation;
1106 let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
1107 if ok == 0 {
1108 return None;
1109 }
1110
1111 Some((u64::from(creation.dwHighDateTime) << 32) | u64::from(creation.dwLowDateTime))
1112}
1113
1114#[cfg(windows)]
1115struct ProcessHandle(HANDLE);
1116
1117#[cfg(windows)]
1118impl Drop for ProcessHandle {
1119 fn drop(&mut self) {
1120 unsafe {
1121 CloseHandle(self.0);
1122 }
1123 }
1124}
1125
1126#[cfg(windows)]
1127fn open_process_handle(pid: u32) -> std::io::Result<ProcessHandle> {
1128 let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
1129 if handle.is_null() {
1130 return Err(std::io::Error::last_os_error());
1131 }
1132 Ok(ProcessHandle(handle))
1133}
1134
1135#[cfg(not(any(target_os = "linux", target_os = "macos", windows)))]
1136fn process_start_token(pid: u32) -> Option<u64> {
1137 let mut system = sysinfo::System::new();
1138 let sysinfo_pid = sysinfo::Pid::from_u32(pid);
1139 system.refresh_processes(ProcessesToUpdate::Some(&[sysinfo_pid]), true);
1140 system
1141 .process(sysinfo_pid)
1142 .map(|process| process.start_time())
1143}
1144
1145#[cfg(target_os = "linux")]
1146fn open_pidfd(pid: u32) -> std::io::Result<OwnedFd> {
1147 let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
1148 if fd < 0 {
1149 return Err(std::io::Error::last_os_error());
1150 }
1151 Ok(unsafe { OwnedFd::from_raw_fd(fd as i32) })
1152}
1153
1154#[cfg(target_os = "linux")]
1155fn pidfd_is_running(pidfd: &OwnedFd) -> bool {
1156 match try_pidfd_is_running(pidfd) {
1157 Ok(running) => running,
1158 Err(err) => {
1159 warn!("failed to poll pidfd {}: {err}", pidfd.as_raw_fd());
1160 true
1161 }
1162 }
1163}
1164
1165#[cfg(target_os = "linux")]
1166fn try_pidfd_is_running(pidfd: &OwnedFd) -> std::io::Result<bool> {
1167 let mut pollfd = libc::pollfd {
1168 fd: pidfd.as_raw_fd(),
1169 events: libc::POLLIN,
1170 revents: 0,
1171 };
1172 let result = unsafe { libc::poll(&mut pollfd, 1, 0) };
1173 if result < 0 {
1174 return Err(std::io::Error::last_os_error());
1175 }
1176 Ok(result == 0)
1177}
1178
1179#[cfg(target_os = "linux")]
1180fn signal_pidfds(members: &[(u32, OwnedFd)], signal: i32, signal_name: &str) -> Result<()> {
1181 for (pid, pidfd) in members {
1182 if !pidfd_is_running(pidfd) {
1183 continue;
1184 }
1185 let result = unsafe {
1186 libc::syscall(
1187 libc::SYS_pidfd_send_signal,
1188 pidfd.as_raw_fd(),
1189 signal,
1190 std::ptr::null::<libc::siginfo_t>(),
1191 0,
1192 )
1193 };
1194 if result == -1 {
1195 let err = std::io::Error::last_os_error();
1196 if err.raw_os_error() == Some(libc::ESRCH) {
1197 continue;
1198 }
1199 return Err(miette::miette!(
1200 "failed to send {signal_name} to pinned process {pid}: {err}"
1201 ));
1202 }
1203 }
1204 Ok(())
1205}
1206
1207#[cfg(target_os = "linux")]
1208fn stop_pidfds(members: &[(u32, OwnedFd)]) -> Result<()> {
1209 signal_pidfds(members, libc::SIGSTOP, "SIGSTOP")?;
1210 for _ in 0..200 {
1211 if members.iter().all(|(pid, pidfd)| {
1212 !pidfd_is_running(pidfd) || matches!(linux_process_state(*pid), Some('T' | 't'))
1213 }) {
1214 return Ok(());
1215 }
1216 std::thread::sleep(std::time::Duration::from_millis(5));
1217 }
1218 Err(miette::miette!(
1219 "timed out while freezing orphan process group"
1220 ))
1221}
1222
1223#[cfg(target_os = "linux")]
1224fn extend_process_group_pidfds(
1225 pgid: i32,
1226 members: &mut Vec<(u32, OwnedFd)>,
1227) -> std::io::Result<usize> {
1228 let entries = std::fs::read_dir("/proc")?;
1229 let mut added = 0;
1230 for entry in entries {
1231 let entry = entry?;
1232 let Some(pid) = entry
1233 .file_name()
1234 .to_str()
1235 .and_then(|name| name.parse::<u32>().ok())
1236 else {
1237 continue;
1238 };
1239 let Some(observed_identity) = linux_process_identity(pid) else {
1240 continue;
1241 };
1242 if observed_identity.0 != pgid {
1243 continue;
1244 }
1245 let mut already_pinned = false;
1246 for (known_pid, pidfd) in members.iter() {
1247 if *known_pid == pid && try_pidfd_is_running(pidfd)? {
1248 already_pinned = true;
1249 break;
1250 }
1251 }
1252 if already_pinned {
1253 continue;
1254 }
1255
1256 let pidfd = match open_pidfd(pid) {
1257 Ok(pidfd) => pidfd,
1258 Err(err) if err.raw_os_error() == Some(libc::ESRCH) => continue,
1259 Err(err) => return Err(err),
1260 };
1261 if linux_process_identity(pid) != Some(observed_identity) {
1262 return Err(std::io::Error::other(format!(
1263 "process {pid} identity changed while pinning group {pgid}"
1264 )));
1265 }
1266 members.push((pid, pidfd));
1267 added += 1;
1268 }
1269 Ok(added)
1270}
1271
1272#[cfg(target_os = "linux")]
1273fn linux_process_identity(pid: u32) -> Option<(i32, u64)> {
1274 let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
1275 let command_end = stat.rfind(')')?;
1276 let fields: Vec<_> = stat.get(command_end + 1..)?.split_whitespace().collect();
1277 Some((fields.get(2)?.parse().ok()?, fields.get(19)?.parse().ok()?))
1280}
1281
1282#[cfg(target_os = "linux")]
1283fn linux_process_state(pid: u32) -> Option<char> {
1284 let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
1285 let command_end = stat.rfind(')')?;
1286 stat.get(command_end + 1..)?
1287 .split_whitespace()
1288 .next()?
1289 .chars()
1290 .next()
1291}
1292
1293#[derive(Debug, Clone, Copy)]
1294pub struct ProcessStats {
1295 pub cpu_percent: f32,
1296 pub memory_bytes: u64,
1297 pub uptime_secs: u64,
1298 pub disk_read_bytes: u64,
1299 pub disk_write_bytes: u64,
1300}
1301
1302impl ProcessStats {
1303 pub fn memory_display(&self) -> String {
1304 format_bytes(self.memory_bytes)
1305 }
1306
1307 pub fn cpu_display(&self) -> String {
1308 format!("{:.1}%", self.cpu_percent)
1309 }
1310
1311 pub fn uptime_display(&self) -> String {
1312 format_duration(self.uptime_secs)
1313 }
1314
1315 pub fn disk_read_display(&self) -> String {
1316 format_bytes_per_sec(self.disk_read_bytes)
1317 }
1318
1319 pub fn disk_write_display(&self) -> String {
1320 format_bytes_per_sec(self.disk_write_bytes)
1321 }
1322}
1323
1324#[derive(Debug, Clone)]
1325pub struct ExtendedProcessStats {
1326 pub name: String,
1327 pub status: String,
1328 pub cpu_percent: f32,
1329 pub memory_bytes: u64,
1330 pub virtual_memory_bytes: u64,
1331 pub uptime_secs: u64,
1332 pub thread_count: usize,
1333}
1334
1335fn format_bytes(bytes: u64) -> String {
1336 humanbyte::to_string(bytes, humanbyte::Format::IEC)
1337}
1338
1339fn format_duration(secs: u64) -> String {
1340 if secs < 60 {
1341 format!("{secs}s")
1342 } else if secs < 3600 {
1343 format!("{}m {}s", secs / 60, secs % 60)
1344 } else if secs < 86400 {
1345 let hours = secs / 3600;
1346 let mins = (secs % 3600) / 60;
1347 format!("{hours}h {mins}m")
1348 } else {
1349 let days = secs / 86400;
1350 let hours = (secs % 86400) / 3600;
1351 format!("{days}d {hours}h")
1352 }
1353}
1354
1355fn format_bytes_per_sec(bytes: u64) -> String {
1356 format!("{}/s", humanbyte::to_string(bytes, humanbyte::Format::IEC))
1357}
1358
1359#[cfg(unix)]
1375fn process_group_terminated(pgid: i32) -> bool {
1376 unsafe { libc::killpg(pgid, 0) != 0 }
1377}
1378
1379#[cfg(unix)]
1380fn signal_name(sig: i32) -> &'static str {
1381 match sig {
1382 libc::SIGHUP => "SIGHUP",
1383 libc::SIGINT => "SIGINT",
1384 libc::SIGQUIT => "SIGQUIT",
1385 libc::SIGTERM => "SIGTERM",
1386 libc::SIGUSR1 => "SIGUSR1",
1387 libc::SIGUSR2 => "SIGUSR2",
1388 libc::SIGKILL => "SIGKILL",
1389 _ => "UNKNOWN",
1390 }
1391}
1392
1393#[cfg(test)]
1394mod format_tests {
1395 use super::*;
1396
1397 #[test]
1398 fn process_start_time_check_rejects_mismatch() {
1399 let procs = Procs::new();
1400 let pid = std::process::id();
1401 procs.refresh_pids(&[pid]);
1402 let actual = procs
1403 .start_time(pid)
1404 .expect("current process should have a start time");
1405
1406 assert_ne!(procs.start_time(pid), Some(actual.saturating_add(1)));
1407 }
1408
1409 #[test]
1410 fn test_format_bytes() {
1411 assert_eq!(format_bytes(512), "512 B");
1412 assert_eq!(format_bytes(1024), "1.0 KiB");
1413 assert_eq!(format_bytes(1536), "1.5 KiB");
1414 assert_eq!(format_bytes(50 * 1024 * 1024), "50.0 MiB");
1415 assert_eq!(format_bytes(3 * 1024 * 1024 * 1024), "3.0 GiB");
1416 assert_eq!(format_bytes(1100 * 1024 * 1024 * 1024), "1.1 TiB");
1418 }
1419
1420 #[test]
1421 fn test_format_bytes_per_sec() {
1422 assert_eq!(format_bytes_per_sec(512), "512 B/s");
1423 assert_eq!(format_bytes_per_sec(1536), "1.5 KiB/s");
1424 assert_eq!(format_bytes_per_sec(2 * 1024 * 1024), "2.0 MiB/s");
1425 }
1426}
1427
1428#[cfg(all(test, unix))]
1429mod tests {
1430 use super::*;
1431 use std::os::unix::process::CommandExt;
1432 use std::process::{Child, Command, Stdio};
1433 use std::time::{Duration, Instant};
1434
1435 struct ChildGuard(Child);
1436
1437 impl Drop for ChildGuard {
1438 fn drop(&mut self) {
1439 let pid = self.0.id() as i32;
1440 let _ = unsafe { libc::killpg(pid, libc::SIGKILL) };
1442 let _ = self.0.wait();
1443 }
1444 }
1445
1446 #[tokio::test]
1447 async fn orphan_identity_checked_group_kill_rejects_mismatch() {
1448 let mut command = Command::new("sleep");
1449 command
1450 .arg("30")
1451 .stdin(Stdio::null())
1452 .stdout(Stdio::null())
1453 .stderr(Stdio::null());
1454 unsafe {
1455 command.pre_exec(|| {
1456 if libc::setsid() == -1 {
1457 return Err(std::io::Error::last_os_error());
1458 }
1459 Ok(())
1460 });
1461 }
1462
1463 let child = command.spawn().expect("failed to spawn test process");
1464 let pid = child.id();
1465 let _child = ChildGuard(child);
1466
1467 PROCS.refresh_pids(&[pid]);
1468 let actual_start_time = PROCS
1469 .start_time(pid)
1470 .expect("test process should have a start time");
1471
1472 let killed = PROCS
1473 .kill_process_group_if_start_time_matches_async(
1474 pid,
1475 Some(actual_start_time.saturating_add(1)),
1476 libc::SIGTERM,
1477 Some(Duration::from_millis(100)),
1478 )
1479 .await
1480 .expect("identity-checked kill should not error");
1481
1482 assert!(!killed);
1483 assert!(PROCS.is_running(pid), "mismatched process must survive");
1484 }
1485
1486 #[cfg(all(unix, not(target_os = "linux")))]
1491 #[tokio::test]
1492 async fn identity_checked_group_kill_reverifies_inside_blocking_op() {
1493 let mut command = Command::new("sleep");
1494 command
1495 .arg("30")
1496 .stdin(Stdio::null())
1497 .stdout(Stdio::null())
1498 .stderr(Stdio::null());
1499 unsafe {
1500 command.pre_exec(|| {
1501 if libc::setsid() == -1 {
1502 return Err(std::io::Error::last_os_error());
1503 }
1504 Ok(())
1505 });
1506 }
1507
1508 let child = command.spawn().expect("failed to spawn test process");
1509 let pid = child.id();
1510 let _child = ChildGuard(child);
1511
1512 PROCS.refresh_pids(&[pid]);
1513 let actual_start_time = PROCS
1514 .start_time(pid)
1515 .expect("test process should have a start time");
1516
1517 let killed = PROCS
1518 .kill_process_group_if_start_time_matches_async(
1519 pid,
1520 Some(actual_start_time),
1521 libc::SIGTERM,
1522 Some(Duration::from_millis(100)),
1523 )
1524 .await
1525 .expect("identity-checked kill should not error");
1526
1527 assert!(killed, "matching generation must be signalled");
1528 assert!(
1529 !PROCS.is_running(pid),
1530 "signalled process group must be gone"
1531 );
1532 }
1533
1534 #[test]
1535 fn get_stats_includes_descendant_rss() {
1536 let mut command = Command::new("sh");
1537 command
1538 .args(["-c", "sleep 30 & wait"])
1539 .stdin(Stdio::null())
1540 .stdout(Stdio::null())
1541 .stderr(Stdio::null());
1542 unsafe {
1543 command.pre_exec(|| {
1544 if libc::setsid() == -1 {
1545 return Err(std::io::Error::last_os_error());
1546 }
1547 Ok(())
1548 });
1549 }
1550
1551 let parent = command.spawn().expect("failed to spawn process tree");
1552 let parent_pid = parent.id();
1553 let _parent = ChildGuard(parent);
1554
1555 let procs = Procs::new();
1556 let deadline = Instant::now() + Duration::from_secs(5);
1557 let mut child_pids = Vec::new();
1558 while Instant::now() < deadline {
1559 procs.refresh_processes();
1560 child_pids = procs.all_children(parent_pid);
1561 if !child_pids.is_empty() {
1562 break;
1563 }
1564 std::thread::sleep(Duration::from_millis(50));
1565 }
1566 assert!(
1567 !child_pids.is_empty(),
1568 "test process tree did not appear under parent pid {parent_pid}"
1569 );
1570
1571 procs.refresh_processes();
1572 child_pids = procs.all_children(parent_pid);
1573 assert!(
1574 !child_pids.is_empty(),
1575 "test process tree disappeared under parent pid {parent_pid}"
1576 );
1577 let root_pid = sysinfo::Pid::from_u32(parent_pid);
1578 let direct_memory = {
1579 let system = procs.lock_system();
1580 system
1581 .process(root_pid)
1582 .expect("parent process should exist")
1583 .memory()
1584 };
1585 let descendant_memory = {
1586 let system = procs.lock_system();
1587 child_pids
1588 .iter()
1589 .filter_map(|pid| system.process(sysinfo::Pid::from_u32(*pid)))
1590 .map(|process| process.memory())
1591 .sum::<u64>()
1592 };
1593 assert!(
1594 descendant_memory > 0,
1595 "descendants {child_pids:?} should have nonzero RSS"
1596 );
1597
1598 let stats = procs
1599 .get_stats(parent_pid)
1600 .expect("parent process should have aggregate stats");
1601
1602 assert_eq!(
1603 stats.memory_bytes,
1604 direct_memory + descendant_memory,
1605 "get_stats should include descendant RSS for parent pid {parent_pid}; \
1606 descendants: {child_pids:?}, direct RSS: {direct_memory}, \
1607 descendant RSS: {descendant_memory}, reported RSS: {}",
1608 stats.memory_bytes
1609 );
1610 }
1611
1612 #[test]
1613 fn full_refresh_is_throttled_by_ttl() {
1614 let procs = Procs::new();
1615
1616 assert!(
1618 procs.last_full_refresh.lock().unwrap().is_none(),
1619 "fresh Procs should have no recorded refresh"
1620 );
1621 procs.refresh_if_stale();
1622 let first = procs.last_full_refresh.lock().unwrap().unwrap();
1623 assert!(first.elapsed() < FULL_REFRESH_INTERVAL);
1624
1625 procs.refresh_if_stale();
1628 let second = procs.last_full_refresh.lock().unwrap().unwrap();
1629 assert_eq!(
1630 second, first,
1631 "refresh_if_stale within TTL must skip the refresh and keep the timestamp"
1632 );
1633
1634 let expired = Instant::now()
1637 .checked_sub(FULL_REFRESH_INTERVAL + Duration::from_secs(1))
1638 .expect("system has been up long enough to backdate by 6s");
1639 *procs.last_full_refresh.lock().unwrap() = Some(expired);
1640 procs.refresh_if_stale();
1641 let third = procs.last_full_refresh.lock().unwrap().unwrap();
1642 assert!(
1643 third > expired,
1644 "refresh_if_stale after expired TTL must refresh and advance the timestamp"
1645 );
1646 assert!(
1647 third.elapsed() < FULL_REFRESH_INTERVAL,
1648 "fresh timestamp after expired-TTL refresh should be recent"
1649 );
1650 }
1651}