Adopt swift-subprocess when running unit tests

Author: owenv

CMakeLists.txt

@@ -47,6 +47,7 @@ if(FIND_PM_DEPS)
   find_package(SwiftCertificates CONFIG REQUIRED)
   find_package(SwiftCrypto CONFIG REQUIRED)
   find_package(SwiftBuild CONFIG REQUIRED)
+  find_package(Subprocess CONFIG REQUIRED)
 endif()
 
 find_package(dispatch QUIET)

Package.swift

@@ -575,6 +575,7 @@ let package = Package(
             dependencies: [
                 .product(name: "ArgumentParser", package: "swift-argument-parser"),
                 .product(name: "OrderedCollections", package: "swift-collections"),
+                .product(name: "Subprocess", package: "swift-subprocess"),
                 "Basics",
                 "BinarySymbols",
                 "Build",
@@ -1113,6 +1114,7 @@ if ProcessInfo.processInfo.environment["SWIFTCI_USE_LOCAL_DEPS"] == nil {
         .package(url: "https://github.com/apple/swift-system.git", from: "1.1.1"),
         .package(url: "https://github.com/apple/swift-collections.git", "1.0.1" ..< "1.2.0"),
         .package(url: "https://github.com/apple/swift-certificates.git", "1.0.1" ..< "1.6.0"),
+        .package(url: "https://github.com/swiftlang/swift-subprocess.git", .upToNextMinor(from: "0.2.0")),
         .package(url: "https://github.com/swiftlang/swift-toolchain-sqlite.git", from: "1.0.0"),
         // For use in previewing documentation
         .package(url: "https://github.com/swiftlang/swift-docc-plugin", from: "1.1.0"),
@@ -1131,6 +1133,7 @@ if ProcessInfo.processInfo.environment["SWIFTCI_USE_LOCAL_DEPS"] == nil {
         .package(path: "../swift-system"),
         .package(path: "../swift-collections"),
         .package(path: "../swift-certificates"),
+        .package(path: "../swift-subprocess"),
         .package(path: "../swift-toolchain-sqlite"),
     ]
     if !swiftDriverDeps.isEmpty {

Sources/Commands/CMakeLists.txt

@@ -61,6 +61,7 @@ add_library(Commands
 target_link_libraries(Commands PUBLIC
   SwiftCollections::OrderedCollections
   SwiftSyntax::SwiftRefactor
+  Subprocess::Subprocess
   ArgumentParser
   Basics
   BinarySymbols

Sources/Commands/SwiftTestCommand.swift

@@ -11,6 +11,12 @@
 //===----------------------------------------------------------------------===//
 
 import ArgumentParser
+import Subprocess
+#if canImport(System)
+import System
+#else
+import SystemPackage
+#endif
 
 @_spi(SwiftPMInternal)
 import Basics
@@ -342,7 +348,7 @@ public struct SwiftTestCommand: AsyncSwiftCommand {
                         observabilityScope: swiftCommandState.observabilityScope
                     )
 
-                    testResults = try runner.run(tests)
+                    testResults = try await runner.run(tests)
                     result = runner.ranSuccessfully ? .success : .failure
                 }
 
@@ -538,7 +544,7 @@ public struct SwiftTestCommand: AsyncSwiftCommand {
         )
 
         // Finally, run the tests.
-        return runner.test(outputHandler: {
+        return await runner.test(outputHandler: {
             // command's result output goes on stdout
             // ie "swift test" should output to stdout
             print($0, terminator: "")
@@ -837,7 +843,7 @@ extension SwiftTestCommand {
                     )
 
                     // Finally, run the tests.
-                    let result = runner.test(outputHandler: {
+                    let result = await runner.test(outputHandler: {
                         // command's result output goes on stdout
                         // ie "swift test" should output to stdout
                         print($0, terminator: "")
@@ -911,7 +917,7 @@ final class TestRunner {
     // The toolchain to use.
     private let toolchain: UserToolchain
 
-    private let testEnv: Environment
+    private let testEnv: Basics.Environment
 
     /// ObservabilityScope  to emit diagnostics.
     private let observabilityScope: ObservabilityScope
@@ -945,7 +951,7 @@ final class TestRunner {
         additionalArguments: [String],
         cancellator: Cancellator,
         toolchain: UserToolchain,
-        testEnv: Environment,
+        testEnv: Basics.Environment,
         observabilityScope: ObservabilityScope,
         library: TestingLibrary
     ) {
@@ -974,10 +980,10 @@ final class TestRunner {
 
     /// Executes and returns execution status. Prints test output on standard streams if requested
     /// - Returns: Result of spawning and running the test process, and the output stream result
-    func test(outputHandler: @escaping (String) -> Void) -> Result {
+    func test(outputHandler: @escaping (String) -> Void) async -> Result {
         var results = [Result]()
         for path in self.bundlePaths {
-            let testSuccess = self.test(at: path, outputHandler: outputHandler)
+            let testSuccess = await self.test(at: path, outputHandler: outputHandler)
             results.append(testSuccess)
         }
         return results.reduce()
@@ -1021,33 +1027,36 @@ final class TestRunner {
         return args
     }
 
-    private func test(at path: AbsolutePath, outputHandler: @escaping (String) -> Void) -> Result {
+    private func test(at path: AbsolutePath, outputHandler: @escaping (String) -> Void) async -> Result {
         let testObservabilityScope = self.observabilityScope.makeChildScope(description: "running test at \(path)")
 
         do {
-            let outputHandler = { (bytes: [UInt8]) in
-                if let output = String(bytes: bytes, encoding: .utf8) {
-                    outputHandler(output)
-                }
+            let args = try args(forTestAt: path)
+            var env: [Subprocess.Environment.Key: String] = [:]
+            for (key, value) in self.testEnv {
+                env[.init(rawValue: key.rawValue)!] = value
             }
-            let outputRedirection = AsyncProcess.OutputRedirection.stream(
-                stdout: outputHandler,
-                stderr: outputHandler
-            )
-            let process = AsyncProcess(arguments: try args(forTestAt: path), environment: self.testEnv, outputRedirection: outputRedirection)
-            guard let terminationKey = self.cancellator.register(process) else {
-                return .failure // terminating
+            let processConfig = Subprocess.Configuration(.path(.init(args[0])), arguments: .init(Array(args.dropFirst())), environment: .custom(env))
+            let result = try await Subprocess.run(processConfig, input: .none, error: .standardOutput) { execution, outputSequence in
+                let token = cancellator.register(name: "Test Execution", handler: { _ in
+                    await execution.teardown(using: [.gracefulShutDown(allowedDurationToNextStep: .seconds(5))])
+                })
+                defer {
+                    token.map { cancellator.deregister($0) }
+                }
+
+                for try await line in outputSequence.lines() {
+                    outputHandler(line)
+                }
             }
-            defer { self.cancellator.deregister(terminationKey) }
-            try process.launch()
-            let result = try process.waitUntilExit()
-            switch result.exitStatus {
-            case .terminated(code: 0):
+
+            switch result.terminationStatus {
+            case .exited(code: 0):
                 return .success
-            case .terminated(code: EXIT_NO_TESTS_FOUND) where library == .swiftTesting:
+            case .exited(code: numericCast(EXIT_NO_TESTS_FOUND)) where library == .swiftTesting:
                 return .noMatchingTests
             #if !os(Windows)
-            case .signalled(let signal) where ![SIGINT, SIGKILL, SIGTERM].contains(signal):
+            case .unhandledException(let signal) where ![SIGINT, SIGKILL, SIGTERM].contains(signal):
                 testObservabilityScope.emit(error: "Exited with unexpected signal code \(signal)")
                 return .failure
             #endif
@@ -1087,21 +1096,9 @@ final class ParallelTestRunner {
     /// Path to XCTest binaries.
     private let bundlePaths: [AbsolutePath]
 
-    /// The queue containing list of tests to run (producer).
-    private let pendingTests = SynchronizedQueue<UnitTest?>()
-
-    /// The queue containing tests which are finished running.
-    private let finishedTests = SynchronizedQueue<TestResult?>()
-
     /// Instance of a terminal progress animation.
     private let progressAnimation: ProgressAnimationProtocol
 
-    /// Number of tests that will be executed.
-    private var numTests = 0
-
-    /// Number of the current tests that has been executed.
-    private var numCurrentTest = 0
-
     /// True if all tests executed successfully.
     private(set) var ranSuccessfully = true
 
@@ -1160,27 +1157,8 @@ final class ParallelTestRunner {
         assert(numJobs > 0, "num jobs should be > 0")
     }
 
-    /// Updates the progress bar status.
-    private func updateProgress(for test: UnitTest) {
-        numCurrentTest += 1
-        progressAnimation.update(step: numCurrentTest, total: numTests, text: "Testing \(test.specifier)")
-    }
-
-    private func enqueueTests(_ tests: [UnitTest]) throws {
-        // Enqueue all the tests.
-        for test in tests {
-            pendingTests.enqueue(test)
-        }
-        self.numTests = tests.count
-        self.numCurrentTest = 0
-        // Enqueue the sentinels, we stop a thread when it encounters a sentinel in the queue.
-        for _ in 0..<numJobs {
-            pendingTests.enqueue(nil)
-        }
-    }
-
     /// Executes the tests spawning parallel workers. Blocks calling thread until all workers are finished.
-    func run(_ tests: [UnitTest]) throws -> [TestResult] {
+    func run(_ tests: [UnitTest]) async throws -> [TestResult] {
         assert(!tests.isEmpty, "There should be at least one test to execute.")
 
         let testEnv = try TestingSupport.constructTestEnvironment(
@@ -1190,77 +1168,75 @@ final class ParallelTestRunner {
             library: .xctest // swift-testing does not use ParallelTestRunner
         )
 
-        // Enqueue all the tests.
-        try enqueueTests(tests)
-
-        // Create the worker threads.
-        let workers: [Thread] = (0..<numJobs).map({ _ in
-            let thread = Thread {
-                // Dequeue a specifier and run it till we encounter nil.
-                while let test = self.pendingTests.dequeue() {
-                    let additionalArguments = TestRunner.xctestArguments(forTestSpecifiers: CollectionOfOne(test.specifier))
-                    let testRunner = TestRunner(
-                        bundlePaths: [test.productPath],
-                        additionalArguments: additionalArguments,
-                        cancellator: self.cancellator,
-                        toolchain: self.toolchain,
-                        testEnv: testEnv,
-                        observabilityScope: self.observabilityScope,
-                        library: .xctest // swift-testing does not use ParallelTestRunner
-                    )
-                    var output = ""
-                    let outputLock = NSLock()
-                    let start = DispatchTime.now()
-                    let result = testRunner.test(outputHandler: { _output in outputLock.withLock{ output += _output }})
-                    let duration = start.distance(to: .now())
-                    if result == .failure {
-                        self.ranSuccessfully = false
+        var pendingTests: [UnitTest] = tests
+        var processedTests: [TestResult] = []
+        observabilityScope.emit(error: "wefhfehjef")
+        await withTaskGroup { group in
+            func runTest(_ test: UnitTest) async -> TestResult {
+                observabilityScope.emit(error: "enqueuing \(test.specifier)")
+                let additionalArguments = TestRunner.xctestArguments(forTestSpecifiers: CollectionOfOne(test.specifier))
+                let testRunner = TestRunner(
+                    bundlePaths: [test.productPath],
+                    additionalArguments: additionalArguments,
+                    cancellator: self.cancellator,
+                    toolchain: self.toolchain,
+                    testEnv: testEnv,
+                    observabilityScope: self.observabilityScope,
+                    library: .xctest // swift-testing does not use ParallelTestRunner
+                )
+                var output = ""
+                let outputLock = NSLock()
+                let start = DispatchTime.now()
+                let result = await testRunner.test(outputHandler: { _output in outputLock.withLock{ output += _output }})
+                let duration = start.distance(to: .now())
+                if result == .failure {
+                    self.ranSuccessfully = false
+                }
+                return TestResult(
+                    unitTest: test,
+                    output: output,
+                    success: result != .failure,
+                    duration: duration
+                )
+            }
+            for _ in 0..<numJobs {
+                if !pendingTests.isEmpty {
+                    let test = pendingTests.removeLast()
+                    group.addTask {
+                        await runTest(test)
                     }
-                    self.finishedTests.enqueue(TestResult(
-                        unitTest: test,
-                        output: output,
-                        success: result != .failure,
-                        duration: duration
-                    ))
                 }
             }
-            thread.start()
-            return thread
-        })
-
-        // List of processed tests.
-        let processedTests = ThreadSafeArrayStore<TestResult>()
-
-        // Report (consume) the tests which have finished running.
-        while let result = finishedTests.dequeue() {
-            updateProgress(for: result.unitTest)
-
-            // Store the result.
-            processedTests.append(result)
+            var completedTests = 0
+            while let result = await group.next() {
+                completedTests += 1
+                progressAnimation.update(step: completedTests, total: tests.count, text: "Testing \(result.unitTest.specifier)")
+
+                if !pendingTests.isEmpty {
+                    let test = pendingTests.removeLast()
+                    group.addTask(operation: {
+                        await runTest(test)
+                    })
+                }
 
-            // We can't enqueue a sentinel into finished tests queue because we won't know
-            // which test is last one so exit this when all the tests have finished running.
-            if numCurrentTest == numTests {
-                break
+                // Store the result.
+                processedTests.append(result)
             }
         }
 
-        // Wait till all threads finish execution.
-        workers.forEach { $0.join() }
-
         // Report the completion.
-        progressAnimation.complete(success: processedTests.get().contains(where: { !$0.success }))
+        progressAnimation.complete(success: processedTests.contains(where: { !$0.success }))
 
         // Print test results.
-        for test in processedTests.get() {
+        for test in processedTests {
             if (!test.success || shouldOutputSuccess) && !productsBuildParameters.testingParameters.experimentalTestOutput {
                 // command's result output goes on stdout
                 // ie "swift test" should output to stdout
                 print(test.output)
             }
         }
 
-        return processedTests.get()
+        return processedTests
     }
 }
 

Sources/Commands/Utilities/PluginDelegate.swift

@@ -288,7 +288,7 @@ final class PluginDelegate: PluginInvocationDelegate {
 
                         // Run the test — for now we run the sequentially so we can capture accurate timing results.
                         let startTime = DispatchTime.now()
-                        let result = testRunner.test(outputHandler: { _ in }) // this drops the tests output
+                        let result = await testRunner.test(outputHandler: { _ in }) // this drops the tests output
                         let duration = Double(startTime.distance(to: .now()).milliseconds() ?? 0) / 1000.0
                         numFailedTests += (result != .failure) ? 0 : 1
                         testResults.append(