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