DEIM Forum 2016 D5-1

汎用分散処理システムでの大規模分散システムシミュレーション
加藤 裕也†

杉野 好宏†

華井 雅俊†

首藤一幸†† 東京工業大学情報理工学研究科 〒 152-8552 東京都目黒区大岡山 2-12-1
E-mail: †kato.y.ay@m.titech.ac.jp
あらまし

今日，インターネット上では様々な分散システムが運用されている．大規模なネットワークで動作する分散システムの場合，システムの挙動を把握するために，しばしばシミュレーションによる検証が行われる．大規模なシミュレーションのために，複数の計算機を用いる場合，汎用分散処理システムを利用することで耐故障処理やジョ
ブスケジューリングといった機能を享受できる．本研究では，汎用分散処理システム上で分散システムシミュレーショ
ンを行い，その妥当性を評価した．また，楽観的時刻管理アルゴリズムを用いた場合のメモリ不足への対応策として，投機的実行を時間窓で制限する手法を適用し，その評価を行った．
キーワード

シミュレーション，MapReduce，分散システム，並列離散事象シミュレーション

1. は じ め に
今日，インターネット上では様々な分散システムが運用されている．他の情報システムがそうであるように，分散システム

シミュレーション規模の拡大が制限されうる．本研究では，現在アルゴリズムレベルのシミュレーションを想定しているが，アンダーレイネットワークも含めた詳細なシミュレーションにも今後取り組んでいく．

においても，開発時点でシステムが運用先でも正しく動作する

オーバーレイネットワークを対象としたシミュレータにことを確認する必要がある．システムの運用先での挙動を確認

Overlay Weaver [5], [6] や PeerSim [7], [8], dPeerSim [9] があするためには，実環境に近い環境を構築した上で検証を行う

る．Overlay Weaver では，シミュレーションのために実装し方法と，実環境に近い仮想環境をシミュレータ上に構築してシたアルゴリズムをそのまま実ネットワーク上で動作させること

ミュレーションを行う方法の 2 つが考えられる．

ができる．最大 100 万ノード程度のシミュレーションが可能で

現在動作している分散システムの 1 つである BitTorrent

あり，複数の計算機での分散シミュレーションも可能である．

Mainline DHT のノード数は 1,000 万ノード規模に達してい

PeerSim では，100 万ノードでのクエリ・サイクル・シミュレー

る [1] ．また，インターネットに接続している機器は 2013 年時

ションが可能である．PeerSim の分散版である dPeerSim はシ

点で 100 億台を突破しており，2020 年には約 500 億台に達す

ミュレート対象の大規模化に対応するため，効率よくスケールると予測されている [2] ．現在インターネットに接続されていアウトするよう改良された．しかし，非分散の PeerSim がシないモノが今後接続されるようになることを想定すると，分散ミュレート可能な範囲では，dPeerSim は同じシミュレーショ

システムの規模は今後も更に大きくなっていくものと考えられンの実行に 100 倍以上の時間を要することもある．

る．大規模な分散システムの場合，実環境に近い環境を構築す

大規模な分散システムを現実的な時間でシミュレートするにることは困難である．シミュレータを用いる場合も，複数の計は，シミュレーションに用いる計算機を増やしてメモリ容量や算機による分散処理が必要となる．

処理速度を稼ぐ必要がある．多くの既存シミュレータでは数

Hadoop MapReduce や Spark のような汎用分散処理システ

十台を超える規模の計算機クラスタで動作することを想定しムは，耐故障処理やジョブスケジューリングといった機能を提供てない．dPeerSim の例から，性能を保ちながら既存のシミュする．汎用分散処理システム上でシミュレータを実現することレータを大規模なシミュレート対象に対応できるよう分散化すで得られるメリットは多い．今回，汎用分散処理システム上でることは困難である．また，規模の大きな計算機クラスタでの大規模分散システムシミュレーションの評価として，Hadoop

は，シミュレーション中に一部の計算機が停止することも考え

MapReduce 及び Spark 上での分散システムシミュレータの評られる．その場合，一部の計算機が停止したのがどのタイミン

価と比較を行った．また，汎用分散処理システム上での大規模グであっても，シミュレーションを最初から実行し直すこととシミュレーションにおけるメモリ不足への対応策として，時刻なる．そのため，大規模分散システムを対象としたシミュレー

管理アルゴリズムの最適化手法を適用し，その評価を行った．

タは，スケーラビリティと耐故障性が不可欠である．その点で2. 関 連 研 究
アンダーレイネットワークを対象としたシミュレータに ns-

汎用分散処理システム上で動作するシミュレータに価値があると考える．汎用分散処理ステム上のシミュレータには Hadoop

MapReduce での実装 [10] があり，本論文で用いる．

