社員のToshiさんによって元のレポートは執筆されています。
第1章はじめに
近年，サービスの状態を保存するためにRDBMSがよく利用されている．代表的なRDBMSとしてMySQL[1] やOracle[2]，PostgreSQL[3] などが挙げられる．RDBMSは，TCOの低さ，トランザクション機能，SQL のような柔軟な問い合わせ言語でのやり取りが可能であるなどのメリットがある．しかしながら，Web アプリケーションにおいて，RDBMS のような複雑なクエリは不要な場合が多く，プライマリキーでアクセスできればよい．また，RDBMS では，一貫性を保つために，ロックなどを用いるため，可用性を犠牲にする．さらに，近年進歩はしているものの，RDBMSにおいてスケールアウトや負荷分散はいまだ難しい問題となっている．
これらの問題を解決する手段として，Key-Value Store が最近注目を集めている．Key-Value Store の例として，memcached[4]，Redis[5]，Google Big Table[6]，TokyoTyrant[7] などが挙げられる．memcached は，C言語で書かれた高性能な分散メモリキャッシュサーバである．主にDB への問い合わせの結果などをキャッシュし，Webシステムにおける性能向上のために利用される．しかしながら，memcached はデータがメモリに保存されるため，プロセスが落ちてしまった場合，データが消失してしまうという問題がある．Redis は，memcached と同様にキーと値の対をメモリ上に保存するが，memcached とはなり，同時に一定量以上の変更が加えられた場合には，非同期でディスクに書き出すことで永続性を実現している．しかし，プロセスが落ちた際に，ディスクに書き出されていないデータは失われてしまう．インメモリDB の高速性とディスクに書き出す一般的なDB との中間的な位置づけである．Google Big Table は，Google が開発した，列からの高速な読み込みに焦点を当てた列指向データベースであり，Google Reader，Google マップ，Youtube などいくつものGoogle のアプリケーションで利用されている．テーブルはGoogle File System[8]に最適化されており，大きなテーブルは複数のタブレットセグメントに自動的に分割される．ただし，Google File System はマスタノードが存在するので，シングルポイントになりやすく，マスタノードで障害が発生した場合サービスが停止してしまう．Tokyo Tyrant はmixi の平林氏が開発したKey-Value Store であり，同氏が開発したTokyo Cabinet DBM[9] のネットワークインターフェースである．Tokyo Tyrant は，データベースサーバを稼働させたまま，データベースファイルのコピーを作成でき，マルチマスタのレプリケーション構成を採用することで可用性を高めている．ただし，Google File System と同様，マスタノードがシングルポイントになりやすいという問題を解決できているとは言えない．
そこで，上記の問題を解決する手段の１つとして，本レポートではP2P 型Key-Value Store を提案する．本実装では，P2Pトポロジを採用しているため，マスタノードが存在せず，スケールアウトや負荷分散が容易になる．また，ローカルストレージとしてメモリとディスクの選択が可能であり，データレプリケーションを行うことで冗長性を確保している．さらに本実装では，可用性を確保するために，EventuallyConsistent を採用している．Eventually Consistent とは，一時的に不整合な状態が生じてもある期間の後には，整合な状態になるような性質のことである．EventuallyConsistent を採用することで，一時的に一貫性を犠牲にするが，高可用性を実現することができる．
本レポートでは，第2 章で実装したP2P 型Key-Value Store のアーキテクチャについて述べ，第3 章で実装について説明し，第4 章でまとめを行う．
第2章アーキテクチャ
本章では，実装したP2P 型Key-Value Store のアーキテクチャについて述べる．まず，本実装の外部インターフェースについて述べ，次にデータパーティショニングやデータレプリケーションについて説明し，最後に，put，get の手順について説明する．
2.1 外部インターフェース
本実装では，外部インターフェースとして，以下の3 つを実装している．
• get(key)
• put(key，context，value)
• delete(key, context)
context はバージョン情報などを含んでいる．アプリケーション側ではバイト列として扱い，内部構造を把握しなくてよい．get(key) は，value(conflict してれば複数) とそのcontext を返す．put(key，context，value) は，格納先を決定し，value とcontextをディスクに書き込む．delete(key, context) は，put 同様，格納先を決定し，key に対応するobject を削除する．
2.2 データパーティショニング
Fig. 2.1: コンシステント・ハッシュ法
本実装ではコンシステント・ハッシュ法[10] を用いることで，データのパーティショニングを実現している．以下にコンシステント・ハッシュ法について述べる．コンシステント・ハッシュ法は，複数のサーバから成るキャッシュを運用する場合にその特性を発揮する．単純なハッシュ法においては，サーバの追加や削除の際に全てのオブジェクトは新しい位置を求めるためにハッシュしなおす必要があるため，対応ができない．このハッシュの全再計算の必要がないのがコンシステント・ハッシュ法である．
ハッシュ関数はオブジェクトとキャッシュを一定の値域に写像する．ハッシュ値の値域が円周に写像されると考える．Fig.2.1 左は，4 つのオブジェクト(1,2,3,4) と3つのキャッシュ(A,B,C) が円周に配置されている．ハッシュでの配置場所には点でマークを表す．各ハッシュ値がどのノードに入るのかを調べるには，ノードの点にぶつかるまでハッシュ値の点を時計周りに動かせばいい．Fig.2.1 左では1,4 がノードA に入り2 はB,3 はC に入る．C が削除されたとするとは3 はA に入る．他の対応は変わらない．Fig.2.1 右に示す位置へノードD が追加されたら3,4 がD に移る．そして1 だけがA に残る．乱数を使用しているので，ノード間でハッシュ値の分布が非一様になることがある．これを解決するのが”仮想ノード” である．仮想ノードではノードの点を円周上で複製する．1 つのノードを追加するたびに，円周には複数の点が配置される．仮想ノードの複製数が大きいほどハッシュ値の分布が一様になる．また，各ノードの仮想ノード数を調節することで，対応オブジェクト数の比率を変更し，性能差を解消することができる．
本実装では，ハッシュ・アルゴリズムとしてMD5(128bit) を用いている．コンシステント・ハッシュ法によって決定されたノードは，コーディネータと呼ばれる．また，2.4 で説明するGossip プロトコルを用いて，コンシステント・ハッシュ法の環情報を共有する．全ノードで環情報を共有することによって，Zero-hop で担当ノードを決定できる．さらに，memcached とは異なり，アプリケーション側は全てのノードを知る必要はなく，どのノードにリクエストを送ってもよい．
2.3 データレプリケーション
本実装では，各データは調整可能なN台のノードに複製される．コンシステント・ハッシュ法によって決定されたコーディネータが複製に責任を持つ．コーディネータがk 番目のノードとした場合，その複製はk+1，k+2· · · k+N-1 番目のノードに複製される．
Fig. 2.2: レプリケーション(k=3 の場合)
2.4 メンバーシップ
本実装では，メンバーシップを維持するために，Gossip プロトコル[11] を用いている．Gossip プロトコルは，多くのノードで情報を共有するときによく使われるプロトコルである．各ノードは，毎秒，ランダムに選んだノードとメンバーシップ履歴(ノードの参加・離脱履歴) を交換する．メカニズムはシンプルだが，情報を指数的な速度で拡散することができる．また，途中でいくつかのノードに障害が発生しても，全てのノードに情報を拡散できる．このようにして，メンバーシップもEventuallyConsistent に管理されている．ただし，Gossip プロトコルは無駄な処理も多いため，オーダーは1K までであり，それ以上の大規模な環境で動作させることは難しい．
Fig. 2.3: Gossip プロトコルの動作例
2.5 put，get の手順
本実装では，複製の整合性を保つためにQuorum プロトコル[12] を用いている．Quorum プロトコルを一言で説明すると，「データはN 個に複製されるが，取得はN個中R個からのみ成功すればよく，同様に保存はN個中W個に成功すればよい」というアルゴリズムである．データを担当するN 個のノードに対して一斉にリクエストを送信し，取得であればR 個，保存であればW個のレスポンスを受信した時点で，残りのレスポンスを待たずにクライアントへレスポンスを返す．なお，保存する際にはデータにバージョン番号を付加し，取得する際には取得したデータの中から最新バージョンを選択して返す．N，R，Wの値は下記の関係を満たすように設定する．
R +W > N (1)
W > N/2 (2)
式(1) を満たすことで，読み取った値のうち少なくともひとつは直前に成功した書き込み(最新バージョン) であるということになり，Write/Read Conflict を防ぐことができる．Write/Read Conflict の例をFig.2.4 に示す．N:R:W = 3:1:2 である場合，式(1) を満たしていないため，Write/Read Conflict が発生する．Fig.2.4 は，2個のノードD, E に保存を行なった例であるが，ノードF から取得を行うと，期待されたデータを取得する事に失敗する．
Fig. 2.4: N:R:W = 3:1:2 の例（Write/Read Conflict 有り）
また，式(2) を満たすことで，連続して書き込みを行ったときに，少なくとも1 つは最新バージョンを置き換えられるこになり，Write/Write Conflict を防ぐことができる．Write/Write Conflict の例をFig.2.5 に示す．N:R:W = 3:3:1 である場合，式(2) を満たしていないため，Write/Write Conflict が発生する．Fig.2.5 は，ノードDに保存を行った後，更にノードF に保存を行った例であるが，この時点で既に，同じキーであるにも関わらず因果関係のない異なる2 つのデータがクラスタ内に存在しているため，取得を行った際には，2 つの値を取得する事になる．
Fig. 2.5: N:R:W = 3:1:3 の例（Write/Write Conflict 有り）
ここで，N:W:R = 3:2:2 の時に，クライアントがput とget した際の処理について述べる．
2.5.1 put の手順
書き込みの際は，コンシステント・ハッシュ法によって対象となる3 個のノードを決定する．そして，決定した3 個のノードに対して一斉にリクエストを送信する．そのうちの2 個のノードからレスポンスが返ってきた時点で残りのレスポンスを待たずにクライアントへ返す．このとき，残りの1 ノードでエラーが発生したとしても無視する．エラーが発生していた場合，一時的に不整合な状態となる．
Fig. 2.6: put の手順
2.5.2 get の手順
読み込みの際も書き込み時と同様，コンシステント・ハッシュ法によって対象となる3 個のノードを決定し，そのノードに対してリクエストを送信する．そのうちの2 個のノードからレスポンスが返ってきた時点で，それらのデータのバージョンが等しければ，クライアントにそのデータを返す．データのバージョンが等しくないときは，最後のレスポンスを待ち，2 つのバージョンの等しいデータをクライアントに返す．そのとき，バージョンの異なるデータを正しいデータで上書きする．こうして，結果的に整合性が保たれる(Eventually Consistent)．
Fig. 2.7: get の手順
Quorum プロトコルのメリットして，たとえ1 台が過負荷や障害に見舞われても，式(1),(2) を満たしている限りデータにアクセスできるため可用性が維持され，その際に応答時間が増加することはない．また，N:W:R を調節することにより，データ取得のレスポンス速度とデータ保存の応答速度のどちらを重視するのかを変更できる．
第3章実装
本実装は，すべてJava 言語で書かれている．ローカルストレージに関しては，Java言語のHashMap か，Tokyo Cabinet DBMを選択できる．また，通信フレームワークとして，Netty[13] を使用した．Netty は，NIO によるサーバ・クライアントフレームワークであり，容易にネットワークアプリケーションを開発することが可能になる．また，外部インターフェースとしてはテキストベースのシェルインターフェースを実装している．今後として，memcached プロトコル互換のインターフェースの実装を考えている．現在では，まだ公開できる状態ではないが，将来的には公開したいと考えている．
第4章まとめ
本レポートでは，RDBMSの可用性，負荷分散，スケールアウト問題やマスタノードのシングルポイント問題を解決すべく，P2P 型Key-Value Store を提案した．本実装では，コンシステント・ハッシュ法を用いることでパーティショニングを実現し，Quorum プロトコルを採用することにより，データレプリケーションを維持しつつ，高可用性を実現した．また，メンバーシップの維持のためにGossip プロトコルを用いた．
今後の課題としては，障害検知処理や，動的にノードの追加・削除の実装，パフォーマンスの検証が挙げられる．
参考文献
[1] MySQL, http://www.mysql.com/
[2] Oracle, http://www.oracle.com/index.html
[3] PostgreSQL, http://www.postgresql.org/
[4] memcached, http://www.danga.com/memcached/
[5] Redis, http://code.google.com/p/redis/
[6] Google Big Table, http://labs.google.com/papers/bigtable-osdi06.pdf
[7] Tokyo Tyrant, http://1978th.net/tokyotyrant/
[8] Google File System, http://labs.google.com/papers/gfs-sosp2003.pdf
[9] Tokyo Cabinet DBM, http://1978th.net/tokyocabinet/
[10] Consistent hashing and random trees: distributed caching protocols
for relieving hot spots on the World Wide Web，
http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.23.3738&rep=rep1&type=pdf
[11] Gossip protocol, http://en.wikipedia.org/wiki/Gossip protocol
[12] Quorum protocol, http://en.wikipedia.org/wiki/Quorum (Distributed Systems)
[13] Netty, http://jboss.org/netty/
※1 2009年9月に開催