分布式并行训练基础样例（GPU）

概述

本篇教程我们主要讲解，如何在GPU处理器硬件平台上，利用MindSpore通过数据并行及自动并行模式，使用CIFAR-10数据集训练ResNet-50网络。

你可以在这里下载完整的样例代码：

https://gitee.com/mindspore/docs/tree/r1.7/docs/sample_code/distributed_training

目录结构如下：

└─sample_code
    ├─distributed_training
    │      rank_table_16pcs.json
    │      rank_table_8pcs.json
    │      rank_table_2pcs.json
    │      cell_wrapper.py
    │      model_accu.py
    │      resnet.py
    │      resnet50_distributed_training.py
    │      resnet50_distributed_training_gpu.py
    │      resnet50_distributed_training_grad_accu.py
    │      run.sh
    │      run_gpu.sh
    │      run_grad_accu.sh
    │      run_cluster.sh

其中，resnet.py和resnet50_distributed_training_gpu.py是定义网络结构的脚本。run_gpu.sh是执行脚本，其余文件为Ascend 910的样例代码。

准备环节

为了保证分布式训练的正常进行，我们需要先对分布式环境进行配置和初步的测试。在完成之后，再做CIFAR-10数据集的准备。

配置分布式环境

OpenMPI-4.0.3：MindSpore采用的多进程通信库。

OpenMPI-4.0.3源码下载地址：https://www.open-mpi.org/software/ompi/v4.0/，选择openmpi-4.0.3.tar.gz下载。

参考OpenMPI官网教程安装：https://www.open-mpi.org/faq/?category=building#easy-build。
主机间免密登陆（涉及多机训练时需要）。若训练涉及多机，则需要配置多机间免密登陆，可参考以下步骤进行配置：
1. 每台主机确定同一个用户作为登陆用户（不推荐root）；
2. 执行ssh-keygen -t rsa -P ""生成密钥；
3. 执行ssh-copy-id DEVICE-IP设置需要免密登陆的机器IP；
4. 执行ssh DEVICE-IP，若不需要输入密码即可登录，则说明以上配置成功；
5. 在所有机器上执行以上命令，确保两两互通。

调用集合通信库

在GPU硬件平台上，MindSpore分布式并行训练中的通信使用的是英伟达集合通信库NVIDIA Collective Communication Library(以下简称为NCCL)。

GPU平台上，MindSpore暂不支持用户进行：

get_local_rank、get_local_size、get_world_rank_from_group_rank、get_group_rank_from_world_rank、create_group操作。

下面是调用通信库的代码样例，设文件名为nccl_allgather.py：

# nccl_allgather.py
import numpy as np
import mindspore.ops as ops
import mindspore.nn as nn
from mindspore import context, Tensor
from mindspore.communication import init, get_rank


class Net(nn.Cell):
    def __init__(self):
        super(Net, self).__init__()
        self.allgather = ops.AllGather()

    def construct(self, x):
        return self.allgather(x)


if __name__ == "__main__":
    context.set_context(mode=context.GRAPH_MODE, device_target="GPU")
    init("nccl")
    value = get_rank()
    input_x = Tensor(np.array([[value]]).astype(np.float32))
    net = Net()
    output = net(input_x)
    print(output)

其中，

mode=context.GRAPH_MODE：使用分布式训练需要指定运行模式为图模式（PyNative模式不支持并行）。
device_target="GPU": 指定设备为GPU。
init("nccl")：使能NCCL通信，并完成分布式训练初始化操作。
get_rank()：获得当前进程的rank号。
ops.AllGather: 在GPU上，该算子会调用NCCL的AllGather通信操作，其含义以及更多的例子可在分布式集合通信原语中找到。

在GPU硬件平台上，MindSpore采用OpenMPI的mpirun来启动进程，通常每一个进程对应一个计算设备。

mpirun -n DEVICE_NUM python nccl_allgather.py

其中，DEVICE_NUM为所在机器的GPU数量。以DEVICE_NUM=4为例，预期的输出为：

