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