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