[[0.],
 [1.],
 [2.],
 [3.]]

输出日志在程序执行后，可在log/1/rank.0中找到。若得到以上输出，则说明OpenMPI和NCCL工作正常，进程正常启动。

下载数据集

本样例采用CIFAR-10数据集，由10类32*32的彩色图片组成，每类包含6000张图片，共60000张。其中训练集共50000张图片，测试集共10000张图片。

CIFAR-10数据集下载链接：http://www.cs.toronto.edu/~kriz/cifar-10-binary.tar.gz，如果点击下载不成功，请尝试复制链接地址后下载。

Linux机器可采用以下命令下载到终端当前路径并解压数据集，解压后的数据所在文件夹为cifar-10-batches-bin。

wget http://www.cs.toronto.edu/~kriz/cifar-10-binary.tar.gz
tar -zxvf cifar-10-binary.tar.gz

数据并行模式加载数据集

分布式训练时，数据是以数据并行的方式导入的。下面我们以CIFAR-10数据集为例，介绍以数据并行方式导入CIFAR-10数据集的方法，data_path是指数据集的路径，即cifar-10-batches-bin文件夹的路径。

from mindspore import dtype as mstype
import mindspore.dataset as ds
import mindspore.dataset.transforms.c_transforms as C
import mindspore.dataset.vision.c_transforms as vision
from mindspore.communication import get_rank, get_group_size


def create_dataset(data_path, repeat_num=1, batch_size=32, rank_id=0, rank_size=1):
    resize_height = 224
    resize_width = 224
    rescale = 1.0 / 255.0
    shift = 0.0

    # get rank_id and rank_size
    rank_id = get_rank()
    rank_size = get_group_size()
    data_set = ds.Cifar10Dataset(data_path, num_shards=rank_size, shard_id=rank_id)

    # define map operations
    random_crop_op = vision.RandomCrop((32, 32), (4, 4, 4, 4))
    random_horizontal_op = vision.RandomHorizontalFlip()
    resize_op = vision.Resize((resize_height, resize_width))
    rescale_op = vision.Rescale(rescale, shift)
    normalize_op = vision.Normalize((0.4465, 0.4822, 0.4914), (0.2010, 0.1994, 0.2023))
    changeswap_op = vision.HWC2CHW()
    type_cast_op = C.TypeCast(mstype.int32)

    c_trans = [random_crop_op, random_horizontal_op]
    c_trans += [resize_op, rescale_op, normalize_op, changeswap_op]

    # apply map operations on images
    data_set = data_set.map(operations=type_cast_op, input_columns="label")
    data_set = data_set.map(operations=c_trans, input_columns="image")

    # apply shuffle operations
    data_set = data_set.shuffle(buffer_size=10)

    # apply batch operations
    data_set = data_set.batch(batch_size=batch_size, drop_remainder=True)

    # apply repeat operations
    data_set = data_set.repeat(repeat_num)

    return data_set

其中，与单机不同的是，在数据集接口需要传入num_shards和shard_id参数，分别对应卡的数量和逻辑序号，建议通过NCCL接口获取：

get_rank：获取当前设备在集群中的ID。
get_group_size：获取集群数量。

数据并行场景加载数据集时，建议对每卡指定相同的数据集文件，若是各卡加载的数据集不同，可能会影响计算精度。

定义网络

在GPU硬件平台上，网络的定义和Ascend 910 AI处理器一致。 数据并行及自动并行 模式下，网络定义方式与单机写法一致，可以参考 ResNet网络样例脚本。。

半自动并行模式时，未配置策略的算子默认以数据并行方式执行。

自动并行模式支持通过策略搜索算法自动获取高效的算子并行策略，同时也支持用户对算子手动配置特定的并行策略。

如果某个parameter被多个算子使用，则每个算子对这个parameter的切分策略需要保持一致，否则将报错。

定义损失函数及优化器

与在Ascend的分布式并行训练基础样例一致。

