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_symmetric_memory_executor.h"
12 :
13 : namespace hccl {
14 :
15 0 : CollRunAlltoAllFullMeshSymmetricMemory::CollRunAlltoAllFullMeshSymmetricMemory(
16 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
17 0 : : CollAlltoAllExecutor(dispatcher, topoMatcher)
18 : {
19 0 : desc_.isZeroCopy = true;
20 0 : }
21 :
22 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
23 : {
24 0 : HcclUs startut = TIME_NOW();
25 0 : HcclResult ret = HCCL_SUCCESS;
26 0 : tag_ = param.tag;
27 0 : algResResp_ = &algRes;
28 0 : AlltoAllVParam_ = param;
29 :
30 0 : ExecMem execMem;
31 0 : execMem.count = 0;
32 0 : execMem.inputPtr = param.inputPtr;
33 0 : execMem.outputPtr = param.outputPtr;
34 0 : execMem.inputMem = algRes.paramInputMem;
35 0 : execMem.outputMem = algRes.paramOutputMem;
36 0 : ret = KernelRun(param, execMem);
37 :
38 0 : CHK_PRT_RET(
39 : ret != HCCL_SUCCESS,
40 : HCCL_ERROR(
41 : "[CollRunAlltoAllFullMeshSymmetricMemory][Orchestrate]errNo[0x%016llx]executor run failed",
42 : HCCL_ERROR_CODE(ret)),
43 : ret);
44 :
45 0 : HCCL_INFO(
46 : "tag[%s], AlltoAllFullMeshSymmetricMemory tempAlg orchestrate success, take time [%lld]us.", param.tag.c_str(),
47 : DURATION_US(TIME_NOW() - startut));
48 0 : return HCCL_SUCCESS;
49 0 : }
50 :
51 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::GetLocalSDMAGroupInfo(u32& devNumInlocalPod, u32& rankIdxInPod) const
52 : {
53 0 : CHK_RET(topoMatcher_->GetLocalSuperPodRankSize(topoAttr_.userRank, devNumInlocalPod, rankIdxInPod));
54 0 : CHK_PRT_RET(
55 : devNumInlocalPod == INVALID_VALUE_RANKSIZE,
56 : HCCL_ERROR(
57 : "[CollRunAlltoAllFullMeshSymmetricMemory][GetLocalSDMAGroupInfo]get local superPod total ranksize failed."),
58 : HCCL_E_PARA);
59 0 : return HCCL_SUCCESS;
60 : }
61 :
62 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::CalcStreamNum(u32& streamNum)
63 : {
64 : // 每个超节点内的卡数
65 0 : u32 devNumInlocalPod = INVALID_VALUE_RANKSIZE;
66 0 : u32 rankIdxInPod = INVALID_VALUE_RANKID;
67 0 : CHK_RET(GetLocalSDMAGroupInfo(devNumInlocalPod, rankIdxInPod));
68 :
69 : // 单超节点场景需要的从流数量，待确认是否需要减去一条主流
70 0 : streamNum = (devNumInlocalPod > ALLTOALLV_DIRECT_FULLMESH_SDMA_CONCURRENT_SIZE) ?
71 : (ALLTOALLV_DIRECT_FULLMESH_SDMA_CONCURRENT_SIZE) :
72 0 : (devNumInlocalPod);
73 :
74 0 : HCCL_INFO("[CollRunAlltoAllFullMeshSymmetricMemory][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
75 0 : return HCCL_SUCCESS;
76 : }
77 :
78 : // level0-level1 打平fullmesh
79 : // 超节点内建SDMA链路；超节点间建RDMA链路
80 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::CalcLevel0CommInfo(
81 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
82 : {
83 0 : CommParaInfo commCombinePara(COMM_COMBINE_ORDER, CommType::COMM_TAG_MESH);
84 0 : CHK_RET(CalcCommPlaneInfo(tag_, commCombinePara, opTransport[COMM_COMBINE_ORDER], inputType, outputType));
85 0 : LevelNSubCommTransport& commTransportLevel0 = opTransport[COMM_COMBINE_ORDER];
86 0 : for (u32 subCommIndex = 0; subCommIndex < commTransportLevel0.size(); subCommIndex++) {
87 0 : commTransportLevel0[subCommIndex].isZeroCopy = true;
88 : }
89 0 : return HCCL_SUCCESS;
90 0 : }
91 :
92 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::CalcTransportMemType(
93 : TransportMemType& inputType, TransportMemType& outputType) const
94 : {
95 0 : inputType = TransportMemType::CCL_INPUT;
96 0 : outputType = TransportMemType::CCL_OUTPUT;
97 :
98 0 : HCCL_INFO(
99 : "[CollRunAlltoAllFullMeshSymmetricMemory][CalcTransportMemType] tag[%s] inputType[%d], outputType[%d]",
100 : tag_.c_str(), inputType, outputType);
101 0 : return HCCL_SUCCESS;
102 : }
103 :
104 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
105 : {
106 0 : TransportMemType inputType = TransportMemType::RESERVED;
107 0 : TransportMemType outputType = TransportMemType::RESERVED;
108 :
109 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
110 : // level0 - level1 全连接通信域
111 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
112 0 : return HCCL_SUCCESS;
113 : }
114 :
115 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::GetLocalSendRecvInfoforAlltoall(const OpParam& param)
116 : {
117 0 : u64 curRecvOffset = 0;
118 0 : for (u32 j = 0; j < topoAttr_.userRankSize; j++) {
119 0 : u64 curSendCounts = param.All2AllDataDes.sendCount;
120 0 : u64 curSendLength = curSendCounts * SIZE_TABLE[param.All2AllDataDes.sendType];
121 0 : sendRecvInfo_.remoteSendOffset[j] = curSendLength * topoAttr_.userRank;
122 :
123 0 : u64 curRecvCounts = param.All2AllDataDes.sendCount;
124 0 : u64 curRecvLength = curRecvCounts * SIZE_TABLE[param.All2AllDataDes.recvType];
125 0 : sendRecvInfo_.localRecvLength[j] = curRecvLength;
126 0 : sendRecvInfo_.localRecvOffset[j] = curRecvOffset;
127 0 : curRecvOffset += curRecvLength;
128 0 : HCCL_DEBUG(
129 : "GetLocalSendRecvInfoforAlltoall rank[%u], remoteSendOffset[j][%llu], localRecvLength[j][%llu] "
130 : "localRecvOffset[j][%llu]",
131 : topoAttr_.userRank, sendRecvInfo_.remoteSendOffset[j], sendRecvInfo_.localRecvLength[j],
132 : sendRecvInfo_.localRecvOffset[j]);
133 : }
134 0 : return HCCL_SUCCESS;
135 : }
136 :
137 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::GetLocalSendRecvInfoforAlltoallVC(const OpParam& param)
138 : {
139 0 : u64 rankSize = topoAttr_.userRankSize;
140 0 : u64 usrRank = topoAttr_.userRank;
141 0 : for (u32 j = 0; j < topoAttr_.userRankSize; j++) {
142 0 : sendRecvInfo_.remoteSendOffset[j] = 0;
143 0 : for (u32 i = 0; i < usrRank; i++) {
144 0 : u64 sendCounts = *(static_cast<const u64*>(param.All2AllDataDes.sendCountMatrix) + i + rankSize * j);
145 0 : u64 sendLength = sendCounts * SIZE_TABLE[param.All2AllDataDes.recvType];
146 0 : sendRecvInfo_.remoteSendOffset[j] += sendLength;
147 : }
148 0 : sendRecvInfo_.localRecvOffset[j] = 0;
149 0 : for (u32 i = 0; i < j; i++) {
150 0 : u64 recvCounts = *(static_cast<const u64*>(param.All2AllDataDes.sendCountMatrix) + usrRank + rankSize * i);
151 0 : u64 recvLength = recvCounts * SIZE_TABLE[param.All2AllDataDes.sendType];
152 0 : sendRecvInfo_.localRecvOffset[j] += recvLength;
153 : }
154 0 : u64 curRecvCounts = *(static_cast<const u64*>(param.All2AllDataDes.sendCountMatrix) + usrRank + rankSize * j);
155 0 : sendRecvInfo_.localRecvLength[j] = curRecvCounts * SIZE_TABLE[param.All2AllDataDes.recvType];
156 :
157 0 : HCCL_DEBUG(
158 : "GetLocalSendRecvInfoforAlltoallVC rank[%u], remoteSendOffset[%llu], "
159 : "localRecvLength[%llu], localRecvOffset[%llu]",
160 : topoAttr_.userRank, sendRecvInfo_.remoteSendOffset[j], sendRecvInfo_.localRecvLength[j],
161 : sendRecvInfo_.localRecvOffset[j]);
162 : }
163 0 : return HCCL_SUCCESS;
164 : }
165 :
166 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::GetAlltoAllTmpRankSendRecvInfo(const OpParam& param)
167 : {
168 0 : sendRecvInfo_.remoteSendOffset.resize(topoAttr_.userRankSize, 0);
169 0 : sendRecvInfo_.localRecvLength.resize(topoAttr_.userRankSize, 0);
170 0 : sendRecvInfo_.localRecvOffset.resize(topoAttr_.userRankSize, 0);
171 :
172 0 : if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALL) {
173 0 : CHK_RET(GetLocalSendRecvInfoforAlltoall(param));
174 0 : } else if (param.opType == HcclCMDType::HCCL_CMD_ALLTOALLVC) {
175 0 : CHK_RET(GetLocalSendRecvInfoforAlltoallVC(param));
176 : } else {
177 0 : HCCL_ERROR("Only support optype AllToAll and AllToAllVC !");
178 : }
179 0 : return HCCL_SUCCESS;
180 : }
181 :
182 0 : HcclResult CollRunAlltoAllFullMeshSymmetricMemory::KernelRun(const OpParam& param, ExecMem& execMem)
183 : {
184 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] AllToAll fullmesh start.", __func__);
185 :
186 : // 准备数据
187 0 : CHK_RET(ActiveSlaveStreams(param.stream));
188 0 : CHK_RET(GetAlltoAllTmpRankSendRecvInfo(param));
189 :
190 : // 获取当前超节点内总卡数
191 0 : u32 devNumInlocalPod = INVALID_VALUE_RANKSIZE;
192 0 : u32 rankIdxInPod = INVALID_VALUE_RANKID;
193 0 : CHK_RET(GetLocalSDMAGroupInfo(devNumInlocalPod, rankIdxInPod));
194 :
195 : // 获取通信域
196 0 : CHK_RET(CheckCommSize(COMM_COMBINE_ORDER, COMM_INDEX_0 + 1));
197 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_COMBINE_ORDER, COMM_INDEX_0);
198 : // isSuPodAsym 表示A2A3卡数不一致场景或者A3多超节点server数不同场景
199 0 : bool isSuPodAsym = false;
200 :
201 : // 执行
202 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
203 0 : TemplateType::TEMPLATE_ALL_2_ALL_FULL_MESH_SYMMETRIC_MEMORY, dispatcher_);
204 0 : HCCL_CONFIG_INFO(
205 : HCCL_ALG, "[%s] Run TEMPLATE_ALL_2_ALL_FULL_MESH_SYMMETRIC_MEMORY in COMM_COMBINE_ORDER", __func__);
206 0 : CHK_SMART_PTR_NULL(tempAlg);
207 :
208 0 : PrepareData prepareData;
209 0 : prepareData.stream = param.stream;
210 0 : prepareData.userRank = topoAttr_.userRank;
211 0 : prepareData.userRankSize = topoAttr_.userRankSize;
212 0 : prepareData.linksPtr = &level0CommInfo.links;
213 0 : prepareData.sendRecvInfoPtr = &sendRecvInfo_;
214 0 : prepareData.devNumInlocalPod = devNumInlocalPod;
215 0 : prepareData.rankIdxInPod = rankIdxInPod;
216 :
217 0 : prepareData.inputMem = algResResp_->paramInputMem;
218 0 : prepareData.outputMem = algResResp_->paramOutputMem;
219 0 : prepareData.cclInMem = execMem.inputMem;
220 0 : prepareData.cclOutMem = execMem.outputMem;
221 0 : prepareData.workMode = workflowMode_;
222 0 : prepareData.subStreamsPtr = &algResResp_->slaveStreams;
223 0 : prepareData.signalPtr = &algResResp_->notifiesMain;
224 0 : prepareData.signalAuxPtr = &algResResp_->notifiesAux;
225 0 : prepareData.isSuPodAsym = isSuPodAsym;
226 0 : prepareData.opType = param.opType;
227 0 : prepareData.algOpContext = algOpContext_;
228 :
229 0 : CHK_RET(tempAlg->Prepare(prepareData));
230 :
231 0 : CHK_RET(tempAlg->RunAsync());
232 :
233 0 : HCCL_INFO("[CollRunAlltoAllFullMeshSymmetricMemory] executor run success.");
234 0 : if (algOpContext_.opRetryHandler.isPostSync == true) {
235 0 : OpParam postSyncParam = param;
236 0 : if ((*prepareData.subStreamsPtr).size() == 0) {
237 0 : CHK_RET(PostSyncWithoutSubstream(postSyncParam, execMem));
238 : } else {
239 0 : PrepareData postSyncPrepareData = prepareData;
240 0 : CHK_RET(PostSyncWithSubstream(postSyncParam, execMem, postSyncPrepareData));
241 0 : }
242 0 : }
243 0 : return HCCL_SUCCESS;
244 0 : }
245 :
246 : REGISTER_EXEC(
247 : "RunAlltoAllFullMeshSymmetricMemory", AlltoAllFullMeshSymmetricMemory, CollRunAlltoAllFullMeshSymmetricMemory);
248 : } // namespace hccl
|
