http client to support configurable max concurrent requests (#4181)

tomerd · tomerd · commit f54f12462c25 · 2022-02-28T19:33:28.000-08:00
motivation: controlling how many concurrent requests are taking place is important for controlling bandwidth requirements, especially as we are transitioning to registry based dependencies

changes:
* add max concurrency controls to http client using a dedicated queue
* remove workspace binary dependencies concurrency control since its not longer necessary (applied at http client level)
* fix a bug in URLSeesion based HTTP client where download can be removed between delegate calls
* improve several http client tests, making them thread safe and robust
* add tests
diff --git a/Sources/Basics/ConcurrencyHelpers.swift b/Sources/Basics/ConcurrencyHelpers.swift
@@ -12,6 +12,8 @@ import Dispatch
 import class Foundation.ProcessInfo
 import TSCBasic
 
+public typealias Lock = TSCBasic.Lock
+
 /// Thread-safe dictionary like structure
 public final class ThreadSafeKeyValueStore<Key, Value> where Key: Hashable {
     private var underlying: [Key: Value]
@@ -194,6 +196,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
@@ -222,6 +230,23 @@ public final class ThreadSafeBox<Value> {
     }
 }
 
+extension ThreadSafeBox where Value == Int {
+    public func increment() {
+        self.lock.withLock {
+            if let value = self.underlying {
+                self.underlying = value + 1
+            }
+        }
+    }
+    public func decrement() {
+        self.lock.withLock {
+            if let value = self.underlying {
+                self.underlying = value - 1
+            }
+        }
+    }
+}
+
 public enum Concurrency {
     public static var maxOperations: Int {
         return ProcessEnv.vars["SWIFTPM_MAX_CONCURRENT_OPERATIONS"].flatMap(Int.init) ?? ProcessInfo.processInfo.activeProcessorCount
diff --git a/Sources/Basics/Errors.swift b/Sources/Basics/Errors.swift
@@ -8,6 +8,10 @@
  See http://swift.org/CONTRIBUTORS.txt for Swift project authors
  */
 
+import struct TSCBasic.StringError
+
+public typealias StringError = TSCBasic.StringError
+
 public struct InternalError: Error {
     private let description: String
     public init(_ description: String) {
diff --git a/Sources/Basics/HTPClient+URLSession.swift b/Sources/Basics/HTPClient+URLSession.swift
@@ -149,7 +149,11 @@ private class DownloadTaskManager: NSObject, URLSessionDownloadDelegate {
             return
         }
 
-        task.location = location
+        do {
+            try task.fileSystem.move(from: AbsolutePath(location.path), to: task.destination)
+        } catch {
+            task.moveFileError = error
+        }
     }
 
     public func urlSession(_ session: URLSession, task downloadTask: URLSessionTask, didCompleteWithError error: Error?) {
@@ -160,10 +164,9 @@ private class DownloadTaskManager: NSObject, URLSessionDownloadDelegate {
         do {
             if let error = error {
                 throw HTTPClientError.downloadError("\(error)")
+            } else if let error = task.moveFileError {
+                throw error
             } else if let response = downloadTask.response as? HTTPURLResponse {
-                if let location = task.location {
-                    try task.fileSystem.move(from: AbsolutePath(location.path), to: task.destination)
-                }
                 task.completionHandler(.success(response.response(body: nil)))
             } else {
                 throw HTTPClientError.invalidResponse
@@ -180,7 +183,7 @@ private class DownloadTaskManager: NSObject, URLSessionDownloadDelegate {
         let completionHandler: HTTPClient.CompletionHandler
         let progressHandler: HTTPClient.ProgressHandler?
 
-        var location: URL?
+        var moveFileError: Error?
 
         init(task: URLSessionDownloadTask, fileSystem: FileSystem, destination: AbsolutePath, progressHandler: HTTPClient.ProgressHandler?, completionHandler: @escaping HTTPClient.CompletionHandler) {
             self.task = task
diff --git a/Sources/Basics/HTTPClient.swift b/Sources/Basics/HTTPClient.swift
@@ -13,6 +13,7 @@ import struct Foundation.Data
 import struct Foundation.Date
 import class Foundation.JSONDecoder
 import class Foundation.NSError
+import class Foundation.OperationQueue
 import struct Foundation.URL
 import TSCBasic
 
@@ -45,6 +46,11 @@ public struct HTTPClient {
     public var configuration: HTTPClientConfiguration
     private let underlying: Handler
 
+    /// DispatchSemaphore to restrict concurrent operations on manager.
+     private let concurrencySemaphore: DispatchSemaphore
+     /// OperationQueue to park pending requests
+     private let requestsQueue: OperationQueue
+
     // static to share across instances of the http client
     private static var hostsErrorsLock = Lock()
     private static var hostsErrors = [String: [Date]]()
@@ -53,6 +59,14 @@ public struct HTTPClient {
         self.configuration = configuration
         // FIXME: inject platform specific implementation here
         self.underlying = handler ?? URLSessionHTTPClient().execute
+
+        // this queue and semaphore is used to limit the amount of concurrent http requests taking place
+        // the default max number of request chosen to match Concurrency.maxOperations which is the number of active CPUs
+        let maxConcurrentRequests = configuration.maxConcurrentRequests ?? Concurrency.maxOperations
+        self.requestsQueue = OperationQueue()
+        self.requestsQueue.name = "org.swift.swiftpm.http-client"
+        self.requestsQueue.maxConcurrentOperationCount = maxConcurrentRequests
+        self.concurrencySemaphore = DispatchSemaphore(value: maxConcurrentRequests)
     }
 
     /// Execute an HTTP request asynchronously
@@ -100,12 +114,14 @@ public struct HTTPClient {
             observabilityScope: observabilityScope,
             progress: progress.map { handler in
                 { received, expected in
+                    // call back on the requested queue
                     callbackQueue.async {
                         handler(received, expected)
                     }
                 }
             },
             completion: { result in
+                // call back on the requested queue
                 callbackQueue.async {
                     completion(result)
                 }
@@ -114,45 +130,65 @@ public struct HTTPClient {
     }
 
     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)")
-            return completion(.failure(HTTPClientError.circuitBreakerTriggered))
+        // wrap completion handler with concurrency control cleanup
+        let originalCompletion = completion
+        let completion: CompletionHandler = { result in
+            // free concurrency control semaphore
+            self.concurrencySemaphore.signal()
+            originalCompletion(result)
         }
 
-        self.underlying(
-            request,
-            { received, expected in
-                if let max = request.options.maximumResponseSizeInBytes {
-                    guard received < max else {
-                        // FIXME: cancel the request?
-                        return completion(.failure(HTTPClientError.responseTooLarge(received)))
-                    }
-                }
-                progress?(received, expected)
-            },
-            { result in
-                switch result {
-                case .failure(let error):
-                    completion(.failure(error))
-                case .success(let response):
-                    // record host errors for circuit breaker
-                    self.recordErrorIfNecessary(response: response, request: request)
-                    // handle retry strategy
-                    if let retryDelay = self.shouldRetry(response: response, request: request, requestNumber: requestNumber) {
-                        observabilityScope?.emit(warning: "\(request.url) failed, retrying in \(retryDelay)")
-                        // TODO: dedicated retry queue?
-                        return self.configuration.callbackQueue.asyncAfter(deadline: .now() + retryDelay) {
-                            self._execute(request: request, requestNumber: requestNumber + 1, observabilityScope: observabilityScope, progress: progress, completion: completion)
+        // we must not block the calling thread (for concurrency control) so nesting this in a queue
+        self.requestsQueue.addOperation {
+            // park the request thread based on the max concurrency allowed
+            self.concurrencySemaphore.wait()
+
+            // apply circuit breaker if necessary
+            if self.shouldCircuitBreak(request: request) {
+                observabilityScope?.emit(warning: "Circuit breaker triggered for \(request.url)")
+                return completion(.failure(HTTPClientError.circuitBreakerTriggered))
+            }
+
+            // call underlying handler
+            self.underlying(
+                request,
+                { received, expected in
+                    if let max = request.options.maximumResponseSizeInBytes {
+                        guard received < max else {
+                            // FIXME: cancel the request?
+                            return completion(.failure(HTTPClientError.responseTooLarge(received)))
                         }
                     }
-                    // check for valid response codes
-                    if let validResponseCodes = request.options.validResponseCodes, !validResponseCodes.contains(response.statusCode) {
-                        return completion(.failure(HTTPClientError.badResponseStatusCode(response.statusCode)))
+                    progress?(received, expected)
+                },
+                { result in
+                    // handle result
+                    switch result {
+                    case .failure(let error):
+                        completion(.failure(error))
+                    case .success(let response):
+                        // record host errors for circuit breaker
+                        self.recordErrorIfNecessary(response: response, request: request)
+                        // 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
+                            self.concurrencySemaphore.signal()
+                            // 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)
+                            }
+                        }
+                        // check for valid response codes
+                        if let validResponseCodes = request.options.validResponseCodes, !validResponseCodes.contains(response.statusCode) {
+                            return completion(.failure(HTTPClientError.badResponseStatusCode(response.statusCode)))
+                        }
+                        completion(.success(response))
                     }
-                    completion(.success(response))
                 }
-            }
-        )
+            )
+        }
     }
 
     private func shouldRetry(response: Response, request: Request, requestNumber: Int) -> DispatchTimeInterval? {
@@ -245,6 +281,7 @@ public struct HTTPClientConfiguration {
     public var authorizationProvider: HTTPClientAuthorizationProvider?
     public var retryStrategy: HTTPClientRetryStrategy?
     public var circuitBreakerStrategy: HTTPClientCircuitBreakerStrategy?
+    public var maxConcurrentRequests: Int?
     public var callbackQueue: DispatchQueue
 
     public init() {
@@ -253,6 +290,7 @@ public struct HTTPClientConfiguration {
         self.authorizationProvider = .none
         self.retryStrategy = .none
         self.circuitBreakerStrategy = .none
+        self.maxConcurrentRequests = .none
         self.callbackQueue = .sharedConcurrent
     }
 }
diff --git a/Sources/Workspace/Workspace.swift b/Sources/Workspace/Workspace.swift
@@ -2515,9 +2515,6 @@ extension Workspace {
             }
         }
 
-        // download max n files concurrently
-        let semaphore = DispatchSemaphore(value: Concurrency.maxOperations)
-
         // finally download zip files, if any
         for artifact in zipArtifacts.get() {
             let parentDirectory =  self.location.artifactsDirectory.appending(component: artifact.packageRef.identity.description)
@@ -2532,7 +2529,6 @@ extension Workspace {
                 }
             }
 
-            semaphore.wait()
             group.enter()
             var headers = HTTPClientHeaders()
             headers.add(name: "Accept", value: "application/octet-stream")
@@ -2553,10 +2549,7 @@ extension Workspace {
                         totalBytesToDownload: totalBytesToDownload)
                 },
                 completion: { downloadResult in
-                    defer {
-                        group.leave()
-                        semaphore.signal()
-                    }
+                    defer { group.leave() }
 
                     // TODO: Use the same extraction logic for both remote and local archived artifacts.
                     switch downloadResult {
diff --git a/Tests/BasicsTests/HTTPClientTests.swift b/Tests/BasicsTests/HTTPClientTests.swift
diff --git a/Tests/WorkspaceTests/WorkspaceTests.swift b/Tests/WorkspaceTests/WorkspaceTests.swift