2 [3]，ns-3 [4] がある．その延長上として分散システムシミュ

時刻管理アルゴリズムに関する研究は様々あり，本研究でもレーションが可能である．アンダーレイネットワークのシミュ

それらの扱う手法の内，いくつかを適用している．それらにつレーションには大きな計算リソースが必要であり，これによりいては付録や提案手法の項で言及する．

るイベントをタイムスタンプに従って処理を行う．このように

iteration ごとにノード ID 単位でのデータの集約とイベント処理を行うことで，メッセージの送受信を表現することができる．
最終的に未実行イベントがなくなるか，iteration 回数が一定の値に達するまで繰り返す．

3. 2 実装本論文では，汎用分散処理システムとして，Hadoop MapRe-duce と Spark を採用した．それぞれの詳細は付録で述べる．また，Hadoop MapReduce 上のシミュレータは先行研究 [10] のものを用い，Spark 上のシミュレータは同様の処理フローとなるよう新たに実装した．
図 1 key-value ペアによる分散システム表現

前述の分散システムの表現形式では，Reduce フェーズにおけるデータの処理順が重要である．データがタイムスタンプ順

3. 汎用分散処理システムでのシミュレーション
手法に並んでいなければ正しいシミュレーション結果は得られない．
また，Reduce フェーズにおいてノード情報の処理を完了する
前にイベント処理を開始してしまった場合にもシミュレーショ

本章では今回用いたシミュレータにおける分散システムの表
現方法について述べる．今回用いたシミュレータでは，ネットワーク上のノードが離散事象シミュレーションにおける論理プ
ロセスと対応する．

3. 1 分散システムの表現
現実の分散システムでは，各ノードが自身の持つ情報を基に自律的に動き，インターネットを介したノード間通信により相互作用することでシステム全体が機能する．汎用分散処理シス
テム上で分散システムのシミュレーションを行うためには，これをシステムが処理可能な計算モデルに落とし込む必要がある．

ン結果に矛盾が生じるため，ノード情報がイベント情報よりも先に処理されるように順序付けしなければならない．

Hadoop MapReduce ではセカンダリソートがサポートされており，key にノード ID だけでなく，タイムスタンプやデータの種類といった情報を付加することで，Shuﬄe フェーズの時点
で正しい順序付けを可能としている．Spark では別途，Reduce
フェーズにおいて受信したデータすべてに対して順序付けを行
うためのソートを実装している．両者の基本的な処理の流れは
前述の分散システムの表現形式に従うが，このような些細な差異は存在する．

今回用いたシミュレータでは計算モデル MapReduce 上で分散システムを表現し，その挙動をシミュレートする．

3. 1. 1 データ形式
MapReduce が扱うデータの形式は key-value ペアであるた

4. 投機的実行への制限
MapReduce モデルでのシミュレーションでは，データの送受信が Shuﬄe フェーズに限定されるため，メッセージのやり取め，分散システムの表現もこの形式をとっている．key にはノーりが生じるたびに同期が必要となる．汎用分散処理システム上

ドの ID，value には表現対象のそれぞれに必要な情報を割り当でのシミュレーションにおいて，同期によるオーバーヘッドはてる．分散システムを表現するために必要な情報は，ノードが保持する情報，ネットワーク情報，シミュレーションのイベン
ト情報である．ノードが保持する情報には隣接ノードの ID や，ノードの持つアプリケーションデータ等が含まれる．シミュ
レーションのイベント情報には，特定の操作の実行スケジュー
ルや，ノード間通信の抽象化であるメッセージを含む．シミュ
レーションはイベントの発生順に従って進むため，イベントにはタイムスタンプを付与する．

実行時間に大きく影響する．そのため，同期回数を抑制できる
楽観的時刻管理アルゴリズムを採用するメリットが大きい．しかし，汎用分散処理システムが動作するマシンはコモディティ
であることが多く，メモリを始めとする計算リソースには制限がある．このような場合に，メモリを大量に消費する楽観的時
刻管理アルゴリズムを用いると，問題が生じることもある．特に Spark はオンメモリ処理であるため，楽観的時刻管理アル
ゴリズムを採用する場合に，メモリ不足が原因で，シミュレー

3. 1. 2 処理フロー

ションが失敗する，あるいは実行時間が非常に長くなるといっ

シミュレーションは MapReduce に従い，Map フェーズ，た事態が起こりやすいと予想される．

Shuﬄe フェーズ，Reduce フェーズを繰り返すことで進行する．

楽観的時刻管理アルゴリズムを用いたシミュレーションにおMap フェーズでは，その iteration で扱うデータをすべて観測

いて，リソースを消費する主な原因となっているのは，イベン

し，最小のタイムスタンプを割り出し，各 Reducer に配布するトキュー内のすべてのイベントを処理することによって発生す

