Conditionally adopt swift-subprocess

This fully resolves working directory thread safety issues with subprocess spawning across all platforms. For now, subprocess is adopted conditionally in order to continue building in certain environments where the Subprocess module may not be available, in which case we fall back to Foundation Process.

Closes #441

Author: jakepetroules

CMakeLists.txt

@@ -81,6 +81,7 @@ add_compile_definitions(USE_STATIC_PLUGIN_INITIALIZATION)
 find_package(ArgumentParser)
 find_package(LLBuild)
 find_package(SwiftDriver)
+find_package(SwiftSubprocess)
 find_package(SwiftSystem)
 find_package(TSC)
 # NOTE: these two are required for LLBuild dependencies

Package.swift

@@ -201,6 +201,7 @@ let package = Package(
                 "SWBCSupport",
                 "SWBLibc",
                 .product(name: "ArgumentParser", package: "swift-argument-parser"),
+                .product(name: "Subprocess", package: "swift-subprocess"),
                 .product(name: "SystemPackage", package: "swift-system", condition: .when(platforms: [.linux, .openbsd, .android, .windows, .custom("freebsd")])),
             ],
             exclude: ["CMakeLists.txt"],
@@ -447,6 +448,7 @@ for target in package.targets {
 if useLocalDependencies {
     package.dependencies += [
         .package(path: "../swift-driver"),
+        .package(path: "../swift-subprocess"),
         .package(path: "../swift-system"),
         .package(path: "../swift-argument-parser"),
     ]
@@ -456,6 +458,7 @@ if useLocalDependencies {
 } else {
     package.dependencies += [
         .package(url: "https://github.com/swiftlang/swift-driver.git", branch: "main"),
+        .package(url: "https://github.com/swiftlang/swift-subprocess.git", branch: "main"),
         .package(url: "https://github.com/apple/swift-system.git", .upToNextMajor(from: "1.5.0")),
         .package(url: "https://github.com/apple/swift-argument-parser.git", from: "1.0.3"),
     ]

Sources/SWBCore/ProcessExecutionCache.swift

@@ -17,11 +17,6 @@ public final class ProcessExecutionCache: Sendable {
     private let workingDirectory: Path?
 
     public init(workingDirectory: Path? = .root) {
-        // FIXME: Work around lack of thread-safe working directory support in Foundation (Amazon Linux 2, OpenBSD). Executing processes in the current working directory is less deterministic, but all of the clients which use this class are generally not expected to be sensitive to the working directory anyways. This workaround can be removed once we drop support for Amazon Linux 2 and/or adopt swift-subprocess and/or Foundation.Process's working directory support is made thread safe.
-        if try! Process.hasUnsafeWorkingDirectorySupport {
-            self.workingDirectory = nil
-            return
-        }
         self.workingDirectory = workingDirectory
     }
 

Sources/SWBTestSupport/Misc.swift

@@ -75,7 +75,7 @@ package func runProcessWithDeveloperDirectory(_ args: [String], workingDirectory
 package func runHostProcess(_ args: [String], workingDirectory: Path? = nil, interruptible: Bool = true, redirectStderr: Bool = true) async throws -> String {
     switch try ProcessInfo.processInfo.hostOperatingSystem() {
     case .macOS:
-        return try await InstalledXcode.currentlySelected().xcrun(args, workingDirectory: workingDirectory, redirectStderr: redirectStderr)
+        return try await InstalledXcode.currentlySelected().xcrun(args, workingDirectory: workingDirectory, interruptible: interruptible, redirectStderr: redirectStderr)
     default:
         return try await runProcess(args, workingDirectory: workingDirectory, environment: .current, interruptible: interruptible, redirectStderr: redirectStderr)
     }

Sources/SWBTestSupport/SkippedTestSupport.swift

@@ -152,11 +152,6 @@ extension Trait where Self == Testing.ConditionTrait {
         })
     }
 
-    /// Constructs a condition trait that causes a test to be disabled if the Foundation process spawning implementation is not thread-safe.
-    package static var requireThreadSafeWorkingDirectory: Self {
-        disabled(if: try Process.hasUnsafeWorkingDirectorySupport, "Foundation.Process working directory support is not thread-safe.")
-    }
-
     /// Constructs a condition trait that causes a test to be disabled if the specified llbuild API version requirement is not met.
     package static func requireLLBuild(apiVersion version: Int32) -> Self {
         let llbuildVersion = llb_get_api_version()

Sources/SWBTestSupport/Xcode.swift

@@ -31,8 +31,8 @@ package struct InstalledXcode: Sendable {
         return try await Path(xcrun(["-f", tool] + toolchainArgs).trimmingCharacters(in: .whitespacesAndNewlines))
     }
 
-    package func xcrun(_ args: [String], workingDirectory: Path? = nil, redirectStderr: Bool = true) async throws -> String {
-        return try await runProcessWithDeveloperDirectory(["/usr/bin/xcrun"] + args, workingDirectory: workingDirectory, overrideDeveloperDirectory: self.developerDirPath.str, redirectStderr: redirectStderr)
+    package func xcrun(_ args: [String], workingDirectory: Path? = nil, interruptible: Bool = true, redirectStderr: Bool = true) async throws -> String {
+        return try await runProcessWithDeveloperDirectory(["/usr/bin/xcrun"] + args, workingDirectory: workingDirectory, overrideDeveloperDirectory: self.developerDirPath.str, interruptible: interruptible, redirectStderr: redirectStderr)
     }
 
     package func productBuildVersion() throws -> ProductBuildVersion {

Sources/SWBUtil/CMakeLists.txt

@@ -77,6 +77,7 @@ add_library(SWBUtil
   POSIX.swift
   Process+Async.swift
   Process.swift
+  ProcessController.swift
   ProcessInfo.swift
   Promise.swift
   PropertyList.swift
@@ -113,6 +114,7 @@ target_link_libraries(SWBUtil PUBLIC
   SWBCSupport
   SWBLibc
   ArgumentParser
+  SwiftSubprocess::Subprocess
   $<$<NOT:$<PLATFORM_ID:Darwin>>:SwiftSystem::SystemPackage>)
 
 set_target_properties(SWBUtil PROPERTIES

Sources/SWBUtil/Process+Async.swift

@@ -90,7 +90,7 @@ extension Process {
     /// - note: This method sets the process's termination handler, if one is set.
     /// - throws: ``CancellationError`` if the task was cancelled. Applies only when `interruptible` is true.
     /// - throws: Rethrows the error from ``Process/run`` if the task could not be launched.
-    public func run(interruptible: Bool = true) async throws {
+    public func run(interruptible: Bool = true, onStarted: () -> () = { }) async throws {
         @Sendable func cancelIfRunning() {
             // Only send the termination signal if the process is already running.
             // We might have created the termination monitoring continuation at this
@@ -115,6 +115,7 @@ extension Process {
                     }
 
                     try run()
+                    onStarted()
                 } catch {
                     terminationHandler = nil
 

Sources/SWBUtil/Process.swift

@@ -11,18 +11,21 @@
 //===----------------------------------------------------------------------===//
 
 public import Foundation
-import SWBLibc
+public import SWBLibc
+import Synchronization
 
-#if os(Windows)
-public typealias pid_t = Int32
+#if canImport(Subprocess)
+import Subprocess
 #endif
 
-#if !canImport(Darwin)
-extension ProcessInfo {
-    public var isMacCatalystApp: Bool {
-        false
-    }
-}
+#if canImport(System)
+public import System
+#else
+public import SystemPackage
+#endif
+
+#if os(Windows)
+public typealias pid_t = Int32
 #endif
 
 #if (!canImport(Foundation.NSTask) || targetEnvironment(macCatalyst)) && canImport(Darwin)
@@ -64,7 +67,7 @@ public typealias Process = Foundation.Process
 #endif
 
 extension Process {
-    public static var hasUnsafeWorkingDirectorySupport: Bool {
+    fileprivate static var hasUnsafeWorkingDirectorySupport: Bool {
         get throws {
             switch try ProcessInfo.processInfo.hostOperatingSystem() {
             case .linux:
@@ -81,6 +84,36 @@ extension Process {
 
 extension Process {
     public static func getOutput(url: URL, arguments: [String], currentDirectoryURL: URL? = nil, environment: Environment? = nil, interruptible: Bool = true) async throws -> Processes.ExecutionResult {
+        #if canImport(Subprocess)
+        #if !canImport(Darwin) || os(macOS)
+        let result = try await Subprocess.run(.path(FilePath(url.filePath.str)), arguments: .init(arguments), environment: environment.map { .custom(.init($0)) } ?? .inherit, workingDirectory: (currentDirectoryURL?.filePath.str).map { FilePath($0) } ?? nil, body: { execution, inputWriter, outputReader, errorReader in
+            try await inputWriter.finish()
+            let cancellationPromise = Promise<Bool, Never>()
+            return try await withTaskCancellationHandler {
+                async let cancellationListener: () = {
+                    if await cancellationPromise.value, interruptible {
+                        try execution.send(signal: .terminate)
+                    }
+                }()
+                async let stdoutBytesAsync = outputReader.collect().flatMap { $0.withUnsafeBytes(Array.init) }
+                async let stderrBytesAsync = errorReader.collect().flatMap { $0.withUnsafeBytes(Array.init) }
+                let stdoutBytes = try await stdoutBytesAsync
+                let stderrBytes = try await stderrBytesAsync
+                cancellationPromise.fulfill(with: false)
+                try await cancellationListener
+                if interruptible {
+                    try Task.checkCancellation()
+                }
+                return (stdoutBytes, stderrBytes)
+            } onCancel: {
+                cancellationPromise.fulfill(with: true)
+            }
+        })
+        return Processes.ExecutionResult(exitStatus: .init(result.terminationStatus), stdout: Data(result.value.0), stderr: Data(result.value.1))
+        #else
+        throw StubError.error("Process spawning is unavailable")
+        #endif
+        #else
         if #available(macOS 15, iOS 18, tvOS 18, watchOS 11, visionOS 2, *) {
             // Extend the lifetime of the pipes to avoid file descriptors being closed until the AsyncStream is finished being consumed.
             return try await withExtendedLifetime((Pipe(), Pipe())) { (stdoutPipe, stderrPipe) in
@@ -110,9 +143,45 @@ extension Process {
                 return Processes.ExecutionResult(exitStatus: exitStatus, stdout: Data(output.stdoutData), stderr: Data(output.stderrData))
             }
         }
+        #endif
     }
 
     public static func getMergedOutput(url: URL, arguments: [String], currentDirectoryURL: URL? = nil, environment: Environment? = nil, interruptible: Bool = true) async throws -> (exitStatus: Processes.ExitStatus, output: Data) {
+        #if canImport(Subprocess)
+        #if !canImport(Darwin) || os(macOS)
+        let (readEnd, writeEnd) = try FileDescriptor.pipe()
+        return try await readEnd.closeAfter {
+            // Direct both stdout and stderr to the same fd. Only set `closeAfterSpawningProcess` on one of the outputs so it isn't double-closed (similarly avoid using closeAfter for the same reason).
+            let result = try await Subprocess.run(.path(FilePath(url.filePath.str)), arguments: .init(arguments), environment: environment.map { .custom(.init($0)) } ?? .inherit, workingDirectory: (currentDirectoryURL?.filePath.str).map { FilePath($0) } ?? nil, output: .fileDescriptor(writeEnd, closeAfterSpawningProcess: true), error: .fileDescriptor(writeEnd, closeAfterSpawningProcess: false), body: { execution in
+                let cancellationPromise = Promise<Bool, Never>()
+                return try await withTaskCancellationHandler {
+                    async let cancellationListener: () = {
+                        if await cancellationPromise.value, interruptible {
+                            try execution.send(signal: .terminate)
+                        }
+                    }()
+                    let bytes: [UInt8]
+                    if #available(macOS 15, iOS 18, tvOS 18, watchOS 11, visionOS 2, *) {
+                        bytes = try await Array(Data(DispatchFD(fileDescriptor: readEnd).dataStream().collect()))
+                    } else {
+                        bytes = try await Array(Data(DispatchFD(fileDescriptor: readEnd)._dataStream().collect()))
+                    }
+                    cancellationPromise.fulfill(with: false)
+                    try await cancellationListener
+                    if interruptible {
+                        try Task.checkCancellation()
+                    }
+                    return bytes
+                } onCancel: {
+                    cancellationPromise.fulfill(with: true)
+                }
+            })
+            return (.init(result.terminationStatus), Data(result.value))
+        }
+        #else
+        throw StubError.error("Process spawning is unavailable")
+        #endif
+        #else
         if #available(macOS 15, iOS 18, tvOS 18, watchOS 11, visionOS 2, *) {
             // Extend the lifetime of the pipe to avoid file descriptors being closed until the AsyncStream is finished being consumed.
             return try await withExtendedLifetime(Pipe()) { pipe in
@@ -138,6 +207,7 @@ extension Process {
                 return (exitStatus: exitStatus, output: Data(output))
             }
         }
+        #endif
     }
 
     private static func _getOutput<T, U>(url: URL, arguments: [String], currentDirectoryURL: URL?, environment: Environment?, interruptible: Bool, setup: (Process) -> T, collect: (T) async throws -> U) async throws -> (exitStatus: Processes.ExitStatus, output: U) {
@@ -294,6 +364,19 @@ public enum Processes: Sendable {
     }
 }
 
+#if canImport(Subprocess)
+extension Processes.ExitStatus {
+    init(_ terminationStatus: TerminationStatus) {
+        switch terminationStatus {
+        case let .exited(code):
+            self = .exit(code)
+        case let .unhandledException(code):
+            self = .uncaughtSignal(code)
+        }
+    }
+}
+#endif
+
 extension Processes.ExitStatus {
     public init(_ process: Process) throws {
         assert(!process.isRunning)