1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
//===----------------------------------------------------------------------===//
//
// This source file is part of the Swift open source project
//
// Copyright (c) 2025 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 the list of Swift project authors
//
//===----------------------------------------------------------------------===//
public actor AsyncOperationQueue {
private let concurrentTasks: Int
private var activeTasks: Int = 0
private var waitingTasks: [CheckedContinuation<Void, Never>] = []
public init(concurrentTasks: Int) {
self.concurrentTasks = concurrentTasks
}
deinit {
if !waitingTasks.isEmpty {
preconditionFailure("Deallocated with waiting tasks")
}
}
public func withOperation<ReturnValue>(
_ operation: @Sendable () async -> sending ReturnValue
) async -> ReturnValue {
await waitIfNeeded()
defer { signalCompletion() }
return await operation()
}
public func withOperation<ReturnValue>(
_ operation: @Sendable () async throws -> sending ReturnValue
) async throws -> ReturnValue {
await waitIfNeeded()
defer { signalCompletion() }
return try await operation()
}
private func waitIfNeeded() async {
if activeTasks >= concurrentTasks {
await withCheckedContinuation { continuation in
waitingTasks.append(continuation)
}
}
activeTasks += 1
}
private func signalCompletion() {
activeTasks -= 1
if let continuation = waitingTasks.popLast() {
continuation.resume()
}
}
}
|
