Envision VM (パート4), 分散実行

Victor Nicollet

この記事は、Envision 仮想マシンの内部動作、すなわち Envision スクリプトを実行するソフトウェアに関する全4部作の第4部目です。 パート1、パート2およびパート3を参照してください。このシリーズでは Envision コンパイラについては扱っていません（多分別の機会に）、ですのでスクリプトが何らかの方法で Envision 仮想マシンが入力として受け取るバイトコードに変換されたと仮定しましょう.

前回の記事では、主に各ワーカーが Envision スクリプトをどのように実行したかを検証しました。しかし、レジリエンスおよびパフォーマンスの両面から、Envision は実際にクラスター内の複数のマシンで実行されます。

各ワーカー内の各レイヤーは、他のワーカーの同じレイヤーまたは同一ワーカー内の他のレイヤーと通信します。これにより、ネットワーク通信が各レイヤーのプライベートな実装上の詳細として保持されることが保証されます。

低レベルでは、各ワーカーはクラスタ内の他の各マシンに対して2つの TLS 接続を開き、各レイヤーからの通信はこれら2つの接続を介して多重化されます（1つの接続は短いメッセージ用、もう1つは大きなデータ転送用として使用されます）。

コントロールレイヤー

このレイヤーは、スケジューラがワーカーに対してミッションの割り当ておよび解除を行うために使用され、ワーカー間の通信は含まれていません。このレイヤーの主なメッセージは次の通りです：

• スケジューラはワーカーに対しミッションの実行を開始するよう依頼します。
• スケジューラはワーカーに対しミッションの実行を停止するよう依頼します。
• ワーカーはミッションの実行中に壊滅的なエラーに遭遇したことをスケジューラに通知します（通常は非決定論的な問題、例えば「NVMe ドライブが発火した」といった例で、これは将来または別のワーカーで同じミッションを再試行できることを意味します）。
• ワーカーは自身の現在の状態に関する統計情報をスケジューラに提供します：ミッションのリスト、各ミッションの DAG のフロンティアのサイズ、各ミッションの DAG 内でまだ実行されていない thunk の総数。

スケジューラは、これらの統計情報を使用してミッションの再割り当てのタイミングを決定します。実際のルールは、優先順位ルール、複数テナント間および同一テナント内のスクリプト間の公平性、そしてその時点でのクラスター全体の負荷に依存するため非常に複雑ですが、一般的な傾向として、フロンティアが十分に大きいミッションは、すでに過負荷でなければ複数のワーカーに分散できるということです。同じ量の作業をする場合、すべてのミッションを全ワーカーに分散するよりも、それぞれのワーカーで4つのミッションを実行する方が効率的です。

実行レイヤー

各ワーカーは現在実行中の thunk を追跡し、新しい thunk をスケジュールするたびにこのリストを他のワーカーに送信します¹。これにより、ネットワーク遅延に関連する非常に短い時間ウィンドウを除いて、二つのワーカーが同じ thunk の実行を開始することはありません。

もちろん、もしワーカーがこれらの更新送信を停止した場合（例えば、クラッシュしたりクラスターの他の部分と切断された場合など）、そのピアは数秒以上前のリストを古いものとみなし、その thunk を実行することを許可します。

メタデータレイヤー

各ワーカーは完全なメタデータのコピーを保持しようとしますが、実際には_同期_は行いません。全てのワーカーが全く同じメタデータに同意する保証は提供せず、代わりに最終的整合性保証を前提としています。これにより、メタデータレイヤーの分散は設計上最も困難なものとなります²。

このレイヤーの最終的整合性は、次の3つの主要なルールに従います：

1. メタデータレイヤーへのすべてのローカルな変更は、直ちに他のすべてのワーカーに送信されます。このブロードキャストは失敗する可能性があり、再試行はされません。
2. 他のワーカーから受信したリモートの変更は、単調な進行³に基づいてローカルのメタデータレイヤーにマージされます： thunk の「結果なし」値は「チェックポイント」値（つまり、thunk の実行が開始されたが完了していないことを意味する）に上書きされ、それが「エイリアス」値（つまり、thunk が代わりに実行されるべき DAG を返したことを意味する）に上書きされ、さらに「結果」値（成功した結果とその関連するアトム、または致命的なエラー）に上書きされる可能性があります。
3. 他のレイヤーがメタデータレイヤー内の値に基づいてネットワークレスポンスを送信するたびに、メタデータレイヤーはその値を再びブロードキャストします。

3番目のルールは、実際に重要な場合に同期のレベルを強制するために設計されています。例えば、次の一連のイベントを考えてみてください：

• スケジューラはワーカーに対しミッションの実行を依頼します（コントロールレイヤーを通じて）
• ワーカーはミッションを実行し、結果をブロードキャストします（メタデータレイヤーを通じて）、しかし メッセージはスケジューラへの途中で失われます。
• スケジューラは、ワーカーがもはやミッションを実行していないことに気付き（コントロールレイヤーを通じて）、再度実行するよう依頼します。
• ワーカーはミッションの thunk にすでにメタデータレイヤー上で結果が存在することを確認し、何も行いません。なぜなら行うべきことが何もないからです。