仮想時刻データを作成する．Shuﬄe フェーズでは，Map フェー

る大量のメッセージとアンチメッセージ，ロールバック先の状

ズで作成した仮想時刻データとともに，ノード情報やイベント

態を保存するための状態のスナップショットである．メッセー

情報データ群をノード ID ごとにまとめられるように Reducer

ジが抑制されれば，新たに発生するイベントが少なくなり，管へ送信する．Reduce フェーズでは，ノード ID ごとに仮想時理しておく必要のあるスナップショットも必然的に少なくなる．

刻データとノード情報の処理を行った後，そのノードに関す

よって，リソースの消費を抑えるには，まずメッセージの大量発生を防ぐ必要がある．また，汎用分散処理システムにおいて

表 1 実験環境と共通パラメータ問題となるのは全体のメッセージ数よりも，iteration ごとのOS

Ubuntu 14.04.2 LTS

メッセージ数である．

CPU

2.40 GHz Intel Xeon E5620 × 2

Java 仮想マシン

Java SE 8 Update 45

Scala

2.11.6

今回，iteration ごとのメッセージ数を抑制するため，Time

Warp の処理範囲を限定したアルゴリズムをシミュレータに適用し，シミュレーション性能の変化を評価する．

yarn.nodemanager

30000.resource.memory-mb

4. 1 Moving Time Window
Moving Time Window (MTW) [22] は bounded lag アルゴ
リズムと Time Warp を組み合わせたアルゴリズムである．

MTW は論理プロセス全体の最小時刻 T 一定の時間窓 W の間のイベント，つまり [ T , T + W ] のタイムスタンプを持つイベントのみを投機的に処理する．この間のイベント処理では，Time Warp と同様，イベントの処理順序に矛盾が生じること

表 2 Hadoop MapReduce のパラメータ

mapreduce.map.memory.mb

2048

mapreduce.map.java.opts-Xmx1500M

mapreduce.reduce.memory.mb

3072

mapreduce.reduce.java.opts-Xmx2500

表 3 Spark のパラメータを許容する．これにより，イベントキューの一部のみを楽観的

spark.executor.instances

19

時刻管理アルゴリズムで処理することとなり，iteration 回数を

spark.executor.memory

12G

抑えながらメッセージ発生量も抑えることができる．MTW で

spark.executor.cores

4は，投機的実行を制御する時間窓 W のサイズは一定であり，spark.driver,memory

5G

シミュレーション実行前に静的に決定する．しかし，時間窓のspark.serializer

サイズを動的に変更すること自体は容易である．

4. 2 Adaptive Time Warporg.apache.spark.
serializer.JavaSerializer

spark.shuﬄe.manager

sort

spark.task.cpus

1

投機的実行の有効範囲に制限を与えるアルゴリズムとして，

Adaptive Time Warp (ATW) [23] がある．ATW では，シミュレーション実行中のパラメータを基に，投機的実行の基準を調整する．参照するシミュレーション実行中のパラメータとしては，ある期間内における送信メッセージやアンチメッセージの数，ロールバックの有無や多寡，ロールバック元とロールバッ
ク先の時間的距離といった様々な要素が挙げられる．投機的実行の有効範囲の基準には，時間窓のサイズや処理可能な未処理メッセージ数の上限が挙げられる．参照するパラメータや投機的実行の有効範囲の基準によって，シミュレーション実行に及ぼす効果に差異がある．メッセージやアンチメッセージ，ロー
ルバック発生の有無に注目し，時間窓や処理可能なメッセージ数に制限をかけることでメモリ消費を抑えたシミュレーション実行が期待できる．

図 2 2 次元正方メッシュネットワークシミュレーションの実行時間

MTW，ATW は Time Warp を用いたシミュレーションの高速化を目的としたものだが，本論文では，汎用分散処理シス

レーションは最もタイムスタンプの小さいイベントを追いかけ

テムで問題となりうるメモリ不足への対応策として適用した．

るように進行する保守的な時間管理に基づく．

5. 実験・評価

2 次元メッシュネットワークにおいて，1 辺のノード数が 1,000，
1,500，2,000，2,500，3,000 である 2 次元正方メッシュネット

今回，シミュレータ上で flooding によるファイル検索を行う

ワーク上で，20%のノードが同時にデータ検索をシミュレート

Gnutella プロトコル [24] のシミュレーションを行った．実験でする．実験結果を図 2 に示す．Hadoop MapReduce，Spark シは，各ノードに固有のアプリケーションデータを与えている．

ミュレータともにノード数の増加に伴って同じように実行時間

実験環境は表 1 の通りで，マスター 1 台，スレーブ 10 台の計が伸びている．両者の実行時間の差は，Hadoop MapReduce

