あっというまに0.8.3

Overlay Weaver 0.8.3 が出てる。
前回、Collective Routing が複数 ID に複数の値を put, get, remove する機能だと書いたが間違いだった。すまん。
正確な説明はメーリングリストを見ると判る。要するに同一の宛先データを別パケットにしないで一個にまとめる
機能のことらしい。
同一 ID への要求は同じノードをたどって到着することが多いという考えなのだろうか。

そのほかの修正は以下。

(1)DHT.put メソッドから TTL 指定が抜けた。
　結構大きな変更。Overlay Weaver を使ってプログラムを書いている人は自分の
ソースを再度チェックが必要。

(2)DHT.put, DHT.remove にオーバーロードメソッドが追加。1 つの ID に対して複数の値を受け付ける。
　正直、どういう時に 1 つの ID に複数の値が出てくるのかわからん。たとえば
とあるメンバーの名前と年齢をひとつのクラスに収めないで、String と Long で
別々に登録するとか？

(3)DHT.put, DHT.remove で同一 ID への複数メッセージを１つにたばねるようになった。
　これは Collective Routing に関係する修正と見た。受け取ったノードに任せない
で、送出元でまとめられるときはまとめちゃうほうが効率いいしね。


ところで Collective だと集めるとか集合って意味で、メッセージをまとめる感が
薄い感じ。代わりに Bundle Forward とかどうだろう。

0.8.2が出てる

Overlay Weaver 0.8.2 が出てる。

注目は２つ。
・recursive ルーティングのバグをいくつか修正した。
　Iterative があんまり好きになれない私にとってはうれしい修正。
・collective routing を実装した。
　第３のルーティングが登場、かと思ったんですが違うみたいです。たぶん。
　ow.dht.DHT インターフェイスの get, put, remove に複数 ID 用のメソッドが
　追加されました。たぶんこれが collective routing かと思われます。
　ソースを全文検索する限り、collective が出てくるのはここだけなので。
　CollectiveRoutingDriver なんてクラスが追加されたのかと思ってワクワク
　したんですけどね…。

ver0.8.1が出てますね

http://sourceforge.net/project/showfiles.php?group_id=156419&package_id=174533
新しいバージョンが出てます。
メーリングリストには何も投稿されなかったので見逃してしまった。

メッセージを受け取るには

忘れてた。メッセージを受け取る方法を書いておかねば。
ow.messaging.MessageHandler インターフェイスを実装したクラスをメッセージの
タグ番号と共に登録することで、メッセージ受信時にコールバックされる。
ow.dht.DHT のインスタンスを dht、MessageHandler のインスタンスを handler とすると、
　dht.getRoutingService().addMessageHandler(tag, handler);
で登録できる。もちろん tag は 0 以上 255 以下だ。

実際には、Map<Integer, List<MessageHandler>> 型の ow.routing.impl.AbstractRoutingDriver.handlerTable 変数に
格納される。つまり１つのタグ値に対して複数の MessageHandler を登録できる。
ただ、ここの実装がちょっと微妙。登録された MessageHandler は、
ow.routing.impl.AbstractRoutingDriver.RoutingDrvMessageHandler.process
メソッドで振り分け処理がなされて呼び出されるんだけど、複数の MessageHandler
がある場合、最後に登録されたハンドラの process メソッドの戻り値を
メッセージの送信元に返すことになる。これだと、最初に登録されたハンドラの
戻り値を返したい、後から登録したハンドラはイベントドリブンのトリガに使いたい
だけ、って時に困る。最初の戻り値を潰す事になるからだ。
たとえば dht.put されたこと（つまり自ノードに put データが来たこと）を
知りたくて ow.messaging.Tag.PUT.getNumber() に addMessageHandler した場合、
本来のハンドラ ow.dht.impl.StandardDHTImpl.PutMessageHandler.process は
DHTMessageFactory.getDHTReplyMessage() の戻り値を put の返答として返すん
だけど、後から登録したハンドラの戻り値に潰されてしまう。
なので、できれば MessageHandler.process の戻り値が null の場合は前のハンドラ
の戻り値を使用する（潰さない）ようにしてほしいなぁ、と。
その旨を Sourceforge の Feature Requests にこっそり匿名で書いといたんだけど、
返事なし。仕方ないので、メッセージ到着を検知したい場合は自分でソースを書き換えよう。

