前回のevelに続き、今回は分散Erlang用のプロセスグループ管理ライブラリであるppg(v0.1.3)の紹介。
"ppg"は"__P__lumtree based distributed __P__rocess __G__roup"の略。

依存関係的にはこちらがevelを使っている形になるが、実装時期的にはppgの方が先となる(後者に適したリーダ選出ライブラリが欲しくて前者を作った)。

完全に同じ、という訳ではないが機能的にはErlangの標準ライブラリが提供するpg2(プロセスグループ管理モジュール)に似せてある。
ただ、(分散Erlang周りの性能測定メモにも記載の通り)pg2がクラスタ内のノード数が増えた場合のスケーラビリティに難があるのに対して、ppgはスケールしやすい作りになっている。
またグループの作成やグループへのメンバ追加・削除のコストも比較的低い(クラスタサイズに依存しない)ので、メンバ数が少ないグループ群を頻繁に作ったり破棄したりする用途にも向いていると思う。

仕組み的には、基本的に以下の二つの論文のアルゴリズムを(おおむね)忠実に実装したものとなっている:

それぞれのアルゴリズムの内容に関しては、後で軽く触れるが、まずはppgの使い方(実行例)の紹介から始める。

使い方(実行例)

create/1を使ってグループ作成、join/2でメンバ追加を行うのはpg2と同様。

%% 'foo'という名前のグループを作る (このグループは他の分散ノードからも利用可能)
> ppg:create(foo).

%% グループ一覧
> ppg:which_groups().
[foo].

%% グループに五名のメンバを追加 (簡単のために全て`self()`)
%% 各メンバには追加の度に、専用の(ブロードキャスト)チャンネルプロセスが割り当てられる
> [{ok, Channel} | _] = [ppg:join(foo, self()) || _ <- lists:seq(1, 5)].

%% グループメンバに対して'hello_world'メッセージをブロードキャスト
%% ※ 効率上の理由から、pg2とは異なりブロードキャスト関数が提供されている
%%   (pg2は「グループ所属のメンバ一覧をローカルETSから取得できるようにしておくから後は好きにして」というスタンスだが、
%%    ppgは、ローカルにメンバ一覧の情報を保持していないため)  
> ppg:broadcast(Channel, hello_world).
> flush().
Shell got hello_world
Shell got hello_world
Shell got hello_world
Shell got hello_world
Shell got hello_world
ok

%% メンバ一覧
> ppg:get_members(foo).
{ok,[{<0.190.0>,<0.198.0>},  % {メンバPID, チャンネルPID}
     {<0.190.0>,<0.201.0>},
     {<0.190.0>,<0.200.0>},
     {<0.190.0>,<0.199.0>},
     {<0.190.0>,<0.193.0>}]}

%% メンバ削除
> ppg:leave(Channel).  % チャンネルに対して`leave/1`を呼びだす
> ppg:get_members(foo).
{ok,[{<0.190.0>,<0.201.0>},
     {<0.190.0>,<0.199.0>},
     {<0.190.0>,<0.200.0>},
     {<0.190.0>,<0.198.0>}]}

> exit(normal). % メンバプロセスが死んでも削除される
> ppg:get_members(foo).
{error,{no_reachable_member,foo}}

%% グループ削除
26> ppg:delete(foo).
27> ppg:which_groups().
[]

他にもいくつか関数はあるが、よく使うのはこれくらい。

性能

包括的な性能測定を行っている訳ではないが、参考までに、クラスタ内のノード数が増えた場合の
メンバの追加コストの変化に関するpg2との比較結果を載せておく。

メンバの追加

測定方法は以下の通り:

一つのマシン上でslaveモジュールを使って__"ノード数"__分のノードを起動する
作成するグループは一つ
メンバは、一ノードに付き一プロセスを追加する
追加方法は__"serial"と"parallel"__の二種類:
- serial: 各メンバを(各ノード上で)順々に追加していく
  - イメージ: ppg:join(foo, A), ppg:join(foo, B), ... .
- parallel: 全メンバを同時に(並行して)追加する
  - イメージ: spawn(ppg, join, [foo, A]), spawn(ppg, join, [foo, B]), ... .
指標としては「一人のメンバ追加に要した時間の平均値」を測定する
- この平均所要時間がノード数やppg/pg2によってどう変化するか

結果:

ノード数	pg2 (serial)	pg2 (parallel)	ppg (serial)	ppg (parallel)
1	0.07ms		0.17ms
50	17.24ms	94.54ms	11.59ms	4.17ms
100	37.37ms	151.65ms	13.59ms	6.76ms

上の表から(期待通り)ppgの方が、ノード数が増えた場合でもメンバ追加の性能劣化度合いが少ない、ことが見てとれる。
※ 理屈上は、クラスタサイズに依存せずに定数コストでのノード追加・削除が可能なはず。
※ 今回の測定でノード数が増えるに従って、微妙に処理時間が長くなっているのは、全ノードを同一マシン上で起動しているためだと思われる (ノード数が増えるほどマシン自体に掛かる負荷が増える)

