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 "coll_scatter_executor.h"
12 : #include "device_capacity.h"
13 :
14 : namespace hccl {
15 1 : CollScatterExecutor::CollScatterExecutor(const HcclDispatcher dispatcher,
16 1 : std::unique_ptr<TopoMatcher> &topoMatcher)
17 1 : : CollCommExecutor(dispatcher, topoMatcher)
18 : {
19 4 : }
20 :
21 4 : HcclResult CollScatterExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
22 : {
23 4 : TransportMemType inputType = TransportMemType::RESERVED;
24 4 : TransportMemType outputType = TransportMemType::RESERVED;
25 4 : CHK_RET(CalcTransportMemType(inputType, outputType));
26 4 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
27 4 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
28 4 : return HCCL_SUCCESS;
29 : }
30 :
31 4 : HcclResult CollScatterExecutor::CalcTransportMemType(TransportMemType &inputType, TransportMemType &outputType)
32 : {
33 4 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
34 4 : inputType = TransportMemType::CCL_INPUT;
35 4 : outputType = TransportMemType::CCL_INPUT;
36 : } else {
37 0 : inputType = TransportMemType::PARAM_INPUT;
38 0 : outputType = TransportMemType::PARAM_INPUT;
39 : }
40 4 : return HCCL_SUCCESS;
41 : }
42 :
43 0 : bool CollScatterExecutor::IsHugeData(u64 curSize)
44 : {
45 0 : bool hugeData = curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE ||
46 : curSize > SDMA_SEND_MAX_SIZE;
47 0 : return hugeData;
48 : }
49 :
50 0 : HcclResult CollScatterExecutor::RunLoop(OpParam ¶m, AlgResourceResponse &algRes)
51 : {
52 0 : auto dataType = param.DataDes.dataType;
53 0 : u32 unitSize = SIZE_TABLE[dataType];
54 0 : RankId root = param.root;
55 :
56 0 : auto totalRecvCount = param.DataDes.count;
57 :
58 0 : u8 *curUserInputPtr = static_cast<u8 *>(param.inputPtr);
59 0 : u8 *curUserOutputPtr = static_cast<u8 *>(param.outputPtr);
60 0 : if (topoAttr_.userRank == root) {
61 0 : CHK_PTR_NULL(curUserInputPtr);
62 : }
63 0 : CHK_PTR_NULL(curUserOutputPtr);
64 :
65 0 : auto inCCLbuffer = algRes.cclInputMem;
66 0 : auto outCCLbuffer = algRes.cclOutputMem;
67 :
68 : // 中转内存单次最多能够接受的output count
69 0 : u64 maxCountPerLoop = inCCLbuffer.size() / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN
70 0 : * HCCL_MIN_SLICE_ALIGN / unitSize;
71 0 : HCCL_DEBUG("[CollScatterExecutor][RunLoop]tag[%s], userRankSize is [%u], root is [%u], "
72 : "maxCountPerLoop is [%llu], totalRecvCount is [%llu]",
73 : tag_.c_str(), topoAttr_.userRankSize, root, maxCountPerLoop, totalRecvCount);
74 :
75 0 : for (u64 countLeft = totalRecvCount, curRecvCount = 0, inputOffset = 0, outputOffset = 0;
76 0 : countLeft > 0; countLeft -= curRecvCount) {
77 0 : curUserInputPtr += inputOffset;
78 0 : curUserOutputPtr += outputOffset;
79 :
80 : // 判断剩余数据量对应的input size是否大于中转input size
81 0 : curRecvCount =
82 0 : ((countLeft * unitSize * topoAttr_.userRankSize) > inCCLbuffer.size()) ? maxCountPerLoop : countLeft;
83 0 : CHK_PRT_RET((curRecvCount == 0), HCCL_ERROR("[RunLoop][Scatter]In OP_BASE curRecvCount is zero"), HCCL_E_PARA);
84 0 : u64 curRecvSize = curRecvCount * unitSize; // 单位：字节
85 0 : u64 curSendSize = topoAttr_.userRankSize * curRecvSize; // 单位：字节
86 :
87 0 : DeviceMem curCCLInputMem(inCCLbuffer.ptr(), curSendSize);
88 0 : DeviceMem curCCLOutputMem(outCCLbuffer.ptr(), curRecvSize);
89 :
90 0 : ExecMem execMem;
91 0 : execMem.count = curRecvCount;
92 0 : execMem.inputMem = curCCLInputMem;
93 0 : execMem.outputMem = curCCLOutputMem;
94 0 : execMem.scratchMem = algRes.scratchMem;
95 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
96 0 : execMem.inputPtr = curUserInputPtr;
97 0 : execMem.outputPtr = curUserOutputPtr;
98 :
99 0 : HCCL_DEBUG("[RunLoop][Scatter] ScatterLoop: inputOffset[%llu], outputOffset[%llu], "
100 : "curUserInputPtr[%p], curUserOutputPtr[%p], curRecvCount[%llu], curRecvSize[%llu], "
101 : "curSendSize[%llu], inCCLbuffer.ptr[%p], outCCLbuffer.ptr[%p]",
102 : inputOffset, outputOffset, curUserInputPtr, curUserOutputPtr, curRecvCount, curRecvSize,
103 : curSendSize, inCCLbuffer.ptr(), outCCLbuffer.ptr());
104 :
105 0 : CHK_RET(RunLoopInner(param, execMem, algRes));
106 :
107 0 : inputOffset = curRecvSize;
108 0 : outputOffset = curRecvSize;
109 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
110 0 : }
111 0 : return HCCL_SUCCESS;
112 0 : }
113 :
114 0 : HcclResult CollScatterExecutor::RunLoopInner(OpParam ¶m, ExecMem &execMem, AlgResourceResponse &algRes)
115 : {
116 0 : auto dataType = param.DataDes.dataType;
117 0 : u32 unitSize = SIZE_TABLE[dataType];
118 0 : RankId root = param.root;
119 :
120 0 : auto totalRecvCount = param.DataDes.count;
121 0 : u64 totalRecvSize = totalRecvCount * unitSize;
122 :
123 0 : u64 recvSize = execMem.outputMem.size();
124 :
125 0 : auto meta = HcclOpMetaInfo::GetOneForScatter(root, IsHugeData(execMem.outputMem.size()));
126 0 : CHK_RET(InitTask(dispatcher_, param.stream, meta.isEnableCache, meta.GetCacheKey()));
127 :
128 0 : DeviceMem dstMem;
129 0 : DeviceMem srcMem;
130 0 : if (topoAttr_.userRank == root) {
131 : // 本rank为root节点，非root节点不需要拷贝到中转内存
132 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
133 : // 拷贝input上每个slice的数据到中转内存，源端每个slice的size固定为totalRecvSize
134 0 : srcMem = DeviceMem::create((u8*)execMem.inputPtr + totalRecvSize * i, recvSize);
135 0 : dstMem = algRes.cclInputMem.range(recvSize * i, recvSize);
136 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
137 : }
138 : }
139 :
140 0 : if (recvSize % HCCL_MIN_SLICE_ALIGN != 0) {
141 : // 不支持内存不对齐的轮次
142 0 : DMAReduceFlag_ = false;
143 : }
144 :
145 : /* 入参的正确性由HCCL确保 */
146 0 : HcclResult ret = KernelRun(param, execMem);
147 :
148 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
149 : HCCL_ERROR("[CollScatterExecutor][RunLoop]errNo[0x%016llx] OP_BASE hcclComm scatter error, tag[%s], "
150 : "input_ptr[%p], output_ptr[%p], recvSize[%llu], data_type[%d], root[%u]",
151 : HCCL_ERROR_CODE(ret), tag_.c_str(), algRes.cclInputMem.ptr(), algRes.cclOutputMem.ptr(),
152 : recvSize, dataType, root),
153 : ret);
154 :
155 : // 将 CCLOut 上的数据搬运到 userOut
156 0 : if (!DMAReduceFlag_) {
157 0 : srcMem = algRes.cclOutputMem.range(0, recvSize);
158 0 : dstMem = DeviceMem::create(execMem.outputPtr, recvSize);
159 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
160 : }
161 0 : return HCCL_SUCCESS;
162 0 : }
163 :
164 0 : HcclResult CollScatterExecutor::PrepareDataSlice(u64 dataCount, u32 unitSize, u32 sliceNum,
165 : std::vector<Slice> &dataSlice)
166 : {
167 0 : CHK_PRT_RET((sliceNum == 0), HCCL_ERROR("[CollScatterExecutor][PrepareDataSlice]sliceNum is zero."), HCCL_E_PARA);
168 :
169 0 : dataSlice.resize(sliceNum);
170 0 : u64 sliceSize = dataCount * unitSize;
171 0 : for (u32 i = 0; i < sliceNum; i++) {
172 0 : dataSlice[i].size = sliceSize;
173 0 : dataSlice[i].offset = (i * sliceSize);
174 : }
175 0 : return HCCL_SUCCESS;
176 : }
177 :
178 0 : HcclResult CollScatterExecutor::ReorderSlice(std::vector<Slice> &dataSlice, std::vector<u32> &order)
179 : {
180 0 : CHK_PRT_RET((dataSlice.size() != order.size()),
181 : HCCL_ERROR("[ReorderSlice] data slice size [%zu], not equal to order size [%zu]",
182 : dataSlice.size(), order.size()), HCCL_E_INTERNAL);
183 0 : std::vector<Slice> tempDataSegsSlice(dataSlice.size());
184 0 : for (size_t i = 0; i < dataSlice.size(); i++) {
185 0 : CHK_PRT_RET(order[i] >= dataSlice.size(),
186 : HCCL_ERROR("[ReorderSlice] order value [%u] >= dataSlice size [%zu]", order[i], dataSlice.size()),
187 : HCCL_E_INTERNAL);
188 0 : tempDataSegsSlice[i] = dataSlice[order[i]];
189 : }
190 0 : dataSlice = tempDataSegsSlice;
191 0 : return HCCL_SUCCESS;
192 0 : }
193 :
194 0 : HcclResult CollScatterExecutor::KernelRunLevel1(DeviceMem& inputMem, u64 count, HcclDataType dataType,
195 : u32 &commIndex, u32 root, u32 &subRoot, CommPlane commLevel, Stream& stream)
196 : {
197 0 : CHK_RET(CheckCommSize(commLevel, commIndex + 1));
198 0 : SubCommInfo subCommInfo = GetSubCommInfo(commLevel, commIndex);
199 :
200 0 : u32 subCommSize = subCommInfo.localRankSize;
201 :
202 0 : if (subCommSize <= 1 || subRoot != topoAttr_.userRank) {
203 0 : HCCL_INFO("[Scatter][KernelRunLevel1]: no need to run intra-server, subCommSize[%u], subRoot[%u]," \
204 : "userRank[%u]", subCommSize, subRoot, topoAttr_.userRank);
205 0 : return HCCL_SUCCESS;
206 : }
207 :
208 0 : HCCL_INFO("[Scatter][KernelRunLevel1]: start to run intra-server, subCommSize[%u], subRoot[%u]," \
209 : "userRank[%u]", subCommSize, subRoot, topoAttr_.userRank);
210 :
211 0 : u32 rootRankLevel1 = 0;
212 0 : CHK_RET(GetRankByUserRank(commLevel, commIndex, root, rootRankLevel1));
213 :
214 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
215 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
216 : // server间NB算法走NB
217 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_SCATTER_NB, dispatcher_);
218 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_SCATTER_NB in COMM_LEVEL1", __func__);
219 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
220 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_SCATTER_NHR, dispatcher_);
221 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_SCATTER_NHR in COMM_LEVEL1", __func__);
222 : } else {
223 0 : level1TempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
224 0 : TemplateType::TEMPLATE_SCATTER_RING, dispatcher_);
225 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_SCATTER_RING in COMM_LEVEL1", __func__);
226 : }
227 :
228 0 : CHK_SMART_PTR_NULL(level1TempAlg);
229 0 : CHK_RET(level1TempAlg->Prepare(inputMem, inputMem, inputMem, count * topoAttr_.userRankSize,
230 : dataType, stream, HCCL_REDUCE_RESERVED, rootRankLevel1, std::vector<Slice>(0))); // count是output的数据个数
231 0 : CHK_RET(level1TempAlg->RegisterProfiler(
232 : (subCommSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + subCommInfo.localRank,
233 : PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, stream));
234 :
235 0 : CHK_RET(RunTemplate(level1TempAlg, subCommInfo));
236 :
237 0 : return HCCL_SUCCESS;
238 0 : }
239 :
240 0 : HcclResult CollScatterExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
241 : {
242 0 : HcclUs startut = TIME_NOW();
243 0 : tag_ = param.tag;
244 0 : algResResp_ = &algRes;
245 0 : HcclResult ret = HCCL_SUCCESS;
246 0 : bool needLaunchAtTheEnd = true; // 是否需要在Orchestrate()结束时launch任务
247 : // 图模式和单卡场景下不需要Loop
248 0 : ExecMem execMem;
249 0 : execMem.count = param.DataDes.count;
250 0 : execMem.inputPtr = param.inputPtr;
251 0 : execMem.outputPtr = param.outputPtr;
252 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
253 0 : execMem.inputMem = algRes.paramInputMem;
254 0 : execMem.outputMem = algRes.paramOutputMem;
255 0 : execMem.scratchMem = algRes.scratchMem;
256 0 : ret = KernelRun(param, execMem);
257 0 : } else if (topoAttr_.userRankSize == 1) {
258 0 : ret = KernelRun(param, execMem);
259 0 : needLaunchAtTheEnd = false;
260 : } else {
261 0 : ret = RunLoop(param, algRes);
262 0 : needLaunchAtTheEnd = false;
263 : }
264 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
265 : HCCL_ERROR("[CollScatterExecutor][Orchestrate]errNo[0x%016llx]Scatter executor kernel run failed",
266 : HCCL_ERROR_CODE(ret)), ret);
267 :
268 : // Enforce task launch at the end of Orchestrate
269 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
270 0 : if (needLaunchAtTheEnd) {
271 0 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
272 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
273 : }
274 :
275 0 : HCCL_INFO("tag[%s] Scatter executor orchestrate success, take time [%lld]us.",
276 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
277 0 : return HCCL_SUCCESS;
278 0 : }
279 :
280 : }
|
