Add concurrency helper
This commit is contained in:
@@ -0,0 +1,42 @@
|
|||||||
|
import Foundation
|
||||||
|
|
||||||
|
/// 同時実行数を抑えつつ並列に変換し、入力と同じ順序で結果を返す。
|
||||||
|
///
|
||||||
|
/// 画像デコードや Vision の推論は CPU を使い切るので、無制限に投げるより
|
||||||
|
/// コア数程度に抑えたほうが速く、メモリも安定する。
|
||||||
|
func mapConcurrently<Item: Sendable, Output: Sendable>(
|
||||||
|
_ items: [Item],
|
||||||
|
limit: Int = max(2, ProcessInfo.processInfo.activeProcessorCount),
|
||||||
|
onProgress: (@Sendable (Int) -> Void)? = nil,
|
||||||
|
_ transform: @escaping @Sendable (Item) async -> Output
|
||||||
|
) async throws -> [Output] {
|
||||||
|
guard !items.isEmpty else { return [] }
|
||||||
|
|
||||||
|
var results = [Output?](repeating: nil, count: items.count)
|
||||||
|
|
||||||
|
try await withThrowingTaskGroup(of: (Int, Output).self) { group in
|
||||||
|
var next = 0
|
||||||
|
let initial = min(limit, items.count)
|
||||||
|
while next < initial {
|
||||||
|
let index = next
|
||||||
|
group.addTask { (index, await transform(items[index])) }
|
||||||
|
next += 1
|
||||||
|
}
|
||||||
|
|
||||||
|
var finished = 0
|
||||||
|
while let (index, value) = try await group.next() {
|
||||||
|
results[index] = value
|
||||||
|
finished += 1
|
||||||
|
onProgress?(finished)
|
||||||
|
try Task.checkCancellation()
|
||||||
|
|
||||||
|
if next < items.count {
|
||||||
|
let index = next
|
||||||
|
group.addTask { (index, await transform(items[index])) }
|
||||||
|
next += 1
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return results.compactMap { $0 }
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user