クラウド Mac で iOS タスクの同時実行数を検証する

20 個の添付ファイルを一括処理するとき、すべてのタスクをタスクグループに追加し、結果を順に取り出すだけでは、同時実行数を制限したことにはなりません。ネットワーク接続、デコード用バッファ、一時ファイルは、それでも同時に増える可能性があります。クラウド Mac でこのロジックを検証する際に重視すべきなのは、処理速度ではありません。上限が 3 なら、4 つ目の処理は空き枠ができるまで開始できないことを確かめる必要があります。

ここでは、スケジューラーをネットワーク処理の実装から切り離します。まず制御可能な処理で同時実行枠の動作を検証し、その後で iOS アプリの業務ロジックに組み込みます。こうすれば、失敗の原因がスケジューリングの不具合なのか、外部サービスの変動なのかを切り分けられます。結論を得るために何度も実行し直す必要はありません。

まず検証可能なスケジューリング契約を定義する

出力は入力と同じ順序に並ぶものとし、処理が完了する順序は問いません。入力が空なら空の配列を返し、同時実行数の上限は必ず正の値とします。いずれかの子タスクがエラーをスローしたら、呼び出し全体を失敗させ、部分的な結果は返しません。親タスクがキャンセルされた後は、未実行の処理を引き続き追加してはいけません。

ここで制限するのは、渡された処理のうち実行中のものの数です。スレッド数でも、リクエストの送信レートでもありません。1 つの処理が内部で複数のリクエストを開始しても、それらは外側のカウント対象にはなりません。1 秒あたりのリクエスト数に制限が必要なら、別途実装する必要があります。同時実行枠で代用することはできません。

キャンセルは協調的な通知であり、強制終了ではありません。タスクグループはスコープを抜ける前に、子タスクの終了を待ちます。キャンセルを無視する処理があると、呼び出し全体がその処理を待ち続けることになります。

完了イベントを受けて次のタスクを追加する

次のコードを BoundedMap.swift として保存します。Swift ツールチェーンがインストールされたクラウド Mac で実行し、まず swift --version でバージョンを記録してください。この例では、厳密な並行性チェックに必要な Sendable 制約を使用しています。GUI セッションやシミュレーターは不要です。

import Foundation

func boundedMap<Input: Sendable, Output: Sendable>(
    _ inputs: [Input],
    limit: Int,
    operation: @escaping @Sendable (Input) async throws -> Output
) async throws -> [Output] {
    precondition(limit > 0)

    return try await withThrowingTaskGroup(
        of: (Int, Output).self
    ) { group in
        var nextIndex = 0
        var completed: [(Int, Output)] = []
        completed.reserveCapacity(inputs.count)

        for index in 0..<min(limit, inputs.count) {
            try Task.checkCancellation()
            let input = inputs[index]
            group.addTask {
                try Task.checkCancellation()
                return (index, try await operation(input))
            }
            nextIndex += 1
        }

        while let result = try await group.next() {
            completed.append(result)
            try Task.checkCancellation()

            if nextIndex < inputs.count {
                let index = nextIndex
                let input = inputs[index]
                group.addTask {
                    try Task.checkCancellation()
                    return (index, try await operation(input))
                }
                nextIndex += 1
            }
        }

        try Task.checkCancellation()
        return completed.sorted { $0.0 < $1.0 }.map { $0.1 }
    }
}

最初は上限の数だけタスクを追加します。その後は、完了した結果を 1 つ受け取るたびに、新しいタスクを 1 つ追加します。このため、入力が大量にあっても、その件数分の待機タスクを生成せずに済みます。結果とともにインデックスを返し、最後に入力順へ並べ直すことで、「先に完了したもの」を「先頭に並ぶもの」と取り違えないようにしています。

エラーは group.next() を通じて伝播します。クロージャーがエラーをスローして終了すると、タスクグループは残りの子タスクをキャンセルし、その終了を待ちます。この実装では、エラーを捕捉してタスクの追加を続けることも、親タスクのライフサイクルから切り離された独立したバックグラウンドタスクを使うこともありません。

actor のプローブで実際の実行中処理数を測る

