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