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
* add cancel method to GitRepositoryProvider protocol so that it can be interuppted
* add cancel methods to RepositoryManager, RegistryDownloadManager and HTTPClient so that it can be interuppted
* change workspace initializer to take a terminator so that CLI and other consumer of libSwiftPM can interrupt
* register intrupption points mentioned above in workspace
* adjust related call sites, ie CLI signal handler now uses the workspace terminator to interuupt

TODO:
* add more tests
* integrate into dependency resolition

rdar://64900054
rdar://63723896

Author: tomerd

Sources/Basics/CMakeLists.txt

@@ -28,6 +28,7 @@ add_library(Basics
   Triple+Extensions.swift
   SwiftVersion.swift
   SQLiteBackedCache.swift
+  Terminator.swift
   Version+Extensions.swift)
 target_link_libraries(Basics PUBLIC
   SwiftCollections::OrderedCollections

Sources/Basics/ConcurrencyHelpers.swift

@@ -194,6 +194,12 @@ public final class ThreadSafeBox<Value> {
         }
     }
 
+    public func get(`default`: Value) -> Value {
+        self.lock.withLock {
+            self.underlying ?? `default`
+        }
+    }
+
     public func put(_ newValue: Value) {
         self.lock.withLock {
             self.underlying = newValue

Sources/Basics/HTTPClient.swift

@@ -45,6 +45,8 @@ public struct HTTPClient {
     public var configuration: HTTPClientConfiguration
     private let underlying: Handler
 
+    private let cancelled = ThreadSafeBox<Bool>(false)
+
     // static to share across instances of the http client
     private static var hostsErrorsLock = Lock()
     private static var hostsErrors = [String: [Date]]()
@@ -113,6 +115,11 @@ public struct HTTPClient {
         )
     }
 
+    /// Cancel any outstanding requests
+    public func cancel() {
+        self.cancelled.put(true)
+    }
+
     private func _execute(request: Request, requestNumber: Int, observabilityScope: ObservabilityScope?, progress: ProgressHandler?, completion: @escaping CompletionHandler) {
         if self.shouldCircuitBreak(request: request) {
             observabilityScope?.emit(warning: "Circuit breaker triggered for \(request.url)")
@@ -128,6 +135,9 @@ public struct HTTPClient {
                         return completion(.failure(HTTPClientError.responseTooLarge(received)))
                     }
                 }
+                guard !self.cancelled.get(default: false) else {
+                    return completion(.failure(TerminationError()))
+                }
                 progress?(received, expected)
             },
             { result in

Sources/Basics/Terminator.swift

@@ -0,0 +1,113 @@
+/*
+ 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 = (DispatchTime) throws -> Void
+
+public class Terminator {
+    public typealias RegistrationKey = String
+
+    private let observabilityScope: ObservabilityScope?
+    private let registry = ThreadSafeKeyValueStore<String, (name: String, handler: TerminationHandler)>()
+    private let terminationQueue = DispatchQueue(label: "org.swift.swiftpm.terminator")
+    private let terminated = ThreadSafeBox<Bool>(false)
+
+    public init(observabilityScope: ObservabilityScope?) {
+        self.observabilityScope = observabilityScope
+    }
+
+    @discardableResult
+    public func register(name: String, handler: @escaping TerminationHandler) -> RegistrationKey? {
+        if self.terminated.get() ?? false {
+            self.observabilityScope?.emit(debug: "not registering '\(name)' with terminator as its already cancelled")
+            return .none
+        }
+        let key = UUID().uuidString
+        self.observabilityScope?.emit(debug: "registering '\(name)' with terminator")
+        self.registry[key] = (name: name, handler: handler)
+        return key
+    }
+
+    @discardableResult
+    public func register(name: String, handler: @escaping () throws -> Void) -> RegistrationKey? {
+        self.register(name: name, handler:  { _ in try handler() })
+    }
+
+    public func register(_ process: TSCBasic.Process) -> RegistrationKey? {
+        self.register(name: "\(process.arguments.joined(separator: " "))", handler:  process.terminate)
+    }
+
+    public func deregister(_ key: RegistrationKey) {
+        self.registry[key] = nil
+    }
+
+    public func terminate(deadline: DispatchTime? = .none) {
+        self.terminated.put(true)
+
+        self.observabilityScope?.emit(info: "starting termination cycle with \(self.registry.count) termination handlers registered")
+
+        let deadline = deadline ?? .now() + .seconds(30)
+        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(deadline)
+                    success.append(name)
+                } catch {
+                    self.observabilityScope?.emit(warning: "failed terminating '\(name)': \(error)")
+                }
+            }
+        }
+
+        if case .timedOut = group.wait(timeout: deadline) {
+            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")
+        }
+    }
+}
+
+public struct TerminationError: Error, CustomStringConvertible {
+    public let description = "program termination in progress"
+
+    public init() {}
+}
+
+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

@@ -102,7 +102,10 @@ struct APIDigesterBaselineDumper {
         try workingCopy.checkout(revision: baselineRevision)
 
         // Create the workspace for this package.
-        let workspace = try Workspace(forRootPackage: baselinePackageRoot)
+        let workspace = try Workspace(
+            forRootPackage: baselinePackageRoot,
+            terminator: swiftTool.terminator
+        )
 
         let graph = try workspace.loadPackageGraph(
             rootPath: baselinePackageRoot,

Sources/Commands/SwiftPackageTool.swift

@@ -1170,21 +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 +1273,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)
@@ -1564,6 +1557,7 @@ extension SwiftPackageTool {
                 projectName: projectName,
                 xcodeprojPath: xcodeprojPath,
                 graph: graph,
+                repositoryProvider: GitRepositoryProvider(),
                 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,10 @@ final class TestRunner {
                 stderr: outputHandler
             )
             let process = Process(arguments: try args(forTestAt: path), environment: self.testEnv, outputRedirection: outputRedirection)
-            try self.processSet.add(process)
+            guard let terminationKey = self.terminator.register(process) else {
+                return false // terminating
+            }
+            defer { self.terminator.deregister(terminationKey) }
             try process.launch()
             let result = try process.waitUntilExit()
             switch result.exitStatus {
@@ -605,8 +608,6 @@ final class TestRunner {
             default:
                 return false
             }
-        } catch ProcessSetError.cancelled {
-            return false
         } catch {
             testObservabilityScope.emit(error)
             return false
@@ -644,7 +645,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 +663,7 @@ final class ParallelTestRunner {
 
     init(
         bundlePaths: [AbsolutePath],
-        processSet: ProcessSet,
+        terminator: Terminator,
         toolchain: UserToolchain,
         numJobs: Int,
         options: SwiftToolOptions,
@@ -671,7 +672,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 +728,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