Line data Source code
1 : /**
2 : * Copyright (c) 2026 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_broadcast_midcount_for_910_93_executor.h"
12 :
13 : namespace hccl {
14 0 : CollBroadcastMidCountFor91093Executor::CollBroadcastMidCountFor91093Executor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 0 : : CollBroadcastExecutor(dispatcher, topoMatcher)
17 : {
18 0 : desc_.level1SupportedAlgos = {
19 : AlgTypeLevel1::ALG_LEVEL1_NHR,
20 0 : };
21 0 : desc_.level2SupportedAlgos = {
22 : AlgTypeLevel2::ALG_LEVEL2_NHR,
23 0 : };
24 0 : }
25 0 : HcclResult CollBroadcastMidCountFor91093Executor::CalcStreamNum(u32& streamNum)
26 : {
27 0 : streamNum = 0;
28 0 : HCCL_INFO("[CollBroadcastCommExecutor][CalcStreamNum]tag[%s] streamNum_ is [%u]", tag_.c_str(), streamNum);
29 0 : return HCCL_SUCCESS;
30 : }
31 :
32 0 : HcclResult CollBroadcastMidCountFor91093Executor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
33 : {
34 0 : TransportMemType inputType = TransportMemType::RESERVED;
35 0 : TransportMemType outputType = TransportMemType::RESERVED;
36 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
37 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
38 0 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
39 0 : return HCCL_SUCCESS;
40 : }
41 :
42 0 : HcclResult CollBroadcastMidCountFor91093Executor::CalcTransportMemType(
43 : TransportMemType& inputType, TransportMemType& outputType) const
44 : {
45 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
46 0 : inputType = TransportMemType::CCL_INPUT;
47 0 : outputType = TransportMemType::CCL_OUTPUT;
48 : } else {
49 0 : HCCL_ERROR("BroadcastMidCountFor91093Executor do not support offload mode");
50 0 : return HCCL_E_UNAVAIL;
51 : }
52 0 : HCCL_INFO(
53 : "[CollBroadcastMidCountFor91093Executor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
54 : tag_.c_str(), inputType, outputType);
55 0 : return HCCL_SUCCESS;
56 : }
57 :
58 0 : HcclResult CollBroadcastMidCountFor91093Executor::CalcLevel1CommInfo(
59 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
60 : {
61 0 : CommParaInfo commParaCombineL1(COMM_COMBINE_L1, CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING);
62 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaCombineL1, opTransport[COMM_COMBINE_L1], inputType, outputType));
63 0 : return HCCL_SUCCESS;
64 0 : }
65 :
66 0 : HcclResult CollBroadcastMidCountFor91093Executor::CalcLevel2CommInfo(
67 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
68 : {
69 0 : CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING);
70 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
71 0 : return HCCL_SUCCESS;
72 0 : }
73 :
74 0 : u64 CollBroadcastMidCountFor91093Executor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
75 : {
76 0 : u64 maxCountPerLoop = cclBuffSize / HCCL_MIN_SLICE_ALIGN_910_93 * HCCL_MIN_SLICE_ALIGN_910_93 / unitSize;
77 0 : if (maxCountPerLoop == 0) {
78 0 : HCCL_ERROR("[CollBroadcastMidCountFor91093Executor][CalcLoopMaxCount] cclbuffer size is too small");
79 : }
80 0 : return maxCountPerLoop;
81 : }
82 :
83 0 : HcclResult CollBroadcastMidCountFor91093Executor::RunLevel2ByNHR(
84 : const OpParam& param, ExecMem& execMem, SubCommInfo& level1CommInfo, SubCommInfo& level2CommInfo) const
85 : {
86 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[MidCountBroadcast][RunLevel2ByNHR] userRank[%u] starts.", topoAttr_.userRank);
87 :
88 0 : u32 unitSize = 0;
89 0 : const HcclDataType dataType = param.GetDataType();
90 0 : CHK_RET(SalGetDataTypeSize(dataType, unitSize));
91 :
92 : std::unique_ptr<AlgTemplateBase> tempAlg
93 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_BROADCAST_NHR_ONESHOT, dispatcher_);
94 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_BROADCAST_NHR_ONESHOT in COMM_LEVEL2", __func__);
95 0 : CHK_SMART_PTR_NULL(tempAlg);
96 :
97 : // 获取root
98 0 : u32 rootRank = param.root / level1CommInfo.localRankSize;
99 0 : CHK_RET(tempAlg->Prepare(
100 : execMem.inputMem, execMem.inputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
101 : HCCL_REDUCE_RESERVED, rootRank));
102 :
103 0 : CHK_RET(RunTemplate(tempAlg, level2CommInfo));
104 :
105 0 : HCCL_INFO("MidCountBroadcast run success in level2");
106 0 : return HCCL_SUCCESS;
107 0 : }
108 :
109 0 : HcclResult CollBroadcastMidCountFor91093Executor::RunLevel1ByNHR(
110 : const OpParam& param, ExecMem& execMem, SubCommInfo& level1CommInfo, [[maybe_unused]] SubCommInfo& level2CommInfo)
111 : {
112 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[MidCountBroadcast][RunLevel1ByNHR] userRank[%u] starts.", topoAttr_.userRank);
113 0 : u32 unitSize = 0;
114 0 : const HcclDataType dataType = param.GetDataType();
115 0 : CHK_RET(SalGetDataTypeSize(dataType, unitSize));
116 :
117 : std::unique_ptr<AlgTemplateBase> tempAlg
118 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_BROADCAST_NHR_ONESHOT, dispatcher_);
119 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_BROADCAST_NHR_ONESHOT in COMM_COMBINE_L1", __func__);
120 0 : CHK_SMART_PTR_NULL(tempAlg);
121 :
122 : // 获取root
123 0 : u32 rootRank = 0;
124 0 : CHK_RET(GetRankByUserRank(COMM_COMBINE_L1, COMM_INDEX_0, param.root, rootRank));
125 0 : CHK_RET(tempAlg->Prepare(
126 : execMem.inputMem, execMem.inputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
127 : HCCL_REDUCE_RESERVED, rootRank));
128 :
129 0 : CHK_RET(RunTemplate(tempAlg, level1CommInfo));
130 :
131 0 : HCCL_INFO("MidCountBroadcast run success in level1");
132 0 : return HCCL_SUCCESS;
133 0 : }
134 :
135 0 : HcclResult CollBroadcastMidCountFor91093Executor::KernelRun(const OpParam& param, ExecMem& execMem)
136 : {
137 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] The MidCountFor91093Executor starts", __func__);
138 :
139 0 : SubCommInfo level1CommInfo;
140 0 : SubCommInfo level2CommInfo;
141 0 : CHK_RET(CheckCommSize(COMM_COMBINE_L1, COMM_INDEX_0 + 1));
142 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
143 0 : level1CommInfo = GetSubCommInfo(COMM_COMBINE_L1, COMM_INDEX_0);
144 0 : level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
145 :
146 0 : u32 unitSize = 0;
147 0 : const HcclDataType dataType = param.GetDataType();
148 0 : CHK_RET(SalGetDataTypeSize(dataType, unitSize));
149 :
150 0 : u32 rootSideIndex = param.root / level1CommInfo.localRankSize;
151 0 : u32 rankSideIndex = topoAttr_.userRank / level1CommInfo.localRankSize;
152 0 : if (rootSideIndex == rankSideIndex) {
153 0 : CHK_RET(RunLevel1ByNHR(param, execMem, level1CommInfo, level2CommInfo));
154 : }
155 0 : CHK_RET(RunLevel2ByNHR(param, execMem, level1CommInfo, level2CommInfo));
156 :
157 0 : HCCL_INFO("MidCountBroadcast run success.");
158 0 : return HCCL_SUCCESS;
159 0 : }
160 :
161 0 : HcclResult CollBroadcastMidCountFor91093Executor::RunLoopInner(OpParam& param, ExecMem& execMem)
162 : {
163 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
164 0 : u64 totalSize = unitSize * param.DataDes.count;
165 0 : bool isRootRank = param.root == topoAttr_.realUserRank ? true : false;
166 0 : u64 curSize = execMem.count * unitSize; // 单位：字节
167 0 : auto inCCLbufferSize = execMem.inputMem.size();
168 :
169 0 : HCCL_DEBUG(
170 : "[CollBroadcastMidCountFor91093Executor][RunLoopInner]inputMem[%p], outputMem[%p]"
171 : "intputPtr[%p], curCount[%llu], curSize[%llu]",
172 : execMem.inputMem.ptr(), execMem.outputMem.ptr(), execMem.inputPtr, execMem.count, curSize);
173 0 : CHK_PRT_RET(
174 : (execMem.count == 0),
175 : HCCL_ERROR("[CollBroadcastMidCountFor91093Executor][RunLoop]In OP_BASE curCount is zero."), HCCL_E_PARA);
176 :
177 0 : bool hugeData
178 0 : = (inCCLbufferSize / topoAttr_.deviceNumPerAggregation > RDMA_SEND_MAX_SIZE) || (curSize > SDMA_SEND_MAX_SIZE);
179 0 : bool isSmallData = IsBroadcastSmallData(curSize, totalSize);
180 0 : u64 sliceNum = 0;
181 0 : CHK_RET(GetSliceNum(curSize, isSmallData, sliceNum));
182 0 : CopyPattern copy = DMAReduceFlag_ ? CopyPattern::ZCOPY : CopyPattern::BCOPY;
183 0 : auto meta = HcclOpMetaInfo::GetOneForBroadcast(isRootRank, param.root, hugeData, isSmallData, sliceNum, copy);
184 0 : CHK_RET(InitTask(dispatcher_, param.stream, meta.isEnableCache, meta.GetCacheKey()));
185 :
186 : // 执行
187 : HcclResult ret;
188 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
189 0 : DeviceMem inCommMem = execMem.inputMem.range(0, curSize);
190 0 : DeviceMem inMem(execMem.inputPtr, curSize);
191 0 : if (topoAttr_.userRank == param.root) {
192 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, param.stream));
193 : }
194 :
195 0 : ret = KernelRun(param, execMem);
196 0 : if (topoAttr_.realUserRank != param.root) {
197 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inMem, inCommMem, param.stream));
198 : }
199 :
200 0 : CHK_PRT_RET(
201 : ret != HCCL_SUCCESS,
202 : HCCL_ERROR(
203 : "[CollBroadcastMidCountFor91093Executor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], "
204 : "inputMem ptr[%p], count[%llu], dataType[%d]",
205 : HCCL_ERROR_CODE(ret), param.tag.c_str(), execMem.inputMem.ptr(), execMem.count, param.DataDes.dataType),
206 : ret);
207 :
208 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
209 0 : return ret;
210 0 : }
211 :
212 0 : HcclResult CollBroadcastMidCountFor91093Executor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
213 : {
214 0 : tag_ = param.tag;
215 0 : algResResp_ = &algRes;
216 : /* ------------执行算法-------------- */
217 0 : HcclUs startut = TIME_NOW();
218 :
219 : // 图模式和单卡场景下不需要Loop
220 0 : ExecMem execMem;
221 0 : execMem.count = param.DataDes.count;
222 0 : execMem.inputPtr = param.inputPtr;
223 0 : execMem.outputPtr = param.inputPtr;
224 0 : HCCL_INFO(
225 : "Orchestrate UserRank[%u], devicePhyId[%u], inputPtr[%p], outputPtr[%p], root[%u]", topoAttr_.userRank,
226 : topoAttr_.devicePhyId, param.inputPtr, param.outputPtr, param.root);
227 :
228 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
229 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
230 0 : CHK_PTR_NULL(curInputPtr);
231 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize);
232 0 : HCCL_DEBUG(
233 : "[CollBroadcastMidCountFor91093Executor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
234 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop);
235 :
236 0 : u64 totalCount = param.DataDes.count;
237 0 : for (u64 countLeft = totalCount, curCount = 0, inputOffset = 0; countLeft > 0; countLeft -= curCount) {
238 0 : curInputPtr += inputOffset;
239 : // 判断剩余数据量对应的output size是否大于中转output size
240 0 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
241 0 : u64 curSize = curCount * unitSize; // 单位：字节
242 :
243 0 : ExecMem execMem;
244 0 : execMem.count = curCount;
245 0 : execMem.inputMem = algRes.cclOutputMem;
246 0 : execMem.outputMem = algRes.cclOutputMem; // ccl buffer 均只使用out buffer
247 0 : execMem.inputPtr = curInputPtr;
248 0 : HCCL_DEBUG(
249 : "[CollBroadcastMidCountFor91093Executor] RunLoop tag[%s], inputOffset[%llu], "
250 : "curInputPtr[%p], sendCount[%llu], sendSize[%llu], dataType[%s], realUserRank[%u]",
251 : param.tag.c_str(), inputOffset, curInputPtr, curCount, curSize,
252 : GetDataTypeEnumStr(param.DataDes.dataType).c_str(), topoAttr_.realUserRank);
253 :
254 0 : CHK_RET(RunLoopInner(param, execMem));
255 :
256 0 : inputOffset = curSize;
257 0 : }
258 :
259 0 : HCCL_INFO(
260 : "tag[%s], Broadcast executor orchestrate success, take time [%lld]us.", param.tag.c_str(),
261 : DURATION_US(TIME_NOW() - startut));
262 0 : return HCCL_SUCCESS;
263 0 : }
264 :
265 : REGISTER_EXEC("BroadcastMidCountFor91093Executor", BroadcastMidCountFor91093, CollBroadcastMidCountFor91093Executor);
266 : } // namespace hccl
|