メッセージを送るには その６ TCPの受信でスレッドいっぱい

TCP のコネクションプールで補足。プールする接続数は ow.messaging.tcp.TCPMessagingConfiguration
で決まるんだけど、その数は動的に変更可能。TCPMessagingConfiguration.setConnectionPoolSize(int size)
で設定する。

で、受信なんだけど ow.messaging.tcp.TCPMessageReceiver で行ってる。具体的には
start() メソッドでスレッドを起動し、run() メソッドで接続の accept を待つ。
accept した接続はどうするかというと、内部クラスの TCPMessageHandler クラスの
インスタンスを生成、そのインスタンスに渡してそれをスレッド起動（！）する。
ということは接続されればされるほどスレッドがどんどん増えると…。
これは看過できない。しかもこちらから積極的に接続を閉じることはしない。
うーん…。

データの受信、これもちょっと…。具体的には ow.messaging.Message.decode(ByteChannel in)
にソケットチャンネルを渡して終わり。そのメソッド内では必要分の情報を受信する
まで channel からデータを読み出す。ブロッキングソケットだから、データが受信
できるまでそのスレッドは止まり続けるわけだ。だからこその別スレッド起動なんだけど。

まあ実装は正しいんだけど、スレッドが大量に起動する可能性があるのはいただけないなぁ。
要はメッセージ境界がハッキリしないからブロッキングソケットで必要分が届くまで
受信し続けるって実装なんじゃないかな、想像だけど。
だったら送信する際にバイト配列にしたメッセージ情報のバイト数を相手に送信して、
メッセージ境界をはっきりさせればいいのでは？１スレッドで複数の受信ソケットを
select なりで扱い、データを溜め込む。メッセージ境界まで読み込んだら decode
する。これでいいと思うのだが。

とまあ６回もメッセージ送受信で書いた結論としては、標準実装の TCP で頑張ってみようということ。
次からは少しずつ実装を行っていけたらいいな。では次の土曜日にまた。

メッセージを送るには その５ 標準実装のTCP

OW 標準実装の TCP はちょっと不思議な感じ。まあ調べればちゃんと理由はあるんだけど。
トランスポート層として TCP を使用し、１つのノードと送受信する場合を考える。
こちらからそのノードへ送信しようとすると、コネクションが張られる。これは ow.messaging.tcp.ConnectionPool でプールされるので再利用可能。
もしプール数が ow.messaging.tcp.TCPMessagingConfiguration.getConnectionPoolSize()
を超えると、ランダムでプールされている中から切断するコネクションが選ばれる。
ランダムってのがちょっと悩むところ。使用しなくなって一番時間が経過した
コネクションの方がよさそうなんだけどな。まあそれは大きな問題ではなかろう。
では受信。あるノードに送信したとする。そうするとそのノードへのコネクション
はすでにプールされている。その状態で相手ノードがメッセージを送りたいとする。
すでに張られているコネクションを利用するのか？

利用しないんだな。

実は OW の TCP 実装は一方通行。全二重（もう古い言い方なんだろうか。私は
世代的には半二重を知ってる世代なもんで）のはずなのになんともったいない。
と貧乏根性を最大発揮してもしかたない。なぜこうなっているのか？
たぶん接続してきた相手が、いったい誰なのかわからないからだろう。
２つのノードが同一の IPv4 アドレスを持つこと、つまりあるマシンが２つの
P2P を起動することは十分に可能だ（ポート変えりゃ簡単だ）。そうなると、
接続してきた元の IPv4 アドレスからだけでは相手を判断できない。
もちろん接続してきたそのコネクションのポート番号を調べても無駄だ。
そのポート番号はランダムに割り当てられたものなのだから。
そうなると、接続された側は相手が誰だかわからない、だからそのコネクション
へは送信しない（できない）わけだ。たとえメッセージをそこから受信し、
そのメッセージの src を取り出しても相手のアドレスとは限らない。
延々とルーティングでたらいまわしにされたメッセージかもしれないからだ（
以前、メッセージの src を詐称できちゃうんじゃ、って書いてたのは間違いで、
実際には送信元を示すために必要なのだ。メッセージは直接相手に届く場合だけじゃ
なく、転々と転送されて届けられることもあるから、送信元情報の格納が必須）。

