2016年に実装した分散Erlang関連のライブラリ(1): evel

今年の前半にevelとppgという分散Erlang関連のライブラリを実装したので、この場を借りて簡単に紹介したいと思います。

---

動機

一言でいえば、evelはリーダ選出(or グローバルな名前付け)を、ppgはプロセス群のグループ管理を行うためのライブラリ。

Erlangの標準ライブラリの中にもglobalとpg2といった似たような役割のモジュールが存在するが(前者がevelと、後者がppgと対応)、どちらもクラスタ内のノード数に対して性能がスケールしない、という問題がある(詳しくは、以前に書いた分散Erlang周りの性能測定メモを参照のこと。あと、この記事には記載はなかったと思うが、一台でも過負荷等で重いノードがいると他のノードもそれに引きずられてしまう可能性があるのも気になるところ)。

globalとpg2にも「整合性が強い」とか「読み込み操作は凄く高速(ローカルキャッシュのみ参照)」といった利点もあるので、用途が合えば便利ではあるけど、もう少し整合性を犠牲にしても良いのでスケール性や可用性が高いものが欲しい、と思い実装してみたのがevelとppgとなる。
※ 誤解を恐れずに書いてしまうとglobalは、更新時にクラスタ全体をスコープとしたロックを獲得することで整合性を保証するような方式 (pg2も内部的にglobalを使用しているので同様)

以降では、この二つのライブラリの使い方や内部の仕組みについて簡単に紹介を行っていく。
※ つもりでしたが、evelだけで結構時間を使ってしまったので、ppgは別の日に分けることにします

evel: v0.1.1

名前は"__EVE__ntual __L__eader election"からとったもので、結果整合性を持つリーダ選出ライブラリのつもり。

使い方は例の見た方が分かりやすいと思うので、まず簡単なサンプルコードを載せる:

% 'foo'という名前の選挙に'self()'が立候補する
% => 複数人(プロセス)が同時に立候補した場合でも、その中の一人のみがリーダとして選出される(返り値となる)
% 
% ※ 表現的に分かりやすいので`self()`および"立候補"を例としているが、自分以外のプロセスを指定することも可能
> Leader = evel:elect(foo, self()).

% 選出されたリーダを取得する
> {ok, Leader} = evel:find_leader(foo).

% 'Leader'の実態は二要素のタプル
% - 'Winner'は立候補したプロセスのPID (上の例だと'self()')
% - 'Certificate'は、そのプロセスがリーダであることを示すプロセス
%    - (例えば)次に出てくる'evel:dismiss/1'によって、リーダを途中で辞めさせることが可能であり、
%      その場合は、'Winner'は生きていても、'Certificate'は死ぬことになる
%    - そのため「あるリーダの有効性」を監視したいなら'Certificate'の方に対してlinkやmonitorを呼ぶのが良い
> {Winner, Certificate} = Leader.

% 実施されていない選挙名が指定された(あるいはリーダが既にいなくなっている)場合は'error'が返る
> error = evel:find_leader(bar).

% 'evel:dismiss/1'を読んだり、リーダプロセスが死んだりすると、その選挙に紐づくリーダはいなくなる
> ok = evel:dismiss(foo).
> error = evel:find_leader(foo).

他にも補助的な関数はいくつか提供されているが、主要な関数はこの三つのみという単純なもの。

「選挙名==プロセス名」と考えれば、globalモジュールと同じように分散環境でのプロセス名管理の目的でも使用が可能(gen_server等のOTPビヘイビアのインタフェースで使えるようにするためのevel_nameというモジュールがあったりもする)。

実装の概要をざっくりと説明すると次のようになる。
(※ 実装したのが結構前なので、もしかしたら怪しい箇所があるかもしれないが、大筋は合っているはず...)

• 選挙(リーダ選出)時: evel:elect/2
  1. その選挙の名前に対応する投票者(ノード)を、Erlangクラスタ内から(例えば)五名選ぶ
     • 選択にはコンシステントハッシュを使用する(ライブラリはhash_ring)
     • => 投票者の数は、クラスタサイズに依存せずに一定数なのでスケールしやすい (and 選挙名毎に集合は変わるので負荷が分散しやすい)
  2. 各候補者は、その投票者群に対して、投票依頼メッセージを投げる
     • 各依頼メッセージの間には全順序が存在し、より小さいものが優先(投票)される
       • 基本的には、より早い時間に発行された依頼の方が、優先度が高くなる
     • 各投票者は(各選挙名に対して)一番優先度が高い立候補者を、自分の投票先として覚えておく
       • 同時にCertificateプロセスをmonitorしておき、それが死んだら投票先はリセットする
  3. 各立候補者は、次のリーダ取得の方法で、選出されたリーダを取得する
• リーダの取得時: evel:find_leader/1
  1. 選挙名に対応する投票者群を取得するところは上と同様
  2. それらに対して、投票先プロセスを教えて貰うように、依頼メッセージを投げる
  3. 応答群の中から、一番小さな(i.e., 優先度が高い)候補者プロセスをリーダと判断して呼び出し元に返す
     • ※ タイムアウト時間を超えても応答を返さなかった投票者プロセスの投票先は単に無視する
  4. その際に{選挙名、リーダ}のペアをETSに保存(キャッシュ)し、次回以降の取得を高速化する
     • なおCertificateプロセスを監視(monitor)しておき、それが死んだらキャッシュは破棄
  5. => globalとは異なり、各ノードは自分が関心のある情報だけを取得・保持するのでスケールしやすい (が読み込み時の初期コストは高くなる)
• リーダを辞める時: evel:dismiss/1
  1. 単にCertificateプロセスを殺せばOK

