長期間使い続けるFoundationのOperationQueueを1つ用意し、計算を送信する前にmaxConcurrentOperationCountを設定します。ボタンを押すたびに1件のリクエストを受け取るMacユーティリティでは、同時に実行する処理を制限しながら、後続のリクエストを待機させられます。

このチュートリアルでは、そのデスクトップタスク用の計算エンジンと、動作を確認するスタンドアロンプログラムを作成します。各リクエストには合成整数が含まれ、結果はすべてメモリ内に保持されます。計算はオペレーション内で同期的に実行されるため、オペレーションの存続期間が実際の処理全体をカバーします。

個別に届くリクエストにはどの並行処理ツールが適していますか？

OperationQueueはオペレーションを個別に受け付けます。また、ブロックベースのaddOperation(_:)メソッドは、送信された各クロージャをオペレーションでラップします。リクエストをまたいで存続するオブジェクトにキューを保持し、届いた項目ごとにクロージャを1つ送信します。AppleのaddOperation(_:)のドキュメントに、この送信メソッドが記載されています。

maxConcurrentOperationCountには、2などの正の上限を設定します。Appleによると、このプロパティは、そのキューのオペレーションのうち同時に実行できる最大数を定めます。この上限はその特定のキューに適用されます。リクエストごとに別のキューを作ると、共有の上限が機能しません。後から上限を下げても、すでに実行中のオペレーションはそのまま実行されます。Appleの同時実行数上限に関するドキュメントでは、これらの制約が説明されています。

Swiftのタスクグループは、存続期間が異なります。withTaskGroupのスコープ内に属し、それを作成したタスクの外部からは使用できず、戻る前に子タスクの完了を待ちます。そのため、今後届く無関係なボタン操作のためにタスクグループをプロパティとして保存することはできません。この同期的な計算エンジンには、保持されるオペレーションキューが必要な存続期間と明示的な同時実行数設定を備えています。Swift標準ライブラリのタスクグループに関するドキュメントに、これらの制約が記載されています。

作業の構成 適した方法
独立した同期計算が個別に届く OperationQueueを1つ保持し、計算ごとにオペレーションを1つ送信します。
親の非同期オペレーションが有限個の子タスクを管理する 親のスコープ内でタスクグループを使います。数値で上限を設ける必要がある場合は、タスクの追加を明示的に制御します。
作業でデスクトップのインターフェースを更新する必要がある 計算を別に実行し、その結果をMainActorに隔離されたコードへ渡します。

最後の行は結果の受け渡しについてです。Swiftの言語ガイドでは、インターフェースの状態を扱う場所としてメインアクターを示し、メインアクターに明示的に隔離されたクロージャの例を紹介しています。負荷の高い演算は、その受け渡し処理の外で行います。The Swift Programming Language: Concurrencyで、メインアクターによる隔離について説明されています。

計算内容と完了時の契約を定義する

このサンプルは、1から1_000_000までの整数識別子と反復回数を受け付けます。反復回数がnの場合、1からnまでのすべての整数の二乗を合計します。結果はUInt64値と識別子を組み合わせて返します。識別子は出力を区別するもので、算術演算には影響しません。

これらは、このプログラム用に選んだ例示的な上限です。受け付ける最大入力での結果は333333833333500000となり、UInt64に収まります。無効なリクエストは送信前に検証されるため、ただちにエラーをスローし、キューにオペレーションは作成されません。入力制限により、ループ範囲も有効に保たれます。

受理された各リクエストについて、オペレーションの完了時に結果が1つ保存されます。結果は配列に蓄積されるため、識別子が重複しても以前の結果は上書きされません。このデモでは昇順の異なる識別子を使い、読みやすい出力になるよう識別子でソートします。この表示順は、実行順を示すものではありません。

この契約は、キューが利用可能なままであり、オペレーションが最後まで実行されることを前提としています。この例ではキャンセルを扱いません。後からキャンセルを追加する場合は、未完了リクエストの結果を別途定義する必要があります。Appleのドキュメントによると、キャンセルは停止要求であり、実行中の処理はその要求に応じる必要があります。AppleのOperationQueueのドキュメントに、オペレーションの存続期間とキャンセルについて説明があります。

共有の上限を設けた完全なプログラムを作る

