test-queueのソース読んでみた

この記事は[アクトインディ Advent Calendar 2015 10日目](http://qiita.com/advent-calendar/2015/actindi "act")になります。

さっそくですが、[いこーよ](http://iko-yo.net "いこーよ")はテスティングフレームワークに`rspec(3.1.0)`を使ってます。
exmpleは2275個もあって開発PC(MacbookPro15 2015)で並列テストを実行すると8分くらいかかってしまっています。遅いです。
[放熱を頑張ることで下がっているクロックを取り戻しテスト実行時間短縮を考えていました](http://tech.actindi.net/3656724226 "放熱を頑張ることで下がっているクロックを取り戻そうとしていました")がCPUの発熱が激しぎて何やっても効果が見込めない感じがしたのでペルチェ素子の購入を最後に放熱は諦めました。本当にありがとうございました。

![damedeshita.png](https://qiita-image-store.s3.amazonaws.com/0/42642/d5c96d44-296f-50af-33ff-28f72f579430.png)

test-queueのREADMEを読むと分散モードというのがあるようで、これは良いソリューションになるのでは？と思いとりあえずソース読むことにしました。

# test-queueについて
テストを並列実行してくれるライブラリです。
masterからforkしたプロセスをworkerとして使って暇になったworkerにテストを振っていきます。

遅いテストから順番に消化するのでworkerの終了時間にばらつきが出にくくなってます。
実行するテストの全情報をすべてのworkerが別々のメモリ空間で保持しているのと、workerプロセス + masterプロセス分が必要なのでメモリをよく食う気がします。

# test-queueがやっているだいたいのこと
単一ホストの複数workerで動かしたとします。

* テストを遅い順番に並び替える
* 環境変数を読み込む
* masterプロセスを開始する
  * UnixSocketをopen
* forkでworkerプロセスを生成する
  * 環境変数`TEST_QUEUE_WORKERS` か CPUコアの数だけworkerを生成する
* masterがworkerからの接続を待つ(A)
* workerがmasterへUnixSocket経由で`POP`という文字を送信する
* `POP`を受信したmasterは1つのテストをUnixSocket経由でworkerへ送信する
* テストを受け取ったworkerはテストを実行する
* masterは終了したworkerがいないかチェックをする
  * 終了したworkerがあったらmasterがworkerの後始末をする
* 実行待ちテストがなくなるまで(A)に戻る
* masterを停止する
  * UnixSocketのclose
* 終了

# test-queueのやっていることを詳しく
## テストを遅い順番に並び替える
### [lib/test_queue/runner/rspec.rb:18](https://github.com/tmm1/test-queue/blob/96fe902c4df69f06974be2ead6ed6b68e30d8d23/lib/test_queue/runner/rspec.rb#L18 "lib/test_queue/runner/rspec.rb:18")
```ruby
module TestQueue
  class Runner
    class RSpec < Runner
      def initialize
        @rspec = ::RSpec::Core::QueueRunner.new
        super(@rspec.example_groups.sort_by{ |s| -(stats[s.to_s] || 0) })
      end
[...]
```
\#statsの戻り値は、前回のテスト実行終了時に保存しておいたテスト毎に掛かった時間のハッシュを返すのでここで遅い順番に並び替えています。
`.test_queue_stats`というファイルに前回のテスト実行終了時の情報を保存してます。

## workerを生成する
### [lib/test_queue/runner.rb:219](https://github.com/tmm1/test-queue/blob/96fe902c4df69f06974be2ead6ed6b68e30d8d23/lib/test_queue/runner.rb#L219 "lib/test_queue/runner.rb:219")

```ruby
def spawn_workers
  @concurrency.times do |i|
    num = i+1
    pid = fork do
      @server.close if @server
      iterator = Iterator.new(relay?? @relay : @socket, @suites, method(:around_filter))
      after_fork_internal(num, iterator)
      ret = run_worker(iterator) || 0
      cleanup_worker
      Kernel.exit! ret
    end
[...]
```
`@concurrency`の数だけループしてworkerを生成しています。
ここでは、EnumerableをincludeしたIteratorクラスをインスタンス化しており、その`Iterator#each`で`master`とSocket通信するコードが書かれています。
トリッキーな感じがしましたが、テスティングフレームワークに左右されない共通処理をまとめるためだと思われます。
ちなみに`Iterator#each`は`#run_worker`の中で呼ばれています。

## masterがworkerにテストを振ってworkerがテストを実行する
### [lib/test_queue/runner.rb:318](https://github.com/tmm1/test-queue/blob/96fe902c4df69f06974be2ead6ed6b68e30d8d23/lib/test_queue/runner.rb#L318 "lib/test_queue/runner.rb:318")
```ruby
# master側
until @queue.empty? && remote_workers == 0
  if IO.select([@server], nil, nil, 0.1).nil? # 
    reap_worker(false) if @workers.any? # check for worker deaths
  else
    sock = @server.accept
    cmd = sock.gets.strip
    case cmd
    when /^POP/
      if obj = @queue.shift
        data = Marshal.dump(obj.to_s)
        sock.write(data) # <= workerへテストを1つ送信してる
      end
```
workerの準備が整う(#run_worker実行してる)とmasterへ`POP`と送信します。するとmasterはworkerへ1つのテストを返信します。
### [lib/test_queue/iterator.rb:18](https://github.com/tmm1/test-queue/blob/96fe902c4df69f06974be2ead6ed6b68e30d8d23/lib/test_queue/iterator.rb#L18 "lib/test_queue/iterator.rb:18")
```ruby
# worker側
def each
  fail "already used this iterator. previous caller: #@done" if @done

  while true
    client = connect_to_master('POP')
    break if client.nil?
    r, w, e = IO.select([client], nil, [client], nil)
    break if !e.empty?
    if data = client.read(65536)
      client.close
      item = Marshal.load(data)
      break if item.nil? || item.empty?
      suite = @suites[item]

      $0 = "#{@procline} - #{suite.respond_to?(:description) ? suite.description : suite}"
      start = Time.now
      if @filter
        @filter.call(suite){ yield suite } # <= ここでテストを実行してる
      else
        yield suite
          end
[...]
```
workerはテストを受け取って`yeild`でテストを実行します。  
workerは一度走りだすと`nil`を受け取るまでmasterに`POP`を送信し続けます。
masterがworkerに何もを送信しない時は実行待ちテストが尽きた時です。workerが`nil`を受け取ると消化したテストを一時ファイルに書き込んでworkerプロセスは役目を終えます。  

masterは、workerにテストを振りつつもも終了したworkerがいないかチェックもしています。終了したworkerがいた場合はそのworkerが終了時に残した実行結果が書かれた一時ファイルの収集と削除を行います。(#reap_worker)

workerが全部なくなると、workerから収集したテスト結果をmasterが標準出力に出力して並列テストは終了します。

# 分散モードについて
slaveホストとして起動したテストはmasterプロセスでSocketをオープンせず、外にあるホストのmasterプロセスに接続しにいくようになります。masterホストから見ると別のホストでテストを実行できるようになります。

環境変数`TEST_QUEUE_SOCKET`に設定されている値によってUnixSocketかTCPSocketのどちらかを作成します。
分散モードで実行したい(TCPSocketをオープンしたい)場合は、masterホスト側で`TEST_QUEUE_SOCKET`に自身のIP:Portの設定が必要です。
slaveホスト側では、環境変数`TEST_QUEUE_RELAY`にmasterホストのIP:Portを設定します。
また、masterとslaveに共通の環境変数`TEST_QUEUE_RELAY_TOKEN`を設定する必要があります。

## 動かしてみます
```shell
TEST_QUEUE_SOCKET=192.168.1.100:12345 bundle exec minitest-queue ./test/sample_minitest5.rb &
sleep 0.1
TEST_QUEUE_RELAY=192.168.1.100:12345 bundle exec minitest-queue ./test/sample_minitest5.rb
wait
```
slaveホストは、masterのmasterプロセスに接続できるまでリトライするのでどちらから実行しても問題はないですが、masterホスト側を先に実行します。
masterホストが順番待ちテストを1つずつworkerに振っているフェーズ(#distribute_queue)に入ると、slaveホストがTCP経由でテストを受けとってテストを実行してくれます。
以降の流れは単体モードとだいたいと同じです。

ということで分散モードというのは、slaveホストの数に制限はなく、slaveホストをたくさん増やすとその分早くなりそうです。
単体モードだとI/Oがボトルネックになりがちなのでこれは便利そうです。

以上。