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

Elixirで試しに何か書いてみる(その1)の続きです。
プロセスによる並行動作で高速化を狙います。

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

今のところ

Elixirで試しに何か書いてみる(その3) - Elixirのアプリケーション
Elixirで試しに何か書いてみる(その4) - Taskを使って簡単にする
Elixirで試しに何か書いてみる(その5) - 失敗したらやり直す

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

並行動作による高速化

その1のプログラムの動作が遅いのは
　- ひとつのURLのデータ取得が終わって結果が出力されるのを待って
　- 次のURLのデータを取りに行っているから
です。

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

プロセス

Elixir(とそのベースになっているErlang)の特徴の一つに仮想マシン上で実現されている軽量なプロセスがあります。

プロセスの立ち上げ

spawn(またはspawn_link)関数でプロセスにしたい関数を呼び出すだけです。

pid = spawn fn -> モジュール名.呼び出したい関数 end
  または
pid = spawn(モジュール名 :呼び出したい関数 [引数])

pidはプロセスIDで、Elixir(Erlang)ではこのIDでプロセスを管理しています。次に述べるプロセス間通信もpidを宛先として使います。
なお、Elixir/Erlangではself()で今まさにself()が実行されているプロセスのIDを取得できます。

プロセス間通信

プロセスにはそれぞれメールボックスがあり、それを介して通信を行います。sendで送信、receiveで受信です。宛先の指定にはpidを使います。

ElixirにはよくできたREPL¹である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

が表示されます³。

設計方針

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

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

プログラムのソース

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

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(並行動作なし)

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

その2(並行動作あり)

real    0m13.993s
user    0m11.314s
sys     0m1.219s

とざっくり倍速になっています。Mac mini(Late 2014)でも試してみました。
その1(並行動作なし)

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

その2(並行動作あり)

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

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

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

real    0m9.694s
user    0m1.361s
sys     0m0.271s

その2(並行動作あり) 30個のURL

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

少し差が出てきました。読み込むURLが増える＝同時に立ち上がるプロセスが増えるほどより高速になると予想されます⁴。
つまり、Javaベースのサーバーなどでよくあるように「一度起動したらなるべく起動しっぱなしでいろんな処理をやらせる」用途向きということですね。

感想

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

REPL(read–eval–print loop)入力を読み取って実行し結果を出力するような環境。Rubyのirb, Pythonのipython, SwiftのPlayGroundもREPL。 ↩
コマンドラインからiexと打てば起動する。iex -S mixとするとmixを使って依存関係のあるライブラリもiex内部で使えるようになる。補完機能もあるのでなかなか強力だと思う。セッションを越えてのヒストリ機能は残念ながらない。 ↩
ElixirはRubyと同様に#{}で文字列内に式を埋め込むことができます。 ↩
とはいえ、CPUコア数にも同時に張れるコネクション数にも上限はあるし、何よりアムダールの法則の示す通りでいくらでも高速化するわけではないでしょうが。 ↩