Swift 6.2以降のコンパイラと、Xcodeまたはコマンドラインツールから利用できるmacOS SDKを搭載したMacを使います。次の完全なプログラムをmain.swiftとして保存します。以下のコマンドでは非隔離のデフォルト隔離を明示的に選択し、計算ヘルパーがアプリターゲットの隔離設定に依存しないようにします。Swiftのコンパイラ設定については、SE-0466: Control default actor isolation inferenceに記載されています。

結果ログではNSLockを使って、配列とアクティビティカウンターの両方を保護します。@unchecked Sendableへの適合は、その同期を実装側の責任とします。変更可能な状態へのアクセスは、すべて同じロックを取得します。AppleのSendableにこの責任が、NSLockにクリティカルセクションの保護が説明されています。

import Foundation

enum InputError: Error {
    case invalidIterations
}

struct Calculation: Sendable {
    let id: Int
    let iterations: Int
}

func sumOfSquares(_ job: Calculation) -> UInt64 {
    var sum: UInt64 = 0
    for integer in 1...job.iterations {
        let value = UInt64(integer)
        sum += value * value
    }
    return sum
}

final class ExecutionLog: @unchecked Sendable {
    private let lock = NSLock()
    private var active = 0
    private var peak = 0
    private var rows: [(id: Int, value: UInt64)] = []

    func begin() {
        lock.lock()
        defer { lock.unlock() }
        active += 1
        peak = max(peak, active)
    }

    func finish(id: Int, value: UInt64) {
        lock.lock()
        defer { lock.unlock() }
        rows.append((id: id, value: value))
        active -= 1
    }

    func snapshot() -> (
        rows: [(id: Int, value: UInt64)], peak: Int
    ) {
        lock.lock()
        defer { lock.unlock() }
        return (rows, peak)
    }
}

final class CalculationPool {
    private let queue = OperationQueue()
    private let log = ExecutionLog()

    init(limit: Int) {
        precondition(limit > 0)
        queue.name = "Synthetic calculations"
        queue.maxConcurrentOperationCount = limit
    }

    func submit(id: Int, iterations: Int) throws {
        guard (1...1_000_000).contains(iterations) else {
            throw InputError.invalidIterations
        }
        let job = Calculation(id: id, iterations: iterations)
        queue.addOperation { [log] in
            log.begin()
            let value = sumOfSquares(job)
            log.finish(id: job.id, value: value)
        }
    }

    // Call only after the producer has stopped submitting work.
    // This blocking method is for the command-line demonstration.
    func finishForDemo() -> (
        rows: [(id: Int, value: UInt64)], peak: Int
    ) {
        queue.waitUntilAllOperationsAreFinished()
        return log.snapshot()
    }
}

@main
struct Demo {
    static func main() throws {
        let limit = 2
        let pool = CalculationPool(limit: limit)

        try pool.submit(id: 1, iterations: 100_000)
        try pool.submit(id: 2, iterations: 200_000)
        try pool.submit(id: 3, iterations: 300_000)

        let report = pool.finishForDemo()
        let rows = report.rows.sorted { $0.id < $1.id }
        precondition(rows.map { $0.id } == [1, 2, 3])
        precondition(rows.map { $0.value } == [
            UInt64(333338333350000),
            UInt64(2666686666700000),
            UInt64(9000045000050000)
        ])
        precondition(report.peak >= 1 && report.peak <= limit)

        for row in rows {
            print("Job \(row.id): \(row.value)")
        }
        print("Observed peak: \(report.peak)")
    }
}

main.swiftがあるディレクトリからコンパイルして実行します。

swiftc -swift-version 6 -default-isolation nonisolated \
  -parse-as-library main.swift -o calculation-demo
./calculation-demo

結果として、次の3行が表示されます。

Job 1: 333338333350000
Job 2: 2666686666700000
Job 3: 9000045000050000

最後の行には、観測されたピーク値として1または2のいずれかが表示されます。上限は重複実行を許可しますが、必ず重複するとは限りません。期待される合計値は、整数演算と公式n * (n + 1) * (2 * n + 1) / 6を使って独立に確認しました。調査環境にはSwift実行ファイルがインストールされていなかったため、Swiftプログラムは確認しましたがコンパイルできませんでした。

この上限が実際の計算にも適用されるのはなぜですか？

