(57)【要約】【課題】待機時間を抑制して処理時間を短縮させることが可能な分散同期処理システムを提供する。【解決手段】分散同期処理システムは、仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、分散処理部間で同期して計算処理を行うシステムであって、処理サーバ３０は、自身の分散処理部が他の処理サーバ上の分散処理部との間で計算結果の待ち状態となり、かつ、他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、他の処理サーバに遅れメッセージを送信する遅れ制御手段３２２と、遅れメッセージに対して、他の処理サーバから待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ３０上にレプリカを構築して計算処理を行わせるレプリカ制御手段３２３と、を備える。【選択図】図６
2019035997.tif 000002
【特許請求の範囲】
【請求項１】
仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、予め定めたグラフトポリジで、前記分散処理部間で同期して計算処理を行う分散同期処理システムであって、
前記処理サーバは、
リソースを管理するリソース管理手段と、
前記分散処理部間で計算結果を含む計算完了メッセージを送受信するデータ転送手段と、
自身の処理サーバ上の分散処理部が他の処理サーバ上の分散処理部との間で前記計算結果の待ち状態となり、かつ、前記リソース管理手段が管理するリソースに前記他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、前記他の処理サーバに処理の遅れを示す遅れメッセージを送信する遅れ制御手段と、
前記遅れメッセージに対して、前記他の処理サーバから前記待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ上に前記レプリカを構築して計算処理を行わせるレプリカ制御手段と、を備え、
前記遅れ制御手段は、前記遅れメッセージを他の処理サーバから受信した場合に、前記レプリカの構築を指示するレプリカ制御メッセージを返信すること
を特徴とする分散同期処理システム。
【請求項２】
前記データ転送手段は、
自身の処理サーバ上の前記レプリカが計算を完了した段階で、前記レプリカのオリジナルである分散処理部が動作する他の処理サーバに、前記レプリカの計算結果を含む前記計算完了メッセージを送信し、
前記他の処理サーバから、前記レプリカの計算結果を含む前記計算完了メッセージを受信した段階で、前記オリジナルの分散処理部の計算を終了させ、前記オリジナルの分散処理部の計算結果の出力先に前記計算完了メッセージを送信すること
を特徴とする請求項１に記載の分散同期処理システム。
【請求項３】
オリジナルの分散処理部に対するレプリカが構築されている状態を管理するレプリカ管理手段を備え、
前記データ転送手段は、自身の処理サーバ上の分散処理部から他の分散処理部に前記計算結果を送信する段階で、前記レプリカ管理手段において、前記自身の処理サーバ上の分散処理部に対するレプリカが構築されている状態である場合に、前記レプリカを構築している他の処理サーバに前記レプリカの動作の中止を指示するレプリカ制御メッセージを送信し、
前記レプリカ制御手段は、他の処理サーバから、前記レプリカの動作の中止を指示するレプリカ制御メッセージを受信した段階で、前記レプリカの動作を中止させること
を特徴とする請求項１または請求項２に記載の分散同期処理システム。
【請求項４】
前記処理サーバは、自身の処理サーバ上で動作する分散処理部を、自身の処理サーバ上で動作するレプリカよりも優先して動作させること
を特徴とする請求項１から請求項３のいずれか一項に記載の分散同期処理システム。
【請求項５】
仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、予め定めたグラフトポリジで、前記分散処理部間で同期して計算処理を行う分散同期処理方法であって、
前記処理サーバは、
自身の処理サーバ上の分散処理部が他の処理サーバ上の分散処理部との間で計算結果の待ち状態となり、かつ、リソース管理手段が管理するリソースに前記他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、前記他の処理サーバに処理の遅れを示す遅れメッセージを送信するステップと、
前記遅れメッセージに対して、前記他の処理サーバから前記待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ上に前記レプリカを構築して計算処理を行わせるステップと、を実行し、
前記遅れメッセージを他の処理サーバから受信した場合に、前記レプリカの構築を指示するレプリカ制御メッセージを返信するステップを実行すること
を特徴とする分散同期処理方法。
【請求項６】
仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、予め定めたグラフトポリジで、前記分散処理部間で同期して計算処理を行う分散同期処理システムにおける前記処理サーバのコンピュータを、
リソースを管理するリソース管理手段、
前記分散処理部間で計算結果を含む計算完了メッセージを送受信するデータ転送手段、
自身の処理サーバ上の分散処理部が他の処理サーバ上の分散処理部との間で前記計算結果の待ち状態となり、かつ、前記リソース管理手段が管理するリソースに前記他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、前記他の処理サーバに処理の遅れを示す遅れメッセージを送信する遅れ制御手段、
前記遅れメッセージに対して、前記他の処理サーバから前記待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ上に前記レプリカを構築して計算処理を行わせるレプリカ制御手段、として機能させるための分散同期処理プログラムであり、
前記遅れ制御手段は、前記遅れメッセージを他の処理サーバから受信した場合に、前記レプリカの構築を指示するレプリカ制御メッセージを返信すること
を特徴とする分散同期処理プログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、分散配置された複数のサーバを同期させて処理を実行する分散同期処理システム、分散同期処理方法および分散同期処理プログラムに関する。
【背景技術】
【０００２】
ネットワーク上に複数のサーバを分散配置する分散処理システムのフレームワークとして、非特許文献１にはＭａｐＲｅｄｕｃｅが開示されている。ただし、このＭａｐＲｅｄｕｃｅは、処理の度に、外部のデータストアからの入力データの読み込みや、結果の書き出し処理が必要であるため、ある処理の結果を次の処理で利用するようなイテレーティブな（反復する）処理には向いていない。この種の処理には、非特許文献２に開示されているＢＳＰ（Bulk Synchronous Parallel：バルク同期並列）が適している。
【０００３】
このＢＳＰは、「スーパーステップ（ＳＳ：superstep)」という処理単位を繰り返し実行することにより、分散環境でのデータ処理を実行する。図１８は、ＢＳＰ計算モデルを説明するための図である。
１つのスーパーステップは、図１８に示すように、３つのフェーズ（ＰＨ：phase）として、「ローカル計算（ＬＣ：Local computation）」（フェーズＰＨ１）、「データ交換（Ｃｏｍ：Communication）」（フェーズＰＨ２）、「同期（Sync）」（フェーズＰＨ３）により構成される。
【０００４】
具体的には、複数のノード（ノード１〜ノード４）のうちのいずれかのノードがデータを受信すると、そのノード（例えば、ノード１）がフェーズＰＨ１において、そのデータについての計算処理（ローカル計算〔ＬＣ〕）を実行する。続いて、フェーズＰＨ２において、各ノードが保持しているローカル計算の結果であるデータについて、ノード間でのデータ交換を実行する。次に、フェーズＰＨ３において、同期処理を行う、より詳細には、すべてのノード間でのデータ交換の終了を待つ。
そして、スーパーステップＳＳ１として、一連のスーパーステップの処理（ＰＨ１〜ＰＨ３）が終了すると、各ノードはその計算結果を保持した上で、次の一連の処理であるスーパーステップＳＳ２へと進む。以下、同様にして、複数のスーパーステップが繰り返される。
【０００５】
従来、分散同期処理システムのフレームワークでは、マスタ（master）／ワーカ（worker）モデルを採用しており、システムを管理する管理サーバ（マスタ）が、対象とする計算処理の全体を所定単位に細分化した個々の計算処理を処理サーバ（ワーカ）に割り振ることとしている。
非特許文献３には、ＢＳＰを採用した分散同期処理システムのフレームワークとして、Ｐｒｅｇｅｌが開示されている。このＰｒｅｇｅｌ等のフレームワークでは、全体の処理をグラフＧ＝（Ｖ，Ｅ）として表現し、これをＢＳＰに適用して実行する。ここで、Ｖは「バーテックス（vertex：頂点）の集合」であり、各頂点は、細分化された個々の処理内容に対応する。また、Ｅは「エッジ（edge：辺）の集合」であり、有向辺は、各頂点間の情報伝達を行う経路に対応する。
【０００６】
ここで、図２０を参照して、従来の分散同期処理システムにおける処理の流れについて説明する。なお、計算対象は、図１９に示すグラフトポロジ（グラフＧ）であることとする。また、図２０に示すように、分散同期処理システム１００Ａが２台のworker（worker１，worker２）で構成され、頂点（vertex）Ｖ１〜Ｖ６のうち、頂点（vertex）Ｖ１〜Ｖ３をworker１が担当し、頂点（vertex）Ｖ４〜Ｖ６をworker２が担当するものとする。以下、全体の処理の流れを通して説明する。
【０００７】
まず、各worker（worker１，worker２）は、masterからの指示で、担当する頂点（vertex）のスーパーステップを実行する（ステップＳ１００）。具体的には、フェーズＰＨ１のローカル計算を実行し、スーパーステップの処理を開始する。
次に、各workerは、自身が担当する頂点（vertex）の処理の進行を監視し、各頂点（vertex）が、フェーズＰＨ２のデータ交換まで完了したか否かを判定する。そして、各workerは、担当するすべての頂点（vertex）が、フェーズＰＨ２までの処理を完了したと確認した場合に、その完了情報をmasterに報告する（ステップＳ１０１，Ｓ１０２）。
【０００８】
続いて、masterは、すべてのworker（ここでは、worker１，worker２）から完了情報を受信した場合に、スーパーステップを「＋１」更新し、各頂点（vertex）に次のスーパーステップの実行を指示する（ステップＳ１０３）。
そして、masterおよび各workerは、ステップＳ１００〜Ｓ１０３の動作を繰り返す。
【０００９】
従来の分散同期処理システム１００Ａは、スーパーステップごとに、計算対象となるすべての頂点（vertex）で同期を行う。具体的には、図２０に示す全体同期ポイントにおいて同期を行うため、最も遅い頂点（vertex）の処理後に同期が完了することになる。例えば、図２０のスーパーステップＳＳ１では、頂点（vertex）Ｖ１〜Ｖ６のうち、最も遅い頂点（vertex）Ｖ２の処理後に同期が完了する。また、スーパーステップＳＳ２では、最も遅い頂点（vertex）Ｖ６の処理後に同期が完了する。よって、著しく処理が遅い頂点（vertex）があると、その頂点（vertex）の処理後に同期が完了するため、頂点（vertex）の処理全体が著しく遅延してしまう。
【００１０】
このような問題を解決するため非特許文献４の分散同期処理システムでは、従来の全頂点（vertex）での同期を行わず、頂点（vertex）ごとに次のスーパーステップへの移行を判断している。具体的には、非特許文献４の分散同期処理システムでは、「自頂点（vertex）およびグラフトポロジの入力辺で接するすべての頂点（vertex）の計算・交換処理（ローカル計算〔フェーズＰＨ１〕およびデータ交換〔フェーズＰＨ２〕）が完了していること」（隣接同期）を満たしている場合に、各頂点（vertex）が次のスーパーステップに移行するものとしている。
【００１１】
図２１は、図２０で説明した従来の分散同期処理システム１００Ａが実行する処理の流れ（図２１（ａ）参照）と、非特許文献４に記載された従来の分散同期処理システム１００Ｂが実行する処理の流れ（図２１（ｂ）参照）とを示す図である。
従来の分散同期処理システム１００Ｂでは、前記したように、「自頂点（vertex）および入力辺で接するすべての頂点（vertex）の計算・交換処理（ローカル計算〔フェーズＰＨ１〕およびデータ交換〔フェーズＰＨ２〕）が完了していること」（隣接同期）により、次のスーパーステップに移行する。
【００１２】
例えば、図２１（ｂ）の頂点（vertex）Ｖ４に着目すると、頂点（vertex）Ｖ４は、入力辺で接する頂点（vertex）Ｖ５，Ｖ６の計算・交換処理と自身の計算・交換処理が終わった時点が隣接同期ポイントとなる。
ここで、頂点（vertex）Ｖ４は、スーパーステップＳＳ１のとき、自身の計算・交換処理が終わった時点では、頂点（vertex）Ｖ４の計算・交換処理は終わっているが、頂点（vertex）Ｖ５，Ｖ６の計算・交換処理が終わっていないため、待機状態となり、頂点（vertex）Ｖ５，Ｖ６の計算・交換処理が終わった時点が隣接同期ポイントとなる。
【００１３】
また、頂点（vertex）Ｖ１に着目すると、頂点（vertex）Ｖ１は、入力辺で接する頂点（vertex）は存在しない。よって、スーパーステップＳＳ１のとき、自頂点（vertex）の計算・送信処理が終了した時点が隣接同期ポイントとなる。
すなわち、分散同期処理システム１００Ｂは、すべての頂点（vertex）の計算・交換処理が完了していなくても、隣接同期ポイントで次のスーパーステップに移行することができる。
このように、分散同期処理システム１００Ｂは、分散同期処理システム１００Ａに比べて、フェーズＰＨ３の同期完了までの時間を大幅に削減することができるため、システム全体としての処理速度や各workerの利用効率を改善することができる。
【先行技術文献】
【特許文献】
【００１４】
【非特許文献１】Dean, J., et al., ”MapReduce: Simplified Data Processing on Large Clusters,” OSDI '04, 2004, p.137-149
【非特許文献２】Valiant, L., et al., ”A bridging model for parallel computation,” Communications of the ACM, 1990, vol.33, No.8, p.103-111
【非特許文献３】Malewicz, G., et al., ”Pregel: A System for Large-Scale Graph Processing,” Proc. of ACM SIGMOD, 2010, p.136-145
【非特許文献４】小林弘明，岡本光浩，「分散処理フレームワークの同期処理に関する一検討」，電子情報通信学会／北海道大学工学部大学院情報科学研究科 共催，2016年ソサイエティ大会講演論文集，Ｂ−７−１，2016年9月20日
【発明の概要】
【発明が解決しようとする課題】
【００１５】
従来のＢＰＳ（バルク同期並列）等のように、分散環境で計算処理を並行して行い、定期的に同期させる分散同期処理システムでは、計算処理（データ量、ＣＰＵ負荷等）が常に変動するため、予め計算処理がサーバ間で均等になるように処理を分散することは困難である。
【００１６】
ここで、図２２を参照して、従来の問題点について説明する。なお、計算対象は、図２２に示すグラフトポロジ（グラフＧ）であることとする。また、図２２に示すように、分散同期処理システムが２台のworker（worker１，worker２）で構成され、頂点（vertex）Ｖ１〜Ｖ３のうち、頂点（vertex）Ｖ１，Ｖ２をworker１が担当し、頂点（vertex）Ｖ３をworker２が担当するものとする。
また、ここでは、説明を簡略化するため、頂点（vertex）Ｖ１〜Ｖ３の各動作ステップは、ＣＰＵの１コア（ＣＯＲＥ）で数字の順に動作するものとする。
【００１７】
図２２に示すように、worker１は、スーパーステップＳＳ１として、頂点（vertex）Ｖ１，Ｖ２において、１〜１２ステップで処理を実行する。一方、worker２は、スーパーステップＳＳ１として、頂点（vertex）Ｖ３において、１〜３ステップで処理を実行する。
ここで、worker１は、スーパーステップＳＳ１の処理が完了した後、頂点（vertex）Ｖ１，Ｖ２において、１３ステップ目でスーパーステップＳＳ２の処理を開始する。
一方、worker２は、スーパーステップＳＳ１の処理が完了した後、頂点（vertex）Ｖ３において、スーパーステップＳＳ２の処理を開始する。このとき、worker２は、スーパーステップＳＳ２の処理を開始するためには、頂点（vertex）Ｖ３において、worker１の頂点（vertex）Ｖ２の計算完了まで待機しなければならない。
【００１８】
このとき、worker１の頂点（vertex）Ｖ２の処理が完了するのは、８ステップ目である。そのため、worker２は、頂点（vertex）Ｖ３において、スーパーステップＳＳ２の処理を開始するのは９ステップ目となる。すなわち、worker２は、４〜８ステップの間、頂点（vertex）Ｖ２の処理の待機状態となり、待機時間が発生してしまう。
【００１９】
本発明は、このような問題に鑑みてなされたもので、分散同期処理を行うシステムにおいて、待機時間を抑制し、システム全体としての処理時間を短縮させることが可能な分散同期処理システム、分散同期処理方法および分散同期処理プログラムを提供することを課題とする。
【課題を解決するための手段】
【００２０】
前記課題を解決するため、請求項１に記載の発明は、仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、予め定めたグラフトポリジで、前記分散処理部間で同期して計算処理を行う分散同期処理システムであって、前記処理サーバは、リソースを管理するリソース管理手段と、前記分散処理部間で計算結果を含む計算完了メッセージを送受信するデータ転送手段と、自身の処理サーバ上の分散処理部が他の処理サーバ上の分散処理部との間で前記計算結果の待ち状態となり、かつ、前記リソース管理手段が管理するリソースに前記他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、前記他の処理サーバに処理の遅れを示す遅れメッセージを送信する遅れ制御手段と、前記遅れメッセージに対して、前記他の処理サーバから前記待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ上に前記レプリカを構築して計算処理を行わせるレプリカ制御手段と、を備え、前記遅れ制御手段は、前記遅れメッセージを他の処理サーバから受信した場合に、前記レプリカの構築を指示するレプリカ制御メッセージを返信することを特徴とする。
【００２１】
また、前記課題を解決するため、請求項５に記載の発明は、仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、予め定めたグラフトポリジで、前記分散処理部間で同期して計算処理を行う分散同期処理方法であって、前記処理サーバは、自身の処理サーバ上の分散処理部が他の処理サーバ上の分散処理部との間で計算結果の待ち状態となり、かつ、リソース管理手段が管理するリソースに前記他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、前記他の処理サーバに処理の遅れを示す遅れメッセージを送信するステップと、前記遅れメッセージに対して、前記他の処理サーバから前記待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ上に前記レプリカを構築して計算処理を行わせるステップと、を実行し、前記遅れメッセージを他の処理サーバから受信した場合に、前記レプリカの構築を指示するレプリカ制御メッセージを返信するステップを実行することを特徴とする。
【００２２】
また、前記課題を解決するため、請求項６に記載の発明は、仮想マシンである分散処理部を一または複数動作させる処理サーバを複数接続し、予め定めたグラフトポリジで、前記分散処理部間で同期して計算処理を行う分散同期処理システムにおける前記処理サーバのコンピュータを、リソースを管理するリソース管理手段、前記分散処理部間で計算結果を含む計算完了メッセージを送受信するデータ転送手段、自身の処理サーバ上の分散処理部が他の処理サーバ上の分散処理部との間で前記計算結果の待ち状態となり、かつ、前記リソース管理手段が管理するリソースに前記他の処理サーバ上の分散処理部を動作させるリソースが存在する場合に、前記他の処理サーバに処理の遅れを示す遅れメッセージを送信する遅れ制御手段、前記遅れメッセージに対して、前記他の処理サーバから前記待ち状態を発生させた分散処理部のレプリカの構築を指示するレプリカ制御メッセージを受信した段階で、自身の処理サーバ上に前記レプリカを構築して計算処理を行わせるレプリカ制御手段、として機能させるための分散同期処理プログラムであり、前記遅れ制御手段は、前記遅れメッセージを他の処理サーバから受信した場合に、前記レプリカの構築を指示するレプリカ制御メッセージを返信することを特徴とする。
【００２３】
請求項１，５，６に記載の発明によれば、分散同期処理システムは、同期を待ち、かつ、空きリソースがある処理サーバが、他の処理サーバの分散処理部の動作をレプリカにより並行して開始することができる。これによって、本発明は、各処理サーバのリソースを効率よく利用することができ、システム全体として、計算処理の処理時間を短縮させることができる。
【００２４】
また、請求項２に記載の発明は、請求項１に記載の分散同期処理システムにおいて、前記データ転送手段が、自身の処理サーバ上の前記レプリカが計算を完了した段階で、前記レプリカのオリジナルである分散処理部が動作する他の処理サーバに、前記レプリカの計算結果を含む前記計算完了メッセージを送信し、前記他の処理サーバから、前記レプリカの計算結果を含む前記計算完了メッセージを受信した段階で、前記オリジナルの分散処理部の計算を終了させ、前記オリジナルの分散処理部の計算結果の出力先に前記計算完了メッセージを送信することを特徴とする。
【００２５】
請求項２に記載の発明によれば、レプリカがオリジナルの分散処理部よりも早く計算処理を終了した段階で、レプリカの計算処理結果を採用し、オリジナルの分散処理部の計算処理を中止させることができる。これによって、本発明は、不要になった計算処理を早期に中止させて、システム全体の効率化を図ることができる。
【００２６】
また、請求項３に記載の発明は、請求項１または請求項２に記載の分散同期処理システムにおいて、オリジナルの分散処理部に対するレプリカが構築されている状態を管理するレプリカ管理手段を備え、前記データ転送手段は、自身の処理サーバ上の分散処理部から他の分散処理部に前記計算結果を送信する段階で、前記レプリカ管理手段において、前記自身の処理サーバ上の分散処理部に対するレプリカが構築されている状態である場合に、前記レプリカを構築している他の処理サーバに前記レプリカの動作の中止を指示するレプリカ制御メッセージを送信し、前記レプリカ制御手段は、他の処理サーバから、前記レプリカの動作の中止を指示するレプリカ制御メッセージを受信した段階で、前記レプリカの動作を中止させることを特徴とする。
【００２７】
請求項３に記載の発明によれば、レプリカよりもオリジナルの分散処理部が早く計算処理を終了した段階で、レプリカの計算処理を中止させることができる。これによって、本発明は、不要になった計算処理を早期に中止させて、システム全体の効率化を図ることができる。
【００２８】
また、請求項４に記載の発明は、請求項１から請求項３のいずれか一項に記載の分散同期処理システムにおいて、前記処理サーバは、自身の処理サーバ上で動作する分散処理部を、自身の処理サーバ上で動作するレプリカよりも優先して動作させることを特徴とする。
【００２９】
請求項４に記載の発明によれば、処理サーバ上で動作する分散処理部の動作を、本来他の処理サーバで動作する分散処理部のレプリカよりも優先して動作させることで、オリジナルの分散処理部の処理速度への影響を抑えることができる。
【発明の効果】
【００３０】
本発明によれば、分散同期処理を行う際に、同期を待ち、かつ、空きリソースがある処理サーバが、他の処理サーバの分散処理部の動作を投機的に実行することができる。
これによって、本発明は、空きリソースがある処理サーバを有効に活用し、待ち状態となっている処理サーバの待機時間を短縮し、システム全体として、同期処理計算の高速化を図ることができる。
【図面の簡単な説明】
【００３１】
【図１】本発明の実施形態に係る分散同期処理システムの処理概要を説明するための図（１／２）である。
【図２】本発明の実施形態に係る分散同期処理システムの処理概要を説明するための図（２／２）である。
【図３】本発明の実施形態に係る分散同期処理システムの全体構成を示す図である。
【図４】本発明の実施形態に係る分散同期処理システムの管理サーバの構成を示す図である。
【図５】本発明の実施形態に係る分散同期処理システムの分散処理部の構成を示す図である。
【図６】本発明の実施形態に係る分散同期処理システムの処理サーバの構成を示す図である。
【図７】本発明の実施形態に係る分散同期処理システムの全体動作を示すフローチャートである。
【図８】本発明の実施形態に係る分散同期処理システムのデータ転送処理の動作を示すフローチャートである。
【図９】本発明の実施形態に係る分散同期処理システムの遅れ制御の動作を示すフローチャートである。
【図１０】本発明の実施形態に係る分散同期処理システムのレプリカ制御の動作を示すフローチャートである。
【図１１】本発明の実施形態に係る分散同期処理システムの適用例の計算対象のグラフトポロジを例示する図である。
【図１２】本発明の実施形態に係る分散同期処理システムの適用例の処理手順を説明するための図（１／６）である。
【図１３】本発明の実施形態に係る分散同期処理システムの適用例の処理手順を説明するための図（２／６）である。
【図１４】本発明の実施形態に係る分散同期処理システムの適用例の処理手順を説明するための図（３／６）である。
【図１５】本発明の実施形態に係る分散同期処理システムの適用例の処理手順を説明するための図（４／６）である。
【図１６】本発明の実施形態に係る分散同期処理システムの適用例の処理手順を説明するための図（５／６）である。
【図１７】本発明の実施形態に係る分散同期処理システムの適用例の処理手順を説明するための図（６／６）である。
【図１８】ＢＳＰ計算モデルを説明するための説明図である。
【図１９】ＢＳＰ計算モデルにおける計算対象のグラフトポロジを例示する図である。
【図２０】従来の分散同期処理システムの処理の流れを説明するための図である。
【図２１】従来の２つ分散同期処理システムの同期ポイントの相違を説明するための図である。
【図２２】従来の分散同期処理システムの問題点を説明するための図である。
【発明を実施するための形態】
【００３２】
本発明の実施形態について図面を参照して説明する。
≪分散同期処理システムの処理概要≫
まず、図１，図２を参照して、本発明の実施形態に係る分散同期処理システム１の処理概要について説明する。なお、計算対象は、図１に示すグラフトポロジ（グラフＧ）であることとする。また、図１に示すように、分散同期処理システム１が２台のworker（worker１，worker２〔処理サーバ〕）で構成され、頂点（vertex）Ｖ１〜Ｖ３（分散処理部）のうち、頂点（vertex）Ｖ１，Ｖ２をworker１が担当し、頂点（vertex）Ｖ３をworker２が担当するものとする。
また、ここでは、説明を簡略化するため、頂点（vertex）Ｖ１〜Ｖ３の各動作ステップは、ＣＰＵの１コア（ＣＯＲＥ）で数字の順に動作するものとする。
【００３３】
図１に示すように、worker１は、スーパーステップＳＳ１として、頂点（vertex）Ｖ１，Ｖ２において、１，２，…の各ステップで処理を実行する。また、worker２は、スーパーステップＳＳ１として、頂点（vertex）Ｖ３において、１〜３ステップで処理を実行する。ここで、worker２は、スーパーステップＳＳ２に移行するには、まだ完了していないworker１の頂点（vertex）Ｖ２の計算処理の結果が必要となる。
【００３４】
そこで、worker２は、リソースに空きがある場合、worker１の頂点（vertex）Ｖ２の複製であるレプリカＲをworker２上に構築し、レプリカＲにおいて、頂点（vertex）Ｖ２と同じ処理を、４〜７ステップで実行する。
図１の場合、worker１は、７ステップでは、オリジナルである頂点（vertex）Ｖ２の処理が完了していない。そこで、オリジナルである頂点（vertex）Ｖ２は、自身の計算途中のデータを、レプリカＲの計算結果であるデータに置換する。
これによって、worker１は、頂点（vertex）Ｖ２の計算処理を早く終了させ、システム全体として計算時間を短縮させることができる。
【００３５】
なお、レプリカＲの計算処理が、オリジナルの頂点（vertex）Ｖ２の計算処理よりも時間がかかった場合、図２に示すように、worker１は、worker２に対して、レプリカＲの計算処理を中止させる。
図２の場合、worker１は、４ステップで頂点（vertex）Ｖ２の処理が完了しているが、レプリカＲは処理を完了していない。そこで、オリジナルである頂点（vertex）Ｖ２は、レプリカＲの計算を中止させる。
これによって、必要以上にシステムに負荷をかけることを防止することができる。
以下、この処理概要で説明した分散同期処理システムを実現するための分散同期処理システムの構成および動作について詳細に説明する。
【００３６】
≪分散同期処理システムの構成≫
図３に示すように、分散同期処理システム１は、管理サーバ１０と、管理サーバ１０にそれぞれ接続され並列に処理を行う複数の処理サーバ３０と、を備える。なお、処理サーバ３０上では、一または複数の分散処理部２０（仮想マシン）が動作する。
【００３７】
管理サーバ１０および処理サーバ３０は、ＣＰＵ（Central Processing Unit）、ＲＡＭ（Random Access Memory）、ＲＯＭ（Read Only Memory）、ＨＤＤ（Hard Disk Drive）等、一般的なコンピュータとしてのハードウエアを備えており、ＨＤＤには、ＯＳ（Operating System）、プログラム、各種データ等が格納されている。ＯＳおよびアプリケーションプログラムは、ＲＡＭに展開され、ＣＰＵによって実行される。なお、図４〜図６において、管理サーバ１０、分散処理部２０および処理サーバ３０の内部は、ＲＡＭに展開されたプログラム（管理プログラム、分散処理プログラムおよび分散同期処理プログラム）によって実現される機能を、ブロックとして示している。
【００３８】
管理サーバ１０（master）は、システム全体を管理するものである。管理サーバ１０は、図４に示すように分散配置手段１１を備える。
分散配置手段１１は、対象とする計算処理に必要な複数の分散処理部２０を複数の処理サーバ３０に対して割り当てるものである。
この分散配置手段１１は、設定された計算対象となる各分散処理部２０（頂点〔vertex〕）の接続関係を示すグラフトポロジに基づいて、処理サーバ３０に対して、仮想マシンである分散処理部２０を配置する。ここで、分散処理部２０を配置するとは、仮想マシンを構成するソフトウェアを処理サーバ３０に送信し、処理サーバ３０上で仮想マシンとして分散処理部２０を動作させることである。
また、分散配置手段１１は、各処理サーバ３０に対して、分散処理部２０の接続関係を示すグラフトポロジや、分散処理部２０が必要とするリソース（ＣＰＵのコア数、メモリ量等）を通知する。
なお、ここでは、図示を省略するが、管理サーバ１０は、外部からの指示により、システム全体の処理を終了させる終了指示メッセージを、処理サーバ３０に送信する終了指示部を備えることとしてもよい。
【００３９】
分散処理部２０（頂点〔vertex〕）は、処理サーバ３０上で動作し、分散された計算処理を実行する仮想マシンである。個々の計算処理には、データ入力、計算、メッセージの送受信等が含まれる。この分散処理部２０は、対象とする計算処理をグラフＧ＝（Ｖ，Ｅ）として表現したときに、グラフ中の個々の頂点（vertex）として機能する。
分散処理部２０は、図５に示すように、数値計算手段２１と、メッセージ送受信手段２２と、を備える。
【００４０】
数値計算手段２１は、所定単位に区分された計算処理を行うものである。この処理は、図１９を参照して説明したＢＳＰ計算モデルのフェーズＰＨ１としてのローカル計算（ＬＣ：Local computation）に相当する。
この数値計算手段２１は、メッセージ送受信手段２２を介して他の分散処理部２０から通知される計算完了メッセージに設定されている計算結果を入力として計算処理を実行し、次のスーパーステップに移行する。なお、数値計算手段２１は、計算処理が他の分散処理部２０からの計算結果を用いない処理である場合（グラフの入力辺がない場合）、入力を待たずに順次計算処理を実行し、次のスーパーステップに移行する。
【００４１】
また、数値計算手段２１は、計算処理がその計算結果を他の分散処理部２０に通知する処理である場合（グラフの出力辺がある場合）、計算結果を計算完了メッセージに設定してメッセージ送受信手段２２に送信する。また、数値計算手段２１は、計算処理がその結果を他の分散処理部２０に通知しない処理である場合（グラフの出力辺がない場合）、計算完了メッセージを送信しない。
【００４２】
メッセージ送受信手段２２は、他の分散処理部２０との間でメッセージを送受信するものである。この処理は、図１８を参照して説明したＢＳＰ計算モデルのフェーズＰＨ２としてのデータ交換（ＣＯＭ：Communication）に相当する。
このメッセージ送受信手段２２は、処理サーバ３０のメッセージ処理手段３２（図６参照）を介して、同じ処理サーバ３０内の他の分散処理部２０、または、他の処理サーバ３０内の分散処理部２０との間でメッセージを送受信する。
【００４３】
なお、メッセージ送受信手段２２は、他の分散処理部２０から通知される計算完了メッセージが揃った段階、すなわち、グラフの入力辺に対応するすべての分散処理部２０から同じスーパーステップの計算完了メッセージを受信した段階で、数値計算手段２１にすべての計算完了メッセージの計算結果を出力する。
また、メッセージ送受信手段２２は、自身の分散処理部２０の数値計算手段２１が他の分散処理部２０からの計算結果を必要とする処理であって、予め定めた時間内に計算完了メッセージを受信しない場合、すなわち、待ち状態となった場合、計算完了メッセージの送信が遅れている他の分散処理部２０を特定する情報（vertex番号等）を設定した遅れメッセージをメッセージ処理手段３２に送信する。
【００４４】
処理サーバ３０（worker）は、個々の計算処理にそれぞれ対応した一または複数の分散処理部２０を動作させるものである。
処理サーバ３０は、図６に示すように、仮想化制御手段３１と、メッセージ処理手段３２と、リソース管理手段３３と、レプリカ管理手段３４と、記憶手段３５と、を備える。
【００４５】
仮想化制御手段３１は、仮想化技術に基づいて、処理サーバ３０上に仮想化プラットホームを構築し、複数の分散処理部２０（仮想マシン）を配置する制御を行うものである。
この仮想化制御手段３１は、管理サーバ１０から入力される仮想マシンを構成するソフトウェアを入力し、仮想化プラットホーム上で動作させる。
また、仮想化制御手段３１は、レプリカ管理手段３４から、他の処理サーバ３０上で動作する分散処理部２０のレプリカ（複製）の構築を指示された場合、管理サーバ１０からソフトウェアを取得し、仮想化プラットホーム上で動作させる。なお、分散処理部２０の接続関係を示すグラフトポロジにより、レプリカとして動作させる可能性がある分散処理部２０については、予め管理サーバ１０からソフトウェアを取得し、記憶手段３５に記憶させておくこととしてもよい。
また、仮想化制御手段３１は、レプリカ管理手段３４から、レプリカの動作を中止する指示があった場合、レプリカの動作を中止し、仮想化プラットホームから廃棄する。
【００４６】
メッセージ処理手段３２は、メッセージの送受信を行うものである。
ここでは、メッセージ処理手段３２は、データ転送手段３２１と、遅れ制御手段３２２と、レプリカ制御手段３２３と、を備える。
【００４７】
データ転送手段３２１は、記憶手段３５に記憶されている各分散処理部２０（頂点〔vertex〕）の接続関係を示すグラフトポロジに基づいて、分散処理部間で計算結果を含む計算完了メッセージを送受信するものである。
このデータ転送手段３２１は、自身または他の処理サーバ３０上で動作する分散処理部２０から受信した計算完了メッセージを、グラフトポロジの送信先（出力辺の相手先）である自身または他の処理サーバ３０上で動作する分散処理部２０に送信する。
このとき、データ転送手段３２１は、自身の処理サーバ３０上の分散処理部２０から計算完了メッセージを受信した場合、当該分散処理部２０のレプリカが他の処理サーバ３０で動作していれば、他の処理サーバ３０にレプリカの動作を中止する指示を含んだレプリカ制御メッセージを送信する。これは、レプリカよりもオリジナルの分散処理部２０の方が早く計算処理を完了し、レプリカの計算処理が不要になったためである。なお、レプリカが動作している否かの情報は、レプリカ管理手段３４から取得する。また、レプリカ制御メッセージには、レプリカを特定する情報（オリジナルの分散処理部２０のvertex番号等）を付加することとする。
【００４８】
また、データ転送手段３２１は、他の処理サーバ３０上のレプリカから計算完了メッセージを受信した場合、自身の処理サーバ３０上のレプリカに対応するオリジナルの分散処理部２０において現時点でのスーパーステップの計算処理を終了させ、オリジナルの分散処理部２０の計算結果の代わりに、受信した計算完了メッセージの計算結果に置き換えた計算完了メッセージ生成し、送信先に送信する。
また、データ転送手段３２１は、自身の処理サーバ３０上のレプリカから計算完了メッセージを受信した場合、レプリカ管理手段３４を介してレプリカを廃棄し、計算完了メッセージをオリジナルの分散処理部２０が動作する処理サーバ３０に送信する。
【００４９】
遅れ制御手段３２２は、自身または他の処理サーバ３０から受信した遅れメッセージに対する処理を行うものである。
この遅れ制御手段３２２は、自身の処理サーバ３０の分散処理部２０から、遅れメッセージを受信した場合、すなわち、分散処理部２０が計算結果の待ち状態となった場合、当該分散処理部２０を計算結果の出力先とする他の分散処理部２０のレプリカを構築するリソースがあれば、他の分散処理部２０を動作させている処理サーバ３０に遅れメッセージを送信する。なお、この遅れメッセージには、遅れの対象となっている分散処理部２０を特定する情報（vertex番号等）を付加することとする。
【００５０】
これによって、自身の処理サーバ３０でレプリカを動作させることが可能であることを他の処理サーバ３０に通知することができる。なお、レプリカを構築するリソースがあるか否かは、リソース管理手段３３に対して問い合わせを行うこととする。なお、レプリカを構築するリソースがなければ、遅れ制御手段３２２は、受信した遅れメッセージを廃棄する。
【００５１】
また、遅れ制御手段３２２は、他の処理サーバ３０の分散処理部２０から、遅れメッセージを受信した場合、メッセージを送信した処理サーバ３０に対して、メッセージを送信した分散処理部２０を計算結果の送信先とする分散処理部２０のレプリカを構築する指示を含んだレプリカ制御メッセージを返信する。なお、このレプリカ制御メッセージには、レプリカのオリジナルとなる分散処理部２０を特定する情報（vertex番号等）、ならびに、オリジナルとなる分散処理部２０に入力辺で接する他の分散処理部２０からの計算結果がある場合にその計算結果を付加することとする。これによって、遅れメッセージを送信した処理サーバ３０上にレプリカが構築され、計算を開始することが可能になる。
【００５２】
レプリカ制御手段３２３は、処理サーバ３０における分散処理部２０のレプリカの動作を制御するものである。
このレプリカ制御手段３２３は、他の処理サーバ３０から、レプリカを構築する指示を含んだレプリカ制御メッセージを受信した場合、分散処理部２０のレプリカの構築する指示をレプリカ管理手段３４に出力する。この指示には、オリジナルとなる分散処理部２０を特定する情報（vertex番号等）が含まれる。なお、レプリカ制御手段３２３は、レプリカの動作よりも、自身が担当する分散処理部２０の動作を優先させることとする。例えば、レプリカ制御手段３２３は、処理サーバ３０で動作するＯＳに、レプリカのプロセスよりも自身が担当する分散処理部２０のプロセスの優先度を高く設定する。
【００５３】
また、レプリカ制御手段３２３は、他の処理サーバ３０から、レプリカの動作を中止する指示を含んだレプリカ制御メッセージを受信した場合、分散処理部２０のレプリカの動作を中止させる指示をレプリカ管理手段３４に出力する。この指示には、レプリカに対応するオリジナルの分散処理部２０を特定する情報（vertex番号等）が含まれる。
なお、メッセージ処理手段３２は、管理サーバ１０から、終了指示メッセージを受信した場合に、すべての分散処理部２０を終了させる終了制御部を備えることとしてもよい。
【００５４】
リソース管理手段３３は、処理サーバ３０のリソース（ＣＰＵのコア数、メモリ量等）を管理するものである。
具体的には、リソース管理手段３３は、仮想化制御手段３１で動作される分散処理部２０が使用するリソースから、現在の空きリソースを把握し、遅れ制御手段３２２からのレプリカを起動させるリソースがあるか否かの問い合わせに対して回答する。
【００５５】
レプリカ管理手段３４は、レプリカの構築および動作を管理するものである。
このレプリカ管理手段３４は、レプリカ制御手段３２３からレプリカを構築する指示があった場合、仮想化制御手段３１に対して、分散処理部２０を特定する情報（vertex番号等）を含んだレプリカ構築の指示を行う。また、このとき、レプリカ管理手段３４は、分散処理部２０に対してレプリカが起動されていることを、例えば、分散処理部２０に対応するフラグをセットする等で管理する。
【００５６】
また、レプリカ管理手段３４は、レプリカ制御手段３２３からレプリカの動作を中止する指示があった場合、仮想化制御手段３１に対して、分散処理部２０を特定する情報（vertex番号等）を含んだレプリカ中止の指示を行う。また、このとき、レプリカ管理手段３４は、分散処理部２０に対してレプリカが動作していないことを、例えば、分散処理部２０に対応するフラグをリセットする等で管理する。
また、レプリカ管理手段３４は、データ転送手段３２１から、分散処理部２０に対するレプリカが動作しているか否かの問い合わせに対して、例えば、フラグのセット、リセット状態を参照して回答する。
【００５７】
記憶手段３５は、処理サーバ３０で使用する各種情報を記憶するものである。
この記憶手段３５は、管理サーバ１０から通知される分散処理部２０の接続関係を示すグラフトポロジや、分散処理部２０が必要とするリソース（ＣＰＵのコア数、メモリ量等）を記憶する。
なお、ここでは、分散同期処理システム１を、管理サーバ１０と処理サーバ３０とを備える構成としたが、複数の処理サーバ３０のうちの１台を代表サーバとして、その代表サーバ内に管理サーバ１０の機能を備える構成としてもよい。
【００５８】
≪分散同期処理システムの動作≫
次に、図７を参照（構成については、適宜図３〜図６参照）して、本発明の実施形態に係る分散同期処理システム１の動作について説明する。図７は、本実施形態に係る分散同期処理システム１の処理の流れを示すフローチャートである。
【００５９】
まず、管理サーバ１０（master）は、分散配置手段１１によって、予め設定された分散処理部２０（頂点〔vertex〕）の接続関係を示すグラフトポロジに基づいて、複数の処理サーバ３０（worker）に対して、分散処理部２０を配置する（ステップＳ１）。ここでは、分散配置手段１１は、分散処理部２０のソフトウェアを、処理サーバ３０に送信する。
【００６０】
そして、処理サーバ３０は、仮想化制御手段３１によって、処理サーバ３０上に仮想化プラットホームを構築し、仮想化プラットホーム上で分散処理部２０を仮想マシンとして動作させることで、計算処理を開始する（ステップＳ２）。なお、このステップＳ２以降の動作は、それぞれの処理サーバ３０で動作する。
【００６１】
ここで、処理サーバ３０は、メッセージ処理手段３２によって、自身の処理サーバ３０で動作する分散処理部２０、または、他の処理サーバ３０からメッセージを受信するまで待機する（ステップＳ３：Ｎｏ）。
そして、メッセージ処理手段３２は、メッセージを受信し（ステップＳ３：Ｙｅｓ）、そのメッセージが計算完了メッセージであった場合（ステップＳ４：Ｙｅｓ）、データ転送手段３２１によって、データ転送処理を行う（ステップＳ５）。
このステップＳ５において、データ転送手段３２１が行うデータ転送処理について、図８を参照してさらに詳細に説明する。
【００６２】
図８に示すように、データ転送手段３２１は、受信した計算完了メッセージの送信元を解析する（ステップＳ２０）。
そして、計算完了メッセージの送信元が他の処理サーバ３０上で動作する分散処理部２０であった場合（ステップＳ２０：他処理サーバの分散処理部）、データ転送手段３２１は、自身の処理サーバ３０上で動作する送信先（出力辺の相手先）の分散処理部２０に計算完了メッセージを送信する（ステップＳ２１）。
【００６３】
また、計算完了メッセージの送信元が自身の処理サーバ３０上で動作する分散処理部２０であった場合（ステップＳ２０：自処理サーバの分散処理部）、データ転送手段３２１は、当該分散処理部２０のレプリカが他の処理サーバ３０で動作しているか否かの判定を、レプリカ管理手段３４に問い合わせることにより行う（ステップＳ２２）。
【００６４】
ここで、他の処理サーバ３０で当該分散処理部２０のレプリカが動作している場合（ステップＳ２２：Ｙｅｓ）、データ転送手段３２１は、このレプリカを動作させている他の処理サーバ３０にレプリカの動作を中止する指示を含んだレプリカ制御メッセージを送信する（ステップＳ２３）。
【００６５】
一方、他の処理サーバ３０で当該分散処理部２０のレプリカが動作していない場合（ステップＳ２２：Ｎｏ）、または、ステップＳ２３の動作後、データ転送手段３２１は、計算完了メッセージを送信先（出力辺の相手先）である自身または他の処理サーバ３０上で動作する分散処理部２０に送信する（ステップＳ２４）。
【００６６】
また、計算完了メッセージの送信元が他の処理サーバ３０で動作するレプリカであった場合（ステップＳ２０：他処理サーバのレプリカ）、データ転送手段３２１は、レプリカに対応するオリジナルの分散処理部２０において現時点でのスーパーステップの計算処理を終了させ、オリジナルの分散処理部２０の計算結果の代わりに、受信した計算完了メッセージの計算結果に置き換えた計算完了メッセージを生成する（ステップＳ２５）。
【００６７】
そして、データ転送手段３２１は、ステップＳ２５で生成した計算完了メッセージをオリジナルの分散処理部２０の送信先（出力辺の相手先）に送信する（ステップＳ２６）。
また、計算完了メッセージの送信元が自身の処理サーバ３０で動作するレプリカであった場合（ステップＳ２０：自処理サーバのレプリカ）、データ転送手段３２１は、レプリカ管理手段３４を介して、レプリカを廃棄する（ステップＳ２７）。
そして、データ転送手段３２１は、計算完了メッセージをオリジナルの分散処理部２０が動作する処理サーバ３０に送信する（ステップＳ２８）。
以上のステップＳ２０〜Ｓ２８によって、データ転送手段３２１は、データ転送処理を行う。
【００６８】
なお、ここでは、図示を省略するが、分散処理部２０は、ステップＳ２１，Ｓ２４，Ｓ２６において送信された計算完了メッセージを、メッセージ送受信手段２２を介して受信する。そして、分散処理部２０は、数値計算手段２１によって、受信した計算完了メッセージの計算結果を入力として計算処理を行い、送信先（出力辺の相手先）があれば、メッセージ処理手段３２を介して、その送信先に計算完了メッセージを送信する。
【００６９】
このとき、分散処理部２０は、計算処理が他の分散処理部２０からの計算結果を必要とする処理である場合、予め定めた時間内に計算完了メッセージが受信されなければ、メッセージ送受信手段２２によって、計算完了メッセージの送信が遅れている他の分散処理部２０を特定する情報（vertex番号等）を設定した遅れメッセージをメッセージ処理手段３２に送信する。
図７に戻って、分散同期処理システム１の動作について説明を続ける。
【００７０】
処理サーバ３０は、ステップＳ３で受信したメッセージが遅れメッセージであった場合（ステップＳ６：Ｙｅｓ）、遅れ制御手段３２２によって、遅れ制御処理を行う（ステップＳ７）。
このステップＳ７において、遅れ制御手段３２２が行う遅れ制御処理について、図９を参照してさらに詳細に説明する。
【００７１】
図９に示すように、遅れ制御手段３２２は、受信した遅れメッセージの送信元を解析する（ステップＳ３０）。
そして、遅れメッセージの送信元が他の処理サーバ３０上で動作する分散処理部２０であった場合（ステップＳ３０：他処理サーバの分散処理部）、遅れ制御手段３２２は、送信元の処理サーバ３０に対して、メッセージを送信した分散処理部２０を計算結果の送信先とするレプリカを構築する指示を含んだレプリカ制御メッセージを送信する（ステップＳ３１）。このレプリカ制御メッセージには、レプリカのオリジナルとなる分散処理部２０を特定する情報（vertex番号等）を付加する。
【００７２】
一方、遅れメッセージの送信元が自身の処理サーバ３０上で動作する分散処理部２０であった場合（ステップＳ３０：自処理サーバの分散処理部）、遅れ制御手段３２２は、レプリカを構築するリソースがあるか否かの判定を、リソース管理手段３３に問い合わせることにより行う（ステップＳ３２）。
【００７３】
ここで、レプリカを構築するリソースがあると判定した場合（ステップＳ３２：Ｙｅｓ）、遅れ制御手段３２２は、遅れメッセージに設定されている他の分散処理部２０を動作させている処理サーバ３０に、遅れメッセージを送信する（ステップＳ３３）。この遅れメッセージには、遅れの対象となっている分散処理部２０を特定する情報（vertex番号等）を付加する。なお、レプリカを構築するリソースがないと判定した場合（ステップＳ３２：Ｎｏ）、遅れ制御手段３２２は、遅れメッセージを廃棄する（不図示）。
以上のステップＳ３０〜Ｓ３３によって、遅れ制御手段３２２は、遅れ制御処理を行う。
図７に戻って、分散同期処理システム１の動作について説明を続ける。
【００７４】
処理サーバ３０は、ステップＳ３で受信したメッセージがレプリカ制御メッセージであった場合（ステップＳ８：Ｙｅｓ）、レプリカ制御手段３２３によって、レプリカ制御処理を行う（ステップＳ９）。
このステップＳ９において、レプリカ制御手段３２３が行うレプリカ制御処理について、図１０を参照してさらに詳細に説明する。
【００７５】
図１０に示すように、レプリカ制御手段３２３は、受信したレプリカ制御メッセージの指示内容を解析する（ステップＳ４０）。
そして、レプリカ制御メッセージの指示内容が、レプリカを構築する指示であった場合（ステップＳ４０：構築）、レプリカ制御手段３２３は、レプリカ制御メッセージに付加されているレプリカのオリジナルとなる分散処理部２０を特定する情報に基づいて、レプリカ管理手段３４により、処理サーバ３０上にレプリカを構築し（ステップＳ４１）、計算処理を開始させる（ステップＳ４２）。
【００７６】
一方、レプリカ制御メッセージの指示内容が、レプリカを中止する指示であった場合（ステップＳ４０：中止）、レプリカ制御手段３２３は、レプリカ制御メッセージに付加されているレプリカを特定する情報（オリジナルの分散処理部２０のvertex番号等）に基づいてレプリカを特定し、レプリカ管理手段３４により、処理サーバ３０上のレプリカの動作を中止し（ステップＳ４３）、仮想化プラットホームから廃棄する（ステップＳ４４）。
以上のステップＳ４０〜Ｓ４４によって、レプリカ制御手段３２３は、レプリカ制御処理を行う。
図７に戻って、分散同期処理システム１の動作について説明を続ける。
【００７７】
処理サーバ３０は、管理サーバ１０から、終了指示メッセージを受信した場合（ステップＳ１０：Ｙｅｓ）、分散処理部２０の動作を終了させ（ステップＳ１１）、システム全体の動作を終了する。
一方、終了指示メッセージを受信しなければ（ステップＳ１０:Ｎｏ）、処理サーバ３０は、ステップＳ３に戻って動作を継続する。
【００７８】
以上説明したように分散同期処理システム１を構成し、動作させることで、分散同期処理システム１は、同期を待ち、かつ、空きリソースがある処理サーバ３０が、他の処理サーバ３０の分散処理部２０の動作をレプリカにより並行して開始（投機実行）することができる。これによって、分散同期処理システム１は、システム全体として、処理時間を短縮することができる。
【００７９】
≪分散同期処理システムの適用例≫
次に、図１２〜図１７を参照して、図１１に示すグラフトポロジ（グラフＧ）で表される計算を、本発明の実施形態に係る分散同期処理システム１で行う適用例について説明する。
ここでは、図１２に示すように、分散同期処理システム１は、２台の処理サーバ３０（worker１，worker２）で構成され、分散処理部２０である頂点（vertex）Ｖ１〜Ｖ３のうち、頂点（vertex）Ｖ１，Ｖ２をworker１が担当し、頂点（vertex）Ｖ３をworker２が担当するものとする。なお、ここでは、説明を簡略化するため頂点（vertex）Ｖ１〜Ｖ５の各動作ステップは、ＣＰＵの１コア（ＣＯＲＥ）で数字の順に動作するものとする。
また、以下の説明では、複数の分散処理部２０および処理サーバ３０を識別するため、それぞれの分散処理部２０をvertex（vertexＶ１，Ｖ２，…）と呼称し、それぞれの処理サーバ３０をworker（worker１，２，…）と呼称する。
【００８０】
図１２に示すように、worker１は、vertexＶ１，Ｖ２，Ｖ３において、１，２，…の各ステップで処理を実行する。worker２は、vertexＶ４において、１，２，…の各ステップで処理を実行する。worker３は、vertexＶ5において、１，２，…の各ステップで処理を実行する。このように、ここでは、各workerの処理量が均等に割り振られていない状況を想定する。
【００８１】
図１３は、各workerが４ステップまでの処理を完了した状態を示している。ここで、vertexＶ４は、自身の計算が完了し、出力辺で接するvertexＶ１，Ｖ３に計算完了メッセージを送信する。しかし、vertexＶ４は、入力辺で接するvertexＶ２から計算完了メッセージを取得しなければ次のスーパーステップに移行することができず、待機状態となる。
そこで、vertexＶ４が予め定めた時間内に計算完了メッセージを受信せず、かつ、worker２のリソースに少なくともvertexＶ２を動作させる余裕があれば、worker２は、worker１に対してvertexＶ２の遅れを示す遅れメッセージを送信する。なお、worker２のリソースに余裕がなければ、worker２は、worker１に対して遅れメッセージを送信しない。ここでは、worker２のリソースに余裕があるものとして、以下説明する。
【００８２】
図１４は、worker１が遅れメッセージを受信した後、worker２に対してvertexＶ２の構築を指示するレプリカ制御メッセージを送信し、worker２がvertexＶ２のレプリカＲを構築した状態を示している。なお、レプリカ制御メッセージには、vertexＶ２を特定する情報や、vertexＶ２を動作させるために必要な情報（例えば、スーパーステップＳＳ２以降であれば、vertexＶ１の計算結果）が含まれる。これによって、worker２は、自身でvertexＶ２を動作させることができる。なお、このとき、worker２は、レプリカよりもworker２上で動作する自身が担当するvertexの動作を優先する。
【００８３】
図１５は、各workerが８ステップまでの処理を完了した状態を示している。ここで、worker２は、vertexＶ２のレプリカＲが計算を完了したことを、オリジナルのvertexＶ２を動作させているworker１に計算完了メッセージを送信する。ここで、worker１は、オリジナルのvertexＶ２の計算を中断し、worker２から送信された計算結果を、vertexＶ２の出力辺で接するvertexＶ１，Ｖ４に計算完了メッセージを送信する。
なお、図１５では、worker３において、vertexＶ３のレプリカを構築した状態を示しているが、vertexＶ２のレプリカＲと同じ動作であるため、ここでは説明を省略する。
【００８４】
図１６は、各workerが９ステップまでの処理を完了した状態を示している。ここで、worker２は、vertexＶ２の計算結果を取得しているため、次のスーパーステップＳＳ２に移行している。
一方、worker１は、worker２で動作したレプリカにより、vertexＶ２の計算は早く完了しているため、１０ステップ以降は、vertexＶ１，Ｖ３だけが動作することになる。これによって、worker１の処理時間を早めることができる。
【００８５】
図１７は、worker１で動作するオリジナルンのvertexＶ２が、レプリカＲよりも早く処理が完了した状態を示している。この場合、worker１は、worker２に対してvertexＶ２のレプリカＲの動作の中止を指示するレプリカ制御メッセージを送信し、さらに、vertexＶ２の出力辺で接するvertexＶ４が動作するworker２に計算完了メッセージを送信する。
そして、worker２は、vertexＶ２のレプリカＲの計算を中止し、vertexＶ４がオリジナルのvertexＶ２の計算結果を取得し、次のスーパーステップに移行する。
このように、分散同期処理システム１は、各workerの処理量が均等に割り振られていない状態であっても、システム全体として、処理時間を短縮することができる。
【符号の説明】
【００８６】
１      分散同期処理システム
１０    管理サーバ（master）
１１    分散配置手段
２０    分散処理部（vertex）
２１    数値計算手段
２２    メッセージ送受信手段
３０    処理サーバ（worker）
３１    仮想化制御手段
３２    メッセージ処理手段
３２１  データ転送手段
３２２  遅れ制御手段
３２３  レプリカ制御手段
３３    リソース管理手段
３４    レプリカ管理手段
３５    記憶手段
【図１】
2019035997.tif 000003
【図２】
2019035997.tif 000004
【図３】
2019035997.tif 000005
【図４】
2019035997.tif 000006
【図５】
2019035997.tif 000007
【図６】
2019035997.tif 000008
【図７】
2019035997.tif 000009
【図８】
2019035997.tif 000010
【図９】
2019035997.tif 000011
【図１０】
2019035997.tif 000012
【図１１】
2019035997.tif 000013
【図１２】
2019035997.tif 000014
【図１３】
2019035997.tif 000015
【図１４】
2019035997.tif 000016
【図１５】
2019035997.tif 000017
【図１６】
2019035997.tif 000018
【図１７】
2019035997.tif 000019
【図１８】
2019035997.tif 000020
【図１９】
2019035997.tif 000021
【図２０】
2019035997.tif 000022
【図２１】
2019035997.tif 000023
【図２２】
2019035997.tif 000024