兄弟タスクが失敗しても成功したSwiftの計算結果を保持するには、各子タスク内で計算エラーを捕捉し、結果を値として返します。このMac向けコマンドラインの例では、すべての結果を収集し、入力順を保ち、成功した値を利用できる状態にします。

対象とするのは、小規模な合成バッチ計算です。非負整数のリストを2乗し、どの入力が成功したかを報告します。負の入力と整数オーバーフローは、それぞれ個別の失敗になります。計算データはすべてメモリ内に保持し、結果はターミナルに出力します。

1つのエラーが結果の収集を中断するのはなぜですか？

子タスクがエラーをスローしても、兄弟タスクがただちにキャンセルされるわけではありません。重要な境界はタスクグループの本体です。エラーがその本体の外まで伝播すると、Swiftは残りの子タスクをキャンセルします。Appleは、この違いを withThrowingTaskGroup(of:returning:isolation:body:) のリファレンスで説明しています。

そのため、外側のcatchを使うthrowingな反復ループでは、最初に検出したエラーで収集が止まることがあります。それ以前の値はすでに保存されている可能性がありますが、catchの後にループが再開することはありません。エラーがグループ本体の外まで伝播すると、残りの子タスクにはキャンセルが通知されます。

独立した計算では、失敗時の方針を明示してください。以下の例では、子タスクが成功、失敗、キャンセルのいずれかを返す、非throwingなタスクグループを使います。計算自体は引き続きエラーをスローしますが、ラッパーがそのエラーを値に変換してからグループへ返します。

これはSwiftで定められた境界に沿った方法です。非throwingなタスクグループの子タスクは、スローされたエラーを伝播できないため、内部で処理するか、戻り値で結果を表す必要があります。AppleのTaskGroupのドキュメントを参照してください。

各入力が返す結果を定義する

このバッチでは、入力位置ごとに1つの結果を返します。計算が成功すると、正確な整数の2乗を返します。計算が拒否されると、失敗の説明を返します。キャンセルが検出されたため計算が省略される場合は、キャンセルの結果を返します。

状況 結果 バッチの動作
2乗がInt64に収まる非負の入力 2乗値を含む成功 値を保持する
負の入力 失敗：negativeInput ほかの結果の収集を続ける
2乗がInt64の範囲を超える 失敗：overflow ほかの結果の収集を続ける
計算前にキャンセルを検出 キャンセル 完了済みの結果を保持し、保留中の処理を省略する

負の入力を拒否するのは、説明のために設定したアプリケーションルールです。負の数も数学的には2乗できます。このルールをオーバーフローとは分けることで、この例では異なる2つの失敗条件を示せます。

乗算にはmultipliedReportingOverflow(by:)を使い、オーバーフローを示すフラグを確認します。Appleの仕様では、このフラグがfalseの場合に限り、返される部分値は完全な積です。trueの場合は値が切り詰められています。この例では、切り詰められた値を破棄してエラーをスローします。Int64.multipliedReportingOverflow(by:)を参照してください。

各結果には入力インデックスも含めます。Swiftでは子タスクの完了順は保証されないため、完了した値をそのまま追加するだけでは元の順序を保てません。Swift言語ガイドの並行処理では、このスケジューリング動作を説明しています。

Macで完全なサンプルを実行する

Swift 6以降のツールチェーンを備えたMacを使います。Swift.orgの入門ガイドの推奨に従い、ターミナルでswift --versionを実行して、インストール済みのコンパイラを確認してください。Swiftが使えない場合は、先にそのガイドのインストールリンクに従ってください。

以下の完全なプログラムを作業フォルダにCalculations.swiftという名前で保存します。Foundationをインポートし、使用可能なプロセッサ数から初期の並行処理上限を選びます。

import Foundation

enum CalculationError: Error {
    case negativeInput
    case overflow
}

enum Outcome: Sendable {
    case success(Int64)
    case failure(String)
    case cancelled
}

struct Row: Sendable {
    let index: Int
    let outcome: Outcome
}

func square(_ input: Int64) throws -> Int64 {
    guard input >= 0 else {
        throw CalculationError.negativeInput
    }
    let product = input.multipliedReportingOverflow(by: input)
    guard !product.overflow else {
        throw CalculationError.overflow
    }
    return product.partialValue
}

func evaluate(index: Int, input: Int64) -> Row {
    do {
        try Task.checkCancellation()
        let value = try square(input)
        return Row(index: index, outcome: .success(value))
    } catch is CancellationError {
        return Row(index: index, outcome: .cancelled)
    } catch {
        return Row(
            index: index,
            outcome: .failure(String(describing: error))
        )
    }
}

func collect(
    _ inputs: [Int64],
    maxConcurrent: Int
) async -> [Row] {
    guard !inputs.isEmpty else { return [] }
    let width = min(inputs.count, max(1, maxConcurrent))

    return await withTaskGroup(of: Row.self) { group in
        var rows: [Row] = []
        rows.reserveCapacity(inputs.count)

        for index in 0..<width {
            let input = inputs[index]
            group.addTask {
                evaluate(index: index, input: input)
            }
        }
        var nextIndex = width

        while let row = await group.next() {
            rows.append(row)

            if Task.isCancelled {
                while nextIndex < inputs.count {
                    rows.append(Row(
                        index: nextIndex,
                        outcome: .cancelled
                    ))
                    nextIndex += 1
                }
            } else if nextIndex < inputs.count {
                let index = nextIndex
                let input = inputs[index]
                nextIndex += 1
                group.addTask {
                    evaluate(index: index, input: input)
                }
            }
        }

        return rows.sorted { $0.index < $1.index }
    }
}

