Quick Start

AllReduce Operator

AllReduce performs reduction (sum, prod, max, and min) over the input data of all ranks in a communicator, and then sends the result to the output buffers of all ranks.

Note: Each rank has only one input.

Sample

You can click here to obtain the complete sample code. In this sample, a communicator is created based on the root rank information, and one AI server is managed in one process. Each NPU is managed by one thread. The main functions are as follows:

• Detects devices and queries the number of available devices by calling aclrtGetDeviceCount().
• Uses rank 0 as the root rank and calls HcclGetRootInfo() to generate the rootInfo identifier of the root rank.

• Initializes the communicator in each thread based on rootInfo by calling HcclCommInitRootInfo().
• Calls HcclAllReduce() to add the input data of all ranks in the communicator, sends the result to all ranks, and prints the result.

Compilation and Running

Run the following commands in the sample code directory:

1
2

make
make test

Result Analysis

The data of each rank is initialized to 0–7. After the AllReduce operation, the result of each rank is the sum of the data at the corresponding positions of all ranks (the data of eight ranks is added).

Found 8 NPU device(s) available
rankId: 0, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 1, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 2, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 3, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 4, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 5, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 6, output: [ 0 8 16 24 32 40 48 56 ]
rankId: 7, output: [ 0 8 16 24 32 40 48 56 ]

Code Analysis

1. Rank 0 is used as the root rank to generate rootInfo identification information, including the device IP address and device ID. The information needs to be broadcast to all ranks in the cluster for communicator initialization.

   1 2 3 4 5 6 7

   int rootRank = 0; ACLCHECK(aclrtSetDevice(rootRank)); // Generate root rank information. The same rootInfo is used for each thread. void *rootInfoBuf = nullptr; ACLCHECK(aclrtMallocHost(&rootInfoBuf, sizeof(HcclRootInfo))); HcclRootInfo *rootInfo = (HcclRootInfo *)rootInfoBuf; HCCLCHECK(HcclGetRootInfo(rootInfo));

2. Allocate memory and construct the input data.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

   // Set the device operated by the current thread. ACLCHECK(aclrtSetDevice(ctx->device)); // Allocate device memory for collective communication. size_t count = ctx->devCount; size_t mallocSize = count * sizeof(float); ACLCHECK(aclrtMalloc(&sendBuf, mallocSize, ACL_MEM_MALLOC_HUGE_ONLY)); ACLCHECK(aclrtMalloc(&recvBuf, mallocSize, ACL_MEM_MALLOC_HUGE_ONLY)); // Allocate host memory for storing input data and initialize the content to 0–7. void *hostBuf = nullptr; ACLCHECK(aclrtMallocHost(&hostBuf, mallocSize)); float *tmpHostBuff = static_cast<float *>(hostBuf); for (uint32_t i = 0; i < count; ++i) { tmpHostBuff[i] = static_cast<float>(i); } // Copy the input data from the host to the device. ACLCHECK(aclrtMemcpy(sendBuf, mallocSize, hostBuf, mallocSize, ACL_MEMCPY_HOST_TO_DEVICE));

3. Initialize the communicator.

   1 2	HcclComm hcclComm; HCCLCHECK(HcclCommInitRootInfo(ctx->devCount, ctx->rootInfo, ctx->device, &hcclComm));

4. Execute the AllReduce collective communication operator.

   1 2 3 4 5 6 7 8

   // Create a task flow. aclrtStream stream; ACLCHECK(aclrtCreateStream(&stream)); // Execute AllReduce to sum up sendBuf of all ranks in the communicator and send the result to recvBuf of all ranks. HCCLCHECK(HcclAllReduce(sendBuf, recvBuf, count, HCCL_DATA_TYPE_FP32, HCCL_REDUCE_SUM, hcclComm, stream)); // Wait until the collective communication task in the task flow is complete. ACLCHECK(aclrtSynchronizeStream(stream));

5. Free resources.

   1 2 3 4 5 6

   ACLCHECK(aclrtFree(sendBuf)); // Free the device memory. ACLCHECK(aclrtFree(recvBuf)); // Free the device memory. ACLCHECK(aclrtFreeHost(hostBuf)); // Free the host memory. ACLCHECK(aclrtDestroyStream(stream)); // Destroy the task flow. HCCLCHECK(HcclCommDestroy(hcclComm)); // Destroy the communicator. ACLCHECK(aclFinalize()); // Deinitialize the device.