Elixirで試しに何か書いてみる(その2)

[**Elixirで試しに何か書いてみる(その1)**](http://qiita.com/HirofumiTamori/items/602d9b326cb4816c2fe6)の続きです。
**プロセス**による並行動作で高速化を狙います。

**[2015/6/5追記:GitHubのURL仕様変更とriot.jsのURL変更に対応して修正しました]**

今のところ 

[**Elixirで試しに何か書いてみる(その3) - Elixirのアプリケーション**]
(http://qiita.com/HirofumiTamori/items/06ab8e85c25f118f8e72)
[**Elixirで試しに何か書いてみる(その4) - Taskを使って簡単にする**](http://qiita.com/HirofumiTamori/items/ed131104494f4bd2f695)
[**Elixirで試しに何か書いてみる(その5) - 失敗したらやり直す**](http://qiita.com/HirofumiTamori/items/dde5d95765c68f2fef3e)

まで徐々に改良をしています。

------

## 並行動作による高速化

[その1](http://qiita.com/HirofumiTamori/items/602d9b326cb4816c2fe6)のプログラムの動作が遅いのは
　- ひとつのURLのデータ取得が終わって結果が出力されるのを待って
　- 次のURLのデータを取りに行っているから
です。

ここを同時にいくつものURLにアクセスするようにすればかなりの改善が期待できます。**Go**のプログラミング例では**goroutine**を使いましたが、**Elixir**では**プロセス**を使います。
## プロセス

**Elixir**(とそのベースになっている**Erlang**)の特徴の一つに仮想マシン上で実現されている軽量な[**プロセス**](http://elixir-lang.org/getting-started/processes.html)があります。

### プロセスの立ち上げ
**spawn**(または**spawn_link**)関数でプロセスにしたい関数を呼び出すだけです。

```ex
pid = spawn fn -> モジュール名.呼び出したい関数 end
  または
pid = spawn(モジュール名 :呼び出したい関数 [引数])
```
**pid**はプロセスIDで、Elixir(Erlang)ではこのIDでプロセスを管理しています。次に述べるプロセス間通信もpidを宛先として使います。
なお、Elixir/Erlangでは```self()```で今まさにself()が実行されているプロセスのIDを取得できます。

### プロセス間通信


プロセスにはそれぞれ**メールボックス**があり、それを介して通信を行います。**send**で送信、**receive**で受信です。宛先の指定には**pid**を使います。
![Elixir150524-1.png](https://qiita-image-store.s3.amazonaws.com/0/46294/6eaf0f15-1bdc-32d8-010e-ee77b9dca824.png)

**Elixir**にはよくできた**REPL**[^1]である**iex**[^2]がついてくるので試してみるのが早いでしょう。
[^1]: REPL(read–eval–print loop)入力を読み取って実行し結果を出力するような環境。**Ruby**の**irb**, **Python**の**ipython**, **Swift**の**PlayGround**もREPL。
[^2]: コマンドラインから`iex`と打てば起動する。`iex -S mix`とするとmixを使って依存関係のあるライブラリもiex内部で使えるようになる。補完機能もあるのでなかなか強力だと思う。セッションを越えてのヒストリ機能は残念ながらない。

Erlang/OTP 17 [erts-6.4.1] [source] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (1.0.4) - press Ctrl+C to exit (type h() ENTER for help)
iex(1)> send self(),{:hello, "world"}
{:hello, "world"}
iex(2)> receive do
...(2)>   {:hello, msg}-> IO.puts "Message is #{msg}"
...(2)>   {:hoge, msg}-> IO.puts "not selected this."
...(2)> end
Message is world
:ok
iex(3)>


**iex(1)>**で**send**で自分自身のpid(self())に対してメッセージ```{:hello, "world"}```(シンボル :hello と 文字列"world"のタプル。あちこち見ているとシンボルでステータスを返し、残りの要素で結果を返すパターンが多い)を送信しています。この状態でメッセージは自分自身(プロセスself)のメールボックスに入ります。
**iex(2)>**のところはメッセージの受信を行う側の処理を書いています。
**receive**ブロックではパターンマッチングを行って一致したメッセージがあれば -> 以下の処理を実行します。この場合**iex(1)>**でメールボックスに```{:hello, "world"}```が入っているのでシンボル```:hello```が一致するため変数msgに"world"が入り、結果として

```Message is world```

が表示されます[^3]。
[^3]: **Elixir**は**Ruby**と同様に```#{}```で文字列内に式を埋め込むことができます。

## 設計方針

先のプログラムで時間のかかっていたWebページの取得を1URLごとに1つのプロセスに割り当てて並行処理化します。
1. 必要なURL個数分のプロセスを立ち上げURL他の引数をメッセージで受信待ち受けさせます。プロセスはリスト"**fetchers**"に格納したpidで管理します。
2. 立ち上げたプロセスに対してメッセージでURLその他の引数を送ります。プロセスは必要な引数を受け取るとページを取得して必要な内容をHTMLパーザで取り出します。
3. 結果をメッセージとして送信した後はプロセスは終了します。
4. 受信側は全ての結果が揃うまで待ち続け、指定順序にソートした後に結果を表示し、プログラムを終了します。

![Elixir150524-2.png](https://qiita-image-store.s3.amazonaws.com/0/46294/e2b53330-091f-e892-4594-c16e1654a035.png)

エラー処理は端折っているのでプロセスの立ち上げには**spawn**ではなく「例外が発生したら起動したプロセスにも通知が届く」**spawn_link**を使っています。

## プログラムのソース

以下にソースを示します。

```ex
defmodule VercheckEx do
  require HTTPoison
  require Floki
  require Timex

  def fetch_content() do
    receive do # メッセージ受信。{送り手のpid, url, type, インデックス}が来たら以下を実行。
      {caller, url, type, i} ->
        #IO.puts "URL = #{url}"
        ret = HTTPoison.get!( url )

        %HTTPoison.Response{status_code: 200, body: body} = ret
        # HTML bodyを取得する
        # HTMLパーザー Flokiで処理
        # 名前、リリース日時を取得
        {_,_,n} = Floki.find(body, ".container strong a") |> List.first
        {_, d} = Floki.find(body, "time") |> Floki.attribute("datetime")
                                          |> List.first
                                          |> Timex.DateFormat.parse("{ISOz}")
        if(type == :type1) do # バージョン番号を取得
          #IO.puts "type1"
          {_,_,x} = Floki.find(body, ".tag-name span") |> List.first
        else
          {_,_,x} = Floki.find(body, ".css-truncate-target span") |> List.first
        end
        d =Timex.Date.local(d, Timex.Date.timezone("JST"))
        send caller, {:ok, {hd(n),hd(x),d,i}}
        # 送り手にメッセージを返したらこのプロセスは終了する
      end
  end

  def put_a_formatted_line(val) do # 1行出力
    {title, ver, date, _} = val
    l = title
    if String.length(title) < 8 do
      l = l <> "\t"
    end
    l = l <> "\t" <> ver
    if String.length(ver) < 8 do
      l = l <> "\t"
    end
    l = l <> "\t" <> Timex.DateFormat.format!(date, "%Y.%m.%d", :strftime)

    now = Timex.Date.now("JST")
    diff =  Timex.Date.diff( date, now, :days)
    if diff < 14 do
      l = l <> "\t<<<<< updated at " <> Integer.to_string(diff) <> " day(s) ago."
    end
    IO.puts(l)
  end

  def receiver(result_list, n) do # URL取得のプロセスからのメッセージを待ち受けする。nはURLの個数。
    if( length(result_list) < n ) do # 指定したURLのデータが全て揃うまで…
      receive do
        {:ok, res} -> # まだ手抜きでエラーケースを入れてない…
          receiver( result_list++[res], n ) # 変数result_listに結果を追加して再帰呼び出し
      end
    else # 結果が全て集まったらリストをソートして終了
      Enum.sort(result_list, fn(a,b) -> # sort by index number
        {_,_,_,i1} = a
        {_,_,_,i2} = b
        i1 < i2 end)|>Enum.each( fn(x) -> put_a_formatted_line x end)
    end
  end
end

urls = [ #{ URL, type, index}
  {"https://github.com/jquery/jquery/releases", :type1},
  {"https://github.com/angular/angular/releases", :type1},
  {"https://github.com/facebook/react/releases", :type2},
  {"https://github.com/PuerkitoBio/goquery/releases", :type1},
  {"https://github.com/revel/revel/releases", :type2},
  {"https://github.com/lhorie/mithril.js/releases", :type1},
  {"https://github.com/riot/riot/releases", :type1},
  {"https://github.com/atom/atom/releases", :type2},
  {"https://github.com/Microsoft/TypeScript/releases", :type2},
  {"https://github.com/docker/docker/releases", :type1},
  {"https://github.com/JuliaLang/julia/releases", :type2},
  {"https://github.com/Araq/Nim/releases", :type1},
  {"https://github.com/elixir-lang/elixir/releases", :type2},
  {"https://github.com/philss/floki/releases", :type1},
  {"https://github.com/takscape/elixir-array/releases", :type2},
]

# URLの数だけプロセスを立ち上げて待たせておく
fetchers = for _ <- 0..length(urls), do: spawn_link fn -> VercheckEx.fetch_content() end

Enum.each( Enum.with_index(urls),  fn(x) ->
 # プロセスにメッセージ{pid, URL, type, インデックス} を送る。メッセージを受け取ったプロセスは処理を開始する。
  {{u,t},i} = x # パターンマッチしている
  send Enum.at(fetchers,i), {self, u, t, i}
end)

result_list = []
VercheckEx.receiver(result_list, length(urls))
```

20行ほどの追加で並行動作ができるようになりました。**Beagle Bone Black**での比較ですと
*その1*(並行動作なし)

```text
real    0m22.564s
user    0m13.391s
sys     0m1.088s
```

*その2*(並行動作あり)

```text
real    0m13.993s
user    0m11.314s
sys     0m1.219s
```
とざっくり倍速になっています。**Mac mini**(Late 2014)でも試してみました。
*その1*(並行動作なし)

```text
real    0m5.877s
user    0m1.035s
sys     0m0.248s
```

*その2*(並行動作あり)

```text
real    0m2.090s
user    0m1.357s
sys     0m0.295s
```

うーん、こちらも3倍までは行かないですね…。
アプリが起動するときに**Erlang VMの起動とバイトコードのロードに時間がかかっている可能性があります…**
なので対象URLを倍にして試してみます。

*その1*(並行動作なし) 30個のURL

```text
real    0m9.694s
user    0m1.361s
sys     0m0.271s
```
*その2*(並行動作あり) 30個のURL

```text
real    0m2.708s
user    0m2.132s
sys     0m0.360s
```

少し差が出てきました。読み込むURLが増える＝同時に立ち上がるプロセスが増えるほどより高速になると予想されます[^4]。
[^4]: とはいえ、CPUコア数にも同時に張れるコネクション数にも上限はあるし、何よりアムダールの法則の示す通りでいくらでも高速化するわけではないでしょうが。

つまり、Javaベースのサーバーなどでよくあるように「**一度起動したらなるべく起動しっぱなしでいろんな処理をやらせる**」用途向きということですね。


### 感想

以上、あんまり関数型っぽくなかったですがElixirで何か書くという目的は達成したと思います。以下感想。
1. 見た目は**Ruby**っぽいけどやはり**Erlang**のシンタックスシュガー。なので「**すごいErlangゆかいに学ぼう！**」あたりはかじっておいたほうがよい。
2. プロセスの使い方にはちょっと戸惑った。いくつかの決まったパターンがありそう。今回のもその一つ。慣れれば使いやすそう。
3. **mix**なかなか強力。まだ機能の1/3も使っていないのでもう少し調べたい。 
4. 次は[**Phoenix**](http://www.phoenixframework.org/)と組み合わせて何か書いてみようかなと思えるぐらいには十分魅力的でした＞Elixir。