2

SignalR(.NET Client)を使用して、イベントの通知とデータの取得を行うプログラムを作りました。
やりたいことはとりあえず出来たのですが、あまり処理を綺麗に書けませんでした。
マルチスレッドの問題やDisposable漏れなどに考慮して、もっと処理をスマートに書けるものでしょうか?
Rxなどを使うともっと綺麗に書けたりするものなのかもしれませんが、自分の力では、これをどうRxで記述すれば良いかイメージができませんでした。

なにかアイディアがあれば力をお貸しください。

■やりたいこと

  • サーバ側は一定間隔でクライアントにデータの発生を通知する。
  • データはキューに保持する。

  • クライアントはデータ発生のイベントを受信すると、サーバからデータ取得を行う。

  • データ発生のイベントだけでなく、接続直後にもデータの取得を行う。
  • データ取得は、サーバのキューが空になるまで続けて呼び出す。
  • クライアントが接続に失敗した場合、一定時間後に再接続を試みる。
  • クライアントは任意の時点で開始/停止ができる。

■クライアント/サーバ共通ソース

public class Data
{
    public int Id { get; set; }
    public string Text { get; set; }
}

■サーバ側ソース

// Hub
[HubName("test")]
public class TestHub : Hub
{
    // データ取得処理
    public Data Query()
    {
        return Service.Deque();
    }
}

// 通知およびキューの管理、事前にStartしておく
public static class Service
{
    private static readonly ConcurrentQueue<Data> queue = new ConcurrentQueue<Data>();

    private static int nextId = 1;

    private static Timer timer;

    // 一定期間毎にキューにデータを入れて、クライアントに通知
    public static void Start(IHubConnectionContext<dynamic> clients)
    {
        timer = new Timer(state =>
        {
            nextId++;
            queue.Enqueue(new Data { Id = nextId, Text = String.Format("Data-{0}", nextId) });

            clients.All.Update();
        }, null, 3000, 3000);
    }

    public static void Stop()
    {
        timer.Dispose();
    }

    public static Data Deque()
    {
        Data data;
        queue.TryFetchAndAction(out data);
        return data;
    }
}

■クライアント側ソース(もっとスマートにしたい部分)

public class Client
{
    // hub操作用の同期オブジェクト
    private readonly object sync = new object();

    // 実行状態、Closed時の再接続と、データ取得の連続処理中断に使用
    private readonly ManualResetEvent running = new ManualResetEvent(false);

    // 再接続用のタイマー
    private Timer timer;

    private HubConnection hub;

    private IHubProxy proxy;

    private void Start()
    {
        if (hub != null)
        {
            return;
        }

        // 実行状態に移行
        running.Set();

        hub = new HubConnection("http://127.0.0.1:10080");
        hub.Closed += () =>
        {
            // 旧タイマーを片付け
            var disposable = Interlocked.Exchange(ref timer, null);
            if (disposable != null)
            {
                disposable.Dispose();
            }

            if (running.WaitOne(0))
            {
                // 一定時間後に再接続の試行
                Interlocked.Exchange(ref timer, new Timer(state =>
                {
                    lock (sync)
                    {
                        if (hub != null)
                        {
                            ConnectHubAsync();
                        }
                    }
                }, null, 5000, 0));
            }
        };

        proxy = hub.CreateHubProxy("test");
        proxy.On("Update", FetchAndAction);

        ConnectHubAsync();
    }

    // 接続と、接続とのデータ取得
    private void ConnectHubAsync()
    {
        hub.Start().ContinueWith(t =>
        {
            if (hub.State == ConnectionState.Connected)
            {
                FetchAndAction();
            }
        });
    }

    // データ取得の連続処理
    private async void FetchAndAction()
    {
        while (running.WaitOne(0))
        {
            Data data;
            try
            {
                data = await proxy.Invoke<Data>("Query");
            }
            catch (Exception ex)
            {
                // Invoke中にHubConnection.Stop()するとInvalidOperationException発生
                break;
            }

            // もうデータがない
            if (data == null)
            {
                break;
            }

            // ここでDataを元に処理を行う
        }
    }

    private void Stop()
    {
        if (hub == null)
        {
            return;
        }

        // 停止状態に移行
        running.Reset();

        lock (sync)
        {
            var disposable = Interlocked.Exchange(ref timer, null);
            if (disposable != null)
            {
                disposable.Dispose();
            }

            hub.Stop();
            hub = null;
        }
    }
}

