ElixirでのMessage Passing入門

Elixir Advent Calendar 2013 22日目。

ElixirといえばErlang/OTP、Erlang/OTPといえばActor Model、Actor ModelといえばMessage Passing。Actor Modelについてちょっと興味を持った人は「どうやって実装するのだろう」という疑問を持つと思います。ここではElixirにおけるMessage Passingのお作法を整理したいと思います。

前提

実際はErlang/OTPの経験があって初めてピンと来ることもElixirには多いのですが、本稿ではErlang/OTPについて知識が無い方でもElixirのMessage Passingについてある程度の理解ができるように記述します。Erlang/OTP由来や、そのものである処理も存在しますが、本稿ではそれを明示しません。
なお、本稿で使用するElixirのバージョンは0.11.2です。昔はErlang/OTPのRPCを直接呼び出していましたが、今はElixir言語の中である程度完結できるようになりました。今後もやり方は変化するかもしれません。あしからず。

用語集

ホスト

ノイマン型コンピュータの最小単位を指し、本稿では1台のMacBook Airを使用します。

ノード

ノードはErlang VM、すなわちBeamのことです。

プロセス

Erlang VMの中で生成されるプロセスです。OSネイティブのプロセスではありませんが、Erlang/OTPがSMPに対応しているので、マルチコアで動作可能です。

iexでやってみる

接続編

まずは対話型のシェルであるiexでやってみましょう。まずは同一のホストでiexを2つ立ち上げます。

% iex --sname foo --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(foo@mba11-keith)1>

% iex --sname bar --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(bar@mba11-keith)1>

--sname は、iexにshort nameを割り当てるオプションです。これによって分散環境でのノード間の識別を行います。ちなみに、--name オプションを使用すると、ドメイン名を含むことができます。
--cookie は、ノード同士が接続する際の、言わば「合言葉」です。平文であることに注意して下さい。

まずは、自分のノード名を確認します。

iex(foo@mba11-keith)1> Node.self
:"foo@mba11-keith"

iex(bar@mba11-keith)1> Node.self
:"bar@mba11-keith"

現在接続されているノードのリストを確認します。

iex(foo@mba11-keith)2> Node.list
[]

iex(bar@mba11-keith)2> Node.list
[]

どちらも、まだ他のノードと接続していません。fooからbarへ接続してみましょう。

iex(foo@mba11-keith)3> Node.connect :"bar@mba11-keith"
true

fooからbarへ接続してみました。barの方からノードのリストを確認してみましょう。

iex(bar@mba11-keith)3> Node.list
[:"foo@mba11-keith"]

もちろんfooからもbarが見えています。

iex(foo@mba11-keith)4> Node.list
[:"bar@mba11-keith"]

ついでに、cookieが異なる場合の挙動を確認しておきましょう。

% iex --sname baz --cookie ELIXIR-EXAMPLE
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(baz@mba11-keith)1> Node.connect :"bar@mba11-keith"
false

このとき、接続先のノードでは以下の様に表示されます。もちろん接続されません。

iex(bar@mba11-keith)4>
=ERROR REPORT==== 22-Dec-2013::14:51:30 ===
** Connection attempt from disallowed node 'baz@mba11-keith' **

nil
iex(bar@mba11-keith)5> Node.list
[:"foo@mba11-keith"]

3つ目のノードとしてbazも加えておきましょう。

% iex --sname baz --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(baz@mba11-keith)1>

iex(foo@mba11-keith)5> Node.connect :"baz@mba11-keith"
true
iex(foo@mba11-keith)6> Node.list
[:"bar@mba11-keith", :"baz@mba11-keith"]

iex(bar@mba11-keith)6> Node.list
[:"foo@mba11-keith", :"baz@mba11-keith"]

iex(baz@mba11-keith)1> Node.list
[:"foo@mba11-keith", :"bar@mba11-keith"]

Message Passing編

環境が構築できましたので、早速Message Passingをしてみましょう。

まずは、実行する関数を定義します。ここでは、Node.selfを実行した結果を検査し書き出す、簡単な関数を定義します。

iex(foo@mba11-keith)7> func = fn -> IO.inspect Node.self end
#Function<20.80484245 in :erl_eval.expr/5>

