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 "log.h"
12 :
13 : #include "ins_coll_alg_registry.h"
14 : #include "ins_batchSendRecv_executor.h"
15 : #include "alg_data_trans_wrapper.h"
16 :
17 : using namespace std;
18 :
19 : namespace Hccl {
20 :
21 : template <typename AlgTopoMatch>
22 0 : InsBatchSendRecvExecutor<AlgTopoMatch>::InsBatchSendRecvExecutor() : InsCollAlgBase()
23 : {
24 0 : }
25 :
26 : template <typename AlgTopoMatch>
27 0 : InsBatchSendRecvExecutor<AlgTopoMatch>::~InsBatchSendRecvExecutor()
28 : {
29 0 : }
30 :
31 : template <typename AlgTopoMatch>
32 0 : void InsBatchSendRecvExecutor<AlgTopoMatch>::SetRmaDataBufferMgr(const RmtDataBufferMgr* rmaDataBufferMgr)
33 : {
34 0 : rmaDataBufferMgr_ = const_cast<RmtDataBufferMgr*>(rmaDataBufferMgr);
35 0 : return;
36 : }
37 :
38 : template <typename AlgTopoMatch>
39 0 : void InsBatchSendRecvExecutor<AlgTopoMatch>::SetOp(const CollAlgOperator &op)
40 : {
41 0 : op_ =op;
42 0 : HcclSendRecvItem* itemPtr = reinterpret_cast<HcclSendRecvItem *>(op.batchSendRecvDataDes.sendRecvItemsPtr);
43 0 : u32 itemNum = op.batchSendRecvDataDes.itemNum;
44 0 : if (itemPtr == nullptr) {
45 0 : THROW<NullPtrException>(StringFormat("itemPtr is null!"));
46 : }
47 0 : commTargetUserRankSet_.clear();
48 0 : for (u32 i = 0; i < itemNum; i++) {
49 0 : commTargetUserRankSet_.insert((itemPtr + i)->remoteRank);
50 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][ParseParam] insert remoteUserRank[%u] to Set ",
51 : (itemPtr + i)->remoteRank);
52 : }
53 0 : HCCL_DEBUG("[SetOp]commTargetUserRankSet_ size[%zu]", commTargetUserRankSet_.size());
54 0 : }
55 :
56 : template <typename AlgTopoMatch>
57 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::InitParams(const CollAlgOperator &op, const CollAlgParams ¶ms)
58 : {
59 0 : opMode_ = params.opMode;
60 0 : maxTmpMemSize_ = params.maxTmpMemSize;
61 0 : CHK_PRT_RET((maxTmpMemSize_ == 0),
62 : HCCL_ERROR("[InitParams] maxTmpMemSize equals to zero for OPBASE."), HcclResult::HCCL_E_PARA);
63 0 : HcclSendRecvItem* itemPtr = reinterpret_cast<HcclSendRecvItem *>(op.batchSendRecvDataDes.sendRecvItemsPtr);
64 0 : u32 itemNum = op.batchSendRecvDataDes.itemNum;
65 0 : CHK_PTR_NULL(itemPtr);
66 0 : commTargetUserRankSet_.clear();
67 0 : for (u32 i = 0; i < itemNum; i++) {
68 0 : commTargetUserRankSet_.insert((itemPtr + i)->remoteRank);
69 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][ParseParam] insert remoteUserRank[%u] to Set ",
70 : (itemPtr + i)->remoteRank);
71 : }
72 0 : HCCL_DEBUG("[InitParams]commTargetUserRankSet_ size[%zu]", commTargetUserRankSet_.size());
73 0 : return HcclResult::HCCL_SUCCESS;
74 : }
75 :
76 : template <typename AlgTopoMatch>
77 0 : bool InsBatchSendRecvExecutor<AlgTopoMatch>::SortSendItems(HcclSendRecvItem* a, HcclSendRecvItem* b) const{
78 0 : u32 aFlag = (a->remoteRank <= static_cast<uint32_t>(myRank_)) ?
79 0 : (a->remoteRank + rankSize_) : a->remoteRank;
80 0 : u32 bFlag = (b->remoteRank <= static_cast<uint32_t>(myRank_)) ?
81 0 : (b->remoteRank + rankSize_) : b->remoteRank;
82 0 : if (aFlag > bFlag) {
83 0 : return true;
84 0 : } else if (aFlag < bFlag) {
85 0 : return false;
86 : }
87 0 : return a->count > b->count;
88 : }
89 :
90 : template <typename AlgTopoMatch>
91 0 : bool InsBatchSendRecvExecutor<AlgTopoMatch>::SortRecvItems(HcclSendRecvItem* a, HcclSendRecvItem* b) const{
92 0 : u32 aFlag = (a->remoteRank < static_cast<uint32_t>(myRank_)) ?
93 0 : (a->remoteRank + rankSize_) : a->remoteRank;
94 0 : u32 bFlag = (b->remoteRank < static_cast<uint32_t>(myRank_)) ?
95 0 : (b->remoteRank + rankSize_) : b->remoteRank;
96 0 : if (aFlag > bFlag) {
97 0 : return false;
98 0 : } else if (aFlag < bFlag) {
99 0 : return true;
100 : }
101 0 : return a->count > b->count;
102 : }
103 :
104 : template <typename AlgTopoMatch>
105 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::GetPairWiseList(HcclSendRecvItem *sendRecvInfo, u32 itemNum)
106 : {
107 0 : HCCL_INFO("[InsBatchSendRecvExecutor][GetPairWiseList] Start sort the batchSendRecv tasklist.");
108 0 : CHK_PTR_NULL(sendRecvInfo);
109 :
110 0 : for (u32 i = 0; i < itemNum; i++) {
111 0 : HCCL_INFO("[InsBatchSendRecvExecutor][GetPairWiseList] index is %u, itemNum is %u,"\
112 : "localRankID is %d, remoteRank is %u, sendRecvType is %u, rankSize is %u.",
113 : i, itemNum, myRank_, sendRecvInfo->remoteRank,
114 : static_cast<u32>(sendRecvInfo->sendRecvType), rankSize_);
115 0 : CHK_PTR_NULL(sendRecvInfo->buf);
116 :
117 0 : if (sendRecvInfo->sendRecvType == HcclSendRecvType::HCCL_SEND) {
118 0 : sendDeque_.push_back(sendRecvInfo);
119 0 : } else if (sendRecvInfo->sendRecvType == HcclSendRecvType::HCCL_RECV) {
120 0 : recvDeque_.push_back(sendRecvInfo);
121 : } else {
122 0 : HCCL_ERROR("[InsBatchSendRecvExecutor][GetPairWiseList] sendRecvType wrong sendrecvType is %d, "\
123 : "rankID is %d, remoteRank is %u.", sendRecvInfo->sendRecvType, myRank_,
124 : sendRecvInfo->remoteRank);
125 0 : return HcclResult::HCCL_E_PARA;
126 : }
127 0 : sendRecvInfo++;
128 : }
129 :
130 : /* 此处的排序逻辑(pair-wise算法):
131 : 1.sendDeque元素顺序是:先放remoteRank号小于等于root rank的第一个任务，依次减小(循环索引)直至放完
132 : 2.recvDeque元素顺序是:先放remoteRank号大于等于root rank的第一个任务，依次增大(循环索引)直至放完
133 : 如果有rank间重复send/recv场景，按照收发数据从大到小排序
134 : */
135 0 : auto sendCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
136 0 : return this->SortSendItems(a, b);
137 : };
138 :
139 0 : auto recvCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
140 0 : return this->SortRecvItems(a, b);
141 : };
142 :
143 0 : std::stable_sort(sendDeque_.begin(), sendDeque_.end(), sendCompare);
144 0 : std::stable_sort(recvDeque_.begin(), recvDeque_.end(), recvCompare);
145 :
146 : // 筛选自收发任务
147 0 : while ((!sendDeque_.empty() && sendDeque_.front()->remoteRank ==
148 0 : static_cast<uint32_t>(myRank_)) &&
149 0 : (!recvDeque_.empty() && recvDeque_.front()->remoteRank == static_cast<uint32_t>(myRank_))) {
150 0 : sendToSelfDeque_.push_back(sendDeque_.front());
151 0 : recvFromSelfDeque_.push_back(recvDeque_.front());
152 0 : sendDeque_.pop_front();
153 0 : recvDeque_.pop_front();
154 : }
155 : // 自收发任务按照收发长度大小排序
156 0 : auto selfDequeCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
157 0 : return a->count > b->count;
158 : };
159 :
160 0 : std::stable_sort(sendToSelfDeque_.begin(), sendToSelfDeque_.end(), selfDequeCompare);
161 0 : std::stable_sort(recvFromSelfDeque_.begin(), recvFromSelfDeque_.end(), selfDequeCompare);
162 :
163 : // 如果自发自收任务没有完全匹配
164 0 : if ((!sendDeque_.empty() && sendDeque_.front()->remoteRank == static_cast<uint32_t>(myRank_)) ||
165 0 : (!recvDeque_.empty() && recvDeque_.front()->remoteRank == static_cast<uint32_t>(myRank_))) {
166 0 : HCCL_ERROR("[CollBatchSendRecvExecutor] SendTask and Recv Task to rank itself do not match,"\
167 : "please check the task list.");
168 0 : return HcclResult::HCCL_E_PARA;
169 : }
170 0 : HCCL_INFO("[CollBatchSendRecvExecutor][GetPairWiseList] End sort the batchSendRecv tasklist.");
171 0 : return HcclResult::HCCL_SUCCESS;
172 : }
173 :
174 : template <typename AlgTopoMatch>
175 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::ProcessSelfSendRecvTasks(InsQuePtr& queue)
176 : {
177 0 : while (!sendToSelfDeque_.empty() && !recvFromSelfDeque_.empty()) {
178 0 : if (sendToSelfDeque_.front()->count == recvFromSelfDeque_.front()->count &&
179 0 : sendToSelfDeque_.front()->dataType == recvFromSelfDeque_.front()->dataType) {
180 0 : HcclDataType hccldataTypeSelf = sendToSelfDeque_.front()->dataType;
181 0 : DataType dataTypeSelf = HcclDataTypeToDataType(hccldataTypeSelf);
182 0 : u64 dataSize = sendToSelfDeque_.front()->count * DataTypeSizeGet(dataTypeSelf);
183 :
184 : // 搬运本卡到本卡的数据 使用扩展InsLocalCopyExtend接口
185 0 : DataBuffer inputBuffer(reinterpret_cast<uintptr_t>(sendToSelfDeque_.front()->buf), dataSize);
186 0 : DataBuffer outputBuffer(reinterpret_cast<uintptr_t>(recvFromSelfDeque_.front()->buf), dataSize);
187 0 : HCCL_DEBUG("inputBuffer[%llu], outputBuffer[%llu], dataSize[%llu]", inputBuffer.GetAddr(),
188 : outputBuffer.GetAddr(), dataSize);
189 0 : queue->Append(std::make_unique<InsLocalCopyExtend>(inputBuffer, outputBuffer)); // localcopy
190 :
191 0 : sendToSelfDeque_.pop_front();
192 0 : recvFromSelfDeque_.pop_front();
193 0 : } else {
194 0 : HCCL_ERROR("[HcclBatchSendRecv] Send task and recv task to self : count or dataType do not equal, please"\
195 : "check the task list.");
196 0 : return HCCL_E_PARA;
197 : }
198 : }
199 0 : return HcclResult::HCCL_SUCCESS;
200 : }
201 :
202 : template <typename AlgTopoMatch>
203 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::ProcessSendRecv(const CollAlgOperator &op, InsQuePtr& queue,
204 : u32 remoteRank, std::vector<SendRecvSlice>& sendRemoteSlices,
205 : std::vector<SendRecvSlice>& recvRemoteSlices, LinkData& link) const
206 : {
207 0 : HCCL_INFO("[InsBatchSendRecvExecutor][ProcessSendRecv] Start to with rank[%u].", remoteRank);
208 0 : u32 maxSendRecvStep = std::max(sendRemoteSlices.size(), recvRemoteSlices.size());
209 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][ProcessSendRecv] maxSendRecvStep[%u].", maxSendRecvStep);
210 :
211 0 : CHK_PTR_NULL(op.scratchMem);
212 0 : uint64_t scratchBufferAddr = op.scratchMem->GetAddr();
213 :
214 : // 当前是write的逻辑，整理一套read的逻辑出来
215 0 : if (dmaMode_ == DmaMode::GET) {
216 0 : for (u32 step = 0; step < maxSendRecvStep; step++) {
217 0 : if (step < sendRemoteSlices.size()) {
218 : // 先做localCopy local copy: usrin->cclin
219 0 : DataBuffer inputBuffer(sendRemoteSlices[step].addr_, sendRemoteSlices[step].size_);
220 0 : DataBuffer inScratchSlice(scratchBufferAddr + (remoteRank % rankSize_) * maxRoundTransferSize_,
221 0 : sendRemoteSlices[step].size_);
222 0 : HCCL_DEBUG("scratchBufferAddr[%llu], offset[%llu], dataSize[%llu]", scratchBufferAddr,
223 : (remoteRank % rankSize_) * maxRoundTransferSize_, sendRemoteSlices[step].size_);
224 :
225 0 : queue->Append(std::make_unique<InsLocalCopyExtend>(inputBuffer, inScratchSlice)); // 这里还没做DMA消减，可以到时看下性能再优化
226 : // 然后通知对端来读
227 0 : queue->Append(std::make_unique<InsPostReady>(static_cast<RankId>(remoteRank), link));
228 0 : }
229 :
230 0 : if (step < recvRemoteSlices.size()) {
231 0 : CHK_RET(ProcessRecvDataSlice(queue, recvRemoteSlices[step], remoteRank, scratchBufferAddr, link));
232 : }
233 :
234 0 : if (step < sendRemoteSlices.size()) {
235 0 : queue->Append(std::make_unique<InsWaitFin>(static_cast<RankId>(remoteRank), link));
236 : }
237 : }
238 :
239 0 : return HcclResult::HCCL_SUCCESS;
240 : }
241 :
242 0 : for (u32 step = 0; step < maxSendRecvStep; step++) {
243 0 : if (step < recvRemoteSlices.size()) {
244 : // tell sendRank ready to write
245 0 : queue->Append(std::make_unique<InsPostReady>(static_cast<RankId>(remoteRank), link));
246 : }
247 0 : if (step < sendRemoteSlices.size()) {
248 0 : CHK_RET(ProcessSendDataSlice(queue, sendRemoteSlices[step], remoteRank, scratchBufferAddr, link));
249 : }
250 0 : if (step < recvRemoteSlices.size()) {
251 : // wait sendRank write done
252 0 : queue->Append(std::make_unique<InsWaitFin>(static_cast<RankId>(remoteRank), link));
253 : // local copy
254 0 : CHK_RET(CopyRecvDataSliceToUsrOut(queue, recvRemoteSlices[step], remoteRank, scratchBufferAddr));
255 : }
256 : }
257 0 : return HcclResult::HCCL_SUCCESS;
258 : }
259 :
260 : template <typename AlgTopoMatch>
261 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::RunLoopSendRecv(const CollAlgOperator &op,
262 : std::vector<InsQuePtr>& queues, InsTempAllGatherMesh1D& tempAlg)
263 : {
264 : // pre sync
265 0 : CHK_RET(tempAlg.PreSyncInterQueues(queues));
266 :
267 : // sendrecv options
268 0 : u32 queIdx = 1;
269 0 : for (const u32& remoteRank : commTargetUserRankSet_) {
270 0 : HCCL_INFO("[InsBatchSendRecvExecutor][RunLoopSendRecv] remoteRank[%u].", remoteRank);
271 0 : if (remoteRank == static_cast<uint32_t>(myRank_)) {
272 0 : continue;
273 : }
274 0 : if (queIdx >= queues.size()) {
275 0 : HCCL_ERROR("[InsBatchSendRecvExecutor][RunLoopSendRecv] queIdx[%u] is bigger than queues size[%u].",
276 : queIdx, queues.size());
277 0 : return HCCL_E_PARA;
278 : }
279 0 : auto sendIt = SendSliceMapByRemoteRank_.find(remoteRank);
280 0 : auto recvIt = RecvSliceMapByRemoteRank_.find(remoteRank);
281 0 : if (sendIt == SendSliceMapByRemoteRank_.end() && recvIt == RecvSliceMapByRemoteRank_.end()) {
282 0 : continue;
283 : }
284 0 : LinkData link = tempResLinks_.at(remoteRank)[0];
285 0 : if (sendIt != SendSliceMapByRemoteRank_.end() && recvIt != RecvSliceMapByRemoteRank_.end()) {
286 0 : CHK_RET(ProcessSendRecv(op, queues[queIdx], remoteRank, SendSliceMapByRemoteRank_[remoteRank],
287 : RecvSliceMapByRemoteRank_[remoteRank], link));
288 0 : } else if (sendIt != SendSliceMapByRemoteRank_.end()) {
289 0 : std::vector<SendRecvSlice> empty;
290 0 : CHK_RET(ProcessSendRecv(op, queues[queIdx], remoteRank, SendSliceMapByRemoteRank_[remoteRank],
291 : empty, link));
292 0 : } else if (recvIt != RecvSliceMapByRemoteRank_.end()) {
293 0 : std::vector<SendRecvSlice> empty;
294 0 : CHK_RET(ProcessSendRecv(op, queues[queIdx], remoteRank, empty,
295 : RecvSliceMapByRemoteRank_[remoteRank], link));
296 0 : }
297 0 : queIdx++;
298 : }
299 :
300 : // post sync
301 0 : CHK_RET(tempAlg.PostSyncInterQueues(queues));
302 :
303 0 : return HcclResult::HCCL_SUCCESS;
304 : }
305 :
306 : template <typename AlgTopoMatch>
307 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::GenSendSlicesMapRank()
308 : {
309 : // 遍历 sendDataSlices_，将每个元素根据其 remoteRank 放入相应的 vector 中
310 0 : for (const auto& slice : sendDataSilces_) {
311 : // 获取 remoteRank
312 0 : int remoteRank = slice.remoteRank_;
313 :
314 : // 将当前 slice 放入对应的 vector 中
315 0 : SendSliceMapByRemoteRank_[remoteRank].emplace_back(slice);
316 : }
317 0 : return HcclResult::HCCL_SUCCESS;
318 : }
319 :
320 : template <typename AlgTopoMatch>
321 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CalcSendSlices(u64 maxRoundTransferSize)
322 : {
323 0 : while (!sendDeque_.empty()) {
324 0 : HcclSendRecvItem* sendRecvItem = sendDeque_.front();
325 0 : HCCL_INFO("[InsBatchSendRecvExecutor][CalcSendSlices] remoteRank[%u], buf[%p], count[%llu],"\
326 : "dataType[%u], sendRecvType[%d].", sendRecvItem->remoteRank, sendRecvItem->buf,
327 : sendRecvItem->count, sendRecvItem->dataType, sendRecvItem->sendRecvType);
328 0 : u8 *curInputPtr = static_cast<u8 *>(sendRecvItem->buf);
329 0 : CHK_PTR_NULL(curInputPtr);
330 :
331 0 : HcclDataType hccldataTypeSend = sendRecvItem->dataType;
332 0 : DataType dataTypeSend = HcclDataTypeToDataType(hccldataTypeSend);
333 0 : u32 unitSize = DataTypeSizeGet(dataTypeSend);
334 :
335 0 : u64 resDataSize = sendRecvItem->count * unitSize;
336 0 : u64 curOffset = 0;
337 :
338 0 : while(resDataSize > 0) {
339 : // 判断本轮需搬运的数据量
340 0 : u64 transferSize = resDataSize > maxRoundTransferSize ? maxRoundTransferSize : resDataSize;
341 0 : curInputPtr = static_cast<u8 *>(sendRecvItem->buf) + curOffset;
342 0 : sendDataSilces_.emplace_back(reinterpret_cast<uintptr_t>(curInputPtr), transferSize, sendRecvItem->remoteRank);
343 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][CalcSendSlices] slice curOffset[%llu], slice size[%llu] curInputPtr [%p].",
344 : curOffset, transferSize, curInputPtr);
345 0 : curOffset += transferSize;
346 0 : resDataSize -= transferSize;
347 : }
348 0 : sendDeque_.pop_front();
349 : }
350 0 : return HcclResult::HCCL_SUCCESS;
351 : }
352 :
353 : template <typename AlgTopoMatch>
354 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::GenRecvSlicesMapRank()
355 : {
356 : // 遍历 recvDataSilces_, 将每个元素根据其 remoteRank 放入相应的 vector 中
357 0 : for (const auto& slice : recvDataSilces_) {
358 : // 获取 remoteRank
359 0 : int remoteRank = slice.remoteRank_;
360 :
361 : // 将当前 slice 放入对应的 vector 中
362 0 : RecvSliceMapByRemoteRank_[remoteRank].emplace_back(slice);
363 : }
364 0 : return HcclResult::HCCL_SUCCESS;
365 : }
366 :
367 : template <typename AlgTopoMatch>
368 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CalcRecvSlices(u64 maxRoundTransferSize)
369 : {
370 0 : while (!recvDeque_.empty()) {
371 0 : HcclSendRecvItem* sendRecvItem = recvDeque_.front();
372 0 : HCCL_INFO("[InsBatchSendRecvExecutor][CalcSendSlices] remoteRank[%u], buf[%p], count[%llu],"\
373 : "dataType[%u], sendRecvType[%d].", sendRecvItem ->remoteRank, sendRecvItem ->buf,
374 : sendRecvItem->count, sendRecvItem->dataType, sendRecvItem->sendRecvType);
375 0 : u8 *curInputPtr = static_cast<u8 *>(sendRecvItem->buf);
376 0 : CHK_PTR_NULL(curInputPtr);
377 :
378 0 : HcclDataType hccldataTypeRecv = sendRecvItem->dataType;
379 0 : DataType dataTypeRecv = HcclDataTypeToDataType(hccldataTypeRecv);
380 0 : u32 unitSize = DataTypeSizeGet(dataTypeRecv);
381 :
382 0 : u64 resDataSize = sendRecvItem->count * unitSize;
383 0 : u64 curOffset = 0;
384 :
385 0 : while(resDataSize > 0) {
386 : // 判断本轮需搬运的数据量
387 0 : u64 transferSize = resDataSize > maxRoundTransferSize ? maxRoundTransferSize : resDataSize;
388 0 : curInputPtr = static_cast<u8 *>(sendRecvItem->buf) + curOffset;
389 0 : recvDataSilces_.emplace_back(reinterpret_cast<uintptr_t>(curInputPtr),
390 0 : transferSize, sendRecvItem->remoteRank);
391 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][CalcRecvSlices] slice curOffset[%llu], slice size[%llu], curInputPtr [%p].",
392 : curOffset, transferSize, curInputPtr);
393 0 : curOffset += transferSize;
394 0 : resDataSize -= transferSize;
395 : }
396 :
397 0 : recvDeque_.pop_front();
398 : }
399 0 : return HcclResult::HCCL_SUCCESS;
400 : }
401 :
402 : template <typename AlgTopoMatch>
403 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::ProcessRecvDataSlice(InsQuePtr& queue,
404 : SendRecvSlice& recvRemoteSlice, u32 remoteRank, uint64_t scratchBufferAddr, LinkData& link) const
405 : {
406 : // 从远端读
407 0 : queue->Append(std::make_unique<InsWaitReady>(static_cast<RankId>(remoteRank), link));
408 :
409 : // 获取远端内存地址, 获取的是scratch的基起始地址
410 0 : DataBuffer remoteBuffer = rmaDataBufferMgr_->GetBuffer(link, BufferType::SCRATCH);
411 0 : uint64_t remoteBufferAddr = remoteBuffer.GetAddr();
412 0 : DataBuffer srcScratchSlice(remoteBufferAddr + (myRank_ % rankSize_) * maxRoundTransferSize_,
413 0 : recvRemoteSlice.size_);
414 :
415 : // 准备数据偏移
416 0 : u64 offsetOfRemoteScratchBase = maxRoundTransferSize_ * rankSize_ + maxRoundTransferSize_ * (remoteRank % rankSize_);
417 0 : DataBuffer dstScratchSlice(scratchBufferAddr + offsetOfRemoteScratchBase, recvRemoteSlice.size_);
418 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][ProcessRecvDataSlice] myRank[%d] recv Size[%llu], remoteBuffer[%llu], remoteUserRank[%u].",
419 : myRank_, recvRemoteSlice.size_, remoteBufferAddr, remoteRank);
420 :
421 : // Recv
422 0 : queue->Append(std::make_unique<InsReadExtend>(static_cast<RankId>(remoteRank),
423 : link, dstScratchSlice, srcScratchSlice));
424 :
425 0 : queue->Append(std::make_unique<InsPostFin>(static_cast<RankId>(remoteRank), link));
426 :
427 : // 最后把数据拷贝回自己的output
428 0 : CHK_RET(CopyRecvDataSliceToUsrOut(queue, recvRemoteSlice, remoteRank, scratchBufferAddr));
429 :
430 0 : return HcclResult::HCCL_SUCCESS;
431 0 : }
432 :
433 : template <typename AlgTopoMatch>
434 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::ProcessSendDataSlice(InsQuePtr& queue,
435 : SendRecvSlice& sendRemoteSlice, u32 remoteRank, uint64_t scratchBufferAddr, LinkData& link) const
436 : {
437 : // local copy: usrin->cclin
438 0 : DataBuffer inputBuffer(sendRemoteSlice.addr_, sendRemoteSlice.size_);
439 0 : DataBuffer inScratchSlice(scratchBufferAddr + (remoteRank % rankSize_) * maxRoundTransferSize_,
440 0 : sendRemoteSlice.size_);
441 0 : HCCL_DEBUG("scratchBufferAddr[%llu], offset[%llu], dataSize[%llu]", scratchBufferAddr,
442 : (remoteRank % rankSize_) * maxRoundTransferSize_, sendRemoteSlice.size_);
443 :
444 0 : queue->Append(std::make_unique<InsLocalCopyExtend>(inputBuffer, inScratchSlice));
445 :
446 0 : CHK_RET(SendRun(inScratchSlice, remoteRank, queue, link));
447 :
448 0 : return HcclResult::HCCL_SUCCESS;
449 0 : }
450 :
451 : template <typename AlgTopoMatch>
452 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CopyRecvDataSliceToUsrOut(InsQuePtr& queue,
453 : SendRecvSlice& slice, u32 remoteRank, uint64_t scratchBufferAddr) const
454 : {
455 : // local copy : cclout->usrout
456 0 : DataBuffer outScratchSlice(scratchBufferAddr +
457 0 : (remoteRank % rankSize_ + rankSize_) * maxRoundTransferSize_, slice.size_);
458 :
459 0 : DataBuffer outputBuffer(slice.addr_, slice.size_);
460 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][CopyRecvDataSliceToUsrOut] scratchMem Addr[%llu] localcopy" \
461 : "size[%llu] to outputBuffer[%llu].", outScratchSlice.GetAddr(), slice.size_, outputBuffer.GetAddr());
462 0 : queue->Append(std::make_unique<InsLocalCopyExtend>(outScratchSlice, outputBuffer));
463 :
464 0 : return HcclResult::HCCL_SUCCESS;
465 0 : }
466 :
467 : template <typename AlgTopoMatch>
468 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::SendRun(DataBuffer &execBufferSlice,
469 : u32 remoteUserRank, InsQuePtr& queue, LinkData& link) const
470 : {
471 0 : if (execBufferSlice.GetSize() == 0) {
472 0 : HCCL_ERROR("[InsBatchSendRecvExecutor][SendRun] SendRun input is null");
473 0 : return HCCL_E_PTR;
474 : }
475 :
476 0 : u64 sendSize = execBufferSlice.GetSize();
477 :
478 : // 准备数据偏移
479 0 : u64 offsetOfRemoteScratchBase = maxRoundTransferSize_ * rankSize_ + maxRoundTransferSize_ * (myRank_ % rankSize_);
480 :
481 : // 获取远端内存地址, 获取的是scratch的基起始地址
482 0 : DataBuffer remoteBuffer = rmaDataBufferMgr_->GetBuffer(link, BufferType::SCRATCH);
483 0 : uint64_t remoteBufferAddr = remoteBuffer.GetAddr();
484 0 : DataBuffer sendRemoteBuffer(remoteBufferAddr + offsetOfRemoteScratchBase, sendSize);
485 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][SendRun] myRank[%d] send Size[%llu], remoteBuffer[%llu], remoteUserRank[%u].",
486 : myRank_, sendSize, remoteBuffer.GetAddr(), remoteUserRank);
487 :
488 : // wait recvRank ready
489 0 : queue->Append(std::make_unique<InsWaitReady>(static_cast<RankId>(remoteUserRank), link));
490 :
491 : // Send
492 0 : queue->Append(std::make_unique<InsWriteWithFinExtend>(static_cast<RankId>(remoteUserRank),
493 : link, execBufferSlice, sendRemoteBuffer));
494 :
495 0 : return HcclResult::HCCL_SUCCESS;
496 0 : }
497 :
498 : // 算子执行ccu接口
499 : template <typename AlgTopoMatch>
500 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::Orchestrate(
501 : const RankGraph *rankGraph,
502 : const CollAlgOperator &op,
503 : const CollAlgParams ¶ms,
504 : InsQuePtr insQue)
505 : {
506 : (void)rankGraph;
507 : (void)op;
508 : (void)params;
509 : (void)insQue;
510 :
511 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
512 : }
513 :
514 : // 算子执行aicpu接口
515 : template <typename AlgTopoMatch>
516 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::Orchestrate(const AlgTopoInfo &topoInfo,
517 : const CollAlgOperator &op,
518 : const CollAlgParams ¶ms,
519 : ConnectedLinkMgr *linkMgr,
520 : InsQuePtr insQue)
521 : {
522 0 : HCCL_INFO("[InsBatchSendRecvExecutor][Orchestrate] Begin to Generate Instruction Queue for BatchSendRecv.");
523 : // init and check params
524 0 : CHK_RET(Init(op, params, insQue));
525 :
526 0 : CHK_PRT_RET(topoInfo.vTopo.size() == 0,
527 : HCCL_ERROR("[InsBatchSendRecvExecutor] Rank[%d], vTopo size is zero.", myRank_),
528 : HcclResult::HCCL_E_PARA);
529 :
530 0 : CHK_PRT_RET(topoInfo.virtRankMap.size() == 0,
531 : HCCL_ERROR("[InsBatchSendRecvExecutor] Rank[%d], virtRankMap size is zero.", myRank_),
532 : HcclResult::HCCL_E_PARA);
533 :
534 0 : CHK_PRT_RET(rankSize_ == 1,
535 : HCCL_ERROR("BatchSendRecv Executor orchestrate failed, do not support single rank."),
536 : HcclResult::HCCL_E_PARA);
537 :
538 0 : virtRankMap_ = topoInfo.virtRankMap[0];
539 0 : vTopo_ = topoInfo.vTopo[0];
540 :
541 0 : InsTempAllGatherMesh1D tempAlg(myRank_, rankSize_, topoInfo.vTopo[0], topoInfo.virtRankMap[0]);
542 :
543 : // calculate required insQues and prepare queue
544 0 : AlgTempResReq tempResReq;
545 0 : CHK_RET(CalcRes(tempResReq));
546 :
547 0 : CHK_RET(InitQueue(tempResReq.queNum, requiredQue_));
548 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], requiredQue Num [%u].", myRank_, tempResReq.queNum);
549 :
550 0 : CHK_PTR_NULL(linkMgr);
551 0 : CHK_RET(PrepResLinks(myRank_, tempResReq.links, linkMgr, tempResLinks_));
552 :
553 0 : if (tempAlg.IsPcieLink(tempResLinks_)) {
554 0 : dmaMode_ = DmaMode::GET;
555 : }
556 :
557 : // cclbuffer
558 0 : buffInfo_.inBuffType = BufferType::SCRATCH;
559 0 : buffInfo_.outBuffType = BufferType::SCRATCH;
560 0 : buffInfo_.inBuffBaseOff = 0;
561 0 : buffInfo_.outBuffBaseOff = maxTmpMemSize_ / 2; // 占据scratch memory的后半部分，除以2
562 :
563 : // batchsendrecv实现
564 0 : CHK_RET(GetPairWiseList(static_cast<HcclSendRecvItem *>(op.batchSendRecvDataDes.sendRecvItemsPtr),
565 : op.batchSendRecvDataDes.itemNum));
566 0 : CHK_RET(ProcessSelfSendRecvTasks(requiredQue_[0]));
567 :
568 : // 当需要多轮搬运时，需保证一次数据的搬运量需为单个数据size的整数倍
569 0 : u64 maxRoundTransferSize = params.maxTmpMemSize / MULTIPLY_TWO / rankSize_; // scratch分成2*ranksize份
570 0 : maxRoundTransferSize_ = maxRoundTransferSize;
571 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][Orchestrate] Max scratch buffer size [%u].",
572 : params.maxTmpMemSize);
573 :
574 0 : CHK_RET(CalcSendSlices(maxRoundTransferSize));
575 0 : CHK_RET(GenSendSlicesMapRank());
576 :
577 0 : CHK_RET(CalcRecvSlices(maxRoundTransferSize));
578 0 : CHK_RET(GenRecvSlicesMapRank());
579 :
580 : // aicpu mode
581 0 : CHK_RET(RunLoopSendRecv(op, requiredQue_, tempAlg));
582 :
583 0 : return HcclResult::HCCL_SUCCESS;
584 0 : }
585 :
586 : template <typename AlgTopoMatch>
587 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CalcResLinksPartialMesh
588 : (const RankId myRank, const std::vector<std::vector<RankId>> &tempVTopo,
589 : const u32 linkNumBtwPeers, AlgTempResReq &tempResReq)
590 : {
591 : u32 myAlgRank;
592 0 : u32 partialRankSize = commTargetUserRankSet_.size() + 1;
593 :
594 0 : if (tempVTopo.size() < 1) {
595 0 : HCCL_ERROR("[InsBatchSendRecvExecutor][CalcResLinksPartialMesh] Rank[%d], tempVTopo size is zero.", myRank);
596 0 : return HCCL_E_PARA;
597 : }
598 0 : for (u32 i = 0; i < tempVTopo.size(); i++) { // 遍历level0的2个平面
599 0 : CHK_RET(GetAlgRank(myRank, tempVTopo[i], myAlgRank));
600 0 : for (u32 queIdx = 0; queIdx < tempResReq.queNum; queIdx++) {
601 : // find neighbors : virtualRank
602 0 : u32 remoteAlgRank = (myAlgRank + 1 + queIdx + partialRankSize) % partialRankSize;
603 0 : if (remoteAlgRank >= tempVTopo[i].size()) {
604 0 : continue;
605 : }
606 0 : RankId neighborRank = tempVTopo[i][remoteAlgRank];
607 0 : HCCL_DEBUG("tempVTopo[%u] index[%u] value[%d]", i, remoteAlgRank, neighborRank);
608 0 : auto rankInRankSet = std::find(commTargetUserRankSet_.begin(), commTargetUserRankSet_.end(),
609 0 : static_cast<u32>(neighborRank));
610 0 : if (rankInRankSet != commTargetUserRankSet_.end() && neighborRank != myRank) {
611 : // LinkNum
612 0 : tempResReq.links[neighborRank] = linkNumBtwPeers;
613 0 : HCCL_DEBUG("myRank[%d] neighborRank[%d] links is [%u]", myRank, neighborRank, linkNumBtwPeers);
614 : }
615 : }
616 : }
617 :
618 0 : return HcclResult::HCCL_SUCCESS;
619 : }
620 :
621 : template <typename AlgTopoMatch>
622 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CalcRes(AlgTempResReq &tempResReq)
623 : {
624 0 : InsTempAllGatherMesh1D tempAlg(myRank_, rankSize_, vTopo_, virtRankMap_);
625 0 : tempResReq.queNum = commTargetUserRankSet_.size() + 1; // 使用n条从流
626 0 : tempResReq.streamNum = tempResReq.queNum;
627 0 : tempResReq.queNotifys = tempAlg.CreateMasterSlaveQueNotifiesRequest(tempResReq.queNum);
628 :
629 0 : QId centerQ = 0;
630 0 : tempResReq.localWaitGroupCntNotify.emplace_back(centerQ, 0);
631 0 : tempResReq.localBcastPostCntNotify.emplace_back(centerQ, 0);
632 :
633 0 : CHK_RET(CalcResLinksPartialMesh(myRank_, vTopo_, 1, tempResReq));
634 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][CalcRes] Rank[%d] vTopoSize[%lu] requiredQue Num[%u].",
635 : myRank_, vTopo_[0].size(), tempResReq.queNum);
636 0 : return HcclResult::HCCL_SUCCESS;
637 0 : }
638 :
639 : template <typename AlgTopoMatch>
640 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CalcResOffload(const RankGraph *rankGraph,
641 : const u64 &dataSize,
642 : CollOffloadOpResReq &resReq)
643 : {
644 : (void)dataSize;
645 0 : resReq.requiredScratchMemSize = 0;
646 : // Topo Match
647 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
648 0 : CHK_RET(topoMatch.SetTargetRanks(commTargetUserRankSet_));
649 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
650 :
651 : // calculate required insQues and prepare queue
652 0 : AlgTempResReq tempResReq;
653 0 : if (enableDetour_) {
654 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
655 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
656 : } else {
657 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
658 0 : CHK_RET(CalcRes(tempResReq));
659 : }
660 :
661 0 : resReq.requiredSubQueNum = commTargetUserRankSet_.size();
662 :
663 0 : return HcclResult::HCCL_SUCCESS;
664 0 : }
665 :
666 : template <typename AlgTopoMatch>
667 0 : HcclResult InsBatchSendRecvExecutor<AlgTopoMatch>::CalcRes(const RankGraph *rankGraph,
668 : CollAlgResReq &algResReq)
669 : {
670 : // Topo Match
671 0 : AlgTopoMatch topoMatch(myRank_, rankSize_, rankGraph, devType_);
672 0 : CHK_RET(topoMatch.SetTargetRanks(commTargetUserRankSet_));
673 0 : CHK_RET(topoMatch.MatchTopo(vTopo_, virtRanks_, virtRankMap_));
674 :
675 0 : algResReq.topoInfo.UpdateSingleLevelTopo(virtRanks_, virtRankMap_, vTopo_);
676 :
677 0 : for (u32 i = 0; i < vTopo_.size(); i++) { // 遍历level0
678 0 : for (u32 j = 0; j < vTopo_[i].size(); j++) { // 遍历平面内的所有rank
679 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][CalcResLinksPartialMesh] vTopo_[%u][%u] is [%d].",
680 : i, j, vTopo_[i][j]);
681 : }
682 : }
683 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor][CalcRes]topoInfo.virtRanks[%u], topoInfo.virtRankMap[%u],"\
684 : "topoInfo.vTopo[%u]", algResReq.topoInfo.virtRanks.size(),
685 : algResReq.topoInfo.virtRankMap.size(), algResReq.topoInfo.vTopo.size());
686 :
687 : // calculate required insQues and prepare queue
688 0 : AlgTempResReq tempResReq;
689 0 : if (enableDetour_) {
690 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], CalcRes with detouring enabled.", myRank_);
691 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
692 : } else {
693 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], CalcRes with detouring disabled.", myRank_);
694 0 : CHK_RET(CalcRes(tempResReq));
695 : }
696 :
697 0 : algResReq.primQueueNum = tempResReq.streamNum;
698 0 : algResReq.queueNotifys = tempResReq.queNotifys;
699 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], requiredQueNum [%u].", myRank_, algResReq.primQueueNum);
700 :
701 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
702 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
703 0 : HCCL_DEBUG("[InsBatchSendRecvExecutor] Rank[%d], algResReq.links size[%zu].", myRank_, algResReq.links.size());
704 :
705 0 : return HcclResult::HCCL_SUCCESS;
706 0 : }
707 :
708 : // 注册
709 : INS_REGISTER_IMPL_BY_TOPO(OpType::BATCHSENDRECV, InsBatchSendRecv, InsBatchSendRecvExecutor, TopoMatchPartialMesh);
710 :
711 : } // namespace Hccl
|