定义损失函数

自动并行以算子为粒度切分模型，通过算法搜索得到最优并行策略，所以与单机训练不同的是，为了有更好的并行训练效果，损失函数建议使用MindSpore算子来实现，而不是直接用封装好的损失函数类。

在Loss部分，我们采用SoftmaxCrossEntropyWithLogits的展开形式，即按照数学公式，将其展开为多个MindSpore算子进行实现，样例代码如下：

import mindspore.ops as ops
from mindspore import Tensor
from mindspore import dtype as mstype
import mindspore.nn as nn


class SoftmaxCrossEntropyExpand(nn.Cell):
    def __init__(self, sparse=False):
        super(SoftmaxCrossEntropyExpand, self).__init__()
        self.exp = ops.Exp()
        self.sum = ops.ReduceSum(keep_dims=True)
        self.onehot = ops.OneHot()
        self.on_value = Tensor(1.0, mstype.float32)
        self.off_value = Tensor(0.0, mstype.float32)
        self.div = ops.RealDiv()
        self.log = ops.Log()
        self.sum_cross_entropy = ops.ReduceSum(keep_dims=False)
        self.mul = ops.Mul()
        self.mul2 = ops.Mul()
        self.mean = ops.ReduceMean(keep_dims=False)
        self.sparse = sparse
        self.max = ops.ReduceMax(keep_dims=True)
        self.sub = ops.Sub()

    def construct(self, logit, label):
        logit_max = self.max(logit, -1)
        exp = self.exp(self.sub(logit, logit_max))
        exp_sum = self.sum(exp, -1)
        softmax_result = self.div(exp, exp_sum)
        if self.sparse:
            label = self.onehot(label, ops.shape(logit)[1], self.on_value, self.off_value)
        softmax_result_log = self.log(softmax_result)
        loss = self.sum_cross_entropy((self.mul(softmax_result_log, label)), -1)
        loss = self.mul2(ops.scalar_to_array(-1.0), loss)
        loss = self.mean(loss, -1)

        return loss

定义优化器

采用Momentum优化器作为参数更新工具，这里定义与单机一致，不再展开，具体可以参考样例代码中的实现。

训练网络

训练之前，我们需要先配置一些自动并行的参数。context.set_auto_parallel_context是配置并行训练模式的接口，必须在初始化网络之前调用。常用参数包括：

parallel_mode：分布式并行模式，默认为单机模式ParallelMode.STAND_ALONE。在本例中，可选择数据并行ParallelMode.DATA_PARALLEL 及自动并行ParallelMode.AUTO_PARALLEL。
parameter_broadcast：训练开始前自动广播0号卡上数据并行的参数权值到其他卡上，默认值为False。
gradients_mean：反向计算时，框架内部会将数据并行参数分散在多台机器的梯度值进行收集，得到全局梯度值后再传入优化器中更新。默认值为False，设置为True对应allreduce_mean 操作，False对应allreduce_sum操作。
device_num和global_rank建议采用默认值，框架内会调用NCCL接口获取。

如脚本中存在多个网络用例，请在执行下个用例前调用context.reset_auto_parallel_context将所有参数还原到默认值。

在下面的样例中我们指定并行模式为自动并行，用户如需切换为数据并行模式只需将parallel_mode改为DATA_PARALLEL。

from mindspore import context, Model
from mindspore.nn import Momentum
from mindspore.train.callback import LossMonitor
from mindspore.context import ParallelMode
from mindspore.communication import init
from resnet import resnet50

context.set_context(mode=context.GRAPH_MODE, device_target="GPU")
init("nccl")


def test_train_cifar(epoch_size=10):
    context.set_auto_parallel_context(parallel_mode=ParallelMode.AUTO_PARALLEL, gradients_mean=True)
    loss_cb = LossMonitor()
    dataset = create_dataset(data_path)
    batch_size = 32
    num_classes = 10
    net = resnet50(batch_size, num_classes)
    loss = SoftmaxCrossEntropyExpand(sparse=True)
    opt = Momentum(filter(lambda x: x.requires_grad, net.get_parameters()), 0.01, 0.9)
    model = Model(net, loss_fn=loss, optimizer=opt)
    model.train(epoch_size, dataset, callbacks=[loss_cb], dataset_sink_mode=True)