11 台で実験を行った．本実験で用いた Hadoop MapReduce 及の欠点であるジョブごとの分散ファイルシステムへの読み書き

び Spark のパラメータを表 2，表 3 に示す．

や，ジョブの準備処理によるものが大きい．

5. 1 2 次元正方メッシュネットワークでの評価

5. 2 複雑ネットワークでの評価

Hadoop MapReduce と Spark 上に実装されたシミュレータ

次に，iteration ごとのイベント数の増加数が非常に大きくなの性能を比較するため，まず iteration ごとのイベント数の増加数が少ないシミュレーションで比較を行う．ネットワーク遅延は一定で，すべてのデータ検索を同時に行う．また，シミュ

りうるシミュレーションで両者の比較を行った．

m=1，2 の Barabasi-Albert (BA) モデル [25] において 100

図 3 複雑ネットワークシミュレーション (m=1) の実行時間

図 4 複雑ネットワークシミュレーション (m=2) の実行時間

件のデータ検索をシミュレートした．ネットワーク遅延は一定
で，すべてのデータ検索を同時に行った．また，シミュレーショ
ンは最もタイムスタンプの小さいイベントを追いかけるように進行する保守的な時間管理に基づく．
ネットワーク上のノード数は 106 から 107 まで，106 ノード
間隔で実験を行った．実験結果を図 3，図 4 に示す．2 次元メッシュネットワークでの実験と同様，Hadoop MapReduce のオー
バーヘッドが顕著である他，Spark シミュレータにおいて，特に m=1 において処理時間の伸び方が不安定になっている．タ
スクごとの実行時間を調査したところ，ネットワークのハブに該当するノードを割り当てられたごく一部のタスクの実行時間
が他のタスクの実行時間に比べて非常に長くなっていた．複雑

図 5 大規模複雑ネットワークシミュレーションの実行時間

ネットワークでは，リンクが集中しているノードが存在し，そのノードが隣接ノードすべてにメッセージを送るような場合，

制限方法は MTW と ATW を用いる．ネットワークトポロジ

イベント数が爆発的に増加した．この性質が，タスクごとの実は 1 辺のノード数が 1,000 ノードの 2 次元正方メッシュネット

行時間に大きな影響を与えている．m=2 のケースにおいては，ワークを用いる．合計 100 万ノードの上で，100 秒間の間にラハブに該当するノードが多く存在するため，他のタスクよりもンダムなタイミングで 1 万件のデータ検索を行う．

実行時間が非常に長くなるというケースが起きにくかったもの

5. 3. 1 MTW による制限と考えられる．

Lazy Cancellation を適用した Time Warp にさらに MTW

汎用分散処理システムの想定するアプリケーションでは，基を適用した．時間窓サイズは，10 秒から 50 秒の間で 10 秒刻

本的には処理を行ってもデータサイズが増加しない，あるい

みで変化させた．実行時間の変化やシミュレーション中に同時は処理を繰り返すたびにデータサイズが小さくなる．一方で，にシミュレータ上に存在したメッセージ及びアンチメッセージ

flooding を行うアプリケーションのシミュレーションような，の数（メッセージ数）を測定した．その結果を図 6 に示す．

処理中に全体データサイズが大きくなっていく場合においては，

今回，データ検索を行うタイミングは一様乱数により決定

想定外の負荷がかかる可能性がある．特に，爆発的にメッセーされており，一定の期間に大きくイベントが集中するような偏ジ数が増加し得る複雑ネットワークとリソース制約の大きいりは生じていない．そのため，時間窓サイズに応じて，線形にSpark の組み合わせでは影響を受けやすいものと考えられる．

メッセージ数が増加していく．実行時間も時間窓サイズに応じSpark 上のシミュレータでは，シミュレーション内容によってて減少していく．しかし，時間窓サイズを大きくするとイベン

は完了までの時間の予測が困難になるという問題が浮かぶ．

ト処理が多く発生する．さらに時間窓の外に存在するイベント

また，m=2 の BA モデルにおいて，Hadoop MapReduce とは未処理のまま残されるため，最終的な iteration 回数の抑制

Spark，双方のシミュレータで 108 ノードまでの動作を確認しは頭打ちとなる．そのため，実行時間の減少は線形にならず，た．その際の実行時間を図 5 に示す．

時間窓のサイズが一定の値に到達するとほぼ減少しなくなった

5. 3 時間窓による投機的実行制限
本項では，シミュレータに楽観的アルゴリズムを適用した状
態で投機的実行に制限を設け，その影響を測る．投機的実行のか，やや増加した．

5. 3. 2 ATW による制限
ナイーブな Time Warp（図中では NTW）及び Lazy Can-図 6 NTW を適用した場合のシミュレータ性能

図 9 ATW を適用した場合の最大メッセージ数

