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