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_reduce_executor.h"
12 :
13 : namespace hccl {
14 :
15 2 : CollReduceExecutor::CollReduceExecutor(const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 2 : : CollCommExecutor(dispatcher, topoMatcher)
17 5 : {}
18 :
19 1 : HcclResult CollReduceExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
20 : {
21 1 : HcclUs startut = TIME_NOW();
22 :
23 1 : tag_ = param.tag;
24 1 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD || algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_HD) {
25 0 : std::string appendTag = "";
26 0 : u32 serverNumPerSuperPod
27 0 : = topoAttr_.superPodNum == 0 ? topoAttr_.moduleNum : topoAttr_.moduleNum / topoAttr_.superPodNum;
28 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_HD) {
29 0 : u32 part1Size = FACTOR_TWO * (serverNumPerSuperPod - (1 << static_cast<u32>(log2(serverNumPerSuperPod))));
30 0 : u32 rootId = param.root / topoAttr_.deviceNumPerAggregation % serverNumPerSuperPod;
31 0 : appendTag += "L1_" + std::to_string((rootId >= part1Size) || ((rootId % FACTOR_TWO) == 0));
32 : }
33 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_HD) {
34 0 : u32 part1Size = FACTOR_TWO * (topoAttr_.superPodNum - (1 << static_cast<u32>(log2(topoAttr_.superPodNum))));
35 0 : u32 rootId = param.root / topoAttr_.deviceNumPerAggregation / serverNumPerSuperPod;
36 0 : appendTag += (appendTag.empty() ? "L2_" : "_L2_")
37 0 : + std::to_string((rootId >= part1Size) || ((rootId % FACTOR_TWO) == 0));
38 : }
39 0 : tag_ = param.tag + '_' + appendTag;
40 0 : if (param.opBaseAtraceInfo != nullptr) {
41 0 : CHK_RET(param.opBaseAtraceInfo->SavealgtypeTraceInfo(appendTag, param.tag));
42 : }
43 0 : }
44 :
45 1 : algResResp_ = &algRes;
46 1 : HcclResult ret = HCCL_SUCCESS;
47 1 : bool needLaunchAtTheEnd = true; // 是否需要在Orchestrate()结束时launch任务
48 1 : ExecMem execMem;
49 1 : execMem.count = param.DataDes.count;
50 1 : execMem.inputPtr = param.inputPtr;
51 1 : execMem.outputPtr = param.outputPtr;
52 : // 图模式和单卡场景下不需要Loop
53 1 : HCCL_DEBUG("[CollReduceExecutor][Orchestrate]workflowMode is %d", workflowMode_);
54 1 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
55 1 : execMem.inputMem = algRes.paramInputMem;
56 1 : execMem.outputMem = algRes.paramOutputMem;
57 1 : execMem.scratchMem = algRes.scratchMem;
58 1 : ret = KernelRun(param, execMem);
59 1 : if (algOpContext_.opRetryHandler.isPostSync == true) {
60 : // post Sync
61 0 : CHK_RET(RetryPostSync(param, execMem));
62 : }
63 0 : } else if (topoAttr_.userRankSize == 1) {
64 0 : execMem.inputMem = algRes.cclInputMem;
65 0 : execMem.outputMem = algRes.cclOutputMem;
66 0 : execMem.scratchMem = algRes.scratchMem;
67 0 : ret = KernelRun(param, execMem);
68 0 : needLaunchAtTheEnd = false;
69 : } else {
70 0 : ret = RunLoop(param, algRes);
71 0 : needLaunchAtTheEnd = false;
72 : }
73 1 : CHK_PRT_RET(
74 : ret != HCCL_SUCCESS,
75 : HCCL_ERROR(
76 : "[CollReduceExecutor][Orchestrate]errNo[0x%016llx]reduce executor kernel run failed", HCCL_ERROR_CODE(ret)),
77 : ret);
78 :
79 : // Enforce task launch at the end of Orchestrate
80 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
81 1 : if (needLaunchAtTheEnd) {
82 1 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
83 1 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
84 : }
85 :
86 1 : HCCL_INFO(
87 : "tag[%s], Reduce executor orchestrate success, take time [%lld]us.", tag_.c_str(),
88 : DURATION_US(TIME_NOW() - startut));
89 1 : return HCCL_SUCCESS;
90 1 : }
91 :
92 0 : HcclResult CollReduceExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
93 : {
94 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
95 0 : ReduceType reduceType
96 0 : = ((param.reduceType != HCCL_REDUCE_PROD) && (param.DataDes.dataType != HCCL_DATA_TYPE_INT64)) ?
97 : ReduceType::INLINE_REDUCE :
98 : ReduceType::TBE_REDUCE;
99 :
100 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
101 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
102 0 : CHK_PTR_NULL(curInputPtr);
103 0 : CHK_PTR_NULL(curOutputPtr);
104 :
105 0 : u64 maxCountPerLoop = CalcLoopMaxCount(unitSize, algRes); // override
106 :
107 0 : HCCL_DEBUG(
108 : "[CollReduceExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].", tag_.c_str(),
109 : topoAttr_.userRankSize, maxCountPerLoop);
110 :
111 0 : u64 inputOffset = 0;
112 0 : u64 outputOffset = 0;
113 0 : u64 countLeft = param.DataDes.count;
114 0 : while (countLeft > 0) {
115 0 : curInputPtr += inputOffset;
116 0 : curOutputPtr += outputOffset;
117 : // 判断剩余数据量对应的output size是否大于中转output size
118 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
119 0 : u64 curSize = curCount * unitSize; // 单位：字节
120 :
121 0 : HCCL_DEBUG(
122 : "[CollReduceExecutor][RunLoop]tag[%s], inputOffset[%llu], outputOffset[%llu], "
123 : "sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d].",
124 : tag_.c_str(), inputOffset, outputOffset, curInputPtr, curOutputPtr, curCount, param.DataDes.dataType);
125 :
126 0 : ExecMem execMem;
127 0 : execMem.count = curCount;
128 0 : execMem.inputMem = algRes.cclInputMem;
129 0 : execMem.outputMem = algRes.cclOutputMem;
130 0 : execMem.scratchMem = algRes.scratchMem;
131 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
132 0 : execMem.inputPtr = curInputPtr;
133 0 : execMem.outputPtr = curOutputPtr;
134 :
135 0 : CHK_RET(RunLoopInner(param, reduceType, execMem));
136 :
137 0 : countLeft -= curCount;
138 0 : inputOffset = curSize;
139 0 : outputOffset = curSize;
140 0 : }
141 0 : if (algOpContext_.opRetryHandler.isPostSync == true) {
142 0 : ExecMem execMem;
143 0 : execMem.count = param.DataDes.count;
144 0 : execMem.inputPtr = param.inputPtr;
145 0 : execMem.outputPtr = param.outputPtr;
146 0 : execMem.inputMem = algRes.cclInputMem;
147 0 : execMem.outputMem = algRes.cclOutputMem;
148 0 : execMem.scratchMem = algRes.scratchMem;
149 : // post Sync
150 0 : CHK_RET(RetryPostSync(param, execMem));
151 0 : }
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : HcclResult CollReduceExecutor::RunLoopInner(OpParam& param, const ReduceType& reduceType, ExecMem& execMem)
156 : {
157 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
158 0 : u64 curSize = execMem.count * unitSize; // 单位：字节
159 0 : HCCL_DEBUG(
160 : "[CollReduceExecutor][RunLoopInner]inputMem[%p][%llu], outputMem[%p][%llu], "
161 : "intputPtr[%p], outputPtr[%p], curCount[%llu], curSize[%llu]",
162 : execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
163 : execMem.inputPtr, execMem.outputPtr, execMem.count, curSize);
164 0 : CHK_PRT_RET(
165 : (execMem.count == 0), HCCL_ERROR("[CollReduceExecutor][RunLoopInner]In OP_BASE curCount is zero."),
166 : HCCL_E_PARA);
167 :
168 : /* 设置子图复用标志 */
169 0 : bool isRootRank = param.root == topoAttr_.realUserRank ? true : false;
170 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
171 0 : bool hugeData = IsHugeData(curSize); // override
172 : /* TBE reduce 当总count数超过INT32_MAX时，不使能子图复用 */
173 0 : if (reduceType == ReduceType::TBE_REDUCE) {
174 0 : hugeData = hugeData || param.DataDes.count > INT32_MAX;
175 : }
176 0 : HCCL_DEBUG("[CollReduceExecutor][RunLoopInner]IsHugeData:[%u]", hugeData);
177 0 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
178 0 : auto opMeta = HcclOpMetaInfo::GetOneForReduce(
179 : isRootRank, param.root, autoSelectedAlgTypeLevel1, param.DataDes.dataType, reduceType, hugeData, deterministic);
180 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
181 :
182 0 : execMem.inputMem = DeviceMem::create(execMem.inputMem.ptr(), curSize);
183 0 : execMem.outputMem = DeviceMem::create(execMem.outputMem.ptr(), curSize);
184 :
185 : // 执行
186 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
187 0 : DeviceMem inMem(execMem.inputPtr, curSize);
188 0 : DeviceMem inCommMem = execMem.inputMem.range(0, curSize);
189 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, param.stream));
190 0 : HCCL_DEBUG("[CollReduceExecutor][RunLoopInner]copy from user in to ccl in.");
191 :
192 0 : HcclResult ret = KernelRun(param, execMem);
193 0 : CHK_PRT_RET(
194 : ret != HCCL_SUCCESS,
195 : HCCL_ERROR(
196 : "[CollReduceExecutor][RunLoopInner]errNo[0x%016llx]kernel run error, tag[%s], "
197 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d], reduce op type[%d]",
198 : HCCL_ERROR_CODE(ret), tag_.c_str(), execMem.inputMem.ptr(), execMem.outputMem.ptr(), execMem.count,
199 : param.DataDes.dataType, param.reduceType),
200 : ret);
201 :
202 0 : if (topoAttr_.realUserRank == param.root) { // 只root rank需要把数据从中转内存拷贝出去
203 0 : DeviceMem outMem(execMem.outputPtr, curSize);
204 0 : DeviceMem outCommMem = execMem.outputMem.range(0, curSize);
205 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outMem, outCommMem, param.stream));
206 0 : }
207 :
208 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
209 0 : return ret;
210 0 : }
211 :
212 0 : u64 CollReduceExecutor::CalcLoopMaxCount(const u32 unitSize, const AlgResourceResponse& algRes)
213 : {
214 : // 中转内存单次最多能够接受的output count
215 0 : u64 maxCountPerLoop = algRes.cclInputMem.size() / unitSize;
216 0 : HCCL_WARNING(
217 : "[CollReduceExecutor][CalcLoopMaxCount]"
218 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.",
219 : maxCountPerLoop);
220 0 : return maxCountPerLoop;
221 : }
222 :
223 0 : bool CollReduceExecutor::IsHugeData(const u64 curSize)
224 : {
225 0 : HCCL_WARNING("[CollReduceExecutor][IsHugeData]opMeta is using the default option.");
226 0 : bool hugeData = (curSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE) || (curSize > SDMA_SEND_MAX_SIZE);
227 0 : return hugeData;
228 : }
229 :
230 0 : HcclResult CollReduceExecutor::RetryPostSync(OpParam& param, ExecMem& execMem)
231 : {
232 0 : if ((algResResp_->slaveStreams).size() == 0) {
233 0 : CHK_RET(PostSyncWithoutSubstream(param, execMem));
234 : } else {
235 0 : PrepareData postSyncPrepareData;
236 0 : postSyncPrepareData.subStreamsPtr = &algResResp_->slaveStreams;
237 0 : postSyncPrepareData.signalPtr = &algResResp_->notifiesMain;
238 0 : postSyncPrepareData.signalAuxPtr = &algResResp_->notifiesAux;
239 0 : postSyncPrepareData.stream = param.stream;
240 0 : CHK_RET(PostSyncWithSubstream(param, execMem, postSyncPrepareData));
241 0 : }
242 0 : return HCCL_SUCCESS;
243 : }
244 : } // namespace hccl
|
