Rx入門 (25) - デッドロックに注意

IObservable<T>は時間軸を基本としたデータを扱うことができます。これまでも述べてきたように、その枠内で非同期処理を扱うことができます。そして、非同期処理で忘れてはならないのがデッドロックです。Rxにおいてもデッドロックが発生するケースがもちろんあるので、今回はその一部を取り上げてみたいと思います。

デッドロックとは

これを読んでいる方で知らない方はほとんどいないと思いますが、簡単に触れておきます。デッドロックとは、処理Aと処理Bがお互いの終了処理を待ち合ってしまうことで、結果としてどちらも処理を先に進めることができなくなってしまう現象のことを言います。コンパイルエラーとして検出されるものではなく、かつ処理Aや処理Bの記述に大きな問題なくても発生する場合があるため、不具合の原因を見つけにくい厄介な問題です。

サンプルコード

パッと思い付いた範囲でサンプルを示します。単純なものばかりですが、結構やってしまう例かと思います。凡ミスをしないためのイメージ作りになれば幸いです。

値がひとつも来ない

ひとつめはFirstメソッドを利用したケースについてです。Firstメソッドは最初の値をIObservable<T>ではなく、単一の値として取得するときに利用するメソッドです。Firstメソッドは呼び出し元スレッドをブロックするという特徴を持っているため、以下のような場合にデッドロックが発生します。

var subject = new Subject<int>();
var value   = subject.First(); //--- 最初の値が来るまで呼び出し元スレッドをブロック
subject.OnNext(0);             //--- 値を発行したいが呼び出されない
Console.WriteLine(value);

もうひとつの例として、ToEnumerableメソッドを利用したものを紹介します。ToEnumerableメソッドは内部でSemaphoreクラスを利用しており、値が流れて来るまで呼び出し元スレッドをブロックして待機します。そのため同一スレッドで値を発行することができず、互いを待ち合う形となってしまいます。

var subject    = new Subject<int>();
var collection = subject.ToEnumerable();
foreach (var item in collection) //--- 値が来るのを待機してしまう
    Console.WriteLine(item);     //--- 呼び出されない

完了通知が来ない

最後に、Lastメソッドを利用したものを紹介します。Lastメソッドは最後の値を単一の値として取得するメソッドです。何をもって最後の値かを判定するかと言うと、OnCompletedが発行されたかどうかです。以下のサンプルでは、Observable.Timerメソッドから発行される最後の値を取得しようとしていますが、Observable.TimerメソッドはOnCompletedを発行しない (= 終わりがない) ので、Lastメソッドは呼び出し元スレッドをブロックしたまま先に進みません。

var value = Observable
          .Timer(TimeSpan.FromSeconds(0), TimeSpan.FromSeconds(1))
          .Do(_ => Console.Clear())
          .Do(_ => Console.WriteLine(DateTime.Now.ToString()))
          .Last(); //--- タイマー系はOnCompletedが発行されない
Console.WriteLine("Last value is {0}", value); //--- 呼び出されない