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 "coll_batch_send_recv_executor.h"
12 :
13 : namespace hccl {
14 : constexpr u32 RANKSIZE_TWO = 2;
15 :
16 64 : CollBatchSendRecvExecutor::CollBatchSendRecvExecutor(const HcclDispatcher dispatcher,
17 64 : std::unique_ptr<TopoMatcher> &topoMatcher)
18 64 : : CollCommExecutor(dispatcher, topoMatcher)
19 : {
20 64 : }
21 :
22 0 : void CollBatchSendRecvExecutor::ParseParam(const OpParam& param)
23 : {
24 0 : tag_ = param.tag;
25 0 : HcclSendRecvItem* itemPtr = param.BatchSendRecvDataDes.sendRecvItemsPtr;
26 0 : u32 itemNum = param.BatchSendRecvDataDes.itemNum;
27 0 : if (itemPtr == nullptr) {
28 0 : HCCL_ERROR("[CollBatchSendRecvExecutor][ParseParam] sendRecvInfo is nullptr.");
29 : }
30 0 : commTargetUserRankSet_.clear();
31 0 : for (u32 i = 0; i < itemNum; i++) {
32 0 : commTargetUserRankSet_.insert((itemPtr + i)->remoteRank);
33 0 : HCCL_INFO("[CollBatchSendRecvExecutor][ParseParam] insert remoteUserRank[%u] to Set ",
34 : (itemPtr + i)->remoteRank);
35 : }
36 0 : aicpuUnfoldMode_ = param.aicpuUnfoldMode;
37 0 : }
38 :
39 0 : HcclResult CollBatchSendRecvExecutor::CalcIncreLinkRequest(const OpParam& param, std::set<u32>& ranksLinked,
40 : AlgResourceRequest& resourceRequest, bool& needIncreLink)
41 : {
42 0 : needIncreLink = false;
43 0 : (void)ParseParam(param);
44 0 : for (auto& remoteRank : commTargetUserRankSet_) {
45 0 : if (ranksLinked.find(remoteRank) == ranksLinked.end()) {
46 0 : needIncreLink = true;
47 0 : ranksLinked.insert(remoteRank);
48 0 : HCCL_INFO("[CollBatchSendRecvExecutor][CalcIncreLinkRequest] Start insert remoteUserRank[%u] to "\
49 : "ranksLinked Set.", remoteRank);
50 : }
51 : }
52 0 : CHK_PRT_RET(!needIncreLink, HCCL_INFO("[CollBatchSendRecvExecutor][CalcIncreLinkRequest] It's "\
53 : "unnecessary to incre alloc link."), HCCL_SUCCESS);
54 :
55 0 : u64 scratchMemSize = 0U;
56 0 : u32 streamNum = 0U;
57 0 : u32 notifyNum = 0U;
58 0 : u64 aivBufferRequest = 0U;
59 :
60 : std::vector<LevelNSubCommTransport> opTransport {
61 0 : std::vector<LevelNSubCommTransport>(static_cast<u32>(COMM_LEVEL_RESERVED))
62 0 : };
63 0 : CHK_RET(CalcCommInfo(opTransport));
64 0 : CHK_RET(BuildResourceRequest(scratchMemSize, streamNum, notifyNum, aivBufferRequest, opTransport, resourceRequest));
65 0 : return HCCL_SUCCESS;
66 0 : }
67 :
68 1 : HcclResult CollBatchSendRecvExecutor::GetPairWiseList(HcclSendRecvItem *sendRecvInfo, u32 itemNum)
69 : {
70 1 : HCCL_INFO("[CollBatchSendRecvExecutor][GetPairWiseList] Start sort the batchSendRecv tasklist.");
71 1 : CHK_PTR_NULL(sendRecvInfo);
72 :
73 0 : for (u32 i = 0; i < itemNum; i++) {
74 0 : HCCL_INFO("[CollBatchSendRecvExecutor][GetPairWiseList] index is %u, itemNum is %u, localRankID is %u, remoteRank is %u, "\
75 : "sendRecvType is %u, rankSize is %u.", i, itemNum, topoAttr_.userRank, sendRecvInfo->remoteRank,
76 : static_cast<u32>(sendRecvInfo->sendRecvType), topoAttr_.userRankSize);
77 0 : CHK_PTR_NULL(sendRecvInfo->buf);
78 :
79 0 : if (sendRecvInfo->sendRecvType == HcclSendRecvType::HCCL_SEND) {
80 0 : sendDeque_.push_back(sendRecvInfo);
81 0 : } else if (sendRecvInfo->sendRecvType == HcclSendRecvType::HCCL_RECV) {
82 0 : recvDeque_.push_back(sendRecvInfo);
83 : } else {
84 0 : HCCL_ERROR("[CollBatchSendRecvExecutor][GetPairWiseList] sendRecvType wrong sendrecvType is %d, "\
85 : "rankID is %u, remoteRank is %u.", sendRecvInfo->sendRecvType, topoAttr_.userRank,
86 : sendRecvInfo->remoteRank);
87 0 : return HCCL_E_PARA;
88 : }
89 0 : sendRecvInfo++;
90 : }
91 :
92 : /* 此处的排序逻辑(pair-wise算法):
93 : 1.sendDeque元素顺序是:先放remoteRank号小于等于root rank的第一个任务，依次减小(循环索引)直至放完
94 : 2.recvDeque元素顺序是:先放remoteRank号大于等于root rank的第一个任务，依次增大(循环索引)直至放完
95 : */
96 0 : auto sendCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
97 0 : u32 aFlag = (a->remoteRank <= topoAttr_.userRank) ? (a->remoteRank + topoAttr_.userRankSize) : a->remoteRank;
98 0 : u32 bFlag = (b->remoteRank <= topoAttr_.userRank) ? (b->remoteRank + topoAttr_.userRankSize) : b->remoteRank;
99 0 : return aFlag > bFlag;
100 0 : };
101 :
102 0 : auto recvCompare = [this](HcclSendRecvItem* a, HcclSendRecvItem* b) {
103 0 : u32 aFlag = (a->remoteRank < topoAttr_.userRank) ? (a->remoteRank + topoAttr_.userRankSize) : a->remoteRank;
104 0 : u32 bFlag = (b->remoteRank < topoAttr_.userRank) ? (b->remoteRank + topoAttr_.userRankSize) : b->remoteRank;
105 0 : return aFlag < bFlag;
106 0 : };
107 :
108 0 : std::stable_sort(sendDeque_.begin(), sendDeque_.end(), sendCompare);
109 0 : std::stable_sort(recvDeque_.begin(), recvDeque_.end(), recvCompare);
110 :
111 0 : while ((!sendDeque_.empty() && sendDeque_.front()->remoteRank == topoAttr_.userRank) &&
112 0 : (!recvDeque_.empty() && recvDeque_.front()->remoteRank == topoAttr_.userRank)) {
113 0 : sendToSelfDeque_.push_back(sendDeque_.front());
114 0 : recvFromSelfDeque_.push_back(recvDeque_.front());
115 0 : sendDeque_.pop_front();
116 0 : recvDeque_.pop_front();
117 : }
118 : // 如果自发自收任务没有完全匹配
119 0 : if ((!sendDeque_.empty() && sendDeque_.front()->remoteRank == topoAttr_.userRank) ||
120 0 : (!recvDeque_.empty() && recvDeque_.front()->remoteRank == topoAttr_.userRank)) {
121 0 : HCCL_ERROR("[CollBatchSendRecvExecutor] SendTask and Recv Task to rank itself do not match,"\
122 : "please check the task list.");
123 0 : return HCCL_E_PARA;
124 : }
125 0 : HCCL_INFO("[CollBatchSendRecvExecutor][GetPairWiseList] End sort the batchSendRecv tasklist.");
126 0 : return HCCL_SUCCESS;
127 : }
128 :
129 0 : HcclResult CollBatchSendRecvExecutor::ProcessSelfSendRecvTasks(Stream& stream)
130 : {
131 0 : while (!sendToSelfDeque_.empty() && !recvFromSelfDeque_.empty()) {
132 0 : if (sendToSelfDeque_.front()->count == recvFromSelfDeque_.front()->count &&
133 0 : sendToSelfDeque_.front()->dataType == recvFromSelfDeque_.front()->dataType) {
134 0 : u64 dataSize = sendToSelfDeque_.front()->count * SIZE_TABLE[sendToSelfDeque_.front()->dataType];
135 :
136 0 : DeviceMem inUserMem = DeviceMem::create(static_cast<u8*>(sendToSelfDeque_.front()->buf), dataSize);
137 0 : DeviceMem outUserMem = DeviceMem::create(static_cast<u8*>(recvFromSelfDeque_.front()->buf), dataSize);
138 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outUserMem, inUserMem, stream));
139 0 : sendToSelfDeque_.pop_front();
140 0 : recvFromSelfDeque_.pop_front();
141 0 : } else {
142 0 : HCCL_ERROR("[HcclBatchSendRecv] Send task and recv task to self : count or dataType do not equal, please"\
143 : "check the task list.");
144 0 : return HCCL_E_PARA;
145 : }
146 : }
147 0 : return HCCL_SUCCESS;
148 : }
149 :
150 0 : HcclResult CollBatchSendRecvExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algResource)
151 : {
152 0 : HcclUs startut = TIME_NOW();
153 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollBatchSendRecvExecutor] batchsendrecv starts.");
154 :
155 0 : algResResp_ = &algResource;
156 0 : CHK_RET(CheckCommSize(COMM_COMBINE_ORDER, COMM_SIZE_TWO));
157 0 : CHK_RET(GetPairWiseList(param.BatchSendRecvDataDes.sendRecvItemsPtr, param.BatchSendRecvDataDes.itemNum));
158 0 : CHK_RET(ProcessSelfSendRecvTasks(param.stream));
159 0 : if (topoAttr_.userRankSize == 1) {
160 0 : HCCL_INFO("tag[%s] BatchSendRecv Executor orchestrate success, take time [%lld]us.",
161 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
162 0 : return HCCL_SUCCESS;
163 : }
164 0 : CHK_RET(CalcSendSlices(algResource));
165 0 : CHK_RET(CalcRecvSlices(algResource));
166 :
167 0 : HCCL_DEBUG("[CollBatchSendRecvExecutor][Orchestrate] aicpuUnfoldMode %d", aicpuUnfoldMode_);
168 0 : if(aicpuUnfoldMode_) {
169 0 : CHK_RET(RunLoopInAicpuUnfoldMode(param));
170 : } else {
171 0 : CHK_RET(RunLoopInHostUnfoldMode(param));
172 : }
173 0 : HCCL_INFO("tag[%s] BatchSendRecv Executor orchestrate success, take time [%lld]us.",
174 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
175 0 : return HCCL_SUCCESS;
176 : }
177 :
178 6 : HcclResult CollBatchSendRecvExecutor::GetSendTargetLink(u32 remoteUserRank, LINK& targetLink) {
179 6 : u32 commIndex = 0;
180 6 : HCCL_INFO("[CollBatchSendRecvExecutor][GetSendTargetLink] remoteUserRank[%u], localUserRank_[%u].",
181 : remoteUserRank, topoAttr_.userRank);
182 6 : if (remoteUserRank < topoAttr_.userRank) {
183 5 : HCCL_INFO("[CollBatchSendRecvExecutor][GetSendTargetLink] CommIndex is 0.");
184 5 : commIndex = COMM_INDEX_0;
185 1 : } else if (remoteUserRank > topoAttr_.userRank) {
186 1 : HCCL_INFO("[CollBatchSendRecvExecutor][GetSendTargetLink] CommIndex is 1.");
187 1 : commIndex = COMM_INDEX_1;
188 : } else {
189 0 : HCCL_ERROR("[CollBatchSendRecvExecutor][GetSendTargetLink] CommIndex doesn't match.");
190 0 : return HCCL_E_PARA;
191 : }
192 6 : CHK_RET(GetTransport(commIndex, remoteUserRank, targetLink));
193 6 : CHK_SMART_PTR_NULL(targetLink);
194 :
195 6 : return HCCL_SUCCESS;
196 : }
197 :
198 2 : HcclResult CollBatchSendRecvExecutor::GetRecvTargetLink(u32 remoteUserRank, LINK& targetLink) {
199 2 : u32 commIndex = 0;
200 2 : HCCL_INFO("[CollBatchSendRecvExecutor][GetRecvTargetLink] remoteUserRank[%u], localUserRank_[%u].",
201 : remoteUserRank, topoAttr_.userRank);
202 2 : if (remoteUserRank > topoAttr_.userRank) {
203 2 : HCCL_INFO("[CollBatchSendRecvExecutor][GetRecvTargetLink] CommIndex is 0.");
204 2 : commIndex = COMM_INDEX_0;
205 0 : } else if (remoteUserRank < topoAttr_.userRank) {
206 0 : HCCL_INFO("[CollBatchSendRecvExecutor][GetRecvTargetLink] CommIndex is 1.");
207 0 : commIndex = COMM_INDEX_1;
208 : } else {
209 0 : HCCL_ERROR("[CollBatchSendRecvExecutor][GetRecvTargetLink] CommIndex doesn't match.");
210 0 : return HCCL_E_PARA;
211 : }
212 2 : CHK_RET(GetTransport(commIndex, remoteUserRank, targetLink));
213 2 : CHK_SMART_PTR_NULL(targetLink);
214 :
215 2 : return HCCL_SUCCESS;
216 : }
217 :
218 0 : HcclResult CollBatchSendRecvExecutor::RunLoopInHostUnfoldMode(OpParam& param)
219 : {
220 0 : if (static_cast<bool>(topoMatcher_->GetExternalInputHcclEnableFfts())) {
221 0 : auto meta = HcclOpMetaInfo::GetOneForBatchSendRecv();
222 0 : CHK_RET(InitTask(dispatcher_, param.stream, meta.isEnableCache, meta.GetCacheKey()));
223 : // 多流子图前后需加空拷贝
224 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(algResResp_->cclInputMem, algResResp_->cclOutputMem, param.stream,
225 : dispatcher_));
226 : }
227 0 : bool isSetNormalMode = false; // 设置过一次就不需要再设置了
228 0 : for (u32 i = 0; i < sendDataSilces_.size(); ++i) {
229 0 : SendRecvSlice& slice = sendDataSilces_[i];
230 0 : LINK targetLink;
231 0 : CHK_RET(GetSendTargetLink(slice.remoteRank, targetLink));
232 0 : if (TransportType::TRANS_TYPE_DEVICE_DIRECT == targetLink->GetTransportType()) {
233 0 : CHK_RET(SetNormalMode(dispatcher_));
234 0 : isSetNormalMode = true;
235 0 : HCCL_INFO("[CollBatchSendRecvExecutor][RunLoopInHostUnfoldMode]Send Set NormalMode true");
236 0 : break;
237 : }
238 0 : }
239 :
240 0 : for (u32 i = 0; i < recvDataSilces_.size() && !isSetNormalMode; ++i) {
241 0 : SendRecvSlice& slice = recvDataSilces_[i];
242 0 : LINK targetLink;
243 0 : CHK_RET(GetRecvTargetLink(slice.remoteRank, targetLink));
244 0 : if (targetLink->GetTransportType() == TransportType::TRANS_TYPE_DEVICE_DIRECT) {
245 0 : CHK_RET(SetNormalMode(dispatcher_));
246 0 : HCCL_INFO("[CollBatchSendRecvExecutor][RunLoopInHostUnfoldMode]Recv Set NormalMode dispatcher");
247 0 : break;
248 : }
249 0 : }
250 :
251 0 : CHK_RET(MainPostSubWait(param.stream, algResResp_->slaveStreams[STREAM_INDEX_0]));
252 0 : HCCL_INFO("[BatchSendRecv] Stream sync: main stream record, subStream wait.");
253 0 : while (!sendDataSilces_.empty() || !recvDataSilces_.empty()) {
254 0 : if(!sendDataSilces_.empty()) {
255 0 : CHK_RET(ProcessSendDataSlice(param.stream, false, false));
256 0 : sendDataSilces_.pop_front();
257 : }
258 0 : if(!recvDataSilces_.empty()) {
259 0 : CHK_RET(ProcessRecvDataSlice(algResResp_->slaveStreams[STREAM_INDEX_0], false));
260 0 : recvDataSilces_.pop_front();
261 : }
262 : }
263 :
264 0 : CHK_RET(SubPostMainWait(param.stream, algResResp_->slaveStreams[STREAM_INDEX_0]));
265 0 : HCCL_INFO("[BatchSendRecv] Stream sync: subStream record, main stream wait.");
266 0 : if (static_cast<bool>(topoMatcher_->GetExternalInputHcclEnableFfts())) {
267 : // 多流子图前后需加空拷贝
268 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(algResResp_->cclInputMem,
269 : algResResp_->cclOutputMem, param.stream, dispatcher_));
270 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
271 : }
272 0 : return HCCL_SUCCESS;
273 : }
274 :
275 0 : HcclResult CollBatchSendRecvExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
276 : {
277 0 : algResResp_ = &algRes;
278 0 : SubCommInfo level1CommInfo = {0};
279 0 : AdjInfo nslbAdjInfo = {0};
280 0 : if (Getlevel1CommRank(level1CommInfo) != HCCL_SUCCESS) {
281 0 : return HCCL_SUCCESS;
282 : }
283 0 : u32 localRank= level1CommInfo.localRank;
284 0 : u32 localRankSize = level1CommInfo.localRankSize;
285 :
286 0 : std::unique_ptr<AlgTemplateBase> level1TempAlg;
287 0 : if (SelectTempAlg(level1TempAlg, localRankSize) != HCCL_SUCCESS) {
288 0 : return HCCL_SUCCESS;
289 : }
290 0 : if(level1TempAlg == nullptr) {
291 0 : return HCCL_SUCCESS;
292 : }
293 0 : CHK_RET(level1TempAlg->GetNslbAdjInfo(localRank, localRankSize, level1CommInfo.links, nslbAdjInfo));
294 :
295 0 : adjInfo.dstRankNum = nslbAdjInfo.dstRankNum;
296 0 : HCCL_INFO("[nslbdp] adjInfo.dstRankNum[%u].", adjInfo.dstRankNum);
297 :
298 0 : for (size_t i = 0; i < nslbAdjInfo.nsAdjInfo.size(); i++) {
299 0 : NslbDpAdjInfo dpAdjInfo = {0};
300 0 : dpAdjInfo.dstLocalRankId = nslbAdjInfo.nsAdjInfo[i].dstLocalRankId;
301 0 : dpAdjInfo.phaseId = nslbAdjInfo.nsAdjInfo[i].phaseId;
302 0 : dpAdjInfo.rev = 0;
303 0 : adjInfo.nsAdjInfo.push_back(dpAdjInfo);
304 0 : HCCL_INFO("[nslbdp]GetAdjInfo dstLocalRankId[%u], phaseId[%u].",
305 : nslbAdjInfo.nsAdjInfo[i].dstLocalRankId, nslbAdjInfo.nsAdjInfo[i].phaseId);
306 : }
307 0 : return HCCL_SUCCESS;
308 0 : }
309 :
310 0 : HcclResult CollBatchSendRecvExecutor::RunLoopInAicpuUnfoldMode(OpParam& param)
311 : {
312 0 : CHK_RET(MainPostSubWait(param.stream, algResResp_->slaveStreams[STREAM_INDEX_0]));
313 0 : u32 loopInOnceLaunch = 0;
314 : // 每隔200个loop launch一次
315 0 : while (!sendDataSilces_.empty() || !recvDataSilces_.empty()) {
316 0 : if(!sendDataSilces_.empty()) {
317 0 : CHK_RET(ProcessSendDataSlice(param.stream, false, false));
318 0 : sendDataSilces_.pop_front();
319 : }
320 0 : if(!recvDataSilces_.empty()) {
321 0 : CHK_RET(ProcessRecvDataSlice(algResResp_->slaveStreams[STREAM_INDEX_0], false));
322 0 : recvDataSilces_.pop_front();
323 : }
324 0 : loopInOnceLaunch++;
325 0 : if (loopInOnceLaunch == MAX_LOOP_IN_ONCE_LAUNCH) {
326 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
327 0 : loopInOnceLaunch = 0;
328 0 : HCCL_INFO("[BatchSendRecv] LaunchTaskExtend, unprocessed send slices[%u], recv slices[%u].",
329 : sendDataSilces_.size(), recvDataSilces_.size());
330 : }
331 : }
332 0 : CHK_RET(SubPostMainWait(param.stream, algResResp_->slaveStreams[STREAM_INDEX_0]));
333 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
334 0 : return HCCL_SUCCESS;
335 : }
336 :
337 0 : HcclResult CollBatchSendRecvExecutor::MainPostSubWait(Stream& mainStream, Stream& subStream)
338 : {
339 0 : CHK_RET(LocalNotify::Post(mainStream, dispatcher_, algResResp_->notifiesAux[STREAM_INDEX_0], PROF_STAGE_0));
340 0 : CHK_RET(LocalNotify::Wait(subStream, dispatcher_,
341 : algResResp_->notifiesAux[STREAM_INDEX_0], PROF_STAGE_0));
342 0 : return HCCL_SUCCESS;
343 : }
344 :
345 0 : HcclResult CollBatchSendRecvExecutor::SubPostMainWait(Stream& mainStream, Stream& subStream)
346 : {
347 0 : CHK_RET(LocalNotify::Post(subStream, dispatcher_,
348 : algResResp_->notifiesMain[STREAM_INDEX_0], PROF_STAGE_0));
349 :
350 0 : CHK_RET(LocalNotify::Wait(mainStream, dispatcher_, algResResp_->notifiesMain[STREAM_INDEX_0],
351 : PROF_STAGE_0));
352 0 : return HCCL_SUCCESS;
353 : }
354 :
355 0 : HcclResult CollBatchSendRecvExecutor::CalcSendSlices(AlgResourceResponse& algRes)
356 : {
357 0 : while (!sendDeque_.empty()) {
358 0 : HcclSendRecvItem* sendRecvItem = sendDeque_.front();
359 0 : HCCL_INFO("[CollBatchSendRecvExecutor][CalcSendSlices] tag[%s], remoteRank[%u], buf[%p], count[%llu],"\
360 : "dataType[%s], sendRecvType[%d].", tag_.c_str(), sendRecvItem->remoteRank, sendRecvItem->buf,
361 : sendRecvItem->count, GetDataTypeEnumStr(sendRecvItem->dataType).c_str(), sendRecvItem->sendRecvType);
362 0 : u8 *curInputPtr = static_cast<u8 *>(sendRecvItem->buf);
363 0 : CHK_PTR_NULL(curInputPtr);
364 0 : u32 unitSize = SIZE_TABLE[sendRecvItem->dataType];
365 0 : u64 maxCountPerLoop = CalcSendLoopMaxCount(const_cast<DeviceMem&>(algRes.cclInputMem), unitSize);
366 :
367 0 : for (u64 countLeft = sendRecvItem->count, curCount = 0, curOffset = 0; countLeft > 0;
368 0 : countLeft -= curCount) {
369 0 : curInputPtr += curOffset;
370 0 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
371 0 : u64 curSize = curCount * unitSize; // 单位：字节
372 0 : sendDataSilces_.emplace_back(curInputPtr, curSize, sendRecvItem->remoteRank);
373 0 : HCCL_DEBUG("[CollBatchSendRecvExecutor][CalcSendSlices] tag[%s], slice userAddr[%p], slice size[%llu].",
374 : tag_.c_str(), curInputPtr, curSize);
375 0 : curOffset = curSize;
376 : }
377 0 : sendDeque_.pop_front();
378 : }
379 0 : return HCCL_SUCCESS;
380 : }
381 :
382 0 : HcclResult CollBatchSendRecvExecutor::CalcRecvSlices(AlgResourceResponse& algRes)
383 : {
384 0 : while (!recvDeque_.empty()) {
385 0 : HcclSendRecvItem* sendRecvItem = recvDeque_.front();
386 0 : HCCL_INFO("[CollBatchSendRecvExecutor][CalcSendSlices] tag[%s], remoteRank[%u], buf[%p], count[%llu],"\
387 : "dataType[%s], sendRecvType[%d].", tag_.c_str(), sendRecvItem ->remoteRank, sendRecvItem ->buf, sendRecvItem->count,
388 : GetDataTypeEnumStr(sendRecvItem->dataType).c_str(), sendRecvItem->sendRecvType);
389 0 : u8 *curOutputPtr = static_cast<u8*>(sendRecvItem->buf);
390 0 : CHK_PTR_NULL(curOutputPtr);
391 0 : u32 unitSize = SIZE_TABLE[sendRecvItem->dataType];
392 0 : u64 maxCountPerLoop = CalcRecvLoopMaxCount(const_cast<DeviceMem&>(algRes.cclOutputMem), unitSize);
393 :
394 0 : for (u64 countLeft = sendRecvItem->count, curCount = 0, curOffset = 0; countLeft > 0;
395 0 : countLeft -= curCount) {
396 0 : curOutputPtr += curOffset;
397 0 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
398 0 : u64 curSize = curCount * unitSize; // 单位：字节
399 0 : recvDataSilces_.emplace_back(curOutputPtr, curSize, sendRecvItem->remoteRank);
400 0 : HCCL_DEBUG("[CollBatchSendRecvExecutor][CalcRecvSlices] tag[%s], slice userAddr[%p], slice size[%llu].",
401 : tag_.c_str(), curOutputPtr, curSize);
402 0 : curOffset = curSize;
403 : }
404 0 : recvDeque_.pop_front();
405 : }
406 0 : return HCCL_SUCCESS;
407 : }
408 :
409 0 : HcclResult CollBatchSendRecvExecutor::ProcessSendDataSlice(Stream& stream, bool needStreamSync, bool retryEnable)
410 : {
411 0 : SendRecvSlice& slice = sendDataSilces_.front();
412 0 : DeviceMem inMem(slice.addr, slice.size);
413 0 : DeviceMem inCommMem = algResResp_->cclInputMem.range(0, slice.size);
414 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, inCommMem, inMem, stream));
415 0 : if (needStreamSync) {
416 0 : CHK_RET(MainPostSubWait(stream, algResResp_->slaveStreams[STREAM_INDEX_0]));
417 : }
418 :
419 0 : ExecMem execMem;
420 0 : execMem.inputMem = inCommMem;
421 0 : HcclResult ret = SendKernelRun(stream, execMem, slice.remoteRank, retryEnable);
422 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
423 : HCCL_ERROR("[CollBatchSendRecvExecutor][ProcessSendDataSlice]errNo[0x%016llx]kernel run error, tag[%s], " \
424 : "input_ptr[%p], size[%llu]", HCCL_ERROR_CODE(ret), tag_.c_str(), execMem.inputMem.ptr(),
425 : slice.size), ret);
426 0 : return HCCL_SUCCESS;
427 0 : }
428 :
429 0 : HcclResult CollBatchSendRecvExecutor::ProcessRecvDataSlice(Stream& stream, bool retryEnable)
430 : {
431 0 : SendRecvSlice& slice = recvDataSilces_.front();
432 0 : ExecMem execMem;
433 0 : execMem.outputMem = algResResp_->cclOutputMem.range(0, slice.size);
434 :
435 0 : HcclResult ret = RecvKernelRun(stream, execMem, slice.remoteRank, retryEnable);
436 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
437 : HCCL_ERROR("[CollBatchSendRecvExecutor][ProcessRecvDataSlice]errNo[0x%016llx]kernel run error, tag[%s], " \
438 : "output_ptr[%p], size[%llu]", HCCL_ERROR_CODE(ret), tag_.c_str(), execMem.outputMem.ptr(),
439 : slice.size), ret);
440 :
441 0 : DeviceMem outMem(slice.addr, slice.size);
442 0 : DeviceMem outCommMem = execMem.outputMem;
443 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outMem, outCommMem, stream));
444 0 : return HCCL_SUCCESS;
445 0 : }
446 :
447 0 : HcclResult CollBatchSendRecvExecutor::SendKernelRun(Stream& stream, ExecMem &execMem, u32 remoteUserRank,
448 : bool retryEnable)
449 : {
450 0 : LINK targetLink;
451 0 : CHK_RET(GetSendTargetLink(remoteUserRank, targetLink));
452 0 : SendReceive executor(dispatcher_, targetLink, INVALID_VALUE_RANKID, HCCL_CHUNK_SIZE, retryEnable);
453 0 : CHK_RET(executor.SendPrepare(execMem.inputMem, remoteUserRank, stream));
454 0 : CHK_RET(executor.RegisterProfiler(0, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, stream));
455 0 : CHK_RET(executor.BatchSendRunAsync());
456 :
457 0 : return HCCL_SUCCESS;
458 0 : }
459 :
460 0 : HcclResult CollBatchSendRecvExecutor::RecvKernelRun(Stream& stream, ExecMem &execMem, u32 remoteUserRank,
461 : bool retryEnable)
462 : {
463 0 : LINK targetLink;
464 0 : CHK_RET(GetRecvTargetLink(remoteUserRank, targetLink));
465 0 : SendReceive executor(dispatcher_, targetLink, INVALID_VALUE_RANKID, HCCL_CHUNK_SIZE, retryEnable);
466 0 : CHK_RET(executor.ReceivePrepare(execMem.outputMem, remoteUserRank, stream));
467 0 : CHK_RET(executor.RegisterProfiler(0, PROF_STAGE_0, HCCL_EXEC_STEP_NOT_SET, stream));
468 0 : CHK_RET(executor.BatchReceiveRunAsync());
469 0 : return HCCL_SUCCESS;
470 0 : }
471 :
472 8 : HcclResult CollBatchSendRecvExecutor::GetTransport(u32 commIndex, u32 remoteUserRank, LINK &targetLink)
473 : {
474 8 : CHK_PRT_RET(commIndex >= algResResp_->opTransportResponse[COMM_COMBINE_ORDER].size(),
475 : HCCL_ERROR("[CollBatchSendRecvExecutor][KernelRun] batchsendrecv op commIndex[%u] is larger than "\
476 : "opTransportResponse size[%zu]",
477 : remoteUserRank, algResResp_->opTransportResponse[COMM_COMBINE_ORDER].size()), HCCL_E_PARA);
478 : SingleSubCommTransport &commCombined =
479 8 : const_cast<SingleSubCommTransport&>(algResResp_->opTransportResponse[COMM_COMBINE_ORDER][commIndex]);
480 :
481 8 : CHK_PRT_RET(remoteUserRank >= commCombined.userRank2subCommRank.size(),
482 : HCCL_ERROR("[CollBatchSendRecvExecutor][KernelRun] batchsendrecv op remoteUserRank[%u] is larger than "\
483 : "userRank2subCommRank map size[%zu]",
484 : remoteUserRank, commCombined.userRank2subCommRank.size()), HCCL_E_PARA);
485 :
486 8 : u32 remoteRank = commCombined.userRank2subCommRank[remoteUserRank];
487 8 : CHK_PRT_RET(remoteRank >= commCombined.links.size(),
488 : HCCL_ERROR("[CollBatchSendRecvExecutor][KernelRun] batchsendrecv op remoteUserRank[%u], get remoteRank[%u]," \
489 : "the size of combinedComm links is [%zu]", remoteUserRank, remoteRank, commCombined.links.size()),
490 : HCCL_E_PARA);
491 8 : targetLink = commCombined.links[remoteRank];
492 8 : return HCCL_SUCCESS;
493 : }
494 :
495 0 : u64 CollBatchSendRecvExecutor::CalcSendLoopMaxCount(DeviceMem& inCCLBuffer, const u32 unitSize)
496 : {
497 : // 中转内存单次最多能够接受的input count
498 0 : u64 maxCountPerLoop = inCCLBuffer.size() / unitSize;
499 0 : HCCL_WARNING("[CollBatchSendRecvExecutor][CalcSendLoopMaxCount]" \
500 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.", maxCountPerLoop);
501 0 : return maxCountPerLoop;
502 : }
503 :
504 0 : u64 CollBatchSendRecvExecutor::CalcRecvLoopMaxCount(DeviceMem& outCCLBuffer, const u32 unitSize)
505 : {
506 : // 中转内存单次最多能够接受的output count
507 0 : u64 maxCountPerLoop = outCCLBuffer.size() / unitSize;
508 0 : HCCL_WARNING("[CollBatchSendRecvExecutor][CalcRecvLoopMaxCount]" \
509 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.", maxCountPerLoop);
510 0 : return maxCountPerLoop;
511 : }
512 :
513 0 : HcclResult CollBatchSendRecvExecutor::CalcStreamNum(u32& streamNum)
514 : {
515 0 : streamNum = 1U;
516 0 : HCCL_INFO("[CollBatchSendRecvExecutor][CalcScratchMemSize] tag_[%s], streamNum[%u].", tag_.c_str(), streamNum);
517 0 : return HCCL_SUCCESS;
518 : }
519 0 : HcclResult CollBatchSendRecvExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
520 : {
521 : CommParaInfo commParaInfo(COMM_COMBINE_ORDER, CommType::COMM_TAG_PARTIAL_MESH_COMBINED, INVALID_VALUE_RANKID,
522 0 : INVALID_VALUE_RANKID, false, false, commTargetUserRankSet_);
523 0 : TransportMemType inputType = TransportMemType::CCL_INPUT;
524 0 : TransportMemType outputType = TransportMemType::CCL_OUTPUT;
525 :
526 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_COMBINE_ORDER], inputType, outputType));
527 0 : return HCCL_SUCCESS;
528 0 : }
529 :
530 : REGISTER_EXEC("BatchSendRecv", BatchSendRecvExecutor, CollBatchSendRecvExecutor);
531 : } // namespace hccl
|
