この記事は、「Elixir or Phoenix Advent Calendar 2017」の5日目です。
昨日は @tuchiro さんの「ElixirでSI開発入門 #1 Ectoで悲観的ロック」でした。
fukuoka.ex
twinbeeことenぺだーしと申します。
福岡でfukuoka.exという活動をやっております。
先日こちらサイトでfukuoka.exのコアメンバーとして取材を受けました。福岡でのElixirのコミュニティー活動に興味がある方は是非ご覧ください。
FlowとGenStage
ElixirにはFlowという並列ストリームを直観的に使えるライブラリがありますが、本格的に使おうとすると、GenStageの知識が必用になってきます。名前の通りStageがどういうものかも直観的に理解ができます。
Flowの簡潔さに比べて、GenStageのコードは一見わかりにくいのですが、定義の一つ一つは簡単です。JavaやC#、Javascriptなどオブジェクト指向から入ってこられた方は、Behaiviorのコールバック実装が継承クラスのメソッド実装に似ているので、こちらのほうが馴染みやすいかもしれません。
公式ドキュメントを元にGenStageに入門してみましょう。
まずはプロジェクトを作成します。
# mix new gentest
# cd gentest
defp deps do
[
{:gen_stage, "~> 0.13.1"} # 追加
]
# mix deps.get
実装の手順
サンプルプログラムは、数字をカウントしながら無限にリストを流すという内容です。
最後のほうにアニメーションGIFがありますので、そちらをご覧になるとイメージが掴めると思います。
今回出来上がったプログラムはA,B,Cはそれぞれ並列ですが、各工程はシングルプロセスで動作します。
ステージには３種類のタイプがあります。
ストリームの入り口から消費先まで、３段階のステージが分かれています。
A) Producer ストリーム発生元(生産者・供給者）
B) Producer-Consumer 中間加工（生産者－消費者）
C) Consumer 結果 （消費者）
今回の例では以下の流れで実装していきます。
１．上記ＡＢＣそれぞれに、コールバックを設置する。
２．ＡＢＣの購読(配管)を行う。
「ＡＢＣの各ステージを作って購読(配管)すれば、勝手にストリームが流れ出す！」それがGenStageです。（GenStageの一部です）
各モジュールの実装
以下ABCモジュールは同一ファイルに書いてもOKです。lib\gen.ex等適宜ファイルを作成してコードを書いて（コピペして）いきましょう。
Aモジュール (Producer)
defmodule直後のuse句はマクロ使用の予約語でGenStageはマクロを利用した言語拡張で書かれていることがわかります。
モジュール内の下記関数はコールバックなので、手動で呼び出すことはなく、処理の流れの中でイベントを受け取った際に勝手に実行されます。
-
initステージのタイプ:producerの宣言、カウンター初期値を設定しています。 -
handle_demand
引数のdemandは要求数、counterは現在値です。現在値～｛現在値＋要求数）のリストを作って次のタプルを返します。
{ :noreply, 結果のリスト, 次回の開始値 }
ソースを見ると一目瞭然なので、解説読むほうがしんどいかもしれませんね。
生産者なのに、需要であるdemandが引数になってるのがポイントですが、何か変だな？と感じた方は鋭いです！ それは次回に解説します。
defmodule A do
use GenStage #「このモジュールはGenStageとして使用する」宣言
def start_link(number) do
GenStage.start_link(A, number) # 本モジュールをプロセスに登録
end
def init(counter) do
{:producer, counter} # "生産者"宣言
end
def handle_demand(demand, counter) when demand > 0 do
events = Enum.to_list(counter..counter+demand-1) #リストを作成
{:noreply, events, counter + demand}
end
end
形式的には、終始この形で下記３種のコールバックを書いていく流れとなります。
これ以上難しいものは、もう出てきません。
- start_link … モジュールをプロセスとして起動する際に呼び出されるコールバック
- init … 初期化コールバック
- handle_events / handle_demand … データの受取、需要発生時のコールバック
Bモジュール (producer-consumer)
ステージAから送られてきたデータに、numberを掛けて次のステージに流す中間加工所です
-
initで生産者-消費者タイプを登録 -
handle_eventsで、リストの各要素にnumberを掛け合わせます。
defmodule B do
use GenStage
def start_link(number) do
GenStage.start_link(B, number) # 本モジュールをプロセスとして登録
end
def init(number) do
{:producer_consumer, number} # 中間加工業者 宣言
end
def handle_events(events, _from, number) do # データ受取った時にする仕事を宣言
events = Enum.map(events, & &1 * number) # 加工処理の本体
{:noreply, events, number} # 次の工程へ
end
end
Cモジュール(cosumer)
initで消費者タイプを登録
handle_eventsで、１秒ごとに前のステージBから送られてきたリストを表示します
defmodule C do
use GenStage
def start_link() do
GenStage.start_link(C, :ok)
end
def init(:ok) do
{:consumer, :the_state_does_not_matter} # "消費者" 宣言
end
def handle_events(events, _from, state) do
Process.sleep(1000) # 1秒待つ
IO.inspect(events) # リストを表示
{:noreply, [], state} # 消費者なので、次のステージに渡すリスト(events)はなし
end
end
購読(配管)
ここも、解説することがないぐらいシンプルです。
前半３行でABCそれぞれのプロセスを引数を与えて起動、後半２行で各ステージの購読先を登録。
これだけです。
{:ok, a} = A.start_link(0) # 0から始めます
{:ok, b} = B.start_link(2) # 2を掛けます
{:ok, c} = C.start_link() # 状態は不要
GenStage.sync_subscribe(c, to: b)
GenStage.sync_subscribe(b, to: a)
５行目を書き終えたら、配管が完成するのでストリームが流れ出します。
１秒ごとに、２倍されたカウンターのリストが表示されると思います。(gifは0.5秒)
永遠に流れ続けるので、Ctrl+Cで適当に終わってくだい。
サンプルを動かすと、どういうものか一目瞭然だと思います。
次回予告
お疲れ様でした！
次回はElixirのGenStageに入門する#2 バックプレッシャーを理解するになります。
明日はzacky1972さんによる
ZEAM開発ログv0.1.0 Flow / GenStage による並列プログラミング入門です。