反対に、pg2はノード数に比例する形で処理時間が伸びているように見える。

また、pg2は並行的にメンバを追加した場合に(逐次的に行った場合に比べて)性能が劣化しているのに対して、ppgでは逆に並行版の方が性能が向上しているのも面白い。

なお、一応公平を期すために書いておくと、グループ構成の変更(メンバ追加・削除)の場合とは反対に、メンバへのメッセージのブロードキャスト時のコストに関しては、pg2の方が基本的には軽くなっている。
pg2では、ブロードキャスト時には「ローカルETSからメンバ一覧を取得し、それらに対して(lists:foreach/2等を用いて)メッセージを送信する」という処理だけで済む。
ppgでは、内部的にブロードキャスト用のツリー(スパニング木)をクラスタ上に構築しており、そのツリーを用いてメッセージをブロードキャストすることになるが、その際にはメッセージに追加の制御情報を載せたり、管理用の追加処理が行われたりするため、pg2に比べると(定数項ではあるが)処理量・メッセージサイズ、がアルゴリズム全体としては増えてしまう。
ただし「単一プロセスの処理量」という観点で見るのであれば、pg2はブロードキャストメッセージを送信するプロセスの負荷がクラスタサイズに比例して増加するのに対して、ppgでは常に一定(各メンバプロセス上に分散する)なので、どちらがより優れているかは最終的には利用目的次第になるのではないかと思われる。

仕組みの概要

ここではppgの仕組みに関して概要レベルで軽く触れておく。

詳しく(or 正確に)知りたい人には、冒頭に記載の二つの論文を読んでみることをお勧めする。
また、昔に書いたppgに関するメモ書きにも意外といろいろ情報が載っていたので、参考までに挙げておく(ただし内容の質はメモレベルでTODOのまま放置な箇所も少なくない)。

大枠

スケーラブルかつ故障耐性の高いブロードキャストアルゴリズムとして、ゴシップ(or epidemic)ベースのものがある

ゴシップ:
- 始めてメッセージを受け取ったメンバは、同じグループに属する隣人群(グループ内の無作為に選択されたサブセット)に、メッセージを転送する
- [ブロードキャスト] 各メンバが自分の隣人群に伝えることを繰り返せば、その内に皆に情報が伝わるはず
- [スケーラブル] 各メンバは、グループサイズに依らず、一定数の隣人にのみメッセージを転送する
- [故障耐性] たまたまその時に誰か一人がいなくて(e.g., 転送中にプロセスがダウンした)も、その人の隣人には、別の経路から(極めて高い確率で)情報が伝わるはず

Q) でもどうやって自分の隣人を把握すれば良いのか？

A) __HyParView__を使って、メンバ管理を行おう

Q) 隣人全てにメッセージを転送するのは無駄では？

各メンバが、複数経路から同一メッセージを重複して受け取ることなってしまう
A) __Plumtree__を使って、正常時には重複メッセージが発生しないようにしよう

HyParViewだけでもグループのメンバ管理やメッセージのブロードキャストは行えるが、その上にPlumtreeを適用することで、より効率的なブロードキャストを可能にする、という枠組みになっている。

HyParView

"__Hy__brid __Par__tial View"の略。
主にメンバ管理を担当し、冗長性を有する接続グラフの構築・維持を効率的に行う。
数万規模の数のノードを扱えるように(スケーラブルに)設計されている。
故障耐性の高さも売りの一つで、例えば全体の半分くらいのノードがほぼ同時にダウンしたとしても、メッセージの配送(ブロードキャスト)が可能だとか(※ ただしあくまでも確率的なもの)。

仕組み(ざっくり):

HyParViewの各メンバは__"部分ビュー(Partial View)"__を管理する:
- 各メンバは、グループに属するメンバ全員ではなく、そのサブセットである部分ビューに含まれる隣人のみを把握している
部分ビューには__ActiveView__と__PassiveView__の二種類が存在する:
- ActiveView:
  - サイズがlog(N)くらいの小さなビュー　 ※ N = グループに属するメンバ数
  - (論文では)ビュー内のメンバとはTCPで接続しており、メッセージの配送はこれを使って行われる
    - ※ Erlang(ppg)では、TCPではなく通常のプロセス間通信の手段(メッセージ配送、monitorによる監視)を採用
  - TCPを使っているのは死活監視を兼ねているため:
    - TCP接続が切断したら「相手が死んだ」と判断 (これにより早急なダウン検知およびリカバリが可能となる)
    - => その際は、PassiveViewから新しいメンバを補填する
  - ActiveViewには対称性がある (i.e., AがBのビューに属しているなら、その逆も同様)
  - 静止状態では、ActiveViewの構造は安定している:
    - 静止状態: メンバの追加・削除がない状態
  - 正常系の処理(通信)は、基本的にActiveView上で行われ、かつ、そのサイズは十分に小さいので、スケーラブルかつ効率的
