SwiftUI APIリクエストのスレッドセーフな排他制御

最近、SwiftUIとFirebase Crashlyticsを連携してクラッシュ対策進めています。
Swiftはクラッシュログの内容から問題となっているコードの特定が難しく、特にマルチスレッドに起因するクラッシュなどはある程度あたりをつけて修正を入れ、しばらく経過を観察して効果を判断する対応となってしまうことも多く、わだかまりを抱えたまま進むこともしばしば・・

今回、排他制御に関するクラッシュについて解決の手段を探り当てたため、備忘録として記録しておきます。

クラッシュログの詳細

今回解消できたクラッシュログはこちらです。

specialized _ArrayBuffer._consumeAndCreateNew(bufferIsUnique:minimumCapacity:growForAppend:)
EXC_BAD_ACCESS (KERN_INVALID_ADDRESS)

調べると、配列への要素の追加や削除がスレッドセーフになっていないために発生するクラッシュのようです。

現在開発しているアプリでは、一部のAPIリクエスト時に複数APIを並行してリクエストしないよう排他処理を実装する必要があり、クラッシュが発生しているスクリーンの傾向からこのあたりに原因があることが濃厚でした。
よくよく精査してみると、実行待ちのメソッドをタスクキューとなる配列に溜めておき、実行完了後に配列から削除するような形で実装していたため、この処理がスレッドセーフではないことに気づきました。

対策

SwiftのArrayがどうやらスレッドセーフではないようで、複数のスレッドからひとつの配列にアクセスが重なるとクラッシュしてしまうようです。
いくつか対応方法はあるかと思いますが、今回はタスクキューを担っている配列をスレッドセーフなActorに置き換える対応となりました。

1. Swift Concurrency対応のSemaphoreをつくる

単にArrayによるタスクキュー管理からSemaphoreに置換できればよかったのですが、APIの処理をSwift Concurrencyに対応させていたため、asyncが使えないSemaphoreをそのまま使ってクリーンに解決できる手立てが見つかりませんでした。

そのため、Actorでasyncの使えるSemaphoreを自作する形で対応します。
こちらの記事を参考にさせていただき、追加で順番待ちを始めてから一定時間順番が来なかった場合はタイムアウトする形にしました。

AsyncSemaphore.swift

import Foundation

/// 並列処理の排他制御とタイムアウト処理を制御する
/// - Parameter value: カウンタの初期値
actor AsyncSemaphore {
    /// 実行待ちタスク数
    private(set) var count: Int
    /// 待機キューリスト
    private var waiters: [CheckedContinuation<Void, Never>] = []

    init(value: Int = 0) {
        self.count = value
    }

    /// 待機キューを追加する
    func append(waiter: CheckedContinuation<Void, Never>) {
        waiters.append(waiter)
    }

    /// カウンタを1減らし、0未満になった場合は0以上になるまで待機する
    /// - Parameter timeout: タイムアウト (秒)
    /// - Returns: タイムアウトせずに順序がきたか
    @discardableResult
    func wait(timeout: Double? = nil) async -> Bool {
        count -= 1
        if count >= 0 { return true }

        // signalが実行されるかタイムアウトするのを待つ
        let result = await withTaskGroup(of: Bool.self) { group in
            group.addTask {
                await withCheckedContinuation { continuation in
                    Task {
                        await self.append(waiter: continuation)
                    }
                }
                return true
            }
            if let timeout = timeout {
                group.addTask {
                    await Task.sleep(seconds: timeout)
                    return false
                }
            }

            let result = await group.next()!

            // タイムアウトした場合は処理をキャンセルして順番待ち終了扱いとする
            group.cancelAll()
            if !result {
                signal()
            }

            return result
        }
        return result
    }

    /// カウンタを増やし、処理が完了したことを通知する
    /// - Parameter count: 追加値
    func signal(count: Int = 1) {
        assert(count >= 1)
        self.count += count
        for _ in 0..<count {
            if waiters.isEmpty { return }
            waiters.removeFirst().resume()
        }
    }
}

2. セマフォを使用したタスク管理を組み込む