次のコードを同じファイルの末尾に追加し、swift BoundedMap.swift を実行します。プローブは、処理の開始時にカウンターを 1 増やし、終了時に 1 減らします。スリープ中も actor はほかの呼び出しを受け付けられるため、意図せずすべての処理が直列化されることはありません。

actor Probe {
    private var active = 0
    private var peak = 0

    func run(_ value: Int) async throws -> Int {
        active += 1
        peak = max(peak, active)
        defer { active -= 1 }

        let delay = UInt64(5 - value % 5) * 10_000_000
        try await Task.sleep(nanoseconds: delay)
        return value * 2
    }

    func snapshot() -> (active: Int, peak: Int) {
        (active, peak)
    }
}

let probe = Probe()
let inputs = Array(0..<20)
let output = try await boundedMap(inputs, limit: 3) {
    try await probe.run($0)
}
let state = await probe.snapshot()
precondition(output == inputs.map { $0 * 2 })
precondition(state.active == 0)
precondition(state.peak <= 3)

待ち時間を変えているのは、完了順を入れ替えるためであり、性能を測定するためではありません。特定のタスクが何ミリ秒で完了するかをアサートしてはいけません。また、マシンの負荷によってピークが 3 に達しなかったとしても、それをエラーと判定しないでください。必ず満たすべき条件は、ピークが上限を超えないこと、結果がすべてそろうこと、呼び出しの終了後に実行中の処理数がゼロになることです。

空の配列、要素が 1 つの配列、上限が 1 の場合についても、それぞれ実行して確認してください。呼び出し側が設定ファイルから上限値を受け取るなら、ゼロと負の値を事前に拒否する必要があります。この例の事前条件は、プログラムの誤りを検出するためのものであり、ユーザー入力をそのまま検証する用途には適していません。

失敗とキャンセルは別々に検証する

成功するテストだけでは、キャンセルが正しく動作するとは証明できません。まず、指定した入力に対して必ずエラーをスローする処理を用意し、呼び出しが失敗することと、プローブのカウンターが最終的にゼロになることをアサートします。失敗するまでに開始したタスク数が特定の値と一致することは、アサートしないでください。エラーがタスクグループに届く前に、ほかのタスクが完了し、空き枠に次のタスクが追加されている可能性があります。

キャンセルのテストには、テスト側で開放できるゲートを使います。最初の処理群から「開始済み」の通知を送らせ、テスト側でその通知を受け取ったら親タスクをキャンセルし、その後でゲートを開放します。各処理はゲートを通過した後にキャンセルを確認します。テストでは親タスクの終了を待ち、最後に、開始済みの処理がすべて終了したことを確認します。

固定時間のスリープで、タスクが開始したかどうかを推測してはいけません。また、ゲート自体がすべての待機者を解放できる必要があります。そうでないと、テスト用フィクスチャのデッドロックをスケジューラーの停止と誤認してしまいます。検証レポートでは、キャンセルを要求した時刻、処理が終了した時刻、親タスクから制御が戻った時刻を区別してください。

業務ロジックへの組み込みでも同時実行枠の境界を守る

ネットワーク経由でダウンロードする場合は、ダウンロード、必要なデコード、一時ファイルの後処理を、同じ処理スコープ内に含めます。レスポンスを受け取った時点で枠を解放し、大きな画像のデコードをバックグラウンドタスクに任せると、ネットワークの同時実行数は制限できても、メモリ使用量のピークは制御できません。

この実装はすべての出力を保持するため、必要なメモリは結果の総量に応じて増えます。大きなファイルを扱う場合は、ファイルの内容をすべて配列に格納するのではなく、ファイルの保存先と小さな要約データを返すのが適切です。完了した結果から順次消費するパイプラインが必要なら、出力のバックプレッシャーに関する契約を別途定義する必要があります。この実装のメモリ使用量が一定だと、そのまま主張することはできません。

最後に、ツールチェーンのバージョン、テスト入力、上限、ピーク、終了件数を記録して残します。スケジューラーのテストは通常のユニットテストターゲットに組み込み、実際の処理についてはキャンセルとリソース解放のチェックを追加してください。同時実行枠、順序、終了処理という 3 つの境界を安定して検証できれば、別のクラウド Mac に移ったときにも、再確認できる動作の基準が得られます。

よくある質問