termination handler

motivation: allow clients of SwiftPM to terminate background activities

changes:
* introduce new terminator class that is a registry for termination handlers
* use the terminator instad of ProcessSet to manage the termination of active processes (eg git, tests) and build system
* adjust call sites

TODO:
* add more tests
* integrate into dependency resolition

rdar://64900054
rdar://63723896

Author: tomerd

Sources/Basics/Terminator.swift

@@ -0,0 +1,112 @@
+/*
+ This source file is part of the Swift.org open source project
+
+ Copyright (c) 2022 Apple Inc. and the Swift project authors
+ Licensed under Apache License v2.0 with Runtime Library Exception
+
+ See http://swift.org/LICENSE.txt for license information
+ See http://swift.org/CONTRIBUTORS.txt for Swift project authors
+ */
+
+import Dispatch
+import Foundation
+import TSCBasic
+
+public typealias TerminationHandler = () throws -> Void
+
+#warning("make observabilityScope non optional")
+public class Terminator {
+    public typealias RegistrationKey = String
+
+    private let killTimeout: DispatchTimeInterval
+    private let observabilityScope: ObservabilityScope?
+
+    private let registry = ThreadSafeKeyValueStore<String, (name: String, handler: TerminationHandler)>()
+    private let terminationQueue = DispatchQueue(label: "org.swift.swiftpm.terminator")
+
+    private let cancelled = ThreadSafeBox<Bool>(false)
+
+
+    /// Create a process set.
+    public init(
+        killTimeout: DispatchTimeInterval = .seconds(30),
+        observabilityScope: ObservabilityScope?
+    ) {
+        self.killTimeout = killTimeout
+        self.observabilityScope = observabilityScope
+    }
+
+    @discardableResult
+    public func register(name: String, handler: @escaping TerminationHandler) -> RegistrationKey {
+        if self.cancelled.get() ?? false {
+            self.observabilityScope?.emit(debug: "not registering '\(name)' with terminator as its already cancelled")
+        }
+        let key = UUID().uuidString
+        self.observabilityScope?.emit(debug: "registering '\(name)' with terminator")
+        self.registry[key] = (name: name, handler: handler)
+        return key
+    }
+
+    public func register(_ process: TSCBasic.Process) -> RegistrationKey {
+        self.register(name: "\(process.arguments.joined(separator: " "))", handler: {
+            process.terminate(timeout: .now() + self.killTimeout)
+        })
+    }
+
+    public func deregister(_ key: RegistrationKey) {
+        self.registry[key] = nil
+    }
+
+    public func terminate() {
+        self.observabilityScope?.emit(info: "starting termination cycle with \(self.registry.count) termination handlers registered")
+
+        self.cancelled.put(true)
+
+        let success = ThreadSafeArrayStore<String>()
+        let group = DispatchGroup()
+        for (_, (name, handler)) in self.registry.get() {
+            group.enter()
+            self.terminationQueue.async {
+                defer { group.leave() }
+                do {
+                    self.observabilityScope?.emit(debug: "terminating '\(name)'")
+                    try handler()
+                    success.append(name)
+                } catch {
+                    self.observabilityScope?.emit(warning: "failed terminating '\(name)': \(error)")
+                }
+            }
+        }
+
+        if case .timedOut = group.wait(timeout: .now() + self.killTimeout) {
+            self.observabilityScope?.emit(warning: "timeout waiting for termination with \(self.registry.count - success.count) termination handlers remaining")
+        } else {
+            self.observabilityScope?.emit(info: "termination cycle completed successfully")
+        }
+    }
+}
+
+extension TSCBasic.Process {
+    fileprivate func terminate(timeout: DispatchTime) {
+        // send graceful shutdown signal
+        self.signal(SIGINT)
+
+        // start a thread to see if we need to terminate more forcibly
+        let forceKillSemaphore = DispatchSemaphore(value: 1)
+        let forceKillThread = TSCBasic.Thread {
+            if case .timedOut = forceKillSemaphore.wait(timeout: timeout) {
+                // send a force-kill signal
+                #if os(Windows)
+                self.signal(SIGTERM)
+                #else
+                self.signal(SIGKILL)
+                #endif
+            }
+        }
+        forceKillThread.start()
+        _ = try? self.waitUntilExit()
+        forceKillSemaphore.signal() // let the force-kill thread know we do not need it any more
+        // join the force-kill thread thread so we don't exit before everything terminates
+        forceKillThread.join()
+    }
+}

