Cancel AsyncIO when the child process exits (#272)

* Cancel AsyncIO when the child process exits

Subprocess used to assume that a child's exit closes the pipe FDs
it inherited. When the child forked a grandchild that also holds
those FDs, the kernel keeps the pipes alive after the child exits
and AsyncIO's read/write loops never see EOF or EPIPE — run()
hangs.

Split process monitoring into two phases:
- waitForProcessTermination waits for the kernel to report exit.
- reapProcess consumes the zombie and returns the status.

Configuration.run now spawns a parallel monitor task and races
it against the body closure via an AtomicCounter. If the process
exits while the body is still doing I/O, the monitor calls
AsyncIO.cancelAsyncIO(for:) to unblock every pending read or
write for that process.

Each platform's AsyncIO backend gains a Registration struct
that groups file descriptors (or IOCP completion keys on Windows)
by process. Cancellation walks those descriptors and either deletes
them from epoll/kqueue or calls CancelIoEx. A cancelledProcesses
set rejects late registrations so a register-vs-cancel race cannot
leak a stream. run clears the marker before reapProcess so PID
reuse is safe.

Adds regression tests covering stdin, stdout, and stderr on Unix
(POSIX shell + sleep grandchild) and Windows (PowerShell +
Start-Process -NoNewWindow grandchild).

* Update macOS CI to using Xcode 26.4

Author: iCharlesHu

.github/workflows/pull_request.yml

@@ -35,17 +35,17 @@ jobs:
           # Test dependencies
           yum install -y procps
         fi
-      linux_build_command: 'swift test && swift test -c release && swift test --disable-default-traits'
+      linux_build_command: "swift test && swift test -c release && swift test --disable-default-traits"
       enable_freebsd_checks: true
-      freebsd_build_command: 'swift test && swift test -c release && swift test --disable-default-traits'
+      freebsd_build_command: "swift test && swift test -c release && swift test --disable-default-traits"
       windows_swift_versions: '["6.2", "nightly-main"]'
       windows_build_command: |
         Invoke-Program swift test
         Invoke-Program swift test -c release
         Invoke-Program swift test --disable-default-traits
       enable_macos_checks: true
-      macos_xcode_versions: '["26.0"]'
-      macos_build_command: 'xcrun swift test && xcrun swift test -c release && xcrun swift test --disable-default-traits'
+      macos_xcode_versions: '["26.4"]'
+      macos_build_command: "xcrun swift test && xcrun swift test -c release && xcrun swift test --disable-default-traits"
       enable_linux_static_sdk_build: true
       enable_android_sdk_build: true
       android_ndk_versions: '["r27d", "r29"]'

Sources/Subprocess/API.swift

@@ -275,7 +275,10 @@ public func run<
         return try await withThrowingTaskGroup(of: _RunGroupResult<Output, Error>.self) { group in
             var writer: StandardInputWriter?
             if inputIOBox != nil {
-                let inputWriter = StandardInputWriter(diskIO: inputIOBox.take()!)
+                let inputWriter = StandardInputWriter(
+                    diskIO: inputIOBox.take()!,
+                    processIdentifier: processIdentifier
+                )
                 writer = inputWriter
 
                 if Input.self != CustomWriteInput.self {
@@ -294,7 +297,8 @@ public func run<
             if Output.self == SequenceOutput.self {
                 var diskIO = outputIOBox.take()
                 outputSequence = AsyncBufferSequence(
-                    diskIO: diskIO!.consumeDescriptor()
+                    diskIO: diskIO!.consumeDescriptor(),
+                    processIdentifier: processIdentifier
                 )
             } else if Output.OutputType.self == Void.self {
                 // No need to capture output
@@ -303,15 +307,18 @@ public func run<
             } else {
                 var diskIO = outputIOBox.take()
                 group.addTask {
-                    let result = try await output.captureOutput(from: diskIO.take())
+                    let result = try await output.captureOutput(
+                        from: diskIO.take(), for: processIdentifier
+                    )
                     return .standardOutputCaptured(result)
                 }
             }
 
             if Error.self == SequenceOutput.self {
                 var diskIO = errorIOBox.take()
                 errorSequence = AsyncBufferSequence(
-                    diskIO: diskIO!.consumeDescriptor()
+                    diskIO: diskIO!.consumeDescriptor(),
+                    processIdentifier: processIdentifier
                 )
             } else if Error.OutputType.self == Void.self {
                 // No need to capture error
@@ -320,7 +327,9 @@ public func run<
             } else {
                 var diskIO = errorIOBox.take()
                 group.addTask {
-                    let result = try await error.captureOutput(from: diskIO.take())
+                    let result = try await error.captureOutput(
+                        from: diskIO.take(), for: processIdentifier
+                    )
                     return .standardErrorCaptured(result)
                 }
             }

Sources/Subprocess/AsyncBufferSequence.swift

@@ -57,11 +57,13 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
         public typealias Element = Buffer
 
         private let diskIO: DiskIO
+        private let processIdentifier: ProcessIdentifier
         private let preferredBufferSize: Int
         private var buffer: [Buffer]
 
-        internal init(diskIO: DiskIO) {
+        internal init(diskIO: DiskIO, processIdentifier: ProcessIdentifier) {
             self.diskIO = diskIO
+            self.processIdentifier = processIdentifier
             self.buffer = []
             // Only need to query it once at beginning of stream
             self.preferredBufferSize = AsyncIO.queryPipeBufferSize(for: diskIO)
@@ -76,6 +78,7 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
             // Read more data
             let data = try await AsyncIO.shared.read(
                 from: self.diskIO,
+                for: self.processIdentifier,
                 upTo: self.preferredBufferSize
             )
             guard let data else {
@@ -97,10 +100,12 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
     }
 
     private let diskIO: DiskIO
+    private let processIdentifier: ProcessIdentifier
     private let state: State
 
-    internal init(diskIO: DiskIO) {
+    internal init(diskIO: DiskIO, processIdentifier: ProcessIdentifier) {
         self.diskIO = diskIO
+        self.processIdentifier = processIdentifier
         self.state = State()
     }
 
@@ -109,7 +114,7 @@ public struct AsyncBufferSequence: AsyncSequence, @unchecked Sendable {
         guard self.state.initializedCount() == 1 else {
             fatalError("AsyncBufferSequence is single pass. It can only be iterated once.")
         }
-        return Iterator(diskIO: self.diskIO)
+        return Iterator(diskIO: self.diskIO, processIdentifier: self.processIdentifier)
     }
 
     /// Splits the buffer into strings using the specified separator.

Sources/Subprocess/Configuration.swift

@@ -38,14 +38,14 @@ public struct Configuration: Sendable {
     public var environment: Environment
     /// The working directory to use when running the executable.
     ///
-    /// If this property is `nil`, the subprocess will inherit
-    /// the working directory from the parent process.
+    /// If this property is `nil`, the subprocess inherits the working
+    /// directory from the parent process.
     public var workingDirectory: FilePath?
     /// The platform-specific options to use when
     /// running the subprocess.
     public var platformOptions: PlatformOptions
 
-    /// Creates a Configuration with the parameters you provide.
+    /// Creates a configuration with the parameters you provide.
     /// - Parameters:
     ///   - executable: The executable to run.
     ///   - arguments: The arguments to pass to the executable.
@@ -107,37 +107,78 @@ public struct Configuration: Sendable {
         let processIdentifier = spawnResults.processIdentifier
 
         defer {
-            // Close process file descriptor now we finished monitoring
+            // Close the process file descriptor now that monitoring has finished.
             processIdentifier.close()
         }
 
         return try await withAsyncTaskCleanupHandler { () throws -> ExecutionOutcome<Result> in
-            let inputIO = spawnResults.inputWriteEnd()
-            let outputIO = spawnResults.outputReadEnd()
-            let errorIO = spawnResults.errorReadEnd()
-
-            let result: Swift.Result<Result, any Swift.Error>
-            do {
-                // Body runs in the same isolation
-                let bodyResult = try await body(processIdentifier, inputIO, outputIO, errorIO)
-
-                result = .success(bodyResult)
-            } catch {
-                result = .failure(error)
+            // The counter coordinates a race between two finishers: the body
+            // closure and the process-termination monitor. Whichever side
+            // increments first observes a value of `1` and owns the response;
+            // the other side sees `2` and stays out of the way.
+            //
+            // If the monitor wins, the child has exited while the body is
+            // still reading or writing. The body might be blocked on a pipe
+            // that an inherited grandchild keeps open, so the monitor calls
+            // `cancelAsyncIO` to unblock it. If the body wins, the I/O
+            // already finished cleanly and no cancellation is needed.
+            let taskFinishFlag = AtomicCounter()
+
+            let (result, monitorError) = try await withThrowingTaskGroup(
+                of: SubprocessError?.self,
+                returning: (Swift.Result<Result, any Swift.Error>, SubprocessError?).self
+            ) { group in
+                group.addTask {
+                    do throws(SubprocessError) {
+                        try await waitForProcessTermination(for: processIdentifier)
+                        if taskFinishFlag.addOne() == 1 {
+                            // The body closure hasn't finished but the child
+                            // process has terminated. Cancel all active
+                            // AsyncIO now.
+                            try AsyncIO.shared.cancelAsyncIO(for: processIdentifier)
+                        }
+                        return nil
+                    } catch {
+                        try? AsyncIO.shared.cancelAsyncIO(for: processIdentifier)
+                        return error
+                    }
+                }
+
+                let inputIO = spawnResults.inputWriteEnd()
+                let outputIO = spawnResults.outputReadEnd()
+                let errorIO = spawnResults.errorReadEnd()
+
+                let result: Swift.Result<Result, any Swift.Error>
+                do {
+                    // Body runs in the same isolation.
+                    let bodyResult = try await body(processIdentifier, inputIO, outputIO, errorIO)
+                    taskFinishFlag.addOne()
+                    result = .success(bodyResult)
+                } catch {
+                    result = .failure(error)
+                }
+
+                // Wait for the monitor child task to finish.
+                let monitorError = try await group.next() ?? nil
+                return (result, monitorError)
             }
 
-            // Ensure that we begin monitoring process termination after `body` runs
-            // and regardless of whether `body` throws, so that the pid gets reaped
-            // even if `body` throws, and we are not leaving zombie processes in the
-            // process table which will cause the process termination monitoring thread
-            // to effectively hang due to the pid never being awaited
-            let terminationStatus = try await monitorProcessTermination(
-                for: processIdentifier
-            )
+            // Drop the cancellation marker before reaping the zombie. After
+            // `reapProcess` runs the kernel can immediately reuse this PID,
+            // so the marker must be gone first; otherwise a concurrent
+            // `run()` that happens to inherit the same PID would see the
+            // stale entry and reject its registrations.
+            AsyncIO.shared.cleanup(processIdentifier: processIdentifier)
+
+            let terminationStatus = try reapProcess(with: processIdentifier)
+
+            if let monitorError {
+                throw monitorError
+            }
 
-            return ExecutionOutcome(
+            return try ExecutionOutcome(
                 terminationStatus: terminationStatus,
-                value: try result.get()
+                value: result.get()
             )
         } onCleanup: {
             let execution = Execution<Input, Output, Error>(
@@ -146,7 +187,7 @@ public struct Configuration: Sendable {
                 outputStream: nil,
                 errorStream: nil
             )
-            // Attempt to terminate the child process
+            // Attempt to terminate the child process.
             await execution.runTeardownSequence(
                 self.platformOptions.teardownSequence
             )