其中，

dataset_sink_mode=True：表示采用数据集的下沉模式，即训练的计算下沉到硬件平台中执行。
LossMonitor：能够通过回调函数返回Loss值，用于监控损失函数。

运行脚本

在GPU硬件平台上，MindSpore采用OpenMPI的mpirun进行分布式训练。在完成了模型、损失函数和优化器的定义之后，我们就完成了模型的并行策略的配置，下面直接执行运行脚本。

单机多卡训练

下面以使用8张卡的分布式训练脚本为例，演示如何运行脚本：

你可以在这里找到样例的运行脚本：

https://gitee.com/mindspore/docs/blob/r1.7/docs/sample_code/distributed_training/run_gpu.sh。

如果通过root用户执行脚本，mpirun需要加上--allow-run-as-root参数。

#!/bin/bash

echo "=============================================================================================================="
echo "Please run the script as: "
echo "bash run_gpu.sh DATA_PATH"
echo "For example: bash run_gpu.sh /path/dataset"
echo "It is better to use the absolute path."
echo "=============================================================================================================="
DATA_PATH=$1
export DATA_PATH=${DATA_PATH}

rm -rf device
mkdir device
cp ./resnet50_distributed_training_gpu.py ./resnet.py ./device
cd ./device
echo "start training"
mpirun -n 8 pytest -s -v ./resnet50_distributed_training_gpu.py > train.log 2>&1 &

脚本会在后台运行，日志文件会保存到device目录下，共跑了10个epoch，每个epoch有234个step，关于Loss部分结果保存在train.log中。将loss值grep出来后，示例如下：

epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854

多机多卡训练

在运行多机多卡训练前，需要保证每个节点上都有相同的OpenMPI、NCCL、Python以及MindSpore版本。

mpirun -H

若训练涉及多机，则需要额外在mpirun命令中设置多机配置。你可以直接在mpirun命令中用-H选项进行设置，比如

mpirun -n 16 -H DEVICE1_IP:8,DEVICE2_IP:8 python hello.py

表示在ip为DEVICE1_IP和DEVICE2_IP的机器上分别起8个进程运行程序。

mpirun –hostfile

GPU的多机多卡的执行也可以通过构造hostfile文件来进行。为方便调试，建议用这种方法来执行多机多卡脚本。之后使用mpirun --hostfile $HOST_FILE的形式来执行。下面我们以hostfile启动方式来给出详细的多机多卡配置。

hostfile文件每一行格式为[hostname] slots=[slotnum]，hostname可以是ip或者主机名。需要注意的是，不同机器上的用户名需要相同，但是hostname不可以相同。如下，表示在DEVICE1上有8张卡；ip为192.168.0.1的机器上也有8张卡：

DEVICE1 slots=8
192.168.0.1 slots=8

两机十六卡的执行脚本如下，需要传入变量DATA_PATH和HOSTFILE，表示数据集的路径和hostfile文件的路径。我们需要设置mpi中mca的btl参数来指定进行mpi通信的网卡，否则可能会在调用mpi接口时初始化失败。btl参数指定了节点间采用tcp协议，节点内采用环路进行通信。btl_tcp_if_include指定节点间通信所经过的网卡的ip地址需要在给定的子网中。更多mpirun的选项设置可见OpenMPI的官网。

#!/bin/bash

DATA_PATH=$1
HOSTFILE=$2

rm -rf device
mkdir device
cp ./resnet50_distributed_training_gpu.py ./resnet.py ./device
cd ./device
echo "start training"
mpirun -n 16 --mca btl tcp,self --mca btl_tcp_if_include 192.168.0.0/24 --hostfile $HOSTFILE -x DATA_PATH=$DATA_PATH -x PATH -mca pml ob1 mpirun_gpu_clusher.sh &

