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_all_gather_v_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllGatherVExecutor::CollAllGatherVExecutor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 0 : : CollCommExecutor(dispatcher, topoMatcher)
17 0 : {}
18 :
19 0 : HcclResult CollAllGatherVExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
20 : {
21 0 : HcclUs startut = TIME_NOW();
22 0 : tag_ = param.tag;
23 0 : algResResp_ = &algRes;
24 :
25 0 : u64 count = static_cast<u64*>(param.VDataDes.counts)[topoAttr_.userRank];
26 0 : HcclResult ret = HCCL_SUCCESS;
27 : // 图模式场景下不需要Loop
28 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
29 0 : ExecMem execMem;
30 0 : execMem.count = count;
31 0 : execMem.inputMem = algRes.paramInputMem;
32 0 : execMem.outputMem = algRes.paramOutputMem;
33 0 : execMem.scratchMem = algRes.scratchMem;
34 0 : execMem.inputPtr = param.inputPtr;
35 0 : execMem.outputPtr = param.outputPtr;
36 0 : HCCL_DEBUG(
37 : "[CollAllGatherVExecutor][Orchestrate]offload inputMem[%p][%llu], outputMem[%p][%llu],"
38 : "scratchMem[%p][%llu], inputPtr[%p] outputPtr[%p], count[%llu]",
39 : execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
40 : execMem.scratchMem.ptr(), execMem.scratchMem.size(), execMem.inputPtr, execMem.outputPtr, execMem.count);
41 0 : ret = KernelRun(param, execMem);
42 0 : } else {
43 0 : ret = RunLoop(param, algRes);
44 : }
45 0 : CHK_PRT_RET(
46 : ret != HCCL_SUCCESS,
47 : HCCL_ERROR(
48 : "[CollAllGatherVExecutor][Orchestrate]errNo[0x%016llx]AllGatherV executor kernel run failed",
49 : HCCL_ERROR_CODE(ret)),
50 : ret);
51 0 : HCCL_INFO(
52 : "tag[%s], AllgatherV executor orchestrate success, take time [%lld]us.", param.tag.c_str(),
53 : DURATION_US(TIME_NOW() - startut));
54 0 : return HCCL_SUCCESS;
55 : }
56 :
57 0 : HcclResult CollAllGatherVExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
58 : {
59 : (void)algRes;
60 : (void)adjInfo;
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 0 : u64 CollAllGatherVExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
65 : {
66 : // 中转内存单次最多能够接受的output count
67 0 : u64 maxCountPerLoop = cclBuffSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
68 :
69 0 : HCCL_WARNING(
70 : "[CollAllGatherVExecutor][CalcLoopMaxCount]"
71 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.",
72 : maxCountPerLoop);
73 0 : return maxCountPerLoop;
74 : }
75 :
76 0 : bool CollAllGatherVExecutor::IsHugeData(const u64 curSize)
77 : {
78 0 : bool hugeData = curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
79 0 : || curSize > SDMA_SEND_MAX_SIZE;
80 0 : return hugeData;
81 : }
82 :
83 0 : HcclResult CollAllGatherVExecutor::CalcCurCountsAndCurDispls(
84 : const u64 maxTotalCount, std::vector<u64>& countsLeft, std::vector<u64>& displs, std::vector<u64>& curCounts,
85 : std::vector<u64>& curDispls, bool& finished)
86 : {
87 0 : HCCL_DEBUG("[CollAllGatherVExecutor][CalcCurCountsAndCurDispls]default func called.");
88 0 : return HCCL_SUCCESS;
89 : }
90 :
91 0 : HcclResult CollAllGatherVExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
92 : {
93 : // 每轮loop需要重新计算counts和displs
94 0 : const auto* countsPtr = static_cast<const u64*>(param.VDataDes.counts);
95 0 : auto countsLeft = std::vector<u64>(countsPtr, countsPtr + topoAttr_.userRankSize);
96 0 : const auto* displsPtr = static_cast<const u64*>(param.VDataDes.displs);
97 0 : auto displs = std::vector<u64>(displsPtr, displsPtr + topoAttr_.userRankSize);
98 :
99 0 : const HcclDataType dataType = param.VDataDes.dataType;
100 0 : u32 unitSize = SIZE_TABLE[dataType];
101 :
102 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
103 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
104 0 : u8* commInputPtr = static_cast<u8*>(algRes.cclInputMem.ptr());
105 0 : u8* commOutputPtr = static_cast<u8*>(algRes.cclOutputMem.ptr());
106 :
107 0 : if (UNLIKELY(countsLeft[topoAttr_.userRank] == 0 && curInputPtr == nullptr)) {
108 : // 若本rank的input count为0，此时允许curInputPtr传入空指针，为保证后续流程正常执行，赋值为cclin的地址
109 0 : curInputPtr = commInputPtr;
110 0 : HCCL_DEBUG("Since the input count is 0, set curInputPtr to ccl input[%p]", curInputPtr);
111 : } else {
112 0 : CHK_PTR_NULL(curInputPtr);
113 : }
114 :
115 0 : CHK_PTR_NULL(curOutputPtr);
116 0 : CHK_PTR_NULL(commInputPtr);
117 0 : CHK_PTR_NULL(commOutputPtr);
118 :
119 : // 计算MaxCountPerLoop
120 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize); // override
121 0 : CHK_PRT_RET(
122 : maxCountPerLoop == 0,
123 : HCCL_ERROR(
124 : "[CollAllGatherVExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
125 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
126 : HCCL_E_PARA);
127 :
128 0 : bool finished = false;
129 0 : while (!finished) {
130 : // 每个块尽可能平分，以均衡利用带宽
131 0 : auto curCounts = std::vector<u64>();
132 0 : auto curDispls = std::vector<u64>();
133 0 : CHK_RET(CalcCurCountsAndCurDispls(maxCountPerLoop, countsLeft, displs, curCounts, curDispls, finished));
134 :
135 : // 打印调测信息
136 0 : PrintCurCountAndCurDispls(curCounts, curDispls);
137 0 : u64 totalCount = 0;
138 0 : CHK_RET(CalcTotalCount(curCounts, totalCount));
139 0 : u64 OutputSize = totalCount * unitSize; // 单位：字节
140 :
141 0 : u64 curCount = curCounts[topoAttr_.userRank];
142 0 : u64 InputSize = curCount * unitSize;
143 0 : u64 curMaxCount = *std::max_element(curCounts.begin(), curCounts.end());
144 0 : u64 curMaxSize = curMaxCount * unitSize;
145 0 : HCCL_DEBUG(
146 : "[CollAllGatherVExecutor][RunLoop]tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d], "
147 : "OutputSize[%llu], curMaxSize[%llu].",
148 : param.tag.c_str(), curInputPtr, curOutputPtr, curCount, dataType, OutputSize, curMaxSize);
149 :
150 0 : if (!is310P3Common_) {
151 : /* 设置子图复用标志 */
152 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
153 0 : bool hugeData = IsHugeData(curMaxSize); // override
154 0 : auto opMeta = HcclOpMetaInfo::GetOneForAllGatherV(
155 : autoSelectedAlgTypeLevel1, hugeData, false, CopyPattern::BCOPY, false);
156 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
157 : }
158 :
159 : // 执行
160 0 : if (!DMAReduceFlag_) {
161 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
162 0 : DeviceMem srcMem = DeviceMem::create(curInputPtr, InputSize);
163 0 : DeviceMem dstMem = DeviceMem::create(commInputPtr, InputSize);
164 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
165 0 : HCCL_DEBUG("[CollAllGatherVExecutor][RunLoop]copy from user in to ccl in.");
166 0 : }
167 :
168 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
169 0 : ExecMem execMem;
170 0 : execMem.count = curCount;
171 0 : execMem.inputMem = algRes.cclInputMem;
172 0 : execMem.outputMem = algRes.cclOutputMem;
173 0 : execMem.scratchMem = algRes.scratchMem;
174 0 : execMem.inputPtr = curInputPtr;
175 0 : execMem.outputPtr = curOutputPtr;
176 :
177 0 : OpParam curParam = param;
178 0 : curParam.VDataDes.counts = curCounts.data();
179 0 : curParam.VDataDes.displs = curDispls.data();
180 0 : curParam.VDataDes.dataType = dataType;
181 0 : HcclResult ret = KernelRun(curParam, execMem);
182 0 : CHK_PRT_RET(
183 : ret != HCCL_SUCCESS,
184 : HCCL_ERROR(
185 : "[CollAllGatherVExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], "
186 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d]",
187 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr, curCount, dataType),
188 : ret);
189 :
190 0 : if (!DMAReduceFlag_) {
191 0 : u64 offSetCount = 0;
192 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
193 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
194 : // 拷贝中转output上每个slice的数据到output内存，目的端中每个slice的size固定为output的size
195 0 : DeviceMem dstMem = DeviceMem::create(curOutputPtr + curDispls[i] * unitSize, curCounts[i] * unitSize);
196 0 : DeviceMem srcMem = DeviceMem::create(commOutputPtr + offSetCount * unitSize, curCounts[i] * unitSize);
197 0 : offSetCount += curCounts[i];
198 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
199 0 : }
200 : }
201 :
202 0 : if (!is310P3Common_) {
203 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
204 : }
205 :
206 0 : curInputPtr += InputSize;
207 : // AllGatherV curOutputPtr不需要偏移，output的偏移由displs计算
208 0 : }
209 0 : return HCCL_SUCCESS;
210 0 : }
211 :
212 0 : void CollAllGatherVExecutor::PrintCurCountAndCurDispls(
213 : const std::vector<u64>& curCounts, const std::vector<u64>& curDispls)
214 : {
215 0 : if (HcclCheckLogLevel(DLOG_DEBUG)) {
216 0 : std::ostringstream curLoopInfo;
217 0 : curLoopInfo << "Counts[ ";
218 0 : for (auto count : curCounts) {
219 0 : curLoopInfo << count << " ";
220 : }
221 0 : curLoopInfo << "], displs[ ";
222 0 : for (auto displ : curDispls) {
223 0 : curLoopInfo << displ << " ";
224 : }
225 0 : curLoopInfo << "]";
226 0 : HCCL_DEBUG(
227 : "[CollAllGatherVExecutor][PrintCurCountAndCurDispls] Current loop info: %s", curLoopInfo.str().c_str());
228 0 : }
229 0 : }
230 :
231 0 : HcclResult CollAllGatherVExecutor::CalcTotalCount(std::vector<u64> curCounts, u64& totalCount)
232 : {
233 0 : for (u64 i = 0; i < topoAttr_.userRankSize; i++) {
234 0 : totalCount += curCounts[i];
235 : }
236 :
237 0 : return HCCL_SUCCESS;
238 : }
239 :
240 : } // namespace hccl
|
