Beyond the basics of structured concurrency

ハイライト

Swift の構造化された同時実行性は、自動キャンセル伝播、優先順位の継承、タスクツリーを介したタスクローカル値 (Task-Local Values) の配信を実装し、新しい機能を追加します。withThrowingDiscardingTaskGroupそしてDiscardingTaskGroupサブタスクの結果を収集する必要のないシナリオを最適化するため。

主な内容

同時実行コードに対処するのが最も難しいのは、ライフサイクル管理です。ネットワーク要求が送信され、ユーザーはページに切り替わります。リクエストを続行する必要がありますか?続行すると、存在しない UI が更新されることになり、良くてもリソースが無駄になり、最悪の場合クラッシュします。

Swift の構造化同時実行では、タスクツリーを使用してこの問題を解決します。構造化されたタスク (async letおよびタスクグループ) は親タスクとツリー階層を形成し、スコープを離れると自動的にキャンセルされます。構造化されていないタスク (Task.initそしてTask.detached) はこのレベルに参加していないため、手動で管理する必要があります。

（02:24）

非構造化から構造化へ

スープを作るときに、同時にスープを調理し、野菜を切り、肉をマリネする必要があるとします。構造化されていない文章:

func makeSoup(order: Order) async throws -> Soup {
    let boilingPot = Task { try await stove.boilBroth() }
    let choppedIngredients = Task { try await chopIngredients(order.ingredients) }
    let meat = Task { await marinate(meat: .chicken) }
    let soup = await Soup(meat: meat.value, ingredients: choppedIngredients.value)
    return await stove.cook(pot: boilingPot.value, soup: soup, duration: .minutes(10))
}

この書き方の問題点は次の 3 つです。Taskすべて構造化されておらず、ライフサイクルは次のようなものです。makeSoup関数には拘束関係がありません。もしmakeSoupキャンセルしても、これら 3 つのタスクは自動的にはキャンセルされません。

構造化された同時実行として書き直されると、次のようになります。

func makeSoup(order: Order) async throws -> Soup {
    async let pot = stove.boilBroth()
    async let choppedIngredients = chopIngredients(order.ingredients)
    async let meat = marinate(meat: .chicken)
    let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
    return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
}

async let作成されたタスクは構造化されており、makeSoup父と子の関係を築きます。makeSoupキャンセルされると、3 つのサブタスクは自動的にキャンセルされます。

（02:42）

タスクがキャンセルされました

キャンセルは協力です。設定解除のみisCancelledフラグを設定すると、タスクは強制的に停止されません。コードは以下について積極的にチェックする必要があります。

func makeSoup(order: Order) async throws -> Soup {
    async let pot = stove.boilBroth()

    guard !Task.isCancelled else {
        throw SoupCancellationError()
    }

    async let choppedIngredients = chopIngredients(order.ingredients)
    async let meat = marinate(meat: .chicken)
    let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
    return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
}

Task.checkCancellation()より簡潔な書き方で、キャンセルされた場合に直接スローされます。CancellationError：

func chopIngredients(_ ingredients: [any Ingredient]) async throws -> [any ChoppedIngredient] {
    return try await withThrowingTaskGroup(of: (ChoppedIngredient?).self) { group in
        try Task.checkCancellation()

        for ingredient in ingredients {
            group.addTask { await chop(ingredient) }
        }

        var choppedIngredients: [any ChoppedIngredient] = []
        for try await choppedIngredient in group {
            if let choppedIngredient {
                choppedIngredients.append(choppedIngredient)
            }
        }
        return choppedIngredients
    }
}

キーポイント: 親タスクをキャンセルすると、すべての子タスクが再帰的にキャンセルされます。タスクグループに電話をかけるgroup.cancelAll()すべてのアクティブなサブタスクと今後追加されるサブタスクはキャンセルされます。

（04:32）

一時停止時のキャンセル処理

タスクが保留中で待機しているときAsyncSequence次の要素をポーリングしてキャンセルステータスを確認する方法はありません。withTaskCancellationHandlerこのシナリオは次のように処理されました。

public func next() async -> Order? {
    return await withTaskCancellationHandler {
        let result = await kitchen.generateOrder()
        guard state.isRunning else {
            return nil
        }
        return result
    } onCancel: {
        state.cancel()
    }
}

プロセッサとメインロジック間の共有状態を解除するには、アトミック操作によって保護する必要があります。ここで使用されますManagedAtomic：

