Swiftでセマフォ（semaphore）にさわってみよう

はじめに

最近セマフォ（semaphore）について振り返る機会があり、Swiftでの動作を確認する意味もあってSwiftでセマフォのサンプルソースコードを作成しました。

そのソースコードを用いて、「セマフォって聞いたことはあるけど？？」と思っているエンジニアのために本記事を書いてみました。

セマフォ（semaphore）とは

下記はwikipediaからの引用です

セマフォ（英: semaphore）とは、計算機科学において、並列プログラミング環境での複数の実行単位（主にプロセス）が共有する資源にアクセスするのを制御する際の、単純だが便利な抽象化を提供する変数または抽象データ型である。

セマフォを理解するキーワードは、**「並列」と「資源」**だと思います。

**「並列」な動作を行う複数のスレッドやプロセスが、共有する「資源」**を正しく利用できる（アクセス制御する）仕組みがセマフォです。

まずは、セマフォがない場合に何が起こるか、の例をSwiftのソースコード ^ code1で以下に提示します。

セマフォ（semaphore）がない場合

このソースコード内では、resourceが「資源」で、スレッド１、スレッド２が「並列」に動作します。
スレッドは何度も繰り返すタスクをもっていて、それを完了すればスレッドも終了します。
それぞれのタスクは、resourceがゼロ以上の場合は、自分のカウンターを１加算してresourceを１減算します。resourceがゼロになれば処理を終了します。

またDispatchQueue.global(qos: .background).async（ リファレンス👉） がそれぞれ使われており、qosが同じレベルのため、ほぼ均等のタイミングで実行状態が切り替わります。

func testHandleResourceWithoutSemaphoreQiita() throws {

    var resource: Int = 100
    var task1_counter: Int = 0
    var task2_counter: Int = 0

    // resourceがゼロ以上の場合は、自分のカウンターを１加算してresourceを１減算する
    func task1() -> Bool {

        var value = resource

        guard value > 0 else {
            return false
        }

        task1_counter += 1
        value -= 1
        resource = value

        return true
    }

    // resourceがゼロ以上の場合は、自分のカウンターを１加算してresourceを１減算する
    func task2() -> Bool {

        var value = resource

        guard value > 0 else {
            return false
        }

        task2_counter += 1
        value -= 1
        resource = value

        return true
    }

    // スレッド 1
    DispatchQueue.global(qos: .background).async {
        var executing = true
        while executing {
            executing = task1()
        }
    }

    // スレッド 2
    DispatchQueue.global(qos: .background).async {
        var executing = true
        while executing {
            executing = task2()
        }
    }

    // ２つのスレッドが完了するまで待つ

    print("task1_counter = \(task1_counter)")
    print("task2_counter = \(task2_counter)")
}

結果は、

task1_counter = 88
task2_counter = 102

だったり

task1_counter = 100
task2_counter = 0

だったり

task1_counter = 13
task2_counter = 101

します。

resourceがゼロ以上の場合は、自分のカウンターを１加算してresourceを１減算します。resourceがゼロになれば処理を終了します。

であればtask1_counterとtask2_counterの合計は100になりそうなものですが、そうはなりません（なることもあります）

これはなぜかというと、

var value = resource

guard value > 0 else {
    break
}

task2_counter += 1
value -= 1
resource = value

この処理を行っている間に、別の処理が割り込んできて、resourceの値を書き換えてしまうからです。
つまり、変数resourceはアクセス制御されていません。スレッドセーフではない、という言い方がよくされます。

セマフォ（semaphore）を使ってみる

関数の先頭で

let semaphore  = DispatchSemaphore(value: 1)

を宣言します。

DispatchSemaphore(value: 1) で、アクセス制御するリソースは１つだけだと宣言しています。リソースとは、

var resource: Int = 100

のことです。
この変数をアクセス制御できれば、おかしな現象は起こりません。

続いて、先ほどの割り込みされるtask1(), task2()の先頭でそれぞれ

defer {
    semaphore.signal()
}
semaphore.wait()

を追加します。
これで、関数に入った時点で semaphore.wait()が実行され、関数を抜けるタイミングで必ず semaphore.signal() が実行されます。
wait()ではDispatchSemaphoreで指定したカウンタを１減算し、signal()で１増加させます。カウンタがゼロの状態でwait()を呼び出したスレッドは、後述のようにBlocking状態となり処理は停止します。このスレッドは、signal()によってリソースが解放されるまでOSによってブロックされます。 [^ block]

[^ block]: https://ja.wikipedia.org/wiki/プロセス#ブロック状態 を参照

