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 "alltoallv_continuous_pipeline.h"
12 :
13 : #include <vector>
14 : #include <algorithm>
15 : #include "alg_template_register.h"
16 :
17 : namespace hccl {
18 0 : AlltoallvContinuousPipeline::AlltoallvContinuousPipeline(const HcclDispatcher dispatcher)
19 0 : : AlgTemplateBase(dispatcher)
20 0 : {}
21 :
22 0 : AlltoallvContinuousPipeline::~AlltoallvContinuousPipeline() {}
23 :
24 0 : HcclResult AlltoallvContinuousPipeline::PrepareSendRecvInfo( std::vector<SendRecvInfo> &sendRecvInfoList)
25 : {
26 0 : if (sendRecvInfoList.size() == 1) {
27 : // 真实业务场景
28 0 : SendRecvInfo &localSendRecvInfo = sendRecvInfoList[0];
29 0 : localSendCounts_ = std::move(localSendRecvInfo.sendCounts);
30 0 : localSendDispls_ = std::move(localSendRecvInfo.sendDispls);
31 0 : localRecvCounts_ = std::move(localSendRecvInfo.recvCounts);
32 0 : localRecvDispls_ = std::move(localSendRecvInfo.recvDispls);
33 0 : needCollectInfo_ = true; // 需要收集信息
34 0 : std::copy(localRecvCounts_.begin(), localRecvCounts_.end(), intraRecvCounts_[intraRankId_].begin());
35 : } else {
36 : // 适配算法分析器，实际业务不会走这个分支
37 0 : SendRecvInfo &localSendRecvInfo = sendRecvInfoList[userRank_];
38 :
39 0 : std::copy(localSendRecvInfo.sendCounts.begin(),
40 : localSendRecvInfo.sendCounts.end(),
41 0 : std::back_inserter(localSendCounts_));
42 0 : std::copy(localSendRecvInfo.sendDispls.begin(),
43 : localSendRecvInfo.sendDispls.end(),
44 0 : std::back_inserter(localSendDispls_));
45 0 : std::copy(localSendRecvInfo.recvCounts.begin(),
46 : localSendRecvInfo.recvCounts.end(),
47 0 : std::back_inserter(localRecvCounts_));
48 0 : std::copy(localSendRecvInfo.recvDispls.begin(),
49 : localSendRecvInfo.recvDispls.end(),
50 0 : std::back_inserter(localRecvDispls_));
51 :
52 0 : for (u32 intraRankIdx = 0; intraRankIdx < intraRankSize_; ++intraRankIdx) {
53 0 : const u32 remoteRank = interRankId_ * intraRankSize_ + intraRankIdx;
54 0 : SendRecvInfo &sendRecvInfo = sendRecvInfoList[remoteRank];
55 0 : std::copy(
56 0 : sendRecvInfo.recvCounts.begin(), sendRecvInfo.recvCounts.end(), intraRecvCounts_[intraRankIdx].begin());
57 : }
58 0 : needCollectInfo_ = false;
59 : }
60 :
61 0 : intraLoopNum_ = GetLocalLoopNum();
62 0 : return HCCL_SUCCESS;
63 : }
64 :
65 0 : HcclResult AlltoallvContinuousPipeline::PrepareTopoInfo(const u32 userRank, const SubCommInfo &level0CommInfo,
66 : const SubCommInfo &level1CommInfo)
67 : {
68 0 : constexpr u32 MIN_RANKSIZE = 2;
69 0 : interRankSize_ = level1CommInfo.localRankSize;
70 0 : CHK_PRT_RET(interRankSize_ < MIN_RANKSIZE,
71 : HCCL_ERROR("[AlltoallvContinuousPipeline][PrepareTopoInfo] Unexpected inter rank size[%u], which should >= 2.",
72 : interRankSize_),
73 : HCCL_E_PARA);
74 :
75 0 : intraRankSize_ = level0CommInfo.localRankSize;
76 0 : CHK_PRT_RET(intraRankSize_ < MIN_RANKSIZE,
77 : HCCL_ERROR("[AlltoallvContinuousPipeline][PrepareTopoInfo] Unexpected intra rank size[%u], which should >= 2.",
78 : intraRankSize_),
79 : HCCL_E_PARA);
80 :
81 0 : userRankSize_ = intraRankSize_ * interRankSize_;
82 0 : userRank_ = userRank;
83 0 : interRankId_ = level1CommInfo.localRank;
84 0 : intraRankId_ = level0CommInfo.localRank;
85 0 : HCCL_INFO("[AlltoallvContinuousPipeline][PrepareTopoInfo] userRank[%u], intraRankId[%u], intraRankSize[%u], "
86 : "interRankId[%u], interRankSize[%u]", userRank_, intraRankId_, intraRankSize_, interRankId_, interRankSize_);
87 :
88 : // 按照module将rank分组
89 0 : ranksPerModule_.resize(interRankSize_);
90 0 : for (u32 interRank = 0; interRank < interRankSize_; ++interRank) {
91 0 : ranksPerModule_[interRank].resize(intraRankSize_);
92 0 : for (u32 intraRank = 0; intraRank < intraRankSize_; ++intraRank) {
93 0 : ranksPerModule_[interRank][intraRank] = interRank * intraRankSize_ + intraRank;
94 : }
95 : }
96 0 : return HCCL_SUCCESS;
97 : }
98 :
99 0 : HcclResult AlltoallvContinuousPipeline::Prepare(const u32 userRank, const A2aPipelineMemory &a2aPipelineMemory,
100 : const SubCommInfo &level0CommInfo, const SubCommInfo &level1CommInfo,
101 : const Stream &mainStream, std::vector<Stream> &subStream,
102 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub,
103 : std::vector<SendRecvInfo> &sendRecvInfoList, const HcclDataType dataType,
104 : const HcclWorkflowMode workMode)
105 : {
106 : // 运行模式：当前只支持单算子
107 0 : workMode_ = workMode;
108 0 : CHK_PRT_RET(workMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE,
109 : HCCL_ERROR("[AlltoallvContinuousPipeline] This template support opbase mode only."),
110 : HCCL_E_INTERNAL);
111 :
112 : // 拓扑信息
113 0 : CHK_RET(PrepareTopoInfo(userRank, level0CommInfo, level1CommInfo));
114 :
115 : // 并发度暂定为1 - 不并发
116 0 : rdmaConcurrentNum_ = 1;
117 :
118 : // 数据类型
119 0 : dataType_ = dataType;
120 0 : unitSize_ = DataUnitSize(dataType_);
121 :
122 : // 内存
123 0 : inputMem_ = a2aPipelineMemory.userInput;
124 0 : outputMem_ = a2aPipelineMemory.userOutput;
125 0 : inBuffer_ = a2aPipelineMemory.cclInBuffer;
126 0 : outBuffer_ = a2aPipelineMemory.cclOutBuffer;
127 :
128 0 : flagAreaRefreshData_.resize(userRankSize_);
129 :
130 : // server内其他卡的recv counts
131 0 : intraRecvCounts_.resize(intraRankSize_);
132 0 : for (auto &countVec : intraRecvCounts_) {
133 0 : countVec.resize(userRankSize_);
134 : }
135 :
136 : // 流和notify
137 0 : mainStream_ = mainStream;
138 0 : CHK_RET(PartitionSubStreamsAndNotifies(subStream, notifyMain, notifySub));
139 :
140 : // 链路
141 0 : intraLinks_ = level0CommInfo.links;
142 0 : interLinks_ = level1CommInfo.links;
143 0 : HCCL_INFO("[AlltoallvContinuousPipeline][Prepare] Link info: interLinksNum[%u], intraLinksNum[%u]",
144 : interLinks_.size(), intraLinks_.size());
145 :
146 : // pingpong模式: module间只有1步：双module
147 0 : enablePingPong_ = rdmaConcurrentNum_ >= interRankSize_ - 1;
148 : // 切分buffer
149 0 : CHK_RET(SplitBuffer(enablePingPong_));
150 :
151 : // 收发信息
152 0 : CHK_RET(PrepareSendRecvInfo(sendRecvInfoList));
153 :
154 :
155 0 : return HCCL_SUCCESS;
156 : }
157 :
158 0 : HcclResult AlltoallvContinuousPipeline::SplitBuffer(const bool enablePingPong)
159 : {
160 : // 单个counts数组的大小
161 0 : const u64 singleRankCountsInfoSize = sizeof(u64) * userRankSize_;
162 : // 全局counts的大小
163 0 : const u64 globalCountsInfoSize = singleRankCountsInfoSize * userRankSize_;
164 :
165 0 : const u64 bufferSize = inBuffer_.size();
166 0 : u32 blockNum = userRankSize_;
167 :
168 0 : if (enablePingPong) {
169 : // 乒乓模式两倍分块
170 0 : blockNum = userRankSize_ * PINGPONG_MEM_NUM;
171 0 : HCCL_INFO("[AlltoallvContinuousPipeline][SplitBuffer] Use ping-pong mode.");
172 : }
173 :
174 0 : CHK_PRT_RET(blockNum == 0,
175 : HCCL_ERROR("[AlltoallvContinuousPipeline][SplitBuffer]Unexpected blockNum[%u].", blockNum), HCCL_E_INTERNAL);
176 :
177 : // 初始化用于记录buffer中每个分块当作存放了多少数据的vector
178 : // 如果是pingpong模式，需要两倍的大小
179 0 : inBufferDataSize_.resize(blockNum);
180 :
181 0 : u64 alignSize = HCCL_MIN_SLICE_ALIGN;
182 0 : const u64 minBufferSize = globalCountsInfoSize + alignSize * blockNum;
183 0 : CHK_PRT_RET(bufferSize < minBufferSize,
184 : HCCL_ERROR("[AlltoallvContinuousPipeline][SplitBuffer]Insufficient buffer size [%llu Byte]; it needs to be "
185 : "greater than [%llu Byte].", bufferSize, minBufferSize), HCCL_E_MEMORY);
186 :
187 0 : if (bufferSize > globalCountsInfoSize + HCCL_MIN_SLICE_ALIGN_910_93 * blockNum) {
188 0 : alignSize = HCCL_MIN_SLICE_ALIGN_910_93;
189 : }
190 :
191 0 : countsPerBlock_ = (((bufferSize - globalCountsInfoSize) / blockNum) /
192 0 : alignSize * alignSize) / unitSize_; // 前面已经可以保证countsPerBlock_大于0，不再检查
193 0 : sizePerBlock_ = countsPerBlock_ * unitSize_;
194 :
195 0 : for (u32 rank = 0; rank < userRankSize_; ++rank) {
196 0 : infoOffsets_.emplace_back(sizePerBlock_ * blockNum + singleRankCountsInfoSize * rank);
197 : }
198 :
199 0 : for (u32 blockIdx = 0; blockIdx < blockNum; ++blockIdx) {
200 0 : dataBlockOffsets_.emplace_back(sizePerBlock_ * blockIdx);
201 : }
202 :
203 0 : HCCL_INFO("[AlltoallvContinuousPipeline][SplitBuffer] Split buffer done, alignSize[%llu], sizePerBlock[%llu], "
204 : "countsPerBlock[%llu], blockNum[%u]", alignSize, sizePerBlock_, countsPerBlock_, blockNum);
205 0 : return HCCL_SUCCESS;
206 : }
207 :
208 0 : HcclResult AlltoallvContinuousPipeline::PartitionSubStreamsAndNotifies(const std::vector<Stream> &subStreams,
209 : const std::vector<std::shared_ptr<LocalNotify>> &signalMainToSub,
210 : const std::vector<std::shared_ptr<LocalNotify>> &signalSubToMain)
211 : {
212 0 : constexpr u32 DEVICE_EIGHT = 8;
213 0 : const u32 sdmaConcurrentNum = std::min(intraRankSize_ - 1, DEVICE_EIGHT);
214 0 : const u32 totalSubstreamSize = rdmaConcurrentNum_ + sdmaConcurrentNum;
215 0 : CHK_PRT_RET(subStreams.size() < totalSubstreamSize || signalMainToSub.size() < totalSubstreamSize ||
216 : signalSubToMain.size() < totalSubstreamSize,
217 : HCCL_ERROR("[AlltoallvContinuousPipeline] subStreams size [%u] or signalMainToSub size [%u] or signalSubToMain "
218 : "size [%u] is small than totalSubstreamSize [%u].",
219 : subStreams.size(),
220 : signalMainToSub.size(),
221 : signalSubToMain.size(),
222 : totalSubstreamSize),
223 : HCCL_E_PARA);
224 :
225 0 : u32 index = 0;
226 :
227 : // 用于SDMA通信的从流和主从同步notify
228 0 : for (u32 i = 0; i < sdmaConcurrentNum; ++i) {
229 0 : subStreams_.push_back(subStreams[index]);
230 0 : sdmaSubStreams_.push_back(subStreams[index]);
231 0 : streamNotifyMainToSdmaSub_.push_back(signalMainToSub[index]);
232 0 : streamNotifySdmaSubToMain_.push_back(signalSubToMain[index]);
233 0 : index++;
234 : }
235 :
236 : // 用于RDMA通信的从流和主从同步notify
237 0 : for (u32 i = 0; i < rdmaConcurrentNum_; ++i) {
238 0 : subStreams_.push_back(subStreams[index]);
239 0 : rdmaSubStreams_.push_back(subStreams[index]);
240 0 : streamNotifyMainToRdmaSub_.push_back(signalMainToSub[index]);
241 0 : streamNotifyRdmaSubToMain_.push_back(signalSubToMain[index]);
242 0 : index++;
243 : }
244 :
245 0 : HCCL_INFO("[AlltoallvContinuousPipeline][PartitionSubStreamsAndNotifies] Done, sdma: #streams[%zu], "
246 : "#notifyMainSub[%zu], #notifySubToMain[%zu]; rdma: #streams[%zu], #notifyMainSub[%zu], "
247 : "#notifySubToMain[%zu].",
248 : sdmaSubStreams_.size(),
249 : streamNotifyMainToSdmaSub_.size(),
250 : streamNotifySdmaSubToMain_.size(),
251 : rdmaSubStreams_.size(),
252 : streamNotifyMainToRdmaSub_.size(),
253 : streamNotifyRdmaSubToMain_.size());
254 :
255 0 : return HCCL_SUCCESS;
256 : }
257 :
258 0 : inline u32 AlltoallvContinuousPipeline::GetSdmaSubStreamIdx(const u32 remoteRank) const
259 : {
260 0 : return (remoteRank > intraRankId_ ? remoteRank - 1 : remoteRank) % sdmaSubStreams_.size();
261 : }
262 :
263 0 : inline u64 AlltoallvContinuousPipeline::GetLocalSendCountOfRank(const u32 targetRank) const
264 : {
265 0 : return localSendCounts_[targetRank];
266 : }
267 :
268 0 : inline u64 AlltoallvContinuousPipeline::GetLocalSendDisplOfRank(const u32 targetRank) const
269 : {
270 0 : return localSendDispls_[targetRank];
271 : }
272 :
273 0 : inline u64 AlltoallvContinuousPipeline::GetLocalRecvCountOfRank(const u32 sourceRank) const
274 : {
275 0 : return localRecvCounts_[sourceRank];
276 : }
277 :
278 0 : inline u64 AlltoallvContinuousPipeline::GetLocalRecvDisplOfRank(const u32 sourceRank) const
279 : {
280 0 : return localRecvDispls_[sourceRank];
281 : }
282 :
283 0 : inline u64 AlltoallvContinuousPipeline::GetDataBlockOffset(const u32 rank, const u32 bufferIdx) const
284 : {
285 0 : if (enablePingPong_) {
286 0 : return dataBlockOffsets_[(bufferIdx % PINGPONG_MEM_NUM) * userRankSize_ + rank];
287 : }
288 0 : return dataBlockOffsets_[rank];
289 : }
290 :
291 0 : u32 AlltoallvContinuousPipeline::GetLocalLoopNum() const
292 : {
293 0 : u64 maxCount = 0;
294 0 : for (u32 rank = 0; rank < userRankSize_; ++rank) {
295 0 : if (rank != userRank_) {
296 0 : const u64 sendCount = GetLocalSendCountOfRank(rank);
297 0 : maxCount = maxCount < sendCount ? sendCount : maxCount;
298 0 : const u64 recvCount = GetLocalRecvCountOfRank(rank);
299 0 : maxCount = maxCount < recvCount ? recvCount : maxCount;
300 : }
301 : }
302 :
303 : // 向上取整
304 0 : const u32 localLoopNum = static_cast<u32>((maxCount + countsPerBlock_ - 1) / countsPerBlock_);
305 0 : HCCL_INFO("[AlltoallvContinuousPipeline][GetLocalLoopNum] maxCount[%llu], localLoopNum[%u]",
306 : maxCount, localLoopNum);
307 0 : return localLoopNum;
308 : }
309 :
310 0 : HcclResult AlltoallvContinuousPipeline::UpdateLocalSendInfo(const u32 targetRank, const u64 count)
311 : {
312 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][UpdateLocalSendInfo]userRank[%u], count[%llu], before "
313 : "update, send info of rank[%u] is [count:%llu, displ:%llu].",
314 : userRank_,
315 : count,
316 : targetRank,
317 : localSendCounts_[targetRank],
318 : localSendDispls_[targetRank]);
319 :
320 0 : const u64 maxCount = std::min(localSendCounts_[targetRank], count);
321 0 : localSendCounts_[targetRank] -= maxCount;
322 0 : localSendDispls_[targetRank] += maxCount;
323 :
324 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][UpdateLocalSendInfo]userRank[%u], count[%llu], after "
325 : "update, send info of rank[%u] is [count:%llu, displ:%llu].",
326 : userRank_,
327 : count,
328 : targetRank,
329 : localSendCounts_[targetRank],
330 : localSendDispls_[targetRank]);
331 :
332 0 : return HCCL_SUCCESS;
333 : }
334 :
335 0 : HcclResult AlltoallvContinuousPipeline::UpdateLocalRecvInfo(const u32 sourceRank, const u64 count)
336 : {
337 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][UpdateLocalRecvInfo]userRank[%u], count[%llu], before "
338 : "update, receive info of rank[%u] is [count:%llu, displ:%llu].",
339 : userRank_,
340 : count,
341 : sourceRank,
342 : localRecvCounts_[sourceRank],
343 : localRecvDispls_[sourceRank]);
344 :
345 0 : const u64 maxCount = std::min(localRecvCounts_[sourceRank], count);
346 0 : localRecvCounts_[sourceRank] -= maxCount;
347 0 : localRecvDispls_[sourceRank] += maxCount;
348 :
349 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][UpdateLocalRecvInfo]userRank[%u], count[%llu], after "
350 : "update, receive info of rank[%u] is [count:%llu, displ:%llu].",
351 : userRank_,
352 : count,
353 : sourceRank,
354 : localRecvCounts_[sourceRank],
355 : localRecvDispls_[sourceRank]);
356 :
357 0 : return HCCL_SUCCESS;
358 : }
359 :
360 0 : HcclResult AlltoallvContinuousPipeline::NotifySdmaSubStreamStart()
361 : {
362 0 : for (u32 streamIndex = 0; streamIndex < sdmaSubStreams_.size(); ++streamIndex) {
363 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifySdmaSubToMain_[streamIndex],
364 : INVALID_VALUE_STAGE));
365 0 : CHK_RET(LocalNotify::Wait(
366 : sdmaSubStreams_[streamIndex], dispatcher_, streamNotifySdmaSubToMain_[streamIndex], INVALID_VALUE_STAGE));
367 : }
368 0 : return HCCL_SUCCESS;
369 : }
370 :
371 0 : HcclResult AlltoallvContinuousPipeline::WaitSdmaSubStreamFinish()
372 : {
373 0 : for (u32 streamIndex = 0; streamIndex < sdmaSubStreams_.size(); ++streamIndex) {
374 0 : CHK_RET(LocalNotify::Post(sdmaSubStreams_[streamIndex], dispatcher_, streamNotifyMainToSdmaSub_[streamIndex],
375 : INVALID_VALUE_STAGE));
376 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMainToSdmaSub_[streamIndex],
377 : INVALID_VALUE_STAGE));
378 : }
379 0 : return HCCL_SUCCESS;
380 : }
381 :
382 0 : HcclResult AlltoallvContinuousPipeline::NotifyRdmaSubStreamStart()
383 : {
384 0 : for (u32 streamIndex = 0; streamIndex < rdmaSubStreams_.size(); ++streamIndex) {
385 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifyRdmaSubToMain_[streamIndex],
386 : INVALID_VALUE_STAGE));
387 0 : CHK_RET(LocalNotify::Wait(
388 : rdmaSubStreams_[streamIndex], dispatcher_, streamNotifyRdmaSubToMain_[streamIndex], INVALID_VALUE_STAGE));
389 : }
390 0 : return HCCL_SUCCESS;
391 : }
392 :
393 0 : HcclResult AlltoallvContinuousPipeline::WaitRdmaSubStreamFinish()
394 : {
395 0 : for (u32 streamIndex = 0; streamIndex < rdmaSubStreams_.size(); ++streamIndex) {
396 0 : CHK_RET(LocalNotify::Post(rdmaSubStreams_[streamIndex], dispatcher_, streamNotifyMainToRdmaSub_[streamIndex],
397 : INVALID_VALUE_STAGE));
398 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMainToRdmaSub_[streamIndex],
399 : INVALID_VALUE_STAGE));
400 : }
401 0 : return HCCL_SUCCESS;
402 : }
403 :
404 0 : HcclResult AlltoallvContinuousPipeline::InterSdmaRx(const LINK& linkLeft, const LINK& linkRight,
405 : std::vector<TxMemoryInfo>& sendMems, std::vector<RxMemoryInfo>& recvMems, Stream& stream)
406 : {
407 0 : const bool needRecvFromLinkLeft = !recvMems.empty();
408 0 : const bool needSendToLinkRight = !sendMems.empty();
409 :
410 : // 前同步，通知right我已准备好，可以从我这里读；等待left通知它已准备好，可以从它那里读
411 0 : if (needSendToLinkRight) {
412 0 : CHK_RET(linkRight->TxAck(stream));
413 : }
414 0 : if (needRecvFromLinkLeft) {
415 0 : CHK_RET(linkLeft->RxAck(stream));
416 : }
417 :
418 : // 从left读
419 0 : for (const auto& memInfo : recvMems) {
420 0 : void *srcMemPtr = nullptr;
421 0 : CHK_RET(linkLeft->GetRemoteMem(memInfo.srcMemType, &srcMemPtr));
422 0 : DeviceMem dstMem = DeviceMem::create(memInfo.dst, memInfo.len);
423 0 : DeviceMem srcMem(static_cast<s8 *>(srcMemPtr) + memInfo.srcOffset, memInfo.len);
424 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, stream, linkLeft->GetRemoteRank(),
425 : linkLeft->GetLinkType()));
426 0 : }
427 :
428 : // 尾同步，通知left我已读完，等待right通知它已读完
429 0 : if (needRecvFromLinkLeft) {
430 0 : CHK_RET(linkLeft->TxDataSignal(stream));
431 : }
432 0 : if (needSendToLinkRight) {
433 0 : CHK_RET(linkRight->RxDataSignal(stream));
434 : }
435 :
436 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][InterSdmaRx] Done. linkLeft.rank[%u], linkRight.rank[%u], "
437 : "recvMems.size[%zu]", linkLeft->GetRemoteRank(), linkRight->GetRemoteRank(), recvMems.size());
438 0 : return HCCL_SUCCESS;
439 : }
440 :
441 : // 跨module通信，通过RDMA从link left读或向link right写
442 0 : HcclResult AlltoallvContinuousPipeline::InterRdmaTxRx(const LINK& linkLeft, const LINK& linkRight,
443 : std::vector<TxMemoryInfo>& sendMems, std::vector<RxMemoryInfo>& recvMems, Stream& stream)
444 : {
445 0 : const bool needRecvFromLinkLeft = !recvMems.empty();
446 0 : const bool needSendToLinkRight = !sendMems.empty();
447 :
448 0 : if (needRecvFromLinkLeft) {
449 0 : CHK_RET(linkLeft->TxAck(stream));
450 : }
451 0 : if (needSendToLinkRight) {
452 0 : CHK_RET(linkRight->RxAck(stream));
453 0 : CHK_RET(linkRight->TxAsync(sendMems, stream));
454 : }
455 0 : if (needRecvFromLinkLeft) {
456 0 : CHK_RET(linkLeft->RxAsync(recvMems, stream));
457 0 : CHK_RET(linkLeft->PostFinAck(stream));
458 : }
459 0 : if (needSendToLinkRight) {
460 0 : CHK_RET(linkRight->WaitFinAck(stream));
461 : }
462 :
463 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][InterRdmaTxRx] Done. linkLeft.rank[%u], linkRight.rank[%u], "
464 : "sendMems.size[%zu], recvMems.size[%zu]", linkLeft->GetRemoteRank(), linkRight->GetRemoteRank(),
465 : sendMems.size(), recvMems.size());
466 0 : return HCCL_SUCCESS;
467 : }
468 :
469 0 : HcclResult AlltoallvContinuousPipeline::LocalCopyFromInputToInBuffer(const u32 targetRank, Stream& stream,
470 : const u32 loopIdx)
471 : {
472 : // 根据send displs来计算input的位置，取min(countsPerBlock_, count)个数
473 0 : const u64 copyCount = std::min(GetLocalSendCountOfRank(targetRank), countsPerBlock_);
474 0 : if (copyCount == 0) {
475 0 : return HCCL_SUCCESS;
476 : }
477 :
478 0 : const u64 copySize = copyCount * unitSize_;
479 :
480 : // 从input拷贝到in buffer对应的分块里
481 0 : const u64 srcOffset = GetLocalSendDisplOfRank(targetRank) * unitSize_;
482 0 : const u64 dstOffset = GetDataBlockOffset(targetRank, loopIdx);
483 0 : DeviceMem src = inputMem_.range(srcOffset, copySize);
484 0 : DeviceMem dst = inBuffer_.range(dstOffset, copySize);
485 :
486 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream));
487 :
488 : // 刷新send info
489 0 : CHK_RET(UpdateLocalSendInfo(targetRank, copyCount));
490 :
491 : // 记录in bufer中该分块存放了多少数据
492 0 : inBufferDataSize_[targetRank] = copySize;
493 :
494 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][LocalCopy][FromInputToInBuffer] done, userRank[%u], targetRank[%u], "
495 : "srcOffset[%llu], dstOffset[%llu], copyCount[%llu], copySize[%llu], loopIdx[%u]", userRank_, targetRank,
496 : srcOffset, dstOffset, copyCount, copySize, loopIdx);
497 0 : return HCCL_SUCCESS;
498 0 : }
499 :
500 0 : HcclResult AlltoallvContinuousPipeline::LocalCopyFromOutBufferToOutput(const u32 sourceRank, Stream& stream,
501 : const u32 loopIdx)
502 : {
503 0 : const u64 copyCount = std::min(GetLocalRecvCountOfRank(sourceRank), countsPerBlock_);
504 0 : if (copyCount == 0) {
505 0 : return HCCL_SUCCESS;
506 : }
507 :
508 : // 从out buffer对应分块拷贝到output
509 0 : const u64 copySize = copyCount * unitSize_;
510 0 : const u64 srcOffset = GetDataBlockOffset(sourceRank, loopIdx);
511 0 : const u64 dstOffset = GetLocalRecvDisplOfRank(sourceRank) * unitSize_;
512 0 : DeviceMem src = outBuffer_.range(srcOffset, copySize);
513 0 : DeviceMem dst = outputMem_.range(dstOffset, copySize);
514 :
515 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream));
516 :
517 : // 刷新receive info
518 0 : CHK_RET(UpdateLocalRecvInfo(sourceRank, copyCount));
519 :
520 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][LocalCopy][FromOutBufferToOutput] done, userRank[%u], sourceRank[%u], "
521 : "srcOffset[%llu], dstOffset[%llu], copyCount[%llu], copySize[%llu], loopIdx[%u]", userRank_, sourceRank,
522 : srcOffset, dstOffset, copyCount, copySize, loopIdx);
523 0 : return HCCL_SUCCESS;
524 0 : }
525 :
526 0 : HcclResult AlltoallvContinuousPipeline::LocalCopySelfDataFromInputToOutput(Stream& stream)
527 : {
528 0 : const u64 copyCount = GetLocalSendCountOfRank(userRank_);
529 0 : if (copyCount == 0) {
530 0 : return HCCL_SUCCESS;
531 : }
532 :
533 : // 从input拷贝到output
534 0 : const u64 copySize = copyCount * unitSize_;
535 0 : const u64 srcOffset = GetLocalSendDisplOfRank(userRank_) * unitSize_;
536 0 : const u64 dstOffset = GetLocalRecvDisplOfRank(userRank_) * unitSize_;
537 0 : DeviceMem src = inputMem_.range(srcOffset, copySize);
538 0 : DeviceMem dst = outputMem_.range(dstOffset, copySize);
539 :
540 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream));
541 :
542 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][LocalCopy][SelfDataFromInputToOutput] done, userRank[%u], "
543 : "srcOffset[%llu], dstOffset[%llu], copyCount[%llu], copySize[%llu]",
544 : userRank_, srcOffset, dstOffset, copyCount, copySize);
545 0 : return HCCL_SUCCESS;
546 0 : }
547 :
548 0 : HcclResult AlltoallvContinuousPipeline::SdmaSendFromInputToRemoteOutBuffer(const u32 targetRank, Stream& stream,
549 : const u32 loopIdx)
550 : {
551 0 : const u64 sendCount = std::min(GetLocalSendCountOfRank(targetRank), countsPerBlock_);
552 0 : if (sendCount == 0) {
553 0 : return HCCL_SUCCESS;
554 : }
555 :
556 : // 从input发送到remote out buffer，目的位置是第[本userRank_]个分块
557 0 : const u64 sendSize = sendCount * unitSize_;
558 0 : const u64 srcOffset = GetLocalSendDisplOfRank(targetRank) * unitSize_;
559 0 : const u64 dstOffset = GetDataBlockOffset(userRank_, loopIdx);
560 0 : DeviceMem src = inputMem_.range(srcOffset, sendSize);
561 :
562 0 : const LINK& link = intraLinks_[targetRank % intraRankSize_];
563 0 : void *remMemPtr = nullptr;
564 0 : CHK_RET(link->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
565 0 : DeviceMem dst = DeviceMem::create(static_cast<u8 *>(remMemPtr) + dstOffset, sendSize);
566 :
567 : // 前后同步在外层处理
568 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream, targetRank, link->GetLinkType()));
569 :
570 : // 刷新send info
571 0 : CHK_RET(UpdateLocalSendInfo(targetRank, sendCount));
572 :
573 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][Sdma][SendFromInputToRemoteOutBuffer] done, userRank[%u], "
574 : "targetRank[%u], srcOffset[%llu], dstOffset[%llu], sendCount[%llu], sendSize[%llu], loopIdx[%u]",
575 : userRank_, targetRank, srcOffset, dstOffset, sendCount, sendSize, loopIdx);
576 0 : return HCCL_SUCCESS;
577 0 : }
578 :
579 0 : HcclResult AlltoallvContinuousPipeline::SdmaReadFromRemoteOutBufferToOutput(const u32 sourceRank, Stream& stream,
580 : const u32 loopIdx)
581 : {
582 : // 需要recv counts信息
583 0 : CHK_PRT_RET(needCollectInfo_,
584 : HCCL_ERROR("[AlltoallvContinuousPipeline][SdmaReadFromRemoteOutBufferToOutput] No receive info."),
585 : HCCL_E_INTERNAL);
586 :
587 0 : const u64 readCount = std::min(GetLocalRecvCountOfRank(sourceRank), countsPerBlock_);
588 0 : if (readCount == 0) {
589 0 : return HCCL_SUCCESS;
590 : }
591 :
592 : // 从remote out buffer读取到output，源位置是第[sourceRank / intraRankSize_ * intraRankSize_ + intraRankId_]个分块
593 0 : const u64 readSize = readCount * unitSize_;
594 0 : const u64 srcBlockIdx = sourceRank / intraRankSize_ * intraRankSize_ + intraRankId_;
595 0 : const u64 srcOffset = GetDataBlockOffset(srcBlockIdx, loopIdx);
596 0 : const u64 dstOffset = GetLocalRecvDisplOfRank(sourceRank) * unitSize_;
597 :
598 0 : const LINK& link = intraLinks_[sourceRank % intraRankSize_];
599 0 : void *remMemPtr = nullptr;
600 0 : CHK_RET(link->GetRemoteMem(UserMemType::OUTPUT_MEM, &remMemPtr));
601 0 : DeviceMem src = DeviceMem::create(static_cast<u8 *>(remMemPtr) + srcOffset, readSize);
602 :
603 0 : DeviceMem dst = outputMem_.range(dstOffset, readSize);
604 :
605 : // 前后同步在外层处理
606 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream, sourceRank, link->GetLinkType()));
607 :
608 : // 刷新receive info
609 0 : CHK_RET(UpdateLocalRecvInfo(sourceRank, readCount));
610 :
611 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][Sdma][ReadFromRemoteOutBufferToOutput] done, userRank[%u], "
612 : "sourceRank[%u], srcOffset[%llu], dstOffset[%llu], readCount[%llu], readSize[%llu], loopIdx[%u]",
613 : userRank_, sourceRank, srcOffset, dstOffset, readCount, readSize, loopIdx);
614 0 : return HCCL_SUCCESS;
615 0 : }
616 :
617 0 : HcclResult AlltoallvContinuousPipeline::InterSendAndReceive(const u32 sendRank, const u32 recvRank, Stream& stream,
618 : const u32 loopIdx)
619 : {
620 : // 计算出对端rank所在module的首rank，因为要向对端rank发送[首rank, 首rank+intraRankSize]rank的数据
621 0 : const u32 sendModuleFirstRank = sendRank / intraRankSize_ * intraRankSize_;
622 0 : const u32 recvModuleFirstRank = recvRank / intraRankSize_ * intraRankSize_;
623 :
624 0 : std::vector<TxMemoryInfo> sendMems;
625 0 : std::vector<RxMemoryInfo> recvMems;
626 0 : sendMems.reserve(intraRankSize_);
627 0 : recvMems.reserve(intraRankSize_);
628 :
629 0 : const LINK& sendLink = interLinks_[sendRank / intraRankSize_];
630 0 : const LINK& recvLink = interLinks_[recvRank / intraRankSize_];
631 0 : const bool isSDMALink = sendLink->IsSpInlineReduce() || recvLink->IsSpInlineReduce();
632 :
633 0 : for (u32 rankOffset = 0; rankOffset < intraRankSize_; ++rankOffset) {
634 0 : const u32 targetRank = sendModuleFirstRank + rankOffset;
635 0 : const u64 sendSrcOffset = GetDataBlockOffset(targetRank, loopIdx);
636 0 : const u64 sendDstOffset = GetDataBlockOffset(interRankId_ * intraRankSize_ + rankOffset, loopIdx);
637 0 : const u64 sendSize = inBufferDataSize_[targetRank];
638 0 : inBufferDataSize_[targetRank] = 0;
639 0 : if (sendSize > 0) {
640 0 : sendMems.emplace_back(TxMemoryInfo{UserMemType::OUTPUT_MEM, sendDstOffset,
641 0 : static_cast<s8*>(inBuffer_.ptr()) + sendSrcOffset, sendSize});
642 : }
643 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][InterSendAndReceive]inter send userRank[%u], sendRank[%u], "
644 : "targetRank[%u], srcOffset[%llu], dstOffset[%llu], sendSize[%llu], loopIdx[%u]",
645 : userRank_, sendRank, targetRank, sendSrcOffset, sendDstOffset, sendSize, loopIdx);
646 :
647 0 : const u32 sourceRank = recvModuleFirstRank + intraRankId_;
648 0 : const u32 actualTargetRank = interRankId_ * intraRankSize_ + rankOffset;
649 0 : const u64 recvSrcOffset = GetDataBlockOffset(actualTargetRank, loopIdx);
650 0 : const u64 recvDstOffset = GetDataBlockOffset(recvModuleFirstRank + rankOffset, loopIdx);
651 0 : const u64 recvCount = std::min(countsPerBlock_, intraRecvCounts_[rankOffset][sourceRank]);
652 0 : const u64 recvSize = recvCount * unitSize_;
653 0 : if (recvCount > 0) {
654 0 : recvMems.emplace_back(RxMemoryInfo{UserMemType::INPUT_MEM, recvSrcOffset,
655 0 : static_cast<s8*>(outBuffer_.ptr()) + recvDstOffset, recvSize});
656 0 : intraRecvCounts_[rankOffset][sourceRank] -= recvCount;
657 : }
658 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][InterSendAndReceive]inter recv userRank[%u], recvRank[%u], "
659 : "sourceRank[%u], targetRank[%u], srcOffset[%llu], dstOffset[%llu], readSize[%llu], loopIdx[%u]",
660 : userRank_, recvRank, sourceRank, actualTargetRank, recvSrcOffset, recvDstOffset, recvSize, loopIdx);
661 : }
662 0 : if (isSDMALink) {
663 : // SDMA读
664 0 : CHK_RET(InterSdmaRx(recvLink, sendLink, sendMems, recvMems, stream));
665 : } else {
666 : // RDMA
667 0 : CHK_RET(InterRdmaTxRx(recvLink, sendLink, sendMems, recvMems, stream));
668 : }
669 :
670 0 : return HCCL_SUCCESS;
671 0 : }
672 :
673 0 : HcclResult AlltoallvContinuousPipeline::DoSdmaSync(const SdmaSyncType syncType)
674 : {
675 0 : for (u32 rank = 0; rank < intraRankSize_; ++rank) {
676 0 : if (rank == intraRankId_) {
677 0 : continue;
678 : }
679 0 : const u32 streamIndex = GetSdmaSubStreamIdx(rank);
680 0 : const LINK& link = intraLinks_[rank];
681 0 : Stream& subStream = sdmaSubStreams_[streamIndex];
682 0 : if (syncType == SdmaSyncType::PRE_SYNC) {
683 : // 前同步
684 0 : CHK_RET(link->TxAck(subStream));
685 0 : CHK_RET(link->RxAck(subStream));
686 : } else {
687 : // 尾同步
688 0 : CHK_RET(link->TxDataSignal(subStream));
689 0 : CHK_RET(link->RxDataSignal(subStream));
690 : }
691 : }
692 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoSdmaSync] Sync done, syncType[%d].", syncType);
693 0 : return HCCL_SUCCESS;
694 : }
695 :
696 0 : HcclResult AlltoallvContinuousPipeline::DoLocalCopy(const u32 beginStepNum, const u32 endStepNum, const u32 loopIdx)
697 : {
698 0 : CHK_PRT_RET(beginStepNum == endStepNum,
699 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoLocalCopy]beginStepNum[%u] == endStepNum[%u], return success.",
700 : beginStepNum, endStepNum),
701 : HCCL_SUCCESS);
702 :
703 0 : for (u32 step = beginStepNum + 1; step < endStepNum + 1; ++step) {
704 0 : const u32 sendModuleId = (interRankId_ + step) % interRankSize_;
705 0 : for (const auto remoteRank : ranksPerModule_[sendModuleId]) {
706 0 : CHK_RET(LocalCopyFromInputToInBuffer(remoteRank, mainStream_, loopIdx));
707 : }
708 : }
709 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoLocalCopy] done. beginStepNum[%u], endStepNum[%u], loopIdx[%u]",
710 : beginStepNum, endStepNum, loopIdx);
711 0 : return HCCL_SUCCESS;
712 : }
713 :
714 0 : HcclResult AlltoallvContinuousPipeline::DoIntraDistribution(const u32 beginStepNum, const u32 endStepNum,
715 : const u32 loopIdx)
716 : {
717 0 : CHK_PRT_RET(beginStepNum == endStepNum,
718 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoIntraDistribution]beginStepNum[%u] == endStepNum[%u], return "
719 : "success.", beginStepNum, endStepNum),
720 : HCCL_SUCCESS);
721 :
722 0 : for (u32 step = beginStepNum + 1; step < endStepNum + 1; ++step) {
723 0 : const u32 recvModuleId = (interRankId_ + interRankSize_ - step) % interRankSize_;
724 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoIntraDistribution] recvModuleId[%u]", recvModuleId);
725 0 : for (const auto remoteRank : ranksPerModule_[recvModuleId]) {
726 0 : const u32 remoteIntraRank = remoteRank % intraRankSize_;
727 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoIntraDistribution] remoteIntraRank[%u]", remoteIntraRank);
728 0 : if (intraRankId_ == remoteIntraRank) {
729 : // 如果是同号卡，直接从out buffer拷贝到output
730 0 : CHK_RET(LocalCopyFromOutBufferToOutput(remoteRank, mainStream_, loopIdx));
731 : } else {
732 : // 如果不是同号卡，从module内它对应的同号卡获取
733 0 : const u32 streamIndex = GetSdmaSubStreamIdx(remoteIntraRank);
734 0 : HCCL_DEBUG(
735 : "[AlltoallvContinuousPipeline][DoIntraDistribution] streamIndex[%u], sdmaSubStreams_.size()[%zu]",
736 : streamIndex, sdmaSubStreams_.size());
737 0 : Stream& subStream = sdmaSubStreams_[streamIndex];
738 0 : CHK_RET(SdmaReadFromRemoteOutBufferToOutput(remoteRank, subStream, loopIdx));
739 : }
740 : }
741 : }
742 :
743 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoIntraDistribution] done. beginStepNum[%u], endStepNum[%u], loopIdx[%u]",
744 : beginStepNum, endStepNum, loopIdx);
745 0 : return HCCL_SUCCESS;
746 : }
747 :
748 0 : HcclResult AlltoallvContinuousPipeline::DoInterSendReceive(const u32 beginStepNum, const u32 endStepNum,
749 : const u32 loopIdx)
750 : {
751 0 : CHK_PRT_RET(beginStepNum == endStepNum,
752 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoInterSendReceive]beginStepNum[%u] == endStepNum[%u], return "
753 : "success.", beginStepNum, endStepNum),
754 : HCCL_SUCCESS);
755 :
756 0 : u32 streamIdx = 0;
757 0 : for (u32 step = beginStepNum + 1; step < endStepNum + 1; ++step) {
758 0 : const u32 sendRank = (userRank_ + step * intraRankSize_) % userRankSize_;
759 0 : const u32 recvRank = (userRank_ + userRankSize_ - step * intraRankSize_) % userRankSize_;
760 0 : CHK_RET(InterSendAndReceive(sendRank, recvRank, rdmaSubStreams_[streamIdx++], loopIdx));
761 : }
762 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoInterSendReceive] done. beginStepNum[%u], endStepNum[%u], loopIdx[%u]",
763 : beginStepNum, endStepNum, loopIdx);
764 0 : return HCCL_SUCCESS;
765 : }
766 :
767 0 : HcclResult AlltoallvContinuousPipeline::DoLevel0LocalCopy(const u32 loopIdx)
768 : {
769 0 : for (const auto remoteRank : ranksPerModule_[interRankId_]) {
770 0 : if (remoteRank == userRank_) {
771 0 : continue;
772 : }
773 0 : CHK_RET(LocalCopyFromOutBufferToOutput(remoteRank, mainStream_, loopIdx));
774 : }
775 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoLevel0LocalCopy] done, loopIdx[%u].", loopIdx);
776 0 : return HCCL_SUCCESS;
777 : }
778 :
779 0 : HcclResult AlltoallvContinuousPipeline::DoLevel0SdmaSend(const u32 loopIdx)
780 : {
781 0 : for (const auto remoteRank : ranksPerModule_[interRankId_]) {
782 0 : if (remoteRank == userRank_) {
783 0 : continue;
784 : }
785 0 : const u32 remoteIntraRank = remoteRank % intraRankSize_;
786 0 : const u32 streamIndex = GetSdmaSubStreamIdx(remoteIntraRank);
787 0 : Stream& subStream = sdmaSubStreams_[streamIndex];
788 0 : CHK_RET(SdmaSendFromInputToRemoteOutBuffer(remoteRank, subStream, loopIdx));
789 : }
790 :
791 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoLevel0SdmaSend] done, loopIdx[%u].", loopIdx);
792 0 : return HCCL_SUCCESS;
793 : }
794 :
795 0 : HcclResult AlltoallvContinuousPipeline::DoIntraInfoBroadcast()
796 : {
797 0 : for (const auto remoteRank : ranksPerModule_[interRankId_]) {
798 0 : if (remoteRank == userRank_) {
799 0 : continue;
800 : }
801 0 : const u32 remoteIntraRank = remoteRank % intraRankSize_;
802 0 : const u32 streamIndex = GetSdmaSubStreamIdx(remoteIntraRank);
803 0 : Stream& subStream = sdmaSubStreams_[streamIndex];
804 :
805 0 : const LINK& link = intraLinks_[remoteRank % intraRankSize_];
806 0 : void *remInPtr = nullptr;
807 0 : void *remOutPtr = nullptr;
808 0 : CHK_RET(link->GetRemoteMem(UserMemType::INPUT_MEM, &remInPtr));
809 0 : CHK_RET(link->GetRemoteMem(UserMemType::OUTPUT_MEM, &remOutPtr));
810 :
811 : // 前后同步在外层处理，直接发送
812 : // 发送counts信息，从output发送到remote out buffer，目的位置是第[本userRank_]个info分块
813 0 : const u64 infoSize = userRankSize_ * sizeof(u64);
814 0 : const u64 infoOffset = infoOffsets_[userRank_];
815 0 : DeviceMem infoSrc = outBuffer_.range(infoOffset, infoSize);
816 0 : DeviceMem infoDst = DeviceMem::create(static_cast<u8 *>(remOutPtr) + infoOffset, infoSize);
817 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, infoDst, infoSrc, subStream, remoteRank, link->GetLinkType()));
818 :
819 : // 发送flag，flag值为[LocalLoopNum + 1]，从input发送到remote in buffer，目的位置是第[本userRank_]个u32
820 0 : const u64 flagSize = sizeof(u32);
821 0 : const u64 flagOffset = infoOffsets_[0] + userRank_ * flagSize;
822 0 : DeviceMem flagSrc = inBuffer_.range(flagOffset, flagSize);
823 0 : DeviceMem flagDst = DeviceMem::create(static_cast<u8 *>(remInPtr) + flagOffset, flagSize);
824 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, flagDst, flagSrc, subStream, remoteRank, link->GetLinkType()));
825 :
826 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoIntraInfoBroadcast] userRank[%u], send info to remoteRank[%u], "
827 : "infoOffset[%llu], infoSize[%llu], flagOffset[%llu], flagSize[%llu]",
828 : userRank_, remoteRank, infoOffset, infoSize, flagOffset, flagSize);
829 0 : }
830 :
831 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoIntraInfoBroadcast] done.");
832 0 : return HCCL_SUCCESS;
833 : }
834 :
835 0 : HcclResult AlltoallvContinuousPipeline::DoLocalWriteInfoAndFlagAndInterSync()
836 : {
837 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][DoLocalWriteInfoAndFlagAndInterSync] start.");
838 :
839 0 : if (!needCollectInfo_) {
840 0 : return HCCL_SUCCESS;
841 : }
842 :
843 : // 将counts信息写到out buffer的info区域
844 0 : void* infoPtr = localRecvCounts_.data();
845 0 : const u64 infoSize = userRankSize_ * sizeof(u64);
846 0 : const u64 infoOffset = infoOffsets_[userRank_];
847 0 : DeviceMem infoSrc = DeviceMem::create(infoPtr, infoSize);
848 0 : DeviceMem infoDst = outBuffer_.range(infoOffset, infoSize);
849 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, infoDst, infoSrc, mainStream_));
850 :
851 : // 将in buffer的flag区域刷0，第[userRank]个u32设为[LocalLoopNum + 1]
852 0 : const u64 flagAreaSize = userRankSize_ * sizeof(u32);
853 0 : flagAreaRefreshData_[userRank_] = intraLoopNum_ + 1; // +1是应对LocalLoopNum为0的情况
854 0 : DeviceMem flagSrc = DeviceMem::create(flagAreaRefreshData_.data(), flagAreaSize);
855 0 : DeviceMem flagDst = inBuffer_.range(infoOffsets_[0], flagAreaSize);
856 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, flagDst, flagSrc, mainStream_));
857 :
858 : // 在主流上下一个搬1的任务，kernel可以通过轮询dst是否为1，确保flag区域已被刷值，避免flag区域还是随机值时就开始轮询。
859 0 : DeviceMem refreshFlagSrc = DeviceMem::create(&flagAreaRefreshValue_, sizeof(flagAreaRefreshValue_));
860 0 : DeviceMem refreshFlagDst = DeviceMem::create(&flagAreaRefreshFlag_, sizeof(flagAreaRefreshFlag_));
861 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, refreshFlagDst, refreshFlagSrc, mainStream_));
862 :
863 0 : CHK_RET(LaunchTask(dispatcher_, mainStream_));
864 :
865 0 : HCCL_INFO("[AlltoallvContinuousPipeline][DoLocalWriteInfoAndFlagAndInterSync] write counts and flag. userRank[%u], "
866 : "infoPtr[%p], infoSize[%llu], infoOffset[%llu]",
867 : userRank_, infoPtr, infoSize, infoOffset);
868 0 : return HCCL_SUCCESS;
869 0 : }
870 :
871 0 : HcclResult AlltoallvContinuousPipeline::WaitValueOfRank(const u32 rank, const HcclUs &startTimeUs, u32 &value)
872 : {
873 0 : const auto* valuePtr = reinterpret_cast<u32 *>(static_cast<u8 *>(inBuffer_.ptr()) + infoOffsets_[0]) + rank;
874 0 : HcclUs lastUt = startTimeUs;
875 0 : constexpr s64 timeout = 1800 * 1000 * 1000; // 超时时间暂定为1800s
876 0 : constexpr s64 printStateInterval = 30 * 1000 * 1000; // 每隔30s打印一次状态
877 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitValueOfRank] start waiting value of rank[%u], valuePtr[%p].",
878 : rank, valuePtr);
879 :
880 0 : while (flagAreaRefreshFlag_ == 0 || *valuePtr == 0) {
881 0 : const HcclUs currentUt = TIME_NOW();
882 : // 等待value过程，每隔30秒打印一次状态
883 0 : if (DURATION_US(currentUt - lastUt).count() > printStateInterval) {
884 0 : lastUt = currentUt;
885 0 : if (flagAreaRefreshFlag_ == 0) {
886 0 : HCCL_RUN_INFO("[AlltoallvContinuousPipeline][WaitValueOfRank] The Previous task has not been completed."
887 : " userRank[%u]", userRank_);
888 : } else {
889 0 : HCCL_RUN_INFO("[AlltoallvContinuousPipeline][WaitValueOfRank] waiting value of rank[%u]", rank);
890 : }
891 : }
892 :
893 0 : CHK_PRT_RET(DURATION_US(currentUt - startTimeUs).count() > timeout,
894 : HCCL_ERROR("[AlltoallvContinuousPipeline][WaitValueOfRank] Waiting for the value of rank[%u] timed out.",
895 : rank),
896 : HCCL_E_TIMEOUT);
897 : }
898 0 : value = *valuePtr;
899 :
900 : // 每次执行算子开头都有重置flag区域的task，所以此处不需要重置为0
901 :
902 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitValueOfRank] Got value of rank[%u], value[%u].", rank, value);
903 0 : return HCCL_SUCCESS;
904 : }
905 :
906 0 : HcclResult AlltoallvContinuousPipeline::WaitAndCalReceiveInfo()
907 : {
908 : // 计算自己以及module内其他卡的receive count
909 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitAndCalReceiveInfo] start.");
910 0 : const HcclUs startUt = TIME_NOW();
911 :
912 0 : for (u32 intraRankIdx = 0; intraRankIdx < intraRankSize_; ++intraRankIdx) {
913 0 : if (intraRankIdx == intraRankId_) {
914 0 : continue;
915 : }
916 0 : const u32 remoteRank = interRankId_ * intraRankSize_ + intraRankIdx;
917 0 : u32 remoteValue = 0;
918 0 : CHK_RET(WaitValueOfRank(remoteRank, startUt, remoteValue));
919 :
920 : const auto *countsPtr =
921 0 : reinterpret_cast<u64 *>(static_cast<u8 *>(outBuffer_.ptr()) + infoOffsets_[remoteRank]);
922 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitAndCalReceiveInfo] remoteRank[%u], infoOffset[%llu], "
923 : "countsPtr[%p]", remoteRank, infoOffsets_[remoteRank], countsPtr);
924 :
925 0 : for (u32 i = 0; i < userRankSize_; ++i) {
926 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitAndCalReceiveInfo] countsPtr[%u]=[%llu]", i, countsPtr[i]);
927 0 : intraRecvCounts_[intraRankIdx][i] = countsPtr[i];
928 : }
929 :
930 0 : const u32 remoteLoopNum = remoteValue - 1; // remoteValue一定大于0
931 0 : intraLoopNum_ = std::max(intraLoopNum_, remoteLoopNum);
932 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitAndCalReceiveInfo] remoteRank[%u], remoteLoopNum[%u], "
933 : "intraLoopNum_[%u]", remoteRank, remoteLoopNum, intraLoopNum_);
934 : }
935 :
936 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][WaitAndCalReceiveInfo] done. loopNum[%u]", intraLoopNum_);
937 0 : return HCCL_SUCCESS;
938 : }
939 :
940 0 : HcclResult AlltoallvContinuousPipeline::RunAsync()
941 : {
942 : // 在开始前，先将counts信息拷贝到info区域，并且刷新一下flag区域
943 0 : CHK_RET(DoLocalWriteInfoAndFlagAndInterSync());
944 :
945 : // 按照机间pairwise的方式计算每轮的步数，等于level1的rank size - 1
946 0 : const u32 stepsPerLoop = interRankSize_ - 1;
947 :
948 : // 需要发给其他module的每块数据都会经历三步：本地拷贝至in buffer、经RDMA链路发送到同号卡、由同号卡用SDMA分发到接收卡
949 0 : TaskState localCopyState;
950 0 : TaskState interState;
951 0 : TaskState intraState;
952 :
953 0 : localCopyState.stepNumNext = std::min(rdmaConcurrentNum_, stepsPerLoop);
954 :
955 : // 外层loop，要重复多少轮，默认为0，在获取到全局counts信息后刷新
956 0 : u32 repeatLoopNum = 0;
957 :
958 : // 第一步，需要把counts信息广播给机内其他rank
959 0 : bool needDoIntraInfoBroadcast = needCollectInfo_;
960 :
961 0 : while (localCopyState.stepNum < stepsPerLoop || interState.stepNum < stepsPerLoop ||
962 0 : intraState.stepNum < stepsPerLoop) {
963 : // 每一轮首次做跨module收发的同时，做level0的SDMA写，每张卡从input写到对端的out buffer
964 0 : const bool needDoLevel0SdmaWrite = (interState.stepNumNext != 0 && interState.stepNum == 0);
965 : // 每一轮首次做机内分发的同时，每张卡从out buffer将level0其他卡发来的数据拷至output
966 0 : const bool needDoLevel0LocalCopy = (intraState.stepNumNext != 0 && intraState.stepNum == 0);
967 :
968 : // intra的stepNum小于stepNumNext，说明本轮需要做intra分发(SDMA)
969 0 : const bool needDoIntraTasks = intraState.stepNum < intraState.stepNumNext;
970 : // inter的stepNum小于stepNumNext，说明本轮需要做inter收发(RDMA)
971 0 : const bool needDoInterTasks = interState.stepNum < interState.stepNumNext;
972 :
973 0 : const bool hasSdmaTask = needDoIntraTasks || needDoLevel0SdmaWrite || needDoIntraInfoBroadcast;
974 0 : const bool hasRdmaTask = needDoInterTasks;
975 :
976 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][RunAsyncLoop][start] localCopy[step:%u, next:%u, loop:%u], "
977 : "inter[step:%u, next:%u, loop:%u], intra[step:%u, next:%u, loop:%u]",
978 : localCopyState.stepNum, localCopyState.stepNumNext, localCopyState.loopNum,
979 : interState.stepNum, interState.stepNumNext, interState.loopNum,
980 : intraState.stepNum, intraState.stepNumNext, intraState.loopNum);
981 :
982 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][RunAsyncLoop] needDoLevel0SdmaWrite[%d], needDoLevel0LocalCopy[%d], "
983 : "hasSdmaTask[%d], hasRdmaTask[%d]", needDoLevel0SdmaWrite, needDoLevel0LocalCopy, hasSdmaTask, hasRdmaTask);
984 :
985 0 : if (hasSdmaTask) {
986 : // 本轮有SDMA任务。主流通知SDMA从流，SDMA从流等待主流，前同步
987 0 : CHK_RET(NotifySdmaSubStreamStart());
988 0 : CHK_RET(DoSdmaSync(SdmaSyncType::PRE_SYNC));
989 :
990 : // 下发一组主从流同步，拉齐SDMA任务，避免任务不同时拉起导致性能下降
991 0 : CHK_RET(WaitSdmaSubStreamFinish());
992 0 : CHK_RET(NotifySdmaSubStreamStart());
993 : }
994 :
995 0 : if (hasRdmaTask) {
996 : // 本轮有RDMA任务。主流通知RDMA从流，RDMA从流等待主流
997 0 : CHK_RET(NotifyRdmaSubStreamStart());
998 : }
999 :
1000 0 : if (needDoLevel0SdmaWrite) {
1001 0 : CHK_RET(DoLevel0SdmaSend(interState.loopNum));
1002 :
1003 0 : if (interState.loopNum == 0) {
1004 : // 首轮，本卡input到output的拷贝也在这时做
1005 0 : CHK_RET(LocalCopySelfDataFromInputToOutput(mainStream_));
1006 : }
1007 : }
1008 :
1009 0 : if (interState.loopNum == 0 && needDoInterTasks) {
1010 : // 第一轮，在做inter分发前，等待、获取receive信息
1011 0 : if (needCollectInfo_) {
1012 0 : CHK_RET(WaitAndCalReceiveInfo()); // 阻塞函数
1013 0 : needCollectInfo_ = false;
1014 : }
1015 :
1016 : // 刷新重复轮数：总轮数-1
1017 0 : repeatLoopNum = intraLoopNum_ > 0 ? intraLoopNum_ - 1 : 0;
1018 0 : if (localCopyState.stepNum == stepsPerLoop && localCopyState.loopNum < repeatLoopNum) {
1019 : // 如果需要做多轮，在此处立即刷新local copy stepNum，让第二轮的任务尽早开始
1020 0 : ++localCopyState.loopNum;
1021 0 : localCopyState.stepNum = 0;
1022 0 : localCopyState.stepNumNext = std::min(rdmaConcurrentNum_, stepsPerLoop);
1023 : }
1024 : }
1025 :
1026 : // intraStepNum小于interStepNum，表示需要做机内分发
1027 0 : CHK_RET(DoIntraDistribution(intraState.stepNum, intraState.stepNumNext, intraState.loopNum));
1028 : // interStepNum小于localCopyStepNum，表示需要做机间收发
1029 0 : CHK_RET(DoInterSendReceive(interState.stepNum, interState.stepNumNext, interState.loopNum));
1030 : // localCopyStepNum小于stepsPerLoop，根据并发度拷贝需要的数据到in buffer
1031 0 : CHK_RET(DoLocalCopy(localCopyState.stepNum, localCopyState.stepNumNext, localCopyState.loopNum));
1032 :
1033 0 : if (needDoIntraInfoBroadcast) {
1034 : // 第一轮第一步，机内广播本卡的counts信息
1035 0 : CHK_RET(DoIntraInfoBroadcast());
1036 0 : needDoIntraInfoBroadcast = false;
1037 : }
1038 :
1039 0 : if (needDoLevel0LocalCopy) {
1040 0 : CHK_RET(DoLevel0LocalCopy(intraState.loopNum));
1041 : }
1042 :
1043 0 : if (hasSdmaTask) {
1044 : // SDMA尾同步，主流等待SDMA从流，SDMA从流通知主流
1045 0 : CHK_RET(DoSdmaSync(SdmaSyncType::POST_SYNC));
1046 0 : CHK_RET(WaitSdmaSubStreamFinish());
1047 : }
1048 0 : if (hasRdmaTask) {
1049 : // 主流等待RDMA从流，RDMA从流通知主流
1050 0 : CHK_RET(WaitRdmaSubStreamFinish());
1051 : }
1052 :
1053 : // 下发task
1054 0 : CHK_RET(LaunchTaskExtend(dispatcher_, mainStream_, subStreams_));
1055 :
1056 : // 更新每种任务的当前步数
1057 0 : intraState.stepNum = intraState.stepNumNext;
1058 0 : interState.stepNum = interState.stepNumNext;
1059 0 : localCopyState.stepNum = localCopyState.stepNumNext;
1060 :
1061 : // 更新每种任务的下一步目标步数
1062 0 : intraState.stepNumNext = interState.stepNumNext;
1063 0 : interState.stepNumNext = localCopyState.stepNumNext;
1064 0 : localCopyState.stepNumNext = std::min(localCopyState.stepNumNext + rdmaConcurrentNum_, stepsPerLoop);
1065 :
1066 : // 检查是否需要重复执行，若需要，将对应的step num刷回为0
1067 0 : if (intraState.stepNum == stepsPerLoop && intraState.loopNum < interState.loopNum) {
1068 0 : ++intraState.loopNum;
1069 0 : intraState.stepNum = 0;
1070 0 : intraState.stepNumNext = std::min(rdmaConcurrentNum_, stepsPerLoop);
1071 : }
1072 0 : if (interState.stepNum == stepsPerLoop && interState.loopNum < localCopyState.loopNum) {
1073 0 : ++interState.loopNum;
1074 0 : interState.stepNum = 0;
1075 0 : interState.stepNumNext = std::min(rdmaConcurrentNum_, stepsPerLoop);
1076 : }
1077 0 : if (localCopyState.stepNum == stepsPerLoop && localCopyState.loopNum < repeatLoopNum) {
1078 0 : ++localCopyState.loopNum;
1079 0 : localCopyState.stepNum = 0;
1080 0 : localCopyState.stepNumNext = std::min(rdmaConcurrentNum_, stepsPerLoop);
1081 : }
1082 :
1083 0 : HCCL_DEBUG("[AlltoallvContinuousPipeline][RunAsyncLoop][end] localCopy[step:%u, next:%u, loop:%u], "
1084 : "inter[step:%u, next:%u, loop:%u], intra[step:%u, next:%u, loop:%u]",
1085 : localCopyState.stepNum, localCopyState.stepNumNext, localCopyState.loopNum,
1086 : interState.stepNum, interState.stepNumNext, interState.loopNum,
1087 : intraState.stepNum, intraState.stepNumNext, intraState.loopNum);
1088 : }
1089 0 : return HCCL_SUCCESS;
1090 : }
1091 :
1092 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_2_ALL_V_CONTINUOUS_PIPELINE, AlltoallvContinuousPipeline);
1093 : } // namespace hccl
|