- PassiveView:
  - ActiveView内のメンバダウンに備えた予備用のビュー
    - 想定サイズはlog(N) * C　 ※ Cは4~7程度の小さな定数
  - 普段は死活監視(TCP接続)は行わずに単にリストを保持するだけ
    - ただし、定期的に他のメンバのActiveViewの内容とシャッフルして、新鮮さが保たれるようにする
    - => これによりActiveViewを補填される際に選択されるメンバは、高い確率で生きていることになる
メンバ追加時の流れ(大枠):
1. 追加メンバは、まずエントリポイントとなる__コンタクトサービス__に追加依頼メッセージを投げる
  - 既にグループに属しているメンバの中から選択された、リーダ的なメンバ
  - ppgでは、(前回に取り上げた)evelを使って、コンタクトサービス用のメンバを決定している
2. コンタクトサービスは、追加依頼メッセージを、自分のActiveViewに属しているメンバ群に転送する
3. メッセージを受け取ったメンバは、自分のActiveView内から無作為に一人を選んで、さらにメッセージを転送する
  - それとは別に、新規メンバを(条件を満たしているなら)自分のPassiveViewに追加する
4. 転送回数が規定回数に達したら、転送は止めて、新規メンバを自分のActiveViewに追加する
メッセージのブロードキャストは、ActiveViewを使って、通常のゴシップアルゴリズムと同様の方法で行う

なおPlumtreeと組み合わせる場合には、HyParViewレイヤーではブロードキャストは行わず、メンバ管理のみに専念することになる。

Plumtree

"__P__ush-__l__azy-push __mu__lticast tree"の略。
グループ(クラスタ)のメンバ管理は管轄外で、ピアサンプリングサービスと呼ばれる別のレイヤーに委譲されている(その実装の一つがHyParView)。
既に接続性を備えた(i.e., ゴシップブロードキャストが行える)クラスタは存在するものとして、その上で(耐障害性は損なわずに)如何に効率的にメッセージ配送(ブロードキャスト)を行うか、がPlumtreeの主関心。
基本的には、クラスタ上にメッセージブロードキャスト用のツリー(スパニング木)を構築し、正常系ではそのツリー上でのみメッセージが転送されるようにすることで、冗長な転送を減らし、効率化している。

仕組み(ざっくり):

Plumtreeの隣人(HyParViewのActiveViewに相当)同士を繋ぐP2Pチャンネルには__eager__と__lazy__の二つの状態が存在する
eager:
- 初期状態は全てこれ
- ブロードキャスト時には、各メンバは__eager__チャンネル上にメッセージを転送する
- メッセージ受信の際に、そのメッセージを既に別経路から受け取っている場合には、送信元にその旨を通知して、チャンネルの状態を__lazy__に変更する
  - これにより次回以降は、この経路から冗長なメッセージを受信することはなくなる
- 最終的に冗長経路は全ての__lazy__な状態となり、__eager__なチャンネル群によるツリー(スパニング木)が形成される
  - 初期状態のゴシップ配送から「メンバ数がNのグループに対して、N - 1回の転送でブロードキャストが行える」という最適な状態に遷移・収束
lazy:
- 冗長性(故障耐性)を確保するためのチャンネル
- ブロードキャスト時には、このチャンネル上で通常のメッセージが転送されることはないが、代わりに小さなIHaveという制御メッセージが送信される
  - IHaveは対応するブロードキャストメッセージのIDのみを保持
- IHaveメッセージを受け取った後、一定時間が経過しても、メッセージ本体を(別の__eager__なチャンネルから)受信しなかった場合には、その__eager__なチャンネルの状態は__lazy__へ変更される:
  - その場合、__lazy__なチャンネルの中から無作為に一つを選択し「このIDのメッセージを頂戴」と依頼する
  - => 無事メッセージを受け取ったら、このチャンネルは__eager__に昇格
- この仕組みにより、ダウンしたり極端にスローダウンしているメンバがグループ内に存在している場合にでも、メッセージのブロードキャスト自体は滞りなく行えるようになっている
メンバの追加・削除は、ピアサンプリングサービス(e.g., HyParView)任せ:
- メンバが(自分のビューに)追加された場合は、ピアサンプリングサービスがそれをPlumtreeレイヤーに通知する
- 追加メンバとの間のチャンネルは、最初は__eager__として扱い、後は通常と同様に処理をする
- 削除の場合も、基本的には自分のビュー内から除去するだけでOK

上の方式をベースとして、効率性を上げるためのいろいろな工夫等も論文内では説明されているが、ここでは割愛。

終わり

これで今年に実装した分散Erlang関連のライブラリの紹介は完了。
PlumtreeとHyParViewは、単純さと性能や信頼性(?)のバランスが結構良い気がしていて、個人的には割と好印象。
あとErlangは(他の言語に比べると)分散アルゴリズムの実装を試すのがとても楽で良いです。