考虑到不同机器上的一些环境变量可能会不一样，我们采用mpirun启动一个mpirun_gpu_cluster.sh的形式，在不同机器上的该脚本文件中指定所需的环境变量。此处我们配置了NCCL_SOCKET_IFNAME，来指定NCCL进行通信时的网卡。

#!/bin/bash
# mpirun_gpu_clusher.sh
# 你可以在这里设置每台机器上不同的环境变量，如下面的网卡名字

export NCCL_SOCKET_IFNAME="en5" # 需进行节点间通信的网卡的名字，不同机器上可能不一致，使用ifconfig查看。
pytest -s -v ./resnet50_distributed_training_gpu.py > train.log 2>&1 &

分布式训练模型参数保存与加载

在GPU上进行分布式训练时，模型参数的保存和加载的方法与Ascend上一致，可参考分布式训练模型参数保存和加载。

不依赖OpenMPI进行训练

出于训练时的安全及可靠性要求，MindSpore GPU还支持不依赖OpenMPI的分布式训练。

OpenMPI在分布式训练的场景中，起到在Host侧同步数据以及进程间组网的功能；MindSpore通过复用Parameter Server模式训练架构，取代了OpenMPI能力。

参考Parameter Server模式训练教程，将多个MindSpore训练进程作为Worker启动，并且额外启动一个Scheduler，对脚本做少量修改，即可执行不依赖OpenMPI的分布式训练。

执行Worker脚本前需要导出环境变量，如环境变量设置:

export MS_SERVER_NUM=0                # Server number
export MS_WORKER_NUM=8                # Worker number
export MS_SCHED_HOST=127.0.0.1        # Scheduler IP address
export MS_SCHED_PORT=6667             # Scheduler port
export MS_ROLE=MS_WORKER              # The role of this process: MS_SCHED represents the scheduler, MS_WORKER represents the worker, MS_PSERVER represents the Server

在此模式下，不建议启动MS_PSERVER角色的进程，因为此角色在数据并行训练中无影响。

运行脚本

在GPU硬件平台上，下面以使用8张卡的分布式训练脚本为例，演示如何运行脚本：

你可以在这里找到样例的运行目录：

https://gitee.com/mindspore/docs/tree/r1.7/docs/sample_code/distributed_training。

相比OpenMPI方式启动，此模式需要调用Parameter Server模式中的set_ps_context接口，告诉MindSpore此次任务使用了PS模式训练架构:

from mindspore import context
from mindspore.communication import init

if __name__ == "__main__":
    context.set_context(mode=context.GRAPH_MODE, device_target="GPU")
    context.set_ps_context(config_file_path="/path/to/config_file.json", enable_ssl=True,
                           client_password="123456", server_password="123456")
    init("nccl")
    ...

其中，

mode=context.GRAPH_MODE：使用分布式训练需要指定运行模式为图模式（PyNative模式不支持并行）。
init("nccl")：使能NCCL通信，并完成分布式训练初始化操作。
默认情况下，安全加密通道是关闭的，需要通过set_ps_context正确配置安全加密通道或者关闭安全加密通道后，才能调用init(“nccl”)，否则初始化组网会失败。

若想使用安全加密通道，请设置context.set_ps_context(config_file_path="/path/to/config_file.json", enable_ssl=True, client_password="123456", server_password="123456") 等配置，详细参数配置说明请参考Python API mindspore.context.set_ps_context，以及本文档安全认证章节。

脚本内容run_gpu_cluster.sh如下，在启动Worker和Scheduler之前，需要添加相关环境变量设置：

#!/bin/bash

echo "=========================================="
echo "Please run the script as: "
echo "bash run_gpu_cluster.sh DATA_PATH"
echo "For example: bash run_gpu_cluster.sh /path/dataset"
echo "It is better to use the absolute path."
echo "==========================================="
DATA_PATH=$1
export DATA_PATH=${DATA_PATH}

