running_process_core/
lib.rs

1use std::collections::VecDeque;
2use std::fs::OpenOptions;
3use std::io::Read;
4use std::io::Write;
5use std::path::PathBuf;
6use std::process::{Child, Command, Stdio};
7use std::sync::atomic::{AtomicI64, Ordering};
8use std::sync::{Arc, Condvar, Mutex};
9use std::thread;
10use std::time::{Duration, Instant};
11
12use thiserror::Error;
13
14pub mod console_detect;
15pub mod containment;
16#[cfg(feature = "originator-scan")]
17pub mod originator;
18pub mod pty;
19mod public_symbols;
20mod rust_debug;
21pub mod spawn;
22
23pub use console_detect::{monitor_console_windows, ConsoleWindowInfo};
24pub use containment::{ContainedProcessGroup, ORIGINATOR_ENV_VAR};
25#[cfg(feature = "originator-scan")]
26pub use originator::{find_processes_by_originator, OriginatorProcessInfo};
27pub use rust_debug::{render_rust_debug_traces, RustDebugScopeGuard};
28pub use spawn::{
29    spawn, spawn_daemon, spawn_daemon_with_clear_env, DaemonChild, SpawnStdio, SpawnedChild,
30    StdioSource,
31};
32
33#[macro_export]
34macro_rules! rp_rust_debug_scope {
35    ($label:expr) => {
36        let _running_process_rust_debug_scope =
37            $crate::RustDebugScopeGuard::enter($label, file!(), line!());
38    };
39}
40
41const CHILD_PID_LOG_PATH_ENV: &str = "RUNNING_PROCESS_CHILD_PID_LOG_PATH";
42
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum StreamKind {
45    Stdout,
46    Stderr,
47}
48
49impl StreamKind {
50    pub fn as_str(self) -> &'static str {
51        match self {
52            Self::Stdout => "stdout",
53            Self::Stderr => "stderr",
54        }
55    }
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct StreamEvent {
60    pub stream: StreamKind,
61    pub line: Vec<u8>,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub enum ReadStatus<T> {
66    Line(T),
67    Timeout,
68    Eof,
69}
70
71#[derive(Debug, Error)]
72pub enum ProcessError {
73    #[error("process already started")]
74    AlreadyStarted,
75    #[error("process is not running")]
76    NotRunning,
77    #[error("process stdin is not available")]
78    StdinUnavailable,
79    #[error("failed to spawn process: {0}")]
80    Spawn(std::io::Error),
81    #[error("failed to read process output: {0}")]
82    Io(std::io::Error),
83    #[error("process timed out")]
84    Timeout,
85}
86
87#[derive(Debug, Clone)]
88pub enum CommandSpec {
89    Shell(String),
90    Argv(Vec<String>),
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
94pub enum StdinMode {
95    Inherit,
96    Piped,
97    Null,
98}
99
100#[derive(Debug, Clone, Copy, PartialEq, Eq)]
101pub enum StderrMode {
102    Stdout,
103    Pipe,
104}
105
106#[cfg(unix)]
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
108pub enum UnixSignal {
109    Interrupt,
110    Terminate,
111    Kill,
112}
113
114#[derive(Debug, Clone)]
115pub struct ProcessConfig {
116    pub command: CommandSpec,
117    pub cwd: Option<PathBuf>,
118    pub env: Option<Vec<(String, String)>>,
119    pub capture: bool,
120    pub stderr_mode: StderrMode,
121    pub creationflags: Option<u32>,
122    pub create_process_group: bool,
123    pub stdin_mode: StdinMode,
124    pub nice: Option<i32>,
125}
126
127#[derive(Default)]
128struct QueueState {
129    stdout_queue: VecDeque<Vec<u8>>,
130    stderr_queue: VecDeque<Vec<u8>>,
131    combined_queue: VecDeque<StreamEvent>,
132    stdout_history: VecDeque<Vec<u8>>,
133    stderr_history: VecDeque<Vec<u8>>,
134    combined_history: VecDeque<StreamEvent>,
135    stdout_history_bytes: usize,
136    stderr_history_bytes: usize,
137    combined_history_bytes: usize,
138    stdout_closed: bool,
139    stderr_closed: bool,
140}
141
142/// Sentinel value for returncode atomic: process has not exited yet.
143const RETURNCODE_NOT_SET: i64 = i64::MIN;
144
145struct SharedState {
146    queues: Mutex<QueueState>,
147    condvar: Condvar,
148    /// Atomic exit code. `RETURNCODE_NOT_SET` means "not exited yet".
149    /// Updated by a background waiter thread — reading is lock-free.
150    returncode: AtomicI64,
151}
152
153#[cfg(windows)]
154struct WindowsJobHandle(usize);
155
156#[cfg(windows)]
157impl Drop for WindowsJobHandle {
158    fn drop(&mut self) {
159        unsafe {
160            winapi::um::handleapi::CloseHandle(self.0 as winapi::shared::ntdef::HANDLE);
161        }
162    }
163}
164
165struct ChildState {
166    child: Child,
167    #[cfg(windows)]
168    _job: WindowsJobHandle,
169}
170
171impl SharedState {
172    fn new(capture: bool) -> Self {
173        let queues = QueueState {
174            stdout_closed: !capture,
175            stderr_closed: !capture,
176            ..QueueState::default()
177        };
178        Self {
179            queues: Mutex::new(queues),
180            condvar: Condvar::new(),
181            returncode: AtomicI64::new(RETURNCODE_NOT_SET),
182        }
183    }
184}
185
186pub struct NativeProcess {
187    config: ProcessConfig,
188    child: Arc<Mutex<Option<ChildState>>>,
189    shared: Arc<SharedState>,
190}
191
192impl NativeProcess {
193    pub fn new(config: ProcessConfig) -> Self {
194        Self {
195            shared: Arc::new(SharedState::new(config.capture)),
196            child: Arc::new(Mutex::new(None)),
197            config,
198        }
199    }
200
201    // Preserve a stable Rust frame here in release user dumps.
202    #[inline(never)]
203    pub fn start(&self) -> Result<(), ProcessError> {
204        public_symbols::rp_native_process_start_public(self)
205    }
206
207    fn start_impl(&self) -> Result<(), ProcessError> {
208        crate::rp_rust_debug_scope!("running_process_core::NativeProcess::start");
209        let mut guard = self.child.lock().expect("child mutex poisoned");
210        if guard.is_some() {
211            return Err(ProcessError::AlreadyStarted);
212        }
213
214        let mut command = self.build_command();
215        match self.config.stdin_mode {
216            StdinMode::Inherit => {}
217            StdinMode::Piped => {
218                command.stdin(Stdio::piped());
219            }
220            StdinMode::Null => {
221                command.stdin(Stdio::null());
222            }
223        }
224        if self.config.capture {
225            command.stdout(Stdio::piped());
226            command.stderr(Stdio::piped());
227        }
228
229        let mut child = command.spawn().map_err(ProcessError::Spawn)?;
230        log_spawned_child_pid(child.id()).map_err(ProcessError::Spawn)?;
231        #[cfg(windows)]
232        let job = public_symbols::rp_assign_child_to_windows_kill_on_close_job_public(&child)
233            .map_err(ProcessError::Spawn)?;
234        if self.config.capture {
235            let stdout = child.stdout.take().expect("stdout pipe missing");
236            let stderr = child.stderr.take().expect("stderr pipe missing");
237            self.spawn_reader(stdout, StreamKind::Stdout, StreamKind::Stdout);
238            self.spawn_reader(
239                stderr,
240                StreamKind::Stderr,
241                match self.config.stderr_mode {
242                    StderrMode::Stdout => StreamKind::Stdout,
243                    StderrMode::Pipe => StreamKind::Stderr,
244                },
245            );
246        }
247        *guard = Some(ChildState {
248            child,
249            #[cfg(windows)]
250            _job: job,
251        });
252        drop(guard);
253        self.spawn_exit_waiter();
254        Ok(())
255    }
256
257    /// Background thread that polls for process exit and stores the exit code
258    /// atomically. This makes `returncode` auto-update without explicit `poll()`.
259    fn spawn_exit_waiter(&self) {
260        let child = Arc::clone(&self.child);
261        let shared = Arc::clone(&self.shared);
262        thread::spawn(move || loop {
263            if shared.returncode.load(Ordering::Acquire) != RETURNCODE_NOT_SET {
264                return;
265            }
266            {
267                let mut guard = child.lock().expect("child mutex poisoned");
268                if let Some(child_state) = guard.as_mut() {
269                    match child_state.child.try_wait() {
270                        Ok(Some(status)) => {
271                            let code = exit_code(status);
272                            shared.returncode.store(code as i64, Ordering::Release);
273                            shared.condvar.notify_all();
274                            return;
275                        }
276                        Ok(None) => {}
277                        Err(_) => return,
278                    }
279                } else {
280                    return;
281                }
282            }
283            thread::sleep(Duration::from_millis(10));
284        });
285    }
286
287    pub fn write_stdin(&self, data: &[u8]) -> Result<(), ProcessError> {
288        let mut guard = self.child.lock().expect("child mutex poisoned");
289        let child = &mut guard.as_mut().ok_or(ProcessError::NotRunning)?.child;
290        let stdin = child.stdin.as_mut().ok_or(ProcessError::StdinUnavailable)?;
291        use std::io::Write;
292        stdin.write_all(data).map_err(ProcessError::Io)?;
293        stdin.flush().map_err(ProcessError::Io)?;
294        drop(child.stdin.take());
295        Ok(())
296    }
297
298    pub fn poll(&self) -> Result<Option<i32>, ProcessError> {
299        // Fast path: check atomic set by background waiter thread.
300        if let Some(code) = self.returncode() {
301            return Ok(Some(code));
302        }
303        let mut guard = self.child.lock().expect("child mutex poisoned");
304        let Some(child_state) = guard.as_mut() else {
305            return Ok(self.returncode());
306        };
307        let child = &mut child_state.child;
308        let status = child.try_wait().map_err(ProcessError::Io)?;
309        if let Some(status) = status {
310            let code = exit_code(status);
311            self.set_returncode(code);
312            return Ok(Some(code));
313        }
314        Ok(None)
315    }
316
317    // Preserve a stable Rust frame here in release user dumps.
318    #[inline(never)]
319    pub fn wait(&self, timeout: Option<Duration>) -> Result<i32, ProcessError> {
320        public_symbols::rp_native_process_wait_public(self, timeout)
321    }
322
323    fn wait_impl(&self, timeout: Option<Duration>) -> Result<i32, ProcessError> {
324        crate::rp_rust_debug_scope!("running_process_core::NativeProcess::wait");
325        if self.child.lock().expect("child mutex poisoned").is_none() {
326            return self.returncode().ok_or(ProcessError::NotRunning);
327        }
328        // Fast path: already exited.
329        if let Some(code) = self.returncode() {
330            public_symbols::rp_native_process_wait_for_capture_completion_public(self);
331            return Ok(code);
332        }
333        let start = Instant::now();
334        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
335        loop {
336            // Check returncode (set by exit-waiter thread via atomic + condvar).
337            let rc = self.shared.returncode.load(Ordering::Acquire);
338            if rc != RETURNCODE_NOT_SET {
339                drop(guard);
340                let code = rc as i32;
341                public_symbols::rp_native_process_wait_for_capture_completion_public(self);
342                return Ok(code);
343            }
344            if let Some(limit) = timeout {
345                let elapsed = start.elapsed();
346                if elapsed >= limit {
347                    return Err(ProcessError::Timeout);
348                }
349                let remaining = limit - elapsed;
350                // Wait on condvar with timeout, capped at 50ms to recheck.
351                let wait_time = remaining.min(Duration::from_millis(50));
352                guard = self
353                    .shared
354                    .condvar
355                    .wait_timeout(guard, wait_time)
356                    .expect("queue mutex poisoned")
357                    .0;
358            } else {
359                // Wait on condvar with periodic recheck.
360                guard = self
361                    .shared
362                    .condvar
363                    .wait_timeout(guard, Duration::from_millis(50))
364                    .expect("queue mutex poisoned")
365                    .0;
366            }
367        }
368    }
369
370    // Preserve a stable Rust frame here in release user dumps.
371    #[inline(never)]
372    pub fn kill(&self) -> Result<(), ProcessError> {
373        public_symbols::rp_native_process_kill_public(self)
374    }
375
376    fn kill_impl(&self) -> Result<(), ProcessError> {
377        crate::rp_rust_debug_scope!("running_process_core::NativeProcess::kill");
378        let mut guard = self.child.lock().expect("child mutex poisoned");
379        let child = &mut guard.as_mut().ok_or(ProcessError::NotRunning)?.child;
380        child.kill().map_err(ProcessError::Io)?;
381        let status = child.wait().map_err(ProcessError::Io)?;
382        self.set_returncode(exit_code(status));
383        Ok(())
384    }
385
386    pub fn terminate(&self) -> Result<(), ProcessError> {
387        self.kill()
388    }
389
390    // Preserve a stable Rust frame here in release user dumps.
391    #[inline(never)]
392    pub fn close(&self) -> Result<(), ProcessError> {
393        public_symbols::rp_native_process_close_public(self)
394    }
395
396    fn close_impl(&self) -> Result<(), ProcessError> {
397        crate::rp_rust_debug_scope!("running_process_core::NativeProcess::close");
398        if self.child.lock().expect("child mutex poisoned").is_none() {
399            return Ok(());
400        }
401        if self.poll()?.is_none() {
402            self.kill()?;
403        } else {
404            public_symbols::rp_native_process_wait_for_capture_completion_public(self);
405        }
406        Ok(())
407    }
408
409    pub fn pid(&self) -> Option<u32> {
410        self.child
411            .lock()
412            .expect("child mutex poisoned")
413            .as_ref()
414            .map(|state| state.child.id())
415    }
416
417    pub fn returncode(&self) -> Option<i32> {
418        let v = self.shared.returncode.load(Ordering::Acquire);
419        if v == RETURNCODE_NOT_SET {
420            None
421        } else {
422            Some(v as i32)
423        }
424    }
425
426    pub fn has_pending_stream(&self, stream: StreamKind) -> bool {
427        if stream == StreamKind::Stderr && self.config.stderr_mode == StderrMode::Stdout {
428            return false;
429        }
430        let guard = self.shared.queues.lock().expect("queue mutex poisoned");
431        match stream {
432            StreamKind::Stdout => !guard.stdout_queue.is_empty(),
433            StreamKind::Stderr => !guard.stderr_queue.is_empty(),
434        }
435    }
436
437    pub fn has_pending_combined(&self) -> bool {
438        let guard = self.shared.queues.lock().expect("queue mutex poisoned");
439        !guard.combined_queue.is_empty()
440    }
441
442    pub fn drain_stream(&self, stream: StreamKind) -> Vec<Vec<u8>> {
443        if stream == StreamKind::Stderr && self.config.stderr_mode == StderrMode::Stdout {
444            return Vec::new();
445        }
446        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
447        let queue = match stream {
448            StreamKind::Stdout => &mut guard.stdout_queue,
449            StreamKind::Stderr => &mut guard.stderr_queue,
450        };
451        queue.drain(..).collect()
452    }
453
454    pub fn drain_combined(&self) -> Vec<StreamEvent> {
455        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
456        guard.combined_queue.drain(..).collect()
457    }
458
459    pub fn read_stream(
460        &self,
461        stream: StreamKind,
462        timeout: Option<Duration>,
463    ) -> ReadStatus<Vec<u8>> {
464        let deadline = timeout.map(|limit| Instant::now() + limit);
465        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
466
467        loop {
468            if stream == StreamKind::Stderr && self.config.stderr_mode == StderrMode::Stdout {
469                return ReadStatus::Eof;
470            }
471
472            let queue = match stream {
473                StreamKind::Stdout => &mut guard.stdout_queue,
474                StreamKind::Stderr => &mut guard.stderr_queue,
475            };
476            if let Some(line) = queue.pop_front() {
477                return ReadStatus::Line(line);
478            }
479
480            let closed = match stream {
481                StreamKind::Stdout => {
482                    if self.config.stderr_mode == StderrMode::Stdout {
483                        guard.stdout_closed && guard.stderr_closed
484                    } else {
485                        guard.stdout_closed
486                    }
487                }
488                StreamKind::Stderr => guard.stderr_closed,
489            };
490            if closed {
491                return ReadStatus::Eof;
492            }
493
494            match deadline {
495                Some(deadline) => {
496                    let now = Instant::now();
497                    if now >= deadline {
498                        return ReadStatus::Timeout;
499                    }
500                    let wait = deadline.saturating_duration_since(now);
501                    let result = self
502                        .shared
503                        .condvar
504                        .wait_timeout(guard, wait)
505                        .expect("queue mutex poisoned");
506                    guard = result.0;
507                    if result.1.timed_out() {
508                        return ReadStatus::Timeout;
509                    }
510                }
511                None => {
512                    guard = self
513                        .shared
514                        .condvar
515                        .wait(guard)
516                        .expect("queue mutex poisoned");
517                }
518            }
519        }
520    }
521
522    // Preserve a stable Rust frame here in release user dumps.
523    #[inline(never)]
524    pub fn read_combined(&self, timeout: Option<Duration>) -> ReadStatus<StreamEvent> {
525        public_symbols::rp_native_process_read_combined_public(self, timeout)
526    }
527
528    fn read_combined_impl(&self, timeout: Option<Duration>) -> ReadStatus<StreamEvent> {
529        crate::rp_rust_debug_scope!("running_process_core::NativeProcess::read_combined");
530        let deadline = timeout.map(|limit| Instant::now() + limit);
531        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
532
533        loop {
534            if let Some(event) = guard.combined_queue.pop_front() {
535                return ReadStatus::Line(event);
536            }
537            if guard.stdout_closed && guard.stderr_closed {
538                return ReadStatus::Eof;
539            }
540
541            match deadline {
542                Some(deadline) => {
543                    let now = Instant::now();
544                    if now >= deadline {
545                        return ReadStatus::Timeout;
546                    }
547                    let wait = deadline.saturating_duration_since(now);
548                    let result = self
549                        .shared
550                        .condvar
551                        .wait_timeout(guard, wait)
552                        .expect("queue mutex poisoned");
553                    guard = result.0;
554                    if result.1.timed_out() {
555                        return ReadStatus::Timeout;
556                    }
557                }
558                None => {
559                    guard = self
560                        .shared
561                        .condvar
562                        .wait(guard)
563                        .expect("queue mutex poisoned");
564                }
565            }
566        }
567    }
568
569    pub fn captured_stdout(&self) -> Vec<Vec<u8>> {
570        self.shared
571            .queues
572            .lock()
573            .expect("queue mutex poisoned")
574            .stdout_history
575            .clone()
576            .into_iter()
577            .collect()
578    }
579
580    pub fn captured_stderr(&self) -> Vec<Vec<u8>> {
581        if self.config.stderr_mode == StderrMode::Stdout {
582            return Vec::new();
583        }
584        self.shared
585            .queues
586            .lock()
587            .expect("queue mutex poisoned")
588            .stderr_history
589            .clone()
590            .into_iter()
591            .collect()
592    }
593
594    pub fn captured_combined(&self) -> Vec<StreamEvent> {
595        self.shared
596            .queues
597            .lock()
598            .expect("queue mutex poisoned")
599            .combined_history
600            .clone()
601            .into_iter()
602            .collect()
603    }
604
605    pub fn captured_stream_bytes(&self, stream: StreamKind) -> usize {
606        if stream == StreamKind::Stderr && self.config.stderr_mode == StderrMode::Stdout {
607            return 0;
608        }
609        let guard = self.shared.queues.lock().expect("queue mutex poisoned");
610        match stream {
611            StreamKind::Stdout => guard.stdout_history_bytes,
612            StreamKind::Stderr => guard.stderr_history_bytes,
613        }
614    }
615
616    pub fn captured_combined_bytes(&self) -> usize {
617        self.shared
618            .queues
619            .lock()
620            .expect("queue mutex poisoned")
621            .combined_history_bytes
622    }
623
624    pub fn clear_captured_stream(&self, stream: StreamKind) -> usize {
625        if stream == StreamKind::Stderr && self.config.stderr_mode == StderrMode::Stdout {
626            return 0;
627        }
628        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
629        match stream {
630            StreamKind::Stdout => {
631                let released = guard.stdout_history_bytes;
632                guard.stdout_history.clear();
633                guard.stdout_history_bytes = 0;
634                released
635            }
636            StreamKind::Stderr => {
637                let released = guard.stderr_history_bytes;
638                guard.stderr_history.clear();
639                guard.stderr_history_bytes = 0;
640                released
641            }
642        }
643    }
644
645    pub fn clear_captured_combined(&self) -> usize {
646        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
647        let released = guard.combined_history_bytes;
648        guard.combined_history.clear();
649        guard.combined_history_bytes = 0;
650        released
651    }
652
653    fn build_command(&self) -> Command {
654        let mut command = match &self.config.command {
655            CommandSpec::Shell(command) => shell_command(command),
656            CommandSpec::Argv(argv) => {
657                let mut command = Command::new(&argv[0]);
658                if argv.len() > 1 {
659                    command.args(&argv[1..]);
660                }
661                command
662            }
663        };
664        if let Some(cwd) = &self.config.cwd {
665            command.current_dir(cwd);
666        }
667        if let Some(env) = &self.config.env {
668            command.env_clear();
669            command.envs(env.iter().map(|(k, v)| (k, v)));
670        }
671        #[cfg(windows)]
672        {
673            use std::os::windows::process::CommandExt;
674
675            let flags =
676                self.config.creationflags.unwrap_or(0) | windows_priority_flags(self.config.nice);
677            if flags != 0 {
678                command.creation_flags(flags);
679            }
680        }
681        #[cfg(unix)]
682        {
683            let create_process_group = self.config.create_process_group;
684            let nice = self.config.nice;
685
686            if create_process_group || nice.is_some() {
687                use std::os::unix::process::CommandExt;
688
689                unsafe {
690                    command.pre_exec(move || {
691                        if create_process_group && libc::setpgid(0, 0) == -1 {
692                            return Err(std::io::Error::last_os_error());
693                        }
694                        if let Some(nice) = nice {
695                            let result = libc::setpriority(libc::PRIO_PROCESS, 0, nice);
696                            if result == -1 {
697                                return Err(std::io::Error::last_os_error());
698                            }
699                        }
700                        Ok(())
701                    });
702                }
703            }
704        }
705        command
706    }
707
708    fn spawn_reader<R>(&self, pipe: R, source_stream: StreamKind, visible_stream: StreamKind)
709    where
710        R: Read + Send + 'static,
711    {
712        let shared = Arc::clone(&self.shared);
713        thread::spawn(move || {
714            let mut reader = pipe;
715            let mut chunk = vec![0_u8; 65536];
716            let mut pending = Vec::new();
717
718            loop {
719                match reader.read(&mut chunk) {
720                    Ok(0) => break,
721                    Ok(n) => {
722                        let lines = feed_chunk(&mut pending, &chunk[..n]);
723                        emit_lines(&shared, visible_stream, lines);
724                    }
725                    Err(_) => break,
726                }
727            }
728
729            if !pending.is_empty() {
730                emit_lines(&shared, visible_stream, vec![std::mem::take(&mut pending)]);
731            }
732
733            let mut guard = shared.queues.lock().expect("queue mutex poisoned");
734            match source_stream {
735                StreamKind::Stdout => guard.stdout_closed = true,
736                StreamKind::Stderr => guard.stderr_closed = true,
737            }
738            shared.condvar.notify_all();
739        });
740    }
741
742    fn set_returncode(&self, code: i32) {
743        self.shared.returncode.store(code as i64, Ordering::Release);
744        self.shared.condvar.notify_all();
745    }
746
747    fn wait_for_capture_completion_impl(&self) {
748        crate::rp_rust_debug_scope!(
749            "running_process_core::NativeProcess::wait_for_capture_completion"
750        );
751        if !self.config.capture {
752            return;
753        }
754
755        let mut guard = self.shared.queues.lock().expect("queue mutex poisoned");
756        while !(guard.stdout_closed && guard.stderr_closed) {
757            guard = self
758                .shared
759                .condvar
760                .wait(guard)
761                .expect("queue mutex poisoned");
762        }
763    }
764}
765
766#[cfg(unix)]
767pub fn unix_set_priority(pid: u32, nice: i32) -> Result<(), std::io::Error> {
768    let result = unsafe { libc::setpriority(libc::PRIO_PROCESS, pid, nice) };
769    if result == -1 {
770        return Err(std::io::Error::last_os_error());
771    }
772    Ok(())
773}
774
775#[cfg(unix)]
776pub fn unix_signal_process(pid: u32, signal: UnixSignal) -> Result<(), std::io::Error> {
777    let result = unsafe { libc::kill(pid as i32, unix_signal_raw(signal)) };
778    if result == -1 {
779        return Err(std::io::Error::last_os_error());
780    }
781    Ok(())
782}
783
784#[cfg(unix)]
785pub fn unix_signal_process_group(pid: i32, signal: UnixSignal) -> Result<(), std::io::Error> {
786    let result = unsafe { libc::killpg(pid, unix_signal_raw(signal)) };
787    if result == -1 {
788        return Err(std::io::Error::last_os_error());
789    }
790    Ok(())
791}
792
793fn log_spawned_child_pid(pid: u32) -> Result<(), std::io::Error> {
794    let Some(path) = std::env::var_os(CHILD_PID_LOG_PATH_ENV) else {
795        return Ok(());
796    };
797
798    let mut file = OpenOptions::new().create(true).append(true).open(path)?;
799    file.write_all(format!("{pid}\n").as_bytes())?;
800    file.flush()?;
801    Ok(())
802}
803
804#[cfg(windows)]
805fn assign_child_to_windows_kill_on_close_job_impl(
806    child: &Child,
807) -> Result<WindowsJobHandle, std::io::Error> {
808    crate::rp_rust_debug_scope!("running_process_core::assign_child_to_windows_kill_on_close_job");
809    use std::mem::zeroed;
810    use std::os::windows::io::AsRawHandle;
811
812    use winapi::shared::minwindef::FALSE;
813    use winapi::um::handleapi::{CloseHandle, INVALID_HANDLE_VALUE};
814    use winapi::um::jobapi2::{
815        AssignProcessToJobObject, CreateJobObjectW, SetInformationJobObject,
816    };
817    use winapi::um::winnt::{
818        JobObjectExtendedLimitInformation, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
819        JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
820    };
821
822    let handle = child.as_raw_handle();
823    let job = unsafe { CreateJobObjectW(std::ptr::null_mut(), std::ptr::null()) };
824    if job.is_null() || job == INVALID_HANDLE_VALUE {
825        return Err(std::io::Error::last_os_error());
826    }
827
828    let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { zeroed() };
829    info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
830    let ok = unsafe {
831        SetInformationJobObject(
832            job,
833            JobObjectExtendedLimitInformation,
834            (&mut info as *mut JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(),
835            std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
836        )
837    };
838    if ok == FALSE {
839        let err = std::io::Error::last_os_error();
840        unsafe { CloseHandle(job) };
841        return Err(err);
842    }
843
844    let ok = unsafe { AssignProcessToJobObject(job, handle.cast()) };
845    if ok == FALSE {
846        let err = std::io::Error::last_os_error();
847        unsafe { CloseHandle(job) };
848        return Err(err);
849    }
850
851    Ok(WindowsJobHandle(job as usize))
852}
853
854fn feed_chunk(pending: &mut Vec<u8>, chunk: &[u8]) -> Vec<Vec<u8>> {
855    let mut lines = Vec::new();
856    let mut start = 0;
857    let mut index = 0;
858
859    while index < chunk.len() {
860        if chunk[index] == b'\n' {
861            let end = if index > start && chunk[index - 1] == b'\r' {
862                index - 1
863            } else {
864                index
865            };
866            pending.extend_from_slice(&chunk[start..end]);
867            if !pending.is_empty() {
868                lines.push(std::mem::take(pending));
869            }
870            start = index + 1;
871        }
872        index += 1;
873    }
874
875    pending.extend_from_slice(&chunk[start..]);
876    lines
877}
878
879fn emit_lines(shared: &Arc<SharedState>, stream: StreamKind, lines: Vec<Vec<u8>>) {
880    if lines.is_empty() {
881        return;
882    }
883    let mut guard = shared.queues.lock().expect("queue mutex poisoned");
884    for line in lines {
885        let line_len = line.len();
886        match stream {
887            StreamKind::Stdout => {
888                guard.stdout_history_bytes += line_len;
889                guard.stdout_history.push_back(line.clone());
890                guard.stdout_queue.push_back(line.clone());
891            }
892            StreamKind::Stderr => {
893                guard.stderr_history_bytes += line_len;
894                guard.stderr_history.push_back(line.clone());
895                guard.stderr_queue.push_back(line.clone());
896            }
897        }
898        let event = StreamEvent { stream, line };
899        guard.combined_history_bytes += line_len;
900        guard.combined_history.push_back(event.clone());
901        guard.combined_queue.push_back(event);
902    }
903    shared.condvar.notify_all();
904}
905
906fn shell_command(command: &str) -> Command {
907    #[cfg(windows)]
908    {
909        use std::os::windows::process::CommandExt;
910
911        let mut cmd = Command::new("cmd");
912        cmd.raw_arg("/D /S /C \"");
913        cmd.raw_arg(command);
914        cmd.raw_arg("\"");
915        cmd
916    }
917    #[cfg(not(windows))]
918    {
919        let mut cmd = Command::new("sh");
920        cmd.arg("-lc").arg(command);
921        cmd
922    }
923}
924
925fn exit_code(status: std::process::ExitStatus) -> i32 {
926    #[cfg(unix)]
927    {
928        use std::os::unix::process::ExitStatusExt;
929        status
930            .code()
931            .unwrap_or_else(|| -status.signal().unwrap_or(1))
932    }
933    #[cfg(not(unix))]
934    {
935        status.code().unwrap_or(1)
936    }
937}
938
939#[cfg(unix)]
940fn unix_signal_raw(signal: UnixSignal) -> i32 {
941    match signal {
942        UnixSignal::Interrupt => libc::SIGINT,
943        UnixSignal::Terminate => libc::SIGTERM,
944        UnixSignal::Kill => libc::SIGKILL,
945    }
946}
947
948#[cfg(windows)]
949fn windows_priority_flags(nice: Option<i32>) -> u32 {
950    const IDLE_PRIORITY_CLASS: u32 = 0x0000_0040;
951    const BELOW_NORMAL_PRIORITY_CLASS: u32 = 0x0000_4000;
952    const ABOVE_NORMAL_PRIORITY_CLASS: u32 = 0x0000_8000;
953    const HIGH_PRIORITY_CLASS: u32 = 0x0000_0080;
954
955    match nice {
956        Some(value) if value >= 15 => IDLE_PRIORITY_CLASS,
957        Some(value) if value >= 1 => BELOW_NORMAL_PRIORITY_CLASS,
958        Some(value) if value <= -15 => HIGH_PRIORITY_CLASS,
959        Some(value) if value <= -1 => ABOVE_NORMAL_PRIORITY_CLASS,
960        _ => 0,
961    }
962}
963#[cfg(test)]
964mod tests;
running_process_core/lib.rs

running_process_core/
lib.rs
