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