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
* add tests

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/HTTPClient.swift

@@ -15,6 +15,7 @@ import class Foundation.JSONDecoder
 import class Foundation.NSError
 import class Foundation.OperationQueue
 import struct Foundation.URL
+import struct Foundation.UUID
 import TSCBasic
 
 #if canImport(Glibc)
@@ -35,7 +36,7 @@ public enum HTTPClientError: Error, Equatable {
 
 // MARK: - HTTPClient
 
-public struct HTTPClient {
+public struct HTTPClient: Cancellable {
     public typealias Configuration = HTTPClientConfiguration
     public typealias Request = HTTPClientRequest
     public typealias Response = HTTPClientResponse
@@ -47,9 +48,12 @@ public struct HTTPClient {
     private let underlying: Handler
 
     /// DispatchSemaphore to restrict concurrent operations on manager.
-     private let concurrencySemaphore: DispatchSemaphore
-     /// OperationQueue to park pending requests
-     private let requestsQueue: OperationQueue
+    private let concurrencySemaphore: DispatchSemaphore
+    /// OperationQueue to park pending requests
+    private let requestsQueue: OperationQueue
+
+    // tracks outstanding requests for cancellation
+    private var outstandingRequests = ThreadSafeKeyValueStore<UUID, (completion: CompletionHandler, progress: ProgressHandler?, queue: DispatchQueue)>()
 
     // static to share across instances of the http client
     private static var hostsErrorsLock = Lock()
@@ -76,7 +80,12 @@ public struct HTTPClient {
     ///   - observabilityScope: the observability scope to emit diagnostics on
     ///   - progress: A progress handler to handle progress for example for downloads
     ///   - completion: A completion handler to be notified of the completion of the request.
-    public func execute(_ request: Request, observabilityScope: ObservabilityScope? = nil, progress: ProgressHandler? = nil, completion: @escaping CompletionHandler) {
+    public func execute(
+        _ request: Request,
+        observabilityScope: ObservabilityScope? = nil,
+        progress: ProgressHandler? = nil,
+        completion: @escaping CompletionHandler
+    ) {
         // merge configuration
         var request = request
         if request.options.callbackQueue == nil {
@@ -107,7 +116,9 @@ public struct HTTPClient {
             request.headers.add(name: "Authorization", value: authorization)
         }
         // execute
-        let callbackQueue = request.options.callbackQueue ?? self.configuration.callbackQueue
+        guard let callbackQueue = request.options.callbackQueue else {
+            return completion(.failure(InternalError("unknown callback queue")))
+        }
         self._execute(
             request: request,
             requestNumber: 0,
@@ -129,13 +140,44 @@ public struct HTTPClient {
         )
     }
 
-    private func _execute(request: Request, requestNumber: Int, observabilityScope: ObservabilityScope?, progress: ProgressHandler?, completion: @escaping CompletionHandler) {
+    /// Cancel any outstanding requests
+    public func cancel(deadline: DispatchTime) throws {
+        let outstanding = self.outstandingRequests.get()
+        self.outstandingRequests.clear()
+        for (completion, _, queue) in outstanding.values {
+            queue.async {
+                completion(.failure(CancellationError()))
+            }
+        }
+    }
+
+    private func _execute(
+        request: Request,
+        requestNumber: Int,
+        observabilityScope: ObservabilityScope?,
+        progress: ProgressHandler?,
+        completion: @escaping CompletionHandler
+    ) {
+        // records outstanding requests for cancellation purposes
+        guard let callbackQueue = request.options.callbackQueue else {
+            return completion(.failure(InternalError("unknown callback queue")))
+        }
+        let requestKey = UUID()
+        self.outstandingRequests[requestKey] = (completion: completion, progress: progress, queue: callbackQueue)
+
         // wrap completion handler with concurrency control cleanup
         let originalCompletion = completion
         let completion: CompletionHandler = { result in
             // free concurrency control semaphore
             self.concurrencySemaphore.signal()
-            originalCompletion(result)
+            // cancellation support
+            // if the callback is no longer on the pending lists it has been canceled already
+            if let (callback, _, queue) = self.outstandingRequests[requestKey] {
+                // remove from outstanding requests
+                self.outstandingRequests[requestKey] = nil
+                // call back on the request queue
+                queue.async { callback(result) }
+            }
         }
 
         // we must not block the calling thread (for concurrency control) so nesting this in a queue
@@ -172,9 +214,11 @@ public struct HTTPClient {
                         // handle retry strategy
                         if let retryDelay = self.shouldRetry(response: response, request: request, requestNumber: requestNumber) {
                             observabilityScope?.emit(warning: "\(request.url) failed, retrying in \(retryDelay)")
-                            // free concurrency control semaphore, since we re-submitting the request with the original completion handler
-                            // using the wrapped completion handler may lead to starving the mac concurrent requests
+                            // free concurrency control semaphore and outstanding request,
+                            // since we re-submitting the request with the original completion handler
+                            // using the wrapped completion handler may lead to starving the max concurrent requests
                             self.concurrencySemaphore.signal()
+                            self.outstandingRequests[requestKey] = nil
                             // TODO: dedicated retry queue?
                             return self.configuration.callbackQueue.asyncAfter(deadline: .now() + retryDelay) {
                                 self._execute(request: request, requestNumber: requestNumber + 1, observabilityScope: observabilityScope, progress: progress, completion: originalCompletion)

Sources/Basics/Terminator.swift

@@ -0,0 +1,133 @@
+/*
+ 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: Cancellable {
+    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", qos: .userInteractive, attributes: .concurrent)
+    private let terminating = ThreadSafeBox<Bool>(false)
+
+    public init(observabilityScope: ObservabilityScope?) {
+        self.observabilityScope = observabilityScope
+    }
+
+    @discardableResult
+    public func register(name: String, handler: @escaping TerminationHandler) -> RegistrationKey? {
+        if self.terminating.get(default: false) {
+            self.observabilityScope?.emit(debug: "not registering '\(name)' with terminator, termination in progress")
+            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: Cancellable) -> RegistrationKey? {
+        self.register(name: name, handler: handler.cancel(deadline:))
+    }
+
+    @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
+    }
+
+    @discardableResult
+    public func terminate(deadline: DispatchTime? = .none) -> Int {
+        self.terminating.put(true)
+
+        self.observabilityScope?.emit(info: "starting termination cycle with \(self.registry.count) termination handlers registered")
+
+        let deadline = deadline ?? .now() + .seconds(30)
+        let terminationHandlers = self.registry.get()
+        let terminated = ThreadSafeArrayStore<String>()
+        let group = DispatchGroup()
+        for (_, (name, handler)) in terminationHandlers {
+            self.terminationQueue.async(group: group) {
+                do {
+                    self.observabilityScope?.emit(debug: "terminating '\(name)'")
+                    // TODO: consider making this configurable
+                    // deadline for individual handlers set slightly before overall deadline
+                    let handlerDeadline = deadline - .seconds(5)
+                    try handler(handlerDeadline)
+                    terminated.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 \(terminationHandlers.count - terminated.count) termination handlers remaining")
+        } else {
+            self.observabilityScope?.emit(info: "termination cycle completed successfully")
+        }
+
+        self.terminating.put(false)
+
+        return terminated.count
+    }
+
+    public func cancel(deadline: DispatchTime) throws {
+        self.terminate(deadline: deadline)
+    }
+}
+
+public protocol Cancellable {
+    func cancel(deadline: DispatchTime) throws -> Void
+}
+
+public struct CancellationError: Error, CustomStringConvertible {
+    public let description = "Operation cancelled"
+
+    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: 0)
+        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/Build/BuildOperation.swift

@@ -140,7 +140,7 @@ public final class BuildOperation: PackageStructureDelegate, SPMBuildCore.BuildS
     }
 
     /// Cancel the active build operation.
-    public func cancel() {
+    public func cancel(deadline: DispatchTime) throws {
         buildSystem?.cancel()
     }
 

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