このような仕組みを導入することで、task1_counterとtask2_counterの合計は必ず100になることが保証されます。

スレッド/プロセスの状態遷移

Running（実行中）状態で、wait()を呼び出して運悪く誰かがリソースを使用していた場合は、Blocked状態になります。その後利用していたスレッドがsignal()を発行してリソースを解放すると、うまくOSスケジューラに拾ってもらえれば、Waiting状態に遷移し、無事Runningに復帰し、リソースにアクセスする権限を得られます。

上記の状態遷移図はiOSのものとは若干異なるのかもしれませんが、おおよそ同じような動きになるはずです。（正式な情報をお持ちでしたらお教えください）

セマフォがない場合のコンテキストスイッチ

アクセス制御がなされていない最初のコードでは、下記のようにスレッドが強制的にWaiting状態にスイッチされる（プリエンプションされてコンテキストスイッチが発生する）ことでtaskが中断しています。この中断の間に別のスレッドで、変数resourceが操作されるのが問題です。

繰り返しになりますが、ここで重要なのはセマフォがアクセス制御しているのは、リソースである変数resourceです。当該のリソースが何なのかを意識しないで漠然とセマフォを利用すると痛い目に遭います。要注意です。

セマフォがある場合のコンテキストスイッチ

セマフォを用いた状態遷移は下図のようになります。task1がセマフォを取得するとtask2はwait()を読んだ時点でBlocking状態に遷移します。のちにOSによって実行許可が出た場合は、Waitingを介してRunning状態に遷移し、wait()関数を抜けることになります。

セマフォが解放されたタイミングで次にどちらのスレッドが割り当てられるかは、OSのスケジューリングに寄ります。これを制御しようと思うのならば、

DispatchQueue.global(qos: .background).async

のQoS（リファレンス👉）を変更することである程度は可能です。QoSとはquality of serviceの略で、優先制御と呼ばれることが多いです。例えばインターネットで動画を途切れなく見るためにはQoSが有効なネットワークを利用することが必要だったりします。この場合は優先制御の他にも帯域制御が行われたりします。（参考記事👉）
ここでは、iOSのスレッドスケジューリングの優先制御に影響を与えるパラメータと言うことができるでしょう。

DispatchQueue.globalでは下記のようなオプションが用意されています。（リファレンス👉）

定義値	説明	補足
userInteractive	The quality-of-service class for user-interactive tasks, such as animations, event handling, or updating your app's user interface.	アニメーションやUIの更新などはモバイルアプリにとっては重要です。そこでiOSではuserInteractive には最も高い優先順位が与えられています^ highqos
userInitiated	The quality-of-service class for tasks that prevent the user from actively using your app.	ユーザが開始したアクション、またそれに付随する動作、結果に対して高い優先順位を与えるオプションです
default	The default quality-of-service class.	標準優先順位
utility	The quality-of-service class for tasks that the user does not track actively.	すぐに実行する必要がない処理に対して指定する
background	The quality-of-service class for maintenance or cleanup tasks that you create.	バックグラウンドでCPUリソースなどに余裕がある時に実行される優先順位
unspecified	The absence of a quality-of-service class.

（👉参考記事）

カウンティングセマフォ(Counting Semaphore)

let semaphore  = DispatchSemaphore(value: 1)

ではリソースが１つのために初期値として1を与えています。

リソースが１つだけのセマフォを**バイナリーセマフォ（Binary Semaphore）**と呼びます。厳密には違う(参考時事👉)のですが、**ミューテックス（Mutex）**と呼ばれることもあります。

リソースが２以上のセマフォはカウンティングセマフォ（計数セマフォ）と呼ばれ、アクセス制御に使うことはできません。
実は筆者も使ったことはありません😅

実際に2にしたらどうなるでしょうか。

サンプルソースコードの中でも実験していますが、興味がある方は試してみてください。🏃‍♂️🏃‍♂️🏃‍♂️🏃‍♂️

イベントフラグ（Event Flag）

イベントフラグにも触れておきたいと思います。
イベントフラグとは、主に組み込みプログラミングなどで利用されるμITRONで使われる用語です。実はPOSIXなどにもこの機能は明示されておらず、良い名称が無いのでイベントフラグという名称を使いたいと思います。

イベントフラグについてはこちらの記事ITRON入門 イベントフラグで学ぶタスク間同期・通信機能を参照してください。

サンプルソースコードに記載されているソースコードを紹介します。

