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 "ccu_context_all_gather_mesh1d_detour.h"
12 : #include "ccu_instruction_all_gather_mesh1d_detour.h"
13 : #include "ccu_assist.h"
14 :
15 : namespace Hccl {
16 :
17 : constexpr int OUTPUT_XN_ID = 1;
18 : constexpr int TOKEN_XN_ID = 2;
19 : constexpr int CKE_IDX_0 = 0;
20 : constexpr int CKE_IDX_1 = 1;
21 : constexpr int CKE_IDX_2 = 2;
22 : constexpr int CKE_IDX_3 = 3;
23 :
24 0 : void CcuContextAllGatherMeshDetour1D::ProcessTransports(const std::vector<CcuTransport *> &transports)
25 : {
26 : // 构建detourTransport
27 0 : if (transports.size() % (rankSize_ - 1) != 0) {
28 0 : THROW<InvalidParamsException>(StringFormat(
29 : "Invalid TransportsNum[%zu] for rankSize_[%llu]", transports.size(), rankSize_));
30 : }
31 :
32 0 : for (uint64_t i = 0; i < pathNumPerPeer_; i++) {
33 : // 到每个对端有pathNum个transport，故detourTransport中共有pathNum组
34 0 : detourTransports_.emplace_back(std::vector<CcuTransport*>());
35 : }
36 0 : uint64_t directPathNum = pathNumPerPeer_ - detourPathNum_;
37 0 : for (uint64_t i = 0; i < directPathNum; i++) {
38 : // 有pathNum-detourPathNum组的直连链路，每组重复
39 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
40 0 : detourTransports_[i].emplace_back(transports[j]);
41 : }
42 0 : HCCL_INFO("Add directTransports[%llu], size[%zu]", i, detourTransports_[i].size());
43 : }
44 0 : for (uint64_t i = 0; i < detourPathNum_; i++) {
45 : // 有detourPathNum组的绕路链路，只添加sendOnly的transport
46 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
47 0 : detourTransports_[i + directPathNum].emplace_back(transports[(i + 1) * (rankSize_ - 1) + j]);
48 : }
49 0 : HCCL_INFO("Add detourTransports_[%llu], size[%zu]", i, detourTransports_[i].size());
50 : }
51 :
52 0 : return;
53 : }
54 :
55 0 : CcuContextAllGatherMeshDetour1D::CcuContextAllGatherMeshDetour1D(const CcuCtxArg &arg,
56 : const std::vector<CcuTransport *> &transports,
57 0 : const CcuTransportGroup &group)
58 0 : : CcuContextAlgBase(arg, transports, group)
59 : {
60 0 : HCCL_INFO("[CcuContextAllGatherMeshDetour1D] Enter Constructor.");
61 0 : const CcuCtxArgAllGatherMeshDetour1D *ctxArg = dynamic_cast<const CcuCtxArgAllGatherMeshDetour1D *>(&arg);
62 0 : if (ctxArg == nullptr) {
63 0 : THROW<NullPtrException>(StringFormat("CcuContextAllGatherMeshDetour1D::ctxArg ptr is null"));
64 : }
65 0 : rankId_ = ctxArg->rankId_;
66 0 : if (ctxArg->dimSize_.size() > 0) {
67 0 : rankSize_ = ctxArg->dimSize_[0];
68 : }
69 0 : singleTransportSize_ = ctxArg->singleTransportSize_;
70 0 : detourPathNum_ = ctxArg->detourPathNum_;
71 0 : pathNumPerPeer_ = ctxArg->pathNumPerPeer_;
72 :
73 0 : ProcessTransports(transports);
74 :
75 : // 申请资源
76 0 : input_ = CreateVariable();
77 0 : baseOffset_ = CreateVariable();
78 0 : tailOffset_ = CreateVariable();
79 0 : loopIterNum_ = CreateVariable();
80 0 : groupOpSize_ = CreateGroupOpSize();
81 :
82 0 : uint16_t transportIdx = 0;
83 : // 按照rank号从小到大遍历transports，遇到本rank就填充本地资源，否则依次取远端资源，要求给框架返回的Link同样是按顺序排列的
84 0 : for (uint64_t peerId = 0; peerId < rankSize_; peerId++) {
85 0 : if (peerId == rankId_) {
86 0 : output_.push_back(CreateVariable());
87 0 : token_.push_back(CreateVariable());
88 : } else {
89 0 : HCCL_INFO("[CcuContextAllGatherMeshDetour1D] MyRank[%u], PeerId[%llu], TransportId[%u]",
90 : rankId_, peerId, transportIdx);
91 0 : CHK_PRT_RET(detourTransports_[0][transportIdx] == nullptr || transportIdx >= detourTransports_[0].size(),
92 : HCCL_ERROR("[CcuContextAllGatherMeshDetour1D] Algorithm transport ptr is null or out of bounds"),);
93 0 : output_.push_back(CreateVariable((*detourTransports_[0][transportIdx]), OUTPUT_XN_ID));
94 0 : token_.push_back(CreateVariable((*detourTransports_[0][transportIdx]), TOKEN_XN_ID));
95 0 : transportIdx++;
96 : }
97 : }
98 0 : for (uint32_t i = 0; i < pathNumPerPeer_; i++) {
99 0 : lengths_.emplace_back(CreateVariable());
100 : }
101 :
102 0 : return;
103 0 : }
104 :
105 0 : void CcuContextAllGatherMeshDetour1D::AllocDetourRes()
106 : {
107 : // 预期给每个对端使用的MS数量都相等
108 0 : u32 interleave = 8;
109 0 : moConfig.loopCount = CcuRep::CCU_MS_DEFAULT_LOOP_COUNT;
110 0 : moConfig.msInterleave = interleave; // Bcast为msNum*1，Reduce为msNum*rankSize_
111 0 : if (moRes.executor.size() == 0) {
112 0 : moRes.executor = CreateBlockExecutor(moConfig.loopCount);
113 0 : moRes.maskSignal = CreateBlockMaskSignal(moConfig.loopCount);
114 0 : moRes.ccuBuffer = CreateBlockCcuBuffer(moConfig.loopCount * moConfig.msInterleave);
115 : }
116 0 : return;
117 : }
118 :
119 0 : void CcuContextAllGatherMeshDetour1D::CreateMultiOpBroadcastDetour()
120 : {
121 : // 设到每个对端有相同数量的多个transport，每个transport需要传输的长度与同下标的lengths中的值对应 <直连--R1,绕路--R1>--<L0,L1>
122 : // 当每个transport不均等切分时，考虑1.添加重复的transport，每个transport都用1片MS；2.lengths中给每个transport填不同的长度
123 0 : AllocDetourRes();
124 :
125 0 : std::string loopType = "broadcastDetour";
126 0 : if (registeredLoop.find(loopType) != registeredLoop.end()) {
127 0 : return;
128 : }
129 :
130 0 : CcuRep::LoopBlock lb(this, loopType + "_loop");
131 : {
132 : // loopblock的形参
133 0 : std::vector<CcuRep::Memory> src; // 每组transport对应一个src
134 0 : std::vector<CcuRep::Memory> dst; // 每组transport对应rankSize_个dst
135 0 : std::vector<CcuRep::Variable> lengths;
136 0 : for (uint64_t i = 0; i < pathNumPerPeer_; i++) {
137 0 : lengths.emplace_back(CreateVariable());
138 0 : src.emplace_back(CreateMemory());
139 0 : for (uint64_t j = 0; j < rankSize_; j++) {
140 0 : dst.emplace_back(CreateMemory());
141 : }
142 : }
143 :
144 0 : lb(src, dst, lengths);
145 0 : std::vector<CcuRep::CcuBuffer> bufs;
146 0 : std::vector<CcuRep::MaskSignal> sems;
147 0 : for (uint32_t i = 0; i < pathNumPerPeer_; i++) {
148 0 : bufs.emplace_back(moRes.ccuBuffer[i]);
149 0 : sems.emplace_back(moRes.maskSignal[i]);
150 : }
151 :
152 : // 从本地搬运多片数据到多个MS
153 0 : for (uint64_t i = 0; i < pathNumPerPeer_; i++) {
154 0 : LocalCopy(bufs[i], src[i], lengths[i], sems[i]);
155 : }
156 : // 等待数据搬到MS
157 0 : for (uint64_t i = 0; i < pathNumPerPeer_; i++) {
158 0 : LocalWait(sems[i]);
159 : }
160 : // 给每个peer搬运多个MS上的数据
161 0 : for (uint64_t i = 0; i < pathNumPerPeer_; i++) {
162 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
163 0 : if (detourTransports_[i][j] == nullptr) {
164 0 : THROW<CcuApiException>("transport is nullptr");
165 : }
166 0 : Write(*detourTransports_[i][j], dst[i * rankSize_ + j], bufs[i], lengths[i], sems[i], 1 << j);
167 : }
168 0 : LocalCopy(dst[i * rankSize_ + rankSize_ - 1], bufs[i], lengths[i], sems[i], 1 << (rankSize_ - 1));
169 : }
170 : // 等待给所有远端写完数据
171 0 : for (uint32_t i = 0; i < pathNumPerPeer_; i++) {
172 0 : LocalWait(sems[i], (1 << rankSize_) - 1);
173 : }
174 0 : }
175 :
176 0 : registeredLoop.insert(loopType);
177 0 : return;
178 0 : }
179 :
180 0 : void CcuContextAllGatherMeshDetour1D::GroupBroadcastDetour(
181 : std::vector<CcuRep::Variable> &lengths, std::vector<CcuRep::Memory> &src, std::vector<CcuRep::Memory> &dst)
182 : {
183 0 : CreateMultiOpBroadcastDetour();
184 0 : uint32_t interLeave = 8;
185 :
186 0 : CCU_IF(loopIterNum_ != 0) {
187 0 : CcuRep::Variable loopParam = CreateVariable();
188 0 : CcuRep::Variable paraCfg = CreateVariable();
189 0 : CcuRep::Variable offsetCfg = CreateVariable();
190 :
191 : // sliceSize：单次搬运量，4K*msNum，必须与lengths的总和相等（链路间可以划分不同流量）
192 0 : loopParam = CcuRep::GetLoopParam(0, singleTransportSize_ * moConfig.loopCount, 0); // 偏移是单次总搬运量*loopNum
193 0 : loopParam += loopIterNum_; // 加上loop的迭代次数构成完整loop参数
194 0 : paraCfg = CcuRep::GetParallelParam(moConfig.loopCount - 1, 0, 1); // loop固定展开到128个
195 0 : offsetCfg = CcuRep::GetOffsetParam(singleTransportSize_, interLeave, pathNumPerPeer_); // 下一个loop偏移量
196 0 : auto lc = Loop("broadcastDetour_loop")(src, dst, lengths);
197 0 : LoopGroup({lc}, {loopParam}, paraCfg, offsetCfg);
198 0 : }
199 0 : return;
200 0 : }
201 :
202 0 : void CcuContextAllGatherMeshDetour1D::FirstStep()
203 : {
204 : // step1，绕路搬整块
205 : // 申请memory地址
206 0 : std::vector<CcuRep::Memory> src;
207 0 : std::vector<CcuRep::Memory> dst;
208 0 : for (uint32_t i = 0; i < pathNumPerPeer_; i++) {
209 0 : src.emplace_back(CreateMemory());
210 0 : for (uint32_t j = 0; j < rankSize_; j++) {
211 0 : dst.emplace_back(CreateMemory());
212 : }
213 : }
214 :
215 : // 地址计算
216 0 : src[0].addr = input_;
217 0 : src[0].token = token_[rankId_];
218 0 : for (uint32_t i = 1; i < pathNumPerPeer_; i++) {
219 0 : src[i].addr = src[i - 1].addr + lengths_[i - 1];
220 0 : src[i].token = token_[rankId_];
221 : }
222 0 : uint32_t dstId = 0;
223 0 : uint32_t curId = 0;
224 0 : for (uint64_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
225 0 : if (rankIdx != rankId_) {
226 0 : curId = dstId;
227 0 : dstId++;
228 : } else {
229 0 : curId = rankSize_ - 1;
230 : }
231 0 : dst[curId].addr = output_[rankIdx]; // 直连链路对应的是下标为0*rankSize_+curId的分片
232 0 : dst[curId].addr += baseOffset_;
233 0 : dst[curId].token = token_[rankIdx];
234 : }
235 0 : for (uint64_t i = 1; i < pathNumPerPeer_; i++) {
236 0 : for (uint64_t j = 0; j < rankSize_; j++) {
237 0 : dst[i * rankSize_ + j].addr = dst[(i - 1) * rankSize_ + j].addr + lengths_[i - 1];
238 0 : dst[i * rankSize_ + j].token = dst[(i - 1) * rankSize_ + j].token;
239 : }
240 : }
241 0 : GroupBroadcastDetour(lengths_, src, dst);
242 :
243 0 : return;
244 0 : }
245 :
246 0 : void CcuContextAllGatherMeshDetour1D::SecondStep()
247 : {
248 : // step2，直连搬尾块
249 0 : CcuRep::Memory tailSrc = CreateMemory();
250 0 : std::vector<CcuRep::Memory> tailDst;
251 0 : for (uint32_t i = 0; i < rankSize_; i++) {
252 0 : tailDst.emplace_back(CreateMemory());
253 : }
254 0 : tailSrc.addr = input_;
255 0 : tailSrc.addr += tailOffset_;
256 0 : tailSrc.token = token_[rankId_];
257 0 : uint32_t dstId = 0;
258 0 : uint32_t curId = 0;
259 0 : for (uint64_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
260 0 : if (rankIdx != rankId_) {
261 0 : curId = dstId;
262 0 : dstId++;
263 : } else {
264 0 : curId = rankSize_ - 1;
265 : }
266 0 : tailDst[curId].addr = output_[rankIdx];
267 0 : tailDst[curId].addr += baseOffset_;
268 0 : tailDst[curId].addr += tailOffset_;
269 0 : tailDst[curId].token = token_[rankIdx];
270 : }
271 0 : GroupBroadcast(detourTransports_[0], tailDst, tailSrc, groupOpSize_);
272 :
273 0 : return;
274 0 : }
275 :
276 0 : void CcuContextAllGatherMeshDetour1D::Algorithm()
277 : {
278 0 : HCCL_INFO("[CcuContextAllGatherMeshDetour1D] AllGatherMeshDetour1D run.");
279 0 : uint16_t selfBit = 1 << rankId_;
280 0 : uint16_t allBit = ((1 << rankSize_) - 1) & (~(1 << rankId_));
281 :
282 0 : Load(input_);
283 0 : Load(output_[rankId_]);
284 0 : Load(token_[rankId_]);
285 0 : Load(baseOffset_);
286 0 : Load(tailOffset_);
287 0 : Load(loopIterNum_);
288 0 : Load(groupOpSize_);
289 0 : for (uint32_t i = 0; i < pathNumPerPeer_; i++) {
290 0 : Load(lengths_[i]);
291 : }
292 :
293 : // 只通过直连链路给对端置位，groupwait仍然关联所有transport
294 0 : for (auto t : detourTransports_[0]) {
295 0 : WriteVariableWithSignal(*t, output_[rankId_], OUTPUT_XN_ID, CKE_IDX_1, selfBit); // index = 1，传递output信息
296 0 : WriteVariableWithSignal(*t, token_[rankId_], TOKEN_XN_ID, CKE_IDX_2, selfBit); // index = 2，传递token信息
297 : }
298 0 : GroupWait(*transportGroup, CKE_IDX_1, allBit); // index = 1，传递output信息
299 0 : GroupWait(*transportGroup, CKE_IDX_2, allBit); // index = 2，传递token信息
300 :
301 0 : FirstStep(); // 绕路整块搬运
302 0 : SecondStep(); // 直连尾块搬运
303 :
304 0 : for (auto t : detourTransports_[0]) {
305 0 : RemotePost(*t, CKE_IDX_0, selfBit);
306 : }
307 0 : GroupWait(*transportGroup, CKE_IDX_0, allBit);
308 0 : HCCL_INFO("[CcuContextAllGatherMeshDetour1D] AllGatherMeshDetour1D end.");
309 0 : return;
310 : }
311 :
312 0 : std::vector<uint64_t> CcuContextAllGatherMeshDetour1D::GeneArgs(const CcuTaskArg &arg)
313 : {
314 0 : const CcuTaskArgAllGatherMeshDetour1D *taskArg = dynamic_cast<const CcuTaskArgAllGatherMeshDetour1D *>(&arg);
315 0 : if (taskArg == nullptr) {
316 0 : THROW<NullPtrException>(StringFormat("CcuContextAllGatherMeshDetour1D::taskArg ptr is null"));
317 : }
318 0 : uint64_t inputAddr = taskArg->inputAddr_;
319 0 : uint64_t outputAddr = taskArg->outputAddr_;
320 0 : uint64_t tokenInfo = taskArg->token_;
321 0 : uint64_t baseOffset = taskArg->baseOffset_;
322 0 : uint64_t tailOffset = taskArg->tailOffset_;
323 0 : uint64_t loopIterNum = taskArg->loopIterNum_;
324 0 : auto goSize = CalGoSize(taskArg->tailSize_);
325 :
326 : std::vector<uint64_t> sqeArgs = {inputAddr, outputAddr, tokenInfo, baseOffset, tailOffset, loopIterNum,
327 0 : goSize[0], goSize[1], goSize[2], goSize[3]};
328 0 : for (auto len : taskArg->lengths_) {
329 0 : sqeArgs.emplace_back(len);
330 : }
331 0 : return sqeArgs;
332 0 : }
333 : }
|
