テスト駆動開発実践 その１「スロットリング制御並列エクステンション」

# テスト駆動開発は難しい

本番の開発でいろいろな問題に出会っていると、やっぱりテスト駆動開発は気持ちいいし、安心感がある。しかし、結構な人はテストを書いていない。多分その理由はやったことがない、知らないとかもあるかもしれないが、テストを書くのが難しいのではないだろうか？サンプル程度のものは独立したものは書きやすいのだが、本番の環境で使うフレームワークやミドルウェアなどが絡んだテストはどう書いていいかわからないときもある。どうやって書いたらいいだろう？

個人的にはテスト駆動開発が好きなので、DevOps 野郎として、どんなコードでもテスト駆動で書けるようになってみたいと思うようになった。これからたまに、これ、どうやってテスト書くんやろう？と思ったお題に対して、どのようにテストを書いていったかの試行錯誤して記録してみたい。

# お題「スロットリング制御並列エクステンション」

同僚から下記のようなカッコいいコードが送られてきた。動作するのはお墨付き。これは何をしているかというと、拡張メソッドで、IReadOnlyList に拡張メソッドをつけて、スロットリング付きの ParalellFor 的な実装をしてくれた。

```csharp
    public static class IReadOnlyListExtensions
    {
        public static async Task ParallelForEachAsync<T>(this IReadOnlyList<T> items, int maxConcurrency, Func<T, Task> action)
        {
            using (var semaphore = new SemaphoreSlim(maxConcurrency))
            {
                var tasks = new Task[items.Count];
                for (int i = 0; i < items.Count; i++)
                {
                    tasks[i] = InvokeThrottledAction(items[i], action, semaphore);
                }

                await Task.WhenAll(tasks);
            }
        }

        static async Task InvokeThrottledAction<T>(T item, Func<T, Task> action, SemaphoreSlim semaphore)
        {
            await semaphore.WaitAsync();
            try
            {
                await action(item);
            }
            finally
            {
                semaphore.Release();
            }
        }

    }
```

## 使い方

```csharp
            List<int> list = Enumerable.Range(0, 1000).ToList<int>();
            IReadOnlyList<int> col = new ReadOnlyCollection<int>(list);
            await col.ParallelForEachAsync(10, async (x) =>
            {
                Console.WriteLine($"{x}: Accepted.");
                await Task.Delay(TimeSpan.FromSeconds(5));
                Console.WriteLine($"{x}: Done.");
                
            });

```

## 実行結果

きれいに１０個づつ並列で実行してくれている。

```
1: Accepted.
2: Accepted.
3: Accepted.
4: Accepted.
5: Accepted.
6: Accepted.
7: Accepted.
8: Accepted.
9: Accepted.
5: Done.
9: Done.
7: Done.
8: Done.
3: Done.
6: Done.
4: Done.
2: Done.
1: Done.
0: Done.
19: Accepted.
15: Accepted.
18: Accepted.
17: Accepted.
12: Accepted.
16: Accepted.
10: Accepted.
14: Accepted.
13: Accepted.
11: Accepted.
　　:
```

＃ テストのアイデア

さて、次回からこのコードを書くときにテスト駆動で書けるようにどうやってテストを書くか考えてみる。並列性を正確にアサートするのは難しそうなので、こんな戦略を考えてみた

* 並列実行するタスクにTask.Delayをつかってある程度ゆっくり実行させる
* 同時に実行されている数を数えてアサートする

完ぺきとは言えないが、async/await などは、スレッドを常に作るとかではないので、完璧に平行で実行されているかはわからないが、少なくとも、このライブラリのポイントである、Max XX 多重で並列実行されていることがアサートできればいいと考えた。

```csharp
        [TestMethod]
        public async Task TestParallelThrottling()
        {
            List<int> list = Enumerable.Range(0, 100).ToList<int>();
            IReadOnlyList<int> col = new ReadOnlyCollection<int>(list);

            var count = 0;
            var concurrentCount = new ConcurrentStack<int>();

            col.ParallelForEachAsync<int>(10, async x => {
                concurrentCount.Push((Interlocked.Increment(ref count)));
                await Task.Delay(1000);
                concurrentCount.Push((Interlocked.Decrement(ref count)));
            });

            Assert.AreEqual(concurrentCount.Max(), 10);

        }
```

