diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift index 713afdb0..a0d4c9e7 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupRequest.swift @@ -8,6 +8,7 @@ import Foundation +/// Replaces all of the app's groups with the given ones public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlatformRequest { public struct Response: Decodable, Sendable {} @@ -16,14 +17,14 @@ public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlat public let platform: DeveloperServicesPlatform public let teamID: DeveloperServicesTeam.ID public let appIDID: String - public let groupID: DeveloperServicesAppGroup.ID + public let groupIDs: [DeveloperServicesAppGroup.ID] var subAction: String { return "assignApplicationGroupToAppId" } var subParameters: [String: Any] { return [ "teamId": teamID.rawValue, "appIdId": appIDID, - "applicationGroups": groupID.rawValue + "applicationGroups": groupIDs.map(\.rawValue) ] } @@ -31,12 +32,12 @@ public final class DeveloperServicesAssignAppGroupRequest: DeveloperServicesPlat platform: DeveloperServicesPlatform, teamID: DeveloperServicesTeam.ID, appIDID: String, - groupID: DeveloperServicesAppGroup.ID + groupIDs: [DeveloperServicesAppGroup.ID] ) { self.platform = platform self.teamID = teamID self.appIDID = appIDID - self.groupID = groupID + self.groupIDs = groupIDs } } diff --git a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift index b13e1a5b..5516bd10 100644 --- a/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift +++ b/Sources/XKit/DeveloperServices/App Groups/DeveloperServicesAssignAppGroupsOperation.swift @@ -8,6 +8,7 @@ import Foundation import DeveloperAPI +import XUtils public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperation { @@ -33,36 +34,23 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat self.client = DeveloperServicesClient(authData: authData) } - private func upsertAppGroup( + private func getOrCreateAppGroup( _ groupID: DeveloperServicesAppGroup.GroupID, existingGroups: [String: DeveloperServicesAppGroup] - ) async throws -> DeveloperServicesAppGroup.GroupID { + ) async throws -> DeveloperServicesAppGroup { let sanitized = ProvisioningIdentifiers.sanitize(groupID: groupID) - let group: DeveloperServicesAppGroup if let existingGroup = existingGroups[sanitized] { - group = existingGroup - } else { - let groupID = ProvisioningIdentifiers.groupID(fromSanitized: sanitized, context: context) - let name = ProvisioningIdentifiers.groupName(fromSanitized: sanitized) - let request = DeveloperServicesAddAppGroupRequest( - platform: .iOS, - teamID: xcodeAuthData.teamID, - name: name, - groupID: groupID - ) - group = try await client.send(request) + return existingGroup } - - _ = try await client.send( - DeveloperServicesAssignAppGroupRequest( - platform: .iOS, - teamID: xcodeAuthData.teamID, - appIDID: appID.id, - groupID: group.id - ) + let groupID = ProvisioningIdentifiers.groupID(fromSanitized: sanitized, context: context) + let name = ProvisioningIdentifiers.groupName(fromSanitized: sanitized) + let request = DeveloperServicesAddAppGroupRequest( + platform: .iOS, + teamID: xcodeAuthData.teamID, + name: name, + groupID: groupID ) - - return group.groupID + return try await client.send(request) } public func perform() async throws -> [DeveloperServicesAppGroup.GroupID] { @@ -71,14 +59,20 @@ public struct DeveloperServicesAssignAppGroupsOperation: DeveloperServicesOperat )) let sanitized = existing.map { (ProvisioningIdentifiers.sanitize(groupID: $0.groupID), $0) } let dict = Dictionary(sanitized, uniquingKeysWith: { $1 }) - return try await withThrowingTaskGroup(of: DeveloperServicesAppGroup.GroupID.self) { group in - for groupID in groupIDs { - group.addTask { - try await upsertAppGroup(groupID, existingGroups: dict) - } - } - return try await group.reduce(into: []) { $0.append($1) } + let groups = try await groupIDs.concurrentMap { groupID in + try await getOrCreateAppGroup(groupID, existingGroups: dict) } + + _ = try await client.send( + DeveloperServicesAssignAppGroupRequest( + platform: .iOS, + teamID: xcodeAuthData.teamID, + appIDID: appID.id, + groupIDs: groups.map(\.id) + ) + ) + + return groups.map(\.groupID) } } diff --git a/Sources/XUtils/Stdlib+Utils.swift b/Sources/XUtils/Stdlib+Utils.swift new file mode 100644 index 00000000..5084948d --- /dev/null +++ b/Sources/XUtils/Stdlib+Utils.swift @@ -0,0 +1,66 @@ +extension Sequence where Element: Sendable { + /// Like `map`, but runs transformations concurrently. + /// + /// - Parameter maxConcurrency: The maximum number of transforms to run concurrently. + /// If `nil`, we choose a reasonable default value. Use `.max` to run all at once. + /// + /// - Parameter transform: The transformation to apply to each element. + /// + /// - Returns: Transformed elements in the same order as the original sequence. + public nonisolated(nonsending) func concurrentMap( + maxConcurrency: Int? = nil, + _ transform: @escaping @Sendable @isolated(any) (Element) async throws(E) -> T, + ) async throws(E) -> [T] { + try await _concurrentMap( + maxConcurrency: maxConcurrency ?? 10, + transform, + willWaitForNext: nil, + ) + } + + nonisolated(nonsending) func _concurrentMap( + maxConcurrency: Int, + _ transform: @escaping @Sendable @isolated(any) (Element) async throws(E) -> T, + willWaitForNext: (() -> Void)?, // hook for testing + ) async throws(E) -> [T] { + precondition(maxConcurrency > 0, "maxConcurrency must be > 0") + + return try await withTaskGroup(of: (Int, Result).self) { group -> Result<[T], E> in + var results: [T?] = [] + results.reserveCapacity(self.underestimatedCount) + var iterator = makeIterator() + var exhausted = false + var inFlight = 0 + while true { + while !exhausted && inFlight < maxConcurrency { + guard let element = iterator.next() else { + exhausted = true + break + } + let index = results.count + results.append(nil) + group.addTask { + let result = await Result { () async throws(E) in + try await transform(element) + } + return (index, result) + } + inFlight += 1 + } + willWaitForNext?() + guard let (index, result) = await group.next() else { break } + inFlight -= 1 + switch result { + case .success(let value): + results[index] = value + case .failure(let error): + group.cancelAll() + return .failure(error) + } + } + // we can't use `as! [T]` because it breaks nested optionals (Foo??) + return .success(results.map { $0! }) + } + .get() + } +} diff --git a/Tests/XToolTests/ConcurrentMapTests.swift b/Tests/XToolTests/ConcurrentMapTests.swift new file mode 100644 index 00000000..32456b7c --- /dev/null +++ b/Tests/XToolTests/ConcurrentMapTests.swift @@ -0,0 +1,227 @@ +import Testing +@testable import XUtils + +@Suite(.timeLimit(.minutes(1))) +struct ConcurrentMapTests { + @Test(arguments: [[], [7], [3, 1, 3, 2], Array(0..<10)], [nil, 1, 2, Int.max] as [Int?]) + func nonThrowingTransform(input: [Int], maxConcurrency: Int?) async { + // No `try` is needed when the transform cannot throw. + let result = await input.concurrentMap(maxConcurrency: maxConcurrency) { String($0 * 2) } + #expect(result == input.map { String($0 * 2) }) + } + + @Test func emptyInputDoesNotCallTransform() async throws { + let result = try await [Int]().concurrentMap { _ throws(MapError) -> Int in + Issue.record("An empty sequence must not invoke the transform") + throw .failed(0) + } + #expect(result.isEmpty) + } + + @Test func singlePassSequence() async { + var next = 0 + let input = AnySequence(AnyIterator { + defer { next += 1 } + return next < 4 ? next : nil + }) + #expect(input.underestimatedCount == 0) + + let result = await input.concurrentMap(maxConcurrency: 1) { $0 * 2 } + + #expect(result == [0, 2, 4, 6]) + } + + @Test func optionalResults() async { + let result = await (0..<4).concurrentMap { value -> Int? in + value.isMultiple(of: 2) ? nil : value + } + #expect(result == [nil, 1, nil, 3]) + } + + @Test func nestedOptionalResults() async { + let input: [Int??] = [.none, .some(.none), .some(.some(3))] + let result = await input.concurrentMap { $0 } + #expect(result == input) + } + + @Test(arguments: [2, 4, Int.max]) + func concurrentExecutionPreservesInputOrder(maxConcurrency: Int) async { + let started = AsyncStream.makeStream() + let finished = AsyncStream.makeStream() + let gates = (0..<4).map { _ in AsyncStream.makeStream() } + let task = Task { + await (0..<4).concurrentMap(maxConcurrency: maxConcurrency) { value in + started.continuation.yield(value) + for await _ in gates[value].stream {} + finished.continuation.yield(value) + return value * 2 + } + } + defer { task.cancel() } + + // Fill the concurrency window before allowing any transform to finish. + let initialCount = min(maxConcurrency, 4) + var starts = started.stream.makeAsyncIterator() + var startedValues: [Int] = [] + for _ in 0...makeStream() + let gates = (0..<25).map { _ in AsyncStream.makeStream() } + let task = Task { + // Both closures below run in the producer task. Only immutable snapshots + // cross to the test through the checkpoint stream. + var submitted = 0 + let input = AnySequence(AnyIterator { + guard submitted < gates.count else { return nil } + defer { submitted += 1 } + return submitted + }) + return await input._concurrentMap(maxConcurrency: maxConcurrency) { value in + for await _ in gates[value].stream {} + return value * 2 + } willWaitForNext: { + checkpoints.continuation.yield(submitted) + } + } + defer { task.cancel() } + + // No transform can finish yet. The hook proves submission has reached + // group.next(), rather than merely being paused by the scheduler. + let initialCount = min(maxConcurrency, gates.count) + var waits = checkpoints.stream.makeAsyncIterator() + #expect(await waits.next() == initialCount) + + // Each completed transform frees exactly one slot. Check the next submission + // boundary before releasing another transform, including while draining. + for value in gates.indices { + gates[value].continuation.finish() + #expect(await waits.next() == min(initialCount + value + 1, gates.count)) + } + let result = await withTaskCancellationHandler { + await task.value + } onCancel: { + task.cancel() + } + + #expect(result == (0..<25).map { $0 * 2 }) + } + + @Test @MainActor func actorIsolatedTransform() async { + var visited: [Int] = [] + let result = await (0..<4).concurrentMap(maxConcurrency: 2) { @MainActor value in + MainActor.assertIsolated() + visited.append(value) + return value * 2 + } + + #expect(visited.sorted() == [0, 1, 2, 3]) + #expect(result == [0, 2, 4, 6]) + } + + @Test func preservesTypedError() async { + // This signature also checks that concurrentMap preserves the error type. + func map() async throws(MapError) -> [Int] { + try await [42].concurrentMap { value throws(MapError) -> Int in + throw .failed(value) + } + } + + await #expect(throws: MapError.failed(42)) { + try await map() + } + } + + @Test func failureCancelsUnfinishedTransforms() async { + let started = AsyncStream.makeStream() + let failureGate = AsyncStream.makeStream() + let siblingGate = AsyncStream.makeStream() + let cancelled = AsyncStream.makeStream() + let task = Task { + try await (0..<10).concurrentMap(maxConcurrency: 3) { value async throws(MapError) -> Int in + #expect(value < 3, "A failure must not schedule the remaining inputs") + started.continuation.yield(value) + if value == 1 { + for await _ in failureGate.stream {} + throw .failed(value) + } + // This stream only finishes when the task is cancelled. + for await _ in siblingGate.stream {} + #expect(Task.isCancelled) + cancelled.continuation.yield(value) + return value + } + } + defer { task.cancel() } + + var starts = started.stream.makeAsyncIterator() + for _ in 0..<3 { + _ = await starts.next() + } + failureGate.continuation.finish() + + await #expect(throws: MapError.failed(1)) { + try await withTaskCancellationHandler { + try await task.value + } onCancel: { + // Let the suite's time limit break a regression in sibling cancellation. + task.cancel() + } + } + // The task group must wait for both cancelled transforms to exit. + cancelled.continuation.finish() + var cancelledValues: [Int] = [] + for await value in cancelled.stream { + cancelledValues.append(value) + } + #expect(cancelledValues.sorted() == [0, 2]) + } + + @Test func parentCancellationReachesTransforms() async { + let started = AsyncStream.makeStream() + let gate = AsyncStream.makeStream() + let task = Task { + try await [1].concurrentMap { _ in + started.continuation.yield(()) + for await _ in gate.stream {} + try Task.checkCancellation() + } + } + defer { task.cancel() } + + var starts = started.stream.makeAsyncIterator() + _ = await starts.next() + task.cancel() + + await #expect(throws: CancellationError.self) { + try await task.value + } + } + + private enum MapError: Error, Equatable { + case failed(Int) + } +}