Process termination monitoring implementation on Linux conflicts with processes spawned by other means

The approach used by Subprocess to monitor subprocess termination on Linux is fundamentally flawed as it calls waitid with P_ALL, and WEXITED without WNOWAIT, which will end up reaping pids that were spawned outside the Subprocess library.

Use an implementation more like swift-testing does for wait tests, which doesn't suffer from this issue.

Closes #82

Author: jakepetroules

Sources/Subprocess/CMakeLists.txt

@@ -13,6 +13,7 @@ target_sources(Subprocess PRIVATE
   Execution.swift
   Buffer.swift
   Error.swift
+  Locked.swift
   Teardown.swift
   Result.swift
   IO/Output.swift

Sources/Subprocess/Locked.swift

@@ -0,0 +1,138 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2025 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+//
+//===----------------------------------------------------------------------===//
+
+/// A protocol defining a type, generally platform-specific, that satisfies the
+/// requirements of a lock or mutex.
+protocol Lockable {
+    /// Initialize the lock at the given address.
+    ///
+    /// - Parameters:
+    ///   - lock: A pointer to uninitialized memory that should be initialized as
+    ///     an instance of this type.
+    static func initializeLock(at lock: UnsafeMutablePointer<Self>)
+
+    /// Deinitialize the lock at the given address.
+    ///
+    /// - Parameters:
+    ///   - lock: A pointer to initialized memory that should be deinitialized.
+    static func deinitializeLock(at lock: UnsafeMutablePointer<Self>)
+
+    /// Acquire the lock at the given address.
+    ///
+    /// - Parameters:
+    ///   - lock: The address of the lock to acquire.
+    static func unsafelyAcquireLock(at lock: UnsafeMutablePointer<Self>)
+
+    /// Relinquish the lock at the given address.
+    ///
+    /// - Parameters:
+    ///   - lock: The address of the lock to relinquish.
+    static func unsafelyRelinquishLock(at lock: UnsafeMutablePointer<Self>)
+}
+
+// MARK: -
+
+/// A type that wraps a value requiring access from a synchronous caller during
+/// concurrent execution.
+///
+/// Instances of this type use a lock to synchronize access to their raw values.
+/// The lock is not recursive.
+///
+/// Instances of this type can be used to synchronize access to shared data from
+/// a synchronous caller. Wherever possible, use actor isolation or other Swift
+/// concurrency tools.
+///
+/// This type is not part of the public interface of the testing library.
+struct LockedWith<L, T>: RawRepresentable where L: Lockable {
+    /// A type providing heap-allocated storage for an instance of ``Locked``.
+    private final class _Storage: ManagedBuffer<T, L> {
+        deinit {
+            withUnsafeMutablePointerToElements { lock in
+                L.deinitializeLock(at: lock)
+            }
+        }
+    }
+
+    /// Storage for the underlying lock and wrapped value.
+    private nonisolated(unsafe) var _storage: ManagedBuffer<T, L>
+
+    init(rawValue: T) {
+        _storage = _Storage.create(minimumCapacity: 1, makingHeaderWith: { _ in rawValue })
+        _storage.withUnsafeMutablePointerToElements { lock in
+            L.initializeLock(at: lock)
+        }
+    }
+
+    var rawValue: T {
+        withLock { $0 }
+    }
+
+    /// Acquire the lock and invoke a function while it is held.
+    ///
+    /// - Parameters:
+    ///   - body: A closure to invoke while the lock is held.
+    ///
+    /// - Returns: Whatever is returned by `body`.
+    ///
+    /// - Throws: Whatever is thrown by `body`.
+    ///
+    /// This function can be used to synchronize access to shared data from a
+    /// synchronous caller. Wherever possible, use actor isolation or other Swift
+    /// concurrency tools.
+    nonmutating func withLock<R>(_ body: (inout T) throws -> R) rethrows -> R where R: ~Copyable {
+        try _storage.withUnsafeMutablePointers { rawValue, lock in
+            L.unsafelyAcquireLock(at: lock)
+            defer {
+                L.unsafelyRelinquishLock(at: lock)
+            }
+            return try body(&rawValue.pointee)
+        }
+    }
+
+    /// Acquire the lock and invoke a function while it is held, yielding both the
+    /// protected value and a reference to the underlying lock guarding it.
+    ///
+    /// - Parameters:
+    ///   - body: A closure to invoke while the lock is held.
+    ///
+    /// - Returns: Whatever is returned by `body`.
+    ///
+    /// - Throws: Whatever is thrown by `body`.
+    ///
+    /// This function is equivalent to ``withLock(_:)`` except that the closure
+    /// passed to it also takes a reference to the underlying lock guarding this
+    /// instance's wrapped value. This function can be used when platform-specific
+    /// functionality such as a `pthread_cond_t` is needed. Because the caller has
+    /// direct access to the lock and is able to unlock and re-lock it, it is
+    /// unsafe to modify the protected value.
+    ///
+    /// - Warning: Callers that unlock the lock _must_ lock it again before the
+    ///   closure returns. If the lock is not acquired when `body` returns, the
+    ///   effect is undefined.
+    nonmutating func withUnsafeUnderlyingLock<R>(_ body: (UnsafeMutablePointer<L>, T) throws -> R) rethrows -> R where R: ~Copyable {
+        try withLock { value in
+            try _storage.withUnsafeMutablePointerToElements { lock in
+                try body(lock, value)
+            }
+        }
+    }
+}
+
+extension LockedWith: Sendable where T: Sendable {}
+
+// MARK: - Additions
+
+extension LockedWith {
+    /// Initialize an instance of this type with a raw value of `[:]`.
+    init<K, V>() where T == Dictionary<K, V> {
+        self.init(rawValue: [:])
+    }
+}

Sources/Subprocess/Platforms/Subprocess+Linux.swift

@@ -266,32 +266,29 @@ extension String {
 internal func monitorProcessTermination(
     forProcessWithIdentifier pid: ProcessIdentifier
 ) async throws -> TerminationStatus {
+    _ = setup
     return try await withCheckedThrowingContinuation { continuation in
         _childProcessContinuations.withLock { continuations in
-            if let existing = continuations.removeValue(forKey: pid.value),
-                case .status(let existingStatus) = existing
-            {
-                // We already have existing status to report
-                continuation.resume(returning: existingStatus)
-            } else {
-                // Save the continuation for handler
-                continuations[pid.value] = .continuation(continuation)
-            }
+            // Save the continuation for handler
+            let oldContinuation = continuations.updateValue(continuation, forKey: pid.value)
+            precondition(oldContinuation == nil)
+
+            _ = pthread_cond_signal(_waitThreadNoChildrenCondition)
         }
     }
 }
 
-private enum ContinuationOrStatus {
-    case continuation(CheckedContinuation<TerminationStatus, any Error>)
-    case status(TerminationStatus)
-}
-
-private let _childProcessContinuations:
-    Mutex<
-        [pid_t: ContinuationOrStatus]
-    > = Mutex([:])
+private let _childProcessContinuations =
+    LockedWith<
+        pthread_mutex_t,
+        [pid_t: CheckedContinuation<TerminationStatus, any Error>]
+    >()
 
-private let signalSource: SendableSourceSignal = SendableSourceSignal()
+private nonisolated(unsafe) let _waitThreadNoChildrenCondition = {
+    let result = UnsafeMutablePointer<pthread_cond_t>.allocate(capacity: 1)
+    _ = pthread_cond_init(result, nil)
+    return result
+}()
 
 private extension siginfo_t {
     var si_status: Int32 {
@@ -316,67 +313,71 @@ private extension siginfo_t {
 }
 
 private let setup: () = {
-    signalSource.setEventHandler {
-        while true {
-            var siginfo = siginfo_t()
-            guard waitid(P_ALL, id_t(0), &siginfo, WEXITED) == 0 || errno == EINTR else {
-                return
-            }
-            var status: TerminationStatus? = nil
-            switch siginfo.si_code {
-            case .init(CLD_EXITED):
-                status = .exited(siginfo.si_status)
-            case .init(CLD_KILLED), .init(CLD_DUMPED):
-                status = .unhandledException(siginfo.si_status)
-            case .init(CLD_TRAPPED), .init(CLD_STOPPED), .init(CLD_CONTINUED):
-                // Ignore these signals because they are not related to
-                // process exiting
-                break
-            default:
-                fatalError("Unexpected exit status: \(siginfo.si_code)")
-            }
-            if let status = status {
-                _childProcessContinuations.withLock { continuations in
+    var thread = pthread_t()
+    _ = pthread_create(
+        &thread,
+        nil,
+        { _ -> UnsafeMutableRawPointer? in
+            while true {
+                var siginfo = siginfo_t()
+                if waitid(P_ALL, id_t(0), &siginfo, WEXITED | WNOWAIT) == 0 {
                     let pid = siginfo.si_pid
-                    if let existing = continuations.removeValue(forKey: pid),
-                        case .continuation(let c) = existing
-                    {
-                        c.resume(returning: status)
-                    } else {
-                        // We don't have continuation yet, just state status
-                        continuations[pid] = .status(status)
+                    guard pid != 0, let c = _childProcessContinuations.withLock({ $0.removeValue(forKey: pid) }) else {
+                        continue
+                    }
+
+                    c.resume(with: Result {
+                        while true {
+                            var siginfo = siginfo_t()
+                            if waitid(P_PID, numericCast(pid), &siginfo, WEXITED) == 0 {
+                                var status: TerminationStatus? = nil
+                                switch siginfo.si_code {
+                                case .init(CLD_EXITED):
+                                    return .exited(siginfo.si_status)
+                                case .init(CLD_KILLED), .init(CLD_DUMPED):
+                                    return .unhandledException(siginfo.si_status)
+                                default:
+                                    fatalError("Unexpected exit status: \(siginfo.si_code)")
+                                }
+                            } else if errno != EINTR {
+                                throw SubprocessError.UnderlyingError(rawValue: errno)
+                            }
+                        }
+                    })
+                } else if errno == SIGCHLD {
+                    _childProcessContinuations.withUnsafeUnderlyingLock { lock, childProcessContinuations in
+                        if childProcessContinuations.isEmpty {
+                            _ = pthread_cond_wait(_waitThreadNoChildrenCondition, lock)
+                        }
                     }
                 }
             }
-        }
-    }
-    signalSource.resume()
+        },
+        nil
+    )
 }()
 
-/// Unchecked Sendable here since this class is only explicitly
-/// initialized once during the lifetime of the process
-final class SendableSourceSignal: @unchecked Sendable {
-    private let signalSource: DispatchSourceSignal
+private func _setupMonitorSignalHandler() {
+    // Only executed once
+    setup
+}
 
-    func setEventHandler(handler: @escaping DispatchSourceHandler) {
-        self.signalSource.setEventHandler(handler: handler)
+extension pthread_mutex_t: Lockable {
+    static func initializeLock(at lock: UnsafeMutablePointer<Self>) {
+        _ = pthread_mutex_init(lock, nil)
     }
 
-    func resume() {
-        self.signalSource.resume()
+    static func deinitializeLock(at lock: UnsafeMutablePointer<Self>) {
+        _ = pthread_mutex_destroy(lock)
     }
 
-    init() {
-        self.signalSource = DispatchSource.makeSignalSource(
-            signal: SIGCHLD,
-            queue: .global()
-        )
+    static func unsafelyAcquireLock(at lock: UnsafeMutablePointer<Self>) {
+        _ = pthread_mutex_lock(lock)
     }
-}
 
-private func _setupMonitorSignalHandler() {
-    // Only executed once
-    setup
+    static func unsafelyRelinquishLock(at lock: UnsafeMutablePointer<Self>) {
+        _ = pthread_mutex_unlock(lock)
+    }
 }
 
 #endif  // canImport(Glibc) || canImport(Android) || canImport(Musl)