このコードのポイントとしては、`ConcurrentStack` を使って、count を `Interlocked` をつかってロックして、開始時にインクリメント、終了時にデクリメントしている。この状態で、現在の count 値を ConcurrentStack にプッシュしていく。最後に、Linq の `Max()` メソッドをつかって、最大値をとると、実行を１秒ぐらいまっているので、最大値が 10 になるはずだ。実際のループは100にしているので、10ずつ実行されているのがアサートされる。

## 他の代替え案

基本的には、平行数を数えるというアプローチなのだが、ReactiveExtension を使って、単位時間当たりの Window で実行された数を数えて 並列数を計算するというアプローチも考えた。多分`OnNext(1)` を呼ぶことは３秒に対して無視できるほど小さいと考えて、それが平行して実行されたあと、３秒まって、無視できるぐらい小さい時間実行されて、３秒まつという感じなので、３秒のウインドウの間に、少しづつずれて実行されるはず。だから、そのウインドウ内の同時実行数は１０になるはずという感じ。

```csharp
        [TestMethod]
        public async Task TestParallelThrottlingRx()
        {
            List<int> list = Enumerable.Range(0, 100).ToList<int>();
            IReadOnlyList<int> col = new ReadOnlyCollection<int>(list);
            var subject = new Subject<int>();
            List<Exception> exs = new List<Exception>();
            subject.Window(TimeSpan.FromSeconds(3))
                .SelectMany(o => o.Sum())
                .Subscribe(x =>
                {
                    try
                    {
                        Assert.AreEqual(10, x);
                    } catch(Exception e)
                    {
                        exs.Add(e);
                    }
                });
           
            await col.ParallelForEachAsync<int>(10, async x =>
            {
                await Task.Delay(TimeSpan.FromSeconds(1));
                subject.OnNext(1);
                await Task.Delay(TimeSpan
                    .FromSeconds(2));
            });

            Assert.AreEqual(0, exs.Count);

        }
```

プログラミングテクニック的には、Stream Analytics のように時間に関するデータを扱うのが得意なリアクティブエクステンションを使った。インポートしたライブラリは下記の通り。Winodow は、Providers のパッケージに入っています。

* System.Reactive
* System.Reactive.Linq
* System.Reactive.Providers

`Window` メソッドで、タイムウインドウを設定して、`SelectMany` の部分でサマリつまり３秒間に発生したイベントの合計を出して、それをアサートしています。なぜかはわかっていませんが、Assert の例外がなぜか飛ばないので、それを内部でキャッチして、最終的に Exception の数でアサートしています。ちなみに、Delay のタイミングを 1 と　2 に分けているのはそうしないとうまくウインドウ枠に入らなかったからです。ちなみに実行時間も最初の案が25ms に対して、こちらは30 sec なので時間的にもメリットがありませんでした。しかし、ReactiveExtensionsで初めて自分だけでコードを書いたので、満足感はありましたｗ

# まとめ

今回初めてやってみましたが、ほかにいいソリューションがある人は是非コメントいただければと思います。また難しそうなネタがあったらシェアしてみたいと思います。

# リソース

* [第1回　Reactive Extensionsの概要と利用方法](http://www.atmarkit.co.jp/fdotnet/introrx/introrx_01/introrx_01_02.html)
* [Reactive Extensions再入門 その２６「値をまとめるWindowメソッド」](https://blog.okazuki.jp/entry/20120121/1327119785)
* [Reactive Extensions再入門 その４８「連続して発生する値の監視」](https://blog.okazuki.jp/entry/20120324/1330956809)
* [Reactive Extensions再入門 その７「LINQスタイルの拡張メソッド」](https://blog.okazuki.jp/entry/20111110/1320849106)
* [Reactive Extensions再入門 その３９「Subject系クラス」](https://blog.okazuki.jp/entry/20120214/1329229786)
* [サンプルアプリケーション](https://github.com/TsuyoshiUshio/TDDPlayground)