RxBlocking Deep Dive

**「RxBlockingってどうやってスレッドをブロックしてるんだろう？:thinking:」**

最近RxTestとRxBlockingについての記事を書いたのですが、記事を書くためにRxBlockingについて調べていく中でそんな疑問を持ちました。
RxBlocking自体はソースコードも少なくコンパクトなライブラリなのですが、その内部で使われている仕組みが意外と奥が深く面白かったので紹介したいと思います。

**2018/11/05 追記**
[Otemachi.swift #02](https://nikkei.connpass.com/event/98887/)というイベントで話したときのスライドを載せておきます。
5分で本テーマを話さなければならなかったため、本記事よりも説明を簡潔にしたり図を入れたりしているので、こちらの方が内容をつかみやすいかもしれません。

<script async class="speakerdeck-embed" data-slide="19" data-id="9d89da93b890493ab5770a1c3259272e" data-ratio="1.77777777777778" src="//speakerdeck.com/assets/embed.js"></script>

## RxBlockingとは？
RxBlockingを使ったテストコードは、例えば以下のように書きます。

```swift
// 非同期にイベントが発行されるObservable
let asyncObservable = Observable.of(10, 20, 30)
    .observeOn(SerialDispatchQueueScheduler(qos: .background))

let elements = try! asyncObservable.toBlocking().toArray()
expect(elements).to(equal([10, 20, 30]))
```

`asyncObservable`は非同期にイベントが発行されるObservableです。
非同期なので普通にサブスクライブして受信したイベントを検証するというテストは書けません。
そんなときに便利なのがRxBlockingです。

RxBlockingをインポートしているファイル内では、Observableに`toBlocking()`というメソッドが生えています。
`toBlocking()`はObservableを`BlockingObservable`に変換するメソッドです。
`BlockingObservable`の`toArray()`メソッドは、カレントスレッドをブロックし、`asyncObservable`をサブスクライブし、全てのイベントを受信したらその全ての要素を返します。

RxBlockingを使えば、非同期にイベントが発行されるObservableのテストが簡単に書けます。

こちらの記事も読んでみてください。
フィードバックお待ちしています。
[RxTest、RxBlockingによるテストパターン](https://qiita.com/takehilo/items/09f4a3077e441e5bb9de)

## スレッドをブロックする仕組み
では、RxBlockingはどのようにしてスレッドをブロックしているのでしょうか？

### materializeResult()
`toArray()`メソッドは、`BlockingObservable`のプライベートな関数である`materializeResult()`メソッドを内部で呼びます。
フルのソースを載せると長いので、ここでは説明に必要な部分を抜粋します。
[ソースはこちら](https://github.com/ReactiveX/RxSwift/blob/master/RxBlocking/BlockingObservable%2BOperators.swift)

```swift
fileprivate func materializeResult(max: Int? = nil, predicate: @escaping (E) throws -> Bool = { _ in true }) -> MaterializedSequenceResult<E> {
    var elements: [E] = Array<E>()
    var error: Swift.Error?
        
    let lock = RunLoopLock(timeout: timeout)
    ...
```

`elements`はソースとなるストリーム（先程の例でいうところの`asyncObservable`）から受信したイベントの要素を格納するための変数です。
`error`はストリームがエラーイベントを発行した場合にそのエラーを格納するための変数です。

スレッドをブロックする仕組みの肝は`RunLoopLock`クラスです。
`RunLoopLock`については後述しますので、ここでは単にスレッドをブロックする機能を持つクラスくらいに思っていてください。

`materializeResult()`の先に進みます。

```swift
    ...
    lock.dispatch {
        let subscription = self.source.subscribe { event in
            if d.isDisposed {
                return
            }
            switch event {
            case .next(let element):
                do {
                    if try predicate(element) {
                        elements.append(element)
                    }
                    if let max = max, elements.count >= max {
                        d.dispose()
                        lock.stop()
                    }
                } catch (let err) {
                    error = err
                    d.dispose()
                    lock.stop()
                }
            case .error(let err):
                error = err
                d.dispose()
                lock.stop()
            case .completed:
                d.dispose()
                lock.stop()
            }
        }

        d.setDisposable(subscription)
    }
    ...
```

`lock.dispatch()`は引数として渡されたクロージャを、言葉として適切ではないかもしれませんが、ブロックしたスレッド内で実行するメソッドです。
クロージャのコードをおおまかに説明すると、ソースとなるストリームをサブスクライブし、`next`イベントを受信したら`elements`に追加し、`completed`イベントを受信したら`lock.stop()`を呼んでスレッドのブロックを解除するということをやっています。
なお、`error`イベントを受信した場合はすぐにブロックを解除します。

`materializeResult()`の最後の部分です。

```swift
    ...
    do {
        try lock.run()
    } catch (let err) {
        error = err
    }
    ...
```

`lock.run()`はスレッドをブロックするメソッドです。
詳細は後述しますが、このメソッドを呼ぶことで実行ループに入ります。

### RunLoopLock
スレッドをブロックする機能の肝となるのは、RxBlocking内で定義されている`RunLoopLock`クラスです。
[ソースはこちら](https://github.com/ReactiveX/RxSwift/blob/master/RxBlocking/RunLoopLock.swift)

`RunLoopLock`クラスは、実行ループを制御するCore Foundationフレームワークの`CFRunLoop`クラスおよび`CFRunLoop`関連関数群のラッパークラスです。
実行ループとは、スレッド上で非同期に届くイベントを管理するための仕組みです。
名前の通りループ状態に入り、例えばユーザの画面操作等のイベントを受信すると、対応するハンドラにイベント渡して処理をするということを繰り返します。

実行ループはスレッドごとに1つ実行することができますが、iOSアプリケーションは起動シーケンスの中でメインスレッドで実行ループを実行済みなので、メインスレッドの実行ループを使用する場合は明示的に実行する必要はありません。
`RunLoopLock`クラスはカレントスレッドの実行ループを利用するようになっているので、これをメインスレッドで呼び出せばメインスレッドの実行ループを制御できるようになります。

それでは`RunLoopLock`のコードを読んでいきましょう。
まずはイニシャライザです。

```swift
init(timeout: RxTimeInterval?) {
    _timeout = timeout
    _currentRunLoop = CFRunLoopGetCurrent()
}
```

`_timeout`はスレッドをブロックする時間のタイムアウト値です。
タイムアウト時間を越えるとブロックが解除されます。

`_currentRunLoop`にはカレントスレッドの`CFRunLoop`オブジェクトが格納されます。
ここではメインスレッドの`CFRunLoop`オブジェクトを想定することとします。

続いて、`materializeResult()`メソッド内で呼ばれていた`dispatch()`メソッドです。
`materializeResult()`では、ストリームをサブスクライブして全てのイベントを受信したら実行ループを抜けるという処理のクロージャを渡していました。

```swift
func dispatch(_ action: @escaping () -> ()) {
    CFRunLoopPerformBlock(_currentRunLoop, runLoopModeRaw) {
        if CurrentThreadScheduler.isScheduleRequired {
            _ = CurrentThreadScheduler.instance.schedule(()) { _ in
                action()
                return Disposables.create()
            }
        }
        else {
            action()
        }
    }
    CFRunLoopWakeUp(_currentRunLoop)
}
```

`CFRunLoopPerformBlock()`は指定した実行ループにコードブロックを渡して実行させるメソッドです。
ここではメインスレッドの実行ループ内で`action()`クロージャを実行するようにしています。

`CFRunLoopWakeUp()`は`CFRunLoopPerformBlock()`で登録したコードブロックをすぐに実行するために必要なようです。

```swift
func stop() {
    if AtomicIncrement(&_calledStop) != 1 {
        return
    }
    CFRunLoopPerformBlock(_currentRunLoop, runLoopModeRaw) {
        CFRunLoopStop(self._currentRunLoop)
    }
    CFRunLoopWakeUp(_currentRunLoop)
}
```

`stop()`メソッドは、実行ループを終了させる`CFRunLoopStop()`メソッドを呼び出します。
これにより、スレッドのブロックが解除されます。

そして最後、`run()`メソッドです。

```swift
func run() throws {
    if AtomicIncrement(&_calledRun) != 1 {
        fatalError("Run can be only called once")
    }
    if let timeout = _timeout {
        #if os(Linux)
            switch Int(CFRunLoopRunInMode(runLoopModeRaw, timeout, false)) {
            case kCFRunLoopRunFinished:
                return
            case kCFRunLoopRunHandledSource:
                return
            case kCFRunLoopRunStopped:
                return
            case kCFRunLoopRunTimedOut:
                throw RxError.timeout
            default:
                fatalError("This failed because `CFRunLoopRunResult` wasn't bridged to Swift.")
            }
        #else
            switch CFRunLoopRunInMode(runLoopMode, timeout, false) {
            case .finished:
                return
            case .handledSource:
                return
            case .stopped:
                return
            case .timedOut:
                throw RxError.timeout
            }
        #endif
    }
    else {
        CFRunLoopRun()
    }
}
```

`CFRunLoopRun()`メソッドは、カレントスレッドの実行ループを実行するメソッドです。
`CFRunLoopRun()`メソッドはカレントスレッドの実行ループを再帰的に実行することが可能で、`CFRunLoopRun()`メソッドの呼び出しはカレントスレッドのコールスタックに追加されます。

![callstack.png](https://qiita-image-store.s3.amazonaws.com/0/81715/55680760-7389-8474-ca1b-f2838d9d5434.png)

`CFRunLoopRun()`メソッドは`CFRunLoopStop()`メソッドを呼ばない限りループし続け、呼び出し元に戻らないので、ここでスレッドがブロックされることになります（タイムアウトを指定している場合は`CFRunLoopRunInMode()`が呼ばれ、タイムアウトしたら強制的にブロックが解除されます）。


ではどこで`CFRunLoopStop()`メソッドが呼ばれるかと言うと、`materializeResult()`のこの部分です。

```swift:再掲
    ...
    lock.dispatch {
        let subscription = self.source.subscribe { event in
            if d.isDisposed {
                return
            }
            switch event {
            case .next(let element):
                do {
                    if try predicate(element) {
                        elements.append(element)
                    }
                    if let max = max, elements.count >= max {
                        d.dispose()
                        lock.stop()
                    }
                } catch (let err) {
                    error = err
                    d.dispose()
                    lock.stop()
                }
            case .error(let err):
                error = err
                d.dispose()
                lock.stop()
            case .completed:
                d.dispose()
                lock.stop()
            }
        }

        d.setDisposable(subscription)
    }
    ...
```

全てのイベントが到着し、最後に`completed`イベントが発行されたら、`RunLoopLock`クラスの`stop()`メソッドが呼ばれます。
`stop()`メソッドは`CFRunLoopStop()`を呼ぶので、ここで実行ループが終了し、呼び出し元に戻ります。

呼び出し元は`materializeResult()`メソッドのこの部分でした。

```swift
    do {
        try lock.run()
    } catch (let err) {
        error = err
    }

    if let error = error {
        return MaterializedSequenceResult.failed(elements: elements, error: error)
    }

    return MaterializedSequenceResult.completed(elements: elements)
}
```

`materializeResult()`メソッドは最後に取得した要素を関連値として持つEnumを返して終了します。
冒頭に出てきた`BlockingObservable`の`toArray()`メソッドは、このEnumから値を取り出して返すメソッドです。

RxBlockingがスレッドをブロックする仕組みの説明は以上となります。

## まとめ
RxBlockingでは、`CFRunLoop`というかなり低レイヤな仕組みを使ってスレッドのブロックをしていることがわかりました。
そして、ブロッキングの仕組みの正体は、メインスレッドで再帰的に実行された実行ループであることがわかりました。

ブロッキングの仕組みがわかったところで何か役に立つわけではないですが、自分としてはiOSの仕組みの一部を知ることができてよかったと思っています。

## 参考資料
公式
- [スレッドプログラミングガイド](https://developer.apple.com/jp/documentation/Multithreading.pdf)
- [CFRunLoop](https://developer.apple.com/documentation/corefoundation/cfrunloop-rht)

ブログ
- [Cocoaのマルチスレッドシステム](http://shin.hateblo.jp/entry/2014/05/04/222104)
- [NSRunLoop](http://i.devris.info/understanding-nsrunloop)