平台适配

优雅容错模式基于故障重调度模式，集成优雅容错模式前请先完成故障重调度模式集成。集群调度组件提供管理进程样例，用户可直接在训练脚本中执行该样例启动管理进程，以下为执行参考样例。

训练脚本启动管理进程reset_process示例：
# 单机多卡和分布式
if [ $# == 6 ]; then
    export DEVICE_NUM=$1
    export SERVER_NUM=$2
    export RANK_SIZE=$1
    export RANK_TABLE_FILE=$3
 
    export SERVER_ID=$4
    device_each_server=$((DEVICE_NUM / SERVER_NUM))
    rank_start=$((${device_each_server} * SERVER_ID))
 
    DATA_PATH=$5
    CONFIG_PATH=$6
 
    # 先启动后台任务，最后留一个前台任务查看日志输出
    for((i=$((${device_each_server}-1)); i>=0; i--))
    do
        rankid=$((rank_start + i))
        export DEVICE_ID=${i}
        export RANK_ID=${rankid}
        rm -rf ${ROOT_PATH}/train_parallel${rankid}
        mkdir ${ROOT_PATH}/train_parallel${rankid}
        cp ${ROOT_PATH}/../*.py ${ROOT_PATH}/train_parallel${rankid}
        cp ${ROOT_PATH}/*.sh ${ROOT_PATH}/train_parallel${rankid}
        cp -r ${ROOT_PATH}/../src ${ROOT_PATH}/train_parallel${rankid}
        cd ${ROOT_PATH}/train_parallel${rankid} || exit
        echo "start training for rank $RANK_ID, device $DEVICE_ID"
        env > env.log
        python ${ROOT_PATH}/../train.py --run_distribute=True --device_num=${RANK_SIZE} --data_path=${DATA_PATH} --config_path=${CONFIG_PATH} &> log &
        train_pids[$i]=$!
    done
else
    echo "Invalid input parameter, usage: main.sh device_count server_count rank_table_file server_id dataset config_file_path" | tee -a log
    exit 1
fi
python -u ${ROOT_PATH}/reset_process.py -p "${train_pids[@]}"
wait

Reset_Process基于Python语言编写，其中关键能力模块示例如下，用户可参考关键模块能力实现该管理进程能力。

# Reset_Process中关键模块示例如下，用户可参考不同模块实现自己的管理进程
# 获取故障卡RANK和恢复卡RANK
    def get_fault_ranks(self):
        fault_rank_list = self._get_ranks_from_cm(self.reset_cm_path, "unrecovered")
        if len(fault_rank_list) != 0:
            self.fault_rank_list = fault_rank_list
        return fault_rank_list
 
    def get_recover_ranks(self):
        recover_rank_list = self._get_ranks_from_cm(self.reset_cm_path, "recovered")
        if len(recover_rank_list) != 0:
            self.recover_rank_list = recover_rank_list
        return recover_rank_list
# 停止训练进程
def _kill_abnormal_process(self, abnormal_rank_list: list):
        if self.killed_abnormal:
            return
        try:
            logger.info(f"to kill abnormal rank {abnormal_rank_list}")
            self._process_manager.kill_fault_process(abnormal_rank_list)
            self.killed_abnormal = True
        except Exception as e:
            logger.error(f"an unexpected error {e} occur when kill abnormal process")
            self.exit_recover_process()
                     
    def _kill_normal_process(self, normal_rank_list: list):
        if self.killed_normal:
            return
        try:
            logger.info(f"to kill normal rank {normal_rank_list}")
            self._process_manager.kill_fault_process(normal_rank_list)
            self.killed_normal = True
        except Exception as e:
            logger.error(f"an unexpected error {e} occur when kill normal process")
            self.exit_recover_process()
# 重启训练进程
def restore_train_process(self):
        """
        Recover all target processes in this node
        """
        new_pid_list = multiprocessing.Manager().list()
        if not self.all_stopped() or self._restart:
            return new_pid_list
 
        process = []
        for rank in self._cmd_dict:
            command = self._cmd_dict[rank]
            pwd_path = self._env_dict[rank]['PWD']
            env_info = self._env_dict[rank]
            p = multiprocessing.Process(target=_run_recover, args=(command, pwd_path, env_info, new_pid_list))
            process.append(p)
            p.start()
 
        for p in process:
            p.join()
        self._restart = True
        logger.info(f"new pids are:{new_pid_list}")
   return new_pid_list