プログラミング/Ｃ＃/Reactive Extentions の履歴(No.3)

概要†

Reactive Extentions ライブラリを用いると、push 式の非同期通知を LINQ を用いて受け取れるそうです。

テストプログラム†

便利そうなので、よく分からないながらテストプログラムを作って試してみました。

LANG:csharp(linenumber)
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;

using System.Reactive.Linq;

namespace WpfApplication1
{
    /// <summary>
    /// コンストラクターに与えられた Action を Dispose 時に実行するクラス
    /// </summary>
    class ActionOnDispose : IDisposable
    {
        Action action;
        public ActionOnDispose(Action action)
        {
            this.action = action;
        }

        public void Dispose()
        {
            action();
        }
    }

    /// <summary>
    /// オブザーバーリストを管理する Observable クラス
    /// 
    /// OnAfterSubscribe / OnBeforeSubscribe が利用可能
    /// </summary>
    /// <typeparam name="T">通知されるデータ型</typeparam>
    public class Observable<T> : IObservable<T>
    {
        /// <summary>
        /// オブザーバーのリスト
        /// </summary>
        private Dictionary<IObserver<T>, bool> observers
             = new Dictionary<IObserver<T>, bool>();

        /// <summary>
        /// オブザーバーを登録する
        /// </summary>
        /// <param name="observer">登録するオブザーバー</param>
        /// <returns>登録解除のための IDisposable</returns>
        public IDisposable Subscribe(IObserver<T> observer)
        {
            // observer を登録して
            lock (observers) {
                observers.Add(observer, true);
            }
            OnAfterSubscription(observer);

            // Dispose 時に Unsubscribe する IDisposable を返す
            return new ActionOnDispose(() => Unsubscribe(observer));
        }

        /// <summary>
        /// オブザーバーの登録を解除する
        /// </summary>
        /// <param name="observer">登録を解除するオブザーバー</param>
        private void Unsubscribe(IObserver<T> observer)
        {
            OnBeforeUnsubscription(observer);
            lock (observers) {
                observers.Remove(observer);
            }
        }

        /// <summary>
        /// オブザーバーが登録された直後に実行する処理
        /// </summary>
        /// <param name="observer">登録されたオブザーバー</param>
        protected virtual void OnAfterSubscription(IObserver<T> observer) { }

        /// <summary>
        /// オブザーバーが登録解除される直前に実行する処理
        /// </summary>
        /// <param name="observer">登録解除されるオブザーバー</param>
        protected virtual void OnBeforeUnsubscription(IObserver<T> observer) { }

        /// <summary>
        /// observer に新しい値を送る
        /// </summary>
        /// <param name="t">observer に送る値</param>
        public void OnNext(T t)
        {
            foreach (var observer in observers.Keys.ToArray())
                if (observers.ContainsKey(observer))
                    observer.OnNext(t);
        }

       /// <summary>
       /// 処理の終了を通知する
       /// </summary>
       public void OnCompleted()
       {
           foreach (var observer in observers.Keys.ToArray())
               if (observers.ContainsKey(observer))
                   observer.OnCompleted();
       }
   }

    /// <summary>
    /// MainWindow.xaml の相互作用ロジック
    /// </summary>
    public partial class MainWindow : Window
    {
        /// <summary>
        /// 文字列値を通知する observable
        /// </summary>
        Observable<string> observable = new Observable<string>();

        public MainWindow()
        {
            InitializeComponent();

            // 同期コンテキスト
            var syncContext = System.Threading.SynchronizationContext.Current;

            // UI スレッドからの通知を別スレッドで待ち受ける
            Task.Run(() => {

                // ４回受け取るまで処理する
                observable.Take(4).Do(s => {
                    textBlock1.Text += s + "\n";
                }).Wait();

                // 以下は４回受け取った後に実行する内容

                // UI を操作するため同期コンテキストで実行する
                // http://ufcpp.wordpress.com/2012/04/26/%E9%9D%9E%E5%90%8C%E6%9C%9F%E5%87%A6%E7%90%86%E3%81%A8%E3%83%87%E3%82%A3%E3%82%B9%E3%83%91%E3%83%83%E3%83%81%E3%83%A3%E3%83%BC/
                syncContext.Post(s => {
                    // s には Post の第２パラメータ "Finished!" が渡される
                    textBlock1.Text += (s as string) + "\n";
                }, "Finished!");
            });
        }

        private void Button_Click(object sender, RoutedEventArgs e)
        {
            // 文字列を通知する
            observable.OnNext("ok");
        }
    }
}

ボタンを４回押すと、それぞれ textBlock1 に ok の文字が追加され、 それが終わるとさらに Finished! の文字が追加されます。

ミソとなるのは 124行目からの

LANG:csharp
    // ４回受け取るまで処理する
    observable.Take(4).Do(s => {
        textBlock1.Text += s + "\n";
    }).Wait();

の部分で、別スレッドから

LANG:csharp
    // 文字列を通知する
    observable.OnNext("ok");

のようにして IObservable<string> observable に送られる文字列を、 LINQ を使って順に処理した上で、処理完了を待ってから次の処理に移ることができます。

Task.Run で実行される別スレッドに記述しているにも関わらず、 Do に記述したラムダ式は UI 同期コンテキストで実行されており、 自由に UI にアクセスできるのが興味深いです。

便利は便利だけれど、実行速度などに心配も残ります？？？

LANG:csharp
   public MainWindow()
   {
       InitializeComponent();

       // ４回受け取るまで処理する
       var awaiter = observable.Take(4).Do(s => {
           textBlock1.Text += s + "\n";
       }).Finally(() => textBlock1.Text += "Finished!\n").GetAwaiter();

       // 必要なところで awaiter.Wait(); として終了まで待機できる
   }

LANG:csharp
   public MainWindow()
   {
       InitializeComponent();

       // ４回受け取るまで処理する
       observable.Take(4).Finally(() => textBlock1.Text += "Finished!\n").Subscribe(s => {
           textBlock1.Text += s + "\n";
       });
   }

パフォーマンス†

LANG:csharp
   var start = DateTime.Now.Ticks;

   var observable = new Observable<int>();
   for (var i = 0; i < 100000; i++) {
       // １回あたり 5μ秒
       observable.Take(1).Do(n => { }).GetAwaiter();
       observable.OnNext(0);
   }

   textBlock1.Text += "\n" + (DateTime.Now.Ticks - start).ToString();

LANG:csharp
   for (var i = 0; i < 100000; i++) {
       using (observable.Take(1).Subscribe(n => { })) { };
   }

LANG:csharp
   var start = DateTime.Now.Ticks;

   var sum = 0L;
   var observable = new Observable<int>();
   var awaiter = observable.Do(i => sum += i }).GetAwaiter();

   Task.Run(() => {
       for (var i = 0L; i < 1000000; i++) {
           observable.OnNext(1);
       }
       observable.OnCompleted();
   });

   awaiter.Wait();

   textBlock1.Text += "\nAnswer=" + sum.ToString() + "\n" + (DateTime.Now.Ticks - start).ToString();

LANG:csharp
   var start = DateTime.Now.Ticks;

   var sum = 0L;
   Task.Run(() => {
       for (var i = 0L; i < 10000000; i++) {
           // １回あたり 3ns
           sum += 1;
       }
   }).Wait();

   textBlock1.Text += "\n" + (DateTime.Now.Ticks - start).ToString();

コメント・質問†