rm -rf device
mkdir device
cp ./resnet50_distributed_training_gpu.py ./resnet.py ./device
cd ./device
echo "start training"

# Launch 8 workers.
for((i=0;i<8;i++));
do
    export MS_WORKER_NUM=8
    export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
    export MS_SCHED_PORT=XXXX             # Scheduler port
    export MS_ROLE=MS_WORKER
    pytest -s -v ./resnet50_distributed_training_gpu.py > worker_$i.log 2>&1 &
done

# Launch 1 scheduler.
export MS_WORKER_NUM=8
export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
export MS_SCHED_PORT=XXXX             # Scheduler port
export MS_ROLE=MS_SCHED
pytest -s -v ./resnet50_distributed_training_gpu.py > scheduler.log 2>&1 &

执行如下指令：

./run_gpu_cluster.sh DATA_PATH

即可单机内部执行8卡分布式训练，若希望执行跨机训练，则需要将脚本拆分，如执行2机8卡训练，每台机器执行启动4Worker：

脚本run_gpu_cluster_1.sh在机器1上启动1Scheduler和Worker1到Worker4：

#!/bin/bash

echo "=========================================="
echo "Please run the script as: "
echo "bash run_gpu_cluster.sh DATA_PATH"
echo "For example: bash run_gpu_cluster.sh /path/dataset"
echo "It is better to use the absolute path."
echo "==========================================="
DATA_PATH=$1
export DATA_PATH=${DATA_PATH}

rm -rf device
mkdir device
cp ./resnet50_distributed_training_gpu.py ./resnet.py ./device
cd ./device
echo "start training"

# Launch 1-4 workers.
for((i=0;i<4;i++));
do
    export MS_WORKER_NUM=8
    export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
    export MS_SCHED_PORT=XXXX             # Scheduler port
    export MS_ROLE=MS_WORKER
    pytest -s -v ./resnet50_distributed_training_gpu.py > worker_$i.log 2>&1 &
done

# Launch 1 scheduler.
export MS_WORKER_NUM=8
export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
export MS_SCHED_PORT=XXXX             # Scheduler port
export MS_ROLE=MS_SCHED
pytest -s -v ./resnet50_distributed_training_gpu.py > scheduler.log 2>&1 &

脚本run_gpu_cluster_2.sh在机器2上启动Worker5到Worker8(无需再执行Scheduler)：

#!/bin/bash

echo "=========================================="
echo "Please run the script as: "
echo "bash run_gpu_cluster.sh DATA_PATH"
echo "For example: bash run_gpu_cluster.sh /path/dataset"
echo "It is better to use the absolute path."
echo "==========================================="
DATA_PATH=$1
export DATA_PATH=${DATA_PATH}

rm -rf device
mkdir device
cp ./resnet50_distributed_training_gpu.py ./resnet.py ./device
cd ./device
echo "start training"

# Launch 5-8 workers.
for((i=4;i<8;i++));
do
    export MS_WORKER_NUM=8
    export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
    export MS_SCHED_PORT=XXXX             # Scheduler port
    export MS_ROLE=MS_WORKER
    pytest -s -v ./resnet50_distributed_training_gpu.py > worker_$i.log 2>&1 &
done

在两台主机分别执行：

./run_gpu_cluster_1.sh DATA_PATH

./run_gpu_cluster_2.sh DATA_PATH

即可执行2机8卡分布式训练任务。

若希望启动数据并行模式训练，需要将脚本resnet50_distributed_training_gpu.py中set_auto_parallel_context入参并行模式改为DATA_PARALLEL:

context.set_auto_parallel_context(parallel_mode=ParallelMode.DATA_PARALLEL, gradients_mean=True)

脚本会在后台运行，日志文件会保存到当前目录下，共跑了10个epoch，每个epoch有234个step，关于Loss部分结果保存在worker_*.log中。将loss值grep出来后，示例如下：

epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854
epoch: 1 step: 1, loss is 2.3025854

安全认证

要支持节点/进程间的SSL安全认证，要开启安全认证，通过Python API mindspore.context.set_ps_context配置enable_ssl=True(不传入时默认为False，表示不启用SSL安全认证)，config_file_path指定的config.json配置文件需要添加如下字段：

{
  "server_cert_path": "server.p12",
  "crl_path": "",
  "client_cert_path": "client.p12",
  "ca_cert_path": "ca.crt",
  "cipher_list": "ECDHE-R SA-AES128-GCM-SHA256:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384:DHE-RSA-AES128-GCM-SHA256:DHE-DSS-AES128-GCM-SHA256:DHE-RSA-AES256-GCM-SHA384:DHE-DSS-AES256-GCM-SHA384:DHE-PSK-AES128-GCM-SHA256:DHE-PSK-AES256-GCM-SHA384:DHE-PSK-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305:ECDHE-PSK-CHACHA20-POLY1305:DHE-RSA-AES128-CCM:DHE-RSA-AES256-CCM:DHE-RSA-CHACHA20-POLY1305:DHE-PSK-AES128-CCM:DHE-PSK-AES256-CCM:ECDHE-ECDSA-AES128-CCM:ECDHE-ECDSA-AES256-CCM:ECDHE-ECDSA-CHACHA20-POLY1305",
  "cert_expire_warning_time_in_day": 90
}

server_cert_path: 服务端包含了证书和秘钥的密文的p12文件（SSL专用证书文件）路径。
crl_path: 吊销列表（用于区分无效不可信证书和有效可信证书）的文件路径。
client_cert_path: 客户端包含了证书和秘钥的密文的p12文件（SSL专用证书文件）路径。
ca_cert_path: 根证书路径。
cipher_list: 密码套件（支持的SSL加密类型列表）。
cert_expire_warning_time_in_day: 证书过期的告警时间。

p12文件中的秘钥为密文存储，在启动时需要传入密码，具体参数请参考Python API mindspore.context.set_ps_context中的client_password以及server_password字段。

容灾恢复

模型训练对分布式训练架构的可靠性、可服务性要求比较高，MindSpore支持数据并行下容灾恢复，多卡数据并行训练场景集群(多个Worker和1个Scheduler)中存在进程异常退出，被重新拉起后，训练任务继续能正常执行；

场景约束：在图模式下，采用MindData进行数据下沉模式训练，开启数据并行模式，采用上述的非OpenMPI的方式拉起Worker进程。

在上述场景下，训练过程中如果有节点挂掉，保证在相同的环境变量（MS_ENABLE_RECOVERY 和 MS_RECOVERY_PATH）下，重新拉起对应进程对应的脚本后训练可继续，并且不影响精度收敛。

1）开启容灾：

通过环境变量开启容灾：

export MS_ENABLE_RECOVERY=1             #开启容灾
export MS_RECOVERY_PATH=“/xxx/xxx”      #配置持久化路径文件夹，Worker和Scheduler进程在执行过程中会进行必要的持久化，如用于恢复组网的节点信息以及训练业务中间状态等

2）配置checkpoint保存间隔，样例如下：

ckptconfig = CheckpointConfig(save_checkpoint_steps=100, keep_checkpoint_max=5)
ckpoint_cb = ModelCheckpoint(prefix='train', directory="./ckpt_of_rank_/"+str(get_rank()), config=ckptconfig)

每个Worker都开启保存checkpoint，并用不同的路径（如上述样例中的directory的设置使用了rank id，保证路径不会相同），防止同名checkpoint保存冲突。checkpoint用于异常进程恢复和正常进程回滚，训练的回滚是指集群中各个Worker都恢复到最新的checkpoint对应的状态，同时数据侧也回退到对应的step，然后继续训练。保存checkpoint的间隔是可配置的，这个间隔决定了容灾恢复的粒度，间隔越小，恢复到上次保存checkpoint所回退的step数就越小，但保存checkpoint频繁也可能会影响训练效率，间隔越大则效果相反。keep_checkpoint_max至少设置为2(防止checkpoint保存失败)。