ナイーブな Time Warp においては，ATW を用いた場合に若干の実行時間の減少が見られた．Lazy Cancellation においては実行時間が増加する結果となった．今回はイベント処理順の矛盾がその後のメッセージ送信に大きく影響しないアプリ
ケーションを用いたため，このような結果となったが，Lazy

Cancellation が上手く働かないようなアプリケーションを用いた際には，ATW を利用することで，実行時間が減少する可能性がある．
また，ナイーブな Time Warp 及び Lazy Cancellation ともに，ATW を用いた場合の方が iteration 回数に対する実行時間が短くなっている．これは，投機的実行可能な範囲に制限を設
図 7 ATW を適用した場合のシミュレーション実行時間

けることによって，1iteration ごとのイベント処理が減少した
ことによる．
シミュレート中の最大メッセージ数に関しては，どちらのケースにおいても 20 ∼ 30% の抑制が確認された．300 万メッ
セージという値は，ナイーブな Time Warp における最大メッセージ数の約 50% の値である．この値の周辺に収まらなかった理由として，Gnutella のような flooding を行うアプリケー
ションでは，メッセージの増加が爆発的であり，閾値を一度に大きく上回ることが挙げられる．

5. 3. 3 MTW vs. ATW
MTW，ATW の双方において，目的であったメッセージ数の抑制は確認できた．ここでは MTW と ATW を用いたシミュ
レーションについて議論を行う．
図 8 ATW を適用した場合の iteration 回数

ATW の実験は，閾値から MTW における時間窓サイズ 50
秒でのシミュレーションに相当する．ATW ではメッセージのcellation（図中では LC）を適用した Time Warp に ATW を

抑制は 20% 程度であり，実行時間もメッセージ抑制の効果も適用した．ATW における投機的実行の有効範囲には，処理可MTW に劣る．しかし，実際の運用を考えると，MTW で時間

能メッセージ数や時間窓等，いくつかのパラメータが考えられ

窓サイズを適切な値に設定するためには，1 度シミュレーショ

るが，今回は MTW との比較を考慮し時間窓を用いた．時間

ンを実行し，メモリ使用量のピーク値を把握しなければならな窓に影響を与えるパラメータは前 iteration 中のメッセージ及い．同じアプリケーションを用いたシミュレーションでも，メ

びアンチメッセージの総数とした．この数が 300 万件を超える

モリ使用の状況はシナリオごとに大きく変化し得る．そのため，と，時間窓サイズが半分になり，そうでない場合は時間窓サイ実行したことのないシナリオのシミュレーションでは，時間窓

ズが 10 秒ずつ拡大する．時間窓サイズの初期値は 100 秒としサイズの設定は困難である．一方で，一度大きな計算機クラス

た．その上で，実行時間の変化やメッセージ数を測定した．そタで実行したことのあるシナリオまたはその類似シナリオを，の結果を図 7，図 8，図 9 に示す．

規模を縮小した計算機クラスタで実行する等の場合には有効である．また，今回のシミュレーションではイベント処理のタイ
ミングに偏りがなかったため問題とはならなかったが，時間窓に収まる一定期間内にイベントが集中すると，MTW ではメッセージ数を抑制することはできない．

ATW の場合，アプリケーションからシミュレーションで扱うことのできるメッセージ数の上限を計算し，それを閾値のヒントとすることが可能である．時間窓のサイズは適当であっても，実行中に調整されていくため，実行したことのないシナリ
オのシミュレーションでも対応できる．さらに，ATW での投機的実行可能な範囲の制限は時間窓に限ったものではないため，処理可能なイベント数や送信可能なメッセージ数を加味することで，メッセージ抑制能力は向上する．また，イベントの発生
時刻の偏りにも柔軟に対応できる．

6. ま と め
本論文では，汎用分散処理システムによる大規模分散システ
ムシミュレーションを目的に，Hadoop MapReduce と Spark でのシミュレーションとその比較を行った．その結果から，Spark
上のシミュレータが Hadoop のものより高速であるが，実行時間の予測が困難になりうることを指摘した．
また，Time Warp を利用したシミュレーションを汎用分散処理システムで行う上で課題となりうるメモリ不足の問題に対して，時間窓による投機的実行の制限により，メッセージ数を削減することで対応することを提案した．MTW，ATW の 2
つの手法で，シミュレーション中のメッセージ量が抑制できることを確認した．今回の実験の対象としたシステムの規模は大きくはないが，同手法は規模を大きくした場合にも適用できる．

