Platform For AI:AIMaster： 弾力性と自動フォールトトレランス機能を備えたエンジン

ジョブのフォールトトレランスモニタリングを有効化した後、フォールトトレランスイベントの通知を設定できます。Workspace Detailsページで、Configure Workspace > Configure Event Notificationを選択し、Create Event Ruleをクリックして、イベントタイプをDLC task > Automatic Fault Toleranceに設定します。詳細については、「ワークスペースイベントセンター」をご参照ください。

NaN のトレーニング損失値など、トレーニングジョブで例外が発生した場合、コード内で AIMaster SDK を使用してカスタム通知メッセージを送信できます。

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 training loss for the job is 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 # 全体的な 30 分間の検出。
}
jm_config = jm.PyTorchConfig(**jm_config_params)
monitor = jm.Monitor(config=jm_config)

dist.init_process_group('nccl')

...

# Implement these two functions in the AIMaster SDK.
# You only need to add annotations to your functions.
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) # この関数のスコープ内でのみ、ハング検出を 3 分にリセットします。
def train():
    ...

@hang_detect(-1) # この関数のスコープ内でのみ、ハング検出を一時的に無効化します。
def test():
    ...

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