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 : [[maybe_unused]] const u64 maxTotalCount, [[maybe_unused]] std::vector<u64>& countsLeft,
85 : [[maybe_unused]] std::vector<u64>& displs, [[maybe_unused]] std::vector<u64>& curCounts,
86 : [[maybe_unused]] std::vector<u64>& curDispls, [[maybe_unused]] bool& finished)
87 : {
88 0 : HCCL_DEBUG("[CollAllGatherVExecutor][CalcCurCountsAndCurDispls]default func called.");
89 0 : return HCCL_SUCCESS;
90 : }
91 :
92 0 : HcclResult CollAllGatherVExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
93 : {
94 : // 每轮loop需要重新计算counts和displs
95 0 : const auto* countsPtr = static_cast<const u64*>(param.VDataDes.counts);
96 0 : auto countsLeft = std::vector<u64>(countsPtr, countsPtr + topoAttr_.userRankSize);
97 0 : const auto* displsPtr = static_cast<const u64*>(param.VDataDes.displs);
98 0 : auto displs = std::vector<u64>(displsPtr, displsPtr + topoAttr_.userRankSize);
99 :
100 0 : const HcclDataType dataType = param.VDataDes.dataType;
101 0 : u32 unitSize = SIZE_TABLE[dataType];
102 :
103 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
104 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
105 0 : u8* commInputPtr = static_cast<u8*>(algRes.cclInputMem.ptr());
106 0 : u8* commOutputPtr = static_cast<u8*>(algRes.cclOutputMem.ptr());
107 :
108 0 : if (UNLIKELY(countsLeft[topoAttr_.userRank] == 0 && curInputPtr == nullptr)) {
109 : // 若本rank的input count为0，此时允许curInputPtr传入空指针，为保证后续流程正常执行，赋值为cclin的地址
110 0 : curInputPtr = commInputPtr;
111 0 : HCCL_DEBUG("Since the input count is 0, set curInputPtr to ccl input[%p]", curInputPtr);
112 : } else {
113 0 : CHK_PTR_NULL(curInputPtr);
114 : }
115 :
116 0 : CHK_PTR_NULL(curOutputPtr);
117 0 : CHK_PTR_NULL(commInputPtr);
118 0 : CHK_PTR_NULL(commOutputPtr);
119 :
120 : // 计算MaxCountPerLoop
121 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize); // override
122 0 : CHK_PRT_RET(
123 : maxCountPerLoop == 0,
124 : HCCL_ERROR(
125 : "[CollAllGatherVExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
126 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
127 : HCCL_E_PARA);
128 :
129 0 : bool finished = false;
130 0 : while (!finished) {
131 : // 每个块尽可能平分，以均衡利用带宽
132 0 : auto curCounts = std::vector<u64>();
133 0 : auto curDispls = std::vector<u64>();
134 0 : CHK_RET(CalcCurCountsAndCurDispls(maxCountPerLoop, countsLeft, displs, curCounts, curDispls, finished));
135 :
136 : // 打印调测信息
137 0 : PrintCurCountAndCurDispls(curCounts, curDispls);
138 0 : u64 totalCount = 0;
139 0 : CHK_RET(CalcTotalCount(curCounts, totalCount));
140 0 : u64 OutputSize = totalCount * unitSize; // 单位：字节
141 :
142 0 : u64 curCount = curCounts[topoAttr_.userRank];
143 0 : u64 InputSize = curCount * unitSize;
144 0 : u64 curMaxCount = *std::max_element(curCounts.begin(), curCounts.end());
145 0 : u64 curMaxSize = curMaxCount * unitSize;
146 0 : HCCL_DEBUG(
147 : "[CollAllGatherVExecutor][RunLoop]tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d], "
148 : "OutputSize[%llu], curMaxSize[%llu].",
149 : param.tag.c_str(), curInputPtr, curOutputPtr, curCount, dataType, OutputSize, curMaxSize);
150 :
151 0 : if (!is310P3Common_) {
152 : /* 设置子图复用标志 */
153 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
154 0 : bool hugeData = IsHugeData(curMaxSize); // override
155 0 : auto opMeta = HcclOpMetaInfo::GetOneForAllGatherV(
156 : autoSelectedAlgTypeLevel1, hugeData, false, CopyPattern::BCOPY, false);
157 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
158 : }
159 :
160 : // 执行
161 0 : if (!DMAReduceFlag_) {
162 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
163 0 : DeviceMem srcMem = DeviceMem::create(curInputPtr, InputSize);
164 0 : DeviceMem dstMem = DeviceMem::create(commInputPtr, InputSize);
165 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
166 0 : HCCL_DEBUG("[CollAllGatherVExecutor][RunLoop]copy from user in to ccl in.");
167 0 : }
168 :
169 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
170 0 : ExecMem execMem;
171 0 : execMem.count = curCount;
172 0 : execMem.inputMem = algRes.cclInputMem;
173 0 : execMem.outputMem = algRes.cclOutputMem;
174 0 : execMem.scratchMem = algRes.scratchMem;
175 0 : execMem.inputPtr = curInputPtr;
176 0 : execMem.outputPtr = curOutputPtr;
177 :
178 0 : OpParam curParam = param;
179 0 : curParam.VDataDes.counts = curCounts.data();
180 0 : curParam.VDataDes.displs = curDispls.data();
181 0 : curParam.VDataDes.dataType = dataType;
182 0 : HcclResult ret = KernelRun(curParam, execMem);
183 0 : CHK_PRT_RET(
184 : ret != HCCL_SUCCESS,
185 : HCCL_ERROR(
186 : "[CollAllGatherVExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], "
187 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d]",
188 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr, curCount, dataType),
189 : ret);
190 :
191 0 : if (!DMAReduceFlag_) {
192 0 : u64 offSetCount = 0;
193 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
194 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
195 : // 拷贝中转output上每个slice的数据到output内存，目的端中每个slice的size固定为output的size
196 0 : DeviceMem dstMem = DeviceMem::create(curOutputPtr + curDispls[i] * unitSize, curCounts[i] * unitSize);
197 0 : DeviceMem srcMem = DeviceMem::create(commOutputPtr + offSetCount * unitSize, curCounts[i] * unitSize);
198 0 : offSetCount += curCounts[i];
199 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
200 0 : }
201 : }
202 :
203 0 : if (!is310P3Common_) {
204 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
205 : }
206 :
207 0 : curInputPtr += InputSize;
208 : // AllGatherV curOutputPtr不需要偏移，output的偏移由displs计算
209 0 : }
210 0 : return HCCL_SUCCESS;
211 0 : }
212 :
213 0 : void CollAllGatherVExecutor::PrintCurCountAndCurDispls(
214 : const std::vector<u64>& curCounts, const std::vector<u64>& curDispls)
215 : {
216 0 : if (HcclCheckLogLevel(DLOG_DEBUG)) {
217 0 : std::ostringstream curLoopInfo;
218 0 : curLoopInfo << "Counts[ ";
219 0 : for (auto count : curCounts) {
220 0 : curLoopInfo << count << " ";
221 : }
222 0 : curLoopInfo << "], displs[ ";
223 0 : for (auto displ : curDispls) {
224 0 : curLoopInfo << displ << " ";
225 : }
226 0 : curLoopInfo << "]";
227 0 : HCCL_DEBUG(
228 : "[CollAllGatherVExecutor][PrintCurCountAndCurDispls] Current loop info: %s", curLoopInfo.str().c_str());
229 0 : }
230 0 : }
231 :
232 0 : HcclResult CollAllGatherVExecutor::CalcTotalCount(std::vector<u64> curCounts, u64& totalCount)
233 : {
234 0 : for (u64 i = 0; i < topoAttr_.userRankSize; i++) {
235 0 : totalCount += curCounts[i];
236 : }
237 :
238 0 : return HCCL_SUCCESS;
239 : }
240 :
241 : } // namespace hccl
|
