Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "aicpu_dmy_cal_allreduce.h"
12 : #include "common/aicpu_kfc_def.h"
13 :
14 : namespace {
15 : constexpr u64 ALL_REDUCE_THRESHOLD = 1024 * 1024; // allreduce确定性计算，小于该值选择AL算法，否则选择RLA算法
16 : constexpr u64 ALL_REDUCE_THRESHOLD_UT = 1024; // allreduce确定性计算ut测试值
17 : }
18 :
19 1 : HcclResult AicpuDmyCalAllreduce::RunAlgorithm(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
20 : u64 dataCount, HcclDataType dataType, u64 strideLen, AivAicpuOpParam * /* nextTask */)
21 : {
22 1 : CHK_PTR_NULL(ctx_);
23 1 : HcclResult ret = HCCL_SUCCESS;
24 : #ifdef RUN_TEST
25 : ctx_->windowSize = ALL_REDUCE_THRESHOLD_UT * ctx_->unitSize;
26 : if (ctx_->commLen < ALL_REDUCE_THRESHOLD_UT) {
27 : #else
28 1 : if (ctx_->commLen < ALL_REDUCE_THRESHOLD) {
29 : #endif
30 0 : ret = RunAllReduceAL(opType, sendBuffer, recvBuffer, dataCount, dataType);
31 : } else {
32 1 : const u64 tailCount = dataCount % (HCCL_COPY_ALIGN / ctx_->unitSize);
33 1 : const u64 tileCount = dataCount - tailCount;
34 1 : CHK_RET(RunAllReduceRLA(opType, sendBuffer, recvBuffer, tileCount, dataType));
35 1 : const u64 offset = tileCount * ctx_->unitSize;
36 1 : CHK_RET(RunAllReduceAL(opType, static_cast<u8 *>(sendBuffer) + offset,
37 : static_cast<u8 *>(recvBuffer) + offset, tailCount, dataType));
38 : }
39 1 : return ret;
40 : }
41 :
42 1 : HcclResult AicpuDmyCalAllreduce::RunAllReduceAL(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
43 : u64 dataCount, HcclDataType dataType)
44 : {
45 1 : if (dataCount == 0UL) {
46 1 : return HCCL_SUCCESS;
47 : }
48 0 : u32 unitSize = ctx_->unitSize;
49 0 : u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
50 0 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
51 :
52 0 : u64 windowSlices[AC_MAX_RANK_NUM] = {0};
53 0 : u64 curCount = dataCount;
54 0 : u64 curSize = curCount * unitSize;
55 :
56 0 : HCCL_INFO("RunDmyCalAllreduceAL:curInputPtr[%p], curOutputPtr[%p], curCount[%llu], curSize[%llu]",
57 : curInputPtr, curOutputPtr, curCount, curSize);
58 :
59 0 : for (u32 i = 0; i < ctx_->rankNum; i++) {
60 0 : windowSlices[i] = i * curSize;
61 : }
62 :
63 : // 1. 片内数据拷贝 snd->win
64 0 : TaskOrchestrator::SelfCpySnd2Win(curInputPtr, curSize, 0, windowSlices[ctx_->rankId], HCCL_REDUCE_RESERVED,
65 : dataType);
66 :
67 : // 2. 前同步
68 0 : TaskOrchestrator::DoPreSync();
69 :
70 : // 3. 跨片SDMA，send->对端win
71 0 : TaskOrchestrator::IpcCpySnd2Win(curInputPtr, curSize, static_cast<u64>(0), windowSlices[ctx_->rankId],
72 : HCCL_REDUCE_RESERVED, dataType);
73 :
74 : // 4. 后同步
75 0 : TaskOrchestrator::DoPostSync();
76 :
77 : // 5. 折半计算
78 0 : TaskOrchestrator::SelfLocalReduce(curSize, opType, dataType);
79 :
80 : // 6. 前同步
81 0 : TaskOrchestrator::DoPreSync();
82 :
83 : // 7. 片内数据拷贝 本端win->当前rcv buff
84 0 : TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, curSize, 0, 0, HCCL_REDUCE_RESERVED, dataType);
85 :
86 : // 8. 后同步
87 0 : TaskOrchestrator::DoPostSync();
88 :
89 0 : TaskOrchestrator::LaunchTasks();
90 0 : return HCCL_SUCCESS;
91 : }
92 :
93 1 : HcclResult AicpuDmyCalAllreduce::RunAllReduceRLA(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
94 : u64 dataCount, HcclDataType dataType) const
95 : {
96 1 : u64 windowSize = ctx_->windowSize;
97 1 : u32 unitSize = ctx_->unitSize;
98 1 : u64 maxCountPerLoop = windowSize / unitSize;
99 :
100 1 : u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
101 1 : u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
102 1 : u64 inputOffset = 0;
103 1 : u64 outputOffset = 0;
104 1 : u64 countLeft = dataCount;
105 1 : u64 windowSlices[AC_MAX_RANK_NUM] = {0};
106 :
107 1 : u32 loopIdx = 0;
108 1 : while (countLeft > 0) {
109 0 : curInputPtr += inputOffset;
110 0 : curOutputPtr += outputOffset;
111 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
112 0 : u64 curSize = curCount * unitSize;
113 0 : u64 sliceSize = curSize / ctx_->rankNum;
114 :
115 0 : HCCL_INFO("DmyCalAllreduceRLA:loop %u, curInputPtr[%p], curOutputPtr[%p], curCount[%llu], curSize[%llu]",
116 : loopIdx++, curInputPtr, curOutputPtr, curCount, curSize);
117 :
118 0 : for (u32 i = 0; i < ctx_->rankNum; i++) {
119 0 : windowSlices[i] = i * sliceSize;
120 : }
121 :
122 : // 1. 片内数据拷贝 snd->win
123 0 : TaskOrchestrator::SelfCpySnd2Win(curInputPtr, sliceSize, windowSlices[ctx_->rankId],
124 0 : windowSlices[ctx_->rankId], HCCL_REDUCE_RESERVED, dataType);
125 : // 2. 前同步
126 0 : TaskOrchestrator::DoPreSync();
127 :
128 : // 3. 跨片SDMA，send->其他window
129 0 : TaskOrchestrator::IpcCpySnd2Win(curInputPtr, sliceSize, windowSlices, windowSlices[ctx_->rankId],
130 : HCCL_REDUCE_RESERVED, dataType);
131 :
132 : // 4. 后同步
133 0 : TaskOrchestrator::DoPostSync();
134 :
135 : // 5. 折半计算
136 0 : TaskOrchestrator::SelfLocalReduce(sliceSize, opType, dataType);
137 :
138 : // 6. 片内数据拷贝 本端win->当前rcv buff
139 0 : TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, sliceSize, 0, windowSlices[ctx_->rankId], HCCL_REDUCE_RESERVED,
140 : dataType);
141 :
142 : // 7. 前同步
143 0 : TaskOrchestrator::DoPreSync();
144 :
145 : // 8. 跨片SDMA send->其他window
146 0 : TaskOrchestrator::IpcCpyWin2Rcv(curOutputPtr, sliceSize, nullptr, windowSlices, HCCL_REDUCE_RESERVED, dataType);
147 :
148 : // 9. 后同步
149 0 : TaskOrchestrator::DoPostSync();
150 :
151 0 : TaskOrchestrator::LaunchTasks();
152 :
153 0 : countLeft -= curCount;
154 0 : inputOffset = curSize;
155 0 : outputOffset = curSize;
156 : }
157 1 : return HCCL_SUCCESS;
158 : }
|