@main
struct CalculationDemo {
    static func main() async {
        let inputs: [Int64] = [3, -2, 4, Int64.max, 0]
        let width = min(
            4,
            max(1, ProcessInfo.processInfo.activeProcessorCount)
        )
        let rows = await collect(inputs, maxConcurrent: width)
        var successes: [Int64] = []

        for row in rows {
            let input = inputs[row.index]
            switch row.outcome {
            case .success(let value):
                successes.append(value)
                print("[\(row.index)] \(input): success \(value)")
            case .failure(let message):
                print("[\(row.index)] \(input): failure \(message)")
            case .cancelled:
                print("[\(row.index)] \(input): cancelled")
            }
        }
        print("Successful values: \(successes)")
    }
}

そのフォルダで、プログラムをコンパイルして実行します。

swiftc -swift-version 6 -parse-as-library Calculations.swift -o Calculations
./Calculations

-parse-as-libraryオプションを使うと、この単一ソースファイルで明示的な@mainエントリーポイントを使用できます。実行ファイルは、アプリケーションデータの読み書きをせずに計算を行います。調査環境ではSwiftを利用できなかったため、このプログラムは確認しましたが、コンパイルも実行もしていません。

キャンセルなしで指定された入力を使った場合、次の出力が期待されます。

[0] 3: success 9
[1] -2: failure negativeInput
[2] 4: success 16
[3] 9223372036854775807: failure overflow
[4] 0: success 0
Successful values: [9, 16, 0]

収集後に入力インデックスでソートするため、出力順は入力順になります。実行順は変わることがありますが、表示順には影響しません。成功値の配列は後から作成するので、すべての結果を確認できる状態に保てます。

収集側は成功した結果をすべて保持するには？

結果配列は親タスクが所有します。各子タスクはインデックスと変更されない入力を1つずつ受け取り、その結果を計算して行を返します。子タスクが親の配列に直接追加することはありません。これは、タスクグループ本体の中で子タスクの結果を蓄積するという言語ガイドの方法に沿っています。並行処理を参照してください。

収集側は、最初に上限数まで子タスクを投入します。行の処理が完了するたびに、空いた枠へ次の入力を投入します。したがって、計算の失敗も成功と同じようにバッチを進めます。どちらも通常の行を返すため、収集は続きます。

FoundationのactiveProcessorCountは、コンピュータで使用可能な処理コア数を返します。この例では、説明のためにその数を4に制限しています。この上限は、測定に基づく最適値でも、4つの計算が同時に実行される保証でもありません。AppleのactiveProcessorCountのドキュメントを参照してください。

並行処理の上限により、同時に存在する子タスクの数は制限されますが、結果配列のサイズは入力数に応じて増えます。小規模なデスクトップのバッチ処理には適しています。ごく小さな整数の2乗計算では、この例は結果の処理方法を示すものであり、逐次ループより高速だと主張するものではありません。

入力が空なら、空の配列を返します。並行処理の上限が0以下の場合は、1に補正します。同じ入力値が重複していても、識別には位置を使うため、それぞれ別の行として扱われます。

親タスクがキャンセルされた場合はどうなりますか？

Swiftのキャンセルは協調的に行われます。親タスクをキャンセルすると、構造化された子タスクにもキャンセルが伝播しますが、子タスクはそのシグナルを検出して応答する必要があります。AppleのTaskGroupのドキュメントでは、キャンセルの伝播と協調的な処理の必要性を説明しています。

ここでは、計算前にevaluateがTask.checkCancellation()を呼び出します。Appleの仕様では、キャンセルが検出されるとこのメソッドはCancellationErrorをスローします。ラッパーはこれを別途捕捉し、キャンセルの行を返します。checkCancellation()を参照してください。

収集側がキャンセルを検出すると、まだ投入されていない入力をキャンセル済みとして記録します。すでに投入された子タスクは、完了するまで結果を回収します。成功した結果はレポートに残り、キャンセルを検出した子タスクはキャンセルの行を返します。

キャンセルは、子タスクがチェックを終えた後に届く場合があります。その場合でも子タスクが正常に完了することがあり、この契約ではそれを許容します。計算が長いループになる場合は、その処理の適切な間隔でキャンセルを確認してください。TaskGroupの仕様にあるとおり、グループは戻る前に投入済みのすべての子タスクが完了するのを待ちます。

throwingなタスクグループはどんな場合に使うべきですか？

既存の子タスクがすでにエラーをスローし、そのインターフェースを維持したい場合でも、throwingなタスクグループを使って個々の結果を収集できます。nextResult()を繰り返し呼び出し、返された成功または失敗を切り替えながら、オプショナルな結果が空になるまで処理します。next()とは異なり、このメソッドは子タスクのエラーを呼び出し元へスローせず、Result値を返します。完了順の動作については、AppleのnextResult()を参照してください。

この方法では、成功値と、入力を特定する必要のあるエラーの両方に入力の識別情報を含めてください。単なる算術エラーでは、複数の入力のどれが原因か分かりません。また、キャンセルを通常の計算失敗として扱わず、明確なキャンセル方針を選んでください。

入力ごとの完全なレポートが主な要件なら、結果を値として返す方法を使います。グループの外へエラーを伝播させるのが望ましいバッチ動作の場合や、既存のthrowingな処理をnextResult()経由で利用するほうが便利な場合は、throwingなグループを使います。

このローカルな計算タスクに必要な契約は単純です。成功した値を保持し、失敗した入力を記録し、収集を続けます。すべての結果に入力インデックスを含めると、計算の完了順が予測できなくても、最終レポートを役立つ形にできます。

出典