ATW により投機的実行を制限する上で適切なパラメータとアルゴリズムの模索は今後の課題である．様々なパラメータからその都度適切なものを選択し，それを用いてシミュレーショ
ンを操作することができれば，メモリ消費を抑えつつ，シミュ
レーションの高速化も狙える．また，シミュレータ上に様々なプロトコルを実装した上で，既存シミュレータとの比較を行う
こともシミュレータの立ち位置を示すために必要である．負荷分散や，多くの計算機を用いたさらに大規模なシミュレーショ
ンも今後の課題である．シミュレート対象となるネットワーク
トポロジやアプリケーションアルゴリズム，シナリオの内容を考慮して負荷分散を行い，計算リソースを効率よく運用することで，更なる高速化が見込めるものと考える．また，より多くの計算機を用いることで，シミュレーションの高速化と対象シ
ステムの大規模化が可能であると考える．

7. 謝

辞

本研究は JSPS 科研費 25700008，26540161 の助成を受けたものである．

文

献[1] Decentralized Systems and Network Services ResearchGroup in Karlsruhe Institute of Technology, “Live Monitoring of the BitTorrent DHT”.

http://dsn.tm.unikarlsruhe.de/english/2936.php .
[2] J. Bradley, J. Barbier, D. Handler, “Embracing the Internet of Everything To Capture Your Share of $14.4 Trillion” ,2013-2. http://www.cisco.com/web/about/ac79/docs/innov/IoE Economy.pdf .
[3] “ns-2”. http://www.isi.edu/nsnam/ns/ .
[4] “ns-3”. https://www.nsnam.org/ .
[5] K. Shudo, “Overlay Weaver: An Overlay Construction
Toolkit”, http://overlayweaver.sourceforge.net/.
[6] K. Shudo, Y. Tanaka, S. Sekiguchi: “Overlay Weaver:
An Overlay Construction Toolkit”, Computer Communications, Vol. 31, No. 2, pp.402–412, 2008.
[7] “PeerSim: A Peer-to-Peer Simulator”,
http://peersim.sourceforge.net/[8] A. A. Jamal, W. S. W. Awang, M. F. A. Kadir, A. A.
Aziz, W. J. Teahan, “Implementation of Resource Discovery Mechanisms on PeerSim: Enabling up to One Million
Nodes P2P Simulation”.
[9] T. T. A. Dinh, M. Lees, G. Theodoropoulos, R. Minson, “Large Scale Distributed Simulation of p2p Networks”,
Proc. Parallel, Distributed and Network-Based, 2008. PDP
2008. 16th Euromicro Conference on. IEEE, 2008.
[10] 杉野 好宏, 華井雅俊, 首藤一幸, “MapReduce による大規模分散システムのシミュレーション” , 電子情報通信学会技術研究報告.
NS, ネットワークシステム 113(244), pp.99–104, 2013-10-10 .
[11] “Apache Hadoop”, http://hadoop.apache.org/.
[12] “Apache Mesos”, http://mesos.apache.org/ .
[13] J. Dean, S. Ghemawat, “MapReduce: Simplified Data Processing on Large Clusters”, Proc. OSDI 04, 2004.
[14] “MapReduce NextGen aka YARN aka MRv2”.
https://hadoop.apache.org/docs/current/hadoop-yarn/
hadoop-yarn-site/index.html .
[15] “Apache Hadoop Project: Hadoop Wiki - PoweredBy”,
2012. http://wiki.apache.org/hadoop/PoweredBy.
[16] “Apache Spark - Lightning-Fast Cluster Computing”,
http://spark.apache.org/ .
[17] M. Zaharia, M. Chowdhury, T. Das, A. Dave, J. Ma, M.
McCauley, M. J. Franklin, S. Shenker, I. Stoica, “Resilient
Distributed Datasets: A Fault-Tolerant Abstraction for InMemory Cluster Computing”, Proc. 9th USENIX conference on Networked Systems Design and Implementation,
April 25–27, 2012, San Jose, CA .
[18] M. Dobashi, K. Saruta, T. Shimogaki, M. Tsuzuki,“NTT DATA: Operating Spark clusters at thousandscore scale and use cases for Telco and IoT”, 2015.
https://databricks.com/blog/2015/05/14/ntt-data-operatingspark-clusters-at-thousands-core-scale-and-use-cases-for-telcoand-iot.html .
[19] B. D. Lubachevsky, “Bounded lag distributed discrete event
simulation”. SCS Multiconference on Distributed Simulation, 1988.
[20] D. R. Jeﬀerson, “Virtual time”. ACM Transactions on Programming Languages and Systems (TOPLAS), Vol 7 Issue
3, pp. 404–425, July 1985.
[21] A. Gagni, “Space management and cancellation mechanisms for Time Warp”. Tech. Rep. TR-85-341, Universityof Southern California, Los Angeles (Ca, USA), 1985.
[22] L. M. Socol, D. P. Briscoo, A. P. Wieland , “MTW: A strategy for scheduling discrete simulation events for concurrent
execution”, SCS Multiconference on Distributed Simulation. 1988.
[23] K. S. Panesar, and R. M. Fujimoto, “Adaptive flow control
in time warp.” Proc. 11th IEEE’s Workshop on Parallel andDistributed Simulation, 1997.
[24] “The Annotated Gnutella Protocol Specification v0.4”,
http://rfc-gnutella.sourceforge.net/developer/stable/index.html.
[25] A. L. Barabasi, R. Albert, “Emergence of scaling in randomnetworks”. Science, Vol. 286, No. 509, 1999.

