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