これは、メタデータレイヤーの thunk の状態についてスケジューラとワーカーが意見の相違を起こす（ワーカーは完了したと判断し、スケジューラはそうではないと判断する）デッドロックです。3番目のルールは、ワーカーが「もはやこのミッションに取り組んでいない」という応答を、thunk に結果があるという観察に基づいて行ったため、メタデータレイヤーが再びこの情報をブロードキャストすべきであると決定することで、これを解決します。こうしてデッドロックは解消されます：

• ワーカーのメタデータレイヤーが thunk の結果を再びブロードキャストし、スケジューラがそれを受信します。
• スケジューラはミッションの thunk に結果が現れたことに反応し、そのミッションを完了とフラグ付けし、そのミッションを依頼したクライアントに通知します。

アトムレイヤー

ワーカーはそれぞれのアトムレイヤーを組み合わせて分散型の Blob ストアを作り出し、各アトムはその識別子（SpookyHash を使用して作成された内容の 128 ビットハッシュ）によりリクエストすることができます。これは_分散ハッシュテーブル_（DHT）ではありません、なぜなら DHT では間違ったトレードオフが発生するからです：DHTでは、アトムの検索は迅速に行える（ハッシュからアトムを保持するワーカーの識別子を単純な関数で計算できる）が、アトムの書き込みは遅くなる（計算したマシンから、DHT の現在のレイアウトに基づいて保持されるべきマシンに送信する必要がある）ためです。ほとんどのアトムはそれを生成したのと同じマシンで消費されると予想されることを考えると、これは無駄です。

代わりに、ワーカーが自分自身のアトムレイヤーからアトムをリクエストする際、まず自分自身の NVMe ドライブ上でそのアトムを探します。見つからない場合は、他のワーカーにそのアトムの存在の確認を問い合わせます。これは、これらのクエリができるだけ迅速に完了されなければならないため、Envision の分散設計における最大のパフォーマンス上の課題となり、レスポンスが得られないワーカーに対処するために複雑なタイムアウト戦略が必要になります：待ちすぎると、決して返答が得られなかった応答を待つために数秒を浪費してしまい、早すぎると、他のワーカーからダウンロードできた可能性のあるアトムを再計算する必要が出てきます。

これを改善するために、アトムレイヤーは複数のリクエストをまとめてバッチ処理し、他のすべてのワーカーが必要なリクエストのパイプラインを常に保持し、ワーカーの応答時間が急上昇した際により容易に検出できるようにしています。

少なくとも別のワーカーがディスク上でアトムの存在を確認すると、アトムをダウンロードするための第二のリクエストが送信されます。このようなダウンロードリクエストは非常に急激な変動を見せる傾向があり、多くの thunk がまずアトムをリクエストし、その後内容の処理を開始するからです。このため、アトムレイヤーは各ワーカー間に単一のダウンロードキューが存在することを認識しており、もし特定のアトムリクエストが数秒間最初のバイトすら受信できなくても（他のアトムがバイトを受信している状態でキューが満杯の場合、心配する必要はありません）、パニックになりません。ある意味で、タイムアウトはアトムリクエスト単体でのものではなく、レイヤー全体のレベルで発生します。

加えて、転送キューには2つの最適化が適用されています：

1. 各リクエストはどの thunk がデータを必要としているかを指定するため、送信者は同じ thunk からのリクエストをまとめようと試みます（特定の thunk のブロック解除が早ければ早いほど、その入力の処理を迅速に開始できるためです）。
2. もし thunk の実行がキャンセルされた場合（エラー、優先順位の変更、または他のワーカーがすでに完了していることが判明した場合など）、アトムレイヤーはこのキャンセルを通知し、その thunk に関連するすべてのリクエストをダウンロードキューから削除できるようにします。

典型的なワーカーは、1GB/s のバーストでデータを送信し、通常は1回のバーストで7GBのデータをカバーします。

ログレイヤー

このレイヤーは実行状態に関する追加情報を保存し、その後の問題調査やパフォーマンス計測に役立てるために使われます。実行された thunk、実行にかかった時間、生成された結果の種類など、非常に詳細な情報を含みます。また、新しい DAG の構築（シリアライズされた DAG 自体を含む）やアトムの欠落の発見などの重要なイベントも記録されます。合計で、各ワーカーごとに毎日数ギガバイトのログが生成されます。

パフォーマンスへの影響を最小限にするために、各ワーカーは60秒ごと、または4メガバイトが蓄積されるたびに（これがアクティビティのバースト時に頻繁に発生します）、蓄積されたログを書き出します。これらは Azure Blob Storage のブロックブロブ⁴に書き込まれ、各ワーカーは単一のブロブで複数のライターをサポートする必要がないように専用のブロブを持ちます。

その後、Envision の本番環境外の他のマシンが、これらのログブロブを後から読み取り、クラスター上で発生した事象の詳細な統計をまとめることができます。

遠慮のない宣伝ですが、我々はソフトウェアエンジニアを募集しています。リモートワークも可能です.