2006-02-28

昨夜作ってみた"org.jgroups.stack.Protocol"実装クラスをもう1つ増やしてプロトコルスタックとしての動きをログから調べてみた．

public static final String STACK2 =
    "UDP(mcast_addr=228.1.2.3;mcast_port=45566;ip_ttl=32):" +
    "MIR:" +
    "MIR2";

最上位層にMIR2，中間層にMIR，最下位層にUDPが置かれるという設定．

ほんとはJGroupsの全Protocol実装クラスにjavaassistとかでログ出力させるように改変とかしちゃうとすこぶる格好良いのだろうけど，恥ずかしながらjavaassistは使ったことが無いので今回はパス．MIR2は昨日のMIRと同じく，各メソッドを呼ばれると単に"プロトコル名.メソッド名（引数値）戻り値"をstdoutに出力するだけ．

呼び出し元コードはこれ．昨日はconnect()してすぐにdisconnect()だったけど，今回からはsend(Message)でメッセージ送信もやる．

public static void main(String[] args) throws Exception {
    System.out.println("\n----- start new JChannel() -----\n");
    JChannel channel = new JChannel(STACK2);
    System.out.println("\n----- start channel.connect() -----\n");
    channel.connect("demo-group");
    
    Message msg = new Message(null, null,6 "Hello World");
    System.out.println("\n----- start channel.send(" + msg + ") -----\n");
    channel.send(msg);
    
    System.out.println("\n----- start channel.disconnect() -----\n");
    channel.disconnect();
    System.out.println("\n----- start channel.close() -----\n");
    channel.close();
    System.out.println("\n----- main method end -----\n");
}

これでJGroupsプロトコルスタックの動きがだいたいわかるでしょう．シーケンス図の1つでも書けるってもんだ．

肝心の実行結果．上位層であるMIR2を青色に、中間層であるMIRを赤色にしてみた．下位層のUDPについては脳内補完してみたのを括弧付きで書いてみた．

 ----- start new JChannel() -----
（たぶんここにUDPの処理，下から上へ．このコンストラクタで下から上なのは設定情報の
　書き順が下のプロトコルから順番にであるから．これは上位の層は下位の層に依存すると
　いうのを考慮してのことでしょう．）
MIR.setProtocolStack(org.jgroups.stack.ProtocolStack@32fb4f)
MIR.setProperties({}) returns true
MIR.setPropertiesInternal({}) returns true
MIR.init()
MIR2.setProtocolStack(org.jgroups.stack.ProtocolStack@32fb4f)
MIR2.setProperties({}) returns true
MIR2.setPropertiesInternal({}) returns true
MIR2.init()
（最上位へ到達）
（たぶんここにUDPの処理，下から上へ）
MIR.requiredUpServices() returns null
MIR.requiredDownServices() returns null
MIR.providedUpServices() returns null
MIR.providedDownServices() returns null
MIR2.requiredUpServices() returns null
MIR2.requiredDownServices() returns null
MIR2.providedUpServices() returns null
MIR2.providedDownServices() returns null
（最上位へ到達）
（たぶんここにUDPの処理，下から上へ）
MIR.setDownProtocol(Protocol UDP(local address: null))
MIR.setUpProtocol(org.jgroups.protocols.MIR2@1608e05)
MIR2.setDownProtocol(org.jgroups.protocols.MIR@bf32c)
MIR2.setUpProtocol(org.jgroups.stack.ProtocolStack@32fb4f)
（最上位へ到達）
（JChannelが下位層へ呼び出し）
MIR2.getDownProtocol() returns org.jgroups.protocols.MIR@bf32c
MIR.getDownProtocol() returns Protocol UDP(local address: null)
（たぶんここにUDPの呼び出し，下位層へ到達）
（たぶんJChannelの呼び出しによりここにUDPの処理，下から上へ）
MIR.startDownHandler()
MIR.startUpHandler()
MIR.getUpProtocol() returns org.jgroups.protocols.MIR2@1608e05
MIR2.startDownHandler()
MIR2.startUpHandler()
MIR2.getUpProtocol() returns org.jgroups.stack.ProtocolStack@32fb4f
（最上位へ到達）

 ----- start channel.connect() -----
