Erlang で並列プログラミングをやってみた

こんにちは。アドベントカレンダー23日目の記事を担当します、エンジニアの澤田です。

昨年は Template Haskell を使ってメタプログラミングをやってみた という記事を書き、Haskell を勉強しつつ関数型言語に触れてみました。
その中で、関数型言語は並列処理との親和性が高いということを知ったので、また、違う言語に触れてみようと思い、今回は Erlang で並列プログラミングをやってみます！

なお、Erlang のバージョンは Erlang/OTP 22 を使用しています。

並列プログラミングと Erlang

並列処理には大きく「マルチプロセス」と「マルチスレッド」があり、両者には主に以下のような違いがあります。

• マルチプロセス: 各プロセスが固有のメモリ空間を持つ。基本的にプロセス同士が干渉することがなく安全に処理を行えるが、プロセス生成やプロセス間でのデータのやり取りのオーバーヘッドが高い。
• マルチスレッド: スレッド間でメモリを共有する。メモリを共有するため処理が高速だが、共有メモリにアクセスする際の排他制御など、処理が複雑になる傾向がある。

Erlangの特徴として、並列処理を強力にサポートしている点が挙げられます。
マルチプロセスは安全だけど処理が重いというデメリットがありますが、Erlang のプロセスは非常に軽量です。
これは Erlang のプロセスが、OSのプロセスとは異なり、Erlang VM上で生成・管理されているためです。
OSのプロセスとは異なるものの、プロセスの特性(固有のメモリ空間を持つなど) は OSのプロセスと同じで安全に扱うことができ、プロセス間の通信も Erlangのメッセージング機構によって高速かつ簡単に行うことができます。すごいですね。

それでは、実際にプロセスを生成してみましょう！

プロセスを生成してメッセージを送る

プロセスの生成は組み込み関数(BIF: Built-in function) の spawn/3 (※) を使用して、以下のように記述します。
※この 3 は引数の数を表しています。Erlangでは引数の数が異なると、同じ名前でも別の関数として扱われます。

spawn(モジュール名, 関数名, 引数リスト)

プロセスを生成するにはモジュールが必要なので、簡単なモジュールを書いてみます。

-module(echo).
-export([receive_message/0]).

receive_message() ->
	receive
		{Pid, Message} ->
			io:format('~w~n', [Message]),
			Pid ! ok
	end.

上記のコードを、モジュール名と同じファイル名 echo.erl で保存し、保存したディレクトリで erl コマンドを実行して Erlangシェルを起動します。
Erlangシェル上で関数 c(echo). を実行するとコンパイルされて、 echo モジュールと、 -export 属性の指定によって公開された関数が外から使えるようになります。

ここで、receive節に到達するとメッセージを受け取るまで待機することになります。
メッセージを受け取るとその中の各節でパターンマッチが行われて、マッチした場合にその節の本体が実行されます。
そして io:format('~w~n', [Message]) のところで受け取ったメッセージを出力しています( ~ は制御シーケンスで、~w には変数 Message の中身が入り、 ~n には改行が入ります)。

Erlangシェルで、以下の通り spawn関数を実行するとプロセスが生成され、プロセス識別子が返されます。
そして返された結果( <0.80.0> ) を変数 P1 に結合させます。

1> P1 = spawn(echo, receive_message, []).
<0.80.0>

それでは、プロセスにメッセージを送ってみましょう！
メッセージを送るには以下のように記述します。

プロセス識別子 ! メッセージ

プロセス P1 にメッセージを送ります。

2> P1 ! {self(), hoge}.
hoge
{<0.78.0>,hoge}

おお、送ったメッセージ hoge と、評価された式の結果( {<0.78.0>,hoge} ) が返ってきましたね！
もう一度送ってみましょう。

3> P1 ! {self(), piyo}.
{<0.78.0>,piyo}

おや･･･最後に評価された式の結果( {<0.78.0>,piyo} ) は表示されますが、送ったメッセージ piyo が表示されません･･･。 プロセスは生きているのでしょうか？ 確認してみます。

4> is_process_alive(P1).
false

なんと･･･ false が返ってきたので、既に存在していないことがわかりました。
生成したプロセスは、一通り処理が完了すると終了してしまうのです。

プロセスを継続するにはどうすれば良いでしょうか？

再帰でプロセスを存続させる

プロセスを存続させるには再帰を使います。
先のモジュールを少し書き換えて、receive節内の最後で自分自身を実行します。

-module(echo).
-export([receive_message/0]).

receive_message() ->
	receive
		{Pid, Message} ->
			io:format('~w~n', [Message]),
			Pid ! ok,
			receive_message()
	end.

では、もう一度やってみましょう。

1> P2 = spawn(echo, receive_message, []).
<0.80.0>
2> P2 ! {self(), hoge}.
hoge
{<0.78.0>,hoge}
3> P2 ! {self(), piyo}.
piyo
{<0.78.0>,piyo}

今度は上手くいきましたね！

プロセスを並列で動作させる

それでは、いよいよプロセスを並列で動作させてみましょう。

並列動作を確認しやすくするため、指定した回数分、1秒間隔でメッセージを送る関数 send_message/3 を追加して使うことにします。

-module(echo).
-export([receive_message/0, send_message/3]).

receive_message() ->
	receive
		{Pid, Message} ->
			io:format('~w~n', [Message]),
			Pid ! ok,
			receive_message()
	end.

send_message(_, _, 0) -> ok;
send_message(Pid, Message, N) ->
	Pid ! {self(), Message},
	receive _ -> ok end,
	timer:sleep(1000),
	send_message(Pid, Message, N - 1).

まず、メッセージを受け取って表示するプロセスを生成しておきます。

1> P3 = spawn(echo, receive_message, []).
<0.80.0>

そしてプロセスを3つ生成しつつ、非同期でメッセージを送ります。

2> spawn(echo, send_message, [P3, hoge, 5]),
2> spawn(echo, send_message, [P3, piyo, 5]),
2> spawn(echo, send_message, [P3, foobar, 5]).
hoge
<0.84.0>
piyo
foobar
hoge
piyo
foobar
hoge
piyo
foobar
hoge
piyo
foobar
hoge
piyo
foobar

無事並列で実行することができました！

さいごに

個人的にすぐに理解できなかったのが self() を実行して得られるプロセス識別子( <0.78.0> ) が一体何なのか？ ということでした。
Erlangシェルで実行している場合、これはErlangシェル自身のプロセス識別子なんですね。

シェル自身にメッセージを送って、処理されなかったメッセージを flush/0 関数で取り出してみます。

1> self() ! hoge.
2> self() ! piyo.
3> flush().
Shell got hoge
Shell got piyo

面白いですね :)