Platform For AI:AIMaster: 弾性フォールトトレランスエンジン

ジョブの再起動などのフォールトトレランスイベントに関する通知を受信する場合は、次の手順を実行して、そのような通知のルールを設定できます。[ワークスペースの詳細] ページに移動し、[イベント] タブをクリックし、[イベントルールの作成] をクリックします。 表示されるパネルで、[イベントタイプ] ドロップダウンリストから [DLCジョブ] > [自動フォールトトレランス] を選択します。 詳細については、「ワークスペース通知」をご参照ください。

AIMaster SDKを使用して、損失関数がNaNを返すときなど、エラーが発生したときにカスタム通知を設定することもできます。 サンプルコード：

from aimaster import job_monitor as jm

job_monitor_client = jm.Monitor(config=jm.PyTorchConfig())

...

if loss == Nan and rank == 0:
  st = job_monitor_client.send_custom_message(content="The loss function returns NaN")
  if not st.ok():
      print('failed to send message, error %s' % st.to_string())

フォールトトレランス監視機能は、一般的な再試行可能なエラーを自動的に検出できる組み込みモジュールを提供します。 エラーログに特定のキーワードに一致する情報が含まれている場合にフォールトトレランスをトリガーする場合は、次の方法を使用してカスタムキーワードを設定できます。 カスタムキーワードを設定した後、モジュールは失敗したインスタンスの末尾ログをスキャンして、カスタムキーワードに一致するキー情報を探します。

• PyTorchジョブのサンプル設定:

  from aimaster import job_monitor as jm
  
  jm_config_params = {}
  jm_config = jm.PyTorchConfig(**jm_config_params)
  monitor = jm.Monitor(config=jm_config)
  monitor.set_retryable_errors(["connect timeout", "error_yyy", "error_zzz"])

  上記のコードでは、カスタムキーワードはmonitor.set_retryable_errors関数を使用して設定されています。

• TensorFlowジョブのサンプル設定:

  from aimaster import job_monitor as jm
  
  jm_config_params = {}
  jm_config = jm.TFConfig(**jm_config_params)
  monitor = jm.Monitor(config=jm_config)
  monitor.set_retryable_errors(["connect timeout", "error_yyy", "error_zzz"])

デフォルトでは、ハング検出はジョブランタイム全体に対して設定されます。 ただし、異なるランタイムステージで異なる設定が必要になる場合があります。 例えば、大規模なジョブが初期化されている場合、インスタンス間の接続に時間がかかるが、ジョブの実行は高速である。 DLCでは、次の方法を使用して、ジョブの現在のステータスに基づいてハング検出ポリシーを設定できます。

monitor.reset_config(jm_config_params)

# Example:
#     monitor.reset_config(job_hang_interval=10)
#     or
#     config_params = {"job_hang_interval": 10, }
#     monitor.reset_config(**config_params)

PyTorchジョブのサンプル設定:

import torch
import torch.distributed as dist
from aimaster import job_monitor as jm

jm_config_params = {
    "job_hang_interval": 1800 # Configure a global hang detection policy. This value specifies that the maximum duration during which the job can be non-responsive is 30 minutes.
}
jm_config = jm.PyTorchConfig(**jm_config_params)
monitor = jm.Monitor(config=jm_config)

dist.init_process_group('nccl')

...

# impl these two funcs in aimaster sdk
# user just need to add annotations to their func
def reset_hang_detect(hang_seconds):
    jm_config_params = {
        "job_hang_interval": hang_seconds
    }
    monitor.reset_config(**jm_config_params)

def hang_detect(interval):
    reset_hang_detect(interval)
    ...

@hang_detect(180) # reset hang detect to 3 min, only for func scope
def train():
    ...

@hang_detect(-1) # disable hang detect temperally, only for func scope
def test():
    ...

for epoch in range(0, 100):
    train(epoch)
    test(epoch)
    self.scheduler.step()