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