各オペレーションはアクティビティカウンターを開始し、計算全体を同期的に実行し、結果を記録してから戻ります。このクロージャから別の非同期計算が開始されることはありません。そのため、実行中のオペレーションを制限すると、これらの計算も制限されます。

プログラムを応用する際も、この関係を保ってください。オペレーションが別のTaskを作成してすぐ戻るだけの場合、キューから見るとオペレーションは完了していますが、そのタスクは計算を続けている可能性があります。その場合、オペレーションの上限は起動用の短いクロージャだけを制御します。これは、ドキュメントに記載されたオペレーションの存続期間とブロック送信APIから導かれます。AppleのOperationQueueの概要とaddOperation(_:)に、それぞれの仕組みが定義されています。

ログのロックが保護するのは、短い記録処理の区間だけです。sumOfSquares(_:)の実行中にロックを保持すると、計算へのアクセスが直列化され、意図した重複実行を妨げます。ロックの取得と解放は、いずれも1つの同期メソッド内で行われます。これは、NSLockをロックしたスレッドと同じスレッドでアンロックするというAppleの要件に沿っています。AppleのNSLockのドキュメントに、この要件が説明されています。

応用したプログラムを確認するには、同じ3件のリクエストをlimitを1に設定して実行し、その後2に戻します。どちらの実行でも、3件すべてについて正確な結果が保持されなければなりません。直列実行ではピーク値が1になるはずです。2回目の実行では2以下でなければなりません。これらの確認では、特定のスケジュール順序を前提とせずに、結果の保持と観測された計算数の上限を検証できます。

インターフェースをブロックせずにデスクトップの操作を受け取る

デスクトップの計算機能を所有するコンポーネントに、CalculationPoolを1つ保持します。操作が届くたびにsubmit(id:iterations:)を1回呼び出します。操作ハンドラー内でプールを作り直さないでください。新しいインスタンスごとに独立したキューと上限ができてしまいます。

このデモでは、コマンドラインプロセスが終了前に結果を収集できるよう、最後の送信後にのみfinishForDemo()を呼び出します。このメソッドは呼び出し元のスレッドをブロックします。デスクトップアプリでは、ボタンハンドラーやメインアクターからこれを呼び出さず、結果ごとに受け渡す方法を使います。AppleはwaitUntilAllOperationsAreFinished()のブロック動作を明示的に説明しています。

インターフェースに統合するには、MainActorに隔離され、@Sendableとマークされた結果コールバックを送信APIに追加します。オペレーション内で結果を計算した後、明示的な@MainActorクロージャを使ってTaskをスケジュールし、そのコールバックを呼び出します。不変の結果値とコールバックをキャプチャします。そのクロージャ内では表示の更新だけを行い、算術演算はオペレーション内で続けます。Swiftのドキュメントでは、メインアクターのクロージャ構文とSendableクロージャのキャプチャの両方が説明されています。

このインターフェース版では、完了の定義をさらに設ける必要があります。計算の終了と結果の表示は別々のイベントです。コマンドラインプログラムは、計算と同期的な結果保存を待ちます。別途スケジュールされたインターフェースのコールバックは、その後に実行される場合があります。

送信が計算の処理速度を上回るとどうなりますか？

同時実行数の設定が制限するのは実行中のオペレーションです。追加で送信された処理は、キューに残ることがあります。また、このサンプルでは完了した結果を格納する配列に上限がなく、増え続ける可能性があります。そのため、この例で制限できるのは計算の同時実行数であり、メモリ使用量やバックログの総量ではありません。Appleのドキュメントによると、キューはオペレーションが完了するまでそれを保持します。OperationQueueの概要に、保持の仕組みが説明されています。

デスクトップ機能で継続的に大量のリクエストが届く可能性がある場合は、拡張する前に受付方針を決めます。未完了リクエスト数を固定して超過分を拒否する、さらなる送信を一時停止する、またはその動作が明確に意図されている場合には、保留中の合成プレビューを新しいリクエストで置き換える、といった方法があります。受付数と完了時の更新は、1つの同期された管理主体で調整します。実行中の作業数の上限だけでは、こうした方針は実装できません。

個別に届く同期リクエストに必要な設計はシンプルです。キューを1つ保持し、その上限を設定し、各計算をオペレーション内で完結させ、共有結果を同期します。デスクトップの操作と接続するときも、これらの境界を維持し、リクエストの到着パターンに応じて明示的なバックログ方針を追加します。

参照資料