A 付録・汎用分散処理システム

B 1 保守的アルゴリズム：bounded lag
保守的アルゴリズムでは，シミュレーション中のイベントの発生順を正確に追いかける形で処理を進める．そのため，イベ

本論文では，Apache Hadoop [11] や Apache Mesos [12] とントの処理順序に矛盾が生じないようにシミュレーション全体

いった信頼性とスケーラビリティを備えた汎用的な分散並列処を監視し管理する必要がある．一般に，シミュレーション全体

理基盤の上で動作するシステムのうち，用途の広いシステムを監視するには，全論理プロセスの同期を頻繁に行う必要がを汎用分散処理システムと呼ぶ．今回評価対象とした Hadoop

ある．しかし，分散処理において，全体を同期するコストは大MapReduce や Spark が汎用分散処理システムに該当する．

きく，頻繁な同期は多大なオーバーヘッドを招く．保守的アルA 1 Hadoop MapReduce

ゴリズムの例として，bounded lag アルゴリズム [19] が挙げら

Hadoop MapReduce は，大量データ処理向けプログラミングれる．

モデル MapReduce [13] のオープンソース・ソフトウェア実装でbounded lag アルゴリズムは，論理プロセス全体の最小時刻ある．現在の Hadoop MapReduce 2.0 は Hadoop YARN [14]T から一定の時間 Dmin の間，つまり [T ,T + Dmin ] のタイム

ベースの大規模データ向け分散処理システムである．ジョブのスタンプを持つイベントの処理のみを許可する時刻管理アルゴ

実行中に一部の計算機が故障した場合にも，計算途中で失われリズムである．Dmin はネットワーク遅延の最小値であるため，たタスクを再実行することで高い耐障害性を達成している．大

この間におけるイベント処理は処理順序の矛盾を生じさせない．

規模な計算クラスタでの実用例として，4,500 台での運用実績

これにより，最小のタイムスタンプのみを追いかけるよりは一がある [15] ．

度に多くのイベントを処理できる．しかし，ネットワーク遅延

Hadoop MapReduce を繰り返しの計算処理に用いる場合，ジョブに必要なファイル群の配布や Java 仮想マシンの起動，分散ファイルシステムへの中間データの読み書きといった，実の最小値に対し，ネットワーク遅延の最大値が非常に大きい場
合には，非効率である．

B 2 楽観的アルゴリズム：Time Warp際の計算とは関係のない部分に時間がかかってしまう．このた

楽観的アルゴリズムでは，シミュレーション中のイベント処

め，反復処理が必要なアプリケーションには適していない．

理順序の矛盾を許容し，投機的にイベントを処理することで同A 2 Spark

期回数を抑制する．本論文では楽観的時刻管理アルゴリズムでSpark [16] は Hadoop 同様，汎用的なオープンソースデー

ある Time Warp [20] を用いる．ここでは Time Warp の概要

タ処理フレームワークである．Spark は DAG という計算モを述べる．

デルを採用しており，中間データの書き込みのようなディス

各論理プロセスは，自身の持つイベントキュー内のイベント

ク I/O を極力起こさないように動作する．そのため，Hadoop

をタイムスタンプの昇順で処理する．時刻の管理は論理プロセ

MapReduce のようにデータを処理してその都度ディスクに書き

スごとに独立して行い，イベントキュー内のすべてのイベント出す方式よりも高速である．Spark では Resilient Distributed

を投機的に処理する．処理を行ったイベントには処理済みを意

Dataset (RDD) [17] という不変なコレクションを導入してい味するラベルを付ける．論理プロセス間での通信により，イベる．RDD を用いることで，ジョブの実行中に計算機の故障等

ントキューに追加されたイベントのタイムスタンプが処理済みの原因でデータの一部が失われたとしても再構築が可能である．

のイベントのタイムスタンプよりも小さかった場合，イベント

これにより，耐故障性を達成している．4,000 コアの大規模なの処理順序に矛盾が生じたこととなる．このようにしてイベン

計算機クラスタで試行が行われた例もある [18] ．

トの処理順序に矛盾が生じた場合，新たにイベントキューに追Spark では，ジョブ実行中の Java 仮想マシンの常駐化，一般加された未処理のイベントのタイムスタンプの時刻まで論理プ化された計算モデル，データキャッシング等の恩恵により，反ロセスの状態を戻す．その後，その時刻を論理プロセスの保持

