(57)【特許請求の範囲】
【請求項１】
それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置され、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を所定期間毎に停止させる停止部と、
前記停止部によりデータの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得する取得部と、
前記取得部により取得された複数のノードのルーティングテーブルのバージョンを比較する比較部と、
前記比較部による比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する更新部と、
を有することを特徴とするサーバ装置。 
【請求項２】
それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置され、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードの間でクエリが移動した場合、クエリが移動してから所定期間が経過したときに、前記複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を停止させる停止部と、
前記停止部によりデータの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得する取得部と、
前記取得部により取得された複数のノードのルーティングテーブルのバージョンを比較する比較部と、
前記比較部による比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する更新部と、
を有することを特徴とするサーバ装置。
【請求項３】
クエリ毎に前記所定期間を記憶する記憶部をさらに有し、
前記停止部は、クエリが移動してから当該クエリに対応する所定期間が経過したときにデータの送信を停止させ、
前記取得部は、前記複数のノードから、移動したクエリに関するルーティングテーブルのバージョンを取得する
ことを特徴とする請求項２に記載のサーバ装置。 
【請求項４】
前記複数のクエリを複数のグループに分けたグループ毎に前記所定期間を記憶する記憶部をさらに有し、
前記停止部は、クエリが移動してから当該クエリが属するグループに対応する所定期間が経過したときにデータの送信を停止させ、
前記取得部は、前記複数のノードから、移動したクエリが属するグループの各クエリに関するルーティングテーブルのバージョンを取得する
ことを特徴とする請求項２に記載のサーバ装置。 
【請求項５】
前記所定期間が、前記複数のノードの間でのデータが転送される最大期間よりも短い場合、前記停止部、前記取得部、前記比較部、前記更新部の処理を実行する実行部をさらに有することを特徴とする請求項３または４に記載のサーバ装置。 
【請求項６】
前記複数のノードから一定期間毎に受信する、当該一定期間の間に最初に他のノードへデータを転送した日時を示す日時情報に基づき、当該一定期間で最も過去に他のノードへデータを転送した日時を特定する特定部をさらに有し、
前記ルーティングテーブルは、前記バージョンとして、更新日時を記憶し、
前記停止部は、前記特定部により特定された日時が前記ルーティングテーブルにバージョンとして記憶された更新日時以降の日時である場合に、複数のノードに対するデータの送信を停止させる
ことを特徴とする請求項３〜５の何れか１つに記載のサーバ装置。 
【請求項７】
コンピュータに、
それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置され、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を所定期間毎に停止させ、
データの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得し、
取得された複数のノードのルーティングテーブルのバージョンを比較し、
比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する
処理を実行させることを特徴とするデータ順序保証プログラム。 
【請求項８】
コンピュータが、
それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置され、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を所定期間毎に停止させ、
データの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得し、
取得された複数のノードのルーティングテーブルのバージョンを比較し、
比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する
処理を実行することを特徴とするデータ順序保証方法。 
【請求項９】
処理を実行する複数の処理要求文が配置され、当該処理要求文が処理対象とするデータ毎に当該処理要求文の配置先および処理要求文の移動に伴って更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を停止させる停止部と、
前記停止部によりデータの送信が停止した場合に、前記複数のノードから前記ルーティングテーブルのバージョンを取得する取得部と、
前記取得部により取得した複数のノードの前記ルーティングテーブルのバージョンを比較する比較部と、
前記比較部による比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する更新部と、
を有することを特徴とするサーバ装置。
【請求項１０】
コンピュータに、
それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置され、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードの間でクエリが移動した場合、クエリが移動してから所定期間が経過したときに、前記複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を停止させ、
データの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得し、
取得された複数のノードのルーティングテーブルのバージョンを比較し、
比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する
処理を実行させることを特徴とするデータ順序保証プログラム。 
【請求項１１】
コンピュータが、
それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置され、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する複数のノードの間でクエリが移動した場合、クエリが移動してから所定期間が経過したときに、前記複数のノードに対してデータを送信する送信元へデータの送信停止要求を送信してデータの送信を停止させ、
データの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得し、
取得された複数のノードのルーティングテーブルのバージョンを比較し、
比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する
処理を実行することを特徴とするデータ順序保証方法。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、サーバ装置、データ順序保証プログラム、およびデータ順序保証方法に関する。
【背景技術】
【０００２】
近年、様々な対象から刻々と収集される多数のデータを並列に処理する技術として複合イベント処理（ＣＥＰ（Complex  Event  Processing））が知られている。この複合イベント処理では、受信したデータに対してイベントを検出し、検出したイベントに関する処理が実行される。なお、複合イベント処理は、ＥＳＰ（Event  Stream  Processing）と呼ぶ場合もあるが、ここでは、ＥＳＰも包含してＣＥＰと総称する。
【０００３】
複合イベント処理を行うＣＥＰシステムでは、一時的に大量の受信データを処理する場合があり、処理負荷が高くなり、処理性能が低下する場合がある。また、ＣＥＰシステムは、データをリアルタイムで処理するシステムであるために常時稼働が期待され、停止が許容されない場合がある。
【０００４】
そこで、ＣＥＰシステムでは、クラウド等に用いられる柔軟な資源割り当てが可能なシステム技術を用いて、処理負荷の変動に応じて複数のサーバや仮想マシン（ＶＭ：Virtual  Machine）に処理の分散が行われる。例えば、処理負荷が高いサーバや仮想マシンに配置されたクエリと呼ばれる処理要求文やクエリに付随するデータを処理要素として他のサーバや仮想マシンに移動させて処理の分散を行う。
【０００５】
しかし、ＣＥＰシステムでは、処理要素の移動が発生した場合、データの順序が入れ替わる可能性がある。例えば、データの送信元サーバをＡとし、処理要素の移動元サーバをＢとし、処理要素の移動先サーバをＣとする。送信元サーバＡがデータを移動元サーバＢへ送信し、処理要素が移動元サーバＢから移動先サーバＣへ移動する場合を想定する。処理要素が移動元サーバＢから移動先サーバＣへ移動するタイミングで送信元サーバＡから移動元サーバＢへ転送されたデータは、移動元サーバＢに到着する。移動元サーバＢは、処理要素が移動先サーバＣへ移動したため、到着したデータを移動先サーバＣへ転送する。移動先サーバＣは、処理要素の移動が完了すると送信元サーバＡへ処理要素の移動完了の通知を行う。送信元サーバＡは、移動完了の通知以降、データを移動先サーバＣへ直接送信する。このような場合、送信元サーバＡから移動先サーバＣへ直接送信されたデータが、送信元サーバＡから移動元サーバＢを介して移動先サーバＣへ転送されたデータよりも先に移動先サーバＣに到着する可能性がある。
【０００６】
このように、データの順序が入れ替わると、データの順序により示される意味が変わる場合がある。例えば、家屋の玄関ドアの鍵の解錠を検知するドアセンサと、室内の人を検知する人感センサを設け、ドアセンサと人感センサから送信されるデータを監視する防犯システムをＣＥＰシステムにより実現した場合を考える。この場合、ドアセンサによる鍵の開閉の検知と人感センサによる室内での動作の検知のデータの順序が入れ替わると意味が変わる。例えば、鍵の解錠が検知された後に、室内で物体が検知された場合、家主の帰宅と考えられる。一方、室内で物体が検知された後に、鍵の解錠が検知された場合、窃盗犯の逃走と考えられる。
【０００７】
そこで、処理要素の移動が発生した場合でもデータの順序を保つ技術が提案されている。この技術では、例えば、ＣＥＰシステムを管理するコントローラが、処理要素の移動元サーバと移動先サーバを決定すると、移動要求を移動元サーバと移動先サーバに送信する。移動元サーバは、移動元サーバにデータを送信する全ての送信元にｐａｕｓｅ要求を送り、ｐａｕｓｅ要求に対するａｃｋを待つことで、送信元でデータの送信をｌｏｃｋさせ、移動対象の処理要素へのデータを送信元でバッファリングさせる。移動元サーバが全ての送信元からａｃｋを受け取った場合、転送中のデータがなくなったことになる。移動元サーバは、全ての送信元からａｃｋを受信するまでデータの処理を行う。そして、移動元サーバは、全ての送信元からａｃｋを受信すると、処理要素の移動を開始する。移動先サーバは、処理要素の移動完了後、全ての送信元へｒｅｓｔａｒｔ要求を送信する。送信元は、ｒｅｓｔａｒｔ要求を受信すると、ルーティングテーブルを変更してデータの送信先を移動先サーバへ変更し、データの送信を再開してバッファリングしたデータを移動先サーバへ送信する。
【先行技術文献】
【特許文献】
【０００８】
【特許文献１】特開２０１１−３９８２０号公報
【非特許文献】
【０００９】
【非特許文献１】Mehul  A.  Shah,  Joseph  M.  Hellerstein,  Sirish  Chandrasekaran  and  Michael  J.  Franklin著、“Flux:  An  Adaptive  Partitioning  Operator  for  Continuous  Query  Systems”、ICDE、2003
【発明の概要】
【発明が解決しようとする課題】
【００１０】
しかし、先行技術は、処理要素を移動させる際、移動元サーバが全ての送信元にｐａｕｓｅ要求を送信して送信元でのデータの送信を停止させる。このため。先行技術では、処理要素の移動毎に、データの処理が停止するオーバーヘッドが発生する。特に、多数の処理要素が移動する場合、各処理要素を移動させる毎にデータの処理が停止するため、オーバーヘッドが多くなる。
【００１１】
１つの側面では、クエリの移動に伴うオーバーヘッドを抑制できるサーバ装置、データ順序保証プログラム、およびデータ順序保証方法を提供することを目的とする。
【課題を解決するための手段】
【００１２】
第１の案では、サーバ装置は、停止部と、取得部と、比較部と、更新部と、を有する。複数のノードは、それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリが配置される。また、複数のノードは、各クエリが処理対象とするデータ毎に当該クエリの配置先およびクエリの移動に伴い更新されるバージョンを格納したルーティングテーブルを記憶する。停止部は、複数のノードに対するデータの送信を所定期間毎に停止させる。取得部は、前記停止部によりデータの送信を停止させた場合に前記複数のノードからルーティングテーブルのバージョンを取得する。比較部は、前記取得部により取得された複数のノードのルーティングテーブルのバージョンを比較する。更新部は、前記比較部による比較の結果、古いバージョンのルーティングテーブルが記憶されたノードがある場合、当該ノードのルーティングテーブルを更新する。
【発明の効果】
【００１３】
クエリの移動に伴うオーバーヘッドを抑制できる。
【図面の簡単な説明】
【００１４】
【図１】図１は、実施例１に係るサーバ装置を含むＣＥＰシステムの全体構成を示す図である。
【図２】図２は、実施例２に係るＣＥＰシステムの構成の一例を示す図である。
【図３】図３は、データの流れに沿ってＣＥＰシステムの構成を模式的に示した図である。
【図４Ａ】図４Ａは、クエリの移動の一例を示す図である。
【図４Ｂ】図４Ｂは、クエリの移動の他の一例を示す図である。
【図５】図５は、クエリのグループ分けの一例を示す図である。
【図６】図６は、ルーティングテーブルのデータ構成の一例を示す図である。
【図７】図７は、格納先テーブルのデータ構成の一例を示す図である。
【図８】図８は、管理サーバの構成の一例を示す図である。
【図９】図９は、停止期間テーブルのデータ構成の一例を示す図である。
【図１０】図１０は、データ順序保証処理の手順を示すフローチャートである。
【図１１】図１１は、実施例３に係るＣＥＰシステムの構成を模式的に示した図である。
【図１２】図１２は、ルーティングテーブルのデータ構成の一例を示す図である。
【図１３】図１３は、管理サーバの構成の一例を示す図である。
【図１４】図１４は、データ順序保証処理の手順を示すフローチャートである。
【図１５】図１５は、データ順序保証プログラムを実行するコンピュータを示す図である。
【発明を実施するための形態】
【００１５】
以下に、本発明にかかるサーバ装置、データ順序保証プログラム、およびデータ順序保証方法の実施例を図面に基づいて詳細に説明する。なお、この実施例によりこの発明が限定されるものではない。そして、各実施例は、処理内容を矛盾させない範囲で適宜組み合わせることが可能である。
【実施例１】
【００１６】
実施例１に係るサーバ装置について説明する。図１は、実施例１に係るサーバ装置を含むＣＥＰシステムの全体構成を示す図である。サーバ装置１０は、ＣＥＰシステムを管理する物理サーバであり、例えば、データセンサや各企業に設けられた管理用のサーバコンピュータである。サーバ装置１０は、複数のノード１１と通信可能とする。ノード１１は、サーバ装置またはサーバ装置上で動作する仮想マシンである。各ノード１１には、それぞれ設定された条件にデータが合致した場合に処理を実行するための複数のクエリ１２が分散配置される。各ノード１１には、様々な対象から収集されたデータが配信される。ノード１１は、受信したデータがクエリ１２に設定された条件に合致した場合にクエリ１２に設定された処理を実行する。
【００１７】
各ノード１１に配置されたクエリ１２は、ノード１１間の移動が可能とされている。各ノード１１は、例えば、処理負荷の分散などで、クエリ１２を移動させる場合に、データの送信元でのデータの送信を停止させることなく、クエリ１２を移動する。各ノード１１は、各クエリ１２が処理対象とするデータ毎に当該クエリ１２の配置先およびクエリ１２の移動に伴い更新されるバージョンを格納したルーティングテーブル１３を記憶する。各ノード１１は、クエリ１２を移動した場合、ルーティングテーブル１３に格納された移動したクエリ１２の配置先およびバージョンを更新する。例えば、クエリ１２が移動した場合、移動先のノード１１は、ルーティングテーブル１３に格納された移動したクエリ１２の配置先およびバージョンを更新すると共に、他のノード１１に移動したクエリ１２および移動先を通知する。各ノード１１は、移動したクエリ１２および移動先を通知された場合、ルーティングテーブル１３に格納された移動したクエリ１２の配置先およびバージョンを更新する。
【００１８】
図１に示すように、サーバ装置１０は、停止部１４と、取得部１５と、比較部１６と、更新部１７とを有する。
【００１９】
停止部１４は、データの配信を一時的に停止させる。例えば、停止部１４は、各ノード１１に対するデータの送信を所定期間毎に停止させる。停止部１４は、所定期間の周期でデータを送信する送信元へデータの送信停止要求を送信してデータの送信を停止させる。また、停止部１４は、後述する比較部１６の比較の結果、ルーティングテーブル１３のバージョンが同じ場合、又は後述する更新部１７によるルーティングテーブル１３の更新が完了した場合、送信元へデータの送信再開要求を送信してデータの送信を再開させる。
【００２０】
取得部１５は、各種情報を取得する。例えば、取得部１５は、停止部１４によりデータの送信を停止させた場合に各ノード１１からルーティングテーブル１３のバージョンを取得する。取得部１５は、例えば、各ノード１１に対してルーティングテーブル１３のバージョンの送信要求を送信する。そして、取得部１５は、各ノード１１からバージョンの通知を受信することにより、各ノード１１からルーティングテーブル１３のバージョンを取得する。
【００２１】
比較部１６は、各種の比較を行う。例えば、比較部１６は、取得部１５により取得された各ノード１１のルーティングテーブル１３のバージョンを比較する。
【００２２】
更新部１７は、各種の更新を行う。例えば、更新部１７は、比較部１６による比較の結果、古いバージョンのルーティングテーブル１３が記憶されたノード１１がある場合、当該ノード１１のルーティングテーブル１３を更新する。更新部１７は、例えば、何れのノード１１に古いバージョンのルーティングテーブル１３が記憶されている場合、新しいバージョンのルーティングテーブル１３を記憶したノード１１に対してルーティングテーブル１３の情報の送信を要求する。そして、比較部１６は、ノード１１からルーティングテーブル１３を受信した場合、当該ルーティングテーブル１３の内容を古いバージョンのルーティングテーブル１３を記憶したノード１１に送信して当該ノード１１のルーティングテーブル１３を更新する。なお、サーバ装置１０がルーティングテーブル１３を記憶しており、当該ルーティングテーブル１３が新しいものである場合、記憶したルーティングテーブル１３の内容を送信してもよい。
【００２３】
なお、図１の例では、機能的な構成を示したため、停止部１４と、取得部１５と、比較部１６と、更新部１７を別に分けているが、例えば、１つのデバイスで構成してもよい。デバイスの一例としては、ＣＰＵ（Central  Processing  Unit）やＭＰＵ（Micro  Processing  Unit）などの電子回路が挙げられる。なお、デバイスとして、ＡＳＩＣ（Application  Specific  Integrated  Circuit）やＦＰＧＡ（Field  Programmable  Gate  Array）などの集積回路を採用することもできる。
【００２４】
ここで、ＣＥＰシステムでは、システム化する処理の性質によっては厳密なデータの処理順序の保証まで要求しておらず、システム化する処理が要求する精度でデータの処理順序が保証されれば十分なケースが存在する。例えば、家屋の玄関ドアの鍵の解錠を検知するドアセンサと、室内の人を検知する人感センサを設け、ドアセンサと人感センサから送信されるデータを監視する防犯システムをＣＥＰシステムにより実現した場合を考える。このような防犯システムでは、人が玄関ドアの鍵の解錠してから室内に入るまでに数秒程度の期間がある。
【００２５】
そこで、サーバ装置１０では、要求される精度のデータ順序を保証する所定期間毎に各ノード１１に対するデータの送信を停止させ、各ノード１１のルーティングテーブル１３のバージョンを比較し、古いバージョンのルーティングテーブル１３を更新する。例えば、上記の防犯システムをＣＥＰシステムにより実現する場合、所定期間を数百ミリから数秒単位に定める。この所定期間は、サーバ装置１０に設けられた記憶部に記憶させてもよい。また、所定期間は、外部の記憶装置に記憶させ、外部の記憶装置から読み出すようにしてもよい。また、所定期間は、例えば、各ノード１１に対するデータの送信を所定期間毎に停止させるプログラム内に定義されてもよい。このように、サーバ装置１０では、システム化した処理が要求する精度のデータ順序を保証する期間毎にデータの処理を停止させて各ノード１１のルーティングテーブル１３の一貫性を確保する。これにより、サーバ装置１０によれば、クエリ１２の移動毎にデータの処理を停止させた場合と比較して、オーバーヘッドの発生を抑制できる。特に、ノード１１間を多数のクエリ１２が移動する場合、クエリ１２の移動毎にデータの処理を停止させるとオーバーヘッドが多くなるが、所定期間毎にデータの処理を停止させてルーティングテーブル１３の一貫性を確保することで、オーバーヘッドを抑制できる。
【００２６】
このように、サーバ装置１０は、複数のノード１１に対するデータの送信を所定期間毎に停止させる。また、サーバ装置１０は、データの送信を停止させた場合に複数のノード１１からルーティングテーブル１３のバージョンを取得する。そして、サーバ装置１０は、取得した複数のノード１１のルーティングテーブル１３のバージョンを比較する。サーバ装置１０は、比較の結果、古いバージョンのルーティングテーブル１３が記憶されたノード１１がある場合、当該ノード１１のルーティングテーブル１３を更新する。これにより、サーバ装置１０によれば、システム化する処理が要求する精度のデータの処理順序を保証できる。また、サーバ装置１０によれば、クエリ１２の移動に伴うオーバーヘッドを抑制できる。
【実施例２】
【００２７】
実施例２について説明する。実施例２では、複数のサーバ装置３０上でそれぞれＶＭ（Virtual  Machine：仮想マシン）３１が動作し、ＶＭ３１に複数のクエリ３３が分散配置されたＣＥＰシステム２０について説明する。図２は、実施例２に係るＣＥＰシステムの構成の一例を示す図である。ＣＥＰシステム２０は、サーバ装置２１と、複数のサーバ装置３０と、管理サーバ４０とを有する。サーバ装置２１と各サーバ装置３０と管理サーバ４０との間は、ネットワーク２２を介して通信可能に接続される。かかるネットワーク２２の一態様としては、有線または無線を問わず、ＬＡＮ（Local  Area  Network）やＶＰＮ（Virtual  Private  Network）などの任意の通信網が挙げられる。
【００２８】
サーバ装置３０では、ＶＭ３１が動作する。各ＶＭ３１には、複数のクエリ３３が分散配置される。サーバ装置２１は、外部ネットワーク２３を介して様々な対象からデータを受信し、受信したデータを当該データを処理対象とするクエリ３３が配置されたＶＭ３１に配信する。管理サーバ４０は、サーバ装置２１および各サーバ装置３０を管理しており、各ＶＭ３１に配置されたクエリ３３の移動を制御する。なお、図２の例では、サーバ装置３０を３つ図示したが、サーバ装置３０の数は何れであってもよい。
【００２９】
図３は、データの流れに沿ってＣＥＰシステムの構成を模式的に示した図である。サーバ装置２１、ＶＭ３１および管理サーバ４０は、各クエリ３３が処理対象とするデータ毎に当該クエリ３３の配置先およびクエリ３３の移動に伴い更新されるバージョンを格納したルーティングテーブル５０を記憶する。
【００３０】
サーバ装置２１は、外部ネットワーク２３を介して様々な対象から多数のデータを受信する。サーバ装置２１は、受信したデータを所定のデータ構成に整える前処理を行う。そして、サーバ装置２１は、ルーティングテーブル５０に基づき、受信したデータを処理対象とするクエリ３３の配置先を特定し、特定した配置先のＶＭ３１へ前処理後のデータを送信する。なお、本実施例では、サーバ装置２１を１つ図示したが、サーバ装置２１を複数設け、複数のサーバ装置２１により受信したデータを振り分けてもよい。
【００３１】
ＶＭ３１は、それぞれデータストリーム処理エンジン３２を動作させる。データストリーム処理エンジン３２は、複合イベント処理を実現するソフトウェアである。データストリーム処理エンジン３２は、受信したデータと、クエリ３３の条件式とを突合させてイベントを検出し、検出したイベントの処理を実行する制御を行う。また、ＣＥＰシステム２０では、複数のクエリ３３により一連の処理を実行させることもできる。例えば、上位のクエリ３３による実行結果のデータを下位のクエリ３３へ送信し、下位のクエリ３３の入力のデータとすることもできる。データストリーム処理エンジン３２は、複数のクエリ３３により一連の処理を実行する場合、クエリ３３の実行結果のデータを下位のクエリ３３が動作するＶＭ３１へ転送する。また、データストリーム処理エンジン３２は、管理サーバ４０からのクエリ３３の移動の指示に応じて、他のＶＭ３１上で動作するデータストリーム処理エンジン３２との間でクエリ３３を移動させる制御を行う。データストリーム処理エンジン３２は、移動したクエリ３３が処理対象とするデータを受信した場合、受信したデータをクエリ３３の移動先のＶＭ３１へ転送する。また、データストリーム処理エンジン３２は、管理サーバ４０から後述する送信停止要求を受信した場合、ＶＭ３１間を転送中のデータがあるかを確認する確認処理を行う。例えば、データストリーム処理エンジン３２は、他の全てのＶＭ３１のデータストリーム処理エンジン３２へ制御パケットを送信する。そして、データストリーム処理エンジン３２は、他のＶＭ３１から制御パケットの受信待ちを行う。
【００３２】
ここで、送信停止要求を受信したタイミングでＶＭ３１間を転送中のデータがある場合、ＶＭ３１は、他のＶＭ３１へ制御パケットの送信後、他の全てのＶＭ３１から制御パケットを受信する前に、他のＶＭ３１への転送データが受信される。一方、ＶＭ３１間を転送中のデータがない場合、ＶＭ３１は、他のＶＭ３１へ制御パケットの送信後、転送データを受信することなく、他の全てのＶＭ３１から制御パケットを受信する。
【００３３】
データストリーム処理エンジン３２は、他の全てのＶＭ３１から制御パケットの受信する前に、転送データを受信した場合、当該転送データを転送先のＶＭ３１へ転送する。そして、データストリーム処理エンジン３２は、確認処理の再実施の要求を送信する。各ＶＭ３１のデータストリーム処理エンジン３２は、確認処理の再実施の要求を受信した場合、それぞれ他の全てのＶＭ３１のデータストリーム処理エンジン３２へ制御パケットを再度送信する。
【００３４】
一方、データストリーム処理エンジン３２は、転送データを受信することなく、他の全てのノード１１から制御パケットの受信した場合、サーバ装置１０へ全てのＶＭ３１から制御パケットを受信した通知を行う。なお、この通知は、各ＶＭ３１のデータストリーム処理エンジン３２がサーバ装置１０へ通知してもよい。また、いずれかのＶＭ３１のデータストリーム処理エンジン３２が代表となり、確認処理の再実施の要求を受信することなく、他の全てのＶＭ３１から通知を受けた場合にサーバ装置１０へ通知してもよい。
【００３５】
ここで、クエリ３３の移動について説明する。クエリ３３は、付随するデータを有する場合がある。例えば、クエリ３３が処理対象とするデータの一定期間の平均を求める処理を行う場合、平均を算出するために一定期間に受信したデータを保持する。また、クエリ３３は、処理にテーブルなどの所定のデータを用いる場合もある。データストリーム処理エンジン３２は、クエリ３３を移動させる場合、クエリ３３およびクエリ３３に付随するデータを処理要素として移動させる。
【００３６】
また、クエリ３３の移動には、以下のようなケースが考えられる。図４Ａは、クエリ３３の移動の一例を示す図である。図４Ａに示すように、クエリ３３毎に処理対象とするデータが異なる場合、クエリ３３を移動させ、移動先で処理対象とするデータを処理させる。例えば、処理を分散させる場合、各クエリ３３をそれぞれ別なＶＭ３１に移動させて、各ＶＭ３１によりデータを処理させる。一方、例えば、処理負荷が低い場合、クエリ３３を１つのＶＭ３１に移動させ、１つのＶＭ３１によりデータを処理させる。
【００３７】
一方、クエリ３３は、処理対象とするデータが複数である場合がある。例えば、データとして証券コードおよび株価がＶＭ３１に配信され、証券コード毎に株価の一定期間の移動平均を求める処理を１つのクエリ３３により行う場合がある。図４Ｂは、クエリ３３の移動の他の一例を示す図である。図４Ｂの左側に示すように、クエリ３３は、データ１〜ｎを処理対象とする。この場合、処理対象のデータ１〜ｎ毎に、クエリ３３を移動させる。例えば、処理の分散を行う場合、クエリ３３を複製して別なＶＭ３１に移動させ、処理対象のデータ１〜ｎをそれぞれ別なＶＭ３１により処理させる。一方、例えば、処理負荷が低い場合、クエリ３３を１つのＶＭ３１にまとめ、処理対象のデータ１〜ｎを１つのＶＭ３１により処理させる。
【００３８】
ところで、各クエリ３３に対して個別に移動を指示して移動させた場合、移動対象のクエリ３３の数が多くなるほど移動の指示回数が多くなる。また、クエリ３３が処理対象とするデータ毎に、ルーティングテーブル５０に配置先を格納する場合、ルーティングテーブル５０は、クエリ３３が処理対象とするデータが多くなるほどデータ量が大きくなる。そこで、本実施例では、複数のクエリ３３を複数のグループに分け、グループ毎に移動する。また、ルーティングテーブル５０は、グループ毎に格納先を管理する。図５は、クエリ３３のグループ分けの一例を示す図である。例えば、処理対象とする各データに、３２ビットの整数データにより、各データを識別するキーが含まれ、それぞれのクエリ３３の処理の条件に異なるキーが設定されているものとする。この場合、例えば、処理の条件に設定されたキーを所定の整数値で割った余りが同じクエリ３３を同一のグループとしてグループ分けする。一例としては、処理の条件に設定されたキーを８１９１で割った余りが同じクエリ３３を同一のグループとして静的にグループを分ける。これにより、各クエリ３３は、８１９１個のグループに分かれる。図５の例では、クエリ３３をグループに分けした各グループをＶ−Ｎｏｄｅと示している。そして、本実施例では、Ｖ−Ｎｏｄｅ毎に、クエリ３３をＶＭ３１に配置する。
【００３９】
ルーティングテーブル５０は、Ｖ−Ｎｏｄｅ毎に配置先およびバージョンを格納する。図６は、ルーティングテーブルのデータ構成の一例を示す図である。図６に示すように、ルーティングテーブル５０は、ＩＤ、格納先、版数の各項目を有する。ＩＤの項目は、Ｖ−Ｎｏｄｅを識別するＩＤを記憶する領域である。格納先の項目は、格納先のＶＭ３１を示すＩＤを記憶する領域である。版数の項目は、バージョン示す値を記憶する領域である。図６の例では、ＩＤが「１２」のＶ−Ｎｏｄｅは、格納先のＶＭ３１のＩＤが「１」であり、バージョンが「３」であることを示す。
【００４０】
サーバ装置２１および各ＶＭ３１のデータストリーム処理エンジン３２は、キーに基づいてクエリ３３の格納先を判別する。例えば、サーバ装置２１は、受信したデータに含まれるキーが８２０３であった場合、８２０３を８１９１で割った余り、１２を求める。そして、サーバ装置２１は、ルーティングテーブル５０からＶ−ＮｏｄｅのＩＤが１２の格納先を求める。図６の例では、Ｖ−ＮｏｄｅのＩＤが１２の格納先のＶＭ３１のＩＤが１であると求まる。
【００４１】
各ＶＭ３１は、それぞれ通信を行うＩＰアドレスおよびポート番号が定められ、各ＶＭ３１のＩＰアドレスおよびポート番号に対応してファイルディスクリプタが割り当てられる。サーバ装置２１、各ＶＭ３１および管理サーバ４０は、それぞれファイルディスクリプタの番号と各ＶＭ３１を示すＩＤを格納先テーブルに記憶する。図７は、格納先テーブルのデータ構成の一例を示す図である。図７に示すように、格納先テーブルは、ＩＤ、ファイルディスクリプタの各項目を有する。ＩＤの項目は、ＶＭ３１のＩＤを記憶する領域である。ファイルディスクリプタの項目は、ＩＤが示すＶＭ３１への通信路に割り当てられたファイルディスクリプタの番号を記憶する領域である。図７の例では、ＩＤが「０」のＶＭ３１への通信路に割り当てられたファイルディスクリプタの番号が「６」であることを示す。
【００４２】
ＶＭ３１と通信を行う場合、格納先テーブルから通信対象のＶＭ３１のＩＤに対応するファイルディスクリプタの番号を求める。例えば、通信対象のＶＭ３１のＩＤが「１」である場合、格納先テーブルからＶＭ３１のＩＤが１のファイルディスクリプタの番号が「７」と求まる。そして、求めたファイルディスクリプタの番号を用いてＴＣＰ／ＩＰのソケットにより通信対象のＶＭ３１と通信を行う。
【００４３】
また、各ＶＭ３１は、それぞれサーバ装置３０から割り当てられたＣＰＵやメモリの使用率などのリソースの負荷状態を検出し、検出した負荷状態を示す負荷情報を管理サーバ４０へ送信する。管理サーバ４０は、各ＶＭ３１から送信された負荷情報に基づき、移動対象のクエリ３３を特定する。そして、管理サーバ４０は、移動対象のクエリ３３の移動指示を送信する。なお、本実施例では、Ｖ−Ｎｏｄｅ単位でクエリ３３の移動を行う。管理サーバ４０は、移動先のＶＭ３１のＩＤおよび移動対象のＶ−ＮｏｄｅのＩＤを含んだ移動指示を移動元のＶＭ３１に送信する。移動指示を受信したＶＭ３１のデータストリーム処理エンジン３２は、移動対象のＶ−Ｎｏｄｅに属するクエリ３３や当該クエリ３３に付随するデータなどの処理要素をシリアライズして移動先のＶＭ３１へ送信する。
【００４４】
移動先のＶＭ３１のデータストリーム処理エンジン３２は、送信されたデータをデシリアライズして復元し、復元されたクエリ３３の駆動を開始する。また、データストリーム処理エンジン３２は、ルーティングテーブル５０に格納された移動対象のＶ−Ｎｏｄｅに属するクエリ３３の格納先を自身が動作するＶＭ３１のＩＤに更新すると共にバージョンを更新する。例えば、バージョンの値を１加算した値に更新する。そして、データストリーム処理エンジン３２は、各ＶＭ３１、サーバ装置２１および管理サーバ４０に対して移動対象のＶ−ＮｏｄｅのＩＤおよび移動先のＶＭ３１のＩＤを通知する。
【００４５】
各ＶＭ３１、サーバ装置２１および管理サーバ４０は、移動が通知された場合、ルーティングテーブル５０の通知された移動対象のＶ−Ｎｏｄｅに属するクエリ３３の格納先を通知された移動先のＶＭ３１のＩＤに更新すると共にバージョンを更新する。例えば、バージョンの値を１加算した値を受け取り更新する。
【００４６】
図８は、管理サーバの構成の一例を示す図である。図８に示すように、管理サーバ４０は、通信制御Ｉ／Ｆ部４１と、記憶部４２と、制御部４３とを有する。
【００４７】
通信制御Ｉ／Ｆ部４１は、少なくとも１つのポートを有し、各ＶＭ３１、サーバ装置２１との間の通信を制御するインタフェースである。通信制御Ｉ／Ｆ部４１は、各ＶＭ３１およびサーバ装置２１と各種情報を送受信する。例えば、通信制御Ｉ／Ｆ部４１は、ＶＭ３１から負荷情報を受信する。また、通信制御Ｉ／Ｆ部４１は、ＶＭ３１へ各種指示を送信する。
【００４８】
記憶部４２は、各種情報を記憶する。例えば、記憶部４２は、上述したルーティングテーブル５０と、最大期間情報５１と、停止期間テーブル５２を記憶する。記憶部４２のデバイスの一例としては、フラッシュメモリやＮＶＳＲＡＭ(Non  Volatile  Static  Random  Access  Memory)などのデータを書き換え可能な半導体メモリや、ハードディスク、光ディスクなどの記憶装置が挙げられる。
【００４９】
最大期間情報５１は、ＣＥＰシステム２０において複数のＶＭ３１の間でのデータが転送された場合に伝送にかかる最大期間を記憶したものである。最大期間情報５１は、ＣＥＰシステム２０の判定部４３ａにより、管理サーバ４０と通信可能とされたクライアントコンピュータなどの端末装置あるいは管理サーバ４０に設けられたマウス、キーボードなどの入力装置から登録される。
【００５０】
停止期間テーブル５２は、複数のクエリ３３を複数のグループに分けたグループ毎に、データの送信を停止させる所定期間を記憶するテーブルである。図９は、停止期間テーブルのデータ構成の一例を示す図である。図９に示すように、停止期間テーブル５２は、ＩＤ、期間の各項目を有する。ＩＤの項目は、Ｖ−ＮｏｄｅのＩＤを記憶する領域である。期間の項目は、処理順序を保証する期間をミリ秒単位で記憶する領域である。図９の例では、ＩＤが「１２」のＶ−Ｎｏｄｅの所定期間が「１００」ｍｓであることを示す。
【００５１】
制御部４３は、例えば、内部メモリ等を有するＣＰＵなどの電子回路であり、判定部４３ａと、移動部４３ｂと、実行部４３ｃと、停止部４３ｄと、取得部４３ｅと、比較部４３ｆと、更新部４３ｇとを有する。
【００５２】
判定部４３ａは、各種の判定を行う。例えば、判定部４３ａは、各ＶＭ３１から送信された負荷情報により示される各ＶＭ３１の負荷状態に許容される負荷以上のものがあるか否か判定する。
【００５３】
移動部４３ｂは、クエリ３３の移動を行う。例えば、移動部４３ｂは、判定部４３ａにより許容される負荷以上と判定されたＶＭ３１がある場合、当該ＶＭ３１の配置された何れのＶ−Ｎｏｄｅの移動を行う。移動部４３ｂは、例えば、各ＶＭ３１から送信された負荷情報から負荷状態の低いＶＭ３１を特定する。そして、移動部４３ｂは、許容される負荷以上と判定されたＶＭ３１を移動元とし、負荷状態の低いＶＭ３１を移動先とし、許容される負荷以上と判定されたＶＭ３１に配置された何れかのＶ−Ｎｏｄｅを移動対象として、移動指示を送信する。
【００５４】
実行部４３ｃは、停止期間テーブル５２から移動したＶ−Ｎｏｄｅに対応する所定期間を読み出す。また、実行部４３ｃは、最大期間情報５１を読み出す。そして、実行部４３ｃは、移動したＶ−Ｎｏｄｅに対応する所定期間が最大期間情報５１が示す最大期間よりも短い場合、後述する停止部４３ｄ、取得部４３ｅ、比較部４３ｆ、更新部４３ｇの処理を実行する。一方、実行部４３ｃは、求めた期間が最大期間以上の場合、停止部４３ｄ、取得部４３ｅ、比較部４３ｆ、更新部４３ｇの処理を停止する。
【００５５】
ここで、移動したＶ−Ｎｏｄｅに対応する期間が最大期間よりも長い場合、クエリ３３を移動させてからＶ−Ｎｏｄｅに対応する期間を経過するまでに、移動元のＶＭ３１から各ＶＭ３１を介して中継されたデータが移動先のＶＭ３１に到着する。よって、Ｖ−Ｎｏｄｅに対応する期間の前後でのデータ順序が保証される。このため、実行部４３ｃは、移動したＶ−Ｎｏｄｅに対応する期間が最大期間以上の場合に、停止部４３ｄ、取得部４３ｅ、比較部４３ｆ、更新部４３ｇの処理を停止する。これにより、管理サーバ４０によれば、不要な処理の実行を抑制できる。
【００５６】
停止部４３ｄは、Ｖ−Ｎｏｄｅが移動してから当該Ｖ−Ｎｏｄｅに対応する所定期間が経過したときにデータの送信を停止させる。例えば、停止部４３ｄは、Ｖ−Ｎｏｄｅの移動先から移動対象のＶ−ＮｏｄｅのＩＤおよび移動先のＶＭ３１のＩＤが通知されてから、当該Ｖ−Ｎｏｄｅに対応する所定期間を経過したときにデータの送信元および各ＶＭ３１へ送信停止要求を送信する。各ＶＭ３１のデータストリーム処理エンジン３２は、送信停止要求を受信すると、ＶＭ３１間を転送中のデータがあるかの確認処理を行い、転送中のデータがない場合にサーバ装置１０へ全てのＶＭ３１から制御パケットを受信した通知を行う。
【００５７】
停止部４３ｄは、後述する比較部４３ｆの比較の結果、ルーティングテーブル５０のバージョンが同じ場合、又は後述する更新部４３ｇによるルーティングテーブル５０の更新が完了した場合、送信元へデータの送信再開要求を送信してデータの送信を再開させる。
【００５８】
取得部４３ｅは、全てのＶＭ３１から制御パケットを受信した旨が通知されると、各ＶＭ３１およびサーバ装置２１からルーティングテーブル５０に格納された移動したＶ−Ｎｏｄｅに対応するバージョンを取得する。例えば、取得部４３ｅは、各ＶＭ３１およびサーバ装置２１に対して移動したＶ−Ｎｏｄｅに対応するバージョンの送信要求を送信する。そして、取得部４３ｅは、各ＶＭ３１およびサーバ装置２１からバージョンの通知を受信することにより、各ＶＭ３１およびサーバ装置２１からバージョンを取得する。
【００５９】
比較部４３ｆは、取得した各ＶＭ３１およびサーバ装置２１のルーティングテーブル５０のバージョンを比較する。
【００６０】
更新部４３ｇは、比較の結果、各ＶＭ３１およびサーバ装置２１の何れかに古いバージョンのルーティングテーブル５０が記憶されている場合、ルーティングテーブル５０を更新する。例えば、記憶部４２に記憶されたルーティングテーブル５０のバージョンが新しい場合、更新部４３ｇは、古いバージョンのルーティングテーブル５０が記憶されたＶＭ３１やサーバ装置２１に記憶部４２に記憶されたルーティングテーブル５０の情報を送信する。また、記憶部４２に記憶されたルーティングテーブル５０のバージョンが古い場合、更新部４３ｇは、新しいバージョンのルーティングテーブル５０を記憶したＶＭ３１やサーバ装置２１に対してルーティングテーブル５０の情報の送信を要求する。そして、更新部４３ｇは、新しいルーティングテーブル５０の情報を受信した場合、受信した情報を古いバージョンのルーティングテーブル５０が記憶されたＶＭ３１やサーバ装置２１に送信する。また、更新部４３ｇは、受信した情報で記憶部４２に記憶されたルーティングテーブル５０を更新する。
【００６１】
次に、本実施例に係る管理サーバ４０がデータ順序の保証を行う処理の流れを説明する。図１０は、データ順序保証処理の手順を示すフローチャートである。このデータ順序保証処理は、例えば、Ｖ−Ｎｏｄｅの移動先から移動対象のＶ−ＮｏｄｅのＩＤおよび移動先のＶＭ３１のＩＤが通知されたタイミングで実行される。
【００６２】
図１０に示すように、実行部４３ｃは、停止期間テーブル５２から移動したＶ−Ｎｏｄｅに対応する所定期間を読み出す（ステップＳ１０）。また、実行部４３ｃは、最大期間情報５１として記憶された最大期間を読み出す（ステップＳ１１）。そして、実行部４３ｃは、移動したＶ−Ｎｏｄｅに対応する所定期間が最大期間よりも短いか否かを判定する（ステップＳ１２）。所定期間が最大期間以上の場合（ステップＳ１２否定）、実行部４３ｃは、処理を終了する。一方、所定期間が最大期間よりも短い場合（ステップＳ１２肯定）、停止部４３ｄは、Ｖ−Ｎｏｄｅの移動が通知されてから、当該Ｖ−Ｎｏｄｅに対応する所定期間を経過したか否かを判定する（ステップＳ１３）。所定期間を経過していない場合（ステップＳ１３否定）、再度ステップＳ１３へ移行して所定期間の経過待ちを行う。一方、所定期間を経過した場合（ステップＳ１３肯定）、停止部４３ｄは、データの送信元および各ＶＭ３１へ送信停止要求を送信する（ステップＳ１４）。停止部４３ｄは、全ての送信元から送信停止要求に対するＡＣＫが得られたか判定する（ステップＳ１５）。ＡＣＫが得られていない場合（ステップＳ１５否定）、再度ステップＳ１５へ移行してＡＣＫ待ちを行う。一方、ＡＣＫが得られた場合（ステップＳ１５肯定）、取得部４３ｅは、全てのＶＭ３１から制御パケットを受信した通知を受信したか否かを判定する（ステップＳ１６）。全てのＶＭ３１から通知が得られていない場合（ステップＳ１６否定）、再度ステップＳ１６へ移行して通知待ちを行う。
【００６３】
一方、全てのＶＭ３１から通知が得られた場合（ステップＳ１６肯定）、取得部４３ｅは、各ＶＭ３１およびサーバ装置２１からルーティングテーブル５０に格納された移動したＶ−Ｎｏｄｅに対応するバージョンを取得する（ステップＳ１７）。比較部４３ｆは、取得した各ＶＭ３１およびサーバ装置２１のルーティングテーブル５０のバージョンを比較し、古いバージョンのルーティングテーブル５０があるか否か判定する（ステップＳ１８）。古いバージョンのルーティングテーブル５０がある場合（ステップＳ１８肯定）、更新部４３ｇは、古いバージョンのルーティングテーブル５０を更新する（ステップＳ１９）。一方、古いバージョンのルーティングテーブル５０がない場合（ステップＳ１８否定）、後述するステップＳ２０へ移行する。停止部４３ｄは、送信元へデータの送信再開要求を送信してデータの送信し（ステップＳ２０）、処理を終了する。
【００６４】
このように、管理サーバ４０は、複数のクエリ３３を複数のグループに分けたＶ−Ｎｏｄｅ毎に所定期間を記憶する。また、管理サーバ４０は、Ｖ−Ｎｏｄｅが移動してから当該Ｖ−Ｎｏｄｅに対応する所定期間が経過したときにデータの送信を停止させる。そして、管理サーバ４０は、各ＶＭ３１およびサーバ装置２１から、移動したＶ−Ｎｏｄｅに関するルーティングテーブル５０のバージョンを取得する。管理サーバ４０は、取得したバージョンを比較し、古いバージョンのルーティングテーブル５０を更新してルーティングテーブル５０の一貫性を確保する。これにより、管理サーバ４０によれば、要求される精度でのデータ順序を保証できる。また、管理サーバ４０によれば、クエリ３３の移動させる毎にデータの送信を停止させる必要が無くなるため、オーバーヘッドを抑制できる。
【００６５】
また、管理サーバ４０によれば、Ｖ−Ｎｏｄｅが移動から所定期間が経過したときにデータの送信を停止させて、バージョンの確認を行う。これにより、管理サーバ４０によれば、要求される精度のデータ順序を効率よく保証できる。
【００６６】
また、管理サーバ４０によれば、移動したＶ−Ｎｏｄｅに対応する所定期間が最大期間よりも短い場合、ルーティングテーブル５０の一貫性の確認を行い、所定期間が最大期間以上の場合、一貫性の確認を行わないことで、処理負荷を軽減できる。
【実施例３】
【００６７】
実施例３について説明する。実施例３に係るＣＥＰシステムの構成は、上述の実施例２とほぼ同様であるため、異なる部分について説明する。本実施例は、上述の実施例１と同様に、管理サーバ４０が、要求される精度のデータ順序を保証する所定期間毎に各ＶＭ３２からルーティングテーブル５０のバージョンを取得して比較し、古いバージョンのルーティングテーブル５０を更新する。図１１は、実施例３に係るＣＥＰシステムの構成を模式的に示した図である。なお、上述の実施例２と同一の部分については、同一の符号を付してその説明を省略する。
【００６８】
各ＶＭ３１、サーバ装置２１および管理サーバ４０は、ルーティングテーブル５０を更新した場合、更新日時をバージョンとして格納する。Ｖ−Ｎｏｄｅの各クエリ３３を移動させた場合、移動先のＶＭ３１のデータストリーム処理エンジン３２は、ルーティングテーブル５０の移動したＶ−Ｎｏｄｅの各クエリ３３の格納先を更新すると共にバージョンに更新した更新日時を記憶する。そして、データストリーム処理エンジン３２は、各ＶＭ３１、サーバ装置２１および管理サーバ４０に対して移動したＶ−ＮｏｄｅのＩＤ、移動先のＶＭ３１のＩＤ、および更新日時を通知する。各ＶＭ３１、サーバ装置２１および管理サーバ４０は、移動が通知された場合、ルーティングテーブル５０の通知された移動したＶ−Ｎｏｄｅの各クエリ３３の格納先を更新すると共にバージョンを通知された更新日時に更新する。図１２は、ルーティングテーブルのデータ構成の一例を示す図である。なお、上述の実施例２と同一の部分については、同一の符号を付してその説明を省略する。図１２に示すように、ルーティングテーブル５０は、版数の項目に更新日時が記憶される。図１２の例では、ＩＤが「１２」のＶ−Ｎｏｄｅは、格納先のＶＭ３１のＩＤが「１」であり、更新日時が「２０１１年１１月１日  １０時１分２０．１２秒」であることを示す。
【００６９】
また、各ＶＭ３１のデータストリーム処理エンジン３２は、一定期間毎に、当該一定期間において最初にデータを転送した日時に関する日時情報５３を記憶する。この日時情報５３は、例えば、サーバ装置２１が受信したデータに受信日時を付加などデータに日時が含まれる場合、データに含まれる日時を示す日時情報としてもよい。また、日時情報５３は、データストリーム処理エンジン３２がデータを転送した日時を示す日時情報としてもよい。そして、データストリーム処理エンジン３２は、一定期間毎に、記憶した日時情報５３を管理サーバ４０へ送信する。なお、データストリーム処理エンジン３２は、一定期間の間にデータの転送を行っていない場合、一定期間毎に、現在時刻を日時情報として管理サーバ４０へ送信する。この一定期間は、管理サーバ４０が、各ＶＭ３２からルーティングテーブル５０のバージョンを取得して比較を行う所定期間以下であればよい。これにより、管理サーバ４０には、一定期間毎に各ＶＭ３１からデータを転送した日時または現在日時が通知される。
【００７０】
図１３は、管理サーバの構成の一例を示す図である。なお、上述の実施例２と同一の部分については、同一の符号を付してその説明を省略する。
【００７１】
制御部４３は、特定部４３ｈをさらに有する。特定部４３ｈは、各種の特定を行う。例えば、特定部４３ｈは、各ＶＭ３１から通知された日時情報に基づき、直前の一定期間で最も過去に他のＶＭ３１へデータを転送した転送日時を特定する。特定部４３ｈは、例えば、各ＶＭ３１から通知された日時情報のうち日時が最も過去の日時情報の日時を直前の一定期間で最も過去に他のＶＭ３１へデータを転送した転送日時を特定する。
【００７２】
停止部４３ｄは、要求される精度のデータ順序を保証する所定期間毎に、特定部４３ｈにより特定された直前の一定期間での最も過去の転送日時と、ルーティングテーブル５０にバージョンとして記憶された更新日時とを比較する。そして、停止部４３ｄは、転送日時が更新日時以降である場合に、データの送信元および各ＶＭ３１へ送信停止要求を送信してデータの送信を停止させる。一方、停止部４３ｄは、転送日時が更新日時よりも前である場合に、送信停止要求の送信を行わない。
【００７３】
取得部４３ｅは、停止部４３ｄが送信停止要求を送信した場合、全てのＶＭ３１から制御パケットを受信した旨が通知されると、各ＶＭ３１およびサーバ装置２１からルーティングテーブル５０のバージョンを取得する。すなわち、管理サーバ４０は、直前の一定期間での最も過去の転送日時がルーティングテーブル５０の更新日時以降である場合、データの送信元および各ＶＭ３１でのデータの送信を停止させてルーティングテーブル５０のバージョンを取得する。一方、取得部４３ｅは、停止部４３ｄが送信停止要求の送信を行わない場合、所定期間毎に、各ＶＭ３１およびサーバ装置２１からルーティングテーブル５０のバージョンを取得する。すなわち、管理サーバ４０は、直前の一定期間での最も過去の転送日時がルーティングテーブル５０の更新日時よりも前である場合、データの送信元および各ＶＭ３１でのデータの送信を停止させずにルーティングテーブル５０のバージョンを取得する。
【００７４】
比較部４３ｆは、取得した各ＶＭ３１およびサーバ装置２１のルーティングテーブル５０のバージョンを比較する。更新部４３ｇは、比較の結果、各ＶＭ３１およびサーバ装置２１の何れかに古いバージョンのルーティングテーブル５０が記憶されている場合、ルーティングテーブル５０を更新する。
【００７５】
図１４は、データ順序保証処理の手順を示すフローチャートである。なお、上述の実施例２と同一の部分については、同一の符号を付してその説明を省略する。このデータ順序保証処理は、例えば、要求される精度のデータ順序を保証する所定期間毎に実行される。
【００７６】
特定部４３ｈは、各ＶＭ３１から通知された日時情報に基づき、直前の一定期間での最も過去の転送日時を特定する（ステップＳ３０）。停止部４３ｄは、直前の一定期間での最も過去の転送日時がルーティングテーブル５０の更新日時以降であるか否かを判定する（ステップＳ３１）。転送日時が更新日時以降である場合（ステップＳ３１肯定）、ステップＳ１４へ移行し、送信停止要求を送信してデータの送信を停止させる。一方、転送日時が更新日時より新しい場合（ステップＳ３１否定）、ステップＳ１７へ移行し、各ＶＭ３１およびサーバ装置２１からルーティングテーブル５０のバージョンを取得する。
【００７７】
このように、管理サーバ４０は、直前の一定期間での最も過去の転送日時がルーティングテーブル５０の更新日時以降である場合、データの送信元および各ＶＭ３１でのデータの送信を停止させてルーティングテーブル５０のバージョンを取得する。これにより、管理サーバ４０によれば、転送日時が更新日時以降である場合、データ順序が変る可能性があるため、データの送信を停止させることにより、データ送信を停止させた前後のデータ順序を保証できる。
【００７８】
また、管理サーバ４０は、直前の一定期間での最も過去の転送日時がルーティングテーブル５０の更新日時よりも新しい場合、データの送信元および各ＶＭ３１でのデータの送信を停止させずにルーティングテーブル５０のバージョンを取得する。これにより、管理サーバ４０によれば、転送日時が更新日時よりも前である場合、データ順序が保たれるため、データ送信を停止させずにルーティングテーブル５０のバージョンを取得することにより、データ送信を停止による遅延を抑制できる。
【実施例４】
【００７９】
さて、これまで開示の装置に関する実施例について説明したが、開示の技術は上述した実施例以外にも、種々の異なる形態にて実施されてよいものである。そこで、以下では、本発明に含まれる他の実施例を説明する。
【００８０】
例えば、上記の実施例１では、複数のノード１１に対するデータの送信を所定期間毎に停止させる場合について説明したが、開示の装置はこれに限定されない。例えば、複数のノード１１の間でクエリ３３が移動した場合、クエリ３３が移動してから所定期間が経過したときにデータの送信を停止させてもよい。これにより、要求される精度のデータ順序を保証できる。
【００８１】
また、上記の実施例２では、Ｖ−Ｎｏｄｅが移動から所定期間が経過したときにデータの送信を停止させる場合について説明したが、開示の装置はこれに限定されない。例えば、所定期間を１つ定め、Ｖ−Ｎｏｄｅが移動から当該所定期間が経過したときに停止させてもよい。
【００８２】
また、上記の実施例２では、Ｖ−Ｎｏｄｅが移動から所定期間が経過したときにデータの送信を１回停止させる場合について説明したが、開示の装置はこれに限定されない。例えば、所定期間毎に複数回停止させてもよい。すなわち、所定期間毎で少なくとも１回停止させればよい。
【００８３】
また、上記の実施例２では、複数のクエリ３３を複数のグループに分けたＶ−Ｎｏｄｅ毎に所定期間を停止期間テーブル５２に記憶させた場合について説明したが、開示の装置はこれに限定されない。例えば、ルーティングテーブル５０に所定期間を記憶させるフィールドを設けてルーティングテーブル５０に記憶させてもよい。
【００８４】
また、上記の実施例２では、Ｖ−Ｎｏｄｅが移動から所定期間が経過したときにデータの送信を１回停止させる場合について説明したが、開示の装置はこれに限定されない。例えば、実施例３と同様に、ルーティングテーブル５０にバージョンとして更新日時を記憶させる。また、各ＶＭ３１のデータストリーム処理エンジン３２は、一定期間毎に、当該一定期間において最初にデータを転送した日時に関する日時情報を管理サーバ４０へ送信する。管理サーバ４０は、Ｖ−Ｎｏｄｅが移動から所定期間が経過したときに、直前の一定期間での最も過去の転送日時と、ルーティングテーブル５０にバージョンとして記憶された更新日時とを比較する。管理サーバ４０は、転送日時が更新日時以降である場合に、データの送信元および各ＶＭ３１でのデータの送信を停止させてルーティングテーブル５０のバージョンを取得する。一方、管理サーバ４０は、転送日時が更新日時よりも前である場合に、データの送信元および各ＶＭ３１でのデータの送信を停止させずにルーティングテーブル５０のバージョンを取得するようにしてもよい。
【００８５】
また、上記の実施例２および実施例３では、複数のクエリ３３を複数のグループに分けたＶ−Ｎｏｄｅ単位にクエリ３３を移動させる場合について説明したが、開示の装置はこれに限定されない。例えば、クエリ３３単位で移動させてもよい。この場合、ルーティングテーブル５０は、クエリ３３単位に格納先を記憶できるように構成すればよい。また、この場合、例えば、停止期間テーブル５２にクエリ３３単位に所定期間を記憶させ、クエリ３３が移動してから当該クエリ３３に対応する所定期間が経過したときにデータの送信を停止させてもよい。
【００８６】
また、図示した各装置の各構成要素は機能概念的なものであり、必ずしも物理的に図示の如く構成されていることを要しない。すなわち、各装置の分散・統合の具体的状態は図示のものに限られず、その全部または一部を、各種の負荷や使用状況などに応じて、任意の単位で機能的または物理的に分散・統合して構成することができる。例えば、図１に示す停止部１４と、取得部１５と、比較部１６と、更新部１７の各処理部が適宜統合されてもよい。また、例えば、図８、図１３に示す判定部４３ａと、移動部４３ｂと、実行部４３ｃと、停止部４３ｄと、取得部４３ｅと、比較部４３ｆと、更新部４３ｇ、特定部４３ｈの各処理部が適宜統合されてもよい。さらに、各処理部にて行なわれる各処理機能は、その全部または任意の一部が、ＣＰＵおよび当該ＣＰＵにて解析実行されるプログラムにて実現され、あるいは、ワイヤードロジックによるハードウェアとして実現され得る。
【００８７】
［データ順序保証プログラム］
また、上記の実施例で説明した各種の処理は、あらかじめ用意されたプログラムをパーソナルコンピュータやワークステーションなどのコンピュータシステムで実行することによって実現することもできる。そこで、以下では、上記の実施例と同様の機能を有するプログラムを実行するコンピュータシステムの一例を説明する。図１５は、データ順序保証プログラムを実行するコンピュータを示す図である。
【００８８】
図１５に示すように、コンピュータ３００は、ＣＰＵ（Central  Processing  Unit）３１０、ＲＯＭ（Read  Only  Memory）３２０、ＨＤＤ（Hard  Disk  Drive）３３０、ＲＡＭ（Random  Access  Memory）３４０を有する。これら３１０〜３４０の各部は、バス４００を介して接続される。
【００８９】
ＲＯＭ３２０には上記実施例１の停止部１４、取得部１５、比較部１６、更新部１７と同様の機能を発揮するデータ順序保証プログラム３２０ａが予め記憶される。あるいは、ＲＯＭ３２０には、上記実施例２、実施例３の実行部４３ｃ、停止部４３ｄ、取得部４３ｅ、比較部４３ｆ、更新部４３ｇ、特定部４３と同様の機能を発揮するデータ順序保証プログラム３２０ａが予め記憶される。なお、データ順序保証プログラム３２０ａについては、適宜分離しても良い。
【００９０】
そして、ＣＰＵ３１０が、データ順序保証プログラム３２０ａをＲＯＭ３２０から読み出して実行することで、実施例１〜３の各制御部と同様の動作を実行する。すなわち、データ順序保証プログラム３２０ａは、上記実施例１の停止部１４、取得部１５、比較部１６、更新部１７、または上記実施例２、３の実行部４３ｃ、停止部４３ｄ、取得部４３ｅ、比較部４３ｆ、更新部４３ｇ、特定部４３と同様の動作を実行する。
【００９１】
なお、上記したデータ順序保証プログラム３２０ａについては、必ずしも最初からＲＯＭ３２０に記憶させることを要しない。データ順序保証プログラム３２０ａはＨＤＤ３３０に記憶させてもよい。
【００９２】
例えば、コンピュータ３００に挿入されるフレキシブルディスク（ＦＤ）、ＣＤ−ＲＯＭ、ＤＶＤディスク、光磁気ディスク、ＩＣカードなどの「可搬用の物理媒体」にプログラムを記憶させておく。そして、コンピュータ３００がこれらからプログラムを読み出して実行するようにしてもよい。
【００９３】
さらには、公衆回線、インターネット、ＬＡＮ、ＷＡＮなどを介してコンピュータ３００に接続される「他のコンピュータ（またはサーバ）」などにプログラムを記憶させておく。そして、コンピュータ３００がこれらからプログラムを読み出して実行するようにしてもよい。
【符号の説明】
【００９４】
１０    サーバ装置
１１    ノード
１２    クエリ
１３    ルーティングテーブル
１４    停止部
１５    取得部
１６    比較部
１７    更新部
２０    ＣＥＰシステム
２１    サーバ装置
３０    サーバ装置
３１    ＶＭ
３２    データストリーム処理エンジン
３３    クエリ
４０    管理サーバ
４３    制御部
４３ａ  判定部
４３ｂ  移動部
４３ｃ  実行部
４３ｄ  停止部
４３ｅ  取得部
４３ｆ  比較部
４３ｇ  更新部
５０    ルーティングテーブル
５１    最大期間情報
５２    停止期間テーブル
【図１】
0005821578.tif 000002
【図２】
0005821578.tif 000003
【図３】
0005821578.tif 000004
【図４Ａ】
0005821578.tif 000005
【図４Ｂ】
0005821578.tif 000006
【図５】
0005821578.tif 000007
【図６】
0005821578.tif 000008
【図７】
0005821578.tif 000009
【図８】
0005821578.tif 000010
【図９】
0005821578.tif 000011
【図１０】
0005821578.tif 000012
【図１１】
0005821578.tif 000013
【図１２】
0005821578.tif 000014
【図１３】
0005821578.tif 000015
【図１４】
0005821578.tif 000016
【図１５】
0005821578.tif 000017