2 件の回答 2

1

クライアント側の同期機構がlock (sync)Interlocked.ExchangeManualResetEventと三種類も存在するのが気になります。ロック時間が特別にクリティカルでもない限り単純にlock (sync)で制御すれば大分シンプルになるのではないでしょうか。

また

var disposable = Interlocked.Exchange(ref timer, null);
if (disposable != null)
{
    disposable.Dispose();
}

usingステートメントで

using(Interlocked.Exchange(ref timer, null)) { }

と書けます。

5
  • 回答ありがとうございます。 再度確認してみますが、最初はsyncだけで制御してみたのですが、Client.Stop()のlock(sync)の中でhub.Stop()するとClosedイベントが別スレッドで発生し、そこでlock(sync)するとデッドロックが発生したような動きとなったため、今の形にしています。 自分としても、同期制御や非同期処理の呼び出しかたが混在しているのが気になっているのですが、そこをうまくまとめられずにいます。 Commented 2015年3月23日 11:49
  • @ホッシー 確認したところConnection.StopClosedを同スレッドで呼び出しているのでlockステートメントが原因のデッドロックは通常発生しないように思います。もし再度発生した場合はVisual Studioで中断し、「スレッド」および「呼び出し履歴」ウィンドウで各スレッドの実行箇所を参照して実際にデッドロックとなっているか確かめてみて下さい。
    – pgrho
    Commented 2015年3月23日 12:22
  • 自分でも調べてみたのですが、スタックトレースを見たところConnection.Stop()の中で_connectTask.Wait()しており、ここで待ちになっているようです。 github.com/SignalR/SignalR/blob/master/src/… Closedイベントは以下ののDisconnect()内で処理されており、デバッガでも別スレッドで実行されていることを確認しました。 github.com/SignalR/SignalR/blob/master/src/… Commented 2015年3月23日 12:50
  • @ホッシー なるほど、Start()失敗のコールバックでデッドロックしているのですね。では接続完了後にClosed等を設定するフローにしてみてはいかがでしょうか。
    – pgrho
    Commented 2015年3月23日 13:19
  • その後、参考になりそうなソースを探していました。 自分の当初のイメージは次のようなものだったのですが、この内容は自分には敷居が高いので一端保留。 github.com/AdaptiveConsulting/ReactiveTrader/tree/master/src/… そこで、次のようなイメージで、HubConnectionの再接続に関する部分を別クラスに分離することでClientクラスの状態管理を軽くする方向でもう少し考えてみたいと思います。 github.com/killnine/HubConnectionManager/blob/master/… Commented 2015年3月25日 12:12
1

何らかのリソースを抱えるクラスは、IDisposableインターフェースを実装します。
例えば、Clientクラスでは、Timer、ManualResetEventといったものがリソースと見なされます。
それらの特徴として、いずれもIDisposableを実装しています。

Clientクラスのインスタンスが破棄されるとき、それらオブジェクトのDisposeメソッドを呼び出す必要があります。

Clientクラスを今後、継承して利用する必要がなければ、sealed を指定しておいたほうが良いでしょう。
次にIDisposable パターンを実装した例を示します。

public sealed class Client : IDisposable {
    private int disposed = 0;
    private void Dispose( bool disposing ) {
        if( Interlocked.CompareExchange( ref disposed , 1 , 0 ) == 0 ) {
            // ここでリソースを解放する、つまり timer.Dispose(); などを呼び出す。
            GC.SuppressFinalize( this );
        }
    }

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

    ~Client(){
        Dispose( false );
    }
}

~Client() の部分は、デストラクタを定義しています。
こうすることで、仮にDisposeが呼ばれないケースがあったとしても、GCがそのうちDisposeを呼び出してくれます。
GC.SuppressFinalize( this ) は、GCにデストラクタを呼び出す必要がないことを通知します。
しかし、デストラクタは、デストラクタを実装しないオブジェクトに比べ、回収されにくくにります。(後回しにされるイメージ)
そのため、無闇にデストラクタを実装することは、推奨されませんが、今回のようなケースでは、有効な方法だと思われます。

もし、Clientクラスを派生させる場合は、 void Dispose( bool disposing )メソッドを
virtual指定し、オーバーライドして下さい。

また非同期パターンにおけるロック機構として、System.Threading.SemaphoreSlimを使用する方法があります。
詳しくは、非同期:awaitを含むコードをロックするには?(SemaphoreSlim編)

を参考にして下さい。

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。