（JChannelが下へイベント発信）
MIR2.(protected) receiveDownEvent(Event[type=START, arg=null])
MIR2.(protected) handleSpecialDownEvent(Event[type=START, arg=null])
MIR2.start()
MIR2.down(Event[type=START, arg=null])
MIR2.passDown(Event[type=START, arg=null])
MIR.(protected) receiveDownEvent(Event[type=START, arg=null])
MIR.(protected) handleSpecialDownEvent(Event[type=START, arg=null])
MIR.start()
MIR.down(Event[type=START, arg=null])
MIR.passDown(Event[type=START, arg=null])
（たぶんここにUDPの処理，最下位層へ到達）
（今度はUDPが下から上へイベントを発信）
MIR.(protected) receiveUpEvent(Event[type=SET_LOCAL_ADDRESS, arg=192.168.1.100:2620])
MIR.up(Event[type=SET_LOCAL_ADDRESS, arg=192.168.1.100:2620])
MIR.passUp(Event[type=SET_LOCAL_ADDRESS, arg=192.168.1.100:2620])
MIR2.(protected) receiveUpEvent(Event[type=SET_LOCAL_ADDRESS, arg=192.168.1.100:2620])
MIR2.up(Event[type=SET_LOCAL_ADDRESS, arg=192.168.1.100:2620])
MIR2.passUp(Event[type=SET_LOCAL_ADDRESS, arg=192.168.1.100:2620])
（最上位へイベントが到達）
（さらにUDPの処理，下から上へイベントを発信）
MIR.(protected) receiveUpEvent(Event[type=START_OK, arg=true])
MIR.up(Event[type=START_OK, arg=true])
MIR.passUp(Event[type=START_OK, arg=true])
MIR2.(protected) receiveUpEvent(Event[type=START_OK, arg=true])
MIR2.up(Event[type=START_OK, arg=true])
MIR2.passUp(Event[type=START_OK, arg=true])
（最上位層へイベントが到達）
（たぶんここでJChannnel側が上から下へ呼び出し開始）
MIR2.(protected) receiveDownEvent(Event[type=CONNECT, arg=demo-group])
MIR2.down(Event[type=CONNECT, arg=demo-group])
MIR2.passDown(Event[type=CONNECT, arg=demo-group])
MIR.(protected) receiveDownEvent(Event[type=CONNECT, arg=demo-group])
MIR.down(Event[type=CONNECT, arg=demo-group])
MIR.passDown(Event[type=CONNECT, arg=demo-group])
（たぶんここにUDPの処理，最下位層へ到達）
（UDPが上へイベントを投げる，下から上へ）
MIR.(protected) receiveUpEvent(Event[type=CONNECT_OK, arg=null])
MIR.up(Event[type=CONNECT_OK, arg=null])
MIR.passUp(Event[type=CONNECT_OK, arg=null])
MIR2.(protected) receiveUpEvent(Event[type=CONNECT_OK, arg=null])
MIR2.up(Event[type=CONNECT_OK, arg=null])
MIR2.passUp(Event[type=CONNECT_OK, arg=null])
（最上位にイベントが到達）

 ----- start channel.send([dst: , src: , size = 18 bytes]) -----
（JChannelが下位層へ向けてイベントを発信）
MIR2.(protected) receiveDownEvent(Event[type=MSG, arg=[dst: , src: , size = 18 bytes]])
MIR2.down(Event[type=MSG, arg=[dst: , src: , size = 18 bytes]])
MIR2.passDown(Event[type=MSG, arg=[dst: , src: , size = 18 bytes]])
MIR.(protected) receiveDownEvent(Event[type=MSG, arg=[dst: , src: , size = 18 bytes]])
MIR.down(Event[type=MSG, arg=[dst: , src: , size = 18 bytes]])
MIR.passDown(Event[type=MSG, arg=[dst: , src: , size = 18 bytes]])
（たぶんここにUDPの処理，最下位層へ到達）

 ----- start channel.disconnect() -----
