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_v_staged_executor.h"
12 : namespace hccl {
13 :
14 0 : CollRunAlltoAllVStaged::CollRunAlltoAllVStaged(const HcclDispatcher dispatcher,
15 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
16 0 : : CollAlltoAllExecutor(dispatcher, topoMatcher)
17 : {
18 0 : }
19 :
20 0 : HcclResult CollRunAlltoAllVStaged::ParallelTaskLoaderProcess(const std::string &tag, Stream &stream,
21 : SubCommInfo &level0CommInfo, std::vector<Stream> &ringStreams)
22 : {
23 : (void) tag;
24 : u32 streamIndex;
25 0 : std::vector<Stream *> streamsPtr;
26 0 : streamsPtr.resize(ringStreams.size() + 1);
27 :
28 0 : for (streamIndex = 0; streamIndex < ringStreams.size(); streamIndex++) { // slaveStreams
29 0 : streamsPtr[streamIndex] = &ringStreams[streamIndex];
30 : }
31 0 : streamsPtr[streamIndex] = &stream;
32 :
33 0 : HCCL_INFO("[ParallelTaskLoaderProcess]main stream[%p], streams size[%zu]", stream.ptr(), streamsPtr.size());
34 :
35 : // 准备多线程启动参数
36 0 : CHK_RET(parallelTaskLoader_->Prepare(streamsPtr, level0CommInfo));
37 :
38 : // 启动多线程处理
39 0 : CHK_RET(parallelTaskLoader_->StartTaskLoad());
40 :
41 : // 等待多线程处理结果
42 0 : CHK_RET(parallelTaskLoader_->WaitTaskLoadFinish());
43 :
44 : // 销毁通信域
45 0 : CHK_RET(parallelTaskLoader_->ClearTagCommInfo());
46 0 : return HCCL_SUCCESS;
47 0 : }
48 :
49 0 : HcclResult CollRunAlltoAllVStaged::CalcStreamNum(u32& streamNum)
50 : {
51 0 : streamNum = 0U;
52 0 : if (FullmeshPairwiseSatisfyHighPerfAlltoallMeshCondition(topoAttr_.deviceType,
53 0 : topoAttr_.userRankSize, topoAttr_.useSuperPodMode, topoMatcher_->GetAlgoConfig(HcclCMDType::HCCL_CMD_ALLTOALL))) {
54 0 : streamNum = topoAttr_.meshAggregationRankSize - 1;
55 : } else {
56 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE || isAlltoAllZCopyMode_) {
57 0 : std::vector<HcclAlgoType> algoTypeArr = topoMatcher_->GetAlgoConfig();
58 0 : if ((algoTypeArr[HCCL_ALGO_LEVEL_0] != HcclAlgoType::HCCL_ALGO_TYPE_PAIRWISE ||
59 0 : algoTypeArr[HCCL_ALGO_LEVEL_1] != HcclAlgoType::HCCL_ALGO_TYPE_PAIRWISE) &&
60 0 : const_cast<HcclTopoInfo &>(topoAttr_).pairLinkCounter[static_cast<u32>(
61 0 : LinkTypeInServer::HCCS_SW_TYPE)] == 0 && topoAttr_.meshAggregationRankSize != 1) {
62 0 : streamNum = topoAttr_.meshAggregationRankSize - MINORS_NUM_TWO;
63 : }
64 0 : }
65 : }
66 :
67 0 : HCCL_INFO("[CollRunAlltoAllVStaged][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 0 : void CollRunAlltoAllVStaged::CalcWorkSpaceMemSize(const AlltoAllUserRankInfo &userRankInfo,
72 : const std::vector<SendRecvInfo> &allMeshAggregationSendRecvInfo, u64 &workspaceMemSize,
73 : u32 meshAggregationRankSize)
74 : {
75 0 : if (allMeshAggregationSendRecvInfo.size() % meshAggregationRankSize != 0 ||
76 0 : allMeshAggregationSendRecvInfo.size() == 0) {
77 0 : workspaceMemSize = 0;
78 0 : HCCL_ERROR("Invalid Send Recv Info Size[%zu]", allMeshAggregationSendRecvInfo.size());
79 0 : return;
80 : }
81 0 : workspaceMemSize = 0;
82 0 : u32 meshAggregationIndex = userRankInfo.userRank / meshAggregationRankSize;
83 0 : u32 meshAggregationRankBegin = meshAggregationIndex * meshAggregationRankSize;
84 0 : for (u32 infoIndex = userRankInfo.userRank % meshAggregationRankSize; infoIndex < userRankInfo.userRankSize;
85 0 : infoIndex += meshAggregationRankSize) {
86 0 : for (u32 k = meshAggregationRankBegin; k < meshAggregationRankBegin + meshAggregationRankSize; k++) {
87 0 : workspaceMemSize += allMeshAggregationSendRecvInfo[k].sendLength[infoIndex];
88 : }
89 : }
90 0 : HCCL_INFO("[AlltoAllVStagedCalculator][CalcWorkSpaceMemSize] workspaceMemSize[%llu]", workspaceMemSize);
91 : }
92 :
93 0 : HcclResult CollRunAlltoAllVStaged::CalcScratchMemSize(u64& scratchMemSize)
94 : {
95 0 : scratchMemSize = 0U;
96 0 : u64 maxWorkSpaceMemSize = 0;
97 : AlltoAllUserRankInfo tmpUserRankInfo;
98 0 : tmpUserRankInfo.userRankSize = topoAttr_.userRankSize;
99 0 : tmpUserRankInfo.userRank = INVALID_VALUE_RANKID;
100 :
101 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
102 0 : u64 workSpaceMemSize = 0;
103 0 : tmpUserRankInfo.userRank = topoAttr_.userRank;
104 0 : CalcWorkSpaceMemSize(tmpUserRankInfo, allMeshAggregationSendRecvInfo_, workSpaceMemSize,
105 0 : topoAttr_.meshAggregationRankSize);
106 0 : scratchMemSize = CalAlltoAllVScratchMemSize(workSpaceMemSize);
107 : } else {
108 0 : if(isAlltoAllZCopyMode_ && topoAttr_.isSingleMeshAggregation) {
109 0 : scratchMemSize = 0;
110 0 : HCCL_INFO("[CollRunAlltoAllVStaged][CalcScratchMemSize] scratchMemSize[%llu]", scratchMemSize);
111 0 : return HCCL_SUCCESS;
112 : }
113 0 : for (u32 rank = 0; rank < topoAttr_.userRankSize; rank++) {
114 0 : u64 workSpaceMemSize = 0;
115 0 : tmpUserRankInfo.userRank = rank;
116 0 : CalcWorkSpaceMemSize(tmpUserRankInfo, allMeshAggregationSendRecvInfo_, workSpaceMemSize,
117 0 : topoAttr_.meshAggregationRankSize);
118 0 : maxWorkSpaceMemSize = std::max(workSpaceMemSize, maxWorkSpaceMemSize);
119 : }
120 0 : scratchMemSize = CalAlltoAllVScratchMemSize(maxWorkSpaceMemSize);
121 0 : HCCL_DEBUG("[CollRunAlltoAllVStaged][CalcScratchMemSize] OpBase branch, "
122 : "maxWorkSpaceMemSize[%llu], scratchMemSize[%llu]",
123 : maxWorkSpaceMemSize, scratchMemSize);
124 : }
125 :
126 0 : HCCL_INFO("[CollRunAlltoAllVStaged][CalcScratchMemSize] workflowMode[%d], scratchMemSize[%llu]",
127 : workflowMode_, scratchMemSize);
128 0 : return HCCL_SUCCESS;
129 : }
130 :
131 0 : HcclResult CollRunAlltoAllVStaged::CheckNeedRecreateComm(u64 lastScratchMemSize, bool& needRecreateAlltoallComm)
132 : {
133 0 : u64 tmpScratchMemSize = 0;
134 0 : CHK_RET(CalcScratchMemSize(tmpScratchMemSize));
135 0 : needRecreateAlltoallComm = ((lastScratchMemSize < tmpScratchMemSize) ? (true) : (false));
136 0 : return HCCL_SUCCESS;
137 : }
138 :
139 0 : HcclResult CollRunAlltoAllVStaged::CheckNeedCreateVirtualLinks(AlgResourceRequest &resourceRequest)
140 : {
141 0 : bool alltoallMeshReadOnly = FullmeshPairwiseSatisfyHighPerfAlltoallMeshCondition(topoAttr_.deviceType,
142 0 : topoAttr_.userRankSize, topoAttr_.useSuperPodMode, topoMatcher_->GetAlgoConfig(HcclCMDType::HCCL_CMD_ALLTOALL));
143 0 : HCCL_DEBUG("[CollRunAlltoAllVStaged][CheckNeedCreateVirtualLinks] AllToAllVMeshReadOnly[%d]," \
144 : "resourceRequest.streamNum[%u], GetExternalInputHcclEnableFfts()[%d], isAlltoAllZCopyMode_[%d]",
145 : alltoallMeshReadOnly, resourceRequest.streamNum, GetExternalInputHcclEnableFfts(), isAlltoAllZCopyMode_);
146 0 : if (!alltoallMeshReadOnly && (resourceRequest.streamNum != 0) && (!GetExternalInputHcclEnableFfts())
147 0 : && isAlltoAllZCopyMode_) {
148 0 : for (auto &levelNSubCommTransport : resourceRequest.opTransport) {
149 0 : for (auto &singleSubCommTransport : levelNSubCommTransport) {
150 0 : singleSubCommTransport.needVirtualLink = true;
151 0 : HCCL_INFO("[CollRunAlltoAllVStaged][CheckNeedCreateVirtualLinks] needVirtualLink is true");
152 : }
153 : }
154 : }
155 0 : return HCCL_SUCCESS;
156 : }
157 :
158 0 : HcclResult CollRunAlltoAllVStaged::CalcLevel0CommInfo(TransportMemType inputType, TransportMemType outputType,
159 : std::vector<LevelNSubCommTransport>& opTransport)
160 : {
161 0 : CommParaInfo commParaLevel0(COMM_MESH_L0, CommType::COMM_TAG_MESH);
162 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_MESH_L0], inputType, outputType));
163 0 : return HCCL_SUCCESS;
164 0 : }
165 :
166 0 : HcclResult CollRunAlltoAllVStaged::CalcLevel1CommInfo(TransportMemType inputType,
167 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
168 : {
169 0 : CommParaInfo commParaInfo(COMM_MESH_L1, CommType::COMM_TAG_MESH);
170 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_MESH_L1], inputType, outputType));
171 0 : return HCCL_SUCCESS;
172 0 : }
173 :
174 0 : HcclResult CollRunAlltoAllVStaged::CalcLevel2CommInfo(TransportMemType inputType,
175 : TransportMemType outputType,
176 : std::vector<LevelNSubCommTransport>& opTransport)
177 : {
178 0 : CommParaInfo commParaLevel2(COMM_LEVEL2, CommType::COMM_TAG_MESH);
179 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel2, opTransport[COMM_LEVEL2], inputType, outputType));
180 0 : return HCCL_SUCCESS;
181 0 : }
182 :
183 0 : HcclResult CollRunAlltoAllVStaged::CalStagedAlltoallVCommInfo(TransportMemType inputType,
184 : TransportMemType outputType,
185 : std::vector<LevelNSubCommTransport>& opTransport)
186 : {
187 : (void) inputType;
188 : (void) outputType;
189 : // 将网卡初始化判断，提到上层调用，减少无必要的循环依赖。
190 0 : bool alltoallMeshReadOnly = FullmeshPairwiseSatisfyHighPerfAlltoallMeshCondition(topoAttr_.deviceType,
191 0 : topoAttr_.userRankSize, topoAttr_.useSuperPodMode, topoMatcher_->GetAlgoConfig(HcclCMDType::HCCL_CMD_ALLTOALL));
192 :
193 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
194 0 : !isAlltoAllZCopyMode_) { // 单算子 && BCopy模式
195 0 : HCCL_INFO("cal comm in opbase and Bcopy mode");
196 0 : CHK_RET(CalcLevel0CommInfo(TransportMemType::CCL_INPUT, TransportMemType::CCL_OUTPUT, opTransport));
197 0 : CHK_RET(CalcLevel1CommInfo(TransportMemType::CCL_INPUT, TransportMemType::CCL_OUTPUT, opTransport));
198 0 : CHK_RET(CalcLevel2CommInfo(TransportMemType::CCL_INPUT, TransportMemType::CCL_OUTPUT, opTransport));
199 0 : } else if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
200 0 : isAlltoAllZCopyMode_) { // 单算子 && ZCopy模式
201 0 : HCCL_INFO("cal comm in opbase and Zcopy mode");
202 0 : if (topoAttr_.isSingleMeshAggregation) {
203 0 : CHK_RET(CalcLevel0CommInfo(TransportMemType::CCL_INPUT, TransportMemType::CCL_OUTPUT, opTransport));
204 : } else {
205 0 : CHK_RET(CalcLevel0CommInfo(TransportMemType::CCL_INPUT, (alltoallMeshReadOnly ?
206 : TransportMemType::CCL_OUTPUT : TransportMemType::SCRATCH), opTransport));
207 0 : CHK_RET(CalcLevel1CommInfo(TransportMemType::SCRATCH, TransportMemType::CCL_OUTPUT, opTransport));
208 : }
209 0 : CHK_RET(CalcLevel2CommInfo(TransportMemType::CCL_INPUT, TransportMemType::CCL_OUTPUT, opTransport));
210 0 : } else {
211 0 : HCCL_INFO("cal comm in graph mode");
212 0 : CHK_RET(CalcLevel0CommInfo(TransportMemType::PARAM_INPUT, TransportMemType::SCRATCH, opTransport));
213 0 : CHK_RET(CalcLevel1CommInfo(TransportMemType::SCRATCH, TransportMemType::PARAM_OUTPUT, opTransport));
214 0 : CHK_RET(CalcLevel2CommInfo(TransportMemType::PARAM_INPUT, TransportMemType::PARAM_OUTPUT, opTransport));
215 : }
216 0 : HCCL_DEBUG("[CollRunAlltoAllVStaged][CalStagedAlltoallVCommInfo] ends");
217 0 : return HCCL_SUCCESS;
218 : }
219 :
220 0 : HcclResult CollRunAlltoAllVStaged::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
221 : {
222 0 : TransportMemType inputType = TransportMemType::RESERVED;
223 0 : TransportMemType outputType = TransportMemType::RESERVED;
224 :
225 0 : CHK_RET(CalStagedAlltoallVCommInfo(inputType, outputType, opTransport));
226 0 : return HCCL_SUCCESS;
227 : }
228 :
229 0 : HcclResult CollRunAlltoAllVStaged::PrepareAlltoAllVStaged1(DeviceMem &sendBuf, DeviceMem &recvBuf, DeviceMem &scratchMem,
230 : std::map<u32, std::list<OneSendRecvAddrInfo>> &sendAddrInfosIntra,
231 : std::map<u32, std::list<OneSendRecvAddrInfo>> &recvAddrInfosIntra,
232 : Stream &stream, const std::string &tag, std::unique_ptr<AlgTemplateBase> &alltoallLevel0,
233 : ExecMem &execMem)
234 : {
235 : (void) tag;
236 : // opbase BCopy 不支持fullmesh算法，因此不必做算法选择
237 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
238 0 : !isAlltoAllZCopyMode_) { // 单算子 && Buffer拷贝模式
239 0 : HCCL_INFO("Running AllToAllV Staged Pairwise intra Server");
240 0 : alltoallLevel0 = AlgTemplateRegistry::Instance().GetAlgTemplate(
241 0 : TemplateType::TEMPLATE_ALL_2_ALL_V_STAGED_PAIRWISE, dispatcher_);
242 0 : CHK_SMART_PTR_NULL(alltoallLevel0);
243 0 : CHK_RET(alltoallLevel0->Prepare(sendBuf, scratchMem, execMem.inputMem, execMem.outputMem, sendAddrInfosIntra,
244 : recvAddrInfosIntra, isAlltoAllZCopyMode_, stream));
245 0 : } else {
246 0 : bool isOpBaseZCopy = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE && isAlltoAllZCopyMode_;
247 0 : DeviceMem inBuf = (isOpBaseZCopy) ? execMem.inputMem : sendBuf;
248 : // 单MeshAggregation下, 分级算法不做第二级, 结果输出到outCCLbuffer_
249 0 : DeviceMem outBuf = (isOpBaseZCopy && topoAttr_.isSingleMeshAggregation) ? recvBuf : scratchMem;
250 : // opbase ZCopy 与 graph，除input buffer差异外，其余行为应保持一致
251 0 : if (isOpBaseZCopy) { // 单算子 && ZCopy模式
252 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, execMem.inputMem, sendBuf, stream));
253 : }
254 0 : std::vector<HcclAlgoType> algoTypeArr = topoMatcher_->GetAlgoConfig();
255 : // 互联场景, alltoall暂不支持走fullmesh+pairwise
256 0 : if ((algoTypeArr[HCCL_ALGO_LEVEL_0] == HcclAlgoType::HCCL_ALGO_TYPE_PAIRWISE &&
257 0 : algoTypeArr[HCCL_ALGO_LEVEL_1] == HcclAlgoType::HCCL_ALGO_TYPE_PAIRWISE) ||
258 0 : const_cast<HcclTopoInfo &>(topoAttr_).pairLinkCounter[static_cast<u32>(LinkTypeInServer::HCCS_SW_TYPE)] != 0 ||
259 0 : topoAttr_.meshAggregationRankSize == 1) {
260 0 : HCCL_INFO("Running AllToAllV Staged Pairwise intra Server");
261 0 : alltoallLevel0 = AlgTemplateRegistry::Instance().GetAlgTemplate(
262 0 : TemplateType::TEMPLATE_ALL_2_ALL_V_STAGED_PAIRWISE, dispatcher_);
263 0 : CHK_SMART_PTR_NULL(alltoallLevel0);
264 0 : CHK_RET(alltoallLevel0->Prepare(inBuf, outBuf, sendAddrInfosIntra, recvAddrInfosIntra,
265 : isAlltoAllZCopyMode_, stream));
266 : } else {
267 0 : HCCL_INFO("Running AllToAllV Staged Mesh intra Server");
268 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
269 0 : CHK_RET(ActiveSlaveStreams(AlltoAllVParam_.stream));
270 : }
271 0 : if (GetExternalInputHcclEnableFfts() || algResResp_->slaveStreams.size() == 0) {
272 0 : alltoallLevel0 = AlgTemplateRegistry::Instance().GetAlgTemplate(
273 0 : TemplateType::TEMPLATE_ALL_2_ALL_V_STAGED_MESH, dispatcher_);
274 : } else {
275 0 : alltoallLevel0 = AlgTemplateRegistry::Instance().GetAlgTemplate(
276 0 : TemplateType::TEMPLATE_ALL_2_ALL_V_STAGED_MESH, vDispatcher_);
277 : }
278 0 : CHK_SMART_PTR_NULL(alltoallLevel0);
279 0 : CHK_RET(alltoallLevel0->Prepare(inBuf, outBuf, sendAddrInfosIntra, recvAddrInfosIntra,
280 : isAlltoAllZCopyMode_, topoAttr_.userRank, stream, algResResp_->slaveStreams,
281 : algResResp_->notifiesMain, algResResp_->notifiesAux));
282 : }
283 0 : }
284 0 : return HCCL_SUCCESS;
285 : }
286 :
287 0 : void CollRunAlltoAllVStaged::CalcInterMeshAggregationRecvRemoteOffset(const AlltoAllUserRankInfo &userRankInfo,
288 : const std::vector<SendRecvInfo> &allSendRecvInfo, u32 index, u64 &remoteOffset, u32 meshAggregationRankSize)
289 : {
290 : // 对于stage1 来说，相当于是从rand index 发送给 userRankInfo.userRank, 然后计算这种情况下的stage1 的接收偏移
291 0 : remoteOffset = 0;
292 0 : u32 anchoruserRank_ = index;
293 0 : u32 anchorIndex = userRankInfo.userRank;
294 0 : u32 beginIndex = anchorIndex % meshAggregationRankSize;
295 0 : u32 beginRank = anchoruserRank_ / meshAggregationRankSize * meshAggregationRankSize;
296 0 : bool getAnchor = false;
297 0 : for (index = beginIndex; index <= anchorIndex; index += meshAggregationRankSize) {
298 0 : for (u32 rank = beginRank; rank < beginRank + meshAggregationRankSize; rank++) {
299 0 : if (index == anchorIndex && rank == anchoruserRank_) {
300 0 : getAnchor = true;
301 0 : break;
302 : }
303 0 : remoteOffset += allSendRecvInfo[rank].sendLength[index];
304 : }
305 0 : if (getAnchor) {
306 0 : break;
307 : }
308 : }
309 0 : }
310 :
311 0 : void CollRunAlltoAllVStaged::CalcInterMeshAggregationAlltoAllMemInfo(
312 : const AlltoAllUserRankInfo &userRankInfo, const std::vector<SendRecvInfo> &allSendRecvInfo,
313 : std::map<u32, std::list<OneSendRecvAddrInfo>> &sendAddrInfosInter,
314 : std::map<u32, std::list<OneSendRecvAddrInfo>> &recvAddrInfosInter,
315 : u32 meshAggregationRankSize)
316 : {
317 0 : sendAddrInfosInter.clear();
318 0 : recvAddrInfosInter.clear();
319 :
320 0 : u64 localOffsetMarker = 0;
321 0 : for (u32 toRank = 0; toRank < userRankInfo.userRankSize; toRank++) {
322 0 : u32 myRank = userRankInfo.userRank;
323 0 : u32 myMeshAggregationRankBegin = myRank / meshAggregationRankSize * meshAggregationRankSize;
324 0 : u32 myMeshAggregationRankEnd = myMeshAggregationRankBegin + meshAggregationRankSize;
325 :
326 0 : for (u32 myMeshAggregationRank = myMeshAggregationRankBegin; myMeshAggregationRank < myMeshAggregationRankEnd;
327 : myMeshAggregationRank++) {
328 0 : if (toRank % meshAggregationRankSize == myRank % meshAggregationRankSize) {
329 : OneSendRecvAddrInfo sendAddrInfo;
330 0 : sendAddrInfo.localLength = allSendRecvInfo[myMeshAggregationRank].sendLength[toRank];
331 0 : sendAddrInfo.localOffset = localOffsetMarker;
332 0 : localOffsetMarker += sendAddrInfo.localLength;
333 0 : sendAddrInfo.remoteOffset = allSendRecvInfo[toRank].recvOffset[myMeshAggregationRank];
334 0 : sendAddrInfo.remoteLength = allSendRecvInfo[toRank].recvLength[myMeshAggregationRank];
335 0 : u32 remoteRankInter = toRank / meshAggregationRankSize;
336 0 : sendAddrInfosInter[remoteRankInter].push_back(sendAddrInfo);
337 0 : HCCL_DEBUG("[CalcInterMeshAggregationAlltoAllMemInfo] sendAddrInfo localOffset[%llu], "\
338 : "localLength[%llu], remoteOffset[%llu], remoteLength[%llu]", sendAddrInfo.localOffset,
339 : sendAddrInfo.localLength, sendAddrInfo.remoteOffset, sendAddrInfo.remoteLength);
340 : }
341 : }
342 : }
343 :
344 : // 构造接收数据结构
345 0 : for (u32 index = 0; index < userRankInfo.userRankSize; index++) {
346 : OneSendRecvAddrInfo recvAddrInfo;
347 0 : u32 meshAggregationIndex = index / meshAggregationRankSize;
348 :
349 0 : recvAddrInfo.localOffset = allSendRecvInfo[userRankInfo.userRank].recvOffset[index];
350 0 : recvAddrInfo.localLength = allSendRecvInfo[userRankInfo.userRank].recvLength[index];
351 : // index 是 从那个rank 来的
352 0 : recvAddrInfo.remoteLength = allSendRecvInfo[index].sendLength[userRankInfo.userRank];
353 0 : u64 remoteOffset = 0;
354 0 : CalcInterMeshAggregationRecvRemoteOffset(userRankInfo, allSendRecvInfo, index, remoteOffset,
355 : meshAggregationRankSize);
356 :
357 0 : recvAddrInfo.remoteOffset = remoteOffset;
358 0 : recvAddrInfosInter[meshAggregationIndex].push_back(recvAddrInfo);
359 0 : HCCL_DEBUG("[CalcInterMeshAggregationAlltoAllMemInfo] recvAddrInfo localOffset[%llu], "\
360 : "localLength[%llu], remoteOffset[%llu], remoteLength[%llu]", recvAddrInfo.localOffset,
361 : recvAddrInfo.localLength, recvAddrInfo.remoteOffset, recvAddrInfo.remoteLength);
362 : }
363 0 : }
364 :
365 0 : HcclResult CollRunAlltoAllVStaged::PrepareAlltoAllVStaged2(DeviceMem &recvBuf, DeviceMem &scratchMem,
366 : std::map<u32, std::list<OneSendRecvAddrInfo>> &sendAddrInfosInter,
367 : std::map<u32, std::list<OneSendRecvAddrInfo>> &recvAddrInfosInter,
368 : Stream &stream, const std::string &tag, std::unique_ptr<AlgTemplateBase> &alltoallLevel1,
369 : ExecMem &execMem)
370 : {
371 : (void) tag;
372 0 : alltoallLevel1 = AlgTemplateRegistry::Instance().GetAlgTemplate(
373 0 : TemplateType::TEMPLATE_ALL_2_ALL_V_STAGED_PAIRWISE, dispatcher_);
374 0 : CHK_SMART_PTR_NULL(alltoallLevel1);
375 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
376 0 : !isAlltoAllZCopyMode_) { // 单算子 && BCopy模式
377 0 : CHK_RET(alltoallLevel1->Prepare(scratchMem, recvBuf, execMem.inputMem, execMem.outputMem, sendAddrInfosInter,
378 : recvAddrInfosInter, isAlltoAllZCopyMode_, stream));
379 0 : } else if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
380 0 : isAlltoAllZCopyMode_) { // 单算子 && ZCopy模式
381 0 : CHK_RET(alltoallLevel1->Prepare(scratchMem, execMem.outputMem, execMem.inputMem, execMem.outputMem,
382 : sendAddrInfosInter, recvAddrInfosInter, isAlltoAllZCopyMode_, stream));
383 0 : } else {
384 0 : CHK_RET(alltoallLevel1->Prepare(scratchMem, recvBuf, sendAddrInfosInter, recvAddrInfosInter,
385 : isAlltoAllZCopyMode_, stream));
386 : }
387 0 : return HCCL_SUCCESS;
388 : }
389 :
390 0 : HcclResult CollRunAlltoAllVStaged::KernelRun(const OpParam ¶m, ExecMem &execMem)
391 : {
392 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollRunAlltoAllVStaged][KernelRun] AllToAllV staged starts");
393 0 : CHK_PRT_RET(topoAttr_.userRankSize % topoAttr_.meshAggregationRankSize != 0,
394 : HCCL_ERROR("userRankSize[%u] is not an Integer multiple of MeshAggregation Dev Num[%u]",
395 : topoAttr_.userRankSize, topoAttr_.meshAggregationRankSize), HCCL_E_PARA);
396 :
397 : AlltoAllUserRankInfo userRankInfo;
398 0 : userRankInfo.userRank = topoAttr_.userRank;
399 0 : userRankInfo.userRankSize = topoAttr_.userRankSize;
400 0 : bool alltoallMeshReadOnly = FullmeshPairwiseSatisfyHighPerfAlltoallMeshCondition(topoAttr_.deviceType,
401 0 : topoAttr_.userRankSize, topoAttr_.useSuperPodMode, topoMatcher_->GetAlgoConfig(HcclCMDType::HCCL_CMD_ALLTOALL));
402 :
403 0 : std::map<u32, std::list<OneSendRecvAddrInfo>> sendAddrInfosIntra;
404 0 : std::map<u32, std::list<OneSendRecvAddrInfo>> recvAddrInfosIntra;
405 0 : bool isSingleMesh = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
406 0 : isAlltoAllZCopyMode_ && topoAttr_.isSingleMeshAggregation;
407 0 : CalcIntraMeshAggregationAlltoAllMemInfo(userRankInfo, allMeshAggregationSendRecvInfo_, sendAddrInfosIntra,
408 0 : recvAddrInfosIntra, topoAttr_.meshAggregationRankSize, isSingleMesh);
409 :
410 0 : CHK_RET(CheckCommSize(COMM_MESH_L0, COMM_INDEX_0 + 1));
411 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_MESH_L0, COMM_INDEX_0);
412 :
413 0 : std::unique_ptr<AlgTemplateBase> alltoallLevel0 = nullptr;
414 0 : CHK_RET(PrepareAlltoAllVStaged1(algResResp_->paramInputMem, algResResp_->paramOutputMem, execMem.scratchMem,
415 : sendAddrInfosIntra, recvAddrInfosIntra, const_cast<Stream&>(param.stream), tag_, alltoallLevel0, execMem));
416 0 : if ((algResResp_->slaveStreams.size() != 0) &&
417 0 : (!GetExternalInputHcclEnableFfts()) && isAlltoAllZCopyMode_) {
418 0 : HCCL_INFO("[AlltoAllOperator][RunAlltoAllVStaged] staged 0 use parallel multi-thread delivery of tasks");
419 0 : CHK_RET(RunTemplateWithVirtualLink(alltoallLevel0, level0CommInfo));
420 : // 多流场景下，并行多线程下发task处理
421 0 : CHK_RET(ParallelTaskLoaderProcess(tag_, const_cast<Stream&>(param.stream), level0CommInfo,
422 : algResResp_->slaveStreams));
423 : } else {
424 0 : CHK_RET(RunAlltoAllVTemplateStaged(alltoallLevel0, level0CommInfo));
425 : }
426 :
427 0 : HCCL_INFO("[hcclImpl][RunAlltoAllVStaged] stage0 run success!");
428 :
429 0 : std::map<u32, std::list<OneSendRecvAddrInfo>> sendAddrInfosInter;
430 0 : std::map<u32, std::list<OneSendRecvAddrInfo>> recvAddrInfosInter;
431 0 : CalcInterMeshAggregationAlltoAllMemInfo(userRankInfo, allMeshAggregationSendRecvInfo_,sendAddrInfosInter,
432 0 : recvAddrInfosInter, topoAttr_.meshAggregationRankSize);
433 :
434 0 : if (((workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
435 0 : isAlltoAllZCopyMode_) || alltoallMeshReadOnly) && topoAttr_.isSingleMeshAggregation) {
436 0 : HCCL_DEBUG("we don't need to do stage 2 when there is only one mesh aggregation");
437 : // we don't need to do stage 2 when there is only one mesh aggregation
438 : } else {
439 0 : HCCL_INFO("[hcclImpl][RunAlltoAllVStaged] stage1 run starts!");
440 0 : CHK_RET(CheckCommSize(COMM_MESH_L1, COMM_INDEX_0 + 1));
441 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_MESH_L1, COMM_INDEX_0);
442 0 : std::unique_ptr<AlgTemplateBase> alltoallLevel1 = nullptr;
443 0 : CHK_RET(PrepareAlltoAllVStaged2(algResResp_->paramOutputMem, execMem.scratchMem, sendAddrInfosInter,
444 : recvAddrInfosInter, const_cast<Stream&>(param.stream), tag_, alltoallLevel1, execMem));
445 0 : CHK_RET(RunAlltoAllVTemplateStaged(alltoallLevel1, level1CommInfo));
446 0 : }
447 :
448 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
449 0 : isAlltoAllZCopyMode_ && !topoAttr_.isSingleMeshAggregation) {
450 0 : DeviceMem srcMem = (execMem.outputMem).range(0, algResResp_->paramOutputMem.size());
451 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, algResResp_->paramOutputMem, srcMem, const_cast<Stream&>(param.stream)));
452 0 : }
453 :
454 0 : HCCL_INFO("[CollRunAlltoAllVStaged][kernelRun] AllToAllV staged ends");
455 0 : return HCCL_SUCCESS;
456 0 : }
457 :
458 : REGISTER_EXEC("RunAlltoAllVStaged", AlltoAllVStaged, CollRunAlltoAllVStaged);
459 : } // namespace hccl
|
