Documentation
Download

Introduction to ATB Communication Operators

The ATB communication operator integrates the HCCL (Huawei Collective Communication Library) and the small-packet communication mechanism as the communication backend, and provides single-server multi-device and multi-server multi-device collective communication primitives. For details about HCCL-related documents, see API in HCCL documentation. The small-packet communication mechanism has great advantages in small-packet communication (the communication data volume is less than 512 KB), development convenience, delivery overhead, and synchronization overhead.

The ATB supports the following communication operators:

  • AllReduceOperation: Computes the data on multiple communication cards, including addition, multiplication, minimum, and maximum calculations, and sends the result to each card.
    Figure 1 AllReduce
  • BroadcastOperation: broadcasts the data on the primary communication card to each of the other cards.
    Figure 2 Broadcast
  • AllGatherOperation: Aggregates data of a plurality of communications cards in the first dimension according to rank sequences and sends data to each card.
    Figure 3 AllGather
  • LinearParallelOperation: A typical tensor parallelism policy inserts dependent communication tasks into a computing flow, so that the computing communication tasks must be executed in serial mode during scheduling. As a result, computing resources are idle when the communication tasks are executed (so it is called "unmaskable communication"), linearParallelOperation integrates the fused operator. Through proper task scheduling and parallel execution, most communication time can be covered by computing tasks, implementing efficient parallel processing of computing and communication. This operator is a combination of LinearOperation and communication operator Operation. Three communication combinations are supported: linear+AllReduce, linear+reduce_scatter, and AllGather+linear.
    Figure 4 LinearParallel linear+AllReduce

Process of Using the ATB Communication Operator

The ATB communication operator implements multi-device and multi-device communication in two steps: initializing the communicator (see ) and executing the communication operator.

A communicator refers to the scope of communication. For example, if you want to perform AllReduce once, you need to perform AllReduce between cards. The communicator can be initialized in either of the following ways:

  • The ATB initializes the communicator for the user. Each device obtains its own communicator handle to execute the operator.
  • You can configure HcclComm in AllReduceOperation to transfer the communication handle of the corresponding communication card. After the communication handle is transferred, the ATB is not initialized.

Generally, the ATB initializes the communicator for the user. As shown in Figure 5, there are two methods:

  • Creating a communicator based on the rankRoot node information applies only to single-node systems.
  • Creating a communicator based on rankTableFile applies to both single-node and multi-node communication.

You can configure different Operation Params to create Operations and initialize the communicator in different ways. After the communicator is initialized successfully, the corresponding communication operator API is executed to implement communication.

Figure 5 Creating a communicator
  • Single-server two-card multi-process communication and computing parallel test case: The running mode of the following demos is the same as that in Operator Usage Guide (ATB C++ APIs).
      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. Set the device ID corresponding to each process. Subsequent communication operators are executed on the device.
    int deviceId = rank;
    aclError status = aclrtSetDevice(deviceId);

    //2. LinearParallelOperation parameter configuration. For multi-card and multi-process communication, the rank number of the operator parameter in each process needs to be configured. The rank number ranges from 0 to 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. Construct the input and output.
    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. The operator preparation and execution process is the same as that of a single-operator. The difference is that the communication operator is executed only after the communicator is successfully initialized.
    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. Release resources. The following cases are the same.
    status = aclrtDestroyStream(stream); // Destroy the stream.
    status = aclrtFree(workspace);       // Destroy the workspace.
    st = atb::DestroyOperation(op);      // Destroy the op object.
    st = atb::DestroyContext(context);   // Destroy the context.
    // The following code is an example of releasing a tensor. In practice, all tensors in the VariantPack need to be released.
    status = aclrtFree(tensor.deviceData);
    tensor.deviceData = nullptr;
    tensor.dataSize = 0;
    aclFinalize();
}

int main(int argc, const char *argv[])
{
    // Multi-process two-card communication
    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;
}
  • The following uses two servers with four cards as an example to describe how to implement multi-device communication.

    For multi-machine communication, you need to configure rankTableFile (currently only supports communication backend as "hccl"). rankTableFile is a JSON file used to describe the cluster information of the collective communication, including the IP and ID information of the communication server and device. The rank number of each card is identified by the rank_id in the rankTableFile provided by the user. For detailed configuration, refer to the following code.

     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 availability flag. The value completed indicates that the ranktable is available.
"version":"1.0",        // Ranktable template version. Must be 1.0.
"server_count":"2",     // AI Server count. In this example, two AI servers are used for communication.
"server_list":
[
   {
        "device":[      // List of devices on the server
                       {
                        "device_id":"0",           // HDC channel ID of the processor
                        "device_ip":"192.168.1.8", // Actual NIC IP address of the processor
                        "rank_id":"0"              // Rank ID, indexed starting at 0.
                        },
                        {
                         "device_id":"1",
                         "device_ip":"192.168.1.9", 
                         "rank_id":"1"
                         }
                  ],
         "server_id":"10.0.0.10"                    // AI server ID, which is an IP address in dotted decimal notation
    },
	{
          "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"
      }
]
}

    The process of using the ATB communication operator on multiple servers is similar to that of using the ATB communication operator on a single server. You only need to start multiple processes on the communication server. The following describes only the configuration differences between the first and second points in the parallel test case of single-server dual-card multi-process communication computing.

    1