送信は Encode された byte 列を書き込むだけ。ここはぜひとも channel で並列処理
したいところだ。で、受信なんだけどここはちょっと気にしておきたいってことで
明日に続く。たぶん明日だろう、きっと。

メッセージを送るには その４

Message の tag が byte 型の範囲内って書いたんだけど、ちょっと違ったので
補足。
byte 型だと -128～127 だけど、tag は 0～255 でした。Message.decode の実装
を見ればわかるけど、
　int tag = buf.get() & 0xff;
ってことで、0～255。
初めて OW でアプリ作るとき、意外にここにはまるんだよね。tag が int だから
適当に 65536 とか指定しちゃうと、ぜんぜんメッセージとどかないんだよね。

ちなみに OW の内部で使用しているタグは、
　ow.messaging.Tag
に public static final で定義されてるので参照しよう。
60個以上あるので、だいたい 0～70 あたりは tag の値として使用できない
ってことになる。自前アプリのメッセージタグは 255 から使うのが吉。

今日は平日なのでこんなところ。次回は TCP の実装の気になるところと
システムメッセージをフックしようとしても現在の実装のままだと上手く
いかないよ、ってところを書くと思うので以下続く。

メッセージを送るには その３

OW はトランスポート層を取り替えることができる。
トランスポート層の実装は ow.messaging.MessageSender とか
ow.messaging.MessageReceiver を実装してるので、これらを JavaDoc で見れば
どういう通信手段があるかわかる。
現在のところ、TCP, UDP, シミュレーション, 分散シミュレーションの４種がある。

シミュレーション用実装は、メッセージを送る場合、Sender が Receiver に
直接データを渡しちゃう実装。なのでパケットロスとかはありえない。
分散シミュレーションは、自分に送る場合はシミュレーションと同様に直接
渡しちゃうが、他ノードへは UDP もしくは TCP で送るらしい。なんか入り組んでて
完全には判らなかった。というか、ポート番号変えて自ノードで複数 OW を起動
すりゃいいんじゃ…。

実際にインターネットで運用する気なら、最終的には TCP もしくは UDP となる
わけで、あとはどう使い分けるか。
と以前は思ってたのだが、どうにも UDP じゃ無理っぽいと思い始めている。
UDP でやるには結局のところ、パケットロスの補償が要るわけだが、だったら
TCP でやりゃいいじゃん、車輪再発明し過ぎという意見がある。
http://www.kt.rim.or.jp/~ksk/sock-faq/indexj.html
ここの５章 UDP の項目を見れば UDP でっていう気力が萎えてくる。
（ちなみに同一HPの http://www.kt.rim.or.jp/~ksk/wskfaq-ja/index.html
もオススメ。このHPの FAQ は、TCP/IP を扱う技術者は必読だと思う）
さらに UDP は、どのバイト数のパケットなら相手に届くかわからない。経路の途中
で断片化されるとパケットロスの危険が高まる。TCP なら Path MTU Discovery で
できる限りの効率で通信ができる。
http://www.microsoft.com/japan/technet/community/columns/cableguy/cg0704.mspx
http://wakita.no-ip.com/server/ProblemMTU.html

実際、Infoseek の無料 HP に申し込んで Perl スクリプトを動かし、自宅に向けて
パケットを送ってみたところ、1426byte のパケットなら届いたが 1427byte は
届かなかった。自宅は Flet's ADSL なので、MTU 1454byte が効いているものと
思われる。IP パケットのヘッダ 20byte と UDP のヘッダ 8byte のあわせて
28byte。1454-28=1426byte となってるわけだ。
じゃあ全員が全員このバイト数で良いかというと、そうもいかないだろう。経路
の途中でもっと小さい UDP パケットしか受け付けない経路があるかもしれない。
どこでも質の高い経路、ってわけじゃないし。
そう考えると、UDP でやる利点ってパケット単位で受け取れる（ストリーム志向
じゃない）ってのが最大の利点なのかなぁ、と。

