加速库通信算子介绍

加速库通信算子使用流程

加速库的通信算子实现多卡、多机通信主要有两个步骤：初始化通信域（参考《HCCL集合通信库接口参考》的“通信域管理”章节）和执行通信算子。

​通信域是指发生通信的范围，例如想做一次AllReduce，需要在哪些卡之间做，这就是一个通信域。初始化通信域方式主要有两种：

• 加速库内部为用户初始化通信域，每张卡获得各自的通信域句柄去执行算子。
• 用户通过配置“ALLReduceParam”中的“HcclComm”来传入对应通信卡的通信句柄，当用户传入后，加速库内部不会再初始化，“HcclComm”具体含义参考《HCCL集合通信库接口参考》。

​一般来说，主要是由加速库内部为用户初始化通信域，如图5所示有两种方式：

• 基于rankRoot节点的信息创建通信域仅适用于单机场景。
• 基于rankTableFile创建通信域同时适用于单机和多机通信。

用户通过配置不同的Operation Param来创建Operation，就可以通过不同的方式来初始化通信域，初始化通信域成功后，会执行对应的通信算子接口，实现通信。

• 加速库单机二卡多进程通信计算并行用例：以下demo运行方式都与算子使用指导相同。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100

  #include <acl/acl.h> #include <atb/atb_infer.h> #include <iostream> #include <unistd.h> #include <sys/wait.h> void LinearallreduceSample(int rank, int worldsize) { int ret = aclInit(nullptr); //1、设置每个进程对应的deviceId，后续通信算子在该device上执行 int deviceId = rank; aclError status = aclrtSetDevice(deviceId); //2、LinearParallelOperation参数配置，对于多卡多进程通信，需要配置每个进程中算子参数的rank号，按照从0到ranksize-1的顺序为每个算子配置 atb::infer::LinearParallelParam param; param.rank = rank; param.rankRoot = 0; param.rankSize = worldsize; param.commMode = atb::infer::CommMode::COMM_MULTI_PROCESS; param.backend = "hccl"; param.transWeight = false; param.type = atb::infer::LinearParallelParam::ParallelType::LINEAR_ALL_REDUCE; atb::Operation *op = nullptr; atb::Status st = atb::CreateOperation(param, &op); //3、构造其输入输出 atb::Tensor input; input.desc.dtype = ACL_FLOAT16; input.desc.format = ACL_FORMAT_ND; input.desc.shape.dimNum = 2; input.desc.shape.dims[0] = 3; input.desc.shape.dims[1] = 32; input.dataSize = atb::Utils::GetTensorSize(input); status = aclrtMalloc(&input.deviceData, input.dataSize, ACL_MEM_MALLOC_HUGE_FIRST); atb::Tensor weight; weight.desc.dtype = ACL_FLOAT16; weight.desc.format = ACL_FORMAT_ND; weight.desc.shape.dimNum = 2; weight.desc.shape.dims[0] = 32; weight.desc.shape.dims[1] = 5; weight.dataSize = atb::Utils::GetTensorSize(weight); status = aclrtMalloc(&weight.deviceData, weight.dataSize, ACL_MEM_MALLOC_HUGE_FIRST); atb::Tensor output; output.desc.dtype = ACL_FLOAT16; output.desc.format = ACL_FORMAT_ND; output.desc.shape.dimNum = 2; output.desc.shape.dims[0] = 3; output.desc.shape.dims[1] = 5; output.dataSize = atb::Utils::GetTensorSize(output); status = aclrtMalloc(&output.deviceData, output.dataSize, ACL_MEM_MALLOC_HUGE_FIRST); atb::VariantPack variantPack; variantPack.inTensors = { input, weight }; variantPack.outTensors = { output }; // 4、算子准备与执行，与单算子准备执行流程相同，区别在于通信算子需要初始化通信域成功才会执行。 atb::Context *context = nullptr; st = atb::CreateContext(&context); aclrtStream stream = nullptr; status = aclrtCreateStream(&stream); context->SetExecuteStream(stream); uint64_t workspaceSize = 0; st = op->Setup(variantPack, workspaceSize, context); void *workspace = nullptr; if (workspaceSize > 0) { status = aclrtMalloc(&workspace, workspaceSize, ACL_MEM_MALLOC_HUGE_FIRST); } st = op->Execute(variantPack, (uint8_t *)workspace, workspaceSize, context); std::cout << "rank: " << rank << " executed END." << std::endl; //5、资源释放，下面用例均相同 status = aclrtDestroyStream(stream); // 销毁stream status = aclrtFree(workspace); // 销毁workspace st = atb::DestroyOperation(op); // 销毁op对象 st = atb::DestroyContext(context); // 销毁context // 下面代码为释放Tensor的示例代码，实际使用时需释放VariantPack中的所有tensor status = aclrtFree(tensor.deviceData); tensor.deviceData = nullptr; tensor.dataSize = 0; aclFinalize(); } int main(int argc, const char *argv[]) { // 多进程两卡通信 const int worldsize = 2; for (int i = 0; i < worldsize; ++i) { pid_t pid = fork(); if (pid == 0) { LinearallreduceSample(i, worldsize); return 0; } else if (pid < 0) { std::cerr << "Failed to create child process." << std::endl; return 1; } } for (int i = 0; i < worldsize; ++i) { wait(NULL); } std::cout << "The communication operator is successfully executed. Parent process exits." << std::endl; return 0; }

