(57)【要約】【課題】  複数のアクセラレータ装置を用いた分散並列計算における性能を最大化すること。【解決手段】  本発明のホストコンピュータは、データサイズの範囲についてデータ転送速度および計算速度を性能分析するプロファイラ部と、性能分析されたデータ転送速度および前記計算速度から、ネットワークを経由するデータ転送に最適なチャンクサイズを決定するサイズ最適化部と、ホストコンピュータのメモリ内に格納されたあるサイズのデータを分割または集約することによって、最適なチャンクサイズを有するチャンクに該データをカプセル化するカプセル化部と、カプセル化されたデータをアクセラレータ装置に振り分け、かつ、該アクセラレータ装置に対し、受信したカプセル化されたデータに関してパイプライン計算を命令するディスパッチャ部と含む。【選択図】  図３
2013140530.tif 000002
【特許請求の範囲】
【請求項１】
ネットワークを介して相互接続される、ホストコンピュータと、少なくとも１つのアクセラレータ装置との間で実行される分散計算方法であって、前記分散計算方法は、
データサイズの範囲についてデータ転送速度および計算速度を性能分析し、前記ネットワークを経由するデータ転送に最適なチャンクサイズを決定するステップと、
前記ホストコンピュータのメモリ内に格納されたあるサイズのデータを分割または集約して、前記最適なチャンクサイズを有するチャンクに該データをカプセル化するステップと、
カプセル化されたデータを前記アクセラレータ装置に振り分けるステップと、
前記アクセラレータ装置に対し、受信した前記カプセル化されたデータに関してパイプライン計算を命令するステップと
を含む、分散計算方法。
【請求項２】
前記データ転送速度と前記計算速度の比は、実際にかかった転送時間から計算される、請求項１に記載の分散計算方法。
【請求項３】
前記最適なチャンクサイズは、計算速度に関して前記性能分析するステップ中に送信された最小データサイズと最大データサイズとの間で、最大のデータ転送速度を有するデータサイズに決定される、請求項１に記載の分散計算方法。
【請求項４】
前記最適なチャンクサイズは、計算および通信の重なり比が１に最も近いデータサイズに決定され、そのようなデータサイズが複数存在する場合は、最小データサイズと最大データサイズとの間で最大のデータ転送速度を有するものに決定される、請求項１に記載の分散計算方法。
【請求項５】
前記パイプライン計算を命令するステップは、
前記アクセラレータ装置内のバッファ・オブジェクトへのチャンク数分の前記カプセル化されたデータの複数書き込みを命令するステップと、
前記カプセル化されたデータの受信に際して、前記カプセル化されたデータへの前記アクセラレータ装置の演算の実行を命令するステップと
を含む、請求項１に記載の分散計算方法。
【請求項６】
前記データのカプセル化について、前記データのサイズが前記最適なチャンクサイズよりも大きい場合に前記データは前記分割に施され、前記データのサイズが前記最適なチャンクサイズよりも小さい場合に前記データは前記集約に施される、請求項１に記載の分散計算方法。
【請求項７】
前記アクセラレータ装置は、アプリケーション・プログラムを実装するコンピュータから選択される、請求項１に記載の分散計算方法。
【請求項８】
前記アクセラレータ装置は、ＴＣＰ／ＩＰネットワークを用いて前記ホストコンピュータとネットワーク接続される、請求項１に記載の分散計算方法。
【請求項９】
ネットワークを介して相互接続される、ホストコンピュータと、少なくとも１つのアクセラレータ装置との間で分散計算するためのプログラムであって、前記プログラムは、前記ホストコンピュータに対し、
データサイズの範囲についてデータ転送速度および計算速度を性能分析し、前記ネットワークを経由するデータ転送に最適なチャンクサイズを決定するステップと、
前記ホストコンピュータのメモリ内に格納されたあるサイズのデータを分割または集約して、前記最適なチャンクサイズを有するチャンクに該データをカプセル化するステップと、
カプセル化されたデータを前記アクセラレータ装置に振り分けるステップと、
前記アクセラレータ装置に対し、受信した前記カプセル化されたデータに関してパイプライン計算を命令するステップと
を実行させる、プログラム。
【請求項１０】
前記データ転送速度と前記計算速度の比は、実際の転送時間から計算される、請求項９に記載のプログラム。
【請求項１１】
前記最適なチャンクサイズは、計算速度に関して前記性能分析するステップ中に送信された最小データサイズと最大データサイズとの間で、最大のデータ転送速度を有するデータサイズに決定される、請求項９に記載のプログラム。
【請求項１２】
前記最適なチャンクサイズは、計算および通信の重なり比が１に最も近いデータサイズに決定され、そのようなデータサイズが複数存在する場合は、最小データサイズと最大データサイズとの間で最大のデータ転送速度を有するものに決定される、請求項９に記載のプログラム。
【請求項１３】
前記パイプライン計算を命令するステップは、
前記アクセラレータ装置内のバッファ・オブジェクトへのチャンク数分の前記カプセル化されたデータの複数書き込みを命令するステップと、
前記カプセル化されたデータの受信に際して、前記カプセル化されたデータへの前記アクセラレータ装置の演算の実行を命令するステップと
を含む、請求項９に記載のプログラム。
【請求項１４】
前記データのカプセル化について、前記データのサイズが前記最適なチャンクサイズよりも大きい場合に前記データは前記分割に施され、前記データのサイズが前記最適なチャンクサイズよりも小さい場合に前記データは前記集約に施される、請求項９に記載のプログラム。
【請求項１５】
前記アクセラレータ装置は、アプリケーション・プログラムを実装するコンピュータから選択され、前記アクセラレータ装置は、ＴＣＰ／ＩＰネットワークを用いて前記ホストコンピュータとネットワーク接続される、請求項９に記載のプログラム。
【請求項１６】
ネットワークを介して相互接続される、ホストコンピュータと、少なくとも１つのアクセラレータ装置との間の分散計算を実行する前記ホストコンピュータであって、前記ホストコンピュータは、
データサイズの範囲についてデータ転送速度および計算速度を性能分析するプロファイラ部と、
性能分析された前記データ転送速度および前記計算速度から、前記ネットワークを経由するデータ転送に最適なチャンクサイズを決定するサイズ最適化部と、
前記ホストコンピュータのメモリ内に格納されたあるサイズのデータを分割または集約することによって、前記最適なチャンクサイズを有するチャンクに該データをカプセル化するカプセル化部と、
カプセル化されたデータを前記アクセラレータ装置に振り分け、かつ、該アクセラレータ装置に対し、受信した前記カプセル化されたデータに関してパイプライン計算を命令するディスパッチャ部と
含む、ホストコンピュータ。
【請求項１７】
前記最適なチャンクサイズは、計算速度に関して前記プロファイラ部により送信された最小データサイズと最大データサイズとの間で、最大のデータ転送速度を有するデータサイズに決定される、請求項１６に記載のホストコンピュータ。
【請求項１８】
前記最適なチャンクサイズは、計算および通信の重なり比が１に最も近いデータサイズに決定され、そのようなデータサイズが複数存在する場合は、最小データサイズと最大データサイズとの間で最大のデータ転送速度を有するものに決定される、請求項１６に記載のホストコンピュータ。
【請求項１９】
前記ディスパッチャ部は、前記アクセラレータ装置内のバッファ・オブジェクトへのチャンク数分の前記カプセル化されたデータの複数書き込みを命令し、前記カプセル化されたデータの受信に際して、前記カプセル化されたデータへの前記アクセラレータ装置の演算の実行を命令する、請求項１６に記載のホストコンピュータ。
【請求項２０】
ＴＣＰ／ＩＰネットワークを介して相互接続される、ホストコンピュータと、少なくとも１つのアクセラレータ装置とを含む分散計算システムであって、前記アクセラレータ装置は、アプリケーション・プログラムを実装し、前記ホストコンピュータは、
データサイズの範囲についてデータ転送速度および計算速度を性能分析するプロファイラ部と、
性能分析された前記データ転送速度および前記計算速度から、前記ネットワークを経由するデータ転送に最適なチャンクサイズを決定するサイズ最適化部と、
前記ホストコンピュータのメモリ内に格納されたあるサイズのデータを分割または集約することによって、前記最適なチャンクサイズを有するチャンクに該データをカプセル化するカプセル化部と、
カプセル化されたデータを前記アクセラレータ装置に振り分け、かつ、該アクセラレータ装置に対し、受信した前記カプセル化されたデータに関してパイプライン計算を命令するディスパッチャ部と
を含み、
前記最適なチャンクサイズは、計算および通信の重なり比が１に最も近いデータサイズに決定し、そのようなデータサイズが複数存在する場合は、最小データサイズと最大データサイズとの間で最大のデータ転送速度を有するものに決定され、
前記ディスパッチャ部は、前記アクセラレータ装置内のバッファ・オブジェクトへのチャンク数分の前記カプセル化されたデータの複数書き込みを命令し、前記カプセル化されたデータの受信に際して、前記カプセル化されたデータへの前記アクセラレータ装置の演算の実行を命令する、分散計算システム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、分散計算技術に関し、より詳細には、複数のアクセラレータ装置を用いた分散並列計算に関する。
【背景技術】
【０００２】
近年、計算効率および／または計算速度を高めるためにＧＰＵ（Graphic Processor Units）を用いたマルチプロセッサ計算が広く利用されている。ＧＰＵは、典型的には、計算性能を高めるためメインＣＰＵのアクセラレータとして用いられる。上述のようなマルチプロセッサ計算アーキテクチャでは、ＰＣＩまたはＰＣＩ−Ｅｘｐｒｅｓｓなどの内部バスを介して接続されたＧＰＵネットワークがよく用いられる。ここでは、上述のような内部バスを介して接続されたＧＰＵを、密結合型アクセラレータ装置（Tightly-Coupled Accelerator Devices）と参照する。
【０００３】
ＧＰＵは、適切なプログラム言語によって、ホストＣＰＵの制御の下、並列動作して計算性能を高めている。上述したようなプログラム言語の一例としては、ＯｐｅｎＣＬを挙げることができる。ＯｐｅｎＣＬは、ホストＣＰＵおよびＧＰＵ間のデータ転送を管理するために適用することができ、本発明では、このデータ転送における性能コストを最小化するため利用することができる。
【０００４】
他のスキームのマルチプロセッサ計算アーキテクチャは、例えば、分散コンピューティングまたはグリッド・コンピューティングとして知られている。これらのマルチプロセッサ計算アーキテクチャは、ホストコンピュータまたはマスタコンピュータの制御下で計算を分担する複数のサーバまたはコンピュータを含むことができる。このタイプのマルチプロセッサ計算アーキテクチャでは、コンピュータ間は、イーサネット（登録商標）などの外部バスネットワークと種々の物理接続プロトコルを用いたネットワーク・インタフェース・カードとを用いて接続される。コンピュータは、ネットワーク内で実行される計算全体をサポートすることができ、分散計算を担当するコンピュータは、アクセラレータと見なすことができる。しかしながら、上記分散計算アーキテクチャにおけるコンピュータは、外部ネットワークを介してＴＣＰ／ＩＰで接続されており、よって、分散計算システム内のコンピュータは、疎結合型アクセラレータ（loosely-coupled accelerators）と見なすことができる。
【０００５】
疎結合型マルチプロセッサ・システムでは、コンピュータまたはノードは、外部ネットワークを介して接続される。このため、ホストコンピュータおよびアクセラレータ間のデータ転送は、データサイズ、ランタイム実装やネットワーク状態などの転送条件の影響を受ける可能性がある。
【０００６】
ＴＣＰネットワークを経由した計算の性能強化は、これまでも開発されており、米国特許出願公開第２００８／０２９５０９８号明細書（特許文献１）は、大きなＴＣＰセグメントを小さなＴＣＰセグメントで動的にセグメント化し、割り込み頻度を低減するコンピュータ・システムを開示する。特開２０１１−１７０７３２号公報（特許文献２）は、機能ブロックをストランドに分け、計算時間に依存して機能ブロックを修正する並列コンピューテーション方法を開示する。
【０００７】
密結合型アクセラレーション・アーキテクチャにおいては、多数の少量の転送を１つの大きな転送にバッチ化して、データ転送性能を向上する技術が提案されている（NVIDIA，OpenCL Best Practice Guide，Section 3.1 ”Data Transfer between Host and Device”を参照のこと）。さらに、Ｋｉｍ等は、非特許文献１において、データ並列プログラム中で通信レイテンシを隠蔽するために通信および計算をオーバラップさせるモデルを開示する。
【先行技術文献】
【特許文献】
【０００８】
【特許文献１】米国特許出願公開第２００８／０２９５０９８号明細書
【特許文献２】特開２０１１−１７０７３２号公報
【非特許文献】
【０００９】
【非特許文献１】Junghwan Kim，Sangyong Han，Heungwan Kim，and Seungho Cho、”A New Communication and Computation Overlapping Model with Loop Sub-Partitioning and Dynamic Scheduling”
【発明の概要】
【発明が解決しようとする課題】
【００１０】
これまで開発されてきた並列計算アーキテクチャのための性能強化では、計算が開始可能となる前のネットワークを介したデータ転送のレイテンシを低減する点については充分に検討されてこなかった。さらに、本質的に相異するプラットフォーム・アーキテクチャ（ＰｏｗｅｒＰＣ（登録商標）アーキテクチャ，インテル（登録商標）アーキテクチャおよび／またはＩＢＭ  Ｓｙｓｔｅｍ／３９０のようなメインフレームアーキテクチャなど）を接続するには、プラットフォームを横断してアプリケーションを動作させる疎結合型アクセラレータによる並列計算の性能を最適化しながら、計算プラットフォーム間の差異を乗り越える必要があった。
【００１１】
本発明の目的は、アプリケーション環境を構成する異プラットフォーム間の差異を緩和する、新規な分散並列計算アーキテクチャを提供することである。
【００１２】
本発明の他の目的は、データ転送速度および並列計算性能を最適化することができる、新規な分散計算アーキテクチャを提供することである。
【００１３】
本発明のさらに他の目的は、プラットフォーム・アーキテクチャ間の差異を乗り越えて並列分散計算を実行するための方法、プログラム、ホストコンピュータおよびシステムを提供することである。
【課題を解決するための手段】
【００１４】
本発明では、ホストコンピュータおよびアクセラレータ装置が外部ネットワークを介して相互接続され、ホストは、アプリケーションの作業用データをアクセラレータ装置に送信する前に、現在の環境で分散計算に用いるデータサイズを最適化する。このチャンクサイズと呼ばれる最適データサイズは、ネットワーク内におけるアクセラレータ装置との通信速度および計算速度の実測に基づく。
【００１５】
ホストコンピュータは、アプリケーション・データをセグメントに分割して、データを最適なチャンクにカプセル化し、そして、カプセル化されたデータと、該カプセル化されたデータに対する並列計算の命令とをアクセラレータ装置に振り分ける。
【００１６】
アクセラレータ装置は、各カプセル化されたデータを受信した際には、アプリケーション・データ全体を待たずに、データ転送と並列して計算を開始し、これにより、疎結合型ネットワークを経由するデータ転送に起因した通信オーバヘッドが最小化され、並列計算効率が改善される。
【００１７】
言い換えると、各アクセラレータ装置は、カプセル化されたデータおよび命令を受信し、必要に応じて（ホスト命令によって規定された通り）、計算に必要なすべてのデータが到着するのを待機する。ひとたびデータが集まると、ホスト命令によって規定された特定の計算（すなわち「カーネル」）をディスパッチする。完了すると、再びホスト・カプセル化命令によって規定された通り、出力データがアクセラレータ装置からホストへ送信される。
【図面の簡単な説明】
【００１８】
【図１】ＯＳＩ基本参照モデルを用いて疎結合型システムにおける分散計算のデータ転送コストを説明する図。
【図２】本発明の例示的な分散計算システム２００のブロック図。
【図３】疎結合型システムを構築するためのプラットフォームの機能ブロック図。
【図４】本分散計算システムの処理を示すフローチャート。
【図５】図４におけるステップＳ４００「最適なチャンクサイズを決定する」の詳細な処理を示す図。
【図６】図５のステップＳ５００においてホストからアクセラレータ装置への経路についてのデータ転送コストおよび計算コストを性能分析する詳細な処理を示す図。
【図７】図５のステップＳ５００においてアクセラレータ装置からホストへの経路についてデータ転送コストおよび計算コストを性能分析する詳細な処理を示す図。
【図８】転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）８００、計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）８１０およびデータサイズの典型的な関係を示す図。
【図９】最適なチャンクサイズおよびアプリケーション・データサイズに依存する、アクセラレータ装置のためのデータ・バッファ割り当てを例示する図。
【図１０】アプリケーション・データが最適なチャンクサイズより大きい場合のデータ転送を例示する図。
【図１１】アプリケーション・データが最適なチャンクサイズより小さく、したがってアクセラレータ装置に転送される際に集約が行われる場合のホストおよびアクセラレータ装置のデータ処理を例示する図。
【図１２】従来技術のカーネル計算１２１０から本発明のカーネル計算１２２０への変形についてのホスト１２００の疑似コードを表す図。
【図１３】図４に示した本発明のパイプライン計算上の改善の仕組みを説明する図。
【図１４】本発明によるシステム実装における計算性能の改善結果を示す図。
【発明を実施するための形態】
【００１９】
以下、本発明について特定の実施形態に基づいて説明するが、説明する実施形態は、単に最良の実施形態を説明するものであり、本発明を限定するものではない。以下、図１を参照して、疎結合型システムにおける分散計算のデータ転送コストについて、ＯＳＩ基本参照モデルを用いて説明する。システム内のノード各々は、通常、物理層１０３、ネットワーク層１０２およびトランスポート層１０１を含む。物理層１０３は、イーサネット（登録商標）フレームの受信および送信を担当する。ネットワーク層１０２は、適切なデータサイズのペイロードを持つＩＰパケットの形成を担当する。トランスポート層１０１は、アプリケーション層１００内のアプリケーションによって処理されたデータからＴＣＰパケットを形成する。アプリケーション層１００は、アプリケーションにおけるデータの計算によって、ランタイム・オーバヘッドとしてそのコストを支払う。
【００２０】
トランスポート層１０１、ネットワーク層１０２および物理層１０３は、それぞれ対応する処理コストに寄与し、アプリケーション全体に対するコストは、ノードから他のノードへＴＣＰ／ＩＰパケットとして転送されるデータサイズに影響される可能性がある。上記データサイズの影響は、密結合型システムとは対照的に、計算およびスケーラビリティに余分なコストを生じさせる可能性がある。
【００２１】
図２は、本発明の例示的な分散計算システム２００のブロック図である。本システム２００は、例えばメインフレーム２０１、サーバ２０２、ブレード型サーバがラック内に配置されるラックマウント・サーバ２０３といった種々の計算プラットフォームを含む。メインフレーム２０１は、ＩＢＭ  Ｓｙｓｔｅｍ／３９０といったＳｙｓｔｅｍ  ｚシリーズから選択することができるが、他のアーキテクチャのメインフレームを用いてもよい。サーバ２０２およびラックマウント・サーバ２０３内のブレード型サーバは、ＩＢＭ  ＰＯＷＥＲ、Ｐｏｗｅｒ  Ｓｅｒｉｅｓ（登録商標）で実装されるＰｏｗｅｒＰＣ（登録商標）などのＣＰＵ／ＣＰＵｓを含み構成される。
【００２２】
これに対して、プラットフォーム２０４は、ｘ８６アーキテクチャとも参照される、Ｉｎｔｅｌ  ＸＥＯＮ（登録商標）といったＣＰＵ／ＣＰＵｓを含むことができる。図２に示すプラットフォームは、それぞれ、異なるオペレーティング・システムおよびアプリケーション・プログラムがインストールされ、オペレータまたはユーザから要求されたサービスを提供する。プラットフォームは、ハブ／スイッチ２１０およびルータ２３０を経由して、イーサネット（登録商標）または光通信を用いるＦＴＴＨといった適切なネットワーク２２０によって接続される。プラットフォームは、さらに、ＲＡＭ、ＲＯＭ、ハードディスク・ドライブ、ギガビット・レートのネットワーク・インタフェース・カードなどを含み、オペレータに業務アプリケーション・サービスを提供する。
【００２３】
プラットフォームは、ＴＣＰ／ＩＰプロトコルを通じて、ネットワークを経由してデータ、コマンドおよびメッセージを通信し、本発明においては、プラットフォームは、疎結合型アーキテクチャによる分散計算環境を提供する。分散計算は、これまで知られた如何なるプロトコルまたは技術を用いることによって可能であるが、既存のアプリケーション・リソースを用いてプラットフォーム間の差異を乗り越える観点からは、ＯｐｅｎＣＬアーキテクチャが有用である。
【００２４】
図３は、疎結合型システムを構築するためのプラットフォームの機能ブロック図３００である。プラットフォーム３１０はホストの役割を担い、プラットフォーム３２０はアクセラレータ装置の役割を担う。ここで、用語「アクセラレータ装置」または、より簡単に「アクセラレータ」は、分離されたコンピュータまたは情報処理装置を参照し、このコンピュータまたは情報処理装置は、ホストと通信して、該ホストの全体計算を高速化する。ホストは、その機能手段として、ホスト・アプリケーション３１１と、ホスト３１０に実装されたアプリケーション・カーネル・プログラム３１２と、ランタイム環境３１３と、ソケット・インタフェース３１４とを含み構成される。
【００２５】
ホスト・アプリケーション３１１は、プロファイラ部３１１ａと、サイズ最適化部３１１ｂと、カプセル化部３１１ｃと、ディスパッチャ部３１１ｄとを含み構成される。プロファイラ部は、最小サイズ（ＭＩＮ＿ＳＩＺＥ）から最大サイズ（ＭＡＸ＿ＳＩＺＥ）までのデータサイズの範囲について、データ転送速度および計算速度を性能分析する。ここで、最小サイズ（ＭＩＮ＿ＳＩＺＥ）および最大サイズ（ＭＡＸ＿ＳＩＺＥ）は、アプリケーション・プログラムにおいて、ホストおよびアクセラレータ装置間で転送される可能なデータサイズである。サイズ最適化部３１１ｂは、性能分析されたデータ転送速度および計算速度に基づいて、ネットワークを経由するデータ転送に最適なチャンクサイズを見つけ出し、決定する。
【００２６】
カプセル化部３１１ｃは、ホストコンピュータのメモリ内に格納されたデータを、最適なチャンクサイズを有するチャンクに分割または集約することによって、カプセル化する。ディスパッチャ部３１１ｄは、カプセル化されたデータをアクセラレータ装置に振り分けると共に、アクセラレータ装置に対し、受信したカプセル化されたデータに関してパイプライン計算を命令する。なお、プロファイラ部３１１ａ、サイズ最適化部３１１ｂ、カプセル化部３１１ｃおよびディスパッチャ部３１１ｄは、ホスト・アプリケーション３１１の一部である必要はなく、これらのいくつかはランタイム環境３１３の一部であってもよい。
【００２７】
ホスト・アプリケーション３１１は、カーネル・プログラム３１２の制御により、ユーザに対し種々のサービスを提供し、説明する実施形態では、ＯｐｅｎＣＬ  ＡＰＩが、ホスト・アプリケーションの一つのモジュールとして実装されてもよい。カーネル・プログラム３１２は、ｚ／ＯＳ、ＵＮＩＸ（登録商標）、Ｌｉｎｕｘ（登録商標）、またはＷｉｎｄｏｗｓ（登録商標）２００Ｘ  Ｓｅｒｖｅｒ用に実装され、ホストの種々の動作を制御する。
【００２８】
特定の実施形態では、ホストは、疎結合型システムを実装するため、カーネル３１２の１つのコンポーネントとしてＯｐｅｎＣＬカーネル・プログラムを含む。ランタイム環境３１３は、ＯｐｅｎＣＬランタイム、ホストのランタイム状態をサポートするダイナミック・リンク・ライブラリなどのランタイム・ライブラリを含む。ソケット・インタフェース３１４は、ソケット通信を用いて、ＴＣＰ／ＩＰパケットをアクセラレータ装置３２０に送信し、説明する実施形態では、Ｓｏｃｋｅｔ  Ｓｅｎｄ／Ｒｅｃｉｅｖｅ、ＲＤＭＡ  Ｒ／Ｗメソッドまたはクラスが、ＯｐｅｎＣＬアーキテクチャをサポートするための１つのモジュールとして実装される。
【００２９】
アクセラレータ装置３２０は、アプリケーション・カーネル３２１と、バッファ・オブジェクト３２２と、ランタイム環境３２３と、ホスト３１０へのプロキシ機能部３２４と、ソケット・インタフェース３２５とを含み構成される。アプリケーション・カーネル３２１は、要求されたサービスを提供し、バッファ・オブジェクト３２２は、アクセラレータ装置３２０で使用される種々の情報を格納する。説明する実施形態では、バッファ・オブジェクト３２２は、疎結合型システムのアクセラレータ装置として、入力データおよびコマンドを受信し、またアプリケーション・カーネル・プログラム３２１からの計算結果を送信する。
【００３０】
ランタイム環境３２３は、その機能コンポーネントとして、プラットフォームのアーキテクチャに適合するＯｐｅｎＣＬコンパイラおよびランタイム・ライブラリを含み、ＯｐｅｎＣＬ関数を実装するプロキシ機能部３２４とともに、ホスト３１０から命令された演算を実行する。ソケット・インタフェース３２５は、ソケット通信を介してホストと通信し、ネットワーク３３０を経由したホスト３１０に対するＳｏｃｋｅｔ  Ｓｅｎｄ／Ｒｅｃｅｉｖｅ、ＲＤＭＡ  Ｒ／Ｗメソッドまたはクラスを含み構成される。
【００３１】
したがって、疎結合型システムは、２種類のデータ処理コストを含み、１つは計算およびフレーム形成コストであり、他の１つはネットワークを経由するデータ転送コストである。この２種類のコスト、つまり、計算コストおよびデータ転送コストは、分散計算環境における円滑かつ効率的な計算のため良くバランスされ、通信レイテンシの待機時間を最小化することが好ましい。換言すれば、データ転送速度および計算速度は、アクセラレータ装置内のハードウェア・リソースを浪費しないように最適化される必要がある。
【００３２】
図４は、本分散計算システムの処理を示すフローチャートである。図４に示す処理は、ステップＳ４００から開始され、ホストは、ネットワーク状態およびアクセラレータ装置のハードウェア状態を検査し、システムにおける最適なチャンクサイズを決定する。ステップＳ４００の詳細は後述するが、ステップＳ４００では、ホストは、ネットワークを経由して、アクセラレータ装置によるテスト計算を行い、その応答時間を測定する。ステップＳ４０１では、ホストは、転送データのためのバッファおよびサブバッファのサイズを割り当てる。ここで、バッファのサイズは、対象とされる計算のデータサイズにより決定することができ、サブバッファのサイズは、最適なチャンクサイズに設定することができる。あるいはその反対とすることができる。説明する好適な実施形態において、いずれとなるかは、データ分割またはデータ集約のどちらが適用されるかに依存する。
【００３３】
ステップＳ４０２では、対象とされる計算用のデータが分割または集約されて、これによって、アクセラレータ装置に転送されるデータサイズが最適なサイズを有するサブバッファ内に収容されるようになる。ステップＳ４０３では、ホストは、最適なサイズを有するデータを、アクセラレータ装置内での計算のためのコマンドまたは命令と共に送信する。説明する実施形態では、アクセラレータ装置上での計算の命令は、ＯｐｅｎＣＬ言語によりコード化されるものとするが、しかしながら、その他の分散計算プロトコルを使用してもよい。
【００３４】
ステップＳ４０４では、ホストは、データ分割が適用され、かつアクセラレータ装置に割り当てられたアプリケーション・カーネル・タスクが結合可能型であるか、つまり、タスクが分割データについてパイプライン実行を適用できるかを判定する。肯定的である場合（ｙｅｓ）は、ホストは、アクセラレータ装置にパイプライン化された計算を開始させるための命令を転送し、アクセラレータ装置で、データ通信と重複させて計算を実行させる。一方、タスクがパイプライン計算に適合していない場合（ｎｏ）は、ホストは、パイプライン処理を行わない通常の疎結合型計算命令コードを転送し、図４に示す処理は、アクセラレータ装置によるステップＳ４０５へ進められる。
【００３５】
ステップＳ４０５〜ステップＳ４０８の処理は、計算を担当するアクセラレータ装置において実行される処理である。ステップＳ４０５では、アクセラレータ装置は、データおよび命令を受信し、アクセラレータ装置は、ここでは計算が結合可能型ではなく、計算の投機的な開始が不正であるため、すべてのデータチャンクが受信されるまで待機する。ステップＳ４０５に処理が進められると、アクセラレータ装置は、計算に必要なデータ全体を受信した後に計算を開始し、ステップＳ４０８で、計算の結果をホストに返す。
【００３６】
ステップＳ４０４の判定で、肯定的な結果が戻された場合（ｙｅｓ）は、ステップＳ４０６へ処理が進められ、アクセラレータ装置は、最適化された計算および通信の重なりをもって、順次受信されたデータチャンクおよび命令に対して、パイプライン化された計算の実行を開始する。アクセラレータ装置は、連続して送信されたデータチャンクに対してパイプライン実行による計算を最後まで続け、アクセラレータ装置は、結合演算を呼び出して、パイプライン化された計算で得られた各部分結果を結合する。アクセラレータ装置は、計算された結果をホストに返し、ホストから受信したコマンドを完了させる。
【００３７】
図５は、図４におけるステップＳ４００「最適なチャンクサイズを決定する」の詳細な処理を示す。本処理は、ステップＳ５００から開始し、ホストは、所与のランタイムおよびネットワーク環境において、ホストからアクセラレータへの経路とアクセラレータからホストへの経路との両方について別個に、データ転送コストおよび計算パイプライン効果を性能分析する。そして、ステップＳ５０１では、ホストは、適切なサンプル計算セットを用いて、経過時間（ｅｌａｐｓｅｄＴｉｍｅ１およびｅｌａｐｓｅｄＴｉｍｅ２で参照する）に相当するデータ転送速度および計算速度を決定する。ここで、「ｅｌａｐｓｅｄＴｉｍｅ１」および「ｅｌａｐｓｅｄＴｉｍｅ２」は、性能分析処理で得られる変数である。
【００３８】
ステップＳ５０２では、ホストは、パラメータ（ｅｌａｐｓｅｄＴｉｍｅ１およびｅｌａｐｓｅｄＴｉｍｅ２）を用いて重なり比を計算する。本実施形態において、「重なり比」とは、計算速度に対する転送速度の比として定義され、好ましくは、データ転送速度と計算速度とが等しい場合に１となる。
【００３９】
続いてステップＳ５０３では、ホストは、ネットワークおよびデバイスの与えられた性能の下で、より高いデータ転送速度を求めつつ、重なり比が１に最も近くなるような、最適化されるべきチャンクのデータサイズを決定する。そして、ホストは、図４中ステップＳ４００で決定された最適なチャンクサイズを用いて、アクセラレータ装置へ転送すべき計算の命令の準備を開始する。
【００４０】
図６は、図５（ステップＳ５００）においてホストからアクセラレータ装置への経路（以下、経路ｈ２ａと参照する）についてのデータ転送コストおよび計算コストを性能分析する詳細な処理を示す。図６に示す処理は、ステップＳ６００から開始され、ホストは、ステップＳ６０１では、テスト計算用データを送信するために用いる書き込み用のグローバル・バッファを割り当て、その後、ホストは、タイマ・オブジェクトを開始する。ホストは、最小サイズ（ＭＩＮ＿ＳＩＺＥ）から最大サイズ（ＭＡＸ＿ＳＩＺＥ）までの範囲でサンプルデータのサイズを、このデータサイズでのデータ転送速度を調査するべくバッファにセットする。ステップＳ６０４では、ホストは、書き込み、つまりアクセラレータ装置内の空のカーネル・プログラムまたはアプリケーション・カーネル・プログラムを呼び出すための操作コマンドを振り分けて、ステップＳ６０５で、所定の繰り返し回数（ＮＵＭ＿ＩＴＥＲ）の計算を実行させる。
【００４１】
ステップＳ６０６で、所定の繰り返し回数（ＮＵＭ＿ＩＴＥＲ）が実行された後（ｙｅｓ）は、ホストは、アクセラレータ装置に投入したすべてのイベントの完了が確認されるまで、終了したか否かを判定する。ステップＳ６０６は、パラメータ（ＮＵＭ＿ＩＴＥＲ）で規定される回数だけステップＳ６０４およびステップＳ６０５の処理を繰り返すことによって、転送速度および計算速度のより精密な値を得るように繰り返されても良い。また用語「イベント」は、ここでは、転送速度および／または計算速度を決定するために用いられる、グローバル・バッファへのデータ書き込みの単位トランザクションおよび後続するカーネル・プログラムの実行として定義される。すべてのイベントが完了すると、ホストは、ステップＳ６０７で、タイマ・オブジェクトを停止させ、ステップＳ６０８で、タイマ・オブジェクトのタイマー値（Ｔｉｍｅｒ＿Ｖａｌｕｅ）を経過時間のパラメータ（ｅｌａｐｓｅｄＴｉｍｅ１またはｅｌａｐｓｅｄＴｉｍｅ２）に設定する。いずれのパラメータに設定されるかは、測定が、転送速度のみを目的とするか、転送速度および計算速度の両方を目的とするかに依存する。２種類の測定が異なる種類のパラメータをセットするが、ソフトウェア・モジュールは２つの測定間で共有されてもよい。
【００４２】
次にステップＳ６０９では、ホストは、サンプルデータのサイズが最大サイズ（ＭＡＸ＿ＳＩＺＥ）に達したかを判定し、サンプルデータが最大サイズ（ＭＡＸ＿ＳＩＺＥ）に達していない場合（ｎｏ）は、ステップＳ６０２に処理が戻され、サンプルデータのサイズが最大サイズ（ＭＡＸ＿ＳＩＺＥ）に達するまで繰り返し実行する。一方、ステップＳ６０９の判定で肯定的な結果が返された場合（ｙｅｓ）は、ステップＳ６１０で処理を終了させて、性能分析を終了する。パラメータｅｌａｐｓｅｄＴｉｍｅ１は、空のカーネル・プログラムが呼び出されたときの全実行時間に相当し、パラメータｅｌａｐｓｅｄＴｉｍｅ２は、サンプル計算のために、アクセラレータ装置からアプリケーション・カーネル・プログラムが呼び出されたときの全実行時間に相当する。空のカーネル・プログラムは、データ転送に要する時間のみを得るべく、何も計算を行わずに単純にホスト・アプリケーションから入力データを受信すると直ちにアクノレッジを返すものである。一方で、アプリケーション・カーネル・プログラムは、入力データを用いてアプリケーションで要求された計算を実行し、カーネル計算の完了によりアクノレッジを返すものである。
【００４３】
非結合型の計算について経路ｈ２ａにおける転送速度は、最小サイズ（ＭＩＮ＿ＳＩＺＥ）および最大サイズ（ＭＡＸ＿ＳＩＺＥ）間のサイズのサンプルデータを用いて空のカーネル・プログラムを呼び出すことによって概算することができ、最適なチャンクサイズは、最もデータ転送速度が速いデータサイズに決定することができる。
【００４４】
図７は、図５（ステップＳ５００）においてアクセラレータ装置からホストへの経路（以下、経路ａ２ｈとも参照する）についてのデータ転送コストおよび計算コストを性能分析する詳細な処理を示す。経路ａ２ｈの転送速度は、アクセラレータからのホストのデータ読み出し速度によって性能分析することができる。本質的に処理は、図６に示した処理と大部分において類似するので、ここでは、詳細な説明は省略する。
【００４５】
本実施形態によれば、データ転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）、計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）および重なり比（ｏｖｅｒｌａｐｐｉｎｇＲａｔｉｏ）といったパラメータは、性能分析の結果から得られ、以下のように定義される：
transfeRate = dataSize * NUM_ITER / elapsedTime1
computationRate = dataSize * NUM_ITER / (elapsedTime2- elapsedTime1)
overlappingRatio = transfeRate / computationRate
ここで、データサイズと繰り返し回数との積（ｄａｔａＳｉｚｅ＊ＮＵＭ＿ＩＴＥＲ）は、パラメータ（ｅｌａｐｓｅｄＴｉｍｅ１またはｅｌａｐｓｅｄＴｉｍｅ２）の測定中に転送されたデータの総量である。
【００４６】
より一般的には、最適なデータサイズは、可能なデータサイズの中で最も速い転送速度を与えながら、値｜１−ｏｖｅｒｌａｐｐｉｎｇＲａｔｉｏ｜が閾値を超えないという条件を満たすデータサイズになるように決定することができる。上記閾値は、転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）および計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）の取り得る範囲にわたり、その重なりの要件を考慮して、可能な限りゼロ（０）に近い値となるように決定することができる。
【００４７】
図８は、転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）８００、計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）８１０およびデータサイズの典型的な関係を示す図である。転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）は、データサイズが小さな場合およびデータサイズが大きい場合の両端において、比較的高いＴＣＰ／ＩＰランタイム・オーバヘッドに起因して、低い値となり、したがって、そのプロファイルは、典型的には、データサイズの中間点で最大の転送速度を示す凸形状を有するようになっている。
【００４８】
一方、計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）は、アプリケーション・カーネル・プログラムのオーバヘッドが増加することに起因して、典型的には、データサイズが増加するにつれ減少する。データサイズの範囲にわたって計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）が転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）よりも高いケースでは、これは従来のネットワーク通信インフラ基盤に典型的なケースであるかもしれないが、交点が存在せず、最適なチャンクサイズは、凸曲線の最大値８４０のデータサイズに一意に決定することで、上記定義した最大の重なり比（ｏｖｅｒｌａｐｐｉｎｇＲａｔｉｏ）を達成することができる。
【００４９】
続いて、転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）および計算速度（ｃｏｍｐｕｔａｔｉｏｎＲａｔｅ）が拮抗する場合は、これは近年のギガビット・イーサネット（登録商標）ネットワーク通信または光通信で起こりうるが、図８に示すように、重なり比（ｏｖｅｒｌａｐｐｉｎｇＲａｔｉｏ）が１に可能な限り近接するという条件を満たす複数の交点が存在する。本発明によれば、交点が複数存在する場合は、より高い転送速度（ｔｒａｎｓｆｅｒＲａｔｅ）を有するデータサイズが最適なチャンクサイズとして採用される。
【００５０】
上記決定されたデータサイズは、疎結合型分散計算システムにおいてデータ転送速度および計算速度を最適化する。上述までは、性能分析処理および最適なチャンクサイズの決定について説明した。以下、本発明における効率的な並列計算のためのデータ処理について説明する。
【００５１】
図９は、データがアクセラレータ装置に転送されたときのデータ処理を示す。図９に示す実施形態においては、疑似コード９００に示すように、データ・バッファおよびサブバッファが、それぞれアプリケーション・データサイズおよび最適なチャンクサイズに基づいて割り当てられる。典型的な実施形態としてｎ＝２を仮定したアプリケーション・データ９１０で示すように、ホスト上のアプリケーション・データサイズが所定の数とチャンクサイズとの積（ｎ＊ｃｈｕｎｋ  ｓｉｚｅ）より大きい場合は、アプリケーション・データは、アクセラレータ装置上のサブバッファに対応して、最適なチャンクサイズを有するｎ個のチャンクに分割される。典型的な実施形態としてｎ＝２を仮定したアプリケーション・データ９２０で示すように、ホスト上のアプリケーション・データサイズが、チャンクサイズを所定数で割った値（ｃｈｕｎｋ  ｓｉｚｅ／ｎ）より小さい場合は、少なくともｎ個のアプリケーション・データが１つの最適なチャンクサイズを有するチャンクに集約される。
【００５２】
その後、最適なチャンクサイズを有するデータは、アクセラレータ装置に転送される。図１０は、アプリケーション・データサイズが最適なチャンクのサイズよりも大きい場合のデータ転送操作を表す。図１０に示す疑似コード１０００は、ＯｐｅｎＣＬ言語での特定の実施形態を記述し、行１０では、ホストは、アプリケーション・データのデータサイズ（ｄａｔａＳｉｚｅ）がチャンクサイズの２倍（ｃｈｕｎｋＳｉｚｅ＊２）より大きいか否かを判定する。データサイズ（ｄａｔａＳｉｚｅ）がチャンクサイズの２倍（ｃｈｕｎｋＳｉｚｅ＊２）より大きい場合は、ホストは、行２０で、サブチャンクの数を決定する。ここで、変数「ｄａｔａＳｉｚｅ」は、ホストからアクセラレータ装置へ転送すべきアプリケーション・データのサイズであり、変数「ｃｈｕｎｋＳｉｚｅ」は、転送されるデータをカプセル化するのに最適なチャンクのサイズである。
【００５３】
続いて、行３０−７０で、ホストは、ホストメモリ１０１０内のアプリケーション・データすべてがアクセラレータ装置のサブバッファ［ｉ］１０２０へ転送されるまで、アプリケーション・データをチャンクサイズに分割して転送する。ホストメモリ内のアプリケーション・データが最適なチャンクサイズよりも小さい場合のデータ集約にも同様のデータ処理が適用される。
【００５４】
図１１は、アクセラレータ装置に転送される際にアプリケーション・データが集約される場合のホストおよびアクセラレータ装置のデータ処理を示す。ホストは、ステップＳ１１００で、ホストメモリ１内のデータおよびホストメモリ２内のデータに対し特定の演算を実行して、アプリケーション・データが最適なチャンクサイズの２分の１より小さくなるようにする。ここでは、ステップＳ１１１０で示すように、変数ｎｕｍが２に設定されている。続いてステップＳ１１２０では、ホストは、アプリケーション・データを最適なサイズのチャンクに集約ないしカプセル化する。そして、ホストは、最適なサイズのチャンクにカプセル化されたアプリケーション・データをアクセラレータ装置に転送する。
【００５５】
ステップＳ１１３０では、アクセラレータ装置は、アプリケーション・データをホストから受信すると、アプリケーション・データをアクセラレータ装置内の集約された数に対応するバッファに格納する。アクセラレータ装置は、ステップＳ１１４０およびＳ１１５０で、アプリケーション・カーネル・プログラムを呼び出して、アクセラレータ装置バッファ１内のデータおよびアクセラレータ装置バッファ２内のデータへのカーネル演算を開始する。
【００５６】
図１２は、従来技術のカーネル計算１２１０および本発明での結合可能型の演算に対するカーネル計算１２２０を実行するためのホストの疑似コードを表す。従来技術の処理１２１０においては、アプリケーション・データは、ホストによって準備されたアプリケーション・データそのままのデータサイズで転送され、アクセラレータ装置は、そのデータを一度に受信する。そしてアクセラレータ装置は、アプリケーション・カーネル／プログラムを呼び出して、カーネル演算を完了させる。
【００５７】
代わりに、本発明の実施形態によれば、疑似コード１２２０で示されるように、アプリケーション・データが最適なチャンクサイズで転送され、アクセラレータ装置は、計算が結合可能型である場合には、各チャンクを受信すると、アプリケーション・カーネル・プログラムを呼び出して、データへの操作を開始する。最後に、アクセラレータ装置は、結合演算を呼び出して、個々のチャンクについて得られた結果を結合する。本処理においては、アクセラレータ装置は、ＴＣＰ／ＩＰネットワークを経由したデータ転送を最適化するとともに、アプリケーション・データへパイプライン計算を適用することによって、アプリケーション・データに対する全実行時間をさらに削減する。
【００５８】
図１３は、従来技術の計算効率と比較して、結合可能計算の場合の本発明のパイプライン計算における改善の仕組みを示す。従来技術の計算１３００では、ホストで準備されたアプリケーション・データは、そのままのアプリケーション・データサイズでアクセラレータ装置に転送される。すなわち、最適化されずに用意されたアプリケーション・データがアクセラレータ装置に転送されるのである。
【００５９】
ブロック１３１０に示すように、アプリケーション・データが最適なサイズで転送される場合、データ転送効率は、本発明に従って改善され得る。しかしながら、ブロック１３２０に示すように、カーネル計算がアプリケーション・データの全体を受信した後に呼び出される場合は、アクセラレータ装置でパイプライン計算が適用されないため、アクセラレータ装置は、実質的には、アプリケーション・データが揃うまで計算リソースを浪費することになる。この場合は、アクセラレータ装置におけるアプリケーション・データに対する計算は、時間（Ｔｉｍｅ１）で終了する。
【００６０】
本発明においては、ホストは、ブロック１３３０において示すように、最適なチャンクサイズを有するチャンク内にアプリケーション・データを生成する。そして、カーネル計算が結合可能型であれば、パイプライン計算の命令が投入され、アクセラレータ装置は、あるデータチャンクを受信すると直ちにアプリケーション・カーネル・プログラムを呼び出して、そのチャンクに対するパイプライン計算を開始させる。アクセラレータ装置がすべてのデータチャンクに対する計算を完了させると、アクセラレータ装置は、個々のデータチャンクに対する部分結果を結合するタスクを呼び出し、時間（Ｔｉｍｅ２）で、アクセラレータ装置に対して割り当てたタスクを完了させる。
【００６１】
カーネル計算がデータチャンクの転送と並列に実行されて、計算時間における浪費が最小化される。ブロック１３４０に示すように、同一のタスクをパイプライン処理で実行するために必要な時間（Ｔｉｍｅ２）は、パイプライン操作を行わない場合の時間（Ｔｉｍｅ１）よりも明らかに短縮され、本発明は、アクセラレータ装置を用いた疎結合型アーキテクチャにおいて分散計算の効率を著しく向上させることができる。
【００６２】
図１４は、本発明を用いた実装システムにおける計算性能の改善結果を示す。ＩＢＭ  ｚＥｎｔｅｒｐｒｉｚｅプラットフォーム（ｚ１９６）と、ＩＢＭ  ＰＯＷＥＲ７を実装するブレードサーバとを１Ｇｂｐｓおよび１０Ｇｂｐｓのイーサネット（登録商標）に接続して疎結合型システムを構築した。
【００６３】
ＳＰＳＳ（インターナショナル・ビジネス・マシーンズ・コーポレーションから提供される。例えば、＜URL=http://www-01.ibm.com/software/analytics/spss/＞を参照されたい）の２ステップ・クラスタリング・アルゴリズムを用いて、ＯｐｅｎＣＬで実装し、実験を行った。実験は、図１３のブロック１３００に示した従来技術の疎結合型システムを含めて、データ分割およびデータ集約の両方について試行した。
【００６４】
図１４において、左側のグラフは、１Ｇｂｐｓネットワーク環境における結果を表し、右側のグラフは、１０Ｇｂｐｓネットワーク環境における結果を表す。両方のグラフにおいて、左側のバーは参照用結果であり、右側のバーは、本発明の結果を表す。図１４に示すように、本発明では、データ分割およびデータ集約の両方のケースについて、参照用結果と比較して実行時間の明瞭な低減が確認された。最適なチャンクサイズは、上述した条件で決定され、１Ｇｂｐｓネットワークについては６４Ｋバイトに決定され、１０Ｇｂｐｓネットワークについては１２８Ｋバイトに決定され、これらは、それぞれアプリケーション・データサイズの４Ｋおよび８Ｋに対応している。
【００６５】
本発明について図面に示した実施形態を参照しながら説明してきた。しかしながら、本発明は、図面に示した実施形態に限定されるものではなく、当業者が導出できる種々の変更または他の実施形態が可能であり、本発明の範囲は、付記する請求項によって定められる。
【図１】
2013140530.tif 000003
【図２】
2013140530.tif 000004
【図３】
2013140530.tif 000005
【図４】
2013140530.tif 000006
【図５】
2013140530.tif 000007
【図６】
2013140530.tif 000008
【図７】
2013140530.tif 000009
【図８】
2013140530.tif 000010
【図９】
2013140530.tif 000011
【図１０】
2013140530.tif 000012
【図１１】
2013140530.tif 000013
【図１２】
2013140530.tif 000014
【図１３】
2013140530.tif 000015
【図１４】
2013140530.tif 000016
【外国語明細書】
D E S C R I P T I O N
[Title of Invention]
DISTRIBUTED PARALLEL COMPUTATION WITH ACCELERATION DEVICES
[Technical Field]
[0001]
The present invention relates to a distributed computing technology, and more particularly relates to a distributed parallel computation using a plurality of accelerator devices.
[Background Art]
[0002]
Recently, a multi-processor computation using GPUs (Graphic Processor Units) is widely used to enhance computation efficiency and／or computation speed.  The GPUs are typically used as accelerators of a main CPU for enhancing the computation performance.  Such multi-processor computation architecture often uses a GPU network connected over internal buses such as PCI or PCI-Express etc.  Such GPUs connected by the internal buses are herein referred to tightly-coupled accelerator devices.
[0003]
The GPUs are operated in parallel under control by a host CPU by an adequate programming language to enhance the computation performance.  One example of such programming language may include OpenCL.  OpenCL may be applied to manage data transfer between the host CPU and GPUs and may be utilized by this invention to minimize the performance cost of that transfer.
[0004]
The multi-processor computation architecture in another scheme has been known such as for example, distributed computation or grid computation.  These multi-processor computation architectures may include a plurality of servers or computers which share computations under control by a host computer or a master computer.  In this type of multi-processor architecture, the computers are connected with an external bus network such as Ethernet (Trade Mark) and a network interface card using various physical connection protocols.  The computers may support the entire computation executed within the network and hence the computers responsible to the distributed computing may also be regarded as the accelerators.  However, the computers in such distributed computation architecture are connected by the external network though TCP／IP and the computers in the distributed computing system may be regarded as loosely-coupled accelerators.
[0005]
In the loosely-coupled multi-processor system, the computers or nodes are connected by the external network and hence, data transfer between the host computer and the accelerator devices may be affected by transport conditions including data sizes, runtime implementation and network conditions.
[0006]
Enhancement of the computation performance through TCP network also has been developed so far and US 2008-029098A1 (Patent Literature 1) discloses a computer system which dynamically segments a large TCP segment with smaller TCP segments so as to reduce interrupt frequency.  JP2011-170732 (Patent Literature 2) discloses the parallel computation method which divides a functional block into strands and modifies the functional block depending on computation time.
[0007]
In the tightly-coupled acceleration architecture, it has been proposed that batching many small transfers into one larger transfer will improve the data transfer performance (reference NVIDIA OpenCL Best Practice Guide, Section 3.1 “Data Transfer between Host and Device”).  In addition, Kim et.al. discloses, in “A New Communication and Computation Overlapping Model with Loop Sub-Partitioning and Dynamic Scheduling”, a communication and computation overlapping model to hide the communication latency in data parallel programs.
[Prior Art Literature]
[Patent Literature]
[0008]
[Patent Literature 1] US 2008-029098A1
[Patent Literature 2] JP 2011-107732
[Non-patent Literature] 
[0009]
[Non-patent Literature 1]
Junghwan Kim, Sangyong Han, Heungwan Kim, and Seungho Cho, A New Communication and Computation Overlapping Model with Loop Sub-Partitioning and Dynamic Scheduling
[Summary of Invention]
[Object of Invention]
[0010]
For the parallel computing architecture developed thus far, enhancement to reduce latency of the data transport through the network before computation can begin has not been fully addressed.  Further, connecting fundamentally different platform architectures (such as Power PC (Trade Mark) architecture, Intel (Trade Mark) architecture, and／or a mainframe architecture such as IBM SYSTEM 390 etc.) must override the differences between the computing platforms while optimizing the performance of the parallel computation through the loosely-coupled accelerators running application across the platforms.
[0011]
An object of this invention is to provide a novel distributed parallel computation architecture which will mitigate differences between incongruous platforms comprising the application environment.
[0012]
Still another object of the present invention is to provide a novel distributed computation architecture which may optimize data transfer rate and performance of the parallel computation.
[0013]
Further another object of the present invention is to provide a method, a program, a host computer and a system for executing parallel distributed computation for overriding differences between the platform architecture.
[Means for Addressing Object]
[0014]
According to the present invention, the host computer and the accelerator devices are interconnected through the external network and the host optimizes a data size used for distributed computation in the present environment prior to transmitting application working set data to the accelerator devices.  This optimal data size, termed the "chunk size", is based on actual measurements of the communication rate and the computation rate the accelerator device in the network.     
[0015]
The host computers splits the application data into segments to encapsulate the data into the optimal chunk and then dispatches the encapsulated data to the accelerator devices as well as instructions for parallel computation for the encapsulated data.
[0016]
The accelerator devices start the computation upon receiving each encapsulated data in parallel to the data transfer without waiting the entire application data such that the parallel computation efficiency may be improved by minimizing communication overhead caused by the data transfer through the loosely-coupled network.
In other words, each accelerator device receives the encapsulated data and instructions and, if necessary, waits for all data required for the computation to arrive (as dictated by the host instructions). Once the datum is assembled, the particular computation dictated by the host instructions (that is, the "kernel") is dispatched. Upon completion, output data is sent by the accelerator device to the host, again according to host encapsulation instructions.
[Brief Description of Drawings]
[0017]
Fig. 1 shows data transfer costs of the distributed computing in the loosely-coupled system will be explained depending on OSI basic reference model.
Fig. 2 shows the block diagram of the sample distributed computing system 200 of the present invention.
Fig. 3 shows the functional block diagram 300 of the platforms for constructing the loosely-coupled system.
Fig. 4 shows the flowchart of the process of the present distributed computing system.
Fig. 5 shows the detailed process of the step S400 “Find optimal chunk size” in 
Fig. 6 shows the detailed process of profiling the data transfer cost and computation cost for the host-to-accelerator path in the step of S500 of Fig. 5
Fig. 7 shows the detailed process of profiling the data transfer cost and computation cost for the accelerator-to-host path in the step S500 of Fig. 5
Fig. 8 shows the exemplary relations between the transferRate 800, the computationRate 810, and the datasize.
Fig. 9 shows an example of data buffer allocation for the accelerator devices, depending on the optimal chunk size and application data size.
Fig. 10 shows an example of transferring the data in the case that the application data is larger than the size of optimal chunk.
Fig. 11 shows an example of the data processing of the host and the accelerator device when the application data is smaller than the size of optimal chunk and thus to be aggregated upon transferring to the accelerator device.
Fig. 12 depicts the pseudo codes of the host 1200, transforming the conventional kernel computation 1210 to the present invention kernel computation 1220.
Fig. 13 shows the scheme of the improvement on the pipeline computation of the present invention Fig. 4.
Fig, 14 shows the result of improvement of the computation performance in the system implemented with the present invention.
[Embodiment for Practicing Invention]
[0018]
Hereafter, the present invention will be described depending on particular embodiments; the described embodiments should be understood mere description of the best mode and should not be limitation of the present invention.  Now referring to Fig. 1, data transfer costs of the distributed computing in the loosely-coupled system will be explained depending on OSI basic reference model.  Each of the nodes in the system generally comprises the physical layer 103， a network layer 102, a transport layer 101 and an application layer 100.  The physical layer 103 is responsible for receiving and sending ethernet (Trade Mark) frames; the network layer 102 is responsible to form IP packets with proper data sizes as payloads; the transport layer 101 forms TCP packets from data processed by the application in the application layer 100.  The application layer 100 pays its cost as the runtime overhead by the computation of data with the applications.
[0019]
The transport layer 101, the network layer 102 and the physical layer 103 contribute their respective processing costs and the overall cost to the application may depend on the data size transferred from the node to another node as TCP／IP packets.  The above data size dependency, as opposed to the tightly-coupled system, may create extra costs for computation as well as scalability of the computation.
[0020]
Fig. 2 shows the block diagram of the sample distributed computing system 200 of the present invention.  The system 200 comprises various computing platforms such as, for example, the main frame 201, the servers 202, the rack-mounted servers 203 in which the blade type servers are disposed in the rack.  The main frame 201 may be selected from z system such as IBM system 390; however, other main frame with architecture may be used.  The server 202 and the blade servers in the rack-mounted server 203 may comprise a CPU／CPUs such as IBM POWER and POWER PC (Trade Mark) which may be implemented as POWER (Trade Mark) series.
[0021]
On the other hand, the platform 204 may comprise the CPU／CPUs such as Intel XEON (Trade Mark) which may conventionally referred to x86 architecture.  The platforms depicted in Fig. 2 are each installed with different operating system and application programs so as to serve requested services from operators or users.  The platforms are connected by an adequate network 220 such as Ethernet (trade Mark) or FTTH using optical communication through the hub and／or switch 210 and the router 230.  The platforms may further comprise a RAM, a ROM, a hard disk drive, a network interface card in Giga bit rate etc. for providing business application services to operators.
[0022]
The platforms communicate the data, commands and messages through TCP／IP protocol through the network and in the present invention, the platforms provide distributed computation environment as a loosely-coupled architecture.  Although the distributed computation may be possible by using any known protocols or techniques, the OpenCL architecture may be useful in order to override the differences between the platforms while using already existing application resources.
[0023]
Fig. 3 shows the functional block diagram 300 of the platforms for constructing the loosely-coupled system.  The platform 310 has a role of the host and the platform 320 has the role of the accelerator device.  The term “accelerator device” or more simply referred to “accelerator” herein refers a separated computer or an information processing apparatus which communicate with the host to accelerate the entire computation of the host.  The host comprises as its functional means the host application 311, the application kernel programs 312 implemented in the host 310, the runtime environment 313, and the socket interface 314.  
[0024]
The host application 311 comprises a profiler part 311a, a size optimizer part 311b, a capsulation part 311c, and a dispatcher part 311d.  The profiler part profiles the data transfer rate and the computation rate for a range of data sizes from MIN_SIZE to MAX_SIZE; here, MIN_SIZE and MAX_SIZE are possible data sizes transferred between the host and the accelerator device in the application program.  The size optimizer part 311b finds and determines an optimal chunk size for the data transfer through the network based on the profiled data transfer rate and the computation rate.
[0025]
The capsulation part 311c encapsulates the data stored in a memory of the host computer by splitting or aggregating into a chunk with the optimal chunk size.  The dispatcher part 311d dispatches the encapsulated data to the accelerator device together with instructions for pipeline computation to the accelerator device with respect to the encapsulated data received.  The profiler 311a, the size optimizer 311b, the capsulation 311c, and dispatcher 311d may not be necessarily part of Host application 311, but some of them may be a part of runtime environment 313.
[0026]
The host application 311 provides various services to users controlled by the kernel program 312 and in the described embodiment, the OpenCL APIs may be implemented as one module of the host application.  The kernel program 312 may be implemented for ZOS, Unix (Trade Mark), Linux (Trade Mark), or Windows (Trade Mark) 200X server to control various operation of the host.
[0027]
Particular embodiment of the host includes OpenCL kernel programs as one component of the kernel 312 for implementing loosely-coupled system.  The runtime environment 313 comprises runtime libraries including OpenCL runtime, dynamic link libraries which support runtime condition of the host.  The socket interface 314 transmits to the accelerator device 320 the TCP／IP packets by using socket communication and in the described embodiment, the socket send／receive, RDMA R／W methods or class may be implemented as one module for supporting OpenCL architecture.
[0028]
The accelerator device 320 comprises the application kernels 321, buffer objects 322, the runtime environment 323, the proxy facilities 324 to the host 310, and the socket interface 325.  The application kernels 321 provide requested services and the buffer objects 322 store various information used in the accelerator device 320 and in the described embodiments, the buffer objects 322 receive the input data and the commands as well as send the computed results from the application kernel program 321 as the accelerator device of the loosely-coupled system.
[0029]
The runtime environment 323 comprises as its functional component OpenCL compiler and runtime libraries suitable for the architecture of the platform as well as executes the operation instructed by the host 310 together with the proxy facilities 324 being implemented with OpenCL functions.  The socket interface 325 communicates to the host through the socket communication and comprises send／receive, RDMA R／W method or class to the host 310 through the network 330.
[0030]
Thus, the loosely-coupled system includes two type of data processing costs; one is the computation and frame formation costs and the other is data transfer cost through the network.  The two kinds of costs i.e., computation and data transfer may be preferably well balanced for smooth and efficient computation in the distributed computing environments to minimize wait time for communication latency.  In other words, the data transfer rate and the computation must be optimized so as not to waste the hardware resources in the accelerator devices.
[0031]
Fig. 4 shows a flowchart of the process of the present distributed computing system.  The process of Fig. 4 starts from the step S400 and the host examines the network conditions and hardware conditions of the accelerator devices to determine the optimal chunk size in the system.  Although the detail of the step S400 will be described elsewhere, the host measures a response time of a test computation by the accelerator devices through the network in the step S400.  In the step S401, the host allocates buffers and sub-buffer sizes for transferring the data.  Here, the size of the buffer may be determined by the data size of the objected computation and the sub-buffer size may be set to the optimal chunk size, or vice versa, depending on whether data splitting or aggregation is applied in the described preferred embodiment.
[0032]
In the step S402, the data for the objected computation may be divided or aggregated such that the data size transferred to the accelerator device may be contained in the sub-buffer with the optimized size.  Next in the step S403, the host transmits the data with the optimal size together with the commands or instructions for computation in the accelerator device.  In the described embodiment, the instruction of the computation on the accelerator device is coded by OpenCL language; however, other any distributed computation protocols may be used.
[0033]
In the step S404, the host determines whether or not the data splitting is applied and the application kernel task allocated to the accelerator device is associative, that is to say, whether the task can be applied with pipelined execution for split data.  If so (yes), the host transfers instructions for initiating the pipelined computations to the accelerator device so as to execute the computation in the accelerators overlapped with data communication.  If the task is not suitable to the pipeline computing (no), the host transfers conventional loosely-coupled computation instruction codes without pipelining and the process of Fig. 4 flows to the step S405 to the accelerator devices.
[0034]
The steps S405-S408 are processes executed in the accelerator devices responsible for the computations.  The accelerator devices receive the data and the instructions in the step S405, the accelerator device waits until all of the data chunks are received because the computation is not associative and the speculative initiation of the computation is illegal.  When the process flows to the step S405, the accelerator device starts the computation after receiving the entire data for the computation and then returns the results of the computation to the host in the step S408.
[0035]
Now, when the determination in the step S404 returns an affirmative result (yes), the process goes to the step S406 and the accelerator device initiates the execution of the pipelined computation over the data chunks and the instructions received sequentially with the optimized computation-communication overlapping.  The accelerator device continues the pipelined computation over the sequentially transmitted data chunks to the last and then the accelerator device invokes the associative operation to combine the partial results with the pipelined computation.  Then, the accelerator device returns the computed results to the host to complete the commands received from the host.
[0036]
Fig. 5 shows the detailed process of the step S400 “Find optimal chunk size” in Fig. 4.  The process begins from the Step S500 and the host starts to profile the data transfer cost and the computation pipeline effect in a given runtime and network environment for both host-to-accelerator and accelerator-to-host paths independently.  Then in the step S501, the host determines the data transfer rate and computation rate using proper sample computation set which are represented time duration referred to as elapsedTime1 and elapsedTime2; here elapsedTime1 and elapsedTime2 are variables obtained in the profiling process.
[0037]
Then the host computes in the step S502 the overlapping ratio using the parameters elapsedTime1 and elapsed Time2.  The overlapping ratio is defined in the present embodiment that the ratio of transfer rate to computation rate which may be preferably one (1) when the data transfer rate and the computation rate are equal.
[0038]
Further next, the host determines in the step S503 the data size of the chunk to be optimized under the network-device performances such that the overlapping ratio may become closest to one (1) while providing higher data transfer rate.  The host starts to prepare the instruction of computation to be transferred to the accelerator device using the optimal chunk size determined in the step S400 in Fig. 4.
[0039]
Fig. 6 shows the detailed process of profiling the data transfer cost and computation cost in Fig. 5 (S500) for the host-to-accelerator (h2a) path.  The process of Fig. 6 begins from the step S600 and the host allocates the global buffer for write used for transferring the test computation data in the step S 601, and then the host starts the timer object.  Then, the host sets the size of sample data ranging from MIN_SIZE to MAX_SIZE in the buffer in order to find data transfer rate on the data size.  In the step S604, the host writes, i.e., dispatches the operation commands to invoke an empty kernel program or an application kernel program in the accelerator devices so as to execute the computation for a predetermined number of iterations NUM_ITER in the step S605.
[0040]
After executing a predetermined number of iterations (NUM_ITER) in the step S606 (yes), the host determines whether or not all events dispatched to the accelerator device end until terminations of all events are confirmed.  The step S606 may be executed so as to obtain more precise values of the transfer rate and the computation rate by repeating the process of S604 and S605 with the number defined by the parameter NUM_ITER.  The term “events” is herein defined as a unit transaction of data write to the global buffer and subsequent execution of kernel program used to determine the transfer rate and/or computation rate.  When all events complete, the host stops the timer object in the step S607 and the Timer_value of the Timer object is set to the parameter elapsedTime1 or elapsedTime2 in the step S608, depending on whether the purpose of measurements is for transfer rate only or both transfer rate and computation rate: the two kind of measurements sets different kind of parameters; however, the software module used therefore may be shared between the two measurements.
[0041]
Next, the host examines whether or not the sample data size reaches MAX_SIZE and when the sample data does not reach MAX_SIZE in the step S609 (no), the process is reverted to the step S602 to repeat until the sample data size reaches MAX_SIZE.  When the determination returns an affirmative result in the step S609 (yes), then the process ends at the step S610 to terminate the profiling.  The parameter elapsedTime1 represents the total execution time when an empty kernel program is invoked and the parameter elapsedTime2 represents the total execution time when an application kernel program is invoked for the sample computation from the accelerator device, i.e., accelerator.  The empty kernel program simply returns acknowledgements as soon as receiving input data from the host application without executing any computation in order to obtain the time required only for the data transfer, while the application kernel program executes the required computation in the application using the input data and returns the acknowledgements upon completion of the kernel computation.
[0042]
In the h2a path for the non-associative computation, the transfer rate may be estimated by invoking an empty kernel program using the sample data of the size between MIN_SIZE and MAX_SIZE and the optimal chunk size may be determined as the data size which shows the highest data transfer rate.
[0043]
Fig. 7 shows the detailed process of profiling data transfer cost and computation cost in Fig. 5 (S500) for the accelerator-to-host (a2h) path.  The transfer rate of the a2h path may be profiled by the data read rate of the host from the accelerator.  The principal processes are almost similar to the process of Fig. 6 and hence further detailed explanation will be omitted.
[0044]
According to the present embodiment, the parameters such as data transferRate, computationRate, and overlappingRatio are obtained from the results of the profiling and defined as follows:
transferRate=datasize*NUM_ITER／elapsedTime1
computationRate=datasize*NUM_ITER／(elapsedTime2-elapsedTime1)
overlappingRatio=transferRate／computationRate
wherein datasize*NUM_ITER is the total amount of data transferred during the measurements of elapsedTime1 or elapsedTime2.
[0045]
More generally, the optimal data size may be determined to be the data size satisfying the condition that the value |1?overlappingRatio| is not more than a threshold while providing the highest transfer rate among the possible data sizes.  The above threshold may be determined to be as close as zero (0) considering the overlapping requirement over the ranges of the transferRate and computationRate.
[0046]
Fig. 8 shows an exemplary relation between the transferRate 800, the computationRate 810, and the data size.  The transferRate may be low both at the small data size end and the large data size end because of relatively higher TCP／IP runtime overheads such that the profile of the transferRate may typically be a convex showing the highest transfer rate at a medium data size point.
[0047]
On the other hand, the computationRate may typically get lower with respect to the increase of the data size because the overhead of application kernel program can increase.  In the case that the computationRate is higher than transferRate over the range of data size, which may be typically a case in the conventional network communication infra base, the crossing point is not present and then the optimal chunk size may be uniquely determined to be the data size of the maximum 840 of the convex curve so as to attain the largest overlappingRatio as defined above.
[0048]
In turn, in the case that the transferRate and the computationRate are almost comparative, which is expected to occur in recent Gigabit Ethernet (Trade Mark) network communications or optical communications, it may have multiple crossing points as shown in Fig. 8; both of the crossing points satisfy the condition that the overlappingRatio to be as close as one (1).  When the crossing points are present, the data size which has higher transferRate is adopted as the optimal chunk size according to the present invention.
[0049]
The above determined data size may optimize the data transfer rate and the computation rate in the loosely-coupled distributed computing system.  Herein above, the profiling process and the determination of the optimal chunk size has been explained.  Now, the data processing for efficient parallel computation in the present invention will be explained.
[0050]
Fig. 9 shows data processing when the data are transferred to the accelerator devices.  In the embodiment of Fig. 9, the data buffers and sub-buffers are allocated based on the application data size and the optimal chunk size as shown in the pseudo code 900.  When the application data size on the host is larger than (n*chunk size), the application data are split into at least n chunks corresponding to the sub-buffers on the accelerator device with the optimal chunk size as the application data 910 where the exemplary embodiment is explained by assuming n=2.  When the application data size on the host is smaller than (chunk size／n), at least n application data are aggregated into one chunk with the optimal chunk size as the application data 920 where the exemplary embodiment is also explained with n=2.
[0051]
Then, the data with the optimal chunk size are transferred to the accelerator device. Fig. 10 depicts the data transfer operation in the case that the application data size is larger than the size of optimal chunk.  The pseudo code 1000 shown in Fig. 10 is described as a particular embodiment in OpenCL language and at line 10, the host determines whether or not the size of the application data dataSize is larger than chunkSize*2.  When the dataSize is larger than chunkSize*2, the host determines at line 20 the number of sub-chunks.  Here, the variable “dataSize” is the application data size to be transferred from the host to the accelerator device and the variable “chunkSize” is the optimal size of the chunk for encapsulating the data transferred.
[0052]
Subsequently, the host transfers the application data at lines 30-70 by splitting the application data into the chunkSize until all of the application data in the host memory 1010 are transferred to the sub-buffer [i] 1020 of the accelerator device.  Similar data processing for the data aggregation may be applied to the application data in the host memory when the application data size is smaller than the optimal chunk size.
[0053]
Fig. 11 shows the data processing of the host and the accelerator device when the application data are aggregated upon transferring to the accelerator device.  The host executes a particular operation on the data in the host memory 1 as well as the data in the host memory 2 in the step S1100 so that the application data being smaller than one-half of the optimal chunk size; here the variable num is set to be equal to two (2) as shown at S1110.  The host subsequently aggregates or encapsulates the application data into the chunk of optimal size in the step S1120.  The host then transfers the application data encapsulated into the chunk of optimal size to the accelerator device.
[0054]
When the accelerator device receives the application data from the host, the accelerator device stores the application data in the accelerator device buffers corresponding to the numbers of aggregated data in the step S1130.  Then the accelerator device invokes application kernel programs to start kernel operation on the data in the accelerator device buffer 1 and in the accelerator device buffer 2 in the step S1140 and S1150.
[0055]
Fig. 12 depicts the pseudo codes of the host 1200 for executing the conventional kernel computation 1210 and the present invention kernel computation 1220 for associative operation.  In the conventional process 1210, the application data are transferred in the data size determined by the application data prepared by the host and the accelerator device receives the data at one time; then accelerator device invokes the application kernel program to complete the kernel operation.
[0056]
Alternatively according to the present invention provided as the pseudo code 1220, the application data are transferred with the optimal chunk size and the accelerator device invokes the application kernel programs to start the operation on the data upon receiving each chunk if the computation is associative. At the end, the accelerator device invokes the combination operation to combine the results obtained for the different chunks.  In the present process, the accelerator device may apply pipeline operation on the application data so that the entire execution time on the application data may further be reduced together with the optimization of data transfer through TCP／IP network.
[0057]
Fig. 13 shows the scheme of the improvement on the pipeline computation of the present invention with referencing the conventional computation efficiency in the case of the associative computation.  In the conventional computation 1300, the application data prepared in the host are transferred to the accelerator device with the application data size as is.  The prepared non-optimized application data are transferred to the accelerator device.
[0058]
When the application data are transferred with the optimal size as shown in the block 1310, the data transfer efficiency may be improved according to the present invention; however, when the kernel computation are invoked upon receiving the whole application data as shown in the block 1320, the accelerator device substantially wastes the computation resources until the application data has been completed because the pipeline computation is not applied in the accelerator device.  In this case, the computation on the application data in the accelerator device ends at the time of Time 1. 
[0059]
In the present invention, the host generates the application data in the chunks with the optimal size as shown in the block 1330 and, if the kernel computation is associative, dispatches the instruction of pipeline computation such that the accelerator device immediately invokes the application kernel programs upon receiving a data chunk and starts the pipeline computation about the chunk.  When the accelerator device completes the computation on all of the data chunks, the accelerator device invokes a task to combine those partial results on data chunks and to complete the task allocated to the accelerator device at the Time 2.
[0060]
While the kernel computations are executed in parallel with the transmission of the data chunk such that the waste in the computation time may be minimized.  As shown in the block 1340, the time required to execute the same task with pipelining, Time 2, is significantly shorter than Time 1 without pipeline operation and the present invention may significantly improve efficiency of the distributed computation with loosely-coupled architecture together with the usage of the accelerators.
[0061]
Fig, 14 shows the result of improvement of the computation performance in the system implemented with the present invention.  The loosely-coupled system was constructed by IBM z／Enterprise platform (z196) and a blade server implemented with IBM POWER 7 connected with Ethernet (Trade Mark) of 1G bps and 10G bps.
[0062]
The experimental was conducted by using Two-Step Cluster algorithm from SPSS (from IBM Corporation, e.g. referenced by the URL= http:／／www-01.ibm.com／software／analytics／spss／) with the OpenCL implementation.  The experiments were conducted for data splitting and data aggregating trials with the conventional loosely-coupling system represented by block 1300 in Fig. 13.
[0063]
In Fig. 14, the left hand side graph represents the results in 1 Gbps network environment and the right hand side graph represents the results in 10 Gbps network environment.  In the both graphs, left-bars correspond to references and the right-bars correspond to the results of the present invention.  As shown in Fig. 14, the present invention provide significant reduction of execution time when compared to the reference both in the data splitting case and the data aggregation case.  The optimal chunk size was determined in the above condition to be 64K bytes for 1G bps network and 128K bytes for 10G bps network, which corresponds to 4K and 8K of the application data size, respectively.
[0064]
The present invention has been described with reference to the embodiments shown in the drawings.  However, the present invention is not limited to the embodiments shown in the drawings and various changes or other embodiments may be possible which are derived by a person skilled in the art, the true scope of the present invention may be determined by appended claims.
C L A I M S
Claim 1     A method for distributed computing between a host computer and at least one accelerator device interconnected through a network, the method comprising the steps of:
profiling a data transfer rate and a computation rate for a range of data sizes to find an optimal chunk size for the data transfer through the network;
splitting or aggregating a size of the data stored in a memory in the host computer for encapsulating the data into a chunk with the optimal chunk size;
dispatching the encapsulated data to the accelerator device; and
instructing pipeline computation to the accelerator device with respect to the encapsulated data received.
Claim 2     The method of claim 1, wherein the ratio between the data transfer rate and the computation rate is computed from actual transmission time durations.
Claim 3     The method of claim 1, wherein the optimal chunk size is determined to be the data size which has the highest data transfer rate between a minimum data size and a maximum data size transmitted during the profiling step in relation to the computation rate. 
Claim 4     The method of claim 1, wherein the optimal chunk size is determined to be the data size where the computation and communication overlapping ratio is closest to 1 and, if there are multiple such data sizes, which has the highest data transfer rate between a minimum data size and a maximum data size.
Claim 5     The method of claim 1, wherein the instructing pipeline computation further comprises the step of:
instructing multiple writes of the encapsulated data for numbers of chunks into buffer objects in the accelerator device; and
instructing execution of operations of the accelerator device on the encapsulated data upon receiving the encapsulated data.
Claim 6     The method of claim 1, wherein the data size is subjected to the splitting in a case where the data size is larger than the optimal chunk size and the data size is subjected to the aggregating in a case where the data size is smaller than the optimal chunk size for the encapsulation of the data.
Claim 7     The method of claim 1, wherein the accelerator device is selected from a computer being implemented with an application program.
Claim 8     The method of claim 1, wherein the accelerator device is networked to the host computer with the TCP／IP network.
Claim 9     The program for distributed computing between a host computer and at least one accelerator device interconnected through a network, the program making the host computer execute the steps of:
profiling a data transfer rate and a computation rate for a range of data sizes to find an optimal chunk size for the data transfer through the network;
splitting or aggregating a size of the data stored in a memory in the host computer for encapsulating the data into a chunk with the optimal chunk size;
dispatching the encapsulated data to the accelerator device; and
instructing pipeline computation to the accelerator device with respect to the encapsulated data received.
Claim 10     The program of claim 9, wherein the ratio between the data transfer rate and the computation rate is computed from actual transmission time durations.
Claim 11     The program of claim 9, wherein the optimal chunk size is determined to be the data size which has the highest data transfer rate between a minimum data size and a maximum data size transmitted during the profiling step in relation to the computation rate.
Claim 12     The program of claim 9, wherein the optimal chunk size is determined to be the data size where the computation and communication overlapping ratio is closest to 1 and, if there are multiple such data sizes, which has the highest data transfer rate between a minimum data size and a maximum data size.
Claim 13     The program of claim 9, wherein the instructing pipeline computation further comprises the step of:
instructing multiple writes of the encapsulated data for numbers of chunks into buffer objects in the accelerator device; and
instructing execution of operations of the accelerator on the encapsulated data upon receiving the encapsulated data.
Claim 14     The program of claim 9, wherein the data size is subjected to the splitting in the case where the data size is larger than the optimal chunk size and the data size is subjected to the aggregation in the case where the data size is smaller than the optimal chunk size for the encapsulation of the data.
Claim 15     The program of claim 9, wherein the accelerator device is selected from a computer being implemented with an application program and the accelerator device is networked to the host computer with the TCP／IP network.
Claim 16     A host computer for distributed computing between the host computer and at least one accelerator device interconnected through a network, the host computer comprising:
a profiler part profiling a data transfer rate and a computation rate for a range of data sizes;
a size optimizer part for finding an optimal chunk size for the data transfer through the network from the profiled data transfer rate and the computation rate;
a capsulation part for encapsulating the data stored in a memory of the host computer by splitting or aggregating into a chunk with the optimal chunk size; and 
a dispatch part for dispatching the encapsulated data to the accelerator device and instructing pipeline computation to the accelerator device with respect to the encapsulated data received.
Claim 17     The host computer of claim 16, wherein the optimal chunk size is determined to be the data size which has the highest data transfer rate between a minimum data size and a maximum data size transmitted by the profiler part in relation to the computation rate.
Claim 18     The host computer of claim 16, wherein the optimal chunk size is determined to be the data size where the computation and communication overlapping ratio is closest to 1 and, if there are multiple such data sizes, which has the highest data transfer rate between a minimum data size and a maximum data size.
Claim 19     The host computer of claim 16, wherein the dispatch part instructs multiple writes of the encapsulated data for numbers of chunks into buffer objects in the accelerator device and instructs execution of operations of the accelerator on the encapsulated data upon receiving the encapsulated data.
Claim 20     A distributed computing system between a host computer and at least one accelerator device interconnected through a TCP／IP network, the accelerator device being implemented with an application program, the host computer comprising:
a profiler part profiling a data transfer rate and a computation rate for a range of data sizes;
a size optimizer part for finding an optimal chunk size for the data transfer through the network from the profiled data transfer rate and the computation rate;
a capsulation part for encapsulating the data stored in a memory of the host computer by splitting or aggregating into a chunk with the optimal chunk size; and 
a dispatch part for dispatching the encapsulated data to the accelerator and instructing pipeline computation to the accelerator with respect to the encapsulated data received
wherein the optimal chunk size is determined to be the data size where the computation and communication overlapping ratio is closest to 1 and, if there are multiple such data sizes, which has the highest data transfer rate between a minimum data size and a maximum data size; and
wherein the dispatch part instructs multiple writes of the encapsulated data for numbers of chunks and instructs execution of operations of the accelerator on the encapsulated data upon receiving the encapsulated data.
A B S T R A C T
[Abstract]
[Object] To maximize performance in distributed parallel computation using a plurality of accelerator devices.
[Means For Addressing Object]  A host computer of the present invention comprises a profiler part profiling a data transfer rate and an computation rate for a range of data sizes; a size optimizer part for finding an optimal chunk size for the data transfer through the network from the profiled data transfer rate and the computation rate; a capsulation part for encapsulating the data stored in a memory of the host computer by splitting or aggregating into a chunk with the optimal chunk size; and a dispatch part for dispatching the encapsulated data to the accelerator device and instructing pipeline computation to the accelerator device with respect to the encapsulated data received.
[Selected Drawings]  Fig. 3
2013140530.tif 000017
2013140530.tif 000018
2013140530.tif 000019
2013140530.tif 000020
2013140530.tif 000021
2013140530.tif 000022
2013140530.tif 000023
2013140530.tif 000024
2013140530.tif 000025
2013140530.tif 000026
2013140530.tif 000027
2013140530.tif 000028
2013140530.tif 000029
2013140530.tif 000030