マルチコア時代の高並列性IOアーキテクチャ Wavy
シングルスレッドではもう遅い。
以前にマルチコア時代の高速サーバーの実装で、「ネットワークIOはマルチスレッドで動かすが、その他の部分はシングルスレッドで動かす」というIOアーキテクチャの実装（mp::iothreads）を紹介しました。iothreadsはロジック部分をシングルスレッドで書けるため実装の手間を抑えることができ、ネットワークIOがボトルネックになるプログラムには特に適していると思われます。
しかし実際にiothreadsを使ってプログラムを書いてみると、非常に負荷が高い状況でシングルスレッドの部分の処理速度がボトルネックになってしまうことがありました。
そこでマルチコアCPUの性能を引き出すために、徹頭徹尾マルチスレッドで動かすIOアーキテクチャを実装してみました。
- 1つのスレッドが、ある時はepoll_wait()し、ある時はread(2)を行い、ある時はイベントを処理するように動作する
- read(2)・プロトコルパース用のスレッドとイベント処理用のスレッドを別々に用意せずに済む
- スレッドの数を節約できる
- スレッド間のデータの受け渡しを省略できるのでメモリ管理がしやすい
- コンテキストスイッチを省略できるため遅延が小さい
- イベント駆動型でも使える一方で、ストリームも扱える
- タスクキュー+スレッドプールの機能が統合されている
- 並列性が高い
複数のスレッドでread(2)を途切れなく次々に実行し続けられるため（波状攻撃。process in waves!）、特に小さいメッセージが次々にやってくるような場合に遅延を減らす効果があると思われます。
以下の図は基盤となっているアーキテクチャを示しています。
epoll_wait()の呼び出しは排他制御し、同時に一つのスレッドだけが実行できるようにします。そして読み込み可能なファイルディスクリプタを取り出したら、すぐにロックを解除します。そのスレッドはそのまま取り出したファイルディスクリプタからread(2)します。このとき既にロックは解除されているので、このスレッドがread(2)している間にも他のスレッドがepoll_wait()→read(2)を実行することができます。
ここではepollにファイルディスクリプタを登録するときに、EPOLLONESHOTフラグを設定しているところがポイントです。読み込み可能なファイルディスクリプタが取り出されると、そのファイルディスクリプタは再度有効化されるまでepoll_wait()で報告されません。このためread(2)がEAGAINになるまで完了していなくても、他のスレッドがすぐにepoll_wait()を実行することができます。
また、取り出されたファイルディスクリプタを再度有効化する操作（epoll_ctl()でEPOLL_CTL_MODを指定）は、スレッドセーフであるようです。つまりread(2)がEAGAINになったら、ロックを取得せずに再有効化する操作を行うことができます（図中の「reactivate」）。
以上の手順によって複数のコネクションを高い並列性で扱うことができます。
さらに、ここにタスクキュー+スレッドプールの機能を統合することができます。
epoll_wait()するときにタスクキューをチェックし、タスクが入っているようならそれを取り出して実行します。このときの方針として以下の方法が考えられます：
- epoll_wait()が実行されていなかったら、タスクキューをチェックせずにepoll_wait()→read(2)を実行する
- 必ずタスクキューにタスクをチェックし、タスク処理を優先して処理する
- タスクキューに溜まっているタスクがN個以下であればepoll_wait()→read(2)を優先するが、そうでなければタスク処理を優先する（N個以上のタスクが溜まらないようにする）
mp::wavyでは３つ目の方針を採用しています。この部分の処理に関しては、タスクキューの実装を優先順位付きキューにするなどの発展が考えられます。
これでイベント駆動型のアーキテクチャは簡単に実装できるようになりました。read(2)して取り出されたメッセージが、処理に時間のかかりそうなものであればタスクキューにpushし、そうでなければそのまま実行します。（時間のかからないメッセージだと処理にかかる時間全体に対して、スレッドの切り替えに発生する遅延の割合が大きくなってしまうため）
ここで、read(2)をしている間やLogicを実行している間も他のスレッドが並列して処理し続けられることから、read(2)をブロッキングIOで行うこともできます。
イベント駆動型のアーキテクチャでは、イベントを処理し終わるたびに状態を変数に保存しておく必要があります。そのため一連の処理が終了するまでに複数のイベントが絡む場合は、状態を保持する変数が増えてプログラムがややこしくなる傾向にあります。あるいはメッセージ指向ではなくストリーム指向のプロトコルの場合は、ストリームをイベント単位に区切って扱わなければならいため大変です。*1
そのような場合は無理にイベント駆動にせず、上の図のようにスレッドを１本占有して処理した方がうまくいくと思われます。
しかしスレッドを占有してしまうと、スレッドの数より多い数のリクエストが同時に到着したときに、スレッドの数が足りないため並列して処理できない可能性があります。スレッドを占有して行っている処理がCPUを使うものであればしょうがないですが、read(2)を待っているのであれば、この待ち時間の間に別のイベントを処理したいところです。スレッドの数を十分増やせば済むことですが、あまり多くしたくない、かもしれません。
そこで、スレッドを占有する代わりにFiber（coroutine, 協調スレッド）を占有する方法が考えられます。※Win32以外の環境ではfiberはスレッド固有だという情報。この方法はダメそうです（コメント欄参照）
メッセージ指向のプロトコルであれば、上の図のようにread(2)して取り出されたメッセージがどのFiberに関するものなのかを探します。もし見つかればそのFiberの処理を再開し、見つからなければ新たなFiberを作ります。Fiberを処理している間もread(2)は並列して行われるため、パイプライン化されたメッセージは並列して処理することができます。
ストリーム指向のプロトコルであれば、下の図のようにファイルディスクリプタをキーとしてFiberを探します。
以上のアーキテクチャではwrite(2)（送信側）に関しては触れてきませんでした。ソケットへのwrite(2)は別のモジュールに分けて扱います。
LogicやFiberの中でソケットへデータを送信したいとき、ブロックしてしまうとスレッドを占有してしまうため並列性が低下する可能性があります。これを避けるため送信用の別のスレッドにデータを受け渡します。
このとき、writev(2)システムコール１回で書き込みきれなかった場合や、既に書き込みきれなかったデータが溜まっている場合のみ、別のスレッドにデータを受け渡します。データが小さくwritev(2)システムコール１回で書き込み切れた場合は受け渡しません（スレッドの切り替えによる遅延を避けるため。また多少時間がかかっても他のスレッドには影響しないこともポイント）。
送信用のスレッドはそれぞれepoll_wait()を行い、ソケットが書き込み可能になるまで待っています。そのスレッドにデータを受け渡したいときは、そのスレッドが持っているキューをロックしてにデータを受け渡し、ロックを解除してepollにソケットを登録します（epollはスレッドセーフ）。
データを受け渡すとき、データはコピーせずに参照を受け渡します。ソケットに書き込むときはwritev(2)を使い、連続していないアドレスを効率よく送信できるようにしています。
この送信用モジュールの実装ではwritev(2)が非同期に行われるために、writev(2)のエラーを受け取れないという問題があります。この問題に関して私の実装では、writev(2)が失敗したら積極的にソケットを壊す（shutdown(2)）ようにしています。これによって確実にread(2)側でもエラーを検出できるため、そのコネクションに結びつけられていたホストがダウンしたことにするなど、異常系の処理を行うことができます。
以上のアーキテクチャはkqueueを使っても実装することができます。Linuxでepoll、Mac OS Xでkqueueを使った実装が動作することを確認しています。同様にSolaris（Solaris 10 3/05以降）ではEvent Portsを使って実装できると思います。
このアーキテクチャを実装したmp::wavyは、mpioライブラリの中に含まれています（Apache License 2.0）。ソースコードは以下の手順でチェックアウトするか、Webから読むこともできます：http://bazaar.launchpad.net/~frsyuki/mpio/trunk/files
$ aptitude install bzr # bazaar >= 1.0 をインストール $ bzr branch lp:msgpack