これを、fooからfoo, bar, bazに対して実行し、結果を返させます。

iex(foo@mba11-keith)8> Node.spawn(:"foo@mba11-keith", func)
#PID<0.65.0>
:"foo@mba11-keith"
iex(foo@mba11-keith)9> Node.spawn(:"bar@mba11-keith", func)
:"bar@mba11-keith"
#PID<8620.66.0>
iex(foo@mba11-keith)10> Node.spawn(:"baz@mba11-keith", func)
#PID<8692.59.0>
:"baz@mba11-keith"

Node.spawnの返り値として、#PIDというものが返ってきました。先頭のフィールドはノード番号を示すもので、localのノードである場合は0、リモートのノードである場合は正の整数が返ります。

プログラムを書いてみる

メッセージへの応答を実装する場合は、receive文を使います。その中でメッセージの内容と引数ごとの処理を定義します。

hello0.ex

defmodule Hello do
  def hello do
    receive do
      {:english, target} ->
        IO.puts "Hello, #{target}."
      {:japanese, target} ->
        IO.puts "こんにちは、#{target}。"
    end
  end
end

このコードに対してiexからメッセージを送ってみます。

% iex --sname qux --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(qux@mba11-keith)1> c("hello0.ex")
[Hello]
iex(qux@mba11-keith)2> target = spawn(Hello, :hello, [])
#PID<0.55.0>
iex(qux@mba11-keith)3> target <- {:english, 'World'}
Hello, World.
{:english, 'World'}
iex(qux@mba11-keith)4> target <- {:japanese, '世界'}
{:japanese, [19990, 30028]}
iex(qux@mba11-keith)5> target <- {:english, 'World'}
{:english, 'World'}

hello0.exをコンパイルして、プロセスとして立ち上げ、 <- 演算子でメッセージを送っています。
立ち上げたプロセスは、{:english, "World"}というメッセージに対しては Hello, World. と返してくれていますが、その後{:japanese, '世界'}, {:english, 'World'}というメッセージに対しては、何も返してくれていません。どうなってしまったのでしょうか？

これは、receive文がループしないため、{:english, 'World'}に対する応答を実行した後に終了してしまったことによるものです。
そのため、明示的に待ち受けループを起こさせる必要があります。

hello1.ex

defmodule Hello do
  def hello do
    receive do
      {:english, target} ->
        IO.puts "Hello, #{target}."
        hello
      {:japanese, target} ->
        IO.puts "こんにちは、#{target}。"
        hello
    end
  end
end

各メッセージに対応する処理の定義の中で、最後にhello関数を呼ぶ処理を追加しました。これで、プロセスは何度でもメッセージに応答することができます。

% iex --sname qux --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(qux@mba11-keith)1> c("hello1.ex")
hello1.ex:1: redefining module Hello
[Hello]
iex(qux@mba11-keith)2> target = spawn(Hello, :hello, [])
#PID<0.55.0>
iex(qux@mba11-keith)3> target <- {:english, 'World'}
Hello, World.
{:english, 'World'}
iex(qux@mba11-keith)4> target <- {:japanese, '世界'}
こんにちは、世界。
{:japanese, [19990, 30028]}
iex(qux@mba11-keith)5> target <- {:english, 'World'}
{:english, 'World'}
Hello, World.

プロセスは何度でも処理を実行できるようになりました。
ここで、このプロセスが知らないメッセージを投げてみます。

iex(qux@mba11-keith)6> target <- {:german, 'Welt'}
{:german, 'Welt'}

メッセージに対して何も処理は行われませんでした。ここで、知らない処理に対する応答を定義してみます。この場合、anonymous variable(無名変数と呼べばいいでしょうか) _ を使用します。

hello2.ex

defmodule Hello do
  def hello do
    receive do
      {:english, target} ->
        IO.puts "Hello, #{target}."
        hello
      {:japanese, target} ->
        IO.puts "こんにちは、#{target}。"
        hello
      _ ->
        IO.puts "Huh?"
        hello
    end
  end
end