private final class OrderState: Sendable {
    let protectedIsRunning = ManagedAtomic<Bool>(true)
    var isRunning: Bool {
        get { protectedIsRunning.load(ordering: .acquiring) }
        set { protectedIsRunning.store(newValue, ordering: .relaxed) }
    }
    func cancel() { isRunning = false }
}

（05:47）

＃＃詳細

同時実行数を制限する

タスクグループのデフォルトはそれぞれaddTask同時実行性が制御されていない新しいタスクを作成します。もしingredients100 個ある場合は、100 個のチョッピングタスクが同時に作成されます。

同時実行を制限するモード: 最初に固定数のタスクを開始し、各タスクが完了した後に新しいタスクを開始します。

func chopIngredients(_ ingredients: [any Ingredient]) async -> [any ChoppedIngredient] {
    return await withTaskGroup(of: (ChoppedIngredient?).self) { group in
        let maxChopTasks = min(3, ingredients.count)
        for ingredientIndex in 0..<maxChopTasks {
            group.addTask { await chop(ingredients[ingredientIndex]) }
        }

        var choppedIngredients: [any ChoppedIngredient] = []
        var nextIngredientIndex = maxChopTasks
        for await choppedIngredient in group {
            if nextIngredientIndex < ingredients.count {
                group.addTask { await chop(ingredients[nextIngredientIndex]) }
                nextIngredientIndex += 1
            }
            if let choppedIngredient {
                choppedIngredients.append(choppedIngredient)
            }
        }
        return choppedIngredients
    }
}

このモデルの核となるロジックは次のとおりです。

一番最初に始めるmaxChopTasksタスク 2.でfor awaitループ内ではタスクが完了するたびに、未処理のタスクがあれば新たなタスクが追加されます。
常に最大限のものを維持するmaxChopTasksタスクが実行中です

（10:55）

DiscardingTaskGroup

従来のタスクグループは、すべてのサブタスクの結果を収集する必要があります。ただし、一部のシナリオでは「すべてのタスクが完了した」ことのみを考慮し、結果を必要としません。たとえば、レストランサービスでは、複数のシェフが同時に働きます。 1つのタスクが完了する（終了時間が過ぎる）限り、サービス全体が終了します。

func run() async throws {
    try await withThrowingTaskGroup(of: Void.self) { group in
        for cook in staff.keys {
            group.addTask { try await cook.handleShift() }
        }

        group.addTask {
            try await Task.sleep(for: shiftDuration)
        }

        try await group.next()
        group.cancelAll()
    }
}

この書き方の問題は、明示的に呼び出す必要があることです。cancelAll()、そしてもしgroup.next()終了タスクの代わりにシェフタスクを返すには、追加のロジックが必要です。

Swift 5.9の新機能withThrowingDiscardingTaskGroup：

func run() async throws {
    try await withThrowingDiscardingTaskGroup { group in
        for cook in staff.keys {
            group.addTask { try await cook.handleShift() }
        }

        group.addTask {
            try await Task.sleep(for: shiftDuration)
            throw TimeToCloseError()
        }
    }
}

DiscardingTaskGroupサブタスクの結果は収集されません。 1 つのサブタスクがエラーをスローするか、すべてが完了する限り、グループ全体が終了します。マニュアルは不要ですcancelAll()。

（11:56）

Task-Local Values

タスクローカル値を使用すると、タスクツリー内でコンテキスト情報を自動的に伝達できます。サブタスクは、親タスクのタスクローカル値を自動的に継承します。

actor Kitchen {
    @TaskLocal static var orderID: Int?
    @TaskLocal static var cook: String?

    func logStatus() {
        print("Current cook: \(Kitchen.cook ?? "none")")
    }
}

let kitchen = Kitchen()
await kitchen.logStatus()           // "Current cook: none"
await Kitchen.$cook.withValue("Sakura") {
    await kitchen.logStatus()       // "Current cook: Sakura"
}
await kitchen.logStatus()           // "Current cook: none"

@TaskLocalタグパスの静的属性$プレフィックスアクセス。withValue値はクロージャの実行中に設定され、クロージャの終了後に自動的に復元されます。

（14:10）

MetadataProvider を使用してロギングを簡素化する

サーバーサイド開発では、各関数は注文 ID、シェフ名などのコンテキストを記録する必要があります。タスクローカル値がない場合、これらのコンテキストはパラメーターとして渡す必要があります。

