running_process_platform_internal/
sync_spawn_group.rs

1use std::io;
2use std::process::{Command, Stdio};
3use std::sync::{Arc, Mutex};
4use std::thread;
5use std::time::{Duration, Instant};
6
7const DEFAULT_KILL_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
8const KILL_DRAIN_TIMEOUT_ENV: &str = "RUNNING_PROCESS_KILL_DRAIN_TIMEOUT_MS";
9
10fn kill_drain_deadline() -> Instant {
11    let timeout = std::env::var(KILL_DRAIN_TIMEOUT_ENV)
12        .ok()
13        .and_then(|raw| raw.trim().parse::<u64>().ok())
14        .map(Duration::from_millis)
15        .unwrap_or(DEFAULT_KILL_DRAIN_TIMEOUT);
16    Instant::now() + timeout
17}
18
19fn poll_until<T>(
20    deadline: Instant,
21    interval: Duration,
22    mut poll: impl FnMut() -> io::Result<Option<T>>,
23) -> io::Result<Option<T>> {
24    loop {
25        if let Some(value) = poll()? {
26            return Ok(Some(value));
27        }
28        let now = Instant::now();
29        if now >= deadline {
30            return Ok(None);
31        }
32        thread::sleep(interval.min(deadline.saturating_duration_since(now)));
33    }
34}
35
36trait UnixChild: Send {
37    fn kill(&mut self) -> io::Result<()>;
38    fn wait(&mut self) -> io::Result<i32>;
39    fn try_wait(&mut self) -> io::Result<Option<i32>>;
40}
41
42impl UnixChild for std::process::Child {
43    fn kill(&mut self) -> io::Result<()> {
44        std::process::Child::kill(self)
45    }
46
47    fn wait(&mut self) -> io::Result<i32> {
48        std::process::Child::wait(self).map(crate::platform::process::exit_code)
49    }
50
51    fn try_wait(&mut self) -> io::Result<Option<i32>> {
52        Ok(std::process::Child::try_wait(self)?.map(crate::platform::process::exit_code))
53    }
54}
55
56impl crate::platform::process::DaemonChildControl for std::process::Child {
57    fn kill(&mut self) -> io::Result<()> {
58        std::process::Child::kill(self)
59    }
60
61    fn wait(&mut self) -> io::Result<i32> {
62        std::process::Child::wait(self).map(crate::platform::process::exit_code)
63    }
64
65    fn try_wait(&mut self) -> io::Result<Option<i32>> {
66        Ok(std::process::Child::try_wait(self)?.map(crate::platform::process::exit_code))
67    }
68}
69
70pub struct SpawnedInner {
71    child: Arc<Mutex<Option<Box<dyn UnixChild>>>>,
72    pgid: i32,
73    retain_exit_identity: bool,
74}
75
76impl SpawnedInner {
77    // WNOWAIT observes exit without releasing the leader's PID. New spawn-mode
78    // handles retain that identity until group control is finished. If another
79    // reaper has consumed it, waitid fails and control must fail closed.
80    fn observe_owned_exit(&self) -> io::Result<Option<i32>> {
81        super::observe_owned_child_exit(self.pgid)
82    }
83
84    pub fn kill(&self) -> io::Result<()> {
85        // Try the child first, then the process group, to make sure
86        // any siblings spawned inside go down too.
87        let mut guard = self.child.lock().expect("child mutex poisoned");
88        if self.retain_exit_identity {
89            if guard.is_none() {
90                return Ok(());
91            }
92            self.observe_owned_exit()?;
93        }
94        if let Some(child) = guard.as_mut() {
95            let _ = child.kill();
96        }
97        drop(guard);
98        let _ = crate::platform::process::unix_signal_process_group(
99            self.pgid,
100            crate::platform::process::UnixSignalKind::Kill,
101        );
102        Ok(())
103    }
104
105    pub fn wait(&self) -> io::Result<i32> {
106        if self.retain_exit_identity {
107            loop {
108                if let Some(code) = self.try_wait()? {
109                    return Ok(code);
110                }
111                thread::sleep(Duration::from_millis(10));
112            }
113        }
114        let mut guard = self.child.lock().expect("child mutex poisoned");
115        let Some(child) = guard.as_mut() else {
116            return Err(io::Error::other("child handle absent"));
117        };
118        child.wait()
119    }
120
121    pub fn try_wait(&self) -> io::Result<Option<i32>> {
122        let mut guard = self.child.lock().expect("child mutex poisoned");
123        if self.retain_exit_identity {
124            return self.observe_owned_exit();
125        }
126        let Some(child) = guard.as_mut() else {
127            return Ok(None);
128        };
129        child.try_wait()
130    }
131
132    pub fn shutdown(&mut self) {
133        self.shutdown_with_deadline(kill_drain_deadline());
134    }
135
136    fn shutdown_with_deadline(&mut self, deadline: Instant) {
137        let identity_owned = !self.retain_exit_identity || self.observe_owned_exit().is_ok();
138        let group_signaled = identity_owned && crate::platform::process::unix_signal_process_group(
139            self.pgid,
140            crate::platform::process::UnixSignalKind::Kill,
141        )
142        .is_ok();
143        let Some(mut child) = self.child.lock().expect("child mutex poisoned").take() else {
144            return;
145        };
146        if !group_signaled && identity_owned {
147            let _ = child.kill();
148        }
149        match poll_until(deadline, Duration::from_millis(10), || child.try_wait()) {
150            Ok(Some(_)) => {}
151            Ok(None) | Err(_) => spawn_background_reaper(child),
152        }
153    }
154}
155
156impl crate::platform::process::SpawnedChildControl for SpawnedInner {
157    #[cfg(feature = "independent-spawn")]
158    fn retain_exit_identity(&mut self) {
159        self.retain_exit_identity = true;
160    }
161    fn kill(&mut self) -> io::Result<()> {
162        SpawnedInner::kill(self)
163    }
164
165    fn wait(&mut self) -> io::Result<i32> {
166        SpawnedInner::wait(self)
167    }
168
169    fn try_wait(&mut self) -> io::Result<Option<i32>> {
170        SpawnedInner::try_wait(self)
171    }
172
173    fn shutdown(&mut self) {
174        SpawnedInner::shutdown(self);
175    }
176}
177
178impl Drop for SpawnedInner {
179    fn drop(&mut self) {
180        // Detached handles do not invoke shutdown. Release their wait ownership
181        // without killing the process, including a retained terminal leader.
182        if self.retain_exit_identity {
183            if let Some(mut child) = self.child.lock().expect("child mutex poisoned").take() {
184                if !matches!(child.try_wait(), Ok(Some(_))) {
185                    spawn_background_reaper(child);
186                }
187            }
188        }
189    }
190}
191
192fn spawn_background_reaper(mut child: Box<dyn UnixChild>) {
193    thread::spawn(move || {
194        // Once ownership is off the caller's teardown path, a blocking wait is
195        // the most reliable terminal policy: it reaps exactly once without a
196        // retry loop, spinning, or retaining the shared child mutex.
197        let _ = child.wait();
198    });
199}
200
201fn slot_to_stdio(slot: &crate::platform::process::StdioSource<'_>) -> io::Result<Stdio> {
202    match slot {
203        crate::platform::process::StdioSource::Null => Ok(Stdio::null()),
204        crate::platform::process::StdioSource::Parent => Ok(Stdio::inherit()),
205        crate::platform::process::StdioSource::File(file) => Ok(Stdio::from(file.try_clone()?)),
206        crate::platform::process::StdioSource::Pipe => Ok(Stdio::piped()),
207    }
208}
209
210fn daemon_slot_to_stdio(
211    slot: &crate::platform::process::DaemonStdioSource<'_>,
212) -> io::Result<Stdio> {
213    match slot {
214        crate::platform::process::DaemonStdioSource::Null => Ok(Stdio::null()),
215        crate::platform::process::DaemonStdioSource::File(file) => {
216            Ok(Stdio::from(file.try_clone()?))
217        }
218    }
219}
220
221pub fn spawn_sync_daemon(
222    command: &mut Command,
223    stdio: crate::platform::process::DaemonStdio<'_>,
224    environment: crate::platform::process::SyncEnvironment,
225    _breakaway: bool,
226) -> io::Result<crate::platform::process::DaemonChild> {
227    spawn_sync_daemon_inner(command, stdio, environment, None)
228}
229
230pub fn spawn_sync_daemon_with_inheritance(
231    command: &mut Command,
232    stdio: crate::platform::process::DaemonStdio<'_>,
233    environment: crate::platform::process::SyncEnvironment,
234    _breakaway: bool,
235    inheritance: crate::platform::process::DaemonExecInheritance,
236) -> io::Result<crate::platform::process::DaemonChild> {
237    spawn_sync_daemon_inner(command, stdio, environment, Some(inheritance))
238}
239
240fn spawn_sync_daemon_inner(
241    command: &mut Command,
242    stdio: crate::platform::process::DaemonStdio<'_>,
243    environment: crate::platform::process::SyncEnvironment,
244    inheritance: Option<crate::platform::process::DaemonExecInheritance>,
245) -> io::Result<crate::platform::process::DaemonChild> {
246    apply_environment(command, environment);
247    command
248        .stdin(Stdio::null())
249        .stdout(daemon_slot_to_stdio(&stdio.stdout)?)
250        .stderr(daemon_slot_to_stdio(&stdio.stderr)?);
251
252    match inheritance {
253        Some(inheritance) => {
254            crate::platform::process::configure_sync_daemon_command_with_inheritance(
255                command,
256                inheritance,
257            )?;
258        }
259        None => crate::platform::process::configure_sync_daemon_command(command)?,
260    }
261
262    let child = command.spawn()?;
263    let pid = child.id();
264    Ok(crate::platform::process::DaemonChild {
265        pid,
266        inner: Box::new(child),
267    })
268}
269
270pub fn spawn_sync(
271    command: &mut Command,
272    stdio: crate::platform::process::SpawnStdio<'_>,
273    environment: crate::platform::process::SyncEnvironment,
274) -> io::Result<crate::platform::process::SpawnedChild> {
275    spawn_sync_inner(command, stdio, environment, false)
276}
277
278#[cfg(feature = "independent-spawn")]
279pub(crate) fn spawn_sync_owned_daemon(command: &mut Command, stdio: crate::platform::process::SpawnStdio<'_>, environment: crate::platform::process::SyncEnvironment) -> io::Result<crate::platform::process::SpawnedChild> {
280    spawn_sync_inner(command, stdio, environment, true)
281}
282
283fn spawn_sync_inner(command: &mut Command, stdio: crate::platform::process::SpawnStdio<'_>, environment: crate::platform::process::SyncEnvironment, detached: bool) -> io::Result<crate::platform::process::SpawnedChild> {
284    apply_environment(command, environment);
285    command.stdin(slot_to_stdio(&stdio.stdin)?);
286    command.stdout(slot_to_stdio(&stdio.stdout)?);
287    command.stderr(slot_to_stdio(&stdio.stderr)?);
288
289    if detached { crate::platform::process::configure_sync_daemon_command(command)?; }
290    else { crate::platform::process::configure_sync_contained_command(command)?; }
291
292    let mut child = command.spawn()?;
293    let pid = child.id();
294    let pgid = pid as i32;
295
296    let stdin = child.stdin.take();
297    let stdout = child.stdout.take();
298    let stderr = child.stderr.take();
299
300    let child: Arc<Mutex<Option<Box<dyn UnixChild>>>> = Arc::new(Mutex::new(Some(Box::new(child))));
301
302    // Drain watcher: wait for exit, then sleep `drain_timeout`. We
303    // don't proactively close anything on Unix — Rust's ChildStdin/etc.
304    // own their fds; once the child exits and the kernel ref-counts
305    // its copies to zero, parent reads will EOF naturally.
306    if let Some(timeout) = stdio.drain_timeout {
307        let child_clone = Arc::clone(&child);
308        thread::spawn(move || {
309            // Borrow child for try_wait.  We do a polling loop so
310            // shutdown() taking the inner Child during Drop doesn't
311            // wedge us.
312            loop {
313                {
314                    let mut guard = child_clone.lock().expect("child mutex poisoned");
315                    match guard.as_mut() {
316                        Some(c) => match c.try_wait() {
317                            Ok(Some(_)) => break,
318                            Ok(None) => {}
319                            Err(_) => break,
320                        },
321                        None => return,
322                    }
323                }
324                // #199: intentional — try_wait poll on the contained
325                // child, 50ms cadence inside a bounded outer drain
326                // loop. waitpid(WNOHANG)-equivalent semantics.
327                thread::sleep(std::time::Duration::from_millis(50));
328            }
329            // #199: intentional — post-mortem pipe drain. Children's
330            // write-ends of the captured stdio pipes are still being
331            // closed by the kernel after exit; this gives readers a
332            // chance to see the final bytes before the watcher
333            // releases its keep-alive.
334            thread::sleep(timeout);
335        });
336    }
337
338    Ok(crate::platform::process::SpawnedChild {
339        kill_on_drop: true,
340        stdin,
341        stdout,
342        stderr,
343        pid,
344        inner: Box::new(SpawnedInner { child, pgid, retain_exit_identity: false }),
345    })
346}
347
348fn apply_environment(
349    command: &mut Command,
350    environment: crate::platform::process::SyncEnvironment,
351) {
352    let crate::platform::process::SyncEnvironment::Explicit(base) = environment else {
353        return;
354    };
355
356    // `env_clear` also clears Command's mutation map. Preserve additions,
357    // overrides, and removals so they are replayed after the selected base.
358    let explicit: Vec<_> = command
359        .get_envs()
360        .map(|(key, value)| (key.to_os_string(), value.map(std::ffi::OsStr::to_os_string)))
361        .collect();
362    command.env_clear();
363    command.envs(base);
364    for (key, value) in explicit {
365        match value {
366            Some(value) => {
367                command.env(key, value);
368            }
369            None => {
370                command.env_remove(key);
371            }
372        }
373    }
374}
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379    use std::sync::atomic::{AtomicUsize, Ordering};
380    use std::sync::{mpsc, Condvar};
381
382    struct FakeChild {
383        wait_gate: Arc<(Mutex<bool>, Condvar)>,
384        waits: Arc<AtomicUsize>,
385        kills: Arc<AtomicUsize>,
386    }
387
388    impl UnixChild for FakeChild {
389        fn kill(&mut self) -> io::Result<()> {
390            self.kills.fetch_add(1, Ordering::SeqCst);
391            Ok(())
392        }
393
394        fn wait(&mut self) -> io::Result<i32> {
395            self.waits.fetch_add(1, Ordering::SeqCst);
396            let (lock, condvar) = &*self.wait_gate;
397            let mut released = lock.lock().expect("wait gate mutex poisoned");
398            while !*released {
399                released = condvar.wait(released).expect("wait gate mutex poisoned");
400            }
401            Ok(0)
402        }
403
404        fn try_wait(&mut self) -> io::Result<Option<i32>> {
405            self.waits.fetch_add(1, Ordering::SeqCst);
406            let released = *self.wait_gate.0.lock().expect("wait gate mutex poisoned");
407            Ok(released.then_some(0))
408        }
409    }
410
411    struct BlockedFixture {
412        inner: SpawnedInner,
413        child: Arc<Mutex<Option<Box<dyn UnixChild>>>>,
414        wait_gate: Arc<(Mutex<bool>, Condvar)>,
415        waits: Arc<AtomicUsize>,
416        kills: Arc<AtomicUsize>,
417    }
418
419    fn blocked_inner() -> BlockedFixture {
420        let wait_gate = Arc::new((Mutex::new(false), Condvar::new()));
421        let waits = Arc::new(AtomicUsize::new(0));
422        let kills = Arc::new(AtomicUsize::new(0));
423        let child: Arc<Mutex<Option<Box<dyn UnixChild>>>> =
424            Arc::new(Mutex::new(Some(Box::new(FakeChild {
425                wait_gate: Arc::clone(&wait_gate),
426                waits: Arc::clone(&waits),
427                kills: Arc::clone(&kills),
428            }))));
429        BlockedFixture {
430            inner: SpawnedInner {
431                child: Arc::clone(&child),
432                pgid: i32::MAX,
433                retain_exit_identity: false,
434            },
435            child,
436            wait_gate,
437            waits,
438            kills,
439        }
440    }
441
442    fn release_wait(wait_gate: &Arc<(Mutex<bool>, Condvar)>) {
443        let (lock, condvar) = &**wait_gate;
444        *lock.lock().expect("wait gate mutex poisoned") = true;
445        condvar.notify_all();
446    }
447
448    struct ShutdownOnDrop {
449        inner: Option<SpawnedInner>,
450        deadline: Instant,
451    }
452
453    impl Drop for ShutdownOnDrop {
454        fn drop(&mut self) {
455            self.inner
456                .as_mut()
457                .expect("test wrapper missing inner")
458                .shutdown_with_deadline(self.deadline);
459        }
460    }
461
462    #[test]
463    fn drop_is_bounded_when_child_wait_does_not_complete() {
464        // Regression for #619: SpawnedChild::drop delegates directly to
465        // SpawnedInner::shutdown, modeled by this wrapper around the fake child.
466        let BlockedFixture {
467            inner, wait_gate, ..
468        } = blocked_inner();
469        let (tx, rx) = mpsc::channel();
470        let started = Instant::now();
471        let worker = thread::spawn(move || {
472            drop(ShutdownOnDrop {
473                inner: Some(inner),
474                deadline: Instant::now() + Duration::from_millis(50),
475            });
476            let _ = tx.send(started.elapsed());
477        });
478
479        // The property is causal, not a stopwatch reading: Drop must return
480        // WITHOUT waiting for the child, so it must report back before the
481        // gate is released. A blocked Drop cannot, whatever the machine load.
482        //
483        // Asserting a wall-clock bound instead conflated that with "finished
484        // inside 100ms", which a loaded runner broke by 0.2ms. The window
485        // below is generous because it only bounds how long a *failure* takes
486        // to detect: a correct Drop returns at its 50ms deadline and never
487        // approaches it.
488        let timely = rx.recv_timeout(Duration::from_secs(5));
489        release_wait(&wait_gate);
490        let returned_before_release = timely.is_ok();
491        let elapsed = timely
492            .or_else(|_| rx.recv_timeout(Duration::from_secs(5)))
493            .expect("shutdown did not unblock even after releasing fake child");
494        worker.join().expect("shutdown worker panicked");
495        assert!(
496            returned_before_release,
497            "Drop blocked in child.wait() until the fake child was released              (took {elapsed:?}); its deadline should have bounded it"
498        );
499    }
500
501    #[test]
502    fn shutdown_does_not_hold_child_mutex_while_reaping() {
503        let BlockedFixture {
504            mut inner,
505            child,
506            wait_gate,
507            waits,
508            ..
509        } = blocked_inner();
510        let worker = thread::spawn(move || {
511            inner.shutdown_with_deadline(Instant::now() + Duration::from_millis(50));
512        });
513        let deadline = Instant::now() + Duration::from_secs(1);
514        while waits.load(Ordering::SeqCst) == 0 && Instant::now() < deadline {
515            thread::yield_now();
516        }
517        assert_eq!(waits.load(Ordering::SeqCst), 1, "fake wait never started");
518
519        let child_mutex_available = child.try_lock().is_ok();
520        release_wait(&wait_gate);
521        worker.join().expect("shutdown worker panicked");
522        assert!(
523            child_mutex_available,
524            "shutdown held the child mutex across reaping"
525        );
526    }
527
528    struct ReadyChild {
529        polls: Arc<AtomicUsize>,
530        waits: Arc<AtomicUsize>,
531    }
532
533    impl UnixChild for ReadyChild {
534        fn kill(&mut self) -> io::Result<()> {
535            Ok(())
536        }
537
538        fn wait(&mut self) -> io::Result<i32> {
539            self.waits.fetch_add(1, Ordering::SeqCst);
540            Ok(0)
541        }
542
543        fn try_wait(&mut self) -> io::Result<Option<i32>> {
544            self.polls.fetch_add(1, Ordering::SeqCst);
545            Ok(Some(0))
546        }
547    }
548
549    #[test]
550    fn shutdown_reaps_ready_child_exactly_once() {
551        let polls = Arc::new(AtomicUsize::new(0));
552        let waits = Arc::new(AtomicUsize::new(0));
553        let child: Arc<Mutex<Option<Box<dyn UnixChild>>>> =
554            Arc::new(Mutex::new(Some(Box::new(ReadyChild {
555                polls: Arc::clone(&polls),
556                waits: Arc::clone(&waits),
557            }))));
558        let mut inner = SpawnedInner {
559            child,
560            pgid: i32::MAX,
561            retain_exit_identity: false,
562        };
563
564        inner.shutdown_with_deadline(Instant::now() + Duration::from_secs(1));
565
566        assert_eq!(polls.load(Ordering::SeqCst), 1);
567        assert_eq!(waits.load(Ordering::SeqCst), 0);
568    }
569
570    #[test]
571    fn shutdown_falls_back_to_direct_kill_when_group_signal_fails() {
572        let BlockedFixture {
573            mut inner,
574            wait_gate,
575            kills,
576            ..
577        } = blocked_inner();
578        release_wait(&wait_gate);
579
580        inner.shutdown_with_deadline(Instant::now() + Duration::from_secs(1));
581
582        assert_eq!(kills.load(Ordering::SeqCst), 1);
583    }
584}
running_process_platform_internal/sync_spawn_group.rs

running_process_platform_internal/
sync_spawn_group.rs
