1use crate::Result;
2#[cfg(unix)]
3use crate::settings::settings;
4use miette::IntoDiagnostic;
5use once_cell::sync::Lazy;
6use std::collections::HashMap;
7#[cfg(target_os = "linux")]
8use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
9#[cfg(windows)]
10use std::os::windows::process::CommandExt;
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(any(target_os = "linux", 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(
202 &self,
203 pid: u32,
204 expected_start_time: u64,
205 stop_signal: i32,
206 stop_timeout: Option<std::time::Duration>,
207 ) -> Result<bool> {
208 tokio::task::spawn_blocking(move || {
209 PROCS.kill_process_group(pid, stop_signal, stop_timeout, Some(expected_start_time))
210 })
211 .await
212 .into_diagnostic()?
213 }
214
215 #[cfg(unix)]
232 fn kill_process_group(
233 &self,
234 pid: u32,
235 stop_signal: i32,
236 stop_timeout: Option<std::time::Duration>,
237 expected_start_time: Option<u64>,
238 ) -> Result<bool> {
239 let pgid = pid as i32;
240 let signal_name = signal_name(stop_signal);
241
242 #[cfg(target_os = "linux")]
243 if let Some(expected) = expected_start_time {
244 return self.kill_process_group_with_pidfds(pid, expected, stop_signal, stop_timeout);
245 }
246
247 #[cfg(not(target_os = "linux"))]
251 if expected_start_time.is_some() {
252 warn!(
253 "cannot securely identify process group {pgid} on this platform; refusing to signal it"
254 );
255 return Ok(false);
256 }
257
258 debug!("killing process group {pgid} with {signal_name}");
259
260 let ret = unsafe { libc::killpg(pgid, stop_signal) };
265 if ret == -1 {
266 let err = std::io::Error::last_os_error();
267 if err.raw_os_error() == Some(libc::ESRCH) {
268 debug!("process group {pgid} no longer exists");
269 return Ok(false);
270 }
271 if err.raw_os_error() == Some(libc::EPERM) {
272 return Err(miette::miette!(
273 "failed to send {signal_name} to process group {pgid}: permission denied"
274 ));
275 }
276 warn!("failed to send {signal_name} to process group {pgid}: {err}");
277 }
278
279 let stop_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
289 let fast_ms = 10u64;
290 let slow_ms = 50u64;
291 let total_ms = stop_timeout.as_millis().max(1) as u64;
292 let fast_count = ((total_ms / fast_ms) as usize).min(10);
293 let fast_total_ms = fast_ms * fast_count as u64;
294 let remaining_ms = total_ms.saturating_sub(fast_total_ms);
295 let slow_count = (remaining_ms / slow_ms) as usize;
296
297 let fast_checks =
298 std::iter::repeat_n(std::time::Duration::from_millis(fast_ms), fast_count);
299 let slow_checks =
300 std::iter::repeat_n(std::time::Duration::from_millis(slow_ms), slow_count);
301 let mut elapsed_ms = 0u64;
302
303 for sleep_duration in fast_checks.chain(slow_checks) {
304 std::thread::sleep(sleep_duration);
305 elapsed_ms += sleep_duration.as_millis() as u64;
306 if process_group_terminated(pgid) {
307 debug!("process group {pgid} terminated after {signal_name} ({elapsed_ms} ms)",);
308 return Ok(true);
309 }
310 }
311
312 warn!(
314 "process group {pgid} did not respond to {signal_name} after {}ms, sending SIGKILL",
315 stop_timeout.as_millis()
316 );
317 let ret = unsafe { libc::killpg(pgid, libc::SIGKILL) };
318 if ret == -1 {
319 let err = std::io::Error::last_os_error();
320 if err.raw_os_error() != Some(libc::ESRCH) {
321 warn!("failed to send SIGKILL to process group {pgid}: {err}");
322 }
323 }
324
325 for _ in 0..40 {
329 std::thread::sleep(std::time::Duration::from_millis(50));
330 if process_group_terminated(pgid) {
331 return Ok(true);
332 }
333 }
334 Err(miette::miette!(
337 "process group {pgid} still has members after SIGKILL \
338 (possibly stuck in uninterruptible sleep)"
339 ))
340 }
341
342 pub fn process_group_alive(&self, pid: u32) -> bool {
345 #[cfg(unix)]
346 {
347 !process_group_terminated(pid as i32)
348 }
349 #[cfg(not(unix))]
350 {
351 self.is_running(pid)
352 }
353 }
354
355 #[cfg(target_os = "linux")]
356 fn kill_process_group_with_pidfds(
357 &self,
358 pid: u32,
359 expected_start_time: u64,
360 _stop_signal: i32,
361 stop_timeout: Option<std::time::Duration>,
362 ) -> Result<bool> {
363 let leader = match open_pidfd(pid) {
364 Ok(pidfd) => pidfd,
365 Err(err) => {
366 warn!("cannot securely identify process group {pid}: {err}");
367 return Ok(false);
368 }
369 };
370 if !self.start_time_matches(pid, expected_start_time) {
371 debug!("process group {pid} leader identity changed before signaling");
372 return Ok(false);
373 }
374
375 let mut members = vec![(pid, leader)];
376 if let Err(err) = stop_pidfds(&members) {
377 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
378 return Err(err);
379 }
380 if !pidfd_is_running(&members[0].1) {
381 debug!("process group {pid} leader exited before it could be frozen");
382 return Ok(false);
383 }
384
385 loop {
389 let known_members = members.len();
390 let added = match extend_process_group_pidfds(pid as i32, &mut members) {
391 Ok(added) => added,
392 Err(err) => {
393 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
394 return Err(miette::miette!(
395 "failed to scan pinned process group {pid}: {err}"
396 ));
397 }
398 };
399 if added == 0 {
400 break;
401 }
402 if let Err(err) = stop_pidfds(&members[known_members..]) {
403 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
404 return Err(err);
405 }
406 }
407
408 warn!(
409 "force-terminating {} pinned orphan process(es) in group {pid}",
410 members.len()
411 );
412 if let Err(err) = signal_pidfds(&members, libc::SIGKILL, "SIGKILL") {
413 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
414 return Err(err);
415 }
416
417 let exit_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
418 let checks = exit_timeout.as_millis().max(1).div_ceil(50) as usize;
419 for _ in 0..checks {
420 if members.iter().all(|(_, pidfd)| !pidfd_is_running(pidfd)) {
421 return Ok(true);
422 }
423 std::thread::sleep(std::time::Duration::from_millis(50));
424 }
425 if members.iter().all(|(_, pidfd)| !pidfd_is_running(pidfd)) {
426 return Ok(true);
427 }
428
429 warn!("one or more pinned processes in orphan group {pid} remained alive after SIGKILL");
430 let _ = signal_pidfds(&members, libc::SIGCONT, "SIGCONT");
431 Ok(false)
432 }
433
434 #[cfg(not(unix))]
435 fn kill_process_group(
436 &self,
437 pid: u32,
438 _stop_signal: i32,
439 _stop_timeout: Option<std::time::Duration>,
440 expected_start_time: Option<u64>,
441 ) -> Result<bool> {
442 #[cfg(windows)]
445 let _identity_handle = if let Some(expected) = expected_start_time {
446 let handle = match open_process_handle(pid) {
447 Ok(handle) => handle,
448 Err(err) => {
449 warn!("cannot securely identify process {pid}: {err}");
450 return Ok(false);
451 }
452 };
453 if process_start_token_from_handle(handle.0) != Some(expected) {
454 debug!("process {pid} identity changed before taskkill");
455 return Ok(false);
456 }
457 Some(handle)
458 } else {
459 None
460 };
461
462 #[cfg(not(windows))]
463 if let Some(expected) = expected_start_time
464 && !self.start_time_matches(pid, expected)
465 {
466 debug!("process {pid} identity changed before termination");
467 return Ok(false);
468 }
469
470 self.kill(pid, 0, None)
471 }
472
473 pub async fn kill_async(
474 &self,
475 pid: u32,
476 stop_signal: i32,
477 stop_timeout: Option<std::time::Duration>,
478 ) -> Result<bool> {
479 tokio::task::spawn_blocking(move || PROCS.kill(pid, stop_signal, stop_timeout))
480 .await
481 .into_diagnostic()?
482 }
483
484 fn kill(
494 &self,
495 pid: u32,
496 stop_signal: i32,
497 stop_timeout: Option<std::time::Duration>,
498 ) -> Result<bool> {
499 let sysinfo_pid = sysinfo::Pid::from_u32(pid);
500
501 debug!("killing process {pid}");
502
503 #[cfg(windows)]
504 {
505 let _ = (stop_signal, stop_timeout);
506 let output = std::process::Command::new("taskkill")
511 .args(["/F", "/T", "/PID"])
512 .arg(pid.to_string())
513 .creation_flags(0x08000000) .output();
515 let taskkill_succeeded = match output {
516 Ok(o) if o.status.success() => {
517 debug!("taskkill /F /T /PID {pid} succeeded");
518 true
519 }
520 Ok(o) => {
521 debug!(
522 "taskkill /F /T /PID {pid} exited with status {}: {}",
523 o.status,
524 String::from_utf8_lossy(&o.stderr).trim()
525 );
526 false
527 }
528 Err(e) => {
529 debug!("failed to spawn taskkill for pid {pid}: {e}");
530 false
531 }
532 };
533 std::thread::sleep(std::time::Duration::from_millis(200));
537 if !taskkill_succeeded && self.is_running(pid) {
538 return Err(miette::miette!(
539 "taskkill failed and process {pid} is still running"
540 ));
541 }
542 Ok(true)
543 }
544
545 #[cfg(unix)]
546 {
547 let signal_name = signal_name(stop_signal);
548 debug!("sending {signal_name} to process {pid}");
552 let ret = unsafe { libc::kill(pid as i32, stop_signal) };
553 if ret == -1 {
554 let err = std::io::Error::last_os_error();
555 if err.raw_os_error() == Some(libc::ESRCH) {
556 debug!("process {pid} no longer exists");
557 return Ok(false);
558 }
559 if err.raw_os_error() == Some(libc::EPERM) {
560 return Err(miette::miette!(
561 "failed to send {signal_name} to process {pid}: permission denied"
562 ));
563 }
564 return Err(miette::miette!(
565 "failed to send {signal_name} to process {pid}: {err}"
566 ));
567 }
568
569 let stop_timeout = stop_timeout.unwrap_or_else(|| settings().supervisor_stop_timeout());
572 let fast_ms = 10u64;
573 let slow_ms = 50u64;
574 let total_ms = stop_timeout.as_millis().max(1) as u64;
575 let fast_count = ((total_ms / fast_ms) as usize).min(10);
576 let fast_total_ms = fast_ms * fast_count as u64;
577 let remaining_ms = total_ms.saturating_sub(fast_total_ms);
578 let slow_count = (remaining_ms / slow_ms) as usize;
579
580 for i in 0..fast_count {
581 std::thread::sleep(std::time::Duration::from_millis(fast_ms));
582 self.refresh_pids(&[pid]);
583 if self.is_terminated_or_zombie(sysinfo_pid) {
584 debug!(
585 "process {pid} terminated after {signal_name} ({} ms)",
586 (i + 1) * fast_ms as usize
587 );
588 return Ok(true);
589 }
590 }
591
592 for i in 0..slow_count {
594 std::thread::sleep(std::time::Duration::from_millis(slow_ms));
595 self.refresh_pids(&[pid]);
596 if self.is_terminated_or_zombie(sysinfo_pid) {
597 debug!(
598 "process {pid} terminated after {signal_name} ({} ms)",
599 fast_total_ms + (i + 1) as u64 * slow_ms
600 );
601 return Ok(true);
602 }
603 }
604
605 warn!(
607 "process {pid} did not respond to {signal_name} after {}ms, sending SIGKILL",
608 stop_timeout.as_millis()
609 );
610 let ret = unsafe { libc::kill(pid as i32, libc::SIGKILL) };
611 if ret == -1 {
612 let err = std::io::Error::last_os_error();
613 if err.raw_os_error() != Some(libc::ESRCH) {
614 warn!("failed to send SIGKILL to process {pid}: {err}");
615 }
616 }
617
618 std::thread::sleep(std::time::Duration::from_millis(100));
620 Ok(true)
621 }
622 }
623
624 #[cfg(unix)]
628 fn is_terminated_or_zombie(&self, sysinfo_pid: sysinfo::Pid) -> bool {
629 let system = self.lock_system();
630 match system.process(sysinfo_pid) {
631 None => true,
632 Some(process) => {
633 matches!(process.status(), sysinfo::ProcessStatus::Zombie)
634 }
635 }
636 }
637
638 pub(crate) fn refresh_processes(&self) {
639 let mut system = self.lock_system();
640 system.refresh_processes(ProcessesToUpdate::All, true);
641 #[cfg(windows)]
646 system.refresh_cpu_usage();
647 }
648
649 pub(crate) fn refresh_pids(&self, pids: &[u32]) {
652 let sysinfo_pids: Vec<sysinfo::Pid> =
653 pids.iter().map(|p| sysinfo::Pid::from_u32(*p)).collect();
654 self.lock_system()
655 .refresh_processes(ProcessesToUpdate::Some(&sysinfo_pids), true);
656 }
657
658 pub(crate) fn refresh_if_stale(&self) {
675 let mut last = self
676 .last_full_refresh
677 .lock()
678 .unwrap_or_else(|poisoned| poisoned.into_inner());
679 if last.is_none_or(|t| t.elapsed() >= FULL_REFRESH_INTERVAL) {
680 self.refresh_processes();
681 *last = Some(Instant::now());
682 }
683 }
684
685 pub fn get_batch_group_stats(&self, pids: &[u32]) -> Vec<(u32, Option<ProcessStats>)> {
691 if pids.is_empty() {
692 return Vec::new();
693 }
694
695 let system = self.lock_system();
696 let processes = system.processes();
697
698 let now = std::time::SystemTime::now()
699 .duration_since(std::time::UNIX_EPOCH)
700 .map(|d| d.as_secs())
701 .unwrap_or(0);
702
703 let mut children_map: std::collections::HashMap<sysinfo::Pid, Vec<sysinfo::Pid>> =
705 std::collections::HashMap::new();
706 for (child_pid, child) in processes {
707 if child.thread_kind().is_some() {
710 continue;
711 }
712 if let Some(ppid) = child.parent() {
713 children_map.entry(ppid).or_default().push(*child_pid);
714 }
715 }
716
717 pids.iter()
718 .map(|&pid| {
719 let root_pid = sysinfo::Pid::from_u32(pid);
720 let Some(root) = processes.get(&root_pid) else {
721 return (pid, None);
722 };
723
724 let root_disk = root.disk_usage();
725 let mut stats = ProcessStats {
726 cpu_percent: root.cpu_usage(),
727 memory_bytes: root.memory(),
728 uptime_secs: now.saturating_sub(root.start_time()),
729 disk_read_bytes: root_disk.read_bytes,
730 disk_write_bytes: root_disk.written_bytes,
731 };
732
733 let mut queue = std::collections::VecDeque::new();
735 if let Some(direct_children) = children_map.get(&root_pid) {
736 queue.extend(direct_children);
737 }
738 while let Some(child_pid) = queue.pop_front() {
739 if let Some(child) = processes.get(&child_pid) {
740 let disk = child.disk_usage();
741 stats.cpu_percent += child.cpu_usage();
742 stats.memory_bytes += child.memory();
743 stats.disk_read_bytes += disk.read_bytes;
744 stats.disk_write_bytes += disk.written_bytes;
745 }
746 if let Some(grandchildren) = children_map.get(&child_pid) {
747 queue.extend(grandchildren);
748 }
749 }
750
751 (pid, Some(stats))
752 })
753 .collect()
754 }
755 pub fn refresh_and_get_batch_stats(&self, pids: &[u32]) -> HashMap<u32, ProcessStats> {
765 self.refresh_processes();
766 self.get_batch_group_stats(pids)
767 .into_iter()
768 .filter_map(|(pid, stats)| stats.map(|s| (pid, s)))
769 .collect()
770 }
771
772 pub fn refresh_and_get_batch_stats_if_stale(&self, pids: &[u32]) -> HashMap<u32, ProcessStats> {
779 self.refresh_if_stale();
780 self.get_batch_group_stats(pids)
781 .into_iter()
782 .filter_map(|(pid, stats)| stats.map(|s| (pid, s)))
783 .collect()
784 }
785
786 pub fn get_batch_tree_stats_map(&self, pids: &[u32]) -> HashMap<u32, ProcessStats> {
788 self.get_batch_group_stats(pids)
789 .into_iter()
790 .filter_map(|(pid, stats)| stats.map(|stats| (pid, stats)))
791 .collect()
792 }
793
794 pub fn get_stats(&self, pid: u32) -> Option<ProcessStats> {
796 self.get_batch_group_stats(&[pid])
797 .into_iter()
798 .next()
799 .and_then(|(_, stats)| stats)
800 }
801
802 pub fn get_extended_stats(&self, pid: u32) -> Option<ExtendedProcessStats> {
804 let system = self.lock_system();
805 let processes = system.processes();
806 let root_pid = sysinfo::Pid::from_u32(pid);
807 let p = processes.get(&root_pid)?;
808
809 let now = std::time::SystemTime::now()
810 .duration_since(std::time::UNIX_EPOCH)
811 .map(|d| d.as_secs())
812 .unwrap_or(0);
813
814 let root_disk = p.disk_usage();
815 let mut aggregate_stats = ProcessStats {
816 cpu_percent: p.cpu_usage(),
817 memory_bytes: p.memory(),
818 uptime_secs: now.saturating_sub(p.start_time()),
819 disk_read_bytes: root_disk.read_bytes,
820 disk_write_bytes: root_disk.written_bytes,
821 };
822
823 let mut children_map: HashMap<sysinfo::Pid, Vec<sysinfo::Pid>> = HashMap::new();
824 for (child_pid, child) in processes {
825 if let Some(ppid) = child.parent() {
826 children_map.entry(ppid).or_default().push(*child_pid);
827 }
828 }
829
830 let mut queue = std::collections::VecDeque::new();
831 if let Some(direct_children) = children_map.get(&root_pid) {
832 queue.extend(direct_children);
833 }
834 while let Some(child_pid) = queue.pop_front() {
835 if let Some(child) = processes.get(&child_pid) {
836 let disk = child.disk_usage();
837 aggregate_stats.cpu_percent += child.cpu_usage();
838 aggregate_stats.memory_bytes += child.memory();
839 aggregate_stats.disk_read_bytes += disk.read_bytes;
840 aggregate_stats.disk_write_bytes += disk.written_bytes;
841 }
842 if let Some(grandchildren) = children_map.get(&child_pid) {
843 queue.extend(grandchildren);
844 }
845 }
846
847 Some(ExtendedProcessStats {
848 name: p.name().to_string_lossy().to_string(),
849 status: format!("{:?}", p.status()),
850 cpu_percent: aggregate_stats.cpu_percent,
851 memory_bytes: aggregate_stats.memory_bytes,
852 virtual_memory_bytes: p.virtual_memory(),
853 uptime_secs: aggregate_stats.uptime_secs,
854 thread_count: p.tasks().map(|t| t.len()).unwrap_or(0),
855 })
856 }
857}
858
859#[cfg(target_os = "linux")]
860fn process_start_token(pid: u32) -> Option<u64> {
861 let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
862 let command_end = stat.rfind(')')?;
863 stat.get(command_end + 1..)?
865 .split_whitespace()
866 .nth(19)?
867 .parse()
868 .ok()
869}
870
871#[cfg(target_os = "macos")]
872fn process_start_token(pid: u32) -> Option<u64> {
873 let mut info = std::mem::MaybeUninit::<libc::proc_bsdinfo>::zeroed();
874 let size = std::mem::size_of::<libc::proc_bsdinfo>() as i32;
875 let read = unsafe {
876 libc::proc_pidinfo(
877 pid as i32,
878 libc::PROC_PIDTBSDINFO,
879 0,
880 info.as_mut_ptr().cast(),
881 size,
882 )
883 };
884 if read != size {
885 return None;
886 }
887 let info = unsafe { info.assume_init() };
888 info.pbi_start_tvsec
889 .checked_mul(1_000_000)?
890 .checked_add(info.pbi_start_tvusec)
891}
892
893#[cfg(windows)]
894fn process_start_token(pid: u32) -> Option<u64> {
895 let handle = open_process_handle(pid).ok()?;
896 process_start_token_from_handle(handle.0)
897}
898
899#[cfg(windows)]
900fn process_start_token_from_handle(handle: HANDLE) -> Option<u64> {
901 let mut creation = FILETIME {
902 dwLowDateTime: 0,
903 dwHighDateTime: 0,
904 };
905 let mut exit = creation;
906 let mut kernel = creation;
907 let mut user = creation;
908 let ok = unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) };
909 if ok == 0 {
910 return None;
911 }
912
913 Some((u64::from(creation.dwHighDateTime) << 32) | u64::from(creation.dwLowDateTime))
914}
915
916#[cfg(windows)]
917struct ProcessHandle(HANDLE);
918
919#[cfg(windows)]
920impl Drop for ProcessHandle {
921 fn drop(&mut self) {
922 unsafe {
923 CloseHandle(self.0);
924 }
925 }
926}
927
928#[cfg(windows)]
929fn open_process_handle(pid: u32) -> std::io::Result<ProcessHandle> {
930 let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
931 if handle.is_null() {
932 return Err(std::io::Error::last_os_error());
933 }
934 Ok(ProcessHandle(handle))
935}
936
937#[cfg(not(any(target_os = "linux", target_os = "macos", windows)))]
938fn process_start_token(pid: u32) -> Option<u64> {
939 let mut system = sysinfo::System::new();
940 let sysinfo_pid = sysinfo::Pid::from_u32(pid);
941 system.refresh_processes(ProcessesToUpdate::Some(&[sysinfo_pid]), true);
942 system
943 .process(sysinfo_pid)
944 .map(|process| process.start_time())
945}
946
947#[cfg(target_os = "linux")]
948fn open_pidfd(pid: u32) -> std::io::Result<OwnedFd> {
949 let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
950 if fd < 0 {
951 return Err(std::io::Error::last_os_error());
952 }
953 Ok(unsafe { OwnedFd::from_raw_fd(fd as i32) })
954}
955
956#[cfg(target_os = "linux")]
957fn pidfd_is_running(pidfd: &OwnedFd) -> bool {
958 match try_pidfd_is_running(pidfd) {
959 Ok(running) => running,
960 Err(err) => {
961 warn!("failed to poll pidfd {}: {err}", pidfd.as_raw_fd());
962 true
963 }
964 }
965}
966
967#[cfg(target_os = "linux")]
968fn try_pidfd_is_running(pidfd: &OwnedFd) -> std::io::Result<bool> {
969 let mut pollfd = libc::pollfd {
970 fd: pidfd.as_raw_fd(),
971 events: libc::POLLIN,
972 revents: 0,
973 };
974 let result = unsafe { libc::poll(&mut pollfd, 1, 0) };
975 if result < 0 {
976 return Err(std::io::Error::last_os_error());
977 }
978 Ok(result == 0)
979}
980
981#[cfg(target_os = "linux")]
982fn signal_pidfds(members: &[(u32, OwnedFd)], signal: i32, signal_name: &str) -> Result<()> {
983 for (pid, pidfd) in members {
984 if !pidfd_is_running(pidfd) {
985 continue;
986 }
987 let result = unsafe {
988 libc::syscall(
989 libc::SYS_pidfd_send_signal,
990 pidfd.as_raw_fd(),
991 signal,
992 std::ptr::null::<libc::siginfo_t>(),
993 0,
994 )
995 };
996 if result == -1 {
997 let err = std::io::Error::last_os_error();
998 if err.raw_os_error() == Some(libc::ESRCH) {
999 continue;
1000 }
1001 return Err(miette::miette!(
1002 "failed to send {signal_name} to pinned process {pid}: {err}"
1003 ));
1004 }
1005 }
1006 Ok(())
1007}
1008
1009#[cfg(target_os = "linux")]
1010fn stop_pidfds(members: &[(u32, OwnedFd)]) -> Result<()> {
1011 signal_pidfds(members, libc::SIGSTOP, "SIGSTOP")?;
1012 for _ in 0..200 {
1013 if members.iter().all(|(pid, pidfd)| {
1014 !pidfd_is_running(pidfd) || matches!(linux_process_state(*pid), Some('T' | 't'))
1015 }) {
1016 return Ok(());
1017 }
1018 std::thread::sleep(std::time::Duration::from_millis(5));
1019 }
1020 Err(miette::miette!(
1021 "timed out while freezing orphan process group"
1022 ))
1023}
1024
1025#[cfg(target_os = "linux")]
1026fn extend_process_group_pidfds(
1027 pgid: i32,
1028 members: &mut Vec<(u32, OwnedFd)>,
1029) -> std::io::Result<usize> {
1030 let entries = std::fs::read_dir("/proc")?;
1031 let mut added = 0;
1032 for entry in entries {
1033 let entry = entry?;
1034 let Some(pid) = entry
1035 .file_name()
1036 .to_str()
1037 .and_then(|name| name.parse::<u32>().ok())
1038 else {
1039 continue;
1040 };
1041 let Some(observed_identity) = linux_process_identity(pid) else {
1042 continue;
1043 };
1044 if observed_identity.0 != pgid {
1045 continue;
1046 }
1047 let mut already_pinned = false;
1048 for (known_pid, pidfd) in members.iter() {
1049 if *known_pid == pid && try_pidfd_is_running(pidfd)? {
1050 already_pinned = true;
1051 break;
1052 }
1053 }
1054 if already_pinned {
1055 continue;
1056 }
1057
1058 let pidfd = match open_pidfd(pid) {
1059 Ok(pidfd) => pidfd,
1060 Err(err) if err.raw_os_error() == Some(libc::ESRCH) => continue,
1061 Err(err) => return Err(err),
1062 };
1063 if linux_process_identity(pid) != Some(observed_identity) {
1064 return Err(std::io::Error::other(format!(
1065 "process {pid} identity changed while pinning group {pgid}"
1066 )));
1067 }
1068 members.push((pid, pidfd));
1069 added += 1;
1070 }
1071 Ok(added)
1072}
1073
1074#[cfg(target_os = "linux")]
1075fn linux_process_identity(pid: u32) -> Option<(i32, u64)> {
1076 let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
1077 let command_end = stat.rfind(')')?;
1078 let fields: Vec<_> = stat.get(command_end + 1..)?.split_whitespace().collect();
1079 Some((fields.get(2)?.parse().ok()?, fields.get(19)?.parse().ok()?))
1082}
1083
1084#[cfg(target_os = "linux")]
1085fn linux_process_state(pid: u32) -> Option<char> {
1086 let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
1087 let command_end = stat.rfind(')')?;
1088 stat.get(command_end + 1..)?
1089 .split_whitespace()
1090 .next()?
1091 .chars()
1092 .next()
1093}
1094
1095#[derive(Debug, Clone, Copy)]
1096pub struct ProcessStats {
1097 pub cpu_percent: f32,
1098 pub memory_bytes: u64,
1099 pub uptime_secs: u64,
1100 pub disk_read_bytes: u64,
1101 pub disk_write_bytes: u64,
1102}
1103
1104impl ProcessStats {
1105 pub fn memory_display(&self) -> String {
1106 format_bytes(self.memory_bytes)
1107 }
1108
1109 pub fn cpu_display(&self) -> String {
1110 format!("{:.1}%", self.cpu_percent)
1111 }
1112
1113 pub fn uptime_display(&self) -> String {
1114 format_duration(self.uptime_secs)
1115 }
1116
1117 pub fn disk_read_display(&self) -> String {
1118 format_bytes_per_sec(self.disk_read_bytes)
1119 }
1120
1121 pub fn disk_write_display(&self) -> String {
1122 format_bytes_per_sec(self.disk_write_bytes)
1123 }
1124}
1125
1126#[derive(Debug, Clone)]
1127pub struct ExtendedProcessStats {
1128 pub name: String,
1129 pub status: String,
1130 pub cpu_percent: f32,
1131 pub memory_bytes: u64,
1132 pub virtual_memory_bytes: u64,
1133 pub uptime_secs: u64,
1134 pub thread_count: usize,
1135}
1136
1137fn format_bytes(bytes: u64) -> String {
1138 humanbyte::to_string(bytes, humanbyte::Format::IEC)
1139}
1140
1141fn format_duration(secs: u64) -> String {
1142 if secs < 60 {
1143 format!("{secs}s")
1144 } else if secs < 3600 {
1145 format!("{}m {}s", secs / 60, secs % 60)
1146 } else if secs < 86400 {
1147 let hours = secs / 3600;
1148 let mins = (secs % 3600) / 60;
1149 format!("{hours}h {mins}m")
1150 } else {
1151 let days = secs / 86400;
1152 let hours = (secs % 86400) / 3600;
1153 format!("{days}d {hours}h")
1154 }
1155}
1156
1157fn format_bytes_per_sec(bytes: u64) -> String {
1158 format!("{}/s", humanbyte::to_string(bytes, humanbyte::Format::IEC))
1159}
1160
1161#[cfg(unix)]
1177fn process_group_terminated(pgid: i32) -> bool {
1178 unsafe { libc::killpg(pgid, 0) != 0 }
1179}
1180
1181#[cfg(unix)]
1182fn signal_name(sig: i32) -> &'static str {
1183 match sig {
1184 libc::SIGHUP => "SIGHUP",
1185 libc::SIGINT => "SIGINT",
1186 libc::SIGQUIT => "SIGQUIT",
1187 libc::SIGTERM => "SIGTERM",
1188 libc::SIGUSR1 => "SIGUSR1",
1189 libc::SIGUSR2 => "SIGUSR2",
1190 libc::SIGKILL => "SIGKILL",
1191 _ => "UNKNOWN",
1192 }
1193}
1194
1195#[cfg(test)]
1196mod format_tests {
1197 use super::*;
1198
1199 #[test]
1200 fn process_start_time_check_rejects_mismatch() {
1201 let procs = Procs::new();
1202 let pid = std::process::id();
1203 procs.refresh_pids(&[pid]);
1204 let actual = procs
1205 .start_time(pid)
1206 .expect("current process should have a start time");
1207
1208 assert_ne!(procs.start_time(pid), Some(actual.saturating_add(1)));
1209 }
1210
1211 #[test]
1212 fn test_format_bytes() {
1213 assert_eq!(format_bytes(512), "512 B");
1214 assert_eq!(format_bytes(1024), "1.0 KiB");
1215 assert_eq!(format_bytes(1536), "1.5 KiB");
1216 assert_eq!(format_bytes(50 * 1024 * 1024), "50.0 MiB");
1217 assert_eq!(format_bytes(3 * 1024 * 1024 * 1024), "3.0 GiB");
1218 assert_eq!(format_bytes(1100 * 1024 * 1024 * 1024), "1.1 TiB");
1220 }
1221
1222 #[test]
1223 fn test_format_bytes_per_sec() {
1224 assert_eq!(format_bytes_per_sec(512), "512 B/s");
1225 assert_eq!(format_bytes_per_sec(1536), "1.5 KiB/s");
1226 assert_eq!(format_bytes_per_sec(2 * 1024 * 1024), "2.0 MiB/s");
1227 }
1228}
1229
1230#[cfg(all(test, unix))]
1231mod tests {
1232 use super::*;
1233 use std::os::unix::process::CommandExt;
1234 use std::process::{Child, Command, Stdio};
1235 use std::time::{Duration, Instant};
1236
1237 struct ChildGuard(Child);
1238
1239 impl Drop for ChildGuard {
1240 fn drop(&mut self) {
1241 let pid = self.0.id() as i32;
1242 let _ = unsafe { libc::killpg(pid, libc::SIGKILL) };
1244 let _ = self.0.wait();
1245 }
1246 }
1247
1248 #[tokio::test]
1249 async fn orphan_identity_checked_group_kill_rejects_mismatch() {
1250 let mut command = Command::new("sleep");
1251 command
1252 .arg("30")
1253 .stdin(Stdio::null())
1254 .stdout(Stdio::null())
1255 .stderr(Stdio::null());
1256 unsafe {
1257 command.pre_exec(|| {
1258 if libc::setsid() == -1 {
1259 return Err(std::io::Error::last_os_error());
1260 }
1261 Ok(())
1262 });
1263 }
1264
1265 let child = command.spawn().expect("failed to spawn test process");
1266 let pid = child.id();
1267 let _child = ChildGuard(child);
1268
1269 PROCS.refresh_pids(&[pid]);
1270 let actual_start_time = PROCS
1271 .start_time(pid)
1272 .expect("test process should have a start time");
1273
1274 let killed = PROCS
1275 .kill_process_group_if_start_time_matches_async(
1276 pid,
1277 actual_start_time.saturating_add(1),
1278 libc::SIGTERM,
1279 Some(Duration::from_millis(100)),
1280 )
1281 .await
1282 .expect("identity-checked kill should not error");
1283
1284 assert!(!killed);
1285 assert!(PROCS.is_running(pid), "mismatched process must survive");
1286 }
1287
1288 #[cfg(not(target_os = "linux"))]
1289 #[tokio::test]
1290 async fn orphan_identity_checked_group_kill_fails_closed_without_pidfd() {
1291 let mut command = Command::new("sleep");
1292 command
1293 .arg("30")
1294 .stdin(Stdio::null())
1295 .stdout(Stdio::null())
1296 .stderr(Stdio::null());
1297 unsafe {
1298 command.pre_exec(|| {
1299 if libc::setsid() == -1 {
1300 return Err(std::io::Error::last_os_error());
1301 }
1302 Ok(())
1303 });
1304 }
1305
1306 let child = command.spawn().expect("failed to spawn test process");
1307 let pid = child.id();
1308 let _child = ChildGuard(child);
1309
1310 PROCS.refresh_pids(&[pid]);
1311 let actual_start_time = PROCS
1312 .start_time(pid)
1313 .expect("test process should have a start time");
1314
1315 let killed = PROCS
1316 .kill_process_group_if_start_time_matches_async(
1317 pid,
1318 actual_start_time,
1319 libc::SIGTERM,
1320 Some(Duration::from_millis(100)),
1321 )
1322 .await
1323 .expect("identity-checked kill should not error");
1324
1325 assert!(!killed);
1326 assert!(
1327 PROCS.is_running(pid),
1328 "process must survive when identity cannot be pinned"
1329 );
1330 }
1331
1332 #[test]
1333 fn get_stats_includes_descendant_rss() {
1334 let mut command = Command::new("sh");
1335 command
1336 .args(["-c", "sleep 30 & wait"])
1337 .stdin(Stdio::null())
1338 .stdout(Stdio::null())
1339 .stderr(Stdio::null());
1340 unsafe {
1341 command.pre_exec(|| {
1342 if libc::setsid() == -1 {
1343 return Err(std::io::Error::last_os_error());
1344 }
1345 Ok(())
1346 });
1347 }
1348
1349 let parent = command.spawn().expect("failed to spawn process tree");
1350 let parent_pid = parent.id();
1351 let _parent = ChildGuard(parent);
1352
1353 let procs = Procs::new();
1354 let deadline = Instant::now() + Duration::from_secs(5);
1355 let mut child_pids = Vec::new();
1356 while Instant::now() < deadline {
1357 procs.refresh_processes();
1358 child_pids = procs.all_children(parent_pid);
1359 if !child_pids.is_empty() {
1360 break;
1361 }
1362 std::thread::sleep(Duration::from_millis(50));
1363 }
1364 assert!(
1365 !child_pids.is_empty(),
1366 "test process tree did not appear under parent pid {parent_pid}"
1367 );
1368
1369 procs.refresh_processes();
1370 child_pids = procs.all_children(parent_pid);
1371 assert!(
1372 !child_pids.is_empty(),
1373 "test process tree disappeared under parent pid {parent_pid}"
1374 );
1375 let root_pid = sysinfo::Pid::from_u32(parent_pid);
1376 let direct_memory = {
1377 let system = procs.lock_system();
1378 system
1379 .process(root_pid)
1380 .expect("parent process should exist")
1381 .memory()
1382 };
1383 let descendant_memory = {
1384 let system = procs.lock_system();
1385 child_pids
1386 .iter()
1387 .filter_map(|pid| system.process(sysinfo::Pid::from_u32(*pid)))
1388 .map(|process| process.memory())
1389 .sum::<u64>()
1390 };
1391 assert!(
1392 descendant_memory > 0,
1393 "descendants {child_pids:?} should have nonzero RSS"
1394 );
1395
1396 let stats = procs
1397 .get_stats(parent_pid)
1398 .expect("parent process should have aggregate stats");
1399
1400 assert_eq!(
1401 stats.memory_bytes,
1402 direct_memory + descendant_memory,
1403 "get_stats should include descendant RSS for parent pid {parent_pid}; \
1404 descendants: {child_pids:?}, direct RSS: {direct_memory}, \
1405 descendant RSS: {descendant_memory}, reported RSS: {}",
1406 stats.memory_bytes
1407 );
1408 }
1409
1410 #[test]
1411 fn full_refresh_is_throttled_by_ttl() {
1412 let procs = Procs::new();
1413
1414 assert!(
1416 procs.last_full_refresh.lock().unwrap().is_none(),
1417 "fresh Procs should have no recorded refresh"
1418 );
1419 procs.refresh_if_stale();
1420 let first = procs.last_full_refresh.lock().unwrap().unwrap();
1421 assert!(first.elapsed() < FULL_REFRESH_INTERVAL);
1422
1423 procs.refresh_if_stale();
1426 let second = procs.last_full_refresh.lock().unwrap().unwrap();
1427 assert_eq!(
1428 second, first,
1429 "refresh_if_stale within TTL must skip the refresh and keep the timestamp"
1430 );
1431
1432 let expired = Instant::now()
1435 .checked_sub(FULL_REFRESH_INTERVAL + Duration::from_secs(1))
1436 .expect("system has been up long enough to backdate by 6s");
1437 *procs.last_full_refresh.lock().unwrap() = Some(expired);
1438 procs.refresh_if_stale();
1439 let third = procs.last_full_refresh.lock().unwrap().unwrap();
1440 assert!(
1441 third > expired,
1442 "refresh_if_stale after expired TTL must refresh and advance the timestamp"
1443 );
1444 assert!(
1445 third.elapsed() < FULL_REFRESH_INTERVAL,
1446 "fresh timestamp after expired-TTL refresh should be recent"
1447 );
1448 }
1449}