func makeSoup(order: Order) async throws -> Soup {
    log.debug("Preparing dinner", [
        "cook": "\(self.name)",
        "order-id": "\(order.id)",
        "vegetable": "\(vegetable)",
    ])
}

使用MetadataProviderタスクローカル値は各ログに自動的に挿入できます。

let orderMetadataProvider = Logger.MetadataProvider {
    var metadata: Logger.Metadata = [:]
    if let orderID = Kitchen.orderID {
        metadata["orderID"] = "\(orderID)"
    }
    return metadata
}

let chefMetadataProvider = Logger.MetadataProvider {
    var metadata: Logger.Metadata = [:]
    if let chef = Kitchen.chef {
        metadata["chef"] = "\(chef)"
    }
    return metadata
}

let metadataProvider = Logger.MetadataProvider.multiplex([
    orderMetadataProvider,
    chefMetadataProvider
])

LoggingSystem.bootstrap(
    StreamLogHandler.standardOutput,
    metadataProvider: metadataProvider
)

構成後、ロギングコードは次のように簡略化されます。

func makeSoup(order: Order) async throws -> Soup {
    logger.info("Preparing soup order")
    async let pot = stove.boilBroth()
    async let choppedIngredients = chopIngredients(order.ingredients)
    async let meat = marinate(meat: .chicken)
    let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
    return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
}

orderIDそしてchefこれはすべてのログに自動的に表示されるため、手動で渡す必要はありません。

（16:17）

分散トレーシング

Swift Distributed Tracing ライブラリと組み合わせると、次のことが可能になります。withSpanサーバー側コードのパフォーマンス分析:

func makeSoup(order: Order) async throws -> Soup {
    try await withSpan("makeSoup(\(order.id)") { span in
        async let pot = stove.boilWater()
        async let choppedIngredients = chopIngredients(order.ingredients)
        async let meat = marinate(meat: .chicken)
        let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
        return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
    }
}

カスタムプロパティも追加できます。

func makeSoup(order: Order) async throws -> Soup {
    try await withSpan(#function) { span in
        span.attributes["kitchen.order.id"] = order.id
        async let pot = stove.boilWater()
        async let choppedIngredients = chopIngredients(order.ingredients)
        async let meat = marinate(meat: .chicken)
        let soup = try await Soup(meat: meat, ingredients: choppedIngredients)
        return try await stove.cook(pot: pot, soup: soup, duration: .minutes(10))
    }
}

withSpan作成されたスパンにはサブタスクの追跡情報が自動的に含まれ、完全なコールチェーンが形成されます。

（20:30）

重要なポイント

ネットワークリクエストに対する統合キャンセルチェックを追加

データ層の各リクエストメソッドの先頭に追加します。try Task.checkCancellation()、ページに切り替えた後、ユーザーが無駄なネットワーク要求を実行し続けるのを防ぎます。

入口：URLSession.data(for:)カプセル化層にキャンセルチェックを追加します。協力するasync let複数のリクエストが開始されると、親タスクのキャンセルがすべての子リクエストに自動的に伝播されます。

タスクローカル値を使用してリクエスト追跡 ID を実装します

サーバーまたはクライアントのいずれかが使用できます@TaskLocal static var requestID: String?現在のリクエストをマークします。すべてのログにはこの ID が自動的に含まれるため、トラブルシューティング時にリクエスト ID でログをフィルタリングできます。

入口：リクエスト入口にてawait RequestContext.$requestID.withValue(UUID().uuidString)、ログ構成MetadataProvider自動的に読み上げます。

同時ダウンロード数を制限する

画像リストページに同時に多くの画像が読み込まれる場合は、タスクグループの「1 つを完了して 1 つ追加」モードを使用して同時実行数を制限し、同時に開始されるリクエストが多すぎることによって引き起こされるネットワークの混雑を回避します。

入口：参考chopIngredients電流制限モード、置くmaxChopTasksに変更しますmaxDownloadTasks(3 か 5 のように)。

DiscardingTaskGroup を使用してバックグラウンドタスクを管理します

ハートビート検出、ログレポート、位置更新などのバックグラウンドタスクでは、結果を収集する必要はありません。使用withDiscardingTaskGroup管理。メインタスクがキャンセルされると、すべてのバックグラウンドタスクが自動的に終了します。

入口：置くTask { ... }に変更しますgroup.addTask { ... }、使用withDiscardingTaskGroupパック。