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_to_all_staged_aiv_rdma_executor.h"
12 :
13 : namespace hccl {
14 : constexpr u32 A_X_SIZE = 16;
15 :
16 0 : CollRunAlltoAllStagedAivRdmaExecutor::CollRunAlltoAllStagedAivRdmaExecutor(
17 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
18 0 : : CollAlltoAllExecutor(dispatcher, topoMatcher)
19 : {
20 0 : desc_.isAivMode = true;
21 0 : }
22 :
23 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
24 : {
25 0 : HcclUs startut = TIME_NOW();
26 0 : tag_ = param.tag;
27 0 : algResResp_ = &algRes;
28 :
29 0 : ExecMem execMem;
30 0 : execMem.count = 0;
31 0 : execMem.inputPtr = param.inputPtr;
32 0 : execMem.outputPtr = param.outputPtr;
33 :
34 : // alltoall aiv暂不支持图模式
35 0 : execMem.inputMem = algRes.cclInputMem;
36 0 : execMem.outputMem = algRes.cclOutputMem;
37 0 : execMem.scratchMem = algRes.aivInputMem;
38 0 : HcclResult ret = KernelRun(param, execMem);
39 0 : CHK_PRT_RET(
40 : ret != HCCL_SUCCESS,
41 : HCCL_ERROR(
42 : "[CollRunAlltoAllStagedAivRdmaExecutor][Orchestrate]errNo[0x%016llx]executor run failed",
43 : HCCL_ERROR_CODE(ret)),
44 : ret);
45 :
46 0 : HCCL_INFO(
47 : "[CollRunAlltoAllStagedAivRdmaExecutor]tag[%s], orchestrate success, take time [%lld]us.", param.tag.c_str(),
48 : DURATION_US(TIME_NOW() - startut));
49 :
50 0 : return HCCL_SUCCESS;
51 0 : }
52 :
53 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::CalcStreamNum(u32& streamNum)
54 : {
55 0 : streamNum = 0; // AIV通信不需要申请从流
56 :
57 0 : HCCL_INFO("[CollRunAlltoAllStagedAivRdmaExecutor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
58 0 : return HCCL_SUCCESS;
59 : }
60 :
61 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::CalcScratchMemSize(u64& scratchMemSize)
62 : {
63 0 : scratchMemSize = 0U; // AIV模式不需要scratch内存，直接在cclbuffer上进行内存重排
64 0 : HCCL_INFO(
65 : "[CollRunAlltoAllStagedAivRdmaExecutor][CalcScratchMemSize]tag[%s] scratchMemSize_ is [%llu]", tag_.c_str(),
66 : scratchMemSize);
67 0 : return HCCL_SUCCESS;
68 : }
69 :
70 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::CalcLevel0CommInfo(
71 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
72 : {
73 0 : CommParaInfo commParaLevel0(COMM_MESH_L0, CommType::COMM_TAG_MESH);
74 0 : commParaLevel0.meshSinglePlane = true;
75 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_MESH_L0], inputType, outputType));
76 0 : return HCCL_SUCCESS;
77 0 : }
78 :
79 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::CalcLevel1CommInfo(
80 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
81 : {
82 0 : CommParaInfo commParaInfo(COMM_MESH_L1, CommType::COMM_TAG_MESH);
83 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_MESH_L1], inputType, outputType));
84 0 : return HCCL_SUCCESS;
85 0 : }
86 :
87 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
88 : {
89 : // aiv阶段使用cclin做数据搬移，使用aivout做标记
90 0 : CHK_RET(CalcLevel0CommInfo(TransportMemType::CCL_INPUT, TransportMemType::AIV_INPUT, opTransport));
91 0 : CHK_RET(CalcLevel1CommInfo(TransportMemType::CCL_INPUT, TransportMemType::CCL_OUTPUT, opTransport));
92 :
93 0 : HCCL_DEBUG("[CollRunAlltoAllStagedAivRdmaExecutor][CalcCommInfo] ends");
94 0 : return HCCL_SUCCESS;
95 : }
96 :
97 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::CalNumBlocks(
98 : u32& numBlocks, u32 rankSize, [[maybe_unused]] u64 dataSize, [[maybe_unused]] HcclCMDType cmdType)
99 : {
100 0 : numBlocks = rankSize; // 默认情况使用rankSize个AIV
101 0 : u32 bestNumBlocks = numBlocks;
102 :
103 0 : CHK_PRT_RET(
104 : numBlocks_ < numBlocks,
105 : HCCL_WARNING(
106 : "[CollRunAlltoAllStagedAivRdmaExecutor][CalNumBlocks]aivCore[%u] is invalid, at least need [%u].",
107 : numBlocks_, numBlocks),
108 : HCCL_E_PARA);
109 :
110 0 : HCCL_INFO(
111 : "[CollRunAlltoAllStagedAivRdmaExecutor][CalNumBlocks] numBlocks is set to [%u], limit[%u], recommanded[%u]",
112 : numBlocks, numBlocks_, bestNumBlocks);
113 0 : return HCCL_SUCCESS;
114 : }
115 :
116 : // run aiv kernel
117 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::RunAlltoAllStaged1InAIV(const OpParam& param, ExecMem& execMem)
118 : {
119 : void* dataBuffers[MAX_RANK_SIZE];
120 : void* flagBuffers[MAX_RANK_SIZE];
121 :
122 0 : u32 serverNum = innerCommInfo_.localRankSize;
123 0 : u64 sendCount = *(static_cast<const u64*>(param.All2AllDataDes.sendCountMatrix));
124 :
125 0 : CHK_RET(PrepareAivBuffers(execMem.inputMem, execMem.scratchMem, dataBuffers, flagBuffers));
126 :
127 0 : AivOpArgs opArgs{
128 : HcclCMDType::HCCL_CMD_ALLTOALL,
129 0 : execMem.inputPtr,
130 0 : execMem.outputPtr,
131 : sendCount,
132 0 : param.All2AllDataDes.sendType,
133 : HCCL_REDUCE_RESERVED,
134 : 0,
135 0 : true};
136 : AivTopoArgs topoArgs{
137 : outerCommInfo_.localRank, outerCommInfo_.localRankSize,
138 0 : topoAttr_.isDiffDeviceModule ? topoAttr_.devicePhyId : A_X_SIZE, 0, serverNum};
139 0 : topoArgs.identify = algoAttr_.identifier;
140 : u32 numBlocks;
141 0 : CHK_PRT_RET(
142 : CalNumBlocks(numBlocks, outerCommInfo_.localRankSize) != HCCL_SUCCESS,
143 : HCCL_ERROR("[%s] CalNumBlocks failed", __func__), HCCL_E_PARA);
144 0 : numBlocks_ = numBlocks;
145 0 : AivResourceArgs resourceArgs{param.tag, param.stream.ptr(), dataBuffers, flagBuffers, execMem.inputMem.size(),
146 0 : numBlocks_, param.aivTag};
147 0 : AivAlgArgs algArgs{0};
148 0 : algArgs.execTimeOut = topoMatcher_->GetExecTimeOutConfig();
149 0 : algArgs.execTimeOutSet = true;
150 0 : struct AivProfilingInfo aivProfilingInfo;
151 0 : aivProfilingInfo.counter = opCounter_;
152 0 : HCCL_DEBUG("[CollRunAlltoAllStagedAivRdmaExecutor]RunAlltoAllStaged1InAIV for numBlocks is %u", numBlocks_);
153 0 : CHK_RET(ExecuteKernelLaunch(opArgs, topoArgs, resourceArgs, algArgs, aivProfilingInfo));
154 0 : return HCCL_SUCCESS;
155 0 : }
156 :
157 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::RunAlltoAllStaged2(const OpParam& param, ExecMem& execMem)
158 : {
159 0 : std::map<u32, std::list<OneSendRecvAddrInfo>> sendAddrInfosInter;
160 0 : std::map<u32, std::list<OneSendRecvAddrInfo>> recvAddrInfosInter;
161 :
162 0 : CalcInterMeshAggregationAlltoAllMemInfo(param, sendAddrInfosInter, recvAddrInfosInter);
163 :
164 : HcclOpMetaInfoDef opMeta
165 0 : = HcclOpMetaInfo::GetOneForAllToAll(CopyPattern::ZCOPY, algResResp_->paramInputMem.size(), false, true);
166 0 : CHK_RET(InitTask(dispatcher_, const_cast<Stream&>(param.stream), opMeta.isEnableCache, opMeta.GetCacheKey()));
167 :
168 0 : std::unique_ptr<AlgTemplateBase> alltoallInner = AlgTemplateRegistry::Instance().GetAlgTemplate(
169 0 : TemplateType::TEMPLATE_ALL_2_ALL_V_STAGED_PAIRWISE, dispatcher_);
170 0 : CHK_SMART_PTR_NULL(alltoallInner);
171 :
172 0 : CHK_RET(alltoallInner->Prepare(
173 : execMem.inputMem, execMem.outputMem, sendAddrInfosInter, recvAddrInfosInter, true,
174 : const_cast<Stream&>(param.stream)));
175 :
176 0 : CHK_RET(RunAlltoAllVTemplateStaged(alltoallInner, innerCommInfo_));
177 0 : return HCCL_SUCCESS;
178 0 : }
179 :
180 0 : void CollRunAlltoAllStagedAivRdmaExecutor::CalcInterMeshAggregationAlltoAllMemInfo(
181 : const OpParam& param, std::map<u32, std::list<OneSendRecvAddrInfo>>& sendAddrInfosInter,
182 : std::map<u32, std::list<OneSendRecvAddrInfo>>& recvAddrInfosInter)
183 : {
184 0 : u64 sendCount = *(static_cast<const u64*>(param.All2AllDataDes.sendCountMatrix));
185 : // serverLength表示每个rank给每个server需要发送的数据总量
186 0 : u64 serverSendLength = outerCommInfo_.localRankSize * sendCount * sendDataSize_;
187 0 : u64 serverRecvLength = outerCommInfo_.localRankSize * sendCount * recvDataSize_;
188 : // 数据中转时每个rank按server顺序存储对应的中转数据，使用userRankOffset表示根据当前rank所在server计算对端的偏移
189 0 : for (u32 i = 0; i < innerCommInfo_.localRankSize; i++) {
190 : OneSendRecvAddrInfo sendAddrInfo;
191 0 : sendAddrInfo.localOffset = i * serverSendLength;
192 0 : sendAddrInfo.remoteOffset = innerCommInfo_.localRank * serverSendLength;
193 0 : sendAddrInfo.localLength = serverSendLength;
194 0 : sendAddrInfo.remoteLength = serverSendLength;
195 0 : sendAddrInfosInter[i].push_back(sendAddrInfo);
196 :
197 : OneSendRecvAddrInfo recvAddrInfo;
198 0 : recvAddrInfo.localOffset = i * serverRecvLength;
199 0 : recvAddrInfo.remoteOffset = innerCommInfo_.localRank * serverRecvLength;
200 0 : recvAddrInfo.localLength = serverRecvLength;
201 0 : recvAddrInfo.remoteLength = serverRecvLength;
202 0 : recvAddrInfosInter[i].push_back(recvAddrInfo);
203 : }
204 0 : return;
205 : }
206 :
207 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::KernelRun(const OpParam& param, ExecMem& execMem)
208 : {
209 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollRunAlltoAllStagedAivRdmaExecutor][KernelRun] AllToAll staged starts");
210 0 : CHK_PRT_RET(
211 : topoAttr_.userRankSize % topoAttr_.meshAggregationRankSize != 0,
212 : HCCL_ERROR(
213 : "userRankSize[%u] is not an Integer multiple of MeshAggregation Dev Num[%u]", topoAttr_.userRankSize,
214 : topoAttr_.meshAggregationRankSize),
215 : HCCL_E_PARA);
216 :
217 0 : CHK_RET(SalGetDataTypeSize(param.All2AllDataDes.sendType, sendDataSize_));
218 0 : CHK_RET(SalGetDataTypeSize(param.All2AllDataDes.recvType, recvDataSize_));
219 :
220 0 : CHK_RET(CheckCommSize(COMM_MESH_L0, COMM_INDEX_0 + 1));
221 0 : outerCommInfo_ = GetSubCommInfo(COMM_MESH_L0, COMM_INDEX_0);
222 :
223 0 : CHK_RET(CheckCommSize(COMM_MESH_L1, COMM_INDEX_0 + 1));
224 0 : innerCommInfo_ = GetSubCommInfo(COMM_MESH_L1, COMM_INDEX_0);
225 :
226 : // step1：每个server内通过aiv进行数据交换，将中转数据存储在ipc_buffer中
227 0 : CHK_RET(RunAlltoAllStaged1InAIV(param, execMem));
228 0 : HCCL_INFO("[CollRunAlltoAllStagedAivRdmaExecutor][kernelRun] stage0 run aiv in level0 success!");
229 :
230 : // step2：server间通过rdma进行数据交换，将中转数据分发到各个rank的ccl_out中
231 0 : CHK_RET(RunAlltoAllStaged2(param, execMem));
232 0 : HCCL_INFO("[CollRunAlltoAllStagedAivRdmaExecutor][kernelRun] stage1 run rdma in level1 success!");
233 :
234 : // 每个rank上将数据从ccl_out中搬运到用户输出buffer中
235 0 : DeviceMem srcMem = (execMem.outputMem).range(0, algResResp_->paramOutputMem.size());
236 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, algResResp_->paramOutputMem, srcMem, const_cast<Stream&>(param.stream)));
237 :
238 0 : CHK_RET(LaunchTask(dispatcher_, const_cast<Stream&>(param.stream)));
239 0 : HCCL_INFO("[CollRunAlltoAllStagedAivRdmaExecutor][kernelRun] AllToAll staged ends");
240 0 : return HCCL_SUCCESS;
241 0 : }
242 :
243 0 : HcclResult CollRunAlltoAllStagedAivRdmaExecutor::PrepareAivBuffers(
244 : DeviceMem& inputMem, DeviceMem& outputMem, void** dataBuffers, void** flagBuffers)
245 : {
246 0 : void* tmpCCLBufferData = nullptr;
247 0 : void* tmpCCLBufferFlag = nullptr;
248 0 : for (u32 i = 0; i < outerCommInfo_.localRankSize; i++) {
249 0 : if (i != outerCommInfo_.localRank) {
250 0 : if (outerCommInfo_.links[i] != nullptr) {
251 0 : CHK_RET(outerCommInfo_.links[i]->GetRemoteMem(UserMemType::INPUT_MEM, &(tmpCCLBufferData)));
252 0 : CHK_RET(outerCommInfo_.links[i]->GetRemoteMem(UserMemType::OUTPUT_MEM, &(tmpCCLBufferFlag)));
253 : // cclbuffer后32K数据作为数据标志位
254 0 : dataBuffers[i] = static_cast<u8*>(tmpCCLBufferData);
255 0 : flagBuffers[i] = static_cast<u8*>(tmpCCLBufferFlag) + HCCL_MID_COUNT_32_MB;
256 : }
257 : } else {
258 0 : dataBuffers[i] = static_cast<u8*>(inputMem.ptr());
259 0 : flagBuffers[i] = static_cast<u8*>(outputMem.ptr()) + HCCL_MID_COUNT_32_MB;
260 : }
261 : }
262 0 : return HCCL_SUCCESS;
263 : }
264 :
265 : REGISTER_EXEC("AlltoAllStagedAIVRdmaExecutor", AlltoAllStagedAIVRdma, CollRunAlltoAllStagedAivRdmaExecutor);
266 : } // namespace hccl
|