OW では１メッセージを 1 UDP パケットで送るのが標準実装。ってことは、ちょっと
でもデカい情報は送れないってことになる。
とまあこれで UDP じゃダメだなぁってこと、決定。
小さい情報しか送らないとしても、DHT で get しようとしたら同一キーに複数の情報が存在しましたってなったらそれで 1426 byte 超えちゃうこともありうる。

しかし同一 LAN 内に限るのであれば、たぶん問題なく使えるんだろうな。以前、
仕事で UDP でメッセージをやり取りしたんだけど、デカいパケットでも同一 LAN 内
ならぜんぜん問題なかったしね。

UDPについてはこんなところか。TCPについては以下続く。たぶん今週末あたりになるけど。

メッセージを送るには その２

実際に送信する方法は、こんな感じ。
まず大本の ow.dht.DHT クラスのインスタンスを dht とすると、
　dht.getRoutingService().getMessageSender().send(dest, msg);
となる。ちょっと長い。

dest は ow.messaging.MessagingAddress インターフェイス。これは
トランスポート層を取り替えられるため。SocketAddress にしちゃうと
インターネット専用になっちゃうから。ま、実質はインターネットが大半だろう
けど。一応、シミュレーション用のアドレス EmuMessagingAddress が用意されて
いるのでインターネット専用ってわけじゃない。
このアドレス、どう作るかというと、ow.messaging.MessagingProvider.getMessagingAddress(java.lang.String hostAndPort) を使う。
　dht.getRoutingService().getMessagingProvider().getMessagingAddress("www.foobar.com:8888");
こんな感じ。
あとは送信されてきたメッセージから取り出す方法もある。
　ow.messaging.Message.getSource().getAddress()

ちなみにこういうのを調べるにも JavaDoc が大活躍。MessagingAddress のページを開いて、
画面上部の「使用」ってリンクを開けば MessagingAddress を使っているメソッド
が一覧で見れる。便利。

これでメッセージを送れるのだが、問題はトランスポート層。どんな方法で送るか、
それが問題で以下続く。

メッセージを送るには その１

この Blog は基本、備忘録っつーか調べたことを書き散らかす目的なんで、話が飛びまくるけどご容赦。

ネットワークを使ったアプリは、他ノードと通信しないと当然動けない。
DHT 自体は「値を put して、get して」っていう Map が主体で、ぜんぜんイベントドリブンじゃない。検知できなくもないけどその話はまた今度。

OW ではメッセージ ow.messaging.Message が担当。
　Message(IDAddressPair src, int tag, java.io.Serializable... contents)
これ見ると、src は詐称できるんかいとか思うけど、まあそこは置いといて。
tag はこのメッセージの種別。contents がメッセージの本体データ。

ここで注意点。tag は int って書いてあるけど、実は byte の範囲しか対応してない。これは ow.messaging.Message.encode(Message msg) を見れば判る。
　　buf.put((byte)msg.tag);
　　buf.putInt(srcAndContents.length);
　　buf.put((byte)msg.contents.length);
　　buf.put(srcAndContents);
というわけで、msg.tag は byte にキャストされるのでありました。

実際にどういう形式で送られるかというと、上記のようにヘッダ情報＋シリアライズ。
シリアライズは JavaDoc 見ると「効率良くない」みたいな書かれ方してるけど、
車輪の再発明するよりかは良いんじゃないか。標準機能だし、バージョンアップ時
の対応とかあるみたいだし。
ちなみに GZip 圧縮してみたけどあんまり縮まなかったよ、Message クラスの encode 後のデータ。無理にあがいても無駄っぽい。

Message.encode したデータ、実データ以外のシリアライズ用データとかその他もろもろ、どれくらい必要なのかと計ってみたら 408 バイトだった。
これはちょっと多いよな。次に書く予定の送信方法にも関わってくるんだけど、以下続く。