（JChannelが下位層へ向けてイベントを発信）
MIR2.(protected) receiveDownEvent(Event[type=DISCONNECT, arg=192.168.1.100:2620])
MIR2.down(Event[type=DISCONNECT, arg=192.168.1.100:2620])
MIR2.passDown(Event[type=DISCONNECT, arg=192.168.1.100:2620])
MIR.(protected) receiveDownEvent(Event[type=DISCONNECT, arg=192.168.1.100:2620])
MIR.down(Event[type=DISCONNECT, arg=192.168.1.100:2620])
MIR.passDown(Event[type=DISCONNECT, arg=192.168.1.100:2620])
（たぶんここにUDPの処理，イベントが最下位層へ到達）
（たぶんUDPがイベントを上位層へ発信）
MIR.(protected) receiveUpEvent(Event[type=MSG, arg=[dst: 228.1.2.3:45566, src: 192.168.1.100:2620 (1 headers), size = 18 bytes]])
MIR.up(Event[type=MSG, arg=[dst: 228.1.2.3:45566, src: 192.168.1.100:2620 (1 headers), size = 18 bytes]])
MIR.passUp(Event[type=MSG, arg=[dst: 228.1.2.3:45566, src: 192.168.1.100:2620 (1 headers), size = 18 bytes]])
MIR2.(protected) receiveUpEvent(Event[type=MSG, arg=[dst: 228.1.2.3:45566, src: 192.168.1.100:2620 (1 headers), size = 18 bytes]])
MIR2.up(Event[type=MSG, arg=[dst: 228.1.2.3:45566, src: 192.168.1.100:2620 (1 headers), size = 18 bytes]])
MIR2.passUp(Event[type=MSG, arg=[dst: 228.1.2.3:45566, src: 192.168.1.100:2620 (1 headers), size = 18 bytes]])
（最上位層へイベントが到達）
（さらにUDPが何かを処理，イベントを上位層へ向かって発信）
MIR.(protected) receiveUpEvent(Event[type=DISCONNECT_OK, arg=null])
MIR.up(Event[type=DISCONNECT_OK, arg=null])
MIR.passUp(Event[type=DISCONNECT_OK, arg=null])
MIR2.(protected) receiveUpEvent(Event[type=DISCONNECT_OK, arg=null])
MIR2.up(Event[type=DISCONNECT_OK, arg=null])
MIR2.passUp(Event[type=DISCONNECT_OK, arg=null])
（最上位層へイベントが到達）
（JChannelが下位層へ向かってイベントを発信）
MIR2.(protected) receiveDownEvent(Event[type=STOP_QUEUEING, arg=null])
MIR2.down(Event[type=STOP_QUEUEING, arg=null])
MIR2.passDown(Event[type=STOP_QUEUEING, arg=null])
MIR.(protected) receiveDownEvent(Event[type=STOP_QUEUEING, arg=null])
MIR2.(protected) receiveDownEvent(Event[type=STOP, arg=null])
MIR.down(Event[type=STOP_QUEUEING, arg=null])
MIR.passDown(Event[type=STOP_QUEUEING, arg=null])
（ここでUDPの処理が入ると思われるが、ここの直前直後だけ動きが不規則、いまのところ謎）
（なぜMIRがSTOP_QUEUEINGによる処理を行っている合間にMIR2がSTOPをやっているのとか不明，おそらくソース参照）
（たぶんまたJChannelがイベントを発信，でもMIR2だけは既にreceiveDownEvent呼び出し済み，謎）
MIR2.(protected) handleSpecialDownEvent(Event[type=STOP, arg=null])
MIR2.stop()
MIR2.down(Event[type=STOP, arg=null])
MIR2.passDown(Event[type=STOP, arg=null])
MIR.(protected) receiveDownEvent(Event[type=STOP, arg=null])
MIR.(protected) handleSpecialDownEvent(Event[type=STOP, arg=null])
MIR.stop()
MIR.down(Event[type=STOP, arg=null])
MIR.passDown(Event[type=STOP, arg=null])
（UDPまでイベントが到達，UDPの処理）
（UDPがイベントを上位層へ向かって返す）
MIR.(protected) receiveUpEvent(Event[type=STOP_OK, arg=true])
MIR.up(Event[type=STOP_OK, arg=true])
MIR.passUp(Event[type=STOP_OK, arg=true])
MIR2.(protected) receiveUpEvent(Event[type=STOP_OK, arg=true])
MIR2.up(Event[type=STOP_OK, arg=true])
MIR2.passUp(Event[type=STOP_OK, arg=true])
（最上位層にイベントが到達）

 ----- start channel.close() -----
（JChannelが下位層へ向かって呼び出し）
MIR2.stopInternal()
MIR2.destroy()
MIR2.getDownProtocol() returns org.jgroups.protocols.MIR@bf32c
MIR.stopInternal()
MIR.destroy()
MIR.getDownProtocol() returns Protocol UDP(local address: 192.168.1.100:2620)
（ここにUDPの処理）
（最下位層まで呼び出しが伝わって終わり）

 ----- main method end -----

んー，どうでしょう．脳内補完をしているうちにかなり動きのイメージが掴めてきましたが，こうしてみるとJGroupsプロトコルスタックが何をやっているのか何となくわかってきますね．とりあえず現段階ではイメージがつかめればOKだと思います．

要するにコマンドパターンを使いつつイベントを下に投げたり上に投げたりってことです．各Protocolは共通の親クラスを持っていて，透過的に処理されいるというのも重要．

これでかなりJGroupsというのは汎用プロトコルスタックフレームワーク＋各プロトコル実装（コンポーネント）という全体像が見えたと思う．JGroupsの特性は？といった場合，それは全て各コンポーネント次第であり，俗に「JGroupsプロトコルスタック」として紹介されているものは，典型的なコンポーネント組み合わせパターンの1つだったのだ！ｗ　（注：フレームワークといっても、Protocol実装クラスに現時点ではパッケージ名の自由がないためフレームワークとしての提供は全く意図していないと思われます）

さてこのイメージを元に再度ソースコードを見れば，今度はかなり読めるようになるかと思います．

続きはまた後ほど．

mir the developer

プロトコルスタックの動きとその正体（長文注意）

英文ドキュメントの翻訳