作成したセマフォを使ったタスク管理機構を作成します。
試しに、実行5秒後にコンソール出力をするexampleTaskメソッドを用意します。
exampleTaskメソッドにはセマフォを判定してから処理を順次実行するexclusiveTaskメソッドを組み込み、連続して呼び出されると前の呼び出しの処理が完了するのを待ったあと、さらに5秒待ってから出力を実行というように、順序よく処理が実行されます。
実際にはこの5秒の部分にAPIリクエスト処理が入り、APIのレスポンスを受け取る前に次のAPIリクエストが走ることのないように排他制御が入るというわけです。

ExampleSDK.swift

import Foundation

struct API {
    /// 排他処理を制御する (同時に実行するAPIは1つまで)
    static let semaphore = AsyncSemaphore(value: 1)

    /// コールバックの処理を順番に実行する
    /// 複数並列実行された場合は排他制御をする
    /// - Parameter execute: 処理内容
    static func exclusiveTask(execute: @escaping () async throws -> Void) async throws {
        // 他キューの実行完了待ち
        await semaphore.wait(timeout: 10)

        // タスクの実行
        try await execute()
        await semaphore.signal()
    }

    /// 5秒かけてメッセージをコンソール出力する
    /// - Parameter count: 実行カウント
    static func exampleTask(count: Int) async throws {
        try await exclusiveTask {
            // APIリクエストなど時間のかかる処理
            try await Task.sleep(nanoseconds: 5_000_000_000)
            print("[\(Date())] タスク完了 \(count)")
        }
    }
}

さらに動作確認用にexampleTaskメソッドを呼び出すボタンを配置したViewを用意します。

ContentView.swift

struct ContentView: View {
    @State var count = 0

    var body: some View {
        VStack {
            Button {
                count += 1
                print("[\(Date())] タスク予約 \(count)")
                Task {
                    do {
                        try await API.exampleTask(count: count)
                    }
                }
            } label: {
                Text("テスト")
            }
        }
    }
}

この実装でビルドしてテスト用のボタンを連打してみます。
すると、以下のようにコンソールに出力されます。

[2023-10-19 00:00:00 +0000] タスク1 予約
[2023-10-19 00:00:00 +0000] タスク2 予約
[2023-10-19 00:00:01 +0000] タスク3 予約
[2023-10-19 00:00:01 +0000] タスク4 予約
[2023-10-19 00:00:02 +0000] タスク5 予約
[2023-10-19 00:00:02 +0000] タスク6 予約
[2023-10-19 00:00:05 +0000] タスク1 完了
[2023-10-19 00:00:11 +0000] タスク2 完了
[2023-10-19 00:00:16 +0000] タスク3 完了
[2023-10-19 00:00:17 +0000] タスク4 完了
[2023-10-19 00:00:18 +0000] タスク6 完了
[2023-10-19 00:00:18 +0000] タスク5 完了

タスク3まではきちんと順番に5秒ずつ待って出力されていますが、タスク4からは諦めてタスク予約から10秒後に出力されているのが分かります。

タイムアウト時のハンドリングを追加する

今回はいろいろと実装要件もあったため、タイムアウト時に諦めてそのままタスク実行する緩めの排他処理として実装しましたが、タイムアウト時は他の処理に分岐したいことも多いかと思います。
スルーしているsemaphore.wait()の戻り値でタイムアウトの有無が返っているので、ここを判定して例外を投げるなどすればタイムアウト時はcatch句に入る挙動にもできます。

struct API {

    // ...

    static func exclusiveTask(execute: () async throws -> Void) async throws {
        // 他キューの実行完了待ち
        let isNotTimeout = await semaphore.wait(timeout: 10)
        if !isNotTimeout {
            throw NSError(domain: "TaskTimeout", code: 0)
        }

        // タスクの実行
        try await execute()
        await semaphore.signal()
    }
}

Button {
    count += 1
    print("[\(Date())] タスク予約 \(count)")
    Task {
        do {
            try await API.exampleTask(count: count)
        } catch {
            print("[\(Date())] \(error)")
        }
    }
} label: {
    Text("テスト")
}