Call spawn() on a background thread because it might block

Introduce runOnBackgroundThread() to run closures on a background thread without blocking the Swift Concurrency thread pool

Author: iCharlesHu

Sources/Subprocess/CMakeLists.txt

@@ -14,6 +14,7 @@ add_library(Subprocess
   Buffer.swift
   Error.swift
   Teardown.swift
+  Thread.swift
   Result.swift
   IO/Output.swift
   IO/Input.swift

Sources/Subprocess/Configuration.swift

@@ -84,7 +84,7 @@ public struct Configuration: Sendable {
         isolation: isolated (any Actor)? = #isolation,
         _ body: ((Execution, consuming IOChannel?, consuming IOChannel?, consuming IOChannel?) async throws -> Result)
     ) async throws -> ExecutionResult<Result> {
-        let spawnResults = try self.spawn(
+        let spawnResults = try await self.spawn(
             withInput: input,
             outputPipe: output,
             errorPipe: error

Sources/Subprocess/IO/AsyncIO+Windows.swift

@@ -43,7 +43,7 @@ final class AsyncIO: @unchecked Sendable {
         ) rethrows -> ResultType
     }
 
-    private final class MonitorThreadContext {
+    private struct MonitorThreadContext: @unchecked Sendable {
         let ioCompletionPort: HANDLE
 
         init(ioCompletionPort: HANDLE) {
@@ -60,11 +60,9 @@ final class AsyncIO: @unchecked Sendable {
     internal init() {
         var maybeSetupError: SubprocessError? = nil
         // Create the the completion port
-        guard
-            let port = CreateIoCompletionPort(
-                INVALID_HANDLE_VALUE, nil, 0, 0
-            ), port != INVALID_HANDLE_VALUE
-        else {
+        guard let ioCompletionPort = CreateIoCompletionPort(
+            INVALID_HANDLE_VALUE, nil, 0, 0
+        ), ioCompletionPort != INVALID_HANDLE_VALUE else {
             let error = SubprocessError(
                 code: .init(.asyncIOFailed("CreateIoCompletionPort failed")),
                 underlyingError: .init(rawValue: GetLastError())
@@ -73,17 +71,12 @@ final class AsyncIO: @unchecked Sendable {
             self.monitorThread = .failure(error)
             return
         }
-        self.ioCompletionPort = .success(port)
+        self.ioCompletionPort = .success(ioCompletionPort)
         // Create monitor thread
-        let threadContext = MonitorThreadContext(ioCompletionPort: port)
-        let threadContextPtr = Unmanaged.passRetained(threadContext)
-        /// Microsoft documentation for `CreateThread` states:
-        /// > A thread in an executable that calls the C run-time library (CRT)
-        /// > should use the _beginthreadex and _endthreadex functions for
-        /// > thread management rather than CreateThread and ExitThread
-        let threadHandleValue = _beginthreadex(
-            nil, 0,
-            { args in
+        let context = MonitorThreadContext(ioCompletionPort: ioCompletionPort)
+        let threadHandle: HANDLE
+        do {
+            threadHandle = try begin_thread_x {
                 func reportError(_ error: SubprocessError) {
                     let continuations = _registration.withLock { store in
                         return store.values
@@ -93,15 +86,50 @@ final class AsyncIO: @unchecked Sendable {
                     }
                 }
 
-                let unmanaged = Unmanaged<MonitorThreadContext>.fromOpaque(args!)
-                let context = unmanaged.takeRetainedValue()
-
+                // Monitor loop
+                while true {
+                    var bytesTransferred: DWORD = 0
+                    var targetFileDescriptor: UInt64 = 0
+                    var overlapped: LPOVERLAPPED? = nil
                 // Monitor loop
                 while true {
                     var bytesTransferred: DWORD = 0
                     var targetFileDescriptor: UInt64 = 0
                     var overlapped: LPOVERLAPPED? = nil
 
+                    let monitorResult = GetQueuedCompletionStatus(
+                        context.ioCompletionPort,
+                        &bytesTransferred,
+                        &targetFileDescriptor,
+                        &overlapped,
+                        INFINITE
+                    )
+                    if !monitorResult {
+                        let lastError = GetLastError()
+                        if lastError == ERROR_BROKEN_PIPE {
+                            // We finished reading the handle. Signal EOF by
+                            // finishing the stream.
+                            // NOTE: here we deliberately leave now unused continuation
+                            // in the store. Windows does not offer an API to remove a
+                            // HANDLE from an IOCP port, therefore we leave the registration
+                            // to signify the HANDLE has already been resisted.
+                            let continuation = _registration.withLock { store -> SignalStream.Continuation? in
+                                if let continuation = store[targetFileDescriptor] {
+                                    return continuation
+                                }
+                                return nil
+                            }
+                            continuation?.finish()
+                            continue
+                        } else {
+                            let error = SubprocessError(
+                                code: .init(.asyncIOFailed("GetQueuedCompletionStatus failed")),
+                                underlyingError: .init(rawValue: lastError)
+                            )
+                            reportError(error)
+                            break
+                        }
+                    }
                     let monitorResult = GetQueuedCompletionStatus(
                         context.ioCompletionPort,
                         &bytesTransferred,
@@ -149,16 +177,13 @@ final class AsyncIO: @unchecked Sendable {
                     }
                     continuation?.yield(bytesTransferred)
                 }
+
                 return 0
-            }, threadContextPtr.toOpaque(), 0, nil)
-        guard threadHandleValue > 0,
-            let threadHandle = HANDLE(bitPattern: threadHandleValue)
-        else {
-            // _beginthreadex uses errno instead of GetLastError()
-            let capturedError = _subprocess_windows_get_errno()
+            }
+        } catch let underlyingError {
             let error = SubprocessError(
-                code: .init(.asyncIOFailed("_beginthreadex failed")),
-                underlyingError: .init(rawValue: capturedError)
+                code: .init(.asyncIOFailed("Failed to create monitor thread")),
+                underlyingError: underlyingError
             )
             self.monitorThread = .failure(error)
             return

Sources/Subprocess/Platforms/Subprocess+Darwin.swift

@@ -153,11 +153,24 @@ extension PlatformOptions: CustomStringConvertible, CustomDebugStringConvertible
 
 // MARK: - Spawn
 extension Configuration {
+    // @unchecked Sendable because we need to capture UnsafePointers
+    // to send to another thread. While UnsafePointers are not
+    // Sendable, we are not mutating them -- we only need these type
+    // for C interface.
+    internal struct SpawnContext: @unchecked Sendable {
+        let fileActions: posix_spawn_file_actions_t?
+        let spawnAttributes: posix_spawnattr_t?
+        let argv: [UnsafeMutablePointer<CChar>?]
+        let env: [UnsafeMutablePointer<CChar>?]
+        let uidPtr: UnsafeMutablePointer<uid_t>?
+        let gidPtr: UnsafeMutablePointer<gid_t>?
+    }
+
     internal func spawn(
         withInput inputPipe: consuming CreatedPipe,
         outputPipe: consuming CreatedPipe,
         errorPipe: consuming CreatedPipe
-    ) throws -> SpawnResult {
+    ) async throws -> SpawnResult {
         // Instead of checking if every possible executable path
         // is valid, spawn each directly and catch ENOENT
         let possiblePaths = self.executable.possibleExecutablePaths(
@@ -167,7 +180,7 @@ extension Configuration {
         var outputPipeBox: CreatedPipe? = consume outputPipe
         var errorPipeBox: CreatedPipe? = consume errorPipe
 
-        return try self.preSpawn { args throws -> SpawnResult in
+        return try await self.preSpawn { args throws -> SpawnResult in
             let (env, uidPtr, gidPtr, supplementaryGroups) = args
             var _inputPipe = inputPipeBox.take()!
             var _outputPipe = outputPipeBox.take()!
@@ -181,8 +194,6 @@ extension Configuration {
             let errorWriteFileDescriptor: IODescriptor? = _errorPipe.writeFileDescriptor()
 
             for possibleExecutablePath in possiblePaths {
-                var pid: pid_t = 0
-
                 // Setup Arguments
                 let argv: [UnsafeMutablePointer<CChar>?] = self.arguments.createArgs(
                     withExecutablePath: possibleExecutablePath
@@ -389,21 +400,35 @@ extension Configuration {
                 }
 
                 // Spawn
-                let spawnError: CInt = possibleExecutablePath.withCString { exePath in
-                    return supplementaryGroups.withOptionalUnsafeBufferPointer { sgroups in
-                        return _subprocess_spawn(
-                            &pid,
-                            exePath,
-                            &fileActions,
-                            &spawnAttributes,
-                            argv,
-                            env,
-                            uidPtr,
-                            gidPtr,
-                            Int32(supplementaryGroups?.count ?? 0),
-                            sgroups?.baseAddress,
-                            self.platformOptions.createSession ? 1 : 0
-                        )
+                let spawnContext = SpawnContext(
+                    fileActions: fileActions,
+                    spawnAttributes: spawnAttributes,
+                    argv: argv,
+                    env: env,
+                    uidPtr: uidPtr,
+                    gidPtr: gidPtr
+                )
+                let (spawnError, pid) = try await runOnBackgroundThread {
+                    return possibleExecutablePath.withCString { exePath in
+                        return supplementaryGroups.withOptionalUnsafeBufferPointer { sgroups in
+                            var pid: pid_t = 0
+                            var _fileActions = spawnContext.fileActions
+                            var _spawnAttributes = spawnContext.spawnAttributes
+                            let rc = _subprocess_spawn(
+                                &pid,
+                                exePath,
+                                &_fileActions,
+                                &_spawnAttributes,
+                                spawnContext.argv,
+                                spawnContext.env,
+                                spawnContext.uidPtr,
+                                spawnContext.gidPtr,
+                                Int32(supplementaryGroups?.count ?? 0),
+                                sgroups?.baseAddress,
+                                self.platformOptions.createSession ? 1 : 0
+                            )
+                            return (rc, pid)
+                        }
                     }
                 }
                 // Spawn error

Sources/Subprocess/Platforms/Subprocess+Unix.swift

@@ -350,8 +350,8 @@ extension Configuration {
     )
 
     internal func preSpawn<Result: ~Copyable>(
-        _ work: (PreSpawnArgs) throws -> Result
-    ) throws -> Result {
+        _ work: (PreSpawnArgs) async throws -> Result
+    ) async throws -> Result {
         // Prepare environment
         let env = self.environment.createEnv()
         defer {
@@ -378,7 +378,7 @@ extension Configuration {
         if let groupsValue = self.platformOptions.supplementaryGroups {
             supplementaryGroups = groupsValue
         }
-        return try work(
+        return try await work(
             (
                 env: env,
                 uidPtr: uidPtr,