2
3
    //1. Set the device ID corresponding to each process based on the device ID in the configured ranktable.
int deviceId = rank;
aclError status = aclrtSetDevice(deviceId);

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
    //2. The user inputs the configured rankTableFile path and sets the rank IDs of different processes on multiple servers based on the rank_id of the ranktable.
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);

When the communication operator is called, the communicator is specified for communication. commDomain is the communicator name identifier. If the AllReduceOperation parameter is set to the default value, after the communication operator initializes the communicator once, the communication operator is executed for multiple times only in the same communicator for communication. If you create multiple AllReduceOperations and configure different commDomain communicator names for different AllReduceOperation parameters, the communication operators can be in different communicators for parallel communication. Multiple communicators can be reused later.

Figure 6 is an example of performing Allreduce three times on four devices in three communicators. The input is 1, and the output is 32.

Figure 6 Multi-communicator AllReduce
  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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2020-2020. All rights reserved.
 */
#include <acl/acl.h>
#include <atb/atb_infer.h>
#include <iostream>
#include <unistd.h>
#include <sys/wait.h>

void ExecuteImpl(atb::Operation *op, atb::VariantPack variantPack, atb::Context *context)
{
    uint64_t workspaceSize = 0;
    atb::Status st = op->Setup(variantPack, workspaceSize, context);
    void *workspace = nullptr;
    if (workspaceSize > 0) {
        aclError status = aclrtMalloc(&workspace, workspaceSize, ACL_MEM_MALLOC_HUGE_FIRST);
    }
    st = op->Execute(variantPack, (uint8_t *)workspace, workspaceSize, context);
    if (workspace) {
         st = aclrtFree(workspace);       // Destroy the workspace.
    }
}

void AllReduceSample(int rank, int worldsize)
{
    int ret = aclInit(nullptr);
    int deviceId = rank;
    aclError status = aclrtSetDevice(deviceId);

    atb::Context *context = nullptr;
    atb::Status st = atb::CreateContext(&context);
    aclrtStream stream = nullptr;
    status = aclrtCreateStream(&stream);
    context->SetExecuteStream(stream);

    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] = 5;
    input.dataSize = atb::Utils::GetTensorSize(input);
    status = aclrtMalloc(&input.deviceData, input.dataSize, ACL_MEM_MALLOC_HUGE_FIRST);

    atb::SVector<atb::Tensor> outputlist;
    for (int i = 0; i < 3; i++) {
        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);
        outputlist.push_back(output);
    }

    // 1. Ranks 0, 1, 2, and 3 are in communicator atb1, and rankSize is 4.
    atb::infer::AllReduceParam param;
    param.rank = rank;
    param.rankRoot = 0;
    param.rankSize = worldsize;
    param.backend = "hccl";
    param.commDomain = "atb1";
    atb::Operation *op1 = nullptr;
    st = atb::CreateOperation(param, &op1);

    atb::VariantPack variantPack1;
    variantPack1.inTensors = { input };
    variantPack1.outTensors = { outputlist[0] };

    ExecuteImpl(op1, variantPack1, context);
    std::cout << "rank: " << rank << " executed END." << std::endl;

    atb::VariantPack variantPack2;
    variantPack2.inTensors = { outputlist[0] };
    variantPack2.outTensors = { outputlist[1] };
    // 2. Ranks 0 and 1 are in communicator atb2, and rankSize is 2.
    if (rank == 0 || rank == 1) {
        param.rank = rank;
        param.rankSize = 2;
        param.commDomain = "atb2";
        atb::Operation *op2 = nullptr;
        atb::Status st = atb::CreateOperation(param, &op2);
        ExecuteImpl(op2, variantPack2, context);
        std::cout << "rank: " << rank << " executed END." << std::endl;
    }  // 3. Ranks 2 and 3 are in communicator atb3, and rankSize is 2.
    else if (rank == 2 || rank == 3) {
        param.rank = rank - 2;
        param.rankSize = 2;
        param.commDomain = "atb3";
        atb::Operation *op2 = nullptr;
        atb::Status st = atb::CreateOperation(param, &op2);
        ExecuteImpl(op2, variantPack2, context);
        std::cout << "rank: " << rank << " executed END." << std::endl;
    }
    // 4. Ranks 0, 1, 2, and 3 are in communicator atb1.
    param.rank = rank;
    param.rankSize = 4;
    param.commDomain = "atb1";

    atb::VariantPack variantPack3;
    variantPack3.inTensors = { outputlist[1] };
    variantPack3.outTensors = { outputlist[2] };
    ExecuteImpl(op1, variantPack3, context);
    std::cout << "rank: " << rank << " executed END." << std::endl;
    // 5. Release resources.
    status = aclrtDestroyStream(stream); // Destroy the stream.
    st = atb::DestroyOperation(op1);      // Destroy the op object.
    st = atb::DestroyOperation(op2);      // Destroy the op object.
    st = atb::DestroyContext(context);   // Destroy the context.
    aclFinalize();
}

int main(int argc, const char *argv[])
{
    // Multi-process two-card communication
    const int worldsize = 4;
    for (int i = 0; i < worldsize; ++i) {
        pid_t pid = fork();
        if (pid == 0) {
            AllReduceSample(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;
}
Parent topic: Distributed Inference
This site uses cookies to improve your browsing experience. By continuing to browse this site, you agree to our use of cookies.View Details