@twitterこの広告は、90日以上更新していないブログに表示しています。古くなってしまったので、ここでまとめ直しておこうと思います。MessagePack-RPCのプロトコルは、非同期型の呼び出しができる点(非同期プロトコル)が大きな特徴となっています。複数のサーバが相互に通信し合ったり、通信と計算をオーバーラップさせて高速化を図ったりするような、高度なネットワークアプリケーションを実装しやすくなります。RPCライブラリと言うよりも、シンプルな非同期メッセージングライブラリと言えます。ここではMessagePack-RPCのプロトコル仕様と、以上の特徴を活かせるような実装例も合わせて紹介します。0.1MessagePack-RPCのプロトコルは、メソッドの呼び出すRequestメッセージと、それに対するResponseメッセージからなります。Requestに対しては、Responseを返す必要があります。これらに加えて、Notificationメッセージがあります。NotificationはRequestと似ていますが、Responseを返す必要がありません。:type必ず整数0です。このメッセージがRequestであることを示します。この手続き呼び出しに対応するResponseメッセージにも同じ値を入れます。符号無し整数か、文字列です。paramsメソッドの引数を示す配列です。配列の要素は、任意のオブジェクトです。type必ず整数1です。このがResponseメッセージであることを示します。msgidこのResponseが対応しているRequestのシーケンス番号です。以外の任意のオブジェクトresultメソッドの返り値を示す、任意のオブジェクトです。なくても良いでしょう。:type必ず整数2です。このメッセージがNotificationメッセージであることを示します。符号無し整数か、文字列です。paramsメソッドの引数を示す配列です。配列の要素は、任意のオブジェクトです。Responseの送信順序Requestを受け取ったサーバーは、必ずしもRequestを受け取った順番通りにResponseを返す必要はありません。複数のRequestを受け取ったら、順不同で返すことができます。言い換えれば、Requestを送ったあと、Responseを待たずに立て続けに別のRequestを送れるようにクライアントを実装した(=Requestのパイプライン化に対応した)場合は、どちらのRequestに対するResponseが先に返ってきても正常に動作するようにする必要があります。言っても、そもそもクライアントはRequestのパイプライン化に対応しなくても問題ありません。昇格することにします。「このライブラリはMessagePack-RPC/0.2に対応しているよ」と書いてあると、「これを使えば非同期呼び出しができるんだなー」と目に見えて分かって便利かな、と思いました。真面目にRPCライブラリを実装しようと思うと、タイムアウト、コネクションプーリング、非同期などなど、考えるべき難しい問題がたくさんあって大変です。そこでここでは、イベント駆動型のアーキテクチャでMessagePack-RPCを実装する例について少し紹介します。イベントループイベント駆動型のアーキテクチャでは、プログラムはイベントループを中心にして駆動します。サーバの実装では、RPCライブラリがプログラム全体の駆動方法を決めることになるので、イベントループを直接叩いてタイマーなどイベントハンドラを追加できるように、機能は豊富にあったほうが便利そうです。最近は各言語向けに高速なイベントループの実装があるようので、それを使うのが良さそうです:Ruby現状の実装では、Revを使っています。RevはlibevのRubyバインディングです。ベンチマークテストによれば、libeventよりも高速に動作するケースがあるようです。PythonTwistedやTornadoが使えそうです。MINAあたりが使えそうです。使って実装を進めています。イベントハンドラ:MessagePackストリームイベントハンドラでは、ソケットからデータを読み込んで、オブジェクトを切り出していきます。MessagePackのストリームデシリアライザを使えば、簡単に実装できるハズです。静的型付け言語では、切り出されたオブジェクトを静的型に変換する必要があります。判定し、その後で後続の要素を静的型に変換すると良さそうです。Sessionコネクションプーリングや非同期呼び出しをサポートし、実装の詳細の隠蔽するために、セッションの概念を導入します。MessagePackストリームは、Sessionを1つ持ちます。Sessionは、0個か複数のMessagePackストリームを持ちます。(循環参照ができない言語では、こちらの参照は弱参照で。MessagePackストリームを閉じる=ソケットをcloseすると同時に、弱参照を削除する)Sessionは、メッセージのシーケンス番号(msgid)からFuture(後述)を引く対応表を持ちます。Sessionは、メソッドのディスパッチャ(nullable)を持ちます(サーバー用)。Sessionは、宛先アドレス(nullable)とイベントループを持ちます(クライアント用)。MessagePackストリームは、メッセージを受け取ったら、Sessionに渡します。Sessionは、MessagePackストリームからメッセージを受け取ったら:Responseなら対応表からFutureを取り出して、Futureに結果とエラーをセットする。Futureにコールバック関数がattachされていれば、それを呼び出すRequestかNotificationならメソッドと引数をディスパッチャに渡します。クライアントは、Sessionのメソッドを呼び出して要求を送信します。Sessionは、まず新しいmsgidとFutureを作成します。次に、msgidとFutureの対応を対応表に追加します。メッセージはMessagePackを使ってシリアライズします。シリアライズしたデータは、MessagePackストリームの内の1つを使って送信します。SessionがMessagePackストリームを1つも持っていなかったら、新しいコネクションを確立し、イベントループに登録します。ここで新しいコネクションを確立する操作は、イベントループを使って非同期になるように実装すると良さそうです。ここで新しいコネクションを確立するとき、宛先アドレスがnullだったら、例外を投げます。ClientClientは、Sessionを継承します。Sessionに加えて、後述のタイムアウト用の機能を実装します。SessionPoolSessionPoolは、宛先アドレスからSessionを引く、連想配列を持ちます。SessionPoolは、イベントループを持ちます。複数のコンテキストでSessionを使い回せるようになるので、コネクションのプーリングを実現できます。ServerServerは、セッションプールを継承します。新しいコネクションを受け付けたら、MessagePackストリームを作成して、イベントループに登録します。宛先アドレスがnullであるSessionを新しく作成し、そのSessionにMessagePackストリームを追加します。FutureFutureは、まだ結果が返ってきていないRPC要求を表します。メンバには、結果(nullable)、エラー(nullable)、イベントループ、コールバック関数(nullable)を持ちます。結果を待ち受けて同期するには、結果かエラーがセットされるまで、イベントループを回します。マルチスレッドで動作する場合は、相互排除が必要になります。タイムアウトについてFutureのメンバにカウンタを持たせます。ClientやSessionPoolは、コンストラクタでイベントループにタイマーを登録し、一定間隔ごとにすべてのFutureのカウンタを1ずつ減らしていきます。カウンタが0になったら、エラーに「タイムアウトエラー」をセットします。現状のRuby版の実装は、確定していない仕様も実装していたりして若干読み辛いですが、メッセージングライブラリを実装するときの参考になるかもしれません。00:00MessagePack-RPCのプロトコル仕様(ドラフト)と実装例Tweet広告を非表示にする関連記事2011-04-05第3回MessagePackハッカソン開催報告4月3日、MessagePackハッカソン第3回を開催しました。コードは、全体で約2万行です。リトライはいつ成功するかリトライと冪等性のデザインパターン月別アーカイブ検索はてなブログをはじめよう!viverさんは、はてなブログを使っています。ブログを報告するスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。