Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "ccu_context_all_to_all_v_mesh2die.h"
12 : #include "ccu_instruction_all_to_all_v_mesh2die.h"
13 :
14 : namespace Hccl {
15 :
16 : constexpr int CKE_IDX_0 = 0;
17 : constexpr int CKE_IDX_1 = 1;
18 : constexpr int CKE_IDX_2 = 2;
19 :
20 0 : CcuContextAllToAllVMesh2Die::CcuContextAllToAllVMesh2Die(const CcuCtxArg &arg,
21 0 : const std::vector<CcuTransport*> &transports, const CcuTransportGroup &group)
22 0 : : CcuContextAlgBase(arg, transports, group)
23 : {
24 0 : if (transports.empty()) {
25 0 : THROW<InvalidParamsException>(StringFormat("CcuContextAllToAllVMesh2Die transports is empty"));
26 : }
27 :
28 0 : const CcuCtxArgAllToAllVMesh2Die *ctxArg = dynamic_cast<const CcuCtxArgAllToAllVMesh2Die *>(&arg);
29 0 : if (ctxArg == nullptr) {
30 0 : THROW<NullPtrException>(StringFormat("CcuContextAllToAllVMesh2Die::ctxArg ptr is null"));
31 : }
32 :
33 0 : auto dimSize = ctxArg->dimSize;
34 0 : if (dimSize.size() != 1) { // 2Die场景dimSize为1
35 0 : THROW<InvalidParamsException>(StringFormat("CcuContextAllToAllVMesh2Die::dimSize[%u] is invalid",
36 : dimSize.size()));
37 : }
38 :
39 0 : rankSize_ = dimSize[0];
40 0 : if (rankSize_ <= 1 || rankSize_ % RANK_EVEN != 0) {
41 0 : THROW<InvalidParamsException>(StringFormat("CcuContextAllToAllVMesh2Die::rankSize[%u] is invalid", rankSize_));
42 : }
43 :
44 0 : rankId_ = ctxArg->rankId;
45 0 : withMyRank_ = ctxArg->withMyRank;
46 0 : rankGroup_ = ctxArg->rankGroup;
47 :
48 0 : localSize_ = transports.size() + 1;
49 0 : localId_ = localSize_ - 1; // 本rank所在DIE的编号，固定放在末尾
50 :
51 0 : peerSize_ = transports.size() + (withMyRank_ ? 1 : 0);
52 0 : logicId_ = rankId_ % peerSize_;
53 :
54 0 : selfBit_ = 1 << logicId_;
55 0 : allBit_ = ((1 << peerSize_) - 1) & (~(withMyRank_ ? selfBit_ : 0));
56 :
57 0 : HCCL_INFO("[CcuContextAllToAllVMesh2Die] RankId[%u], rankSize[%llu], localSize[%u], peerSize[%u], logicId[%u], "
58 : "withMyRank[%u]", rankId_, rankSize_, localSize_, peerSize_, logicId_, withMyRank_);
59 0 : }
60 :
61 0 : void CcuContextAllToAllVMesh2Die::InitResources()
62 : {
63 0 : locSignal_ = CreateMaskSignal();
64 :
65 0 : input_ = CreateVariable();
66 :
67 0 : for (uint32_t peerId = 0; peerId < transports.size(); peerId++) {
68 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die]RankId[%u], PeerId[%u]", rankId_, peerId);
69 0 : output_.emplace_back(CreateVariable(*(transports[peerId]), CKE_IDX_1));
70 0 : token_.emplace_back(CreateVariable(*(transports[peerId]), CKE_IDX_2));
71 : }
72 : // 本rank固定放在末尾
73 0 : output_.emplace_back(CreateVariable());
74 0 : token_.emplace_back(CreateVariable());
75 :
76 0 : xnMaxTransportSize_ = CreateVariable();
77 0 : xnMaxTransportGoSize_ = CreateGroupOpSize();
78 :
79 0 : xnMaxTransportSize_ = MAX_TRANSPORT_SIZE;
80 0 : auto xnMaxTransportGoSize = CalGoSize(MAX_TRANSPORT_SIZE);
81 0 : xnMaxTransportGoSize_.addrOffset = xnMaxTransportGoSize[GO_ADDR_OFFSET_IDX];
82 0 : xnMaxTransportGoSize_.loopParam = xnMaxTransportGoSize[GO_LOOP_PARAM_IDX];
83 0 : xnMaxTransportGoSize_.parallelParam = xnMaxTransportGoSize[GO_PARALLEL_PARAM_IDX];
84 0 : xnMaxTransportGoSize_.residual = xnMaxTransportGoSize[GO_RESIDUAL_IDX];
85 :
86 0 : sendRecvInfo_.resize(localSize_);
87 0 : for (uint64_t rankIdx = 0; rankIdx < localSize_; rankIdx++) {
88 0 : sendRecvInfo_[rankIdx].sendOffset = CreateVariable();
89 0 : sendRecvInfo_[rankIdx].recvOffset = CreateVariable();
90 0 : sendRecvInfo_[rankIdx].sendTailSize = CreateVariable();
91 0 : sendRecvInfo_[rankIdx].sendTailGoSize = CreateGroupOpSize();
92 0 : sendRecvInfo_[rankIdx].sendLoopNum = CreateVariable();
93 : }
94 :
95 0 : for (uint16_t i = 0; i < localSize_; i++) {
96 0 : src_.emplace_back(CreateMemory());
97 0 : dst_.emplace_back(CreateMemory());
98 : }
99 :
100 0 : curSendTailSize_ = CreateVariable();
101 0 : curSendTailGoSize_ = CreateGroupOpSize();
102 :
103 0 : xnConst1_ = CreateVariable();
104 0 : completedRankCount_ = CreateVariable();
105 0 : }
106 :
107 0 : void CcuContextAllToAllVMesh2Die::LoadArgs()
108 : {
109 0 : Load(input_);
110 0 : Load(output_[localId_]);
111 0 : Load(token_[localId_]);
112 :
113 0 : for (uint64_t rankIdx = 0; rankIdx < localSize_; rankIdx++) {
114 0 : Load(sendRecvInfo_[rankIdx].sendOffset);
115 0 : Load(sendRecvInfo_[rankIdx].recvOffset);
116 0 : Load(sendRecvInfo_[rankIdx].sendTailSize);
117 0 : Load(sendRecvInfo_[rankIdx].sendTailGoSize);
118 0 : Load(sendRecvInfo_[rankIdx].sendLoopNum);
119 : }
120 0 : }
121 :
122 0 : void CcuContextAllToAllVMesh2Die::ExchangeInfoAndSync()
123 : {
124 : // 交换信息并做同步，前同步固定用1,2,3号信号
125 0 : CcuRep::Variable tempDst = CreateVariable();
126 0 : for (u32 peerId = 0; peerId < transports.size(); peerId++) {
127 0 : uint32_t dst = CalcDstRank(peerId);
128 0 : tempDst = output_[localId_];
129 0 : tempDst += sendRecvInfo_[dst].recvOffset;
130 :
131 0 : WriteVariableWithSignal(*transports[peerId], tempDst, CKE_IDX_1, CKE_IDX_1, selfBit_);
132 0 : WriteVariableWithSignal(*transports[peerId], token_[localId_], CKE_IDX_2, CKE_IDX_2, selfBit_);
133 : }
134 0 : GroupWait(*transportGroup, CKE_IDX_1, allBit_);
135 0 : GroupWait(*transportGroup, CKE_IDX_2, allBit_);
136 0 : }
137 :
138 0 : void CcuContextAllToAllVMesh2Die::PostSync()
139 : {
140 0 : for (const auto &t : transports) {
141 0 : if (t == nullptr) {
142 0 : THROW<NullPtrException>(StringFormat("CcuContextAllToAllVMesh2Die::PostSync transport ptr is null"));
143 : }
144 0 : RemotePost(*t, CKE_IDX_0, selfBit_);
145 : }
146 0 : GroupWait(*transportGroup, CKE_IDX_0, allBit_);
147 0 : }
148 :
149 0 : uint32_t CcuContextAllToAllVMesh2Die::CalcDstRank(uint32_t peerId) const
150 : {
151 0 : return peerId;
152 : }
153 :
154 0 : uint32_t CcuContextAllToAllVMesh2Die::CalcTransIdx(uint32_t peerId) const
155 : {
156 0 : return peerId;
157 : }
158 :
159 0 : void CcuContextAllToAllVMesh2Die::DoAll2AllVMultiLoop()
160 : {
161 0 : completedRankCount_ = 0;
162 0 : xnConst1_ = 1;
163 0 : CCU_WHILE(completedRankCount_ != peerSize_) {
164 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die] Algorithm loops[%u].", peerSize_);
165 0 : LoopStep();
166 0 : }
167 0 : }
168 :
169 0 : void CcuContextAllToAllVMesh2Die::WriteToDstOutput(uint32_t peerId)
170 : {
171 0 : uint32_t dstRank = CalcDstRank(peerId);
172 0 : uint32_t transIdx = CalcTransIdx(peerId);
173 :
174 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die] WriteToDstOutput[%u] Start. RankId[%u] dstRank[%u] transIdx[%u]", peerId,
175 : rankId_, dstRank, transIdx);
176 :
177 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum == UINT64_MAX) // 已经搬完了，仅同步
178 : {
179 0 : LocalPost(locSignal_, (1 << peerId));
180 0 : }
181 :
182 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum != UINT64_MAX) // 还没有搬完
183 : {
184 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum == UINT64_MAX - 1) // 最后一次搬运, 发送尾块数据
185 : {
186 0 : curSendTailSize_ = sendRecvInfo_[dstRank].sendTailSize;
187 0 : CCU_IF(curSendTailSize_ == 0)
188 : {
189 0 : LocalPost(locSignal_, (1 << peerId));
190 0 : }
191 0 : CCU_IF(curSendTailSize_ != 0)
192 : {
193 0 : Write(*(transports[transIdx]), dst_[peerId], src_[peerId], curSendTailSize_, locSignal_, (1 << peerId));
194 0 : }
195 0 : completedRankCount_ += xnConst1_;
196 0 : }
197 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum != UINT64_MAX - 1) // 正常搬运
198 : {
199 0 : Write(*(transports[transIdx]), dst_[peerId], src_[peerId], xnMaxTransportSize_, locSignal_,
200 0 : (1 << peerId));
201 0 : dst_[peerId].addr += xnMaxTransportSize_;
202 0 : src_[peerId].addr += xnMaxTransportSize_;
203 0 : }
204 0 : sendRecvInfo_[dstRank].sendLoopNum += xnConst1_;
205 0 : }
206 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die] WriteToDstOutput end.");
207 0 : }
208 :
209 0 : void CcuContextAllToAllVMesh2Die::GroupCopyToDstOutput(uint32_t peerId)
210 : {
211 0 : uint32_t dstRank = CalcDstRank(peerId);
212 :
213 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die] GroupCopyToDstOutput[%u] Start. RankId[%u] dstRank[%u]", peerId, rankId_,
214 : dstRank);
215 :
216 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum == UINT64_MAX) // 已经搬完了，仅同步
217 : {
218 0 : LocalPost(locSignal_, (1 << peerId));
219 0 : }
220 :
221 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum != UINT64_MAX) // 还没有搬完
222 : {
223 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum == UINT64_MAX - 1) // 最后一次搬运, 发送尾块数据
224 : {
225 0 : curSendTailSize_ = sendRecvInfo_[dstRank].sendTailSize;
226 0 : curSendTailGoSize_ = sendRecvInfo_[dstRank].sendTailGoSize;
227 0 : CCU_IF(curSendTailSize_ == 0)
228 : {
229 0 : LocalPost(locSignal_, (1 << peerId));
230 0 : }
231 0 : CCU_IF(curSendTailSize_ != 0)
232 : {
233 0 : GroupCopy(dst_[peerId], src_[peerId], curSendTailGoSize_);
234 0 : LocalPost(locSignal_, (1 << peerId));
235 0 : }
236 0 : completedRankCount_ += xnConst1_;
237 0 : }
238 0 : CCU_IF(sendRecvInfo_[dstRank].sendLoopNum != UINT64_MAX - 1) // 正常搬运
239 : {
240 0 : GroupCopy(dst_[peerId], src_[peerId], xnMaxTransportGoSize_);
241 0 : dst_[peerId].addr += xnMaxTransportSize_;
242 0 : src_[peerId].addr += xnMaxTransportSize_;
243 0 : LocalPost(locSignal_, (1 << peerId));
244 0 : }
245 0 : sendRecvInfo_[dstRank].sendLoopNum += xnConst1_;
246 0 : }
247 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die] GroupCopyToDstOutput end.");
248 0 : }
249 :
250 0 : void CcuContextAllToAllVMesh2Die::CalcGroupSrcDst()
251 : {
252 0 : for (uint32_t peerId = 0; peerId < transports.size(); peerId++) {
253 0 : const u32 dstRank = CalcDstRank(peerId);
254 :
255 0 : src_[peerId].addr = input_;
256 0 : src_[peerId].addr += sendRecvInfo_[dstRank].sendOffset;
257 0 : src_[peerId].token = token_[peerId];
258 :
259 0 : dst_[peerId].addr = output_[peerId]; // recvOffset在前同步时已经计算
260 0 : dst_[peerId].token = token_[peerId];
261 : }
262 :
263 0 : if (withMyRank_) {
264 0 : src_[localId_].addr = input_;
265 0 : src_[localId_].addr += sendRecvInfo_[localId_].sendOffset;
266 0 : src_[localId_].token = token_[localId_];
267 0 : dst_[localId_].addr = output_[localId_];
268 0 : dst_[localId_].addr += sendRecvInfo_[localId_].recvOffset;
269 0 : dst_[localId_].token = token_[localId_];
270 : }
271 0 : }
272 :
273 0 : void CcuContextAllToAllVMesh2Die::LoopStep()
274 : {
275 0 : for (uint32_t peerId = 0; peerId < transports.size(); peerId++) {
276 0 : WriteToDstOutput(peerId);
277 : }
278 :
279 0 : if (withMyRank_) {
280 0 : GroupCopyToDstOutput(localId_);
281 : }
282 :
283 0 : LocalWait(locSignal_, (1 << peerSize_) - 1);
284 0 : }
285 :
286 0 : void CcuContextAllToAllVMesh2Die::Algorithm()
287 : {
288 : // 初始化寄存器资源 & 加载外部输入参数
289 0 : HCCL_INFO("[CcuContextAllToAllVMesh2Die] Algorithm Init Begins.");
290 0 : InitResources();
291 0 : LoadArgs();
292 :
293 0 : HCCL_INFO("[CcuContextAllToAllVMesh2Die] Algorithm begins.");
294 :
295 : // 框架已经默认做了前后轴同步，算法不需要再重复做
296 0 : ExchangeInfoAndSync();
297 :
298 0 : CalcGroupSrcDst();
299 0 : DoAll2AllVMultiLoop();
300 :
301 0 : PostSync();
302 :
303 0 : HCCL_INFO("[CcuContextAllToAllVMesh2Die] Algorithm Ends.");
304 0 : }
305 :
306 0 : std::vector<uint64_t> CcuContextAllToAllVMesh2Die::GeneArgs(const CcuTaskArg &arg)
307 : {
308 0 : const CcuTaskArgAllToAllVMesh2Die *taskArg = dynamic_cast<const CcuTaskArgAllToAllVMesh2Die *>(&arg);
309 0 : if (taskArg == nullptr) {
310 0 : THROW<NullPtrException>(StringFormat("CcuContextAllToAllVMesh2Die::taskArg ptr is null"));
311 : }
312 :
313 0 : uint64_t inputAddr = taskArg->inputAddr;
314 0 : uint64_t outputAddr = taskArg->outputAddr;
315 0 : uint64_t tokenInfo = taskArg->token;
316 :
317 0 : std::vector<uint64_t> taskParams = {inputAddr, outputAddr, tokenInfo}; // 不需要ScratchMem
318 :
319 0 : for (auto peerId : rankGroup_) {
320 0 : const uint64_t floorLoopNum = taskArg->localSendRecvInfo.sendLength[peerId] / MAX_TRANSPORT_SIZE;
321 0 : uint64_t sendLoopNum = UINT64_MAX - 1 - floorLoopNum;
322 0 : uint64_t sendTailSize = taskArg->localSendRecvInfo.sendLength[peerId] - floorLoopNum * MAX_TRANSPORT_SIZE;
323 0 : auto sendTailGoSize = CalGoSize(sendTailSize);
324 0 : uint64_t sendOffset = taskArg->localSendRecvInfo.sendOffset[peerId];
325 0 : uint64_t recvOffset = taskArg->localSendRecvInfo.recvOffset[peerId];
326 0 : taskParams.push_back(sendOffset);
327 0 : taskParams.push_back(recvOffset);
328 0 : taskParams.push_back(sendTailSize);
329 0 : taskParams.insert(taskParams.cend(), sendTailGoSize.cbegin(), sendTailGoSize.cend());
330 0 : taskParams.push_back(sendLoopNum);
331 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die][sliceInfo] RankId[%u], dstRank[%d]: sendOffset[%llu], "
332 : "recvOffset[%llu], sendLength[%llu], sendTailSize[%llu], sendLoopNum[%llu]", rankId_, peerId, sendOffset,
333 : recvOffset, taskArg->localSendRecvInfo.sendLength[peerId], sendTailSize, sendLoopNum);
334 0 : }
335 :
336 0 : HCCL_DEBUG("[CcuContextAllToAllVMesh2Die][GeneArgs] RankId[%u], inputAddr[%#llx], outputAddr[%#llx], "
337 : "xnMaxTransportSize[%llu], args[%zu]", rankId_, inputAddr, outputAddr, MAX_TRANSPORT_SIZE, taskParams.size());
338 :
339 0 : return taskParams;
340 0 : }
341 :
342 : }
|