上述の通り基本的には「クラスタの中から、各選挙名に対応する投票者ノード群(サブセット)をコンシステントハッシュを用いて選択」と「各投票者の投票先を集めて、その中から一番優先度が高いものをリーダとして判断(選出)」を組み合わせただけの単純な方法。
※ 蛇足: コンシステントハッシュの適用や優先度の判断は、各ノードがローカルに行えるのでその際にはノード間の通信は不要
※ 蛇足: 投票者を一人ではなく複数にしているのは、投票者ノード(の一つ)が何らかの理由で重くなったとしても、全体の処理は継続可能にするため(i.e., 可用性のため)

後は、クラスタの構成変更(ノードの追加削除)が行われたケースのことを考慮して「net_kernal:monitor_node/1を使ってノードのup/downを監視し、構成変更があったらコンシステントハッシュリングの再構築、影響を受ける(投票者集合に変化がある)選挙に関しては投票依頼メッセージ等を再送」といったことを行っている。

このようにevelは、(例えば)globalとは異なり、その時点で利用可能な情報のみを使ってリーダの選出(判断)を行っているので、仮に選挙の競合度が極端に高かったり、ノードの大幅な追加や削除が一度に行われたり、ネットワーク故障によりErlangクラスタが分断してしまった場合には、同時に複数のリーダが存在することがありえる (クラスタの接続性が満たされている限りは、最終的には単一のリーダに収束する)。
次の例が分かりやすい:

% 標準の`slave`モジュールを(内部的に)用いて、50ノードをローカルに起動する
> ok = evel_debug:slave_start_n(50).
> nodes().
['1@localhost','2@localhost','3@localhost','4@localhost',
'5@localhost','6@localhost','7@localhost','8@localhost'|...]

% 補助関数: 引数の関数`F`を、クラスタ内の全ノード上で実行し、結果を集める
> RpcMultiApply = fun (F) -> {Result, []} = rpc:multicall(erlang, apply, [F, []]), Result end.

% 各ノード上で、異なる候補者が同時に立候補する
> RpcMultiApply(fun () -> evel:elect(foo, spawn(timer, sleep, [infinity])) end).
> lists:usort(v(-1)).
[{<10721.296.0>,<10721.298.0>},
 {<10720.298.0>,<10720.300.0>},
 {<10729.307.0>,<10729.309.0>}]  % この時点では、三つのリーダプロセスが並存している

> timer:sleep(100). % 適当な時間だけ待機

% もう一度、全ノード上で選挙を行ってみる
> RpcMultiApply(fun () -> evel:elect(foo, spawn(timer, sleep, [infinity])) end).
> lists:usort(v(-1)).
[{<10720.298.0>,<10720.300.0>}] % 上の中から、一つのプロセスにリーダが収束している

このようなデメリットはあるが、スケール性や可用性はglobalよりも高い。

上の続きとして、globalとの比較の例を見てみる。

まずはスケール性の比較:
※ 本来は、ノード数を変えての性能変化を見るべきだが、手抜きでノード台数が多い場合の処理時間の測定のみを行っている

% [global]
% a) 全てのノードが、同じ選挙に参加する場合の処理時間
> timer:tc(fun () -> length(RpcMultiApply(fun () -> global:register_name(bar, spawn(timer, sleep, [infinity])) end)) end).
{7573849,51} % 7.573 秒

% b) 各ノードが、異なる選挙に参加する場合の処理時間
> timer:tc(fun () -> length(RpcMultiApply(fun () -> global:register_name(node(), spawn(timer, sleep, [infinity])) end)) end).
{9992855,51} % 9.992 秒

% [evel]
% a) 全てのノードが、同じ選挙に参加する場合の処理時間
> timer:tc(fun () -> length(RpcMultiApply(fun () -> evel:elect(bar, spawn(timer, sleep, [infinity])) end)) end).
{99444,51} % 0.099 秒

% b) 各ノードが、異なる選挙に参加する場合の処理時間
> timer:tc(fun () -> length(RpcMultiApply(fun () -> evel:elect(node(), spawn(timer, sleep, [infinity])) end)) end).
{324833,51} % 0.324 秒

単一のマシン上で、無理やり多数のノードを動かしていることもあって、globalとevelでは、平行度合が高い場合の処理時間に顕著な差がみられる。
※ 仕組み上、evelはノード数をより増やしていったとしても、(理想的には)処理時間が変わらないことが期待できる

次は可用性の比較の例:

% 準備: あるスレーブノードのbeamプロセスをサスペンドする
> os:cmd("kill -19 ${A_SLAVE_OS_PID}").

% globalの何らかの関数を呼び出す => 上のプロセスが再開するまでブロックしてしまう
% ※ なお、読み込み系の操作であれば、ローカルのETSのみで完結するので、ブロックはしない
> global:register_name(baz, self()). % `Ctrl+G => i => c`でシェルをリセットする

% evelの場合は、単一ノードが停止していても影響を受けずに、処理が継続可能
> evel:elect(baz, self()).
{<0.715.0>,<0.718.0>}

globalはスケーラビリティの問題よりも、こっちの「クラスタ内の不調ノードの影響を受けやすい」という性質の方が恐いな、と個人的には思います。

evelも(明らかに)万能という訳ではないですが、整合性をそこまで重視しない(というかリーダの並存が致命的ではなく許容可能な)利用途であれば、globalよりは使いやすいのではないかと思います。
※ ただ、ちゃんと関連ライブラリの調査を行っている訳ではないので、

---

今日はここまで。
ppgの話はおそらく18日に書きます。
(ちなみにppgは内部的にevelを使っていたりするので、その点でもglobalとpg2の関係に似ているところがあったりします)