Sources/Commands/APIDigester.swift

@@ -90,7 +90,7 @@ struct APIDigesterBaselineDumper {
         }
 
         // Clone the current package in a sandbox and checkout the baseline revision.
-        let repositoryProvider = GitRepositoryProvider()
+        let repositoryProvider = GitRepositoryProvider(terminator: swiftTool.terminator)
         let specifier = RepositorySpecifier(path: baselinePackageRoot)
         let workingCopy = try repositoryProvider.createWorkingCopy(
             repository: specifier,

Sources/Commands/SwiftPackageTool.swift

@@ -394,7 +394,7 @@ extension SwiftPackageTool {
             let apiDigesterTool = SwiftAPIDigester(fileSystem: swiftTool.fileSystem, tool: apiDigesterPath)
 
             let packageRoot = try swiftOptions.packagePath ?? swiftTool.getPackageRoot()
-            let repository = GitRepository(path: packageRoot)
+            let repository = GitRepository(path: packageRoot, terminator: swiftTool.terminator)
             let baselineRevision = try repository.resolveRevision(identifier: treeish)
 
             // We turn build manifest caching off because we need the build plan.
@@ -883,7 +883,7 @@ extension SwiftPackageTool {
 
         func run(_ swiftTool: SwiftTool) throws {
             let packageRoot = try swiftOptions.packagePath ?? swiftTool.getPackageRoot()
-            let repository = GitRepository(path: packageRoot)
+            let repository = GitRepository(path: packageRoot, terminator: swiftTool.terminator)
 
             let destination: AbsolutePath
             if let output = output {
@@ -1170,22 +1170,14 @@ final class PluginDelegate: PluginInvocationDelegate {
 
         // Create a build operation. We have to disable the cache in order to get a build plan created.
         let outputStream = BufferedOutputByteStream()
-        let buildOperation = BuildOperation(
-            buildParameters: buildParameters,
+        let buildOperation = try swiftTool.createBuildOperation(
+            explicitProduct: explicitProduct,
             cacheBuildManifest: false,
-            packageGraphLoader: { try self.swiftTool.loadPackageGraph(explicitProduct: explicitProduct) },
-            pluginScriptRunner: try self.swiftTool.getPluginScriptRunner(),
-            pluginWorkDirectory: try self.swiftTool.getActiveWorkspace().location.pluginWorkingDirectory,
-            outputStream: outputStream,
-            logLevel: logLevel,
-            fileSystem: swiftTool.fileSystem,
-            observabilityScope: self.swiftTool.observabilityScope
+            customBuildParameters: buildParameters,
+            customOutputStream: outputStream,
+            customLogLevel: logLevel
         )
-
-        // Save the instance so it can be canceled from the interrupt handler.
-        self.swiftTool.buildSystemRef.buildSystem = buildOperation
-
-        // Get or create the build description and plan the build.
+ 
         let _ = try buildOperation.getBuildDescription()
         let buildPlan = buildOperation.buildPlan!
 
@@ -1280,7 +1272,7 @@ final class PluginDelegate: PluginInvocationDelegate {
                         let testRunner = TestRunner(
                             bundlePaths: [testProduct.bundlePath],
                             xctestArg: testSpecifier,
-                            processSet: swiftTool.processSet,
+                            terminator: swiftTool.terminator,
                             toolchain: toolchain,
                             testEnv: testEnvironment,
                             observabilityScope: swiftTool.observabilityScope)
@@ -1555,15 +1547,16 @@ extension SwiftPackageTool {
                 dstdir = try swiftTool.getPackageRoot()
                 projectName = graph.rootPackages[0].manifest.displayName // TODO: use identity instead?
             }
-            let xcodeprojPath = Xcodeproj.buildXcodeprojPath(outputDir: dstdir, projectName: projectName)
+            let xcodeprojPath = XcodeProject.makePath(outputDir: dstdir, projectName: projectName)
 
             var genOptions = xcodeprojOptions()
             genOptions.manifestLoader = try swiftTool.getManifestLoader()
 
-            try Xcodeproj.generate(
+            try XcodeProject.generate(
                 projectName: projectName,
                 xcodeprojPath: xcodeprojPath,
                 graph: graph,
+                repositoryProvider: GitRepositoryProvider(terminator: swiftTool.terminator),
                 options: genOptions,
                 fileSystem: swiftTool.fileSystem,
                 observabilityScope: swiftTool.observabilityScope

Sources/Commands/SwiftTestTool.swift

@@ -283,7 +283,7 @@ public struct SwiftTestTool: SwiftCommand {
             let runner = TestRunner(
                 bundlePaths: testProducts.map { $0.bundlePath },
                 xctestArg: xctestArg,
-                processSet: swiftTool.processSet,
+                terminator: swiftTool.terminator,
                 toolchain: toolchain,
                 testEnv: testEnv,
                 observabilityScope: swiftTool.observabilityScope
@@ -327,7 +327,7 @@ public struct SwiftTestTool: SwiftCommand {
             // Run the tests using the parallel runner.
             let runner = ParallelTestRunner(
                 bundlePaths: testProducts.map { $0.bundlePath },
-                processSet: swiftTool.processSet,
+                terminator: swiftTool.terminator,
                 toolchain: toolchain,
                 numJobs: options.numberOfWorkers ?? ProcessInfo.processInfo.activeProcessorCount,
                 options: swiftOptions,
@@ -514,7 +514,7 @@ final class TestRunner {
     /// Arguments to pass to XCTest if any.
     private let xctestArg: String?
 
-    private let processSet: ProcessSet
+    private let terminator: Terminator
 
     // The toolchain to use.
     private let toolchain: UserToolchain
@@ -532,14 +532,14 @@ final class TestRunner {
     init(
         bundlePaths: [AbsolutePath],
         xctestArg: String? = nil,
-        processSet: ProcessSet,
+        terminator: Terminator,
         toolchain: UserToolchain,
         testEnv: [String: String],
         observabilityScope: ObservabilityScope
     ) {
         self.bundlePaths = bundlePaths
         self.xctestArg = xctestArg
-        self.processSet = processSet
+        self.terminator = terminator
         self.toolchain = toolchain
         self.testEnv = testEnv
         self.observabilityScope = observabilityScope.makeChildScope(description: "Test Runner")
@@ -591,7 +591,8 @@ final class TestRunner {
                 stderr: outputHandler
             )
             let process = Process(arguments: try args(forTestAt: path), environment: self.testEnv, outputRedirection: outputRedirection)
-            try self.processSet.add(process)
+            let terminationKey = self.terminator.register(process)
+            defer { self.terminator.deregister(terminationKey) }
             try process.launch()
             let result = try process.waitUntilExit()
             switch result.exitStatus {
@@ -605,8 +606,6 @@ final class TestRunner {
             default:
                 return false
             }
-        } catch ProcessSetError.cancelled {
-            return false
         } catch {
             testObservabilityScope.emit(error)
             return false
@@ -644,7 +643,7 @@ final class ParallelTestRunner {
     /// True if all tests executed successfully.
     private(set) var ranSuccessfully = true
 
-    private let processSet: ProcessSet
+    private let terminator: Terminator
 
     private let toolchain: UserToolchain
 
@@ -662,7 +661,7 @@ final class ParallelTestRunner {
 
     init(
         bundlePaths: [AbsolutePath],
-        processSet: ProcessSet,
+        terminator: Terminator,
         toolchain: UserToolchain,
         numJobs: Int,
         options: SwiftToolOptions,
@@ -671,7 +670,7 @@ final class ParallelTestRunner {
         observabilityScope: ObservabilityScope
     ) {
         self.bundlePaths = bundlePaths
-        self.processSet = processSet
+        self.terminator = terminator
         self.toolchain = toolchain
         self.numJobs = numJobs
         self.shouldOutputSuccess = shouldOutputSuccess
@@ -727,7 +726,7 @@ final class ParallelTestRunner {
                     let testRunner = TestRunner(
                         bundlePaths: [test.productPath],
                         xctestArg: test.specifier,
-                        processSet: self.processSet,
+                        terminator: self.terminator,
                         toolchain: self.toolchain,
                         testEnv: testEnv,
                         observabilityScope: self.observabilityScope