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