% iex --sname qux --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(qux@mba11-keith)1> c("hello2.ex")
hello2.ex:1: redefining module Hello
[Hello]
iex(qux@mba11-keith)2> target = spawn(Hello, :hello, [])
#PID<0.55.0>
iex(qux@mba11-keith)3> target <- {:japanese, '世界'}
こんにちは、世界。
{:japanese, [19990, 30028]}
iex(qux@mba11-keith)4> target <- {:english, 'World'}
Hello, World.
{:english, 'World'}
iex(qux@mba11-keith)5> target <- {:german, 'Welt'}
{:german, 'Welt'}
Huh?

うまく処理できました。
receive文におけるループの性質を利用して、終了処理を書いてみます。

hello3.ex

defmodule Hello do
  def hello do
    receive do
      {:english, target} ->
        IO.puts "Hello, #{target}."
        hello
      {:japanese, target} ->
        IO.puts "こんにちは、#{target}。"
        hello
      :exit ->
        IO.puts "Bye!"
      _ ->
        IO.puts "Huh?"
        hello
    end
  end
end

終了メッセージの後ではメッセージに応答しないことを確認します。

% iex --sname qux --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(qux@mba11-keith)1> c("hello3.ex")
hello3.ex:1: redefining module Hello
[Hello]
iex(qux@mba11-keith)2> target = spawn(Hello, :hello, [])
#PID<0.55.0>
iex(qux@mba11-keith)3> target <- :exit
:exit
Bye!
iex(qux@mba11-keith)4> target <- {:english, 'World'}
{:english, 'World'}

意図した通りになりました。

最後に、簡単なサンプルコードと実行例を紹介して終わりにします。

ticker.ex

defmodule Tick do
  @interval 2000
  @name :ticker

  def start do
    pid = spawn(__MODULE__, :generator, [[]])
    :global.register_name(@name, pid)
  end

  def register(client_pid) do
    :global.whereis_name(@name) <- {:register, client_pid}
  end

  def generator(clients) do
    receive do
      {:register, pid} ->
        IO.puts "registering #{inspect pid}"
        generator([pid|clients])
    after
      @interval ->
        IO.puts "tick"
        Enum.each clients, fn client -> client <- {:tick} end
        generator(clients)
    end
  end
end

defmodule Client do
  def start do
    pid = spawn(__MODULE__, :receiver, [])
    Tick.register(pid)
  end

  def receiver do
    receive do
      {:tick} ->
        IO.puts "tock in client"
        receiver
    end
  end
end

% iex --sname qux --cookie elixir-example
Erlang R16B02 (erts-5.10.3) [source] [64-bit] [smp:4:4] [async-threads:10] [hipe] [kernel-poll:false]

Interactive Elixir (0.11.2) - press Ctrl+C to exit (type h() ENTER for help)
iex(qux@mba11-keith)1> c("ticker.ex")
[Client, Tick]
iex(qux@mba11-keith)2> Node.connect :"foo@mba11-keith"
true
iex(qux@mba11-keith)3> Node.connect :"bar@mba11-keith"
true
iex(qux@mba11-keith)4> Node.connect :"baz@mba11-keith"
true
iex(qux@mba11-keith)5> Tick.start
:yes
tick
tick
tick
tick
tick
iex(qux@mba11-keith)6> Client.start
registering #PID<0.76.0>
{:register, #PID<0.76.0>}
tick
tock in client
tick
tock in client
tick
tock in client
tick
tock in client

iex(foo@mba11-keith)11> c("../nodes/ticker.ex")
[Client, Tick]
iex(foo@mba11-keith)12> Client.start
{:register, #PID<0.86.0>}
tock in client
tock in client
tock in client

iex(bar@mba11-keith)7> Client.start
{:register, #PID<0.82.0>}
tock in client
tock in client
tock in client

iex(baz@mba11-keith)2> Client.start
{:register, #PID<0.75.0>}
tock in client
tock in client
tock in client

まとめ

ElixirにおけるMessage Passingの基本を紹介しました。Actor Modelを使うと並列処理がシンプルに書けますし、Erlang/OTPやScalaだけでなく、Elixirでもちゃんと動くものが書けます。
明日は @ma2ge さんです。

参考文献

• Programming Elixir ( http://pragprog.com/book/elixir/programming-elixir )
• http://www.sitepoint.com/elixir-love-child-ruby-erlang/