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_deter_pipeline_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollReduceScatterDeterPipelineExecutor::CollReduceScatterDeterPipelineExecutor(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollReduceScatterExecutor(dispatcher, topoMatcher)
18 : {
19 0 : scratchMemFlag_ = (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
20 0 : DMAReduceFlag_ = (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
21 0 : }
22 :
23 0 : void CollReduceScatterDeterPipelineExecutor::ParseParam(const OpParam& param)
24 : {
25 0 : tag_ = param.tag;
26 0 : curOffset_ = 0;
27 0 : totalSize_ = topoAttr_.userRankSize * param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
28 0 : }
29 :
30 0 : HcclResult CollReduceScatterDeterPipelineExecutor::CalcScratchMemSize(u64& scratchMemSize)
31 : {
32 0 : scratchMemSize = scratchMemFlag_ ? totalSize_ + topoAttr_.userRankSize * HCCL_MIN_SLICE_ALIGN_910B : 0U;
33 0 : HCCL_INFO(
34 : "[CollReduceScatterDeterPipelineExecutor][CalcScratchMemSize]tag[%s] scratchMemSize[%llu]", tag_.c_str(),
35 : scratchMemSize);
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 0 : HcclResult CollReduceScatterDeterPipelineExecutor::CalcStreamNum(u32& streamNum)
40 : {
41 0 : streamNum = topoAttr_.deviceNumPerAggregation + 3U; // (deviceNum - 1)机内 + 4Reduce + 1机间 - 1主流
42 0 : HCCL_INFO("[CollReduceScatterDeterPipelineExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
43 0 : return HCCL_SUCCESS;
44 : }
45 :
46 0 : HcclResult CollReduceScatterDeterPipelineExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
47 : {
48 0 : TransportMemType inputType = TransportMemType::RESERVED;
49 0 : TransportMemType outputType = TransportMemType::RESERVED;
50 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
51 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
52 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
53 0 : return HCCL_SUCCESS;
54 : }
55 :
56 0 : HcclResult CollReduceScatterDeterPipelineExecutor::CalcLevel0CommInfo(
57 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
58 : {
59 0 : CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_MESH);
60 0 : commParaInfo.meshSinglePlane = true;
61 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
62 0 : return HCCL_SUCCESS;
63 0 : }
64 :
65 0 : HcclResult CollReduceScatterDeterPipelineExecutor::CalcLevel1CommInfo(
66 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
67 : {
68 0 : CommParaInfo commParaInfo(COMM_LEVEL1, CommType::COMM_TAG_MESH);
69 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL1], inputType, outputType));
70 0 : return HCCL_SUCCESS;
71 0 : }
72 :
73 : HcclResult
74 0 : CollReduceScatterDeterPipelineExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
75 : {
76 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
77 0 : inputType = TransportMemType::CCL_INPUT;
78 0 : outputType = TransportMemType::CCL_OUTPUT;
79 : } else {
80 0 : inputType = TransportMemType::PARAM_INPUT;
81 0 : outputType = TransportMemType::SCRATCH;
82 : }
83 0 : HCCL_INFO(
84 : "[CollReduceScatterDeterPipelineExecutor][CalcTransportMemType] tag[%s] inputType[%d], "
85 : "outputType[%d]",
86 : tag_.c_str(), inputType, outputType);
87 0 : return HCCL_SUCCESS;
88 : }
89 :
90 0 : u64 CollReduceScatterDeterPipelineExecutor::CalcLoopMaxCount(const u32 unitSize)
91 : {
92 : // 中转内存单次最多能够接受的output count，这里每一块的地址不要求128byte对齐
93 0 : u64 maxCountPerLoop = ((inCCLbufferSize_ / topoAttr_.userRankSize) - HCCL_MIN_SLICE_ALIGN_910B) / unitSize;
94 0 : HCCL_INFO("[CollReduceScatterDeterPipelineExecutor][CalcLoopMaxCount] maxCountPerLoop[%llu]", maxCountPerLoop);
95 0 : return maxCountPerLoop;
96 : }
97 :
98 0 : HcclResult CollReduceScatterDeterPipelineExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
99 : {
100 0 : HCCL_CONFIG_INFO(
101 : HCCL_ALG, "[CollReduceScatterDeterPipelineExecutor][RunLoop] tag[%s], userRank[%u] begins.", tag_.c_str(),
102 : topoAttr_.userRank);
103 :
104 0 : CHK_PRT_RET(
105 : (param.reduceType == HCCL_REDUCE_PROD) || (param.DataDes.dataType == HCCL_DATA_TYPE_INT64),
106 : HCCL_ERROR(
107 : "[CollReduceScatterDeterPipelineExecutor] unsupported reduceType[%u] or unsupported dataType[%u]",
108 : param.reduceType, param.DataDes.dataType),
109 : HCCL_E_INTERNAL);
110 :
111 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
112 :
113 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
114 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
115 0 : CHK_PTR_NULL(curInputPtr);
116 0 : CHK_PTR_NULL(curOutputPtr);
117 :
118 0 : u64 maxCountPerLoop = CalcLoopMaxCount(unitSize);
119 0 : HCCL_INFO(
120 : "[CollReduceScatterDeterPipelineExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop "
121 : "is [%llu].",
122 : tag_.c_str(), topoAttr_.userRankSize, maxCountPerLoop);
123 :
124 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
125 0 : u8 deterministic = topoMatcher_->GetExternalInputHcclDeterministic();
126 :
127 0 : for (u64 countLeft = param.DataDes.count, curCount = 0, curSize = 0; countLeft > 0; countLeft -= curCount) {
128 0 : curInputPtr += curSize;
129 0 : curOutputPtr += curSize;
130 :
131 0 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
132 0 : curSize = curCount * unitSize;
133 :
134 0 : HCCL_CONFIG_DEBUG(
135 : HCCL_ALG,
136 : "[CollReduceScatterDeterPipelineExecutor][RunLoop]tag[%s], curOffset[%llu],"
137 : "curInputPtr[%p], curOutputPtr[%p], curCount[%llu], dataType[%d].",
138 : tag_.c_str(), curOffset_, curInputPtr, curOutputPtr, curCount, param.DataDes.dataType);
139 :
140 0 : constexpr s64 HCCL_MEDIUM_COUNT_2_MB = 2 * 1024 * 1024;
141 0 : bool smallData = curSize < HCCL_MEDIUM_COUNT_2_MB ? 1 : 0;
142 0 : bool hugeData = IsHugeData(curSize);
143 0 : auto meta = HcclOpMetaInfo::GetOneForReduceScatter(
144 : autoSelectedAlgTypeLevel1, param.DataDes.dataType, ReduceType::INLINE_REDUCE, hugeData, smallData,
145 : CopyPattern::ZCOPY, false, deterministic, false);
146 0 : CHK_RET(InitTask(dispatcher_, const_cast<Stream&>(param.stream), meta.isEnableCache, meta.GetCacheKey(), true));
147 0 : ExecMem execMem;
148 0 : execMem.count = curCount;
149 0 : execMem.inputMem = algRes.cclInputMem;
150 0 : execMem.outputMem = algRes.cclOutputMem;
151 0 : execMem.scratchMem = algRes.cclOutputMem;
152 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
153 0 : execMem.inputPtr = curInputPtr;
154 0 : execMem.outputPtr = curOutputPtr;
155 :
156 0 : CHK_RET(KernelRun(param, execMem));
157 :
158 0 : CHK_RET(LaunchTaskExtend(
159 : dispatcher_, const_cast<Stream&>(param.stream),
160 : const_cast<std::vector<Stream>&>(algResResp_->slaveStreams)));
161 :
162 0 : curOffset_ += curSize;
163 0 : }
164 0 : HCCL_INFO(
165 : "[CollReduceScatterDeterPipelineExecutor][RunLoop] tag[%s], userRank[%u] run loop success.", tag_.c_str(),
166 : topoAttr_.userRank);
167 :
168 0 : return HCCL_SUCCESS;
169 : }
170 :
171 0 : HcclResult CollReduceScatterDeterPipelineExecutor::PrepareDataSlice(
172 : const OpParam& param, const ExecMem& execMem, const SubCommInfo& level0CommInfo, const SubCommInfo& level1CommInfo,
173 : std::vector<Slice>& bufferSlices)
174 : {
175 0 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
176 0 : bufferSlices.resize(topoAttr_.userRankSize);
177 0 : u64 totalOutputSize = param.DataDes.count * unitSize;
178 0 : u32 rankIdLevel0 = level0CommInfo.localRank;
179 0 : u32 rankSizeLevel0 = level0CommInfo.localRankSize;
180 0 : u32 rankIdLevel1 = level1CommInfo.localRank;
181 0 : u32 rankSizeLevel1 = level1CommInfo.localRankSize;
182 :
183 0 : for (u32 i = 0; i < rankSizeLevel1; i++) {
184 0 : u32 inputBlockIndex = (rankIdLevel1 + i) % rankSizeLevel1;
185 0 : u32 outputBlockIndex = (rankIdLevel1 + rankSizeLevel1 - i) % rankSizeLevel1;
186 0 : u32 inputSliceIndex = inputBlockIndex * rankSizeLevel0 + rankIdLevel0;
187 0 : u64 inputSliceOffset = totalOutputSize * inputSliceIndex + curOffset_;
188 0 : for (u32 j = 0; j < rankSizeLevel0; j++) {
189 0 : u32 outputSliceIndex = outputBlockIndex * rankSizeLevel0 + j;
190 0 : bufferSlices[outputSliceIndex].size = execMem.count * unitSize;
191 : u64 outputSliceOffset
192 0 : = (bufferSlices[outputSliceIndex].size + HCCL_MIN_SLICE_ALIGN_910B) * outputSliceIndex;
193 0 : u64 outputInSliceOffset = (HCCL_MIN_SLICE_ALIGN_910B + (inputSliceOffset % HCCL_MIN_SLICE_ALIGN_910B)
194 : - (outputSliceOffset % HCCL_MIN_SLICE_ALIGN_910B))
195 0 : % HCCL_MIN_SLICE_ALIGN_910B;
196 0 : bufferSlices[outputSliceIndex].offset = outputSliceOffset + outputInSliceOffset;
197 0 : HCCL_DEBUG(
198 : "[CollReduceScatterDeterPipelineExecutor][PrepareDataSlice]tag[%s], buffer slice i[%u], "
199 : "size[%llu], offset[%llu], outputInSliceOffset[%llu], inputSliceIndex[%u], inputSliceOffset[%llu], "
200 : "curOffset[%llu]",
201 : tag_.c_str(), outputSliceIndex, bufferSlices[outputSliceIndex].size,
202 : bufferSlices[outputSliceIndex].offset, outputInSliceOffset, inputSliceIndex, inputSliceOffset,
203 : curOffset_);
204 : }
205 : }
206 0 : return HCCL_SUCCESS;
207 : }
208 :
209 0 : HcclResult CollReduceScatterDeterPipelineExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
210 : {
211 0 : HCCL_CONFIG_INFO(
212 : HCCL_ALG, "[CollReduceScatterDeterPipelineExecutor][KernelRun] tag[%s], userRank[%u] starts.", tag_.c_str(),
213 : topoAttr_.userRank);
214 :
215 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
216 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
217 0 : u32 commIndex = level0CommInfo.localRank;
218 :
219 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
220 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
221 :
222 0 : std::vector<Slice> bufferSlices; // 数据分成ranksize份，每份的起始偏移和大小
223 0 : CHK_RET(PrepareDataSlice(param, execMem, level0CommInfo, level1CommInfo, bufferSlices));
224 :
225 0 : CHK_RET(ActiveSlaveStreams(param.stream));
226 :
227 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
228 0 : TemplateType::TEMPLATE_REDUCESCATTER_MULTI_DETERMINISTIC_PIPELINE, dispatcher_);
229 0 : CHK_SMART_PTR_NULL(tempAlg);
230 :
231 0 : HcomCollOpInfo opInfo = {"",
232 0 : execMem.inputPtr,
233 0 : execMem.outputPtr,
234 0 : param.DataDes.count,
235 0 : param.DataDes.dataType,
236 0 : param.root,
237 0 : param.reduceType,
238 0 : 0};
239 :
240 0 : CHK_RET(tempAlg->Prepare(
241 : &opInfo, execMem.scratchMem, execMem.count, curOffset_, bufferSlices, level0CommInfo, level1CommInfo,
242 : const_cast<Stream&>(param.stream), algResResp_->slaveStreams, algResResp_->notifiesMain,
243 : algResResp_->notifiesAux));
244 0 : CHK_RET(tempAlg->RunAsync());
245 :
246 0 : HCCL_INFO(
247 : "[CollReduceScatterDeterPipelineExecutor][KernelRun] tag[%s], userRank[%u] run success.", tag_.c_str(),
248 : topoAttr_.userRank);
249 0 : return HCCL_SUCCESS;
250 0 : }
251 :
252 : REGISTER_EXEC("ReduceScatterDeterPipelineExecutor", ReduceScatterDeterPipeline, CollReduceScatterDeterPipelineExecutor);
253 :
254 : } // namespace hccl
|
