Async git repository opening #8721
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,6 +15,7 @@ import Dispatch | |
| import class Foundation.NSLock | ||
| import class Foundation.ProcessInfo | ||
| import struct Foundation.URL | ||
| import struct Foundation.UUID | ||
| import func TSCBasic.tsc_await | ||
|
|
||
| public enum Concurrency { | ||
|
|
@@ -76,3 +77,183 @@ extension DispatchQueue { | |
| } | ||
| } | ||
| } | ||
|
|
||
| /// A queue for running async operations with a limit on the number of concurrent tasks. | ||
| public final class AsyncOperationQueue: @unchecked Sendable { | ||
|
|
||
| // This implementation is identical to the AsyncOperationQueue in swift-build. | ||
| // Any modifications made here should also be made there. | ||
| // https://github.com/swiftlang/swift-build/blob/main/Sources/SWBUtil/AsyncOperationQueue.swift#L13 | ||
|
|
||
| fileprivate typealias ID = UUID | ||
| fileprivate typealias WaitingContinuation = CheckedContinuation<Void, any Error> | ||
|
|
||
| private let concurrentTasks: Int | ||
| private var activeTasks: Int = 0 | ||
| private var waitingTasks: [WaitingTask] = [] | ||
| private let waitingTasksLock = NSLock() | ||
|
|
||
| fileprivate enum WaitingTask { | ||
| case creating(ID) | ||
| case waiting(ID, WaitingContinuation) | ||
| case cancelled(ID) | ||
|
|
||
| var id: ID { | ||
| switch self { | ||
| case .creating(let id), .waiting(let id, _), .cancelled(let id): | ||
| return id | ||
| } | ||
| } | ||
|
|
||
| var continuation: WaitingContinuation? { | ||
| guard case .waiting(_, let continuation) = self else { | ||
| return nil | ||
| } | ||
| return continuation | ||
| } | ||
| } | ||
|
|
||
| /// Creates an `AsyncOperationQueue` with a specified number of concurrent tasks. | ||
| /// - Parameter concurrentTasks: The maximum number of concurrent tasks that can be executed concurrently. | ||
| public init(concurrentTasks: Int) { | ||
| self.concurrentTasks = concurrentTasks | ||
| } | ||
|
|
||
| deinit { | ||
| waitingTasksLock.withLock { | ||
| if !waitingTasks.filter({ $0.continuation != nil }).isEmpty { | ||
|
There was a problem hiding this comment. Isn't it unexpected if there is anything at all in waitingTasks, not just those with a non-nil continuation? There was a problem hiding this comment. Yep, you're right, I've updated to revert this check. |
||
| preconditionFailure("Deallocated with waiting tasks") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// Executes an asynchronous operation, ensuring that the number of concurrent tasks | ||
| // does not exceed the specified limit. | ||
| /// - Parameter operation: The asynchronous operation to execute. | ||
| /// - Returns: The result of the operation. | ||
| /// - Throws: An error thrown by the operation, or a `CancellationError` if the operation is cancelled. | ||
| public func withOperation<ReturnValue>( | ||
| _ operation: @Sendable () async throws -> sending ReturnValue | ||
plemarquand marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| ) async throws -> ReturnValue { | ||
| try await waitIfNeeded() | ||
| defer { signalCompletion() } | ||
| return try await operation() | ||
| } | ||
|
|
||
| private func waitIfNeeded() async throws { | ||
| guard waitingTasksLock.withLock({ | ||
| let shouldWait = activeTasks >= concurrentTasks | ||
| activeTasks += 1 | ||
| return shouldWait | ||
| }) else { | ||
| return // Less tasks are in flight than the limit. | ||
| } | ||
|
|
||
| let taskId = ID() | ||
| waitingTasksLock.withLock { | ||
| waitingTasks.append(.creating(taskId)) | ||
| } | ||
|
There was a problem hiding this comment. We could move this into the above lock and return an optional |
||
|
|
||
| enum TaskAction { | ||
| case start(WaitingContinuation) | ||
| case cancel(WaitingContinuation) | ||
| } | ||
|
|
||
| try await withTaskCancellationHandler { | ||
| try await withCheckedThrowingContinuation { (continuation: WaitingContinuation) -> Void in | ||
| let action: TaskAction? = waitingTasksLock.withLock { | ||
| guard let index = waitingTasks.firstIndex(where: { $0.id == taskId }) else { | ||
| // If the task was cancelled in onCancelled it will have been removed from the waiting tasks list. | ||
|
There was a problem hiding this comment. This comment is not true right? We are always going to get an index by just looking at the code in There was a problem hiding this comment. Thanks, I've tightend up this comment to reflect how this guard code would actually get called. |
||
| return .cancel(continuation) | ||
| } | ||
|
|
||
| // If the task was cancelled in between creating the task cancellation handler and aquiring the lock, | ||
| // we should resume the continuation with a `CancellationError`. | ||
| if case .cancelled = waitingTasks[index] { | ||
|
There was a problem hiding this comment. Instead of writing an |
||
| return .cancel(continuation) | ||
| } | ||
|
|
||
| // A task may have completed since we iniitally checked if we should wait. Check again in this locked | ||
| // section and if we can start it, remove it from the waiting tasks and start it immediately. | ||
| let shouldWait = activeTasks >= concurrentTasks | ||
| if shouldWait { | ||
| waitingTasks[index] = .waiting(taskId, continuation) | ||
| return nil | ||
| } else { | ||
| waitingTasks.remove(at: index) | ||
|
|
||
| // activeTasks isn't decremented in the `signalCompletion` method | ||
| // when the next task to start is .creating, so we decrement it here. | ||
| activeTasks -= 1 | ||
| return .start(continuation) | ||
| } | ||
| } | ||
|
|
||
| switch action { | ||
| case .some(.cancel(let continuation)): | ||
| continuation.resume(throwing: _Concurrency.CancellationError()) | ||
| case .some(.start(let continuation)): | ||
| continuation.resume() | ||
| case .none: | ||
| return | ||
| } | ||
|
|
||
| } | ||
| } onCancel: { | ||
| let continuation: WaitingContinuation? = self.waitingTasksLock.withLock { | ||
| guard let taskIndex = self.waitingTasks.firstIndex(where: { $0.id == taskId }) else { | ||
| return nil | ||
| } | ||
|
|
||
| switch self.waitingTasks[taskIndex] { | ||
| case .waiting(_, let continuation): | ||
| self.waitingTasks.remove(at: taskIndex) | ||
|
|
||
| // If the parent task is cancelled then we need to manually handle resuming the | ||
| // continuation for the waiting task with a `CancellationError`. Return the continuation | ||
| // here so it can be resumed once the `waitingTasksLock` is released. | ||
| return continuation | ||
| case .creating: | ||
| // If the task was still being created, mark it as cancelled in the queue so that | ||
| // withCheckedThrowingContinuation can immediately cancel it. | ||
| self.waitingTasks[taskIndex] = .cancelled(taskId) | ||
| activeTasks -= 1 | ||
| return nil | ||
| case .cancelled: | ||
| preconditionFailure("Attempting to cancel a task that was already cancelled") | ||
| } | ||
| } | ||
|
|
||
| continuation?.resume(throwing: _Concurrency.CancellationError()) | ||
| } | ||
| } | ||
|
|
||
| private func signalCompletion() { | ||
| let continuationToResume = waitingTasksLock.withLock { () -> WaitingContinuation? in | ||
| guard !waitingTasks.isEmpty else { | ||
| activeTasks -= 1 | ||
| return nil | ||
| } | ||
|
|
||
| while let lastTask = waitingTasks.first { | ||
| switch lastTask { | ||
| case .creating: | ||
| // If the next task is in the process of being created, let the | ||
| return Optional<WaitingContinuation>.none | ||
| case .waiting: | ||
| activeTasks -= 1 | ||
| // Begin the next waiting task | ||
| return waitingTasks.remove(at: 0).continuation | ||
|
There was a problem hiding this comment. Since we are operating under FIFO here it is worth using a There was a problem hiding this comment. I was trying to keep in mind this will make its way back in to swift-build, which has no dependency on swift-collections and looks to be trying to keep its dependencies minimal. @jakepetroules is that accurate, or would swift-build accept a swift-collections dependency for this? |
||
| case .cancelled: | ||
| // If the next task is cancelled, continue removing cancelled | ||
| // tasks until we find one that hasn't run yet or we run out. | ||
| _ = waitingTasks.remove(at: 0) | ||
| continue | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| continuationToResume?.resume() | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need this counter at all or can we just use
waitingTasks.count? Keeping the two in sync can be tricky and I had to triple check that the code does it correctly. That requires us to change theWaitingTaskto be justWorkTaskwith a new case but IMO that would make this code a lot clearer.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We can, but it that would mean the waitingTasks would no longer be able to be modelled as a queue since the actively running tasks would be contained within it.
I think an OrderedDictionary could work instead, but runs in to the swift-collections in swift-build question.