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_reduce_mesh1d_detour.h"
12 : #include "ccu_instruction_all_reduce_mesh1d_detour.h"
13 :
14 : namespace Hccl {
15 :
16 : constexpr int INPUT_XN_ID = 0;
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 : CcuContextAllReduceMeshDetour1D::CcuContextAllReduceMeshDetour1D(const CcuCtxArg &arg,
25 : const std::vector<CcuTransport *> &transports,
26 0 : const CcuTransportGroup &group)
27 0 : : CcuContextAlgBase(arg, transports, group)
28 : {
29 0 : const CcuCtxArgAllReduceMeshDetour1D *ctxArg = dynamic_cast<const CcuCtxArgAllReduceMeshDetour1D *>(&arg);
30 0 : if (ctxArg == nullptr) {
31 0 : THROW<NullPtrException>(StringFormat("CcuContextAllReduceMeshDetour1D::ctxArg ptr is null"));
32 : }
33 0 : rankId_ = ctxArg->rankId_;
34 0 : rankSize_ = ctxArg->dimSize_[0];
35 0 : dataType_ = ctxArg->op_.dataType;
36 0 : outputDataType_ = ctxArg->op_.outputDataType;
37 0 : if (outputDataType_ == DataType::INVALID) {
38 0 : outputDataType_ = dataType_;
39 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] outputDataType is [INVALID], set outputDataType to[%s]",
40 : outputDataType_.Describe().c_str());
41 : }
42 0 : reduceOp_ = ctxArg->op_.reduceOp;
43 0 : singleTransportSize = ctxArg->singleTransportSize_;
44 0 : detourPathNum = ctxArg->detourPathNum_;
45 0 : pathNumPerPeer = ctxArg->pathNumPerPeer_;
46 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] Init, CtxArgs are rankId_[%u], rankSize_[%u], dataType[%s], "
47 : "outputDataType[%s], reduceOp[%s]", rankId_, rankSize_, dataType_.Describe().c_str(),
48 : outputDataType_.Describe().c_str(), reduceOp_.Describe().c_str());
49 :
50 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] transport.size[%zu]", transports.size());
51 0 : if (transports.size() < rankSize_ -1) {
52 0 : THROW<NullPtrException>(StringFormat("CcuContextAllReduceMeshDetour1D transports size is less"));
53 : }
54 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
55 : // 到每个对端有pathNum个transport，故detourTransport中共有pathNum组
56 0 : detourTransports_.emplace_back(std::vector<CcuTransport*>());
57 : }
58 0 : uint64_t directPathNum = pathNumPerPeer - detourPathNum;
59 0 : for (uint64_t i = 0; i < directPathNum; i++) {
60 : // 有pathNum-detourPathNum组的直连链路，每组重复
61 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
62 0 : detourTransports_[i].emplace_back(transports[j]);
63 : }
64 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] Add directTransports[%llu], size[%zu]", i, detourTransports_[i].size());
65 : }
66 0 : for (uint64_t i = 0; i < detourPathNum; i++) {
67 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
68 0 : detourTransports_[i + directPathNum].emplace_back(transports[(i + 1) * (rankSize_ - 1) + j]);
69 0 : detourTransports_[i + directPathNum].emplace_back(transports[(i + 1) * (rankSize_ - 1) + j + detourPathNum * (rankSize_ - 1)]);
70 0 : HCCL_INFO("detourTransports_ emplace_back sendLink[%u], recvLink[%u]",
71 : (i + 1) * (rankSize_ - 1) + j, (i + 1) * (rankSize_ - 1) + j + detourPathNum * (rankSize_ - 1));
72 : }
73 : }
74 0 : }
75 :
76 0 : void CcuContextAllReduceMeshDetour1D::CreateMultiOpReduceDetour(DataType &dataType, DataType &outputDataType, ReduceOp &opType)
77 : {
78 0 : moConfig.loopCount = CcuRep::CCU_MS_DEFAULT_LOOP_COUNT;
79 0 : moConfig.msInterleave = pathNumPerPeer * rankSize_;
80 0 : if (moRes.executor.size() == 0) {
81 0 : moRes.maskSignal = CreateBlockMaskSignal(moConfig.loopCount);
82 0 : moRes.executor = CreateBlockExecutor(moConfig.loopCount);
83 0 : moRes.ccuBuffer = CreateBlockCcuBuffer(moConfig.loopCount * moConfig.msInterleave);
84 : }
85 0 : std::string loopType = "reduceDetour";
86 0 : if (registeredLoop.find(loopType) != registeredLoop.end()) {
87 0 : return;
88 : }
89 0 : CcuRep::LoopBlock lb(this, loopType + "_loop");
90 : {
91 : // loopblock的形参
92 0 : std::vector<CcuRep::Memory> dst;
93 0 : std::vector<CcuRep::Memory> src;
94 0 : std::vector<CcuRep::Variable> lengths;
95 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
96 0 : lengths.emplace_back(CreateVariable());
97 0 : dst.emplace_back(CreateMemory());
98 0 : for (uint64_t j = 0; j < rankSize_; j++) {
99 0 : src.emplace_back(CreateMemory());
100 : }
101 : }
102 :
103 0 : lb(src, dst, lengths);
104 0 : std::vector<std::vector<CcuRep::CcuBuffer>> bufs;
105 0 : bufs.resize(pathNumPerPeer);
106 0 : std::vector<CcuRep::MaskSignal> sems;
107 :
108 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
109 0 : for (uint64_t j = 0; j < rankSize_; j++) {
110 0 : bufs[i].emplace_back(moRes.ccuBuffer[i * rankSize_ + j]);
111 : }
112 0 : sems.emplace_back(moRes.maskSignal[i]);
113 : }
114 :
115 : // 先读远端直连的到本地MS
116 0 : uint64_t directPathNum = pathNumPerPeer - detourPathNum;
117 0 : for (uint64_t i = 0; i < directPathNum; i++) {
118 0 : for (uint32_t j = 0; j < detourTransports_[i].size(); j++) {
119 0 : if (detourTransports_[i][j] == nullptr) {
120 0 : THROW<CcuApiException>("transport is nullptr");
121 : }
122 0 : Read(*detourTransports_[i][j], bufs[i][j], src[i * rankSize_ + j], lengths[i], sems[i], 1 << j);
123 : }
124 : }
125 : // 再读远端绕路的到本地MS
126 0 : for (uint64_t i = directPathNum; i < pathNumPerPeer; i++) {
127 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
128 0 : if (detourTransports_[i][j * 2 + 1] == nullptr) { // j * 2 + 1是recvOnly Link
129 0 : THROW<CcuApiException>("transport is nullptr");
130 : }
131 0 : Read(*detourTransports_[i][j * 2 + 1], bufs[i][j], src[i * rankSize_ + j], lengths[i], sems[i], 1 << j);
132 : }
133 : }
134 :
135 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
136 0 : LocalCopy(bufs[i][rankSize_ - 1], src[i * rankSize_ + rankSize_ - 1], lengths[i], sems[i], 1 << (rankSize_ - 1));
137 : }
138 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
139 0 : LocalWait(sems[i], (1 << rankSize_) - 1);
140 : }
141 0 : if (rankSize_ > 1) {
142 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
143 0 : LocalReduce(bufs[i], rankSize_, dataType, outputDataType, opType, sems[i], lengths[i]);
144 0 : LocalWait(sems[i]);
145 : }
146 : }
147 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
148 0 : LocalCopy(dst[i], bufs[i][0], lengths[i], sems[i]);
149 0 : LocalWait(sems[i]);
150 : }
151 0 : }
152 0 : registeredLoop.insert(loopType);
153 0 : return;
154 0 : }
155 :
156 0 : void CcuContextAllReduceMeshDetour1D::GroupReduceDetour(std::vector<CcuRep::Memory> &src,
157 : std::vector<CcuRep::Memory> &dst, DataType &dataType, DataType &outputDataType, ReduceOp &opType)
158 : {
159 0 : CreateMultiOpReduceDetour(dataType, outputDataType, opType);
160 0 : uint32_t interLeave = 8;
161 :
162 0 : CCU_IF(iterNum_ != 0) {
163 0 : CcuRep::Variable loopParam = CreateVariable();
164 0 : CcuRep::Variable paraCfg = CreateVariable();
165 0 : CcuRep::Variable offsetCfg = CreateVariable();
166 :
167 0 : loopParam = CcuRep::GetLoopParam(0, singleTransportSize * moConfig.loopCount, 0); // 下次迭代的偏移是单次总搬运量*loopNum
168 0 : loopParam += iterNum_; // 加上loop的迭代次数构成完整loop参数
169 0 : paraCfg = CcuRep::GetParallelParam(moConfig.loopCount - 1, 0, 1); // loop固定展开到128个
170 0 : offsetCfg = CcuRep::GetOffsetParam(singleTransportSize, interLeave, pathNumPerPeer); // 下一个loop偏移量
171 0 : auto lc = Loop("reduceDetour_loop")(src, dst, lengths_);
172 0 : LoopGroup({lc}, {loopParam}, paraCfg, offsetCfg);
173 0 : }
174 0 : return;
175 0 : }
176 :
177 0 : void CcuContextAllReduceMeshDetour1D::CreateMultiOpBroadcastDetour()
178 : {
179 0 : moConfig.loopCount = CcuRep::CCU_MS_DEFAULT_LOOP_COUNT;
180 0 : moConfig.msInterleave = pathNumPerPeer * 1; // Bcast为msNum*1，Reduce为msNum*rankSize_
181 0 : if (moRes.executor.size() == 0) {
182 0 : moRes.executor = CreateBlockExecutor(moConfig.loopCount);
183 0 : moRes.maskSignal = CreateBlockMaskSignal(moConfig.loopCount);
184 0 : moRes.ccuBuffer = CreateBlockCcuBuffer(moConfig.loopCount * moConfig.msInterleave);
185 : }
186 :
187 0 : std::string loopType = "broadcastDetour";
188 0 : if (registeredLoop.find(loopType) != registeredLoop.end()) {
189 0 : return;
190 : }
191 :
192 0 : CcuRep::LoopBlock lb(this, loopType + "_loop");
193 : {
194 : // loopblock的形参
195 0 : std::vector<CcuRep::Memory> src;
196 0 : std::vector<CcuRep::Memory> dst;
197 0 : std::vector<CcuRep::Variable> lengths;
198 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
199 0 : lengths.emplace_back(CreateVariable());
200 0 : src.emplace_back(CreateMemory());
201 0 : for (uint64_t j = 0; j < rankSize_; j++) {
202 0 : dst.emplace_back(CreateMemory());
203 : }
204 : }
205 :
206 0 : lb(src, dst, lengths);
207 0 : std::vector<CcuRep::CcuBuffer> bufs;
208 0 : std::vector<CcuRep::MaskSignal> sems;
209 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
210 0 : bufs.emplace_back(moRes.ccuBuffer[i]);
211 0 : sems.emplace_back(moRes.maskSignal[i]);
212 : }
213 :
214 : // 从本地搬运多片数据到多个MS
215 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
216 0 : LocalCopy(bufs[i], src[i], lengths[i], sems[i]);
217 : }
218 : // 等待数据搬到MS
219 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
220 0 : LocalWait(sems[i]);
221 : }
222 : // 给每个peer搬运多个MS上的数据
223 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
224 0 : for (uint64_t j = 0; j < rankSize_ - 1; j++) {
225 0 : if (detourTransports_[i][j * 2] == nullptr) { // j * 2是sendOnly Link
226 0 : THROW<CcuApiException>("transport is nullptr");
227 : }
228 0 : Write(*detourTransports_[i][j * 2], dst[i * rankSize_ + j], bufs[i], lengths[i], sems[i], 1 << j);
229 : }
230 0 : LocalCopy(dst[i * rankSize_ + rankSize_ - 1], bufs[i], lengths[i], sems[i], 1 << (rankSize_ - 1));
231 : }
232 : // 等待给所有远端写完数据
233 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
234 0 : LocalWait(sems[i], (1 << rankSize_) - 1);
235 : }
236 0 : }
237 :
238 0 : registeredLoop.insert(loopType);
239 0 : return;
240 0 : }
241 :
242 0 : void CcuContextAllReduceMeshDetour1D::GroupBroadcastDetour(std::vector<CcuRep::Variable> &lengths, std::vector<CcuRep::Memory> &src,
243 : std::vector<CcuRep::Memory> &dst)
244 : {
245 0 : CreateMultiOpBroadcastDetour();
246 0 : uint32_t interLeave = 8;
247 :
248 0 : CCU_IF(iterNum_ != 0) {
249 0 : CcuRep::Variable loopParam = CreateVariable();
250 0 : CcuRep::Variable paraCfg = CreateVariable();
251 0 : CcuRep::Variable offsetCfg = CreateVariable();
252 :
253 0 : loopParam = CcuRep::GetLoopParam(0, singleTransportSize * moConfig.loopCount, 0); // 偏移是单次总搬运量*loopNum
254 0 : loopParam += iterNum_; // 加上loop的迭代次数构成完整loop参数
255 0 : paraCfg = CcuRep::GetParallelParam(moConfig.loopCount - 1, 0, 1); // loop固定展开到128个
256 0 : offsetCfg = CcuRep::GetOffsetParam(singleTransportSize, interLeave, pathNumPerPeer); // 下一个loop偏移量
257 0 : auto lc = Loop("broadcastDetour_loop")(src, dst, lengths);
258 0 : LoopGroup({lc}, {loopParam}, paraCfg, offsetCfg);
259 0 : }
260 0 : return;
261 0 : }
262 :
263 0 : void CcuContextAllReduceMeshDetour1D::ReduceScatterFirstStep()
264 : {
265 0 : std::vector<CcuRep::Memory> reduceSrc;
266 0 : std::vector<CcuRep::Memory> reduceDst;
267 :
268 : // 为每个直连或绕路transport分别准备reduceSrc与reduceDst
269 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
270 0 : reduceDst.emplace_back(CreateMemory());
271 0 : for (uint64_t j = 0; j < rankSize_; j++) {
272 0 : reduceSrc.emplace_back(CreateMemory());
273 : }
274 : }
275 :
276 : // reduceDst填充
277 0 : reduceDst[0].addr = output_[rankId_];
278 0 : reduceDst[0].addr += offset_;
279 0 : reduceDst[0].token = token_[rankId_];
280 0 : for (uint64_t i = 1; i < pathNumPerPeer; i++) {
281 0 : reduceDst[i].addr = reduceDst[i - 1].addr + lengths_[i - 1];
282 0 : reduceDst[i].token = token_[rankId_];
283 : }
284 : // 直连transport的reduceSrc填充
285 0 : uint32_t srcId = 0;
286 0 : uint32_t curId = 0;
287 0 : for (uint64_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
288 0 : if (rankIdx != rankId_) {
289 0 : curId = srcId;
290 0 : srcId++;
291 : } else {
292 0 : curId = rankSize_ - 1;
293 : }
294 0 : reduceSrc[curId].addr = input_[rankIdx];
295 0 : reduceSrc[curId].addr += offset_;
296 0 : reduceSrc[curId].token = token_[rankIdx];
297 : }
298 : // 绕路transport的reduceSrc相比直连src再做偏移
299 0 : for (uint64_t i = 1; i < pathNumPerPeer; i++) {
300 0 : for (uint64_t j = 0; j < rankSize_; j++) {
301 0 : reduceSrc[i * rankSize_ + j].addr = reduceSrc[(i - 1) * rankSize_ + j].addr + lengths_[i - 1];
302 0 : reduceSrc[i * rankSize_ + j].token = reduceSrc[(i - 1) * rankSize_ + j].token;
303 : }
304 : }
305 :
306 : // 整块数据用绕路Reduce
307 0 : GroupReduceDetour(reduceSrc, reduceDst, dataType_, outputDataType_, reduceOp_);
308 0 : return;
309 0 : }
310 :
311 0 : void CcuContextAllReduceMeshDetour1D::ReduceScatterSecondStep()
312 : {
313 : // 余下的尾块用直连Reduce
314 0 : std::vector<CcuRep::Memory> tailSrc;
315 0 : CcuRep::Memory tailDst = CreateMemory();
316 0 : for (uint64_t i = 0; i < rankSize_; i++) {
317 0 : tailSrc.emplace_back(CreateMemory());
318 : }
319 0 : tailDst.addr = output_[rankId_];
320 0 : tailDst.addr += offset_;
321 0 : tailDst.addr += tailOffset_;
322 0 : tailDst.token = token_[rankId_];
323 0 : uint32_t srcId = 0;
324 0 : uint32_t curId = 0;
325 0 : for (uint64_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
326 0 : if (rankIdx != rankId_) {
327 0 : curId = srcId;
328 0 : srcId++;
329 : } else {
330 0 : curId = rankSize_ - 1;
331 : }
332 0 : tailSrc[curId].addr = input_[rankIdx];
333 0 : tailSrc[curId].addr += tailOffset_;
334 0 : tailSrc[curId].addr += offset_;
335 0 : tailSrc[curId].token = token_[rankIdx];
336 : }
337 :
338 0 : GroupReduce(detourTransports_[0], tailDst, tailSrc, groupOpSize_, dataType_, outputDataType_, reduceOp_);
339 0 : return;
340 0 : }
341 :
342 :
343 0 : void CcuContextAllReduceMeshDetour1D::AllGatherFirstStep()
344 : {
345 : // 开始AllGather
346 0 : std::vector<CcuRep::Memory> allGatherSrc;
347 0 : std::vector<CcuRep::Memory> allGatherDst;
348 :
349 : // 为每个直连或绕路transport分别准备src与dst
350 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
351 0 : allGatherSrc.emplace_back(CreateMemory());
352 0 : for (uint64_t j = 0; j < rankSize_; j++) {
353 0 : allGatherDst.emplace_back(CreateMemory());
354 : }
355 : }
356 : // allGather 的输入就是 reduceScatter 的输出
357 0 : allGatherSrc[0].addr = output_[rankId_]; // 直连源地址
358 0 : allGatherSrc[0].addr += offset_;
359 0 : allGatherSrc[0].token = token_[rankId_];
360 0 : for (uint64_t i = 1; i < pathNumPerPeer; i++) {
361 0 : allGatherSrc[i].addr = allGatherSrc[i - 1].addr + lengths_[i - 1];
362 0 : allGatherSrc[i].token = token_[rankId_];
363 : }
364 :
365 : // 直连的allGatherDst填充
366 0 : uint32_t curId = 0;
367 0 : uint32_t dstId = 0;
368 0 : for (uint64_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
369 0 : if (rankIdx != rankId_) {
370 0 : curId = dstId;
371 0 : dstId++;
372 : } else {
373 0 : curId = rankSize_ - 1;
374 : }
375 0 : allGatherDst[curId].addr = output_[rankIdx];
376 0 : allGatherDst[curId].addr += offset_;
377 0 : allGatherDst[curId].token = token_[rankIdx];
378 : }
379 :
380 : // 绕路的allGatherDst填充，相比直连做偏移
381 0 : for (uint64_t i = 1; i < pathNumPerPeer; i++) {
382 0 : for (uint64_t j = 0; j < rankSize_; j++) {
383 0 : allGatherDst[i * rankSize_ + j].addr = allGatherDst[(i - 1) * rankSize_ + j].addr + lengths_[i - 1];
384 0 : allGatherDst[i * rankSize_ + j].token = allGatherDst[(i - 1) * rankSize_ + j].token;
385 : }
386 : }
387 0 : GroupBroadcastDetour(lengths_, allGatherSrc, allGatherDst);
388 0 : return;
389 0 : }
390 :
391 0 : void CcuContextAllReduceMeshDetour1D::AllGatherSecondStep()
392 : {
393 : // 余下的尾块用直连transport发送
394 0 : CcuRep::Memory bcastTailSrc = CreateMemory();
395 0 : std::vector<CcuRep::Memory> bcastTailDst;
396 0 : for (uint64_t i = 0; i < rankSize_; i++) {
397 0 : bcastTailDst.emplace_back(CreateMemory());
398 : }
399 0 : bcastTailSrc.addr = output_[rankId_];
400 0 : bcastTailSrc.addr += offset_;
401 0 : bcastTailSrc.addr += tailOffset_;
402 0 : bcastTailSrc.token = token_[rankId_];
403 0 : uint32_t dstId = 0;
404 0 : uint32_t curId = 0;
405 0 : for (uint64_t rankIdx = 0; rankIdx < rankSize_; rankIdx++) {
406 0 : if (rankIdx != rankId_) {
407 0 : curId = dstId;
408 0 : dstId++;
409 : } else {
410 0 : curId = rankSize_ - 1;
411 : }
412 0 : bcastTailDst[curId].addr = output_[rankIdx];
413 0 : bcastTailDst[curId].addr += offset_;
414 0 : bcastTailDst[curId].addr += tailOffset_;
415 0 : bcastTailDst[curId].token = token_[rankIdx];
416 : }
417 0 : GroupBroadcast(detourTransports_[0], bcastTailDst, bcastTailSrc, groupOpSize_);
418 0 : return;
419 0 : }
420 :
421 0 : void CcuContextAllReduceMeshDetour1D::Algorithm()
422 : {
423 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] AllReduceMeshDetour1D run.");
424 0 : uint16_t selfBit = 1 << rankId_;
425 0 : uint16_t allBit = ((1 << rankSize_) - 1) & (~(1 << rankId_));
426 :
427 : // 初始化资源
428 0 : uint16_t transportIdx = 0;
429 : // 按照rank号从小到大遍历transports，遇到本rank就填充本地资源，否则依次取远端资源，要求给框架返回的Link同样是按顺序排列的
430 0 : for (uint64_t peerId = 0; peerId < rankSize_; peerId++) {
431 0 : if (peerId == rankId_) {
432 0 : input_.push_back(CreateVariable());
433 0 : output_.push_back(CreateVariable());
434 0 : token_.push_back(CreateVariable());
435 : } else {
436 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] MyRank[%u], PeerId[%llu], TransportId[%u]",
437 : rankId_, peerId, transportIdx);
438 0 : CHK_PRT_RET(detourTransports_[0][transportIdx] == nullptr,
439 : HCCL_ERROR("[CcuContextAllReduceMeshDetour1D] Algorithm transport ptr is null"),);
440 0 : input_.push_back(CreateVariable((*detourTransports_[0][transportIdx]), INPUT_XN_ID));
441 0 : output_.push_back(CreateVariable((*detourTransports_[0][transportIdx]), OUTPUT_XN_ID));
442 0 : token_.push_back(CreateVariable((*detourTransports_[0][transportIdx]), TOKEN_XN_ID));
443 0 : transportIdx++;
444 : }
445 : }
446 0 : offset_ = CreateVariable();
447 0 : iterNum_ = CreateVariable();
448 0 : tailOffset_ = CreateVariable();
449 0 : tailSize_ = CreateVariable();
450 0 : groupOpSize_ = CreateGroupOpSize();
451 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
452 0 : lengths_.emplace_back(CreateVariable());
453 : }
454 :
455 0 : Load(input_[rankId_]);
456 0 : Load(output_[rankId_]);
457 0 : Load(token_[rankId_]);
458 0 : Load(offset_);
459 0 : Load(iterNum_);
460 0 : Load(tailOffset_);
461 0 : Load(tailSize_);
462 0 : Load(groupOpSize_);
463 0 : for (uint64_t i = 0; i < pathNumPerPeer; i++) {
464 0 : Load(lengths_[i]);
465 : }
466 :
467 0 : for (auto &t : detourTransports_[0]) {
468 0 : WriteVariableWithSignal(*t, input_[rankId_], INPUT_XN_ID, CKE_IDX_1, selfBit);
469 0 : WriteVariableWithSignal(*t, output_[rankId_], OUTPUT_XN_ID, CKE_IDX_2, selfBit);
470 0 : WriteVariableWithSignal(*t, token_[rankId_], TOKEN_XN_ID, CKE_IDX_3, selfBit);
471 : }
472 :
473 0 : GroupWait(*transportGroup, CKE_IDX_1, allBit);
474 0 : GroupWait(*transportGroup, CKE_IDX_2, allBit);
475 0 : GroupWait(*transportGroup, CKE_IDX_3, allBit);
476 :
477 0 : ReduceScatterFirstStep();
478 0 : ReduceScatterSecondStep();
479 :
480 0 : AllGatherFirstStep();
481 0 : AllGatherSecondStep();
482 :
483 0 : for (auto t : detourTransports_[0]) {
484 0 : RemotePost(*t, CKE_IDX_0, selfBit);
485 : }
486 0 : GroupWait(*transportGroup, CKE_IDX_0, allBit);
487 :
488 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] AllReduceMeshDetour1D end.");
489 0 : return;
490 : }
491 :
492 0 : std::vector<uint64_t> CcuContextAllReduceMeshDetour1D::GeneArgs(const CcuTaskArg &arg)
493 : {
494 0 : const CcuTaskArgAllReduceMeshDetour1D *taskArg = dynamic_cast<const CcuTaskArgAllReduceMeshDetour1D *>(&arg);
495 0 : if (taskArg == nullptr) {
496 0 : THROW<NullPtrException>(StringFormat("CcuContextAllReduceMeshDetour1D::taskArg ptr is null"));
497 : }
498 0 : uint64_t inputAddr = taskArg->inputAddr_;
499 0 : uint64_t outputAddr = taskArg->outputAddr_;
500 0 : uint64_t tokenInfo = taskArg->token_;
501 0 : uint64_t offset = taskArg->offset_;
502 0 : uint64_t iterNum = taskArg->iterNum_;
503 0 : uint64_t tailOffset = taskArg->tailOffset_;
504 0 : uint64_t tailSize = taskArg->tailSize_;
505 0 : auto goSize = CalGoSize(tailSize);
506 :
507 0 : HCCL_INFO("[CcuContextAllReduceMeshDetour1D] GeneArgs, taskArg are inputAddr[%llu], outputAddr[%llu], "
508 : "offset[%llu], iterNum[%llu], tailOffset[%llu], tailSize[%llu]",
509 : inputAddr, outputAddr, offset, iterNum, tailOffset, tailSize);
510 : std::vector<uint64_t> sqeArgs = {inputAddr, outputAddr, tokenInfo, offset, iterNum, tailOffset, tailSize,
511 0 : goSize[0], goSize[1], goSize[2], goSize[3]};
512 0 : for (auto len : taskArg->lengths_) {
513 0 : sqeArgs.emplace_back(len);
514 : }
515 0 : return sqeArgs;
516 0 : }
517 :
518 : }
|