このサンプルでは、DispatchSemaphore(value: 0)で初期値をゼロとしています。バイナリーセマフォがスレッドをBlockすることを利用して、別のスレッドにイベントフラグとしてをsignalを送信し、Blockを解除します。

    func testSingleEventFlag() throws {

        let semaphore  = DispatchSemaphore(value: 0)  // ⚠️ イベントフラグの初期化

        let expectation1 = XCTestExpectation(description: "expectation1")
        let expectation2 = XCTestExpectation(description: "expectation2")

        semLog.format("✳️ Start")

        DispatchQueue.global(qos: .background).async {

            usleep(1000_000)

            semaphore.signal() // ⚠️ イベントフラグを送信

            semLog.format("✳️ Sent EventFlag")

            expectation1.fulfill()
        }

        semaphore.wait() // ⚠️ イベントフラグを受信

        semLog.format("✴️ Recieved EventFlag")

        expectation2.fulfill()

        wait(for: [expectation1, expectation2], timeout: 10.0)
    }

こちらの実行結果は、下記のようになります（semLog.format()は自作のログ出力ツールです。サンプルソースコードの中にあります）

テストを開始してから約１秒後にbackgroundのスレッドからイベントフラグの送信が起こり、待ち受けていたメインスレッドがBlockingからRunningに状態遷移して続きの処理が実行された様子が分かります。

    /*
     ✳️ Start [18:37:25.837] [main]
     ✳️ Sent EventFlag [18:37:26.847] [com.apple.root.background-qos]
     ✴️ Recieved EventFlag [18:37:26.847] [main]
     */

リソースのアクセス制御という観点から見ると、最初に説明した手法が変数などを守るためにあるのに対して、こちらはスレッドそれ自身を対象にしています。そのためスレッドが、いつ生成されて、いつ消滅するのかを考慮して設計、実装しなければなりません。

上記のようなシンプルなものであれば良いのですが、大規模なマルチスレッドのシステムに動的に適用しようとすると、イベントフラグの初期化と送受信のタイミングが問題になることもあります。

デッドロック（Deadlock）/ リソーススタベーション（Resource Starvation）

最後に、バイナリセマフォ、ミューテックスで必ず取り上げられるデッドロックについても紹介いたします。

下記は、二つのスレッドが、互いの管理するリソースを参照して動作しようとしてデッドロックします。
大きな問題は、wait() → signal()の間に、外部の（しかも相互参照している）オブジェクトを呼び出していることです。
外部のオブジェクトが実は、自分のリソースを操作する関数を呼び出していると知らなければ、容易にデッドロックが発生します。

func testDeadlockBySemaphore() throws {

    var expectations: [XCTestExpectation] = [XCTestExpectation]()

    class SemThread {
        let semaphore = DispatchSemaphore(value: 1)
        var resource: Int = 0
        var other: SemThread?
        let expectation: XCTestExpectation
        let name: String

        // 生成時にセマフォを取得し、外部からのインクリメントをブロックする
        init(name: String) {
            self.name = name
            expectation = XCTestExpectation(description: name)
        }

        // otherに対してカウントアップ要求を出す。その後、外部からのインクリメントを許可
        func run() {
            DispatchQueue.global(qos: .background).async {

                self.increment()

                self.expectation.fulfill()
                semLog.format("⭐️\(self.name) completed")
            }
        }

        func increment() {

            semLog.format("⭐️\(self.name) will wait")
            semaphore.wait()

            usleep(100)

            semLog.format("⭐️\(self.name) make other increment")
            // 自セマフォを取得中に、相互参照している外部オブジェクトを利用する
            self.other?.increment()

            resource += 1
            semaphore.signal()
        }
    }

    let thread1 = SemThread(name: "thread1")
    let thread2 = SemThread(name: "thread2")

    expectations.append(thread1.expectation)
    expectations.append(thread2.expectation)

    // クロス参照する
    thread1.other = thread2
    thread2.other = thread1

    // スレッド開始
    thread1.run()
    thread2.run()

    // ２つのスレッドが完了するまで待つ（が、必ず失敗）
    wait(for: expectations, timeout: 5.0)
}

iOSではあまりセマフォを使う機会はないのですが、安易に導入すると、このようなデッドロックを引き起こす可能性があります。セマフォが本当に必要なのか、対象となるリソースは何なのか、相互参照はあるのか、wait() → signal()区間でreturnしていないか、などを十分に吟味して利用することが必要です。

iOSの場合、このようなプリミティブな操作を使うことなく、マルチスレッドを使いこなすために、DispatchQueueが用意されています。
セマフォの説明を延々した後でなんですが、まずDispatchQueueを使うことを検討した方が良いかもしれません。（自戒😇）

DispatchQueueとマルチスレッド操作については、別途記事を書いてみたいと思います。