• 加速库多机通信，以两机四卡为例：

  对于多机通信，需要配置rankTableFile（当前只支持通信backend为“hccl”），rankTableFile是一个用于描述参与集合通信集群信息的json文件，包括通信Server，Device等的IP、ID信息，每张卡的rank号由用户传入的rankTableFile中rank_id标识，详细配置参考以下代码。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

  { "status":"completed", // ranktable可用标识，completed为可用 "version":"1.0", // ranktable模板版本信息,当前必须为"1.0" "server_count":"2", // AI Server数目，此例中为2机通信 "server_list": [ { "device":[ // AI Server中的Device列表 { "device_id":"0", // 处理器HDC通道号 "device_ip":"192.168.1.8", // 处理器真实网卡IP "rank_id":"0" // rank的标识，rankID从0开始 }, { "device_id":"1", "device_ip":"192.168.1.9", "rank_id":"1" } ], "server_id":"10.0.0.10" //AI Server标识，以点分十进制表示IP字符串 }, { "device": [ { "device_id": "0", "device_ip": "192.168.1.10", "rank_id": "2" }, { "device_id": "1", "device_ip": "192.168.1.11", "rank_id": "3" } ], "server_id": "10.0.0.11" } ] }

  加速库通信算子多机使用流程与单机使用流程基本相同，只需要在通信的机器中都开启多个进程，以下只给出与加速库单机二卡多进程通信计算并行用例中第一点和第二点的配置区别描述：

  1 2 3

  //1、根据配置的ranktable中的device_id设置每个进程对应的deviceId int deviceId = rank; aclError status = aclrtSetDevice(deviceId);

  1 2 3 4 5 6 7 8 9 10 11

  //2、用户传入配置好的rankTableFile文件路径,根据ranktable的rank_id来设置多机不同进程的rank号 atb::infer::LinearParallelParam param; param.rankTableFile = "/demo/ranktable.json"; param.rank = rank; param.commMode = atb::infer::CommMode::COMM_MULTI_PROCESS; param.backend = "hccl"; param.transWeight = true; param.type = atb::infer::LinearParallelParam::ParallelType::LINEAR_ALL_REDUCE; atb::Operation *op = nullptr; atb::Status st = atb::CreateOperation(param, &op);

调用通信算子的时候会指定在哪个通信域进行通信，commDomain是通信域名标识，如果构造AllReduceOperation参数使用默认值，则通信算子初始化通信域一次后，后续执行多次通信算子都只在同一个通信域进行通信；如果用户创建多个AllReduceOperation并给不同的AllReduceOperation参数配置不同的commDomain通信域名就可以让通信算子分别处于不同的通信域中并行通信，多个通信域后续也可以复用。

图6是在三个通信域内四卡做三次Allreduce的样例，输入为1，输出结果为32。