復処理を効率よく行うことが可能である．一方で，オンメモリするシミュレーション時刻とする．シミュレーション時刻の更処理を行う以上，計算機クラスタの総メモリ容量を大きく上回

新後，イベントキュー内のシミュレーション時刻以降のタイム

るサイズのデータを扱うことはできない．

スタンプを持つ処理済みイベントの実行を取り消す．この操作

B 付録・時刻管理アルゴリズム

をロールバックと呼ぶ．ロールバックの例を図 10 に示す．
他の論理プロセスへのメッセージ送信の取り消しには，アン

今回扱うネットワークシミュレーションのような離散事象シ

チメッセージというメッセージ取り消し通知を用いる．他の論

ミュレーションは，シミュレート対象の構成単位である論理プ

理プロセスへのメッセージ送信時には，送信するメッセージとロセスと論理プロセスの状態を変化させるイベントから構成さ対応するアンチメッセージを生成し，メッセージ送信の取り消れる．各論理プロセスはイベントを収容するイベントキューをしが必要になるまでそれを保持する．ロールバック発生時，矛持つ．シミュレーション実行時にはイベントキューの先頭のイ

盾が発生した論理プロセスは，ロールバック先の時刻以降のタ

ベントから順に処理を行う．イベント処理の時刻管理アルゴリ

イムスタンプを持つアンチメッセージを，対応するメッセージ

ズムには，大きく分けて保守的アルゴリズムと楽観的アルゴリの送信先論理プロセスへと送信する．アンチメッセージを受信

ズムの 2 つがある．

した論理プロセスは，自身のイベントキューから対応するメッ
セージを削除し，アンチメッセージのタイムスタンプの時刻まメッセージ m を送信する
再実行の前後でイベントの挙動が変わらず，送信メッセージの内容が一致する場合には，イベント処理の取り消しは他の論理プロセスへ伝搬されなくなる．再実行の前後でイベントの挙動が変化し，再実行前のメッセージが送信されなくなった場合には，再実行前のメッセージは取り消される．加えて，再実行後に内容の異なるメッセージ送信を行う場合には，そのメッセー
ジが送信される．再実行後の送信メッセージが存在しない場合にはまた，再実行前には送信されなかったメッセージは通常通りに送信を行う．
これにより，不要なアンチメッセージの送信とそれによる不図 10 key-value ペアによる分散システム表現

要なロールバックの発生を抑制できる．よって，通常の Time

Warp では不要なロールバックが生じてしまう場合において，より少ない同期回数で正しいシミュレーション結果への収束がでロールバックを行う．これにより，連鎖的にロールバックが起こり，投機的な処理によって発生したイベント処理順序の矛
盾は解消される．最終的に，正確なシミュレーション結果へと収束する．

Time Warp は，保守的アルゴリズムを用いた場合よりもシ
ミュレーション中の同期回数を抑えることができる一方で，アンチメッセージが大量に発生することによりメモリや CPU といった計算リソースを多く必要とする．

B 3 楽観的アルゴリズムの最適化：Lazy Cancellation
Time Warp を用いることで，シミュレーション中の同期回数を抑えることができる．しかし，通常の Time Warp では，
アンチメッセージによるロールバックが連鎖的に発生し，それに伴うイベントの再実行がオーバーヘッドとなり得る．よって，
アンチメッセージの送信を可能な限り回避し，ロールバックの発生回数を抑えることにより，シミュレーション実行時間を抑えることができると考えられる．Lazy Cancellation [21] は，これを実現したアルゴリズムである．
通常の Time Warp を用いる場合，イベントの処理順序に矛盾が生じると，その時点で即座にアンチメッセージを送信し，イベント処理の取り消しと再実行を行う．この際，ロールバッ
クにより再実行されることとなったイベントの挙動は，必ずしも再実行前のものから変化するとは限らない．ロールバックによる再実行前と再実行後でイベントの挙動が一致する場合，そのイベントに対応するアンチメッセージを送信する必要はなくなる．
アンチメッセージ送信の必要性は，再実行の前後の他論理プ
ロセスへの送信メッセージを比較することで判断できる．再実
行前の送信メッセージの集合を M ，再実行後の送信メッセージの集合を M ′ とする．この時，再実行後のメッセージ m に対するアンチメッセージ送信の是非は次のように判断される．
（ 1 ） m ∈ M かつ m ∈ M ′ ならば
アンチメッセージもメッセージ m も送信しない（ 2 ） m ∈ M かつ m ∈/ M ′ ならばアンチメッセージを送信する

m が存在する場合にはメッセージ m を送信する（3） m ∈
/ M かつ m ∈ M ′ ならば可能となる．