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_mix_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllGatherMixExecutor::CollAllGatherMixExecutor(
15 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 0 : : CollAllGatherExecutor(dispatcher, topoMatcher)
17 : {
18 0 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
19 0 : }
20 :
21 0 : HcclResult CollAllGatherMixExecutor::CalcStreamNum(u32& streamNum)
22 : {
23 0 : u32 totalStreamNum = 0;
24 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) { // mesh
25 0 : totalStreamNum = topoAttr_.deviceNumPerAggregation;
26 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) { // dbring
27 : totalStreamNum
28 0 : = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
29 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
30 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
31 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING;
32 : }
33 : }
34 :
35 0 : streamNum = totalStreamNum - 1U;
36 0 : HCCL_INFO("[CollAllGatherMixExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
37 0 : return HCCL_SUCCESS;
38 : }
39 :
40 0 : HcclResult CollAllGatherMixExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
41 : {
42 0 : TransportMemType inputType = TransportMemType::RESERVED;
43 0 : TransportMemType outputType = TransportMemType::RESERVED;
44 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
45 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
46 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
47 :
48 : // mix在server间使用NHR通信域，并在多机A+X场景下当未设置使用RDMA时，默认使用RDMA
49 0 : std::vector<SingleSubCommTransport>& commTransportLevel1 = opTransport[COMM_LEVEL1];
50 0 : for (u32 ringIndex = 0; ringIndex < commTransportLevel1.size(); ringIndex++) {
51 0 : for (auto& transportRequest : commTransportLevel1[ringIndex].transportRequests) {
52 0 : transportRequest.isUsedRdma = true;
53 : }
54 : }
55 0 : return HCCL_SUCCESS;
56 : }
57 :
58 0 : HcclResult CollAllGatherMixExecutor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
59 : {
60 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
61 0 : inputType = TransportMemType::CCL_INPUT;
62 0 : outputType = TransportMemType::CCL_OUTPUT;
63 : } else {
64 0 : inputType = TransportMemType::PARAM_INPUT;
65 0 : outputType = TransportMemType::PARAM_OUTPUT;
66 : }
67 0 : HCCL_INFO(
68 : "[CollAllGatherMixExecutor][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]", tag_.c_str(),
69 : inputType, outputType);
70 0 : return HCCL_SUCCESS;
71 : }
72 :
73 0 : HcclResult CollAllGatherMixExecutor::CalcLevel0CommInfo(
74 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
75 : {
76 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
77 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_MESH);
78 0 : commParaLevel0.meshSinglePlane = topoMatcher_->GetExternalInputHcclDeterministic() == DETERMINISTIC_DISABLE
79 0 : && (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE);
80 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
81 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
82 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
83 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
84 0 : }
85 :
86 0 : return HCCL_SUCCESS;
87 : }
88 :
89 0 : u64 CollAllGatherMixExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
90 : {
91 0 : u64 maxCountPerLoop = cclBuffSize / (topoAttr_.userRankSize * unitSize);
92 0 : return maxCountPerLoop;
93 : }
94 :
95 0 : HcclResult CollAllGatherMixExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
96 : {
97 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherMixExecutor][KernelRun] The AllGatherMixExecutor starts.");
98 0 : u32 perDataSize = SIZE_TABLE[param.DataDes.dataType];
99 :
100 : // 获取子通信域信息
101 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
102 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
103 0 : u32 level0RankSize = level0CommInfo.localRankSize;
104 0 : u32 commIndex = level0CommInfo.localRank;
105 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
106 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
107 0 : u32 serverIndex = level1CommInfo.localRank;
108 0 : u32 level1RankSize = level1CommInfo.localRankSize;
109 0 : HCCL_DEBUG(
110 : "inputSize=%llu, level0RankSize=%u, commIndex=%u, level1RankSize=%u, serverIndex=%u", execMem.inputMem.size(),
111 : level0RankSize, commIndex, level1RankSize, serverIndex);
112 :
113 0 : u64 inputMemSize = execMem.inputMem.size();
114 0 : u64 baseOffset = serverIndex * inputMemSize * level0RankSize;
115 0 : u64 level0Offset = commIndex * inputMemSize;
116 0 : DeviceMem dstMem = execMem.outputMem.range(baseOffset + level0Offset, inputMemSize);
117 0 : CHK_SMART_PTR_NULL(dstMem);
118 :
119 0 : HcomCollOpInfo opInfo = {
120 0 : "", execMem.inputPtr, execMem.outputPtr, param.DataDes.count, param.DataDes.dataType, 0, HCCL_REDUCE_RESERVED,
121 0 : 0};
122 0 : HcomCollOpInfo* opInfoPtr = nullptr;
123 : // 图模式opinfo不为空，但需要将数据从ccl input拷贝到ccl output上
124 0 : HcclResult ret = HCCL_SUCCESS;
125 0 : if (!DMAReduceFlag_) {
126 0 : ret = HcclD2DMemcpyAsync(dispatcher_, dstMem, execMem.inputMem, const_cast<Stream&>(param.stream));
127 0 : CHK_PRT_RET(
128 : ret != HCCL_SUCCESS,
129 : HCCL_ERROR(
130 : "[CollAllGatherMixExecutor][KernelRun]AllGather mix memcpy Failed, Offset[%llu], Size[%llu]",
131 : baseOffset + level0Offset, inputMemSize),
132 : ret);
133 : } else {
134 0 : opInfoPtr = &opInfo;
135 : // 先做server间算法，带有消减拷贝场景数据需要从user input取，拷贝到ccl output上
136 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8*>(execMem.inputPtr), inputMemSize);
137 0 : ret = HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, const_cast<Stream&>(param.stream));
138 0 : CHK_PRT_RET(
139 : ret != HCCL_SUCCESS,
140 : HCCL_ERROR(
141 : "[CollAllGatherMixExecutor][KernelRun]AllGather mix user memcpy "
142 : "Failed, Offset[%llu], Size[%llu]",
143 : baseOffset + level0Offset, inputMemSize),
144 : ret);
145 0 : }
146 :
147 : // 第一步，AI server间AllGather
148 0 : std::unique_ptr<AlgTemplateBase> level1Executor;
149 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
150 : level1Executor
151 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
152 0 : HCCL_INFO("[CollAllGatherMixExecutor][KernelRun]AllGather mix: using ring algo inter-server.");
153 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
154 : level1Executor
155 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
156 0 : HCCL_INFO("[CollAllGatherMixExecutor][KernelRun]AllGather mix: using nhr algo inter-server.");
157 : } else {
158 0 : HCCL_ERROR(
159 : "[CollAllGatherMixExecutor][KernelRun]AllGather mix: algType_[%u] is not supported.", algType_.algoLevel1);
160 0 : return HCCL_E_NOT_SUPPORT;
161 : }
162 0 : CHK_SMART_PTR_NULL(level1Executor);
163 :
164 : // 计算slice
165 0 : std::vector<Slice> level1DataSegsSlice;
166 0 : for (u32 i = 0; i < level1RankSize; i++) {
167 0 : Slice level1Slice;
168 0 : level1Slice.size = inputMemSize;
169 0 : level1Slice.offset = (i * level0RankSize + commIndex) * inputMemSize;
170 0 : level1DataSegsSlice.push_back(level1Slice);
171 : }
172 0 : CHK_RET(level1Executor->Prepare(
173 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
174 : HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level1DataSegsSlice, 0));
175 :
176 0 : CHK_RET(level1Executor->RegisterProfiler(
177 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + serverIndex, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET,
178 : param.stream));
179 :
180 0 : CHK_RET(RunTemplate(level1Executor, level1CommInfo));
181 0 : HCCL_INFO("[CollAllGatherMixExecutor][KernelRun]AllGather mix level1 AllGather run success");
182 :
183 : // 第二步，节点内做AllGather mesh/dbring
184 0 : if (topoAttr_.deviceType == DevType::DEV_TYPE_910_93) {
185 0 : std::vector<Slice> dataSegsSlice; // 数据分成ranksize份，每份的起始偏移和大小
186 0 : CHK_RET(PrepareAllgatherSlice(level0RankSize, inputMemSize, dataSegsSlice));
187 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
188 : // 多环数据切分
189 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
190 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
191 : } else {
192 0 : multRingsSliceZero.push_back(dataSegsSlice);
193 : }
194 0 : std::vector<std::vector<Slice>> multRingsSlice;
195 0 : CHK_RET(CalculateLevel1AllgatherSlice(
196 : inputMemSize, level0RankSize, level1RankSize, multRingsSliceZero, multRingsSlice));
197 :
198 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
199 0 : if (!DMAReduceFlag_) {
200 0 : multRingsUserMemSlice = multRingsSlice;
201 : } else {
202 0 : for (u32 ringIndex = 0; ringIndex < multRingsSlice.size(); ringIndex++) {
203 0 : std::vector<Slice> userMemSlice;
204 0 : for (auto& cclSlice : multRingsSlice[ringIndex]) {
205 0 : Slice tmpSlice;
206 0 : tmpSlice.size = cclSlice.size;
207 0 : tmpSlice.offset = (cclSlice.offset / inputMemSize) * opInfo.count * perDataSize
208 0 : + multRingsSliceZero[ringIndex][0].offset;
209 0 : userMemSlice.push_back(tmpSlice);
210 : }
211 0 : multRingsUserMemSlice.push_back(userMemSlice);
212 0 : }
213 : }
214 0 : CHK_RET(ActiveSlaveStreams(param.stream));
215 0 : if (DMAReduceFlag_) {
216 : // allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
217 0 : opInfo.inputAddr = nullptr;
218 : }
219 0 : CHK_RET(MultiRingAllGather(
220 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, multRingsSlice,
221 : param.stream, PROF_STAGE_1, 0, opInfoPtr, multRingsUserMemSlice));
222 0 : } else if (topoAttr_.deviceType == DevType::DEV_TYPE_910B) {
223 0 : CHK_RET(ActiveSlaveStreams(param.stream));
224 :
225 : std::unique_ptr<AlgTemplateBase> level0Executor
226 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_MESH_MIX, dispatcher_);
227 0 : CHK_SMART_PTR_NULL(level0Executor);
228 0 : CHK_RET(level0Executor->Prepare(
229 : algResResp_->slaveStreams, algResResp_->notifiesMain, algResResp_->notifiesAux, 0, opInfoPtr, serverIndex,
230 : level1RankSize));
231 0 : std::vector<Slice> emptySlices;
232 0 : CHK_RET(level0Executor->Prepare(
233 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, param.stream,
234 : HCCL_REDUCE_RESERVED, LEVEL0_BRIDGE_RANK_ID, emptySlices, 0));
235 0 : CHK_RET(level0Executor->RegisterProfiler(
236 : (level0RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + commIndex, PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET,
237 : param.stream));
238 0 : CHK_RET(RunTemplate(level0Executor, level0CommInfo));
239 0 : }
240 :
241 0 : HCCL_INFO("[CollAllGatherMixExecutor][KernelRun]AllGather mix run success");
242 0 : return HCCL_SUCCESS;
243 0 : }
244 :
245 : REGISTER_EXEC("AllGatherMixExecutor", AllGatherMix, CollAllGatherMixExecutor);
246 : } // namespace hccl
|
