人工智能平台 PAI：AIMaster：弹性自动容错引擎

任务开启容错监控后，如果您希望容错发生时可以进行通知，比如任务发生了重启，那么您可以在工作空间详情，选择事件中心，点击新建事件规则，事件类型选择DLC任务>任务自动容错。具体操作，请参见工作空间事件中心。

除了对任务容错信息通知外，当您判断任务训练出现异常时，比如loss出现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="任务训练loss出现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中配置的参数即为自定义容错关键字。

• TF任务自定义容错关键字配置示例

  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"])

目前任务Hang异常检测的配置是针对整个任务的，但是任务状态是分阶段的。例如：在任务初始化阶段，各个节点建立通信可能耗时比较长，但训练阶段日志更新比较快。为了在训练过程中能快速发现任务Hang异常的节点，DLC提供了分阶段自定义任务Hang异常检测功能，支持您在不同训练阶段配置不同的任务Hang异常检测时长，具体配置方法如下。

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任务分阶段自定义任务Hang异常检测示例如下。

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

jm_config_params = {
    "job_hang_interval": 1800 # 全局30min检测。
}
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()