样例的运行目录：

https://gitee.com/mindspore/docs/tree/r1.7/docs/sample_code/distributed_training。

涉及到的脚本有run_gpu_cluster_recovery.sh, resnet50_distributed_training_gpu_recovery.py, resnet.py 脚本内容run_gpu_cluster_recovery.sh如下

#!/bin/bash

echo "=========================================="
echo "Please run the script as: "
echo "bash run_gpu_cluster_recovery.sh DATA_PATH"
echo "For example: bash run_gpu_cluster_recovery.sh /path/dataset"
echo "It is better to use the absolute path."
echo "==========================================="
DATA_PATH=$1
export DATA_PATH=${DATA_PATH}

export MS_ENABLE_RECOVERY=1      # Enable recovery
export MS_RECOVERY_PATH=/XXX/XXX # Set recovery path

rm -rf device
mkdir device
cp ./resnet50_distributed_training_gpu_recovery.py ./resnet.py ./device
cd ./device
echo "start training"

# Launch 1 scheduler.
export MS_WORKER_NUM=8
export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
export MS_SCHED_PORT=XXXX             # Scheduler port
export MS_ROLE=MS_SCHED
export MS_NODE_ID=sched               # The node id for Scheduler
pytest -s -v ./resnet50_distributed_training_gpu_recovery.py > scheduler.log 2>&1 &

# Launch 8 workers.
for((i=0;i<8;i++));
do
    export MS_WORKER_NUM=8
    export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
    export MS_SCHED_PORT=XXXX             # Scheduler port
    export MS_ROLE=MS_WORKER
    export MS_NODE_ID=worker_$i           # The node id for Workers
    pytest -s -v ./resnet50_distributed_training_gpu_recovery.py > worker_$i.log 2>&1 &
done

在启动Worker和Scheduler之前，需要添加相关环境变量设置, 如Scheduler的IP和Port，当前进程的角色是Worker还是Scheduler。

执行下面的命令即可启动一个单机8卡的数据并行训练

bash run_gpu_cluster_recovery.sh YOUR_DATA_PATH"

分布式训练开始，若训练过程中遇到异常，如进程异常退出，然后再重新启动对应的进程，训练流程即可恢复：例如训练中途Scheduler进程异常退出，可执行下列命令重新启动Scheduler：

export DATA_PATH=YOUR_DATA_PATH
export MS_ENABLE_RECOVERY=1           # Enable recovery
export MS_RECOVERY_PATH=/XXX/XXX      # Set recovery path

cd ./device

# Launch 1 scheduler.
export MS_WORKER_NUM=8
export MS_SCHED_HOST=XXX.XXX.XXX.XXX  # Scheduler IP address
export MS_SCHED_PORT=XXXX             # Scheduler port
export MS_ROLE=MS_SCHED
export MS_NODE_ID=sched               # The node id for Scheduler
pytest -s -v ./resnet50_distributed_training_gpu_recovery.py > scheduler.log 2>&1 &

Worker和Scheduler的组网会自动恢复。

Worker进程出现异常退出处理方式类似(注:Worker进程出现异常退出，需要等30s后再拉起才能恢复训练，在这之前，Scheduler为了防止网络抖动和恶意注册，拒绝相同node id的Worker再次注册)。

在K8s集群中使用ms-operator进行分布式训练

MindSpore Operator 是MindSpore在Kubernetes上进行分布式训练的插件。CRD（Custom Resource Definition）中定义了Scheduler、PS、Worker三种角色，用户只需配置yaml文件，即可轻松实现分布式训练。

当前ms-operator支持普通单Worker训练、PS模式的单Worker训练以及自动并行（例如数据并行、模型并行等）的Scheduler、Worker启动。详细流程请参考ms-operator。