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