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 "allltoall_pipeline_mesh_pairwise_ping_pong.h"
12 : #include <numeric>
13 : #include "alg_template_register.h"
14 :
15 : namespace hccl {
16 :
17 : // 需要将 ccl 切成两份，ping-pong 时也根据收发次数取模2决定使用 ping mem 还是 pong mem
18 : static const u32 PING_PONG_CONST_NUM = 2;
19 : static const u32 INTRA_STREAM_INFO_SENDLEN_INDEX = 0; // intraStreamInfo 中 sendLen 的下标
20 : static const u32 INTRA_STREAM_INFO_RECVLEN_INDEX = 1; // intraStreamInfo 中 recvLen 的下标
21 : static const u32 INTRA_STREAM_INFO_RECV_LOCAL_OFFSET_INDEX = 2; // intraStreamInfo 中 recvRemoteOffset 的下标
22 :
23 0 : AlltoallPipelineMeshPairwisePingPong::AlltoallPipelineMeshPairwisePingPong(
24 0 : const HcclDispatcher dispatcher): AlltoallPipelineBase(dispatcher) {}
25 :
26 0 : AlltoallPipelineMeshPairwisePingPong::~AlltoallPipelineMeshPairwisePingPong() {}
27 :
28 0 : u32 AlltoallPipelineMeshPairwisePingPong::CalcInterNumSteps()
29 : {
30 0 : return interRankSize_ - 1;
31 : }
32 :
33 : // 适配新CollExecutor接口
34 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::Prepare(u32 userRank, A2aPipelineMemory A2aPipelineMemory,
35 : const SubCommInfo &level0CommInfo, const SubCommInfo &level1CommInfo,
36 : Stream &mainStream, std::vector<Stream> &subStream,
37 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub,
38 : std::vector<SendRecvInfo> &allMeshAggregationSendRecvInfo, HcclWorkflowMode workMode)
39 : {
40 0 : AlltoallPipelineBase::Prepare(userRank, A2aPipelineMemory, level0CommInfo, level1CommInfo, mainStream, subStream,
41 : notifyMain, notifySub, allMeshAggregationSendRecvInfo, workMode);
42 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
43 0 : pingPongMemSize_ = (cclIn_.size() / PING_PONG_CONST_NUM);
44 : } else {
45 0 : pingPongMemSize_ = (scratchMem_.size() / PING_PONG_CONST_NUM);
46 : }
47 0 : intraDataBlockSize_ = (pingPongMemSize_ / intraRankSize_);
48 0 : if (intraDataBlockSize_ > HCCL_MIN_SLICE_ALIGN_910B) {
49 0 : intraDataBlockSize_ = (intraDataBlockSize_ / HCCL_MIN_SLICE_ALIGN_910B) * HCCL_MIN_SLICE_ALIGN_910B;
50 : }
51 0 : memStatusInMesh_ = std::vector<bool>(intraRankSize_, false);
52 0 : CHK_RET(DeviceMemMapping());
53 0 : return HCCL_SUCCESS;
54 : }
55 :
56 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::DeviceMemMapping()
57 : {
58 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
59 0 : interTransportSend_ = cclIn_;
60 0 : interTransportRecv_ = cclOut_;
61 0 : intraTransportSend_ = cclOut_;
62 0 : interSendPing_ = interTransportSend_.range(0, pingPongMemSize_);
63 0 : interSendPong_ = interTransportSend_.range(pingPongMemSize_, pingPongMemSize_);
64 0 : interRecvPing_ = interTransportRecv_.range(0, pingPongMemSize_);
65 0 : interRecvPong_ = interTransportRecv_.range(pingPongMemSize_, pingPongMemSize_);
66 : } else {
67 0 : interRecvPing_ = scratchMem_.range(0, pingPongMemSize_);
68 0 : interRecvPong_ = scratchMem_.range(pingPongMemSize_, pingPongMemSize_);
69 : }
70 0 : intraSendPing_ = interRecvPing_;
71 0 : intraSendPong_ = interRecvPong_;
72 :
73 0 : for (u32 intraRank = 0; intraRank < intraRankSize_; intraRank++) {
74 0 : if (intraRank == intraRankId_) {
75 0 : continue;
76 : }
77 0 : LINK& intraNeighboorTransport = intraLinks_[intraRank];
78 0 : void* remDMAMemPtr = nullptr;
79 0 : CHK_RET(intraNeighboorTransport->GetRemoteMem(UserMemType::INPUT_MEM, &remDMAMemPtr));
80 : DeviceMem remoteIntraSend = DeviceMem::create(static_cast<u8 *>(remDMAMemPtr),
81 0 : workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? cclIn_.size() : scratchMem_.size());
82 0 : DeviceMem remoteIntraSendPing = remoteIntraSend.range(0, pingPongMemSize_);
83 0 : DeviceMem remoteIntraSendPong = remoteIntraSend.range(pingPongMemSize_, pingPongMemSize_);
84 0 : intraNeighBoorMemory_[intraRank] = {remoteIntraSendPing, remoteIntraSendPong};
85 0 : }
86 0 : return HCCL_SUCCESS;
87 0 : }
88 :
89 : // 将需要发送给其他 mesh 的数据准备好，并计算好 TxMemoryInfo
90 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PrepareInterSendData(
91 : u32 mainStep,
92 : u32 subStep)
93 : {
94 0 : nextInterSendData_.clear();
95 0 : u32 interSendRankStart = ((interRankId_ + 1 + mainStep) % interRankSize_) * intraRankSize_;
96 0 : DeviceMem interSendMem = (interSendUsePingMem_ ? interSendPing_ : interSendPong_);
97 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][PrepareInterSendData] userRank %u, interRank %u, "
98 : "intraRank %u in main step %llu, sub step %llu send to remote %s", userRank_, interRankId_,
99 : intraRankId_, mainStep, subStep, sendToInterDstMemPing_ ? "interRecvPingMem" : "interRecvPongMem");
100 0 : u64 preStepMaxSend = intraDataBlockSize_ * subStep;
101 0 : for (u32 i = 0; i < intraRankSize_; i++) {
102 0 : u32 dataIndex = i + interSendRankStart;
103 0 : u64 totalSendLen = localSendRecvInfo_.sendLength[dataIndex];
104 0 : u64 sendLen = std::min(intraDataBlockSize_, std::max(totalSendLen, preStepMaxSend) - preStepMaxSend);
105 0 : if (sendLen == 0) {
106 0 : continue;
107 : }
108 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][PrepareInterSendData] userRank %u, interRank %u, "
109 : "intraRank %u data index %llu move from userInput offset %llu length %llu to %s, total size %llu"
110 : "send to remote %s", userRank_, interRankId_, intraRankId_, dataIndex,
111 : localSendRecvInfo_.sendOffset[dataIndex] + preStepMaxSend, sendLen, interSendUsePingMem_ ?
112 : "localInterSendPingMem" : "localInterSendPongMem", totalSendLen,
113 : sendToInterDstMemPing_ ? "interRecvPingMem" : "interRecvPongMem");
114 0 : DeviceMem src = inputMem_.range(localSendRecvInfo_.sendOffset[dataIndex] + preStepMaxSend, sendLen);
115 0 : DeviceMem dst = interSendMem.range(i * intraDataBlockSize_, sendLen);
116 : // 单算子模式需要搬到 CCL，图模式省去这一步
117 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
118 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, mainStream_));
119 : }
120 0 : nextInterSendData_.emplace_back(TxMemoryInfo{UserMemType::OUTPUT_MEM, (sendToInterDstMemPing_ ? 0 :
121 0 : pingPongMemSize_) + i * intraDataBlockSize_, workMode_ ==
122 0 : HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? dst.ptr() : src.ptr(), sendLen});
123 0 : }
124 0 : return HCCL_SUCCESS;
125 0 : }
126 :
127 : // 将需要发送给其他 mesh 的数据准备好，并准备好 TxMemoryInfo
128 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PrepareInterRecvData(
129 : u32 mainStep,
130 : u32 subStep)
131 : {
132 0 : nextInterRecvData_.clear();
133 0 : if (workMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
134 : // 单算子模式本次要接收的数据都放在 CCL，直接整块接收
135 0 : nextInterRecvData_.emplace_back(RxMemoryInfo{
136 0 : UserMemType::INPUT_MEM, recvFromInterSrcMemPing_ ? 0u : pingPongMemSize_,
137 0 : (interRecvUsePingMem_ ? interRecvPing_ : interRecvPong_).ptr(), pingPongMemSize_});
138 : } else {
139 : // 图模式需要计算数据放在对端 userInput 的位置
140 0 : u32 recvFromRank = (userRank_ + groupRankSize_ - (mainStep + 1) * intraRankSize_) % groupRankSize_;
141 0 : const std::vector<u64>& remoteSendLength = (*allMeshAggregationSendRecvInfo_)[recvFromRank].sendLength;
142 0 : const std::vector<u64>& remoteSendOffset = (*allMeshAggregationSendRecvInfo_)[recvFromRank].sendOffset;
143 0 : u64 dataStartOffset = subStep * intraDataBlockSize_;
144 0 : for (u32 i = 0; i < intraRankSize_; i++) {
145 0 : u64 totalRecvDataLen = remoteSendLength[meshRankStart_ + i];
146 0 : u64 recvLen = std::min(std::max(totalRecvDataLen, dataStartOffset) - dataStartOffset, intraDataBlockSize_);
147 0 : if (recvLen == 0) {
148 0 : continue;
149 : }
150 0 : u64 recvRemoteOffset = remoteSendOffset[meshRankStart_ + i] + dataStartOffset;
151 0 : nextInterRecvData_.emplace_back(RxMemoryInfo{UserMemType::INPUT_MEM, recvRemoteOffset,
152 0 : (interRecvUsePingMem_ ? interRecvPing_ : interRecvPong_).range(i * intraDataBlockSize_, recvLen).ptr(),
153 : recvLen});
154 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][PrepareInterRecvData] userRank %u, interRank %u, "
155 : "intraRank %u recv from remote userInput offset %llu length %llu to %s offset %llu", userRank_,
156 : interRankId_, intraRankId_, recvRemoteOffset, recvLen, interRecvUsePingMem_ ?
157 : "localInterRecvPingMem" : "localInterRecvPongMem", i * intraDataBlockSize_);
158 : }
159 : }
160 0 : return HCCL_SUCCESS;
161 : }
162 :
163 : // 准备下一次mesh间需要收发的数据，单算子模式需要从 userInput 搬到 CCLBuffer，图模式则仅需要准备好 TxMemoryInfo
164 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PrepareInterData(
165 : u32 mainStep,
166 : u32 subStep)
167 : {
168 0 : CHK_RET(PrepareInterSendData(mainStep, subStep));
169 0 : CHK_RET(PrepareInterRecvData(mainStep, subStep));
170 0 : return HCCL_SUCCESS;
171 : }
172 :
173 : // 将原先在userInput，且需要发到本mesh内其他卡的数据搬到CCL
174 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PrepareIntraData(u32 subStep)
175 : {
176 0 : u64 dataStartOffset = subStep * intraDataBlockSize_;
177 0 : for (u32 i = 0; i < intraRankSize_; i++) {
178 0 : u32 dataIndex = i + meshRankStart_;
179 0 : u64 totalSendDataLen = localSendRecvInfo_.sendLength[dataIndex];
180 0 : u64 sendLen = std::min(std::max(totalSendDataLen, dataStartOffset) - dataStartOffset, intraDataBlockSize_);
181 0 : if (i == intraRankId_ || sendLen == 0) {
182 0 : continue;
183 : }
184 0 : DeviceMem src = inputMem_.range(localSendRecvInfo_.sendOffset[dataIndex] + dataStartOffset, sendLen);
185 0 : DeviceMem dst = (intraSendUsePingMem_ ? intraSendPing_ : intraSendPong_).range(i * intraDataBlockSize_,
186 0 : sendLen);
187 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, mainStream_));
188 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][PrepareIntraData] userRank %u, interRank %u, intraRank %u"
189 : "data index %u move from userInput offset %llu length %llu to %s, total size %llu ", userRank_,
190 : interRankId_, intraRankId_, dataIndex, localSendRecvInfo_.sendOffset[dataIndex] + dataStartOffset,
191 : sendLen, intraSendUsePingMem_ ? "IntraPingMem" : "IntraPongMem", totalSendDataLen);
192 0 : }
193 0 : return HCCL_SUCCESS;
194 : }
195 :
196 : // 计算mesh内其他卡此时RDMA接收到的数据是在 cclIn 还是 cclOut
197 0 : void AlltoallPipelineMeshPairwisePingPong::UpdateRemoteMemStatusIntra(u32 step)
198 : {
199 0 : for (u32 intraRank = 0; intraRank < intraRankSize_; intraRank++) {
200 0 : if (intraRank == intraRankId_) continue;
201 0 : u32 intraRankHaveRecv = 0;
202 0 : for (u32 i = 1; i <= step; i++) {
203 0 : const std::vector<u64>& intraRankRecvFrom = (*allMeshAggregationSendRecvInfo_)[(meshRankStart_ +
204 0 : groupRankSize_ + intraRank - i * intraRankSize_) % groupRankSize_].sendLength;
205 0 : u64 maxRecvLen = std::accumulate(intraRankRecvFrom.begin() + meshRankStart_,
206 0 : intraRankRecvFrom.begin() + meshRankStart_ + intraRankSize_, 0ULL,
207 0 : [](u64 a, u64 b) {return a > b ? a : b;});
208 0 : intraRankHaveRecv += ((maxRecvLen + intraDataBlockSize_ - 1) / intraDataBlockSize_);
209 : }
210 0 : memStatusInMesh_[intraRank] = ((intraRankHaveRecv % PING_PONG_CONST_NUM) == 0);
211 : }
212 0 : }
213 :
214 : // 计算本卡接收数据的那张卡和本卡将要发数据的那张卡在这个大步骤中的
215 : // 第一个小步骤从哪块ccl收以及发到哪块ccl, 每次切换
216 0 : void AlltoallPipelineMeshPairwisePingPong::UpdateRemoteMemStatusInter(u32 step)
217 : {
218 0 : u32 recvGlobalRank = (userRank_ + groupRankSize_ - (step + 1) * intraRankSize_) % groupRankSize_;
219 0 : u32 recvInterRank = ((interRankId_ + interRankSize_ - (step + 1)) % interRankSize_);
220 0 : u32 numRecvRankHaveSend = 0;
221 0 : u32 numSendRankHaveRecv = 0;
222 0 : const std::vector<u64>& recvRankSendLen = (*allMeshAggregationSendRecvInfo_)[recvGlobalRank].sendLength;
223 0 : for (u32 i = 1; i <= step; i++) {
224 0 : u32 firstBlockIndex = (((recvInterRank + i) % interRankSize_) * intraRankSize_);
225 0 : u64 maxSendLen = std::accumulate(recvRankSendLen.begin() + firstBlockIndex,
226 0 : recvRankSendLen.begin() + firstBlockIndex + intraRankSize_, 0ULL,
227 0 : [](u64 a, u64 b) {return a > b ? a : b;});
228 0 : numRecvRankHaveSend += ((maxSendLen + intraDataBlockSize_ - 1) / intraDataBlockSize_);
229 : const std::vector<u64>& sendRankRecvFrom =
230 0 : (*allMeshAggregationSendRecvInfo_)[(userRank_ + i * intraRankSize_) % groupRankSize_].sendLength;
231 0 : u64 maxRecvLen = std::accumulate(sendRankRecvFrom.begin() + meshRankStart_,
232 0 : sendRankRecvFrom.begin() + meshRankStart_ + intraRankSize_, 0ULL,
233 0 : [](u64 a, u64 b) {return a > b ? a : b;});
234 0 : numSendRankHaveRecv += ((maxRecvLen + intraDataBlockSize_ - 1ULL) / intraDataBlockSize_);
235 : }
236 : // 首次默认都从对端pingMem读，本卡接收数据来源的那张卡每发一次数据切换一次
237 0 : recvFromInterSrcMemPing_ = ((numRecvRankHaveSend % PING_PONG_CONST_NUM) == 0);
238 : // 首次默认发到对端pingMem，本卡发送数据目的地的那张卡每接收一次数据切换一次
239 0 : sendToInterDstMemPing_ = ((numSendRankHaveRecv % PING_PONG_CONST_NUM) == 0);
240 0 : }
241 :
242 : // 收集本次 SDMA 子步骤每条流需要收发的长度，偏移地址，内存状态信息避免重复计算影响性能
243 0 : void AlltoallPipelineMeshPairwisePingPong::UpdateIntraStreamInfo(
244 : u32 interRankDistance,
245 : u32 subStep)
246 : {
247 0 : intraStreamInfo_.clear();
248 0 : u32 firstDataBlockIndex =
249 0 : (meshRankStart_ + groupRankSize_ - interRankDistance * intraRankSize_) % groupRankSize_;
250 0 : const std::vector<u64>& sendInfo = (*allMeshAggregationSendRecvInfo_)[firstDataBlockIndex + intraRankId_].sendLength;
251 0 : u64 dataStartOffset = subStep * intraDataBlockSize_;
252 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][UpdateSDMAStreamInfo] userRank %u, "
253 : "interRank %u, intraRank %u, interRankDistance %llu, sub step %llu", userRank_,
254 : interRankId_, intraRankId_, interRankDistance, subStep);
255 0 : for (u32 i = 0; i < intraRankSize_; i++) {
256 0 : u64 totalSendDataLen = sendInfo[meshRankStart_ + i];
257 0 : u64 totalRecvDataLen = localSendRecvInfo_.recvLength[i + firstDataBlockIndex];
258 0 : u64 sendLen = std::min(std::max(totalSendDataLen, dataStartOffset) - dataStartOffset, intraDataBlockSize_);
259 0 : u64 recvLen = std::min(std::max(totalRecvDataLen, dataStartOffset) - dataStartOffset, intraDataBlockSize_);
260 0 : u64 localOffset = localSendRecvInfo_.recvOffset[i + firstDataBlockIndex] + subStep * intraDataBlockSize_;
261 0 : if (i != intraRankId_) {
262 0 : intraStreamInfo_[i] = {sendLen, recvLen, localOffset};
263 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][UpdateSDMAStreamInfo] userRank %u, interRank %u, "
264 : "intraRank %u, sdma stream %llu need send %llu and read length %llu to local offset %llu",
265 : userRank_, interRankId_, intraRankId_, i, sendLen, recvLen, localOffset);
266 : }
267 : }
268 0 : }
269 :
270 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::SendRecvDataIntraMesh()
271 : {
272 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][ReadDataInMesh] userRank %u, "
273 : "interRank %u, intraRank %u, sdma stream %s wait main stream", userRank_, interRankId_,
274 : intraRankId_, GetStreamIndexString().c_str());
275 0 : bool anySend = false;
276 0 : for (auto& sdmaInfo : intraStreamInfo_) {
277 0 : u32 streamIndex = sdmaInfo.first;
278 0 : u64 recvLen = sdmaInfo.second[INTRA_STREAM_INFO_RECVLEN_INDEX];
279 0 : u64 recvOffset = sdmaInfo.second[INTRA_STREAM_INFO_RECV_LOCAL_OFFSET_INDEX];
280 0 : Stream& currStream = subStream_[streamIndex];
281 0 : LINK& readTransport = intraLinks_[streamIndex];
282 0 : CHK_RET(readTransport->TxAck(currStream));
283 0 : CHK_RET(readTransport->RxAck(currStream));
284 0 : if (recvLen > 0) {
285 0 : DeviceMem src = intraNeighBoorMemory_[streamIndex][(memStatusInMesh_[streamIndex] ? 0 : 1)].range(
286 0 : intraRankId_ * intraDataBlockSize_, recvLen);
287 0 : DeviceMem dst = outputMem_.range(recvOffset, recvLen);
288 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, currStream, readTransport->GetRemoteRank(),
289 : readTransport->GetLinkType()));
290 0 : }
291 0 : CHK_RET(readTransport->TxDataSignal(currStream));
292 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][ReadDataInMesh] userRank %u, interRank %u, "
293 : "intraRank %u, sdma stream %llu read data from remote %s offset %llu len %llu to local %llu",
294 : userRank_, interRankId_, intraRankId_, streamIndex, memStatusInMesh_[streamIndex] ?
295 : "IntraSendPingMem" : "IntraSendPongMem", intraRankId_ * intraDataBlockSize_,
296 : recvLen, recvOffset);
297 0 : memStatusInMesh_[streamIndex] = (!memStatusInMesh_[streamIndex]);
298 0 : CHK_RET(readTransport->RxDataSignal(currStream));
299 0 : anySend = true;
300 : }
301 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][ReadDataInMesh] userRank %u, "
302 : "interRank %u, intraRank %u, sdma stream %s notify main stream", userRank_, interRankId_,
303 : intraRankId_, GetStreamIndexString().c_str());
304 0 : intraSendUsePingMem_ ^= anySend;
305 0 : return HCCL_SUCCESS;
306 : }
307 :
308 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::SendRecvDataInterMesh(
309 : u32 step,
310 : bool doSend,
311 : bool doRecv)
312 : {
313 0 : Stream& interStream = subStream_[intraRankId_];
314 0 : LINK& interRecvTransport = interLinks_[(interRankId_ + interRankSize_ - 1 - step) % interRankSize_];
315 0 : LINK& interSendTransport = interLinks_[(interRankId_ + 1 + step) % interRankSize_];
316 0 : if (doRecv) {
317 0 : CHK_RET(interRecvTransport->TxAck(interStream));
318 : }
319 0 : if (doSend) {
320 0 : CHK_RET(interSendTransport->RxAck(interStream));
321 0 : CHK_RET(interSendTransport->TxAsync(UserMemType::OUTPUT_MEM, (sendToInterDstMemPing_ ? 0u :
322 : pingPongMemSize_), (interSendUsePingMem_ ? interSendPing_ : interSendPong_).ptr(), pingPongMemSize_,
323 : interStream));
324 0 : interSendUsePingMem_ ^= true;
325 0 : sendToInterDstMemPing_ ^= true;
326 : }
327 0 : if (doRecv) {
328 0 : CHK_RET(interRecvTransport->RxAsync(UserMemType::INPUT_MEM, (recvFromInterSrcMemPing_ ? 0u :
329 : pingPongMemSize_), (interRecvUsePingMem_ ? interRecvPing_ : interRecvPong_).ptr(), pingPongMemSize_,
330 : interStream));
331 0 : CHK_RET(interRecvTransport->PostFinAck(interStream));
332 0 : interRecvUsePingMem_ ^= true;
333 0 : recvFromInterSrcMemPing_ ^= true;
334 : }
335 0 : if (doSend) {
336 0 : CHK_RET(interSendTransport->WaitFinAck(interStream));
337 : }
338 0 : CHK_RET(ExecuteBarrier(interRecvTransport, interSendTransport, interStream));
339 0 : return HCCL_SUCCESS;
340 : }
341 :
342 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::LocalCopyDataRecvFromInter(
343 : u32 mainStep,
344 : u32 subStep)
345 : {
346 : u64 localRecvLen = localSendRecvInfo_.recvLength[
347 0 : (userRank_ + groupRankSize_ - (mainStep + 1) * intraRankSize_) % groupRankSize_];
348 : u64 localRecvOff = localSendRecvInfo_.recvOffset[
349 0 : (userRank_ + groupRankSize_ - (mainStep + 1) * intraRankSize_) % groupRankSize_];
350 0 : u64 currStepRecvLen = std::min(localRecvLen - subStep * intraDataBlockSize_, intraDataBlockSize_);
351 0 : DeviceMem src = (interRecvUsePingMem_ ? interRecvPong_ : interRecvPing_).range(
352 0 : intraRankId_ * intraDataBlockSize_, currStepRecvLen);
353 0 : DeviceMem dst = outputMem_.range(localRecvOff + subStep * intraDataBlockSize_, currStepRecvLen);
354 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, mainStream_));
355 0 : return HCCL_SUCCESS;
356 0 : }
357 :
358 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PreProcess()
359 : {
360 0 : HCCL_DEBUG("[AlltoallPipelineMeshPairwisePingPong][PreProcess] userRank %u, interRank %u, intraRank %u, "
361 : "main stream notify RDMA stream %llu start send", userRank_, interRankId_,
362 : intraRankId_, intraRankId_);
363 : // 搬下次要做 Server 间收发的数据到 ccl buffer
364 0 : CHK_RET(PrepareInterData(0u, 0u));
365 : // 主流notify RDMA流
366 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
367 0 : CHK_RET(NotifyInterStreamStart());
368 :
369 : // 先做一部分 mesh 内 SDMA 操作，剩下的数据待到整体 RDMA 做完之后再补
370 0 : CHK_RET(PrepareIntraData(0u));
371 0 : UpdateIntraStreamInfo(0u, 0u);
372 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
373 0 : CHK_RET(NotifyIntraStreamStart());
374 0 : CHK_RET(SendRecvDataIntraMesh());
375 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
376 : // 主流搬本地那块数据
377 0 : DeviceMem src = inputMem_.range(localSendRecvInfo_.sendOffset[userRank_],
378 0 : localSendRecvInfo_.sendLength[userRank_]);
379 0 : DeviceMem dst = outputMem_.range(localSendRecvInfo_.recvOffset[userRank_],
380 0 : localSendRecvInfo_.recvLength[userRank_]);
381 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, mainStream_));
382 0 : return HCCL_SUCCESS;
383 0 : }
384 :
385 : // 分别计算当前大步骤需要做几次mesh间收和发，和mesh内收和发（mesh间收的次数和mesh内发的次数相同）
386 0 : void AlltoallPipelineMeshPairwisePingPong::GetNumSubStep(
387 : u32 step,
388 : u32& interSendSubStep,
389 : u32& interRecvSubStep,
390 : u32& intraSubStep)
391 : {
392 0 : u32 sendRankStart = ((interRankId_ + 1 + step) % interRankSize_) * intraRankSize_;
393 0 : u32 recvRankStart = ((interRankId_ + interRankSize_ - 1 - step) % interRankSize_) * intraRankSize_;
394 0 : const std::vector<u64>& sendInfo = (*allMeshAggregationSendRecvInfo_)[
395 0 : (userRank_ + groupRankSize_ - intraRankSize_ * (step + 1)) % intraRankSize_].sendLength;
396 0 : u64 maxInterSendLen = std::accumulate(localSendRecvInfo_.sendLength.begin() + sendRankStart,
397 0 : localSendRecvInfo_.sendLength.begin() + sendRankStart + intraRankSize_, 0ULL,
398 0 : [](u64 a, u64 b) {return a > b ? a : b;});
399 0 : u64 maxInterRecvLen = std::accumulate(sendInfo.begin() + meshRankStart_,
400 0 : sendInfo.begin() + meshRankStart_ + intraRankSize_, 0ULL,
401 0 : [](u64 a, u64 b) {return a > b ? a : b;});
402 0 : u64 maxIntraRecvLen = std::accumulate(localSendRecvInfo_.recvLength.begin() + recvRankStart,
403 0 : localSendRecvInfo_.recvLength.begin() + recvRankStart + intraRankSize_, 0ULL,
404 0 : [](u64 a, u64 b) {return a > b ? a : b;});
405 0 : interSendSubStep = (maxInterSendLen + intraDataBlockSize_ - 1) / intraDataBlockSize_;
406 0 : interRecvSubStep = (maxInterRecvLen + intraDataBlockSize_ - 1) / intraDataBlockSize_;
407 : // mesh 的收发步数取决于本卡从其它mesh收到的需要转发到mesh内其他卡的数据以及本卡需要做mesh内读的其他卡数据
408 0 : intraSubStep = (std::max(maxInterRecvLen, maxIntraRecvLen) + intraDataBlockSize_ - 1) / intraDataBlockSize_;
409 0 : }
410 :
411 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PipelineSend(u32 step, bool isLastStep)
412 : {
413 0 : CHK_RET(ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_));
414 0 : u64 maxDataBlock = 0;
415 0 : for (const SendRecvInfo& info : (*allMeshAggregationSendRecvInfo_)) {
416 0 : for (u64 sendLen : info.sendLength) {
417 0 : maxDataBlock = std::max(maxDataBlock, sendLen);
418 : }
419 : }
420 0 : u32 totalSubStep = (maxDataBlock + intraDataBlockSize_ - 1) / intraDataBlockSize_;
421 : // 计算需要从源端哪块内存收数据和发到哪块目的内存
422 : u64 localRecvLen = localSendRecvInfo_.recvLength[
423 0 : (userRank_ + groupRankSize_ - (step + 1) * intraRankSize_) % groupRankSize_];
424 0 : for (u32 subStep = 0; subStep < totalSubStep; subStep++) {
425 : // RDMA 收发数据
426 0 : SendRecvDataInterMesh(step, true, true);
427 0 : if ((subStep + 1u) == totalSubStep) {
428 0 : CHK_RET(PrepareInterData(step + 1, 0u));
429 : } else {
430 0 : CHK_RET(PrepareInterData(step, subStep + 1u));
431 : }
432 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
433 0 : CHK_RET(WaitIntraStreamFinish());
434 0 : CHK_RET(WaitInterStreamFinish());
435 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
436 0 : UpdateIntraStreamInfo(step + 1u, subStep);
437 0 : CHK_RET(NotifyIntraStreamStart());
438 0 : CHK_RET(SendRecvDataIntraMesh());
439 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
440 0 : if ((!isLastStep && subStep < totalSubStep) || (isLastStep && subStep < (totalSubStep - 1))) {
441 0 : CHK_RET(NotifyInterStreamStart());
442 : }
443 0 : if (localRecvLen > subStep * intraDataBlockSize_) {
444 0 : CHK_RET(LocalCopyDataRecvFromInter(step, subStep));
445 : }
446 : }
447 0 : return HCCL_SUCCESS;
448 : }
449 :
450 0 : HcclResult AlltoallPipelineMeshPairwisePingPong::PostProcess()
451 : {
452 0 : CHK_RET(ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_));
453 0 : CHK_RET(WaitIntraStreamFinish());
454 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
455 0 : u64 maxDataBlock = 0;
456 0 : for (const SendRecvInfo& info : (*allMeshAggregationSendRecvInfo_)) {
457 0 : for (u64 sendLen : info.sendLength) {
458 0 : maxDataBlock = std::max(maxDataBlock, sendLen);
459 : }
460 : }
461 0 : u64 stepLast = (maxDataBlock + intraDataBlockSize_ - 1) / intraDataBlockSize_;
462 0 : for (u64 i = 1 ; i < stepLast; i++) {
463 0 : UpdateIntraStreamInfo(0, i);
464 0 : CHK_RET(PrepareIntraData(i));
465 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
466 0 : CHK_RET(NotifyIntraStreamStart());
467 0 : ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_);
468 0 : CHK_RET(SendRecvDataIntraMesh());
469 0 : CHK_RET(WaitIntraStreamFinish());
470 0 : CHK_RET(ExecEmptyTask(inputMem_, outputMem_, mainStream_, dispatcher_));
471 : }
472 0 : return HCCL_SUCCESS;
473 : }
474 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_2_ALL_PIPELINE_MESH_PAIRWISE_PING_PONG,
475 : AlltoallPipelineMeshPairwisePingPong);
476 : } // namespace hccl
|
