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 "hccl_communicator.h"
12 : #include <atomic>
13 : #include <chrono>
14 : #include <thread>
15 : #include <algorithm>
16 : #include <numeric>
17 : #include <unordered_set>
18 : #include <memory>
19 : #include "externalinput_pub.h"
20 : #include "opexecounter_pub.h"
21 : // ltm指定config路径
22 : #include "common/src/config.h"
23 : #include "stream_active_manager.h"
24 : #include "profiling_manager_pub.h"
25 : #include "hccl_aiv.h"
26 : #include "adapter_rts_common.h"
27 : #include "coll_alg_utils.h"
28 : #include "../common/src/state_guard.h"
29 : #include "alg_profiling.h"
30 : #include "preempt_port_manager.h"
31 : #include "../nslbdp/hccl_nslbdp.h"
32 : #include "dispatcher_ctx.h"
33 :
34 : using namespace std;
35 :
36 : constexpr u32 MODULE_NUM_FOUR = 4;
37 :
38 : namespace hccl
39 : {
40 : static std::mutex g_hcomInitMutex;
41 : std::mutex HcclCommunicator::linkResMapMutex_;
42 : std::unordered_map<Transport *, LinkInfo> HcclCommunicator::linkResMap_;
43 : constexpr u32 MEMORY_CAPACITY = 256 * 1024;
44 : constexpr u32 WAIT_PREPARE_SLEEP_TIME = 5000;
45 : constexpr u32 SINGLE_SERVER_NUM = 1;
46 : constexpr u32 CONN_LIMIT = 4096;
47 : constexpr u32 COMM_DEV_TYPE_DIGIT_NUM = 8;
48 : constexpr u32 TILINGDATA_BUF_SIZE = 32 * 1024; // 单位：字节
49 : constexpr u32 ALLTOALL_INFO_MATRIX_SIZE = 4;
50 : constexpr u32 AICPU_RETRY_LINKROCE_DEFAULT = 0;
51 : constexpr u32 AICPU_RETRY_LINKROCE_BACKUP = 1;
52 : constexpr u32 SINGLE_PROCESS_MIN_PORT = 1024;
53 : constexpr u32 SINGLE_PROCESS_MAX_PORT = 65535;
54 :
55 : enum TransferMemInfoIdx
56 : {
57 : TRANSFER_MEM_INFO_KEY_IDX = 0,
58 : TRANSFER_MEM_INFO_VALUE_IDX = 1,
59 : TRANSFER_MEM_INFO_RDMA_ENVELOPE_IDX = 2,
60 : TRANSFER_MEM_INFO_IDX_NUM = 3
61 : };
62 :
63 : unordered_map<std::string, std::string> ALGCFG_TO_NAME = {
64 : {"AllGather=level0:ring", "AllGatherRingFor91093Executor"},
65 : {"AllGather=level0:fullmesh", "AllGatherMeshOpbaseExecutor"},
66 : {"AllGather=level0:doublering", "AlignedAllGatherDoubleRingFor91093Executor"},
67 : {"ReduceScatter=level0:ring", "ReduceScatterRingFor91093Executor"},
68 : {"ReduceScatter=level0:fullmesh", "ReduceScatterMeshDmaEliminationExecutor"},
69 : {"ReduceScatter=level0:doublering", "AlignedReduceScatterDoubleRingFor91093Executor"},
70 : {"AllReduce=level0:ring", "AllReduceRingFor91093Executor"},
71 : {"AllReduce=level0:fullmesh", "AllReduceMeshOpbaseLoopExecutor"},
72 : {"AllReduce=level0:doublering", "AlignedAllReduceDoubleRingFor91093Executor"},
73 : {"AlltoAll=level0:fullmesh;level1:pairwise", "RunAlltoAllDirectFullmesh"},
74 : {"AlltoAll=level1:hierarchy", "RunAlltoAllAivDirect"},
75 : {"BatchWrite=level0:fullmesh", "BatchWriteBySdma"},
76 : {"BatchWrite=level1:fullmesh", "DispatchCombineFullmesh"},
77 : {"BatchWrite=level1:hierarchy", "DispatchCombineHierarchy"}};
78 :
79 : struct HcclCMDTypeHash
80 : {
81 287 : size_t operator()(HcclCMDType t) const
82 : {
83 287 : return static_cast<size_t>(t);
84 : }
85 : };
86 :
87 : unordered_map<HcclCMDType, std::string, HcclCMDTypeHash> CMDTYPE_TO_KEYWORD = {
88 : {HcclCMDType::HCCL_CMD_ALLGATHER, "AllGather"},
89 : {HcclCMDType::HCCL_CMD_REDUCE_SCATTER, "ReduceScatter"},
90 : {HcclCMDType::HCCL_CMD_ALLREDUCE, "AllReduce"},
91 : {HcclCMDType::HCCL_CMD_ALLTOALLV, "AlltoAll"},
92 : {HcclCMDType::HCCL_CMD_ALLTOALLVC, "AlltoAll"},
93 : {HcclCMDType::HCCL_CMD_ALLTOALL, "AlltoAll"},
94 : {HcclCMDType::HCCL_CMD_BATCH_WRITE, "BatchWrite"}};
95 :
96 1865 : bool HcclCommunicator::IsEnableCustom()
97 : {
98 1865 : return binCustomHandle_ != nullptr;
99 : }
100 :
101 520 : HcclResult HcclCommunicator::InitOpResPara()
102 : {
103 520 : CHK_SAFETY_FUNC_RET(
104 : memset_s(reinterpret_cast<void *>(&opResPara_), sizeof(HcclOpResParam), 0, sizeof(HcclOpResParam)));
105 520 : ListCommonInit(&opResDeviceParaPtr_->localRes.nextTagRes, &opResPara_.localRes.nextTagRes);
106 520 : opResPara_.remoteResNum = 0;
107 520 : CHK_RET(GetOpCountInfo(opResPara_.opCounterInfo));
108 520 : if (deviceType_ == DevType::DEV_TYPE_910B && GetAicpuUnfoldConfig() == false && IsOneSidedIdentifier(identifier_)) {
109 : // A2单边通信域在非aicpu展开场景下不初始化host与device侧的数据同步内存
110 0 : return HCCL_SUCCESS;
111 : }
112 520 : CHK_RET(CreateWorkSpace(sizeof(HcclOpResParam), opResDevicePara_));
113 :
114 520 : opResDeviceParaPtr_ = static_cast<HcclOpResParam *>(opResDevicePara_.ptr());
115 :
116 520 : hostDeviceLock_.reset(new (std::nothrow) PetersonLock(PetersonLock::DEFAULT_LOCK_TIMEOUT_SEC));
117 520 : CHK_SMART_PTR_NULL(hostDeviceLock_);
118 520 : CHK_RET(hostDeviceLock_->Init());
119 520 : if (aiRMAInfoMem_ == nullptr) {
120 32 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAInfo), aiRMAInfoMem_));
121 : }
122 520 : CHK_PTR_NULL(aiRMAInfoMem_);
123 520 : CHK_PTR_NULL(aiRMAInfoMem_->ptr());
124 :
125 520 : return HCCL_SUCCESS;
126 : }
127 :
128 492 : HcclResult HcclCommunicator::InitRankInfo(const RankTable_t &rankTable)
129 : {
130 492 : CHK_RET(InitTcpMode(rankTable));
131 493 : SetAttrs();
132 495 : localRank_ = attrCollector_.GetLocalRank();
133 495 : deviceLogicId_ = attrCollector_.GetDeviceLogicId();
134 : // 按通信域配置是否使用算子级重执行
135 495 : HcclIpAddress serverIp = !rankInfoList_.empty() ? rankInfoList_[0].hostIp : HcclIpAddress();
136 493 : HcclIpAddress localIp = rankInfoList_.size() > userRank_ ? rankInfoList_[userRank_].hostIp : HcclIpAddress();
137 495 : bool isAivMode = GetAivModeConfig() || GetConfigIsOnlyAivMode();
138 988 : SetRetryEnable(deviceType_, superPodNum_, serverNum_, deviceNumPerAggregation_,
139 494 : isDiffDeviceType_, isAivMode, serverIp, localIp, retryEnable_,
140 495 : commConfig_.GetConfigInterServerRetryEnable(), commConfig_.GetConfigInterSuperPodRetryEnable());
141 : // 校验A+X单机双module场景下通信能否建立
142 495 : CHK_RET(CheckSingleServerComm(rankTable.rankList));
143 : // 解析rank和port的映射信息
144 495 : CHK_RET(SetRanksPort(rankTable.rankList));
145 495 : return HCCL_SUCCESS;
146 495 : }
147 :
148 496 : HcclResult HcclCommunicator::SetRanksPort(const std::vector<RankInfo_t> &rankList)
149 : {
150 496 : bool devicePortSwitchOn = commPortConfig_.devPortSwitchOn;
151 496 : if (devicePortSwitchOn)
152 : {
153 1 : nicRanksPort_.resize(userRankSize_, HCCL_INVALID_PORT);
154 1 : vnicRanksPort_.resize(userRankSize_, HCCL_INVALID_PORT);
155 2 : for (auto &rankInfo : rankList)
156 : {
157 1 : nicRanksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT
158 1 : ? HETEROG_CCL_PORT
159 : : rankInfo.deviceInfo.port;
160 1 : vnicRanksPort_[rankInfo.rankId] = rankInfo.deviceInfo.vnicPort == HCCL_INVALID_PORT
161 1 : ? HETEROG_CCL_PORT
162 : : rankInfo.deviceInfo.vnicPort;
163 : }
164 : }
165 : else
166 : {
167 495 : nicRanksPort_.resize(userRankSize_, HCCL_INVALID_PORT);
168 1969 : for (auto &rankInfo : rankList)
169 : {
170 1477 : nicRanksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT || rankInfo.deviceInfo.port < SINGLE_PROCESS_MIN_PORT || rankInfo.deviceInfo.port > SINGLE_PROCESS_MAX_PORT
171 1477 : ? HETEROG_CCL_PORT
172 : : rankInfo.deviceInfo.port;
173 : }
174 : }
175 493 : isUseRankPort_ = ((devicePortSwitchOn && nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE) || isHaveCpuRank_)
176 986 : ? true
177 492 : : isUseRankPort_;
178 493 : HCCL_INFO("[HcclCommunicator][SetRanksPort] devicePortSwitchOn[%u], isHaveCpuRank[%u], isUseRankPort[%u], "
179 : "nicRanksPort size[%u], vnicRanksPort size[%u].",
180 : devicePortSwitchOn, isHaveCpuRank_, isUseRankPort_, nicRanksPort_.size(), vnicRanksPort_.size());
181 496 : return HCCL_SUCCESS;
182 : }
183 :
184 363 : HcclResult HcclCommunicator::InitNetResource(const RankTable_t &rankTable)
185 : {
186 363 : CHK_RET(InitPreResource(rankTable));
187 363 : CHK_RET(InitRaResource());
188 356 : return HCCL_SUCCESS;
189 : }
190 :
191 488 : HcclResult HcclCommunicator::InitDebug()
192 : {
193 488 : CHK_RET(InitProfiling());
194 488 : CHK_RET(InitATraceInfo());
195 488 : return HCCL_SUCCESS;
196 : }
197 :
198 1 : std::string HcclCommunicator::GetSupportDataType(bool needReduce)
199 : {
200 : std::vector<HcclDataType> supportList = {HCCL_DATA_TYPE_INT8, HCCL_DATA_TYPE_INT16, HCCL_DATA_TYPE_INT32,
201 2 : HCCL_DATA_TYPE_FP16, HCCL_DATA_TYPE_FP32};
202 1 : if (needReduce)
203 : {
204 1 : if (!Is310P3Common(isHaveCpuRank_, deviceType_))
205 : {
206 1 : supportList.insert(supportList.end(), {HCCL_DATA_TYPE_BFP16, HCCL_DATA_TYPE_INT64});
207 : }
208 : }
209 : else
210 : {
211 0 : supportList.insert(supportList.end(), {HCCL_DATA_TYPE_INT64, HCCL_DATA_TYPE_UINT8, HCCL_DATA_TYPE_UINT16,
212 : HCCL_DATA_TYPE_UINT32, HCCL_DATA_TYPE_UINT64, HCCL_DATA_TYPE_FP64});
213 0 : if (!Is310P3Common(isHaveCpuRank_, deviceType_))
214 : {
215 0 : supportList.push_back(HCCL_DATA_TYPE_BFP16);
216 : }
217 : }
218 :
219 1 : std::string supportInfo = "";
220 8 : for (u32 i = 0; i < supportList.size(); i++)
221 : {
222 7 : if (i != 0)
223 : {
224 6 : supportInfo += ", ";
225 : }
226 7 : supportInfo += GetDataTypeEnumStr(supportList[i]);
227 : }
228 :
229 1 : return supportInfo;
230 1 : }
231 :
232 521 : HcclResult HcclCommunicator::InitATraceInfo()
233 : {
234 : /* 申请trace资源信息 */
235 521 : std::string logInfo = "HCCL_";
236 519 : logInfo.append(to_string(SalGetTid()));
237 521 : logInfo.append("_");
238 521 : logInfo.append(to_string(deviceLogicId_));
239 521 : opBaseAtraceInfo_.reset(new (std::nothrow) HcclTraceInfo());
240 519 : CHK_PTR_NULL(opBaseAtraceInfo_);
241 521 : CHK_RET(opBaseAtraceInfo_->Init(logInfo));
242 521 : return HCCL_SUCCESS;
243 521 : }
244 :
245 33 : HcclResult HcclCommunicator::InitDebugSubGroup()
246 : {
247 33 : CHK_RET(InitATraceInfo());
248 33 : CHK_RET(InitProfiler());
249 33 : return HCCL_SUCCESS;
250 : }
251 :
252 521 : HcclResult HcclCommunicator::InitNotifyManager()
253 : {
254 521 : queueNotifyManager_.reset(new (std::nothrow) QueueNotifyManager());
255 521 : CHK_SMART_PTR_NULL(queueNotifyManager_);
256 521 : CHK_RET(queueNotifyManager_->Init());
257 521 : queueNotifyManagerRefac_.reset(new (std::nothrow) QueueNotifyManager());
258 521 : CHK_SMART_PTR_NULL(queueNotifyManagerRefac_);
259 521 : CHK_RET(queueNotifyManagerRefac_->Init());
260 :
261 521 : return HCCL_SUCCESS;
262 : }
263 :
264 41 : void TaskProfilerCallBack(void *userPtr, void *param, u32 length)
265 : {
266 41 : static_cast<ProfilerManager *>(userPtr)->TaskProfilerHandle(param, length);
267 41 : }
268 :
269 5 : void TaskAivProfilerCallBack(void *userPtr, void *param, u32 length)
270 : {
271 5 : static_cast<ProfilerManager *>(userPtr)->TaskAivProfilerHandle(param, length);
272 5 : }
273 :
274 521 : HcclResult HcclCommunicator::InitDispatcher()
275 : {
276 : // 根据设备ID创建dispatcher
277 521 : if ((deviceType_ == DevType::DEV_TYPE_910B) && GetExternalInputHcclEnableFfts())
278 : {
279 98 : CHK_PRT_CONT(GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE && !GetAicpuUnfoldConfig(),
280 : HCCL_RUN_INFO("Will use ffts mode."));
281 : }
282 : else
283 : {
284 : // 不满足ffts+特性开启条件。
285 423 : SetFftsSwitch(false);
286 : }
287 521 : CHK_RET(HcclDispatcherInit(DispatcherType::DISPATCHER_NORMAL, devicePhyId_, &dispatcher_));
288 521 : CHK_SMART_PTR_NULL(dispatcher_);
289 521 : CHK_RET(HcclSetExecTimeOut(dispatcher_, commConfig_.GetConfigExecTimeOut()));
290 :
291 521 : if (!FindDispatcherByCommId(&dispatcherCtx_, identifier_.c_str())) {
292 405 : CHK_RET(CreateDispatcherCtx(&dispatcherCtx_, devicePhyId_, identifier_.c_str()));
293 : }
294 521 : CHK_PTR_NULL(dispatcherCtx_);
295 :
296 521 : hccl::DispatcherCtx *Ctx_tmp = static_cast<DispatcherCtx *>(dispatcherCtx_);
297 521 : HCCL_INFO("[%s] RegisterLoadTaskCallBack Dispatcher = [%p], Ctx_tmp = [%p]",
298 : __func__, Ctx_tmp->GetDispatcher(), static_cast<void *>(Ctx_tmp));
299 521 : (void)RegisterLoadTaskCallBack(Ctx_tmp->GetDispatcher(), static_cast<void *>(profilerManager_.get()), TaskProfilerCallBack);
300 :
301 521 : CHK_RET(HcclDispatcherInit(DispatcherType::DISPATCHER_VIRTURAL, devicePhyId_, &vDispatcher_));
302 521 : CHK_SMART_PTR_NULL(vDispatcher_);
303 521 : CHK_RET(HcclSetExecTimeOut(vDispatcher_, commConfig_.GetConfigExecTimeOut()));
304 :
305 521 : (void)RegisterLoadTaskCallBack(dispatcher_, static_cast<void *>(profilerManager_.get()), TaskProfilerCallBack);
306 : // 此时要确保identify已经全部构造完成
307 521 : AlgWrap::GetInstance().RegisterAlgCallBack(identifier_, static_cast<void *>(profilerManager_.get()), TaskAivProfilerCallBack, deviceLogicId_);
308 521 : return HCCL_SUCCESS;
309 : }
310 :
311 521 : HcclResult HcclCommunicator::InitStreamManager()
312 : {
313 521 : opStreamManager_.reset(static_cast<OpBaseStreamManager *>(new (std::nothrow) OpBaseStreamManager));
314 519 : CHK_SMART_PTR_NULL(opStreamManager_);
315 519 : CHK_RET(StreamActiveManager::GetInstance(deviceLogicId_).Init());
316 521 : return HCCL_SUCCESS;
317 : }
318 :
319 159 : HcclResult HcclCommunicator::InitSocketManager()
320 : {
321 159 : socketManager_.reset(new (std::nothrow) HcclSocketManager(nicDeployment_, deviceLogicId_, devicePhyId_, userRank_));
322 159 : CHK_PTR_NULL(socketManager_);
323 159 : return HCCL_SUCCESS;
324 : }
325 :
326 521 : HcclResult HcclCommunicator::InitTransportManager()
327 : {
328 521 : std::vector<u32> &nicRanksPorts = groupNicRanksPort_.empty() ? nicRanksPort_ : groupNicRanksPort_;
329 521 : std::vector<u32> &vnicRanksPorts = groupVnicRanksPort_.empty() ? vnicRanksPort_ : groupVnicRanksPort_;
330 521 : transportManager_.reset(static_cast<TransportManager *>(new (std::nothrow) TransportManager(
331 521 : cclBufferManager_, socketManager_, dispatcher_, notifyPool_,
332 521 : rankInfoList_, userRank_, identifier_,
333 521 : deviceLogicId_, nicDeployment_, isHaveCpuRank_,
334 521 : isUseRankPort_, isUsedRdmaLevel0_, nicRanksPorts, vnicRanksPorts, useSuperPodMode_,
335 521 : devIpAddr_, hostIp_, localVnicIp_, netDevCtxMap_)));
336 521 : CHK_SMART_PTR_NULL(transportManager_);
337 521 : (void)transportManager_->SetPortConfig(commPortConfig_.devPortSwitchOn);
338 521 : (void)transportManager_->SetIsStandardCard(isStandardCard_);
339 :
340 521 : DispatcherCtx *ctx = static_cast<DispatcherCtx *>(dispatcherCtx_);
341 521 : CHK_PTR_NULL(ctx);
342 521 : indptOpTransportManager_.reset(static_cast<TransportManager *>(new (std::nothrow) TransportManager(
343 1042 : cclBufferManager_, socketManager_, ctx->GetDispatcher(), notifyPool_,
344 521 : rankInfoList_, userRank_, identifier_,
345 521 : deviceLogicId_, nicDeployment_, isHaveCpuRank_,
346 521 : isUseRankPort_, isUsedRdmaLevel0_, nicRanksPorts, vnicRanksPorts, useSuperPodMode_,
347 521 : devIpAddr_, hostIp_, localVnicIp_, netDevCtxMap_)));
348 521 : CHK_SMART_PTR_NULL(indptOpTransportManager_);
349 521 : (void)indptOpTransportManager_->SetPortConfig(commPortConfig_.devPortSwitchOn);
350 521 : (void)indptOpTransportManager_->SetIsStandardCard(isStandardCard_);
351 521 : return HCCL_SUCCESS;
352 : }
353 :
354 33 : HcclResult HcclCommunicator::InitHcclAlg()
355 : {
356 33 : CHK_RET(OpExeCounter::GetInstance(deviceLogicId_).InitCounter());
357 :
358 33 : notifyPool_.reset(new (std::nothrow) NotifyPool());
359 33 : CHK_SMART_PTR_NULL(notifyPool_);
360 33 : CHK_RET(notifyPool_->Init(devicePhyId_));
361 :
362 33 : callbackTask_.reset(new (std::nothrow) HcclCallbackTask(devicePhyId_, deviceLogicId_,
363 33 : dispatcher_, nicDeployment_));
364 33 : CHK_SMART_PTR_NULL(callbackTask_);
365 :
366 33 : workSpaceRes_.reset(new (std::nothrow) WorkspaceResource(devicePhyId_, deviceLogicId_));
367 33 : CHK_SMART_PTR_NULL(workSpaceRes_);
368 :
369 33 : CHK_RET(InitAlgResource());
370 32 : return HCCL_SUCCESS;
371 : }
372 :
373 521 : HcclResult HcclCommunicator::InitAlgResource()
374 : {
375 521 : HcclTopoAttr topoAttr{};
376 521 : attrCollector_.GetTopoAttr(topoAttr);
377 :
378 521 : HcclAlgoAttr algoAttr{};
379 521 : attrCollector_.GetAlgoAttr(algoAttr);
380 :
381 521 : implAlg_.reset(new (std::nothrow) HcclAlg(cclBufferManager_, dispatcher_, vDispatcher_));
382 521 : CHK_SMART_PTR_NULL(implAlg_);
383 521 : CHK_RET(implAlg_->Init(workSpaceRes_, notifyPool_, netDevCtxMap_, queueNotifyManager_,
384 : algoAttr, topoAttr, false));
385 520 : return HCCL_SUCCESS;
386 521 : }
387 :
388 528 : void HcclCommunicator::SetAttrs()
389 : {
390 528 : serverId_ = attrCollector_.GetServerId();
391 526 : superPodId_ = attrCollector_.GetSuperPodId();
392 529 : superDeviceId_ = attrCollector_.GetSuperDeviceId();
393 : // GetServerNum
394 528 : serverNum_ = attrCollector_.GetServerNum();
395 : // IsSuperPodMode
396 529 : useSuperPodMode_ = attrCollector_.GetSuperPodMode();
397 : // GetSuperPodNum
398 528 : superPodNum_ = attrCollector_.GetSuperPodNums();
399 : // GetInnerServerAverageDevice
400 529 : deviceNumPerAggregation_ = attrCollector_.GetDeviceNumPerAggregation();
401 529 : deviceNumPerServer_ = attrCollector_.GetDeviceNumPerServer();
402 529 : isHaveCpuRank_ = attrCollector_.GetHaveCpuRank();
403 : // TransformRankInfoByServerId
404 529 : servRankInfo_ = attrCollector_.GetServRankInfo();
405 : // GetModuleInfo
406 527 : isDiffDeviceModule_ = attrCollector_.GetDiffDeviceModule();
407 529 : isDiffDeviceType_ = attrCollector_.GetDiffDeviceType();
408 529 : gcdDeviceNumPerAggregation_ = attrCollector_.GetGcdDeviceNumPerAggregation();
409 529 : moduleNum_ = attrCollector_.GetModuleNum();
410 529 : multiModuleDiffDeviceNumMode_ = attrCollector_.GetMultiModuleDiffDeviceNumMode();
411 529 : multiSuperPodDiffServerNumMode_ = attrCollector_.GetMultiSuperPodDiffServerNumMode();
412 529 : multiSuperPodDiffDeviceNumMode_ = attrCollector_.GetmultiSuperPodDiffDeviceNumMode();
413 526 : isARSDoubleRing_ = attrCollector_.GetSupportARS();
414 : // 生成nicList
415 529 : nicList_ = attrCollector_.GetNicList();
416 : // InitTopoInfo
417 526 : isSingleMeshAggregation_ = attrCollector_.GetSingleMeshAggregation();
418 529 : isAllRankSamePlane_ = attrCollector_.GetAllRankSamePlane();
419 529 : isStandardCard_ = attrCollector_.GetStandardCard();
420 529 : is310PDuoCard_ = attrCollector_.Get310PDuoCard();
421 529 : isCommon310P3DUO_ = attrCollector_.GetIsCommon310P3DUO();
422 529 : hccsPortNum_ = attrCollector_.GetHccsPortNum();
423 527 : attrCollector_.GetPairLinkCounter(pairLinkCounter_);
424 526 : attrCollector_.GetPairLinkInfo(pairLinkInfo_);
425 : // SetInterModeInSuperPod
426 528 : isUsedInterHccsMode_ = attrCollector_.GetUsedInterHccsMode();
427 : // GetRankInfoList
428 528 : rankInfoList_ = attrCollector_.GetRankInfoList();
429 : // Localinfo
430 528 : devIpAddr_ = attrCollector_.GetDevIpAddr();
431 528 : devBackupIpAddr_ = attrCollector_.GetDevBackupIpAddr();
432 527 : devBackupPort_ = attrCollector_.GetBackupDevPort();
433 528 : devBackupPort_ = devBackupPort_ == HCCL_INVALID_PORT ? AICPU_RETRY_BACKUP_PORT : devBackupPort_;
434 528 : devicePhyId_ = attrCollector_.GetDevicePhyId();
435 528 : hostIp_ = attrCollector_.GetHostIp();
436 529 : hostPort_ = attrCollector_.GetHostPort();
437 :
438 528 : interServer_ = attrCollector_.GetInterServe();
439 529 : nicDeployment_ = attrCollector_.GetNicDeployment();
440 529 : }
441 :
442 75 : void HcclCommunicator::ForceProf(bool isForce) {
443 75 : ForceProfOn(dispatcher_, isForce);
444 71 : }
445 :
446 34 : HcclResult HcclCommunicator::InitRankInfoSubGroup(WorldGroupInfo &groupCommonData)
447 : {
448 34 : SetAttrs();
449 : // inline reduce 开关
450 34 : inlineReduceSwitchOn_ = attrCollector_.GetInlineReduceSwitchOn();
451 : // CalAndSetMeshAggRankSize
452 34 : meshAggregationRankSize_ = attrCollector_.GetMeshAggregationRankSize();
453 : // IsUsedRdmaLevel0AndIpInvalid
454 34 : isUsedRdmaLevel0_ = attrCollector_.GetUsedRdmaLevel0();
455 :
456 34 : CHK_RET(SetWorldGroupInfo(groupCommonData.phyIdNicInfoMap, groupCommonData.worldRankInfoList,
457 : groupCommonData.ranksPort, groupCommonData.vnicRanksPort));
458 34 : for (auto &rankInfo : worldRankInfoList_)
459 : {
460 0 : if (rankInfo.devicePhyId == HOST_DEVICE_ID)
461 : {
462 0 : isUseRankPort_ = true;
463 0 : break;
464 : }
465 : }
466 34 : CHK_RET(IsHostUseDevNic(isHostUseDevNic_));
467 : // 按通信域配置是否使用算子级重执行
468 34 : HcclIpAddress serverIp = !rankInfoList_.empty() ? rankInfoList_[0].hostIp : HcclIpAddress();
469 34 : HcclIpAddress localIp = rankInfoList_.size() > userRank_ ? rankInfoList_[userRank_].hostIp : HcclIpAddress();
470 34 : bool isAivMode = GetAivModeConfig() || GetConfigIsOnlyAivMode();
471 68 : SetRetryEnable(deviceType_, superPodNum_, serverNum_, deviceNumPerAggregation_,
472 34 : isDiffDeviceType_, isAivMode, serverIp, localIp, retryEnable_,
473 34 : commConfig_.GetConfigInterServerRetryEnable(), commConfig_.GetConfigInterSuperPodRetryEnable());
474 34 : groupNicRanksPort_.resize(rankInfoList_.size(), HCCL_INVALID_PORT);
475 34 : if (nicRanksPort_.size() != 0)
476 : {
477 0 : for (auto &rankInfo : rankInfoList_)
478 : {
479 0 : groupNicRanksPort_[rankInfo.userRank] = nicRanksPort_[rankInfo.worldRank];
480 0 : HCCL_INFO("hostIp[%s], nicIp[%s], rankInfo.userRank[%u], rankInfo.worldRank[%u], "
481 : "nic port[%u], devicePhyId[%d]",
482 : rankInfo.hostIp.GetReadableAddress(), rankInfo.nicIp[0].GetReadableAddress(),
483 : rankInfo.userRank, rankInfo.worldRank, groupNicRanksPort_[rankInfo.userRank], rankInfo.devicePhyId);
484 : }
485 : }
486 34 : commPortConfig_.devPortSwitchOn = groupCommonData.devPortSwitchOn;
487 34 : if (commPortConfig_.devPortSwitchOn)
488 : {
489 1 : groupVnicRanksPort_.resize(rankInfoList_.size(), HCCL_INVALID_PORT);
490 1 : if (vnicRanksPort_.size() != 0)
491 : {
492 1 : for (auto &rankInfo : rankInfoList_)
493 : {
494 0 : groupVnicRanksPort_[rankInfo.userRank] = vnicRanksPort_[rankInfo.worldRank];
495 0 : HCCL_INFO("hostIp[%s], nicIp[%s], rankInfo.userRank[%u], rankInfo.worldRank[%u], "
496 : "vnic port[%u], devicePhyId[%d]",
497 : rankInfo.hostIp.GetReadableAddress(), rankInfo.nicIp[0].GetReadableAddress(),
498 : rankInfo.userRank, rankInfo.worldRank, groupVnicRanksPort_[rankInfo.userRank],
499 : rankInfo.devicePhyId);
500 : }
501 : }
502 : }
503 34 : isUseRankPort_ = ((commPortConfig_.devPortSwitchOn && nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE) || isHaveCpuRank_) ? true : isUseRankPort_;
504 191 : for (auto &rank : rankInfoList_)
505 : {
506 157 : if (hostIp_ != rank.hostIp)
507 : {
508 0 : isServerInter_ = true;
509 0 : HCCL_DEBUG(" isServerInter_ is true");
510 0 : break;
511 : }
512 : }
513 34 : return HCCL_SUCCESS;
514 34 : }
515 :
516 0 : HcclResult HcclCommunicator::SetClearAivSyncBuf(bool aivClearEnable)
517 : {
518 0 : aivClearEnable_ = aivClearEnable;
519 0 : return HCCL_SUCCESS;
520 : }
521 :
522 0 : u32 HcclCommunicator::GetRankTableCrc()
523 : {
524 0 : return ranktableCrc_;
525 : }
526 :
527 0 : u32 HcclCommunicator::GetServerNum()
528 : {
529 0 : return serverNum_;
530 : }
531 :
532 0 : u32 HcclCommunicator::GetRealUserRank()
533 : {
534 0 : return realUserRank_;
535 : }
536 :
537 1 : u32 HcclCommunicator::GetModuleNum()
538 : {
539 1 : return moduleNum_;
540 : }
541 :
542 622 : bool HcclCommunicator::GetSupportHDCommunicate()
543 : {
544 622 : HCCL_INFO("%s aicpuUnfold[%d], deviceType_[%d], isHaveCpuRank_[%d]",
545 : __func__, GetAicpuUnfoldConfig(), deviceType_, isHaveCpuRank_);
546 622 : if (deviceType_ == DevType::DEV_TYPE_910B && GetAicpuUnfoldConfig() == false && IsOneSidedIdentifier(identifier_)) {
547 : // A2单边通信域在非aicpu展开场景下不初始化HDC资源
548 0 : return false;
549 : }
550 1157 : return (GetAicpuUnfoldConfig() == true) ||
551 535 : ((deviceType_ == DevType::DEV_TYPE_910_93) || (deviceType_ == DevType::DEV_TYPE_910B) ||
552 990 : Is310P3Common(isHaveCpuRank_, deviceType_));
553 : }
554 :
555 520 : HcclResult HcclCommunicator::InitHDCommunicate()
556 : {
557 520 : if (GetSupportHDCommunicate())
558 : {
559 : // 初始化aicpu进程host-device共享内存
560 257 : EXCEPTION_CATCH((kfcControlTransferH2D_ =
561 : std::make_shared<hccl::HDCommunicate>(deviceLogicId_, HCCL_HDC_TYPE_H2D, sizeof(KfcExecControl))),
562 : return HCCL_E_PTR);
563 257 : CHK_RET(kfcControlTransferH2D_->InitHost());
564 :
565 257 : EXCEPTION_CATCH((kfcStatusTransferD2H_ =
566 : std::make_shared<hccl::HDCommunicate>(deviceLogicId_, HCCL_HDC_TYPE_D2H, sizeof(KfcExecStatus))),
567 : return HCCL_E_PTR);
568 257 : CHK_RET(kfcStatusTransferD2H_->InitHost());
569 :
570 257 : if (IsEnableCustom())
571 : {
572 : // 初始化custom进程host-device共享内存
573 0 : EXCEPTION_CATCH((customControlTransferH2D_ =
574 : std::make_shared<hccl::HDCommunicate>(deviceLogicId_, HCCL_HDC_TYPE_H2D, sizeof(KfcExecControl))),
575 : return HCCL_E_PTR);
576 0 : CHK_RET(customControlTransferH2D_->InitHost());
577 :
578 0 : EXCEPTION_CATCH((customStatusTransferD2H_ =
579 : std::make_shared<hccl::HDCommunicate>(deviceLogicId_, HCCL_HDC_TYPE_D2H, sizeof(KfcExecStatus))),
580 : return HCCL_E_PTR);
581 0 : CHK_RET(customStatusTransferD2H_->InitHost());
582 : }
583 : }
584 520 : return HCCL_SUCCESS;
585 : }
586 :
587 691 : bool HcclCommunicator::IsEnableRoce()
588 : {
589 691 : return attrCollector_.IsEnableRoce();
590 : }
591 :
592 0 : u32 HcclCommunicator::LargestPowerOfTwoLessThan(const u32 localRankSize)
593 : {
594 0 : return (1 << static_cast<int>(std::floor(SalLog2(localRankSize))));
595 : }
596 :
597 0 : u32 HcclCommunicator::CalcStreamNumForReduceOrderPreservation()
598 : {
599 : // Level0RankSize条流给alltoall，剩下的流给LocalReduce使用
600 0 : u32 level0StreamNum = deviceNumPerAggregation_ - 1 + LargestPowerOfTwoLessThan(deviceNumPerAggregation_);
601 : // level1主流分给alltoall，从流给LocalReduce使用
602 0 : u32 level1StreamNum = LargestPowerOfTwoLessThan(moduleNum_);
603 : // 总流数上限：7（alltoall使用，提前的本地拷贝任务不需要并行）+ 4（LocalReduce使用）
604 0 : u32 streamNum = std::min(std::max(level0StreamNum - 1, level1StreamNum),
605 0 : DEVICE_EIGHT + DEVICE_EIGHT / FACTOR_NUM_TWO - 1);
606 :
607 0 : HCCL_INFO("[%s]level0StreamNum[%u], level1StreamNum[%u], streamNum[%u]", __func__,
608 : level0StreamNum, level1StreamNum, streamNum);
609 0 : return streamNum;
610 : }
611 :
612 86 : void HcclCommunicator::DestroyOpTransportResponse(OpCommTransport &opTransportResponse, bool aclGraphDestroyCbk)
613 : {
614 86 : std::unique_lock<std::mutex> commLock(linkResMapMutex_);
615 1319 : for (auto &levelNSubCommTransport : opTransportResponse)
616 : {
617 1505 : for (auto &singleSubCommTransport : levelNSubCommTransport)
618 : {
619 273 : for (u32 i = 0; i < singleSubCommTransport.virtualLinks.size(); i++)
620 : {
621 2 : if (singleSubCommTransport.virtualLinks[i] != nullptr &&
622 1 : (!aclGraphDestroyCbk || singleSubCommTransport.transportRequests[i].isUsedRdma))
623 : {
624 0 : linkResMap_.erase(singleSubCommTransport.virtualLinks[i].get());
625 : }
626 : }
627 769 : for (u32 i = 0; i < singleSubCommTransport.links.size(); i++)
628 : {
629 498 : if (singleSubCommTransport.transportRequests[i].isValid && singleSubCommTransport.links[i] != nullptr &&
630 1 : (!aclGraphDestroyCbk || singleSubCommTransport.transportRequests[i].isUsedRdma))
631 : {
632 0 : linkResMap_.erase(singleSubCommTransport.links[i].get());
633 : }
634 : }
635 : }
636 : }
637 86 : commLock.unlock();
638 1319 : for (auto &levelNSubCommTransport : opTransportResponse)
639 : {
640 1505 : for (auto &singleSubCommTransport : levelNSubCommTransport)
641 : {
642 273 : for (u32 i = 0; i < singleSubCommTransport.virtualLinks.size(); i++)
643 : {
644 2 : if (singleSubCommTransport.virtualLinks[i] != nullptr &&
645 1 : (!aclGraphDestroyCbk || singleSubCommTransport.transportRequests[i].isUsedRdma))
646 : {
647 0 : singleSubCommTransport.virtualLinks[i]->DeInit();
648 0 : singleSubCommTransport.virtualLinks[i] = nullptr;
649 : }
650 : }
651 769 : for (u32 i = 0; i < singleSubCommTransport.links.size(); i++)
652 : {
653 498 : if (singleSubCommTransport.transportRequests[i].isValid && singleSubCommTransport.links[i] != nullptr &&
654 1 : (!aclGraphDestroyCbk || singleSubCommTransport.transportRequests[i].isUsedRdma))
655 : {
656 0 : singleSubCommTransport.links[i]->DeInit();
657 0 : singleSubCommTransport.links[i] = nullptr;
658 : }
659 : }
660 272 : if (!aclGraphDestroyCbk) {
661 271 : singleSubCommTransport.virtualLinks.clear();
662 271 : singleSubCommTransport.links.clear();
663 : }
664 : }
665 : }
666 86 : }
667 :
668 86 : void HcclCommunicator::DestroyAlgResource(AlgResourceResponse &res, bool aclGraphDestroyCbk)
669 : {
670 86 : DestroyOpTransportResponse(res.opTransportResponse, aclGraphDestroyCbk);
671 86 : if (IsEnableBackupLink())
672 : {
673 0 : DestroyOpTransportResponse(res.opTransportResponseBackUp, aclGraphDestroyCbk);
674 0 : HCCL_INFO("[%s]finish DestroyOpTransportResponse", __func__);
675 : }
676 86 : }
677 :
678 597 : HcclResult HcclCommunicator::ReleasePreemptSocket()
679 : {
680 597 : if (commPortConfig_.devNicListen.first)
681 : {
682 2 : CHK_RET(PreemptPortManager::GetInstance(deviceLogicId_).Release(commPortConfig_.devNicListen.first));
683 2 : commPortConfig_.devNicListen.first.reset();
684 2 : if (commPortConfig_.devNicListen.second)
685 : {
686 2 : HcclNetCloseDev(commPortConfig_.devNicListen.second);
687 2 : commPortConfig_.devNicListen.second = nullptr;
688 : }
689 2 : CHK_RET(HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, devicePhyId_, deviceLogicId_));
690 1 : HCCL_INFO("[HcclCommunicator][ReleasePreemptSocket] release preempt socket of device nic success, "
691 : "comm id[%s].",
692 : identifier_.c_str());
693 : }
694 :
695 596 : if (commPortConfig_.devVnicListen.first)
696 : {
697 2 : CHK_RET(PreemptPortManager::GetInstance(deviceLogicId_).Release(commPortConfig_.devVnicListen.first));
698 2 : commPortConfig_.devVnicListen.first.reset();
699 2 : if (commPortConfig_.devVnicListen.second)
700 : {
701 1 : HcclNetCloseDev(commPortConfig_.devVnicListen.second);
702 1 : commPortConfig_.devVnicListen.second = nullptr;
703 : }
704 2 : CHK_RET(HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, devicePhyId_, deviceLogicId_));
705 2 : HCCL_INFO("[HcclCommunicator][ReleasePreemptSocket] release preempt socket of device vnic success, "
706 : "comm id[%s].",
707 : identifier_.c_str());
708 : }
709 :
710 596 : if (commPortConfig_.backupDevNicListen.first)
711 : {
712 1 : CHK_RET(hrtGetPairDevicePhyId(devicePhyId_, deviceBackUpPhyId_));
713 1 : if (hrtGetDeviceIndexByPhyId(deviceBackUpPhyId_, deviceBackUpLogicId_) == HCCL_SUCCESS)
714 : {
715 1 : CHK_RET(PreemptPortManager::GetInstance(deviceBackUpLogicId_)
716 : .Release(commPortConfig_.backupDevNicListen.first));
717 1 : commPortConfig_.backupDevNicListen.first.reset();
718 1 : if (commPortConfig_.backupDevNicListen.second)
719 : {
720 1 : HcclNetCloseDev(commPortConfig_.backupDevNicListen.second);
721 1 : commPortConfig_.backupDevNicListen.second = nullptr;
722 : }
723 1 : CHK_RET(HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, deviceBackUpPhyId_,
724 : deviceBackUpLogicId_, true));
725 1 : HCCL_INFO("[HcclCommunicator][ReleasePreemptSocket] release preempt socket of backup nic success, "
726 : "comm id[%s].",
727 : identifier_.c_str());
728 : }
729 : }
730 :
731 596 : HCCL_INFO("[HcclCommunicator][ReleasePreemptSocket] release all preempt socket success, comm id[%s].",
732 : identifier_.c_str());
733 :
734 596 : return HCCL_SUCCESS;
735 : }
736 :
737 0 : ErrorMessageReport HcclCommunicator::GetAicpuTaskException()
738 : {
739 0 : HcclResult ret = HCCL_SUCCESS;
740 0 : ErrorMessageReport errorMessage;
741 0 : if (kfcStatusTransferD2H_ != nullptr)
742 : {
743 0 : CHK_PRT_RET(isInvalidComm_,
744 : HCCL_ERROR("[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recoverying, "
745 : "this comm is invalid.", __func__, identifier_.c_str(), userRank_, deviceLogicId_), errorMessage);
746 0 : ret = kfcStatusTransferD2H_->Get(sizeof(HcclOpIdentifier) + sizeof(ExecStatusDef),
747 : sizeof(errorMessage), reinterpret_cast<uint8_t *>(&errorMessage));
748 0 : if (ret != HCCL_SUCCESS)
749 : {
750 0 : HCCL_ERROR("GetAicpuTaskException get aicpu task exception failed.ret[%u]", ret);
751 : }
752 : }
753 0 : return errorMessage;
754 : }
755 :
756 804 : HcclResult HcclCommunicator::UnRegisterBackGroundThread()
757 : {
758 804 : CHK_PRT_RET(isInvalidComm_,
759 : HCCL_ERROR("[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recoverying, "
760 : "this comm is invalid.", __func__, identifier_.c_str(), userRank_, deviceLogicId_), HCCL_E_UNAVAIL);
761 804 : CHK_RET(UnRegisterBackGroundThread(kfcControlTransferH2D_, kfcStatusTransferD2H_));
762 804 : if (IsEnableCustom())
763 : {
764 0 : CHK_RET(UnRegisterBackGroundThread(customControlTransferH2D_, customStatusTransferD2H_));
765 : }
766 804 : return HCCL_SUCCESS;
767 : }
768 :
769 804 : HcclResult HcclCommunicator::UnRegisterBackGroundThread(std::shared_ptr<HDCommunicate> &controlH2D,
770 : std::shared_ptr<HDCommunicate> &statusD2H)
771 : {
772 804 : HCCL_INFO("start to stop the backGround Thread");
773 804 : if (deviceType_ == DevType::DEV_TYPE_910 || (deviceType_ == DevType::DEV_TYPE_910B && !GetAicpuUnfoldFlag()))
774 : {
775 489 : if (GetMC2EnvFlag())
776 : {
777 0 : if (controlH2D != nullptr)
778 : {
779 0 : BackgroundCommand request = BackgroundCommand::kStop;
780 0 : CHK_RET(controlH2D->Put(sizeof(KfcCommand),
781 : sizeof(BackgroundCommand),
782 : reinterpret_cast<uint8_t *>(&request))); // 下的停止命令仅仅只修改BackGroundCommand
783 0 : auto waitStopExecCmdTimeoutMs = HcclGetCmdTimeout();
784 0 : auto waitStopExecCmdTimeout = std::chrono::milliseconds(waitStopExecCmdTimeoutMs);
785 0 : auto startTime = std::chrono::steady_clock::now();
786 : while (true)
787 : {
788 0 : if ((std::chrono::steady_clock::now() - startTime) >= waitStopExecCmdTimeout)
789 : {
790 0 : HCCL_ERROR("[NsRecovery]~HcclCommunicator is timeout [%u ms]", waitStopExecCmdTimeoutMs);
791 0 : return HCCL_E_INTERNAL;
792 : }
793 0 : KfcExecStatus status;
794 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus),
795 : reinterpret_cast<uint8_t *>(&status)));
796 0 : if (status.execStatus.backgroundStatus == BackgroundStatus::kStop)
797 : {
798 0 : break;
799 : }
800 0 : }
801 : }
802 : }
803 : }
804 :
805 804 : return HCCL_SUCCESS;
806 : }
807 :
808 804 : HcclResult HcclCommunicator::DestroyAicpuComm()
809 : {
810 804 : CHK_PRT_RET(isInvalidComm_,
811 : HCCL_ERROR("[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recoverying, "
812 : "this comm is invalid.", __func__, identifier_.c_str(), userRank_, deviceLogicId_), HCCL_E_UNAVAIL);
813 804 : CHK_RET(DestroyAicpuComm(kfcControlTransferH2D_, kfcStatusTransferD2H_));
814 804 : if (IsEnableCustom())
815 : {
816 0 : CHK_RET(DestroyAicpuComm(customControlTransferH2D_, customStatusTransferD2H_));
817 : }
818 804 : return HCCL_SUCCESS;
819 : }
820 :
821 804 : HcclResult HcclCommunicator::DestroyAicpuComm(std::shared_ptr<HDCommunicate> &controlH2D,
822 : std::shared_ptr<HDCommunicate> &statusD2H)
823 : {
824 804 : HCCL_INFO("[HcclCommunicator][%s]start to destroy the aicpu comm, group[%s].", __func__, identifier_.c_str());
825 2366 : if (deviceType_ != DevType::DEV_TYPE_910_93 &&
826 804 : !(deviceType_ == DevType::DEV_TYPE_910B && GetAicpuUnfoldFlag()) && (myRankConnectMode_ == 0))
827 : {
828 758 : HCCL_INFO("[HcclCommunicator][%s]Device type[%d] no needs to destroy the aicpu comm.", __func__,
829 : deviceType_);
830 758 : return HCCL_SUCCESS;
831 : }
832 46 : if (controlH2D == nullptr)
833 : {
834 13 : HCCL_WARNING("[HcclCommunicator][%s]controlH2D is nullptr, can not destroy the aicpu comm.",
835 : __func__);
836 13 : return HCCL_SUCCESS;
837 : }
838 33 : if (!GetMC2EnvFlag() && (getAicpuCommState_ == nullptr || !getAicpuCommState_()))
839 : {
840 33 : HCCL_INFO("[HcclCommunicator][%s]Not mc2 or aicpu environment, "\
841 : "no needs to destroy the aicpu comm.", __func__);
842 33 : return HCCL_SUCCESS;
843 : }
844 0 : KfcCommand destroyCmd = KfcCommand::kDestroyComm;
845 0 : CHK_RET(controlH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&destroyCmd)));
846 0 : KfcExecStatus status;
847 0 : auto waitCmdTimeoutMs = HcclGetCmdTimeout();
848 0 : auto waitCmdTimeout = std::chrono::milliseconds(waitCmdTimeoutMs);
849 0 : auto startTime = std::chrono::steady_clock::now();
850 : while (true)
851 : {
852 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&status)));
853 0 : if (status.execStatus.kfcStatus == KfcStatus::kDestroyComm)
854 : {
855 0 : HCCL_RUN_INFO("[HcclCommunicator][%s]ExecStatus[%d]", __func__, status.execStatus.kfcStatus);
856 0 : return HCCL_SUCCESS;
857 : }
858 : else
859 : {
860 0 : if ((std::chrono::steady_clock::now() - startTime) >= waitCmdTimeout)
861 : {
862 0 : HCCL_ERROR("[HcclCommunicator][%s]Wait DestroyExec response status timeout[%u ms] and get the "
863 : "ExecState is [%d].",
864 : __func__, waitCmdTimeoutMs, status.execStatus.kfcStatus);
865 0 : return HCCL_E_INTERNAL;
866 : }
867 : }
868 0 : }
869 :
870 : return HCCL_SUCCESS;
871 : }
872 :
873 2 : u32 HcclCommunicator::GetHostPort(s32 devicePhyId)
874 : {
875 2 : if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT)
876 : {
877 0 : return (devicePhyId + HOST_PARA_BASE_PORT);
878 : }
879 : else
880 : {
881 2 : return (devicePhyId + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM);
882 : }
883 : }
884 :
885 193 : HcclResult HcclCommunicator::setVnicIpToRankInfoList()
886 : {
887 : // 单卡场景不需要获取
888 193 : if (userRankSize_ <= 1)
889 : {
890 2 : HCCL_INFO("user rank size <= 1, ra is not needed for single device.");
891 2 : return HCCL_SUCCESS;
892 : }
893 :
894 191 : HcclIpAddress vnicIp;
895 1137 : for (auto &rankInfo : rankInfoList_)
896 : {
897 943 : if (useSuperPodMode_ && superPodId_ == rankInfo.superPodId)
898 : {
899 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
900 : devicePhyId_, DeviceIdType::DEVICE_ID_TYPE_SDID, rankInfo.superDeviceId, vnicIp));
901 0 : rankInfo.deviceVnicIp = vnicIp;
902 : }
903 943 : else if (serverId_ == rankInfo.serverId)
904 : {
905 704 : if (rankInfo.devicePhyId != HOST_DEVICE_ID)
906 : {
907 704 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
908 : devicePhyId_, DeviceIdType::DEVICE_ID_TYPE_PHY_ID, rankInfo.devicePhyId, vnicIp));
909 704 : rankInfo.deviceVnicIp = vnicIp;
910 : }
911 : }
912 : }
913 194 : return HCCL_SUCCESS;
914 194 : }
915 :
916 0 : HcclResult HcclCommunicator::SetInfoToDevice(const std::unique_ptr<PreProcessMetaInfo> &preMetaInfo,
917 : Stream &stream)
918 : {
919 0 : auto inAlltoAllvParaBuffer = cclBufferManager_.GetInAlltoAllvParaBuffer();
920 0 : auto outAlltoAllvParaBuffer = cclBufferManager_.GetOutAlltoAllvParaBuffer();
921 0 : if ((inAlltoAllvParaBuffer.ptr() == nullptr) || (outAlltoAllvParaBuffer.ptr() == nullptr))
922 : {
923 0 : CHK_RET(
924 : cclBufferManager_.InitAlltoAllvParaBuffer(preMetaInfo->inputSize, preMetaInfo->outputSize));
925 0 : inAlltoAllvParaBuffer = cclBufferManager_.GetInAlltoAllvParaBuffer();
926 0 : outAlltoAllvParaBuffer = cclBufferManager_.GetOutAlltoAllvParaBuffer();
927 : }
928 :
929 0 : auto inCCLbuffer = cclBufferManager_.GetInCCLbuffer();
930 0 : auto outCCLbuffer = cclBufferManager_.GetOutCCLbuffer();
931 0 : auto expBuffer = cclBufferManager_.GetCommExpBuffer();
932 0 : if ((inCCLbuffer.ptr() == nullptr) || (outCCLbuffer.ptr() == nullptr) || (expBuffer.ptr() == nullptr))
933 : {
934 0 : CHK_RET(CreateCommCCLbuffer());
935 0 : inCCLbuffer = cclBufferManager_.GetInCCLbuffer();
936 0 : outCCLbuffer = cclBufferManager_.GetOutCCLbuffer();
937 0 : expBuffer = cclBufferManager_.GetCommExpBuffer();
938 : }
939 :
940 0 : CHK_RET(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
941 0 : CHK_RET(hcclStreamSynchronize(stream.ptr(), commConfig_.GetConfigExecTimeOut()));
942 0 : CHK_RET(hrtMemSyncCopy(inAlltoAllvParaBuffer.ptr(), preMetaInfo->inputSize, preMetaInfo->inputData.data(),
943 : preMetaInfo->inputSize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
944 0 : return HCCL_SUCCESS;
945 0 : }
946 :
947 0 : HcclResult HcclCommunicator::GetInfoFromDevice(const std::unique_ptr<PreProcessMetaInfo> &preMetaInfo,
948 : const HcclWorkflowMode &mode, HostMem &hostCollectBuffer)
949 : {
950 0 : CHK_RET(hrtMemSyncCopy(hostCollectBuffer.ptr(), preMetaInfo->outputSize,
951 : cclBufferManager_.GetOutAlltoAllvParaBuffer().ptr(), preMetaInfo->outputSize,
952 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
953 :
954 : // 非单算子场景，中转内存使用完之后直接释放
955 0 : if (mode != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE)
956 : {
957 0 : cclBufferManager_.ReleaseAlltoAllvParaBuffer();
958 : }
959 :
960 0 : return HCCL_SUCCESS;
961 : }
962 :
963 0 : DevType HcclCommunicator::NslbGetDeviceType()
964 : {
965 0 : return deviceType_;
966 : }
967 :
968 0 : u32 HcclCommunicator::NslbGetServerNum()
969 : {
970 0 : return serverNum_;
971 : }
972 :
973 0 : HcclResult HcclCommunicator::NslbDp_CollectOperTable(HcclCMDType opType, OpParam &opParam,
974 : AlgType nslbAlgType, std::string &algName)
975 : {
976 0 : HCCL_INFO("NSLBDP-HCCL try to collect Table NSLBDP_TYPE_TBL_OPER.");
977 0 : u32 srcLocalRankId = userRank_;
978 0 : u32 rootRank = opParam.root;
979 0 : if (opParam.root == INVALID_VALUE_RANKID)
980 : {
981 0 : rootRank = 0;
982 : }
983 0 : std::string nslb_identifier = identifier_;
984 :
985 0 : HCCL_INFO("NSLBDP-SWK NslbDp_CollectOperTable nslb_identifier[%s] .", nslb_identifier.c_str());
986 0 : u32 rankSize = userRankSize_;
987 0 : u64 count = opParam.outputSize;
988 0 : if (opParam.outputSize == 0)
989 : {
990 0 : count = opParam.inputSize;
991 : }
992 :
993 0 : if (opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV)
994 : {
995 0 : u32 perDataSize = SIZE_TABLE[opParam.BatchSendRecvDataDes.sendRecvItemsPtr->dataType];
996 0 : count = opParam.BatchSendRecvDataDes.sendRecvItemsPtr->count * perDataSize;
997 : }
998 :
999 : /* NSLB 填充表2 */
1000 0 : AlgTypeLevel1 algValue = nslbAlgType.algoLevel1;
1001 0 : uint8_t nslbAlg = hcclNslbDp::GetInstance().GetNslbLevel1AlgType(algValue);
1002 0 : if (algValue != AlgTypeLevel1::ALG_LEVEL1_AHC && algValue != AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
1003 0 : if (NslbGetDeviceType() == DevType::DEV_TYPE_910_93 && NslbGetServerNum() > 1) {
1004 0 : AlgTypeLevel2 algValue2 = nslbAlgType.algoLevel2;
1005 0 : nslbAlg = hcclNslbDp::GetInstance().GetNslbLevel2AlgType(algValue2);
1006 : }
1007 : }
1008 :
1009 0 : HCCL_INFO("NSLB-HCCL algValue:[%u],nslbAlg[%u], count:[%llu].", algValue, nslbAlg, count);
1010 0 : if (hcclNslbDp::GetInstance().CheckAlgoConsistency(opType, algName) == true)
1011 : {
1012 0 : hcclNslbDp::GetInstance().GenerateOpAndAdjTable(opType, rootRank, srcLocalRankId,
1013 : nslbAlg, nslb_identifier, count, rankSize);
1014 : }
1015 0 : return HCCL_SUCCESS;
1016 0 : }
1017 :
1018 0 : HcclResult HcclCommunicator::NslbDp_CollectSendAdjTable(HcclCMDType opType, OpParam &opParam,
1019 : AlgType nslbAlgType, AdjInfo &nslbAdjInfo)
1020 : {
1021 0 : HCCL_INFO("NSLBDP-HCCL try to collect Table NSLBDP_TYPE_TBL_ADJ.");
1022 0 : u32 srcLocalRankId = userRank_;
1023 0 : u32 rootRank = opParam.root;
1024 0 : if (opParam.root == INVALID_VALUE_RANKID)
1025 : {
1026 0 : rootRank = 0;
1027 : }
1028 :
1029 0 : std::string nslb_identifier = identifier_;
1030 0 : AlgTypeLevel1 algValue = nslbAlgType.algoLevel1;
1031 0 : uint8_t nslbAlg = hcclNslbDp::GetInstance().GetNslbLevel1AlgType(algValue);
1032 0 : if (algValue != AlgTypeLevel1::ALG_LEVEL1_AHC && algValue != AlgTypeLevel1::ALG_LEVEL1_AHC_BROKE) {
1033 0 : if (NslbGetDeviceType() == DevType::DEV_TYPE_910_93 && NslbGetServerNum() > 1) {
1034 0 : AlgTypeLevel2 algValue2 = nslbAlgType.algoLevel2;
1035 0 : nslbAlg = hcclNslbDp::GetInstance().GetNslbLevel2AlgType(algValue2);
1036 : }
1037 : }
1038 0 : HCCL_INFO("NSLBDP-SHEN opType:[%u],srcLocalRankId[%u],rootRank[%u],algValue[%u],rankSize[%u]-commDesc[%s].",
1039 : opType, srcLocalRankId, rootRank, algValue, userRankSize_, nslb_identifier.c_str());
1040 0 : if (opType == HcclCMDType::HCCL_CMD_SEND || opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV)
1041 : {
1042 0 : if (nslbAdjInfo.dstRankNum == 0)
1043 : {
1044 0 : u32 ringNextRank = (srcLocalRankId + userRankSize_ / 2) % userRankSize_;
1045 0 : nslbAdjInfo.dstRankNum = 1;
1046 0 : NslbDpAdjInfo adjInfoStep = {0};
1047 0 : adjInfoStep.dstLocalRankId = ringNextRank;
1048 0 : adjInfoStep.phaseId = 1;
1049 0 : adjInfoStep.rev = 0;
1050 0 : nslbAdjInfo.nsAdjInfo.push_back(adjInfoStep);
1051 : }
1052 : }
1053 : // 填充表3
1054 0 : hcclNslbDp::GetInstance().GetAlgAdjacencyTable(opType, srcLocalRankId, rootRank, nslbAlg, nslb_identifier, nslbAdjInfo);
1055 0 : hcclNslbDp::GetInstance().SendAlgorithmInfoTable();
1056 0 : return HCCL_SUCCESS;
1057 0 : }
1058 :
1059 0 : HcclResult HcclCommunicator::updateList() const
1060 : {
1061 0 : return HCCL_SUCCESS;
1062 : }
1063 :
1064 0 : HcclResult HcclCommunicator::SetMC2EnvFlag()
1065 : {
1066 0 : isNsRecovery_ = true;
1067 0 : return HCCL_SUCCESS;
1068 : }
1069 :
1070 522 : bool HcclCommunicator::GetMC2EnvFlag()
1071 : {
1072 522 : return isNsRecovery_;
1073 : }
1074 :
1075 12 : bool HcclCommunicator::GetAicpuCommEngine()
1076 : {
1077 12 : return isAicpuCommEngine_;
1078 : }
1079 :
1080 1 : HcclResult HcclCommunicator::SetAicpuCommEngine(bool isAicpuCommEngine)
1081 : {
1082 1 : HCCL_INFO("SetAicpuCommEngine isAicpuCommEngine_[%u]", isAicpuCommEngine);
1083 1 : isAicpuCommEngine_ = isAicpuCommEngine;
1084 1 : return HCCL_SUCCESS;
1085 : }
1086 :
1087 0 : HcclResult HcclCommunicator::SetAicpuUnfoldFlag()
1088 : {
1089 0 : isAicpuUnfold_ = true;
1090 0 : return HCCL_SUCCESS;
1091 : }
1092 :
1093 344 : bool HcclCommunicator::GetAicpuUnfoldFlag()
1094 : {
1095 344 : return isAicpuUnfold_;
1096 : }
1097 :
1098 234 : HcclResult HcclCommunicator::SetStopFlag(bool value)
1099 : {
1100 234 : if (socketManager_ != nullptr)
1101 : {
1102 0 : CHK_RET(socketManager_->SetStopFlag(value));
1103 : }
1104 :
1105 234 : if (transportManager_ != nullptr)
1106 : {
1107 234 : CHK_RET(transportManager_->SetStopFlag(value));
1108 : }
1109 :
1110 234 : if (indptOpTransportManager_ != nullptr) {
1111 234 : CHK_RET(indptOpTransportManager_->SetStopFlag(value));
1112 : }
1113 :
1114 234 : for (auto &entry : resMap_)
1115 : { // map
1116 0 : for (auto &levelNSubCommTransport : entry.second.opTransportResponse)
1117 : { // vector
1118 0 : for (auto &singleSubCommTransport : levelNSubCommTransport)
1119 : { // vector
1120 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++)
1121 : { // vector
1122 0 : if (singleSubCommTransport.transportRequests[i].isValid && i < singleSubCommTransport.links.size())
1123 : {
1124 0 : auto transport = singleSubCommTransport.links[i];
1125 0 : if (transport != nullptr)
1126 : {
1127 0 : CHK_RET(transport->SetStopFlag(value));
1128 : }
1129 0 : }
1130 : }
1131 : }
1132 : }
1133 : }
1134 234 : return HCCL_SUCCESS;
1135 : }
1136 :
1137 768 : HcclResult HcclCommunicator::SetState(HcclCommState state)
1138 : {
1139 768 : state_.store(state);
1140 768 : return HCCL_SUCCESS;
1141 : }
1142 :
1143 618 : HcclCommState HcclCommunicator::GetState()
1144 : {
1145 618 : return state_.load();
1146 : }
1147 :
1148 0 : u32 HcclCommunicator ::HcclGetCmdTimeout()
1149 : {
1150 0 : return HCCL_AICPU_HOST_BASE_TIME_MS;
1151 : }
1152 :
1153 0 : HcclResult HcclCommunicator::Suspend()
1154 : {
1155 0 : CHK_PRT_RET(isInvalidComm_,
1156 : HCCL_ERROR("[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recoverying, "
1157 : "this comm is invalid.", __func__, identifier_.c_str(), userRank_, deviceLogicId_), HCCL_E_UNAVAIL);
1158 0 : return Suspend(kfcControlTransferH2D_, kfcStatusTransferD2H_);
1159 : }
1160 :
1161 0 : HcclResult HcclCommunicator::Suspend(std::shared_ptr<HDCommunicate> &controlH2D, std::shared_ptr<HDCommunicate> &statusD2H)
1162 : {
1163 0 : isSuspending = true;
1164 0 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine())
1165 : {
1166 0 : HCCL_DEBUG("[NsRecovery]MC2 OR AICPU ENVIRONMENT TO RECOVERY");
1167 0 : KfcExecControl execCommand;
1168 0 : execCommand.kfcCmd = KfcCommand::NsStopLaunch;
1169 0 : execCommand.bgCmd = BackgroundCommand::kNone;
1170 0 : execCommand.suspendingStatus = HcclComSuspendingFlag::isSuspending;
1171 0 : HCCL_RUN_INFO("[NsRecovery][SetOpExecCmd]set the suspending flag [%d] and set KfcCommand [%d], group[%s]",
1172 : execCommand.suspendingStatus, execCommand.kfcCmd, identifier_.c_str());
1173 :
1174 0 : CHK_RET(CheckSetRetryStateToWaitResume());
1175 :
1176 0 : CHK_RET(controlH2D->Put(0, sizeof(KfcExecControl), reinterpret_cast<uint8_t *>(&execCommand)));
1177 0 : KfcExecStatus opInfo;
1178 0 : auto waitStopExecCmdTimeoutMs = HcclGetCmdTimeout();
1179 0 : auto waitStopExecCmdTimeout = std::chrono::milliseconds(waitStopExecCmdTimeoutMs);
1180 0 : auto startTime = std::chrono::steady_clock::now();
1181 : while (true)
1182 : {
1183 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1184 0 : if (opInfo.execStatus.kfcStatus == KfcStatus::kStoplaunch)
1185 : {
1186 0 : HCCL_RUN_INFO("[NsRecovery]opExecState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1187 0 : return HCCL_E_SUSPENDING;
1188 : }
1189 0 : else if (opInfo.execStatus.kfcStatus == KfcStatus::kError)
1190 : {
1191 0 : return HCCL_E_INTERNAL;
1192 : }
1193 : else
1194 : {
1195 0 : if ((std::chrono::steady_clock::now() - startTime) >= waitStopExecCmdTimeout)
1196 : {
1197 0 : HCCL_ERROR("[NsRecovery]Wait suspend response status timeout[%u ms] and get the opExecState is [%u] and opId[%u].", waitStopExecCmdTimeoutMs,
1198 : opInfo.execStatus.kfcStatus, opInfo.opId.index);
1199 :
1200 0 : return HCCL_E_INTERNAL;
1201 : }
1202 0 : continue;
1203 : }
1204 0 : }
1205 : }
1206 : else
1207 : {
1208 0 : HCCL_RUN_INFO("[NsRecovery] not mc2 or aicpu ENVIRONMENT, group[%s]", identifier_.c_str());
1209 0 : return HCCL_SUCCESS;
1210 : }
1211 : }
1212 :
1213 6 : HcclResult HcclCommunicator::StopExec()
1214 : {
1215 6 : CHK_PRT_RET(isInvalidComm_,
1216 : HCCL_ERROR("[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recoverying, "
1217 : "this comm is invalid.", __func__, identifier_.c_str(), userRank_, deviceLogicId_), HCCL_E_UNAVAIL);
1218 6 : return StopExec(kfcControlTransferH2D_, kfcStatusTransferD2H_);
1219 : }
1220 :
1221 6 : HcclResult HcclCommunicator::StopExec(std::shared_ptr<HDCommunicate> &controlH2D, std::shared_ptr<HDCommunicate> &statusD2H)
1222 : {
1223 6 : isSuspending = true;
1224 6 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine())
1225 : {
1226 0 : HCCL_DEBUG("[NsRecovery]MC2 OR AICPU ENVIRONMENT TO RECOVERY");
1227 0 : KfcExecStatus opInfo;
1228 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1229 0 : HCCL_DEBUG("[NsRecovery][GetOpExecInfo] opExeState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1230 0 : if (opInfo.execStatus.kfcStatus == KfcStatus::kStoplaunch)
1231 : {
1232 0 : KfcCommand opCmd = KfcCommand::NsStopExec;
1233 0 : HCCL_RUN_INFO("[NsRecovery][SetOpExecCmd]set KfcCommand [%d], group[%s]", opCmd, identifier_.c_str());
1234 0 : CHK_RET(controlH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
1235 0 : auto waitStopExecCmdTimeoutMs = HcclGetCmdTimeout();
1236 0 : auto waitStopExecCmdTimeout = std::chrono::milliseconds(waitStopExecCmdTimeoutMs);
1237 0 : auto startTime = std::chrono::steady_clock::now();
1238 : while (true)
1239 : {
1240 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1241 0 : if (opInfo.execStatus.kfcStatus == KfcStatus::kStopExec)
1242 : {
1243 0 : HCCL_RUN_INFO("[NsRecovery]opExecState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1244 0 : return HCCL_E_SUSPENDING;
1245 : }
1246 0 : else if (opInfo.execStatus.kfcStatus == KfcStatus::kEnd)
1247 : {
1248 0 : HCCL_RUN_INFO("[NsRecovery]opExecState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1249 0 : return HCCL_SUCCESS;
1250 : }
1251 0 : else if (opInfo.execStatus.kfcStatus == KfcStatus::kError)
1252 : {
1253 0 : return HCCL_E_INTERNAL;
1254 : }
1255 : else
1256 : {
1257 0 : if ((std::chrono::steady_clock::now() - startTime) >= waitStopExecCmdTimeout)
1258 : {
1259 0 : HCCL_ERROR("[NsRecovery]Wait stopExec response status timeout[%u ms] and get the opExecState is [%u] and opId[%u].", waitStopExecCmdTimeoutMs,
1260 : opInfo.execStatus.kfcStatus, opInfo.opId.index);
1261 0 : return HCCL_E_INTERNAL;
1262 : }
1263 0 : continue;
1264 : }
1265 0 : }
1266 : }
1267 : else
1268 : {
1269 0 : return HCCL_SUCCESS;
1270 : }
1271 : }
1272 : else
1273 : {
1274 6 : HCCL_DEBUG("[NsRecovery] not mc2 or aicpu ENVIRONMENT");
1275 6 : return HCCL_SUCCESS;
1276 : }
1277 : }
1278 :
1279 4 : HcclResult HcclCommunicator::Clean()
1280 : {
1281 4 : return Clean(kfcControlTransferH2D_, kfcStatusTransferD2H_);
1282 : }
1283 :
1284 4 : HcclResult HcclCommunicator::Clean(std::shared_ptr<HDCommunicate> &controlH2D, std::shared_ptr<HDCommunicate> &statusD2H)
1285 : {
1286 4 : isSuspending = true;
1287 4 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine())
1288 : {
1289 0 : HCCL_DEBUG("[NsRecovery]MC2 OR AICPU ENVIRONMENT TO RECOVERY");
1290 0 : KfcExecStatus opInfo;
1291 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1292 0 : HCCL_DEBUG("[NsRecovery][GetOpExecInfo] opExeState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1293 0 : if (opInfo.execStatus.kfcStatus == KfcStatus::kStopExec || opInfo.execStatus.kfcStatus == KfcStatus::kEnd)
1294 : {
1295 0 : KfcCommand opCmd = KfcCommand::NsClear;
1296 0 : HCCL_RUN_INFO("[NsRecovery][SetOpExecCmd]set KfcCommand [%d], group[%s]", opCmd, identifier_.c_str());
1297 0 : CHK_RET(controlH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
1298 0 : auto waitStopExecCmdTimeoutMs = HcclGetCmdTimeout();
1299 0 : auto waitStopExecCmdTimeout = std::chrono::milliseconds(waitStopExecCmdTimeoutMs);
1300 0 : auto startTime = std::chrono::steady_clock::now();
1301 : while (true)
1302 : {
1303 0 : CHK_RET(statusD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1304 0 : if (opInfo.execStatus.kfcStatus == KfcStatus::kClear)
1305 : {
1306 0 : HCCL_RUN_INFO("[NsRecovery]opExecState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1307 0 : return HCCL_E_SUSPENDING;
1308 : }
1309 0 : else if (opInfo.execStatus.kfcStatus == KfcStatus::kEnd)
1310 : {
1311 0 : HCCL_RUN_INFO("[NsRecovery]opExecState[%d], opId[%u]", opInfo.execStatus.kfcStatus, opInfo.opId.index);
1312 0 : return HCCL_SUCCESS;
1313 : }
1314 0 : else if (opInfo.execStatus.kfcStatus == KfcStatus::kError)
1315 : {
1316 0 : return HCCL_E_INTERNAL;
1317 : }
1318 : else
1319 : {
1320 0 : if ((std::chrono::steady_clock::now() - startTime) >= waitStopExecCmdTimeout)
1321 : {
1322 0 : HCCL_ERROR("[NsRecovery]Wait clean response status timeout[%u ms] and get the opExecState is [%u] and opId[%u].", waitStopExecCmdTimeoutMs,
1323 : opInfo.execStatus.kfcStatus, opInfo.opId.index);
1324 0 : return HCCL_E_INTERNAL;
1325 : }
1326 0 : continue;
1327 : }
1328 0 : }
1329 : }
1330 : else
1331 : {
1332 0 : return HCCL_SUCCESS;
1333 : }
1334 : }
1335 : else
1336 : {
1337 4 : HCCL_DEBUG("[NsRecovery] not mc2 or aicpu ENVIRONMENT");
1338 4 : return HCCL_SUCCESS;
1339 : }
1340 : }
1341 :
1342 0 : HcclResult HcclCommunicator::CheckNotifyOrQPMaxNum(u64 &existNum, const u64 &MaxNum, const bool &isNotifyRes)
1343 : {
1344 0 : std::string resType = isNotifyRes ? "Notify" : "QP";
1345 0 : if (existNum + 1 > MaxNum)
1346 : {
1347 0 : HCCL_ERROR("[%s]%s resources are insufficient, existNum[%llu], MaxNum is [%llu]",
1348 : __func__, resType.c_str(), existNum, MaxNum);
1349 0 : return HCCL_E_INTERNAL;
1350 : }
1351 0 : HCCL_DEBUG("[%s]%s resources are sufficient, existNum[%llu], MaxNum is [%llu]",
1352 : __func__, resType.c_str(), existNum, MaxNum);
1353 0 : return HCCL_SUCCESS;
1354 0 : }
1355 :
1356 0 : HcclResult HcclCommunicator::CopyHostOpRemoteResToDeviceParam(const std::string &newTag)
1357 : {
1358 0 : HCCL_DEBUG("[%s] remote resource, tag[%s]", __func__, newTag.c_str());
1359 0 : for (u32 userRankIdx = 0; userRankIdx < AICPU_MAX_RANK_NUM; userRankIdx++)
1360 : {
1361 0 : if (opResPara_.remoteRes[userRankIdx].nextHostPtr == 0 &&
1362 0 : opResPara_.remoteRes[userRankIdx].nextDevicePtr == 0)
1363 : {
1364 0 : continue;
1365 : }
1366 : // 1、将rank公共资源，H2D到device
1367 0 : HcclRankRelationResV2 *remoteResHostPtr =
1368 0 : reinterpret_cast<HcclRankRelationResV2 *>(opResPara_.remoteRes[userRankIdx].nextHostPtr);
1369 0 : HcclRankRelationResV2 *remoteResDevicePtr =
1370 0 : reinterpret_cast<HcclRankRelationResV2 *>(opResPara_.remoteRes[userRankIdx].nextDevicePtr);
1371 0 : CHK_RET(hrtMemSyncCopy(static_cast<void *>(remoteResDevicePtr), sizeof(HcclRankRelationResV2),
1372 : static_cast<void *>(remoteResHostPtr), sizeof(HcclRankRelationResV2),
1373 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
1374 0 : HCCL_DEBUG("[%s] remote resource, tag[%s], userRankIx[%u], "
1375 : "cclinbuffer[%p], ccloutbuffer[%p], opResPara_.remoteRes[userRankIdx].nextDevicePtr[%p], "
1376 : "opResPara_.remoteRes[userRankIdx].nextHostPtr[%p]",
1377 : __func__,
1378 : newTag.c_str(), userRankIdx, remoteResHostPtr->windowsIn, remoteResHostPtr->windowsOut,
1379 : reinterpret_cast<HcclRankRelationResV2 *>(opResPara_.remoteRes[userRankIdx].nextDevicePtr),
1380 : reinterpret_cast<HcclRankRelationResV2 *>(opResPara_.remoteRes[userRankIdx].nextHostPtr));
1381 0 : CHK_RET(CopyHostListResToDeviceParam(
1382 : newTag, reinterpret_cast<ListCommon *>(&remoteResHostPtr->nextTagRes), sizeof(HccltagRemoteResV2)));
1383 : }
1384 0 : return HCCL_SUCCESS;
1385 : }
1386 :
1387 0 : HcclResult HcclCommunicator::AicpuResourceRefresh(const AlgResourceResponse &algResource, const std::string &newTag,
1388 : const HcclCMDType opType)
1389 : {
1390 0 : HCCL_INFO("[HcclCommunicator][AicpuResourceRefresh] start refresh aicpu resources newTag[%s] local rankId[%u]",
1391 : newTag.c_str(), userRank_);
1392 0 : LocalResInfoV2 *localResHostPtr = &opResPara_.localRes;
1393 0 : opResPara_.winSize = algResource.cclInputMem.size();
1394 0 : opResPara_.localWindowsIn = reinterpret_cast<u64>(algResource.cclInputMem.ptr());
1395 0 : opResPara_.localWindowsOut = reinterpret_cast<u64>(algResource.cclOutputMem.ptr());
1396 0 : CHK_RET(BuildOpLocalScratchMemResParam(algResource, newTag, localResHostPtr));
1397 0 : CHK_RET(BuildOpRemoteResParam(algResource, newTag, opType));
1398 0 : CHK_RET(BuildZeroCopyParam());
1399 0 : CHK_RET(CopyHostOpResToDeviceParam(newTag));
1400 0 : newTagResAlloced_.insert(newTag);
1401 0 : return HCCL_SUCCESS;
1402 : }
1403 :
1404 24 : HcclResult HcclCommunicator::AddGroupTagInfo(const std::string &tag, bool isAiv)
1405 : {
1406 24 : HCCL_PROFILER_ADD_GROUPRANK(identifier_, userRankSize_, userRank_);
1407 24 : if (isAiv)
1408 : {
1409 5 : HCCL_PROFILER_ADD_TAG_AIV(tag, identifier_, GetWorkflowMode());
1410 : }
1411 : else
1412 : {
1413 19 : HCCL_PROFILER_ADD_TAG(tag, identifier_, GetWorkflowMode());
1414 : }
1415 24 : return HCCL_SUCCESS;
1416 : }
1417 :
1418 21 : HcclResult HcclCommunicator::UnRegisterDfxInfo(const OpParam ¶m, const std::vector<Stream> &slaveStreams)
1419 : {
1420 21 : HCCL_PROFILER_DEL_TAG(param.tag);
1421 21 : HCCL_PROFILER_DEL_GROUPRANK(identifier_);
1422 21 : HCCL_PROFILER_DEL_STREAM_BY_STREAMID(param.stream.id());
1423 21 : HCCL_PROFILER_DEL_OPDATA(param.tag);
1424 21 : if (((GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) &&
1425 12 : !hccl::ProfilingManagerPub::GetAdditionInfoState() &&
1426 45 : !hccl::ProfilingManagerPub::GetTaskApiState()) &&
1427 12 : !param.isCapture)
1428 : {
1429 9 : return HCCL_SUCCESS;
1430 : }
1431 12 : for (auto subStream : slaveStreams)
1432 : {
1433 0 : HCCL_PROFILER_DEL_STREAM_BY_STREAMID(subStream.id());
1434 0 : }
1435 12 : return HCCL_SUCCESS;
1436 : }
1437 :
1438 : // 判断AICPU展开是否需要都走OpBase模式
1439 61 : bool HcclCommunicator::IsForceAicpuOpBaseMode(const OpParam &opParam, const HcclCMDType &opType)
1440 : {
1441 : // 目前alltoall系列算子在aicpu展开场景下仍走原有的OpBase模式
1442 : // ZeroCopy特性也强制走OpBase流程
1443 61 : if (opParam.aicpuUnfoldMode &&
1444 4 : (opType == HcclCMDType::HCCL_CMD_ALLTOALL ||
1445 4 : opType == HcclCMDType::HCCL_CMD_ALLTOALLV ||
1446 4 : opType == HcclCMDType::HCCL_CMD_ALLTOALLVC ||
1447 4 : opParam.isZeroCopy))
1448 : {
1449 0 : return true;
1450 : }
1451 :
1452 61 : return false;
1453 : }
1454 :
1455 48 : HcclResult HcclCommunicator::AllocOpBaseModeScratchMem(HcclCMDType opType, const OpParam &opParam,
1456 : AlgResourceRequest &resRequest, AlgResourceResponse &algResResponse)
1457 : {
1458 48 : if (resRequest.scratchMemSize == 0)
1459 : {
1460 43 : return HCCL_SUCCESS;
1461 : }
1462 :
1463 5 : if (opParam.isZeroCopy)
1464 : {
1465 0 : if (opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER)
1466 : {
1467 : // 零拷贝场景不需要进行scratchMem申请
1468 0 : DeviceMem tmpBuffer = DeviceMem::create(opParam.inputPtr, resRequest.scratchMemSize + CCE_REDUCE_ALIGN_SIZE);
1469 : // cce reduce地址32字节对齐，截取32字节对齐后的内存地址
1470 0 : u32 addOffset = (reinterpret_cast<uintptr_t>(tmpBuffer.ptr())) % CCE_REDUCE_ALIGN_SIZE;
1471 0 : u64 totalSize = userRankSize_ * opParam.DataDes.count * SIZE_TABLE[opParam.DataDes.dataType];
1472 0 : algResResponse.scratchMem = addOffset == 0 ? tmpBuffer.range(addOffset, totalSize) : tmpBuffer.range(CCE_REDUCE_ALIGN_SIZE - addOffset, totalSize);
1473 0 : deviceResOrigMem_.emplace_back(std::move(tmpBuffer));
1474 0 : }
1475 : else
1476 : {
1477 0 : algResResponse.scratchMem = DeviceMem::create(opParam.inputPtr, resRequest.scratchMemSize);
1478 : }
1479 : }
1480 : else
1481 : {
1482 5 : if (opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER)
1483 : {
1484 8 : DeviceMem tmpBuffer;
1485 8 : CHK_RET(DeviceMem::alloc(tmpBuffer, resRequest.scratchMemSize + CCE_REDUCE_ALIGN_SIZE));
1486 : // cce reduce地址32字节对齐，截取32字节对齐后的内存地址
1487 8 : u32 addOffset = (reinterpret_cast<uintptr_t>(tmpBuffer.ptr())) % CCE_REDUCE_ALIGN_SIZE;
1488 8 : algResResponse.scratchMem = addOffset == 0 ? tmpBuffer.range(addOffset, cclBufferManager_.GetInCCLbufferSize()) : tmpBuffer.range(CCE_REDUCE_ALIGN_SIZE - addOffset, cclBufferManager_.GetInCCLbufferSize());
1489 8 : deviceResOrigMem_.emplace_back(std::move(tmpBuffer));
1490 8 : }
1491 : else
1492 : {
1493 0 : CHK_RET(DeviceMem::alloc(algResResponse.scratchMem, resRequest.scratchMemSize));
1494 : }
1495 : }
1496 :
1497 8 : return HCCL_SUCCESS;
1498 : }
1499 :
1500 0 : HcclResult HcclCommunicator::GetAlgInfo(const std::string &algConfig, const std::string &tag,
1501 : HcclCMDType commType, std::string &algName, std::string &newTag)
1502 : {
1503 : // 查表
1504 0 : CHK_PRT_RET((ALGCFG_TO_NAME.find(algConfig) == ALGCFG_TO_NAME.end()),
1505 : HCCL_ERROR("[%s] invalid algConfig=[%s]", __func__, algConfig.c_str()),
1506 : HCCL_E_PARA);
1507 :
1508 0 : auto iter = CMDTYPE_TO_KEYWORD.find(commType);
1509 0 : CHK_PRT_RET((iter == CMDTYPE_TO_KEYWORD.end()),
1510 : HCCL_ERROR("[%s] invalid commType=[%d]", __func__, static_cast<int>(commType)),
1511 : HCCL_E_PARA);
1512 0 : CHK_PRT_RET((algConfig.find(iter->second) == algConfig.npos),
1513 : HCCL_ERROR("[%s] commType=[%d] not support algConfig=[%s]",
1514 : __func__, static_cast<int>(commType), algConfig.c_str()),
1515 : HCCL_E_PARA);
1516 :
1517 0 : algName = ALGCFG_TO_NAME[algConfig];
1518 :
1519 : TopoType topoType;
1520 0 : CHK_RET(implAlg_->GetTopoType(topoType));
1521 0 : if (topoType == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
1522 0 : if (algConfig == "AllGather=level0:doublering" || algConfig == "ReduceScatter=level0:doublering" ||
1523 0 : algConfig == "AllReduce=level0:doublering") {
1524 0 : std::size_t found = algConfig.find(":");
1525 0 : std::string algConfigTmp = algConfig.substr(0, found + 1) + "ring";
1526 0 : algName = ALGCFG_TO_NAME[algConfigTmp];
1527 0 : }
1528 : }
1529 0 : newTag = tag + algName + "_device";
1530 0 : HCCL_INFO("[%s] tag=[%s], algName=[%s], newTag=[%s], topoType=[%d]",
1531 : __func__, tag.c_str(), algName.c_str(), newTag.c_str(), topoType);
1532 0 : return HCCL_SUCCESS;
1533 : }
1534 :
1535 0 : HcclResult HcclCommunicator::CreateAndGetAiCpuNotifyWithNotifyRes(HcclSignalInfo ¬ifyInfo)
1536 : {
1537 0 : if (localAiCpuNotifyRes_.size() > 0)
1538 : {
1539 0 : CHK_RET(CreateAndGetAiCpuNotify(localAiCpuNotifyRes_[0], notifyInfo));
1540 : }
1541 : else
1542 : {
1543 0 : std::shared_ptr<LocalNotify> localNotify = {nullptr};
1544 0 : CHK_RET(CreateAndGetAiCpuNotify(localNotify, notifyInfo));
1545 0 : localAiCpuNotifyRes_.push_back(localNotify);
1546 0 : }
1547 0 : return HCCL_SUCCESS;
1548 : }
1549 :
1550 0 : HcclResult HcclCommunicator::SetDynamicTilingDataAlltoall(const OpParam &opParam, HostMem &dynamicDataMem)
1551 : {
1552 : struct OpTilingAllToAllDataDes *a2ADataPtr =
1553 0 : reinterpret_cast<struct OpTilingAllToAllDataDes *>(dynamicDataMem.ptr());
1554 0 : a2ADataPtr->sendType = static_cast<u8>(opParam.All2AllDataDes.sendType);
1555 0 : a2ADataPtr->recvType = static_cast<u8>(opParam.All2AllDataDes.recvType);
1556 0 : a2ADataPtr->sendCount = opParam.All2AllDataDes.sendCount;
1557 0 : return HCCL_SUCCESS;
1558 : }
1559 :
1560 0 : HcclResult HcclCommunicator::SetDynamicTilingDataAlltoallv(const OpParam &opParam, HostMem &dynamicDataMem, const std::string &algName)
1561 : {
1562 : struct OpTilingAlltoallvDataDes *alltoallvDataPtr =
1563 0 : reinterpret_cast<struct OpTilingAlltoallvDataDes *>(dynamicDataMem.ptr());
1564 0 : alltoallvDataPtr->sendType = static_cast<u8>(opParam.All2AllDataDes.sendType);
1565 0 : alltoallvDataPtr->recvType = static_cast<u8>(opParam.All2AllDataDes.recvType);
1566 0 : u32 rankSize = GetRankSize();
1567 0 : u64 *sendCountsPtr = static_cast<u64 *>(alltoallvDataPtr->sendRecvInfos);
1568 0 : u64 *recvCountsPtr = sendCountsPtr + rankSize;
1569 0 : u64 *sdisplsPtr = recvCountsPtr + rankSize;
1570 0 : u64 *rdisplsPtr = sdisplsPtr + rankSize;
1571 0 : for (u32 i = 0; i < rankSize; i++)
1572 : {
1573 0 : CHK_PTR_NULL(static_cast<const u64 *>(opParam.All2AllDataDes.sendCounts) + i);
1574 0 : sendCountsPtr[i] = *(static_cast<const u64 *>(opParam.All2AllDataDes.sendCounts) + i);
1575 0 : CHK_PTR_NULL(static_cast<const u64 *>(opParam.All2AllDataDes.recvCounts) + i);
1576 0 : recvCountsPtr[i] = *(static_cast<const u64 *>(opParam.All2AllDataDes.recvCounts) + i);
1577 0 : CHK_PTR_NULL(static_cast<const u64 *>(opParam.All2AllDataDes.sdispls) + i);
1578 0 : sdisplsPtr[i] = *(static_cast<const u64 *>(opParam.All2AllDataDes.sdispls) + i);
1579 0 : CHK_PTR_NULL(static_cast<const u64 *>(opParam.All2AllDataDes.rdispls) + i);
1580 0 : rdisplsPtr[i] = *(static_cast<const u64 *>(opParam.All2AllDataDes.rdispls) + i);
1581 0 : HCCL_DEBUG("[SetDynamicTilingDataAlltoallv] sendCounts[%llu], recvCounts[%llu], sdispls[%llu], rdispls[%llu]",
1582 : sendCountsPtr[i], recvCountsPtr[i], sdisplsPtr[i], rdisplsPtr[i]);
1583 : }
1584 :
1585 0 : if (algName == "RunAlltoAllVTwoLevelPipeline")
1586 : {
1587 0 : u64 *sendRecvInfoPtr = rdisplsPtr + rankSize;
1588 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sendRecvInfoPtr, hostCollectBuffer_.size(), hostCollectBuffer_.ptr(), hostCollectBuffer_.size()));
1589 : }
1590 :
1591 0 : HCCL_DEBUG("[SetDynamicTilingDataAlltoallv] set dynamic tiling data for AllToAllV success, alltoallvDataPtr[%p]", alltoallvDataPtr);
1592 0 : return HCCL_SUCCESS;
1593 : }
1594 :
1595 0 : HcclResult HcclCommunicator::SetDynamicTilingDataAlltoallvc(const OpParam &opParam, HostMem &dynamicDataMem)
1596 : {
1597 : struct OpTilingAlltoallvcDataDes *a2ADataPtr =
1598 0 : reinterpret_cast<struct OpTilingAlltoallvcDataDes *>(dynamicDataMem.ptr());
1599 0 : a2ADataPtr->sendType = static_cast<u8>(opParam.All2AllDataDes.sendType);
1600 0 : a2ADataPtr->recvType = static_cast<u8>(opParam.All2AllDataDes.recvType);
1601 0 : u32 rankSize = GetRankSize();
1602 0 : for (u64 i = 0; i < rankSize * rankSize; i++)
1603 : {
1604 0 : a2ADataPtr->sendCountMatrix[i] = *(static_cast<const u64 *>(opParam.All2AllDataDes.sendCountMatrix) + i);
1605 : }
1606 0 : return HCCL_SUCCESS;
1607 : }
1608 :
1609 0 : u64 HcclCommunicator::CalcOpTilingVDataDesVDataLen(const u32 rankSize) const
1610 : {
1611 0 : const u32 vFactor = 2; // counts和displs 2个变长数组
1612 0 : return vFactor * rankSize * sizeof(u64);
1613 : }
1614 :
1615 0 : HcclResult HcclCommunicator::SetDynamicTilingDataV(const OpParam &opParam, HostMem &dynamicDataMem)
1616 : {
1617 0 : CHK_PTR_NULL(static_cast<const u64 *>(opParam.VDataDes.counts));
1618 0 : CHK_PTR_NULL(static_cast<const u64 *>(opParam.VDataDes.displs));
1619 :
1620 0 : const u32 rankSize = GetRankSize();
1621 0 : struct OpTilingVDataDes *vDataPtr = reinterpret_cast<struct OpTilingVDataDes *>(dynamicDataMem.ptr());
1622 0 : vDataPtr->dataType = static_cast<u8>(opParam.VDataDes.dataType);
1623 0 : vDataPtr->vDataLen = CalcOpTilingVDataDesVDataLen(rankSize);
1624 :
1625 0 : u64 *countsPtr = static_cast<u64 *>(vDataPtr->vData);
1626 0 : u64 *displsPtr = countsPtr + rankSize;
1627 0 : for (u32 i = 0; i < rankSize; ++i)
1628 : {
1629 0 : countsPtr[i] = *(static_cast<const u64 *>(opParam.VDataDes.counts) + i);
1630 0 : displsPtr[i] = *(static_cast<const u64 *>(opParam.VDataDes.displs) + i);
1631 0 : HCCL_DEBUG("[SetDynamicTilingDataV][%u] counts[%llu], displs[%llu]", i, countsPtr[i], displsPtr[i]);
1632 : }
1633 :
1634 0 : HCCL_DEBUG("[SetDynamicTilingDataV] set dynamic tiling data success, vDataPtr[%p]", vDataPtr);
1635 0 : return HCCL_SUCCESS;
1636 : }
1637 :
1638 0 : u64 HcclCommunicator::CalcOpTilingDynamicDataSize(
1639 : const OpParam &opParam, const HcclCMDType &opType, const u32 &rankSize, const std::string &algName)
1640 : {
1641 0 : u64 dynamicDataSize = 0ULL;
1642 0 : if (opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV)
1643 : {
1644 0 : dynamicDataSize = sizeof(struct OpTilingBatchSendRecvDataDes) +
1645 0 : opParam.BatchSendRecvDataDes.itemNum * sizeof(HcclSendRecvItem) +
1646 0 : userRankSize_ * sizeof(u8);
1647 : }
1648 0 : else if (opType == HcclCMDType::HCCL_CMD_ALLTOALL)
1649 : {
1650 0 : dynamicDataSize = sizeof(struct OpTilingAllToAllDataDes);
1651 : }
1652 0 : else if (opType == HcclCMDType::HCCL_CMD_ALLTOALLV)
1653 : {
1654 0 : dynamicDataSize = sizeof(struct OpTilingAlltoallvDataDes) + rankSize * ALLTOALL_INFO_MATRIX_SIZE * sizeof(u64);
1655 0 : if (algName == "RunAlltoAllVTwoLevelPipeline")
1656 : {
1657 0 : dynamicDataSize += hostCollectBuffer_.size();
1658 : }
1659 : }
1660 0 : else if (opType == HcclCMDType::HCCL_CMD_ALLTOALLVC)
1661 : {
1662 0 : dynamicDataSize = sizeof(struct OpTilingAlltoallvcDataDes) + rankSize * rankSize * sizeof(u64);
1663 : }
1664 0 : else if (opType == HcclCMDType::HCCL_CMD_ALLGATHER_V || opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER_V)
1665 : {
1666 0 : dynamicDataSize = sizeof(struct OpTilingVDataDes) + CalcOpTilingVDataDesVDataLen(rankSize);
1667 : }
1668 : else
1669 : {
1670 0 : dynamicDataSize = sizeof(struct OpTilingDataDes);
1671 : }
1672 0 : return dynamicDataSize;
1673 : }
1674 :
1675 0 : HcclResult HcclCommunicator::AicpuInitOpTilingDataFromOpParam(const OpParam &opParam, const HcclCMDType &opType,
1676 : struct OpTilingData *opTilingData)
1677 : {
1678 0 : opTilingData->workflowMode = (IsForceAicpuOpBaseMode(opParam, opType) && !opParam.isZeroCopy) ? static_cast<u8>(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) : static_cast<u8>(GetWorkflowMode());
1679 0 : opTilingData->inputPtr = reinterpret_cast<u64>(opParam.inputPtr);
1680 0 : opTilingData->outputPtr = reinterpret_cast<u64>(opParam.outputPtr);
1681 0 : opTilingData->reduceType = static_cast<u8>(opParam.reduceType);
1682 0 : opTilingData->syncMode = static_cast<u8>(opParam.syncMode);
1683 0 : opTilingData->root = opParam.root;
1684 0 : opTilingData->dstRank = opParam.dstRank;
1685 0 : opTilingData->srcRank = opParam.srcRank;
1686 0 : opTilingData->opType = static_cast<u8>(opType);
1687 0 : opTilingData->inplaceSupportRetry = static_cast<u8>(inplaceSupportRetry_);
1688 0 : opTilingData->retryEnable = static_cast<u8>(retryEnable_);
1689 0 : opTilingData->inPlaceSupportRetryStatus = static_cast<u8>(inPlaceSupportRetryStatus_);
1690 0 : opTilingData->isInplacePreSync = static_cast<u8>(isInplacePreSync_);
1691 0 : opTilingData->isPostSync = static_cast<u8>(isPostSync_);
1692 0 : opTilingData->userStreamId = opParam.stream.id();
1693 0 : opTilingData->inputSymWindow = reinterpret_cast<u64>(opParam.inputSymWindow);
1694 0 : opTilingData->inputOffset = opParam.inputOffset;
1695 0 : opTilingData->outputSymWindow = reinterpret_cast<u64>(opParam.outputSymWindow);
1696 0 : opTilingData->outputOffset = opParam.outputOffset;
1697 0 : return HCCL_SUCCESS;
1698 : }
1699 :
1700 1 : HcclResult HcclCommunicator::KernelLaunchChooseAicpuOrCustom(void *inputPtr, void *outputPtr, const rtStream_t stm, u64 addr,
1701 : void *tilingDataPtr, u32 tilingDataSize, const std::string &kernelName, HcclWorkflowMode mode, const std::string &tag, bool isCustom)
1702 : {
1703 1 : return AicpuUnfoldKernelLaunchV2(inputPtr, outputPtr, stm, addr, tilingDataPtr, tilingDataSize,
1704 1 : kernelName, mode, tag, isCustom);
1705 : }
1706 :
1707 633 : HcclResult HcclCommunicator::SaveTraceInfo(std::string &logInfo)
1708 : {
1709 633 : opBaseAtraceInfo_->SaveTraceInfo(logInfo, AtraceOption::Opbasekey);
1710 633 : return HCCL_SUCCESS;
1711 : }
1712 :
1713 0 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>> HcclCommunicator::GetPhyIdNicInfo()
1714 : {
1715 0 : return rankDevicePhyIdNicInfoMap_;
1716 : }
1717 :
1718 1 : vector<u32> HcclCommunicator::GetRanksPort()
1719 : {
1720 1 : return nicRanksPort_;
1721 : }
1722 :
1723 0 : vector<RankInfo> HcclCommunicator::GetRanksList()
1724 : {
1725 0 : return rankInfoList_;
1726 : }
1727 :
1728 291 : std::string HcclCommunicator::GetUniqueId(void)
1729 : {
1730 : static std::atomic<u32> idCounter(0);
1731 :
1732 291 : std::string uniqueId("");
1733 291 : uniqueId += std::to_string(SalGetPid());
1734 291 : uniqueId += '-';
1735 291 : uniqueId += std::to_string(idCounter.fetch_add(1));
1736 291 : uniqueId += '-';
1737 291 : uniqueId += std::to_string(SalGetSysTime());
1738 :
1739 291 : return uniqueId;
1740 0 : }
1741 :
1742 1 : u8 HcclCommunicator::GetDeterministicConfig() const
1743 : {
1744 1 : CHK_SMART_PTR_NULL(implAlg_);
1745 1 : return implAlg_->GetDeterministicConfig();
1746 : }
1747 :
1748 234 : HcclResult HcclCommunicator::SetDeterministicConfig(const u8 deterministic)
1749 : {
1750 234 : CHK_SMART_PTR_NULL(implAlg_);
1751 234 : CHK_RET(implAlg_->SetDeterministicConfig(deterministic));
1752 234 : return HCCL_SUCCESS;
1753 : }
1754 :
1755 0 : HcclResult HcclCommunicator::MigrateLinkToStopOrResume(LINK &link, bool isStop)
1756 : {
1757 0 : if (isStop)
1758 : {
1759 0 : return link->Stop();
1760 : }
1761 0 : return link->Resume();
1762 : }
1763 :
1764 0 : HcclResult HcclCommunicator::MigrateLinkVectorToStopOrResume(const std::vector<LINK> &links, bool isStop)
1765 : {
1766 0 : for (auto it : links)
1767 : {
1768 0 : if (it)
1769 : {
1770 0 : CHK_RET(MigrateLinkToStopOrResume(it, isStop));
1771 : }
1772 0 : }
1773 0 : return HCCL_SUCCESS;
1774 : }
1775 :
1776 0 : HcclResult HcclCommunicator::TraverseLinkVector(std::vector<std::unique_ptr<CommBase>> &commBaseVector, bool isStop)
1777 : {
1778 0 : for (unsigned int i = 0; i < commBaseVector.size(); i++)
1779 : {
1780 0 : auto commBase = commBaseVector[i].get();
1781 0 : if (commBase == nullptr)
1782 : {
1783 0 : continue;
1784 : }
1785 0 : const std::vector<LINK> &ret = commBase->TransportInfo();
1786 0 : CHK_RET(MigrateLinkVectorToStopOrResume(ret, isStop));
1787 : }
1788 0 : return HCCL_SUCCESS;
1789 : }
1790 :
1791 0 : HcclResult HcclCommunicator::TraverseSingleSubCommTransport(SingleSubCommTransport &commTransport, bool isStop)
1792 : {
1793 0 : for (unsigned int i = 0; i < commTransport.transportRequests.size(); i++)
1794 : {
1795 0 : if (!commTransport.transportRequests[i].isValid)
1796 : {
1797 0 : continue;
1798 : }
1799 0 : if (commTransport.links[i] == nullptr)
1800 : {
1801 0 : continue;
1802 : }
1803 :
1804 0 : if (isStop)
1805 : {
1806 0 : CHK_RET(commTransport.links[i]->Stop());
1807 : }
1808 : else
1809 : {
1810 0 : CHK_RET(commTransport.links[i]->Resume());
1811 : }
1812 : }
1813 0 : return HCCL_SUCCESS;
1814 : }
1815 :
1816 0 : HcclResult HcclCommunicator::TraverseLevelNSubCommTransport(LevelNSubCommTransport &levelNSubCommTransport, bool isStop)
1817 : {
1818 0 : for (unsigned int jj = 0; jj < levelNSubCommTransport.size(); jj++)
1819 : {
1820 0 : CHK_RET(TraverseSingleSubCommTransport(levelNSubCommTransport[jj], isStop));
1821 : }
1822 0 : return HCCL_SUCCESS;
1823 : }
1824 :
1825 0 : HcclResult HcclCommunicator::TraverseOpCommTransport(OpCommTransport &opCommTransport, bool isStop)
1826 : {
1827 0 : for (unsigned int ii = 0; ii < opCommTransport.size(); ii++)
1828 : {
1829 0 : CHK_RET(TraverseLevelNSubCommTransport(opCommTransport[ii], isStop));
1830 : }
1831 0 : return HCCL_SUCCESS;
1832 : }
1833 :
1834 7 : HcclResult HcclCommunicator::TraverseAlgResourceResponse(bool isStop)
1835 : {
1836 7 : for (auto &it : resMap_)
1837 : {
1838 0 : CHK_RET(TraverseOpCommTransport(it.second.opTransportResponse, isStop));
1839 : }
1840 7 : return HCCL_SUCCESS;
1841 : }
1842 :
1843 0 : HcclResult HcclCommunicator::ResetNotify()
1844 : {
1845 0 : CHK_SMART_PTR_NULL(notifyPool_);
1846 0 : CHK_SMART_PTR_NULL(queueNotifyManagerRefac_);
1847 0 : notifyPool_->ResetNotify();
1848 0 : queueNotifyManagerRefac_->ResetNotify();
1849 0 : return HCCL_SUCCESS;
1850 : }
1851 :
1852 0 : HcclResult HcclCommunicator::ResetNotifyForDestRank(s64 destRank)
1853 : {
1854 0 : CHK_SMART_PTR_NULL(notifyPool_);
1855 0 : CHK_SMART_PTR_NULL(queueNotifyManagerRefac_);
1856 0 : notifyPool_->ResetNotifyForDestRank(destRank);
1857 0 : return HCCL_SUCCESS;
1858 : }
1859 :
1860 79 : void HcclCommunicator::InsertNewTagToTagMap(std::string &newTag, std::string &tag)
1861 : {
1862 79 : const auto &mapIt = newTagToTagMap_.find(newTag);
1863 78 : if (mapIt == newTagToTagMap_.end())
1864 : {
1865 78 : newTagToTagMap_.insert({newTag, tag});
1866 : }
1867 : else
1868 : {
1869 1 : mapIt->second = tag;
1870 : }
1871 158 : return;
1872 : }
1873 :
1874 0 : HcclResult HcclCommunicator::GetTagFromNewTag(const std::string &newTag, std::string &tag)
1875 : {
1876 0 : const auto &mapIt = newTagToTagMap_.find(newTag);
1877 0 : if (mapIt == newTagToTagMap_.end())
1878 : {
1879 0 : HCCL_ERROR("[OpRetry]newTag[%s] is not in newTagToTagMap_", newTag.c_str());
1880 0 : return HCCL_E_INTERNAL;
1881 : }
1882 : else
1883 : {
1884 0 : tag = mapIt->second;
1885 : }
1886 0 : return HCCL_SUCCESS;
1887 : }
1888 :
1889 0 : HcclResult HcclCommunicator::SetSignalTransport(SingleSubCommTransport &singleSubCommTransport,
1890 : u32 linkIdx, bool statusStop)
1891 : {
1892 0 : RankId loc = singleSubCommTransport.transportRequests[linkIdx].localUserRank;
1893 0 : RankId rmt = singleSubCommTransport.transportRequests[linkIdx].remoteUserRank;
1894 0 : if (statusStop)
1895 : {
1896 0 : if (singleSubCommTransport.links[linkIdx] && singleSubCommTransport.links[linkIdx]->GetLinkType() == LinkType::LINK_ROCE)
1897 : {
1898 0 : CHK_RET(singleSubCommTransport.links[linkIdx]->Stop());
1899 0 : singleSubCommTransport.status[linkIdx] = TransportStatus::STOP;
1900 0 : HCCL_INFO("[SetTransportStatus]set transport status to stop, loc[%u], rmt[%u]", loc, rmt);
1901 : }
1902 : }
1903 : else
1904 : {
1905 0 : if (singleSubCommTransport.links[linkIdx] && singleSubCommTransport.status[linkIdx] == TransportStatus::STOP)
1906 : {
1907 0 : HCCL_INFO("[SetTransportStatus]set transport status to resume, loc[%u], rmt[%u]", loc, rmt);
1908 0 : CHK_RET(singleSubCommTransport.links[linkIdx]->DeInit());
1909 0 : singleSubCommTransport.links[linkIdx] = nullptr; // 赋值为nullptr, 供后面重新建链
1910 0 : singleSubCommTransport.status[linkIdx] = TransportStatus::INIT;
1911 : }
1912 : }
1913 0 : return HCCL_SUCCESS;
1914 : }
1915 :
1916 0 : HcclResult HcclCommunicator::SetBsrTransportStatusImpl(OpCommTransport &opCommTransport, bool statusStop,
1917 : const HcclOpIdentifier &opId, u32 remoteRank)
1918 : {
1919 0 : u32 commIndex = 0;
1920 0 : if ((userRank_ == opId.detRank && remoteRank > userRank_) ||
1921 0 : (userRank_ == opId.srcRank && remoteRank < userRank_))
1922 : {
1923 0 : commIndex = COMM_INDEX_0;
1924 : }
1925 : else
1926 : {
1927 0 : commIndex = COMM_INDEX_1;
1928 : }
1929 0 : CHK_PRT_RET(commIndex >= opCommTransport[COMM_COMBINE_ORDER].size(),
1930 : HCCL_ERROR("[SetBsrTransportStatusImpl] batchsendrecv op commIndex[%u] is larger than "
1931 : "opTransportResponse size[%zu]",
1932 : remoteRank, opCommTransport[COMM_COMBINE_ORDER].size()),
1933 : HCCL_E_PARA);
1934 : SingleSubCommTransport &commCombined =
1935 0 : const_cast<SingleSubCommTransport &>(opCommTransport[COMM_COMBINE_ORDER][commIndex]);
1936 0 : u32 Rank = commCombined.userRank2subCommRank[remoteRank];
1937 0 : CHK_PRT_RET(Rank >= commCombined.links.size(),
1938 : HCCL_ERROR("[SetBsrTransportStatusImpl] batchsendrecv op remoteRank[%u], get Rank[%u],"
1939 : "the size of combinedComm links is [%zu]",
1940 : remoteRank, Rank, commCombined.links.size()),
1941 : HCCL_E_PARA);
1942 0 : CHK_SMART_PTR_NULL(commCombined.links[Rank]);
1943 :
1944 0 : RankId loc = commCombined.transportRequests[Rank].localUserRank;
1945 0 : RankId rmt = commCombined.transportRequests[Rank].remoteUserRank;
1946 0 : if (!commCombined.transportRequests[Rank].isValid)
1947 : {
1948 0 : return HCCL_SUCCESS;
1949 : }
1950 0 : if (statusStop)
1951 : {
1952 0 : if (commCombined.links[Rank]->GetLinkType() == LinkType::LINK_ROCE)
1953 : {
1954 0 : CHK_RET(commCombined.links[Rank]->Stop());
1955 0 : commCombined.status[Rank] = TransportStatus::STOP;
1956 0 : HCCL_INFO("[SetBsrTransportStatusImpl]set bsr transport status to stop, comindex[%u] loc[%u], rmt[%u]",
1957 : commIndex, loc, rmt);
1958 : }
1959 : }
1960 : else
1961 : {
1962 0 : if (commCombined.status[Rank] == TransportStatus::STOP)
1963 : {
1964 0 : HCCL_INFO("[SetBsrTransportStatusImpl]set bsr transport status to resume, comindex[%u] loc[%u], rmt[%u]",
1965 : commIndex, loc, rmt);
1966 0 : CHK_RET(commCombined.links[Rank]->DeInit());
1967 0 : commCombined.links[Rank] = nullptr; // 赋值为nullptr, 供后面重新建链
1968 0 : commCombined.status[Rank] = TransportStatus::INIT;
1969 : }
1970 : }
1971 0 : return HCCL_SUCCESS;
1972 : }
1973 :
1974 0 : HcclResult HcclCommunicator::SetBsrTransportStatusImplforchange(OpCommTransport &opCommTransport,
1975 : const HcclOpIdentifier &opId, u32 remoteRank, const std::map<u32, bool> &remoteRankPortMap, bool isUseDefault,
1976 : const std::map<u32, bool> &isChangeLinkMap, bool isCurTag)
1977 : {
1978 0 : bool isPortSatisfy = (remoteRankPortMap.find(remoteRank) != remoteRankPortMap.end() &&
1979 0 : remoteRankPortMap.find(remoteRank)->second == isUseDefault);
1980 0 : bool isChangeLink = (isChangeLinkMap.find(remoteRank) != isChangeLinkMap.end() &&
1981 0 : isChangeLinkMap.find(remoteRank)->second);
1982 0 : HCCL_INFO("[SetBsrTransportStatusImplforchange]remoteRank[%u], isUseDefault[%d], "
1983 : "isPortSatisfy[%d], isChangeLink[%d], isCurTag[%d]",
1984 : remoteRank, isUseDefault, isPortSatisfy, isChangeLink, isCurTag);
1985 0 : if (!isPortSatisfy || !(isChangeLink || isCurTag))
1986 : {
1987 0 : return HCCL_SUCCESS;
1988 : }
1989 :
1990 0 : u32 commIndex = 0;
1991 0 : if ((userRank_ == opId.detRank && remoteRank > userRank_) ||
1992 0 : (userRank_ == opId.srcRank && remoteRank < userRank_))
1993 : {
1994 0 : commIndex = COMM_INDEX_0;
1995 : }
1996 : else
1997 : {
1998 0 : commIndex = COMM_INDEX_1;
1999 : }
2000 0 : CHK_PRT_RET(commIndex >= opCommTransport[COMM_COMBINE_ORDER].size(),
2001 : HCCL_ERROR("[SetBsrTransportStatusImplforchange] batchsendrecv op commIndex[%u] is larger than "
2002 : "opTransportResponse size[%zu]",
2003 : commIndex, opCommTransport[COMM_COMBINE_ORDER].size()),
2004 : HCCL_E_PARA);
2005 : SingleSubCommTransport &commCombined =
2006 0 : static_cast<SingleSubCommTransport &>(opCommTransport[COMM_COMBINE_ORDER][commIndex]);
2007 0 : u32 rank = commCombined.userRank2subCommRank[remoteRank];
2008 0 : CHK_PRT_RET(rank >= commCombined.links.size(),
2009 : HCCL_ERROR("[SetBsrTransportStatusImplforchange] batchsendrecv op remoteRank[%u], get Rank[%u],"
2010 : "the size of combinedComm links is [%zu]",
2011 : remoteRank, rank, commCombined.links.size()),
2012 : HCCL_E_PARA);
2013 0 : CHK_SMART_PTR_NULL(commCombined.links[rank]);
2014 :
2015 0 : RankId loc = commCombined.transportRequests[rank].localUserRank;
2016 0 : RankId rmt = commCombined.transportRequests[rank].remoteUserRank;
2017 0 : if (!commCombined.transportRequests[rank].isValid)
2018 : {
2019 0 : return HCCL_SUCCESS;
2020 : }
2021 :
2022 0 : if (commCombined.status[rank] == TransportStatus::STOP)
2023 : {
2024 0 : HCCL_INFO("[SetBsrTransportStatusImplforchange]set bsr transport status to resume, comindex[%u] loc[%u], rmt[%u]",
2025 : commIndex, loc, rmt);
2026 0 : CHK_RET(commCombined.links[rank]->DeInit());
2027 0 : commCombined.links[rank] = nullptr; // 赋值为nullptr, 供后面重新建链
2028 0 : commCombined.status[rank] = TransportStatus::INIT;
2029 : }
2030 0 : return HCCL_SUCCESS;
2031 : }
2032 :
2033 0 : HcclResult HcclCommunicator::SetTransportStatusImpl(OpCommTransport &opCommTransport, bool statusStop,
2034 : const HcclOpIdentifier &opId, u32 remoteRank, const std::map<u32, bool> &remoteRankPortMap, bool isUseDefault)
2035 : {
2036 0 : bool isSendRecv = opId.isSendRecv;
2037 :
2038 : // stop阶段及原地重执行的resume阶段
2039 : // bsr判断当前故障的send、recv是否remoterank是否相同的情况，如果是相同只操作故障op，如果不同都操作
2040 0 : u32 sendRemoteRank = userRank_ == opId.bsrInfo[HCCL_SEND].detRank ? opId.bsrInfo[HCCL_SEND].srcRank : opId.bsrInfo[HCCL_SEND].detRank;
2041 0 : u32 recvRemoteRank = userRank_ == opId.bsrInfo[HCCL_RECV].detRank ? opId.bsrInfo[HCCL_RECV].srcRank : opId.bsrInfo[HCCL_RECV].detRank;
2042 0 : bool isBsrPortSatisfy = (remoteRankPortMap.find(remoteRank) != remoteRankPortMap.end() &&
2043 0 : remoteRankPortMap.find(remoteRank)->second == isUseDefault);
2044 0 : bool isQpnSatify = (opId.bsrInfo[HCCL_RECV].tpQpn != 0) && (opId.bsrInfo[HCCL_SEND].tpQpn != 0);
2045 0 : HCCL_INFO("[SetBsrTransportStatusImpl]SendremoteRank[%u], RecvremoteRank[%u], isUseDefault[%d], isQpnSatify[%d], isBsrPortSatisfy[%d]",
2046 : sendRemoteRank, recvRemoteRank, isUseDefault, isQpnSatify, isBsrPortSatisfy);
2047 0 : if (opId.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV &&
2048 0 : sendRemoteRank == recvRemoteRank && isBsrPortSatisfy && isQpnSatify)
2049 : {
2050 0 : CHK_RET(SetBsrTransportStatusImpl(opCommTransport, statusStop, opId, remoteRank));
2051 0 : return HCCL_SUCCESS;
2052 : }
2053 :
2054 0 : for (auto &levelNSubCommTransport : opCommTransport)
2055 : {
2056 0 : for (auto &singleSubCommTransport : levelNSubCommTransport)
2057 : {
2058 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++)
2059 : {
2060 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
2061 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
2062 0 : bool isRankSatisfy = ((!isSendRecv) || (isSendRecv && remoteRank == transportRemoteRank));
2063 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
2064 0 : bool isPortSatisfy = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end() &&
2065 0 : remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
2066 0 : HCCL_INFO("[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], isRankSatisfy[%d], isPortSatisfy[%d]",
2067 : transportRemoteRank, isUseDefault, isValid, isRankSatisfy, isPortSatisfy);
2068 0 : if (isValid && isRankSatisfy && isPortSatisfy)
2069 : {
2070 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, statusStop));
2071 : }
2072 : }
2073 : }
2074 : }
2075 0 : return HCCL_SUCCESS;
2076 : }
2077 :
2078 0 : HcclResult HcclCommunicator::SetTransportStatusImplForChange(OpCommTransport &opCommTransport, const HcclOpIdentifier &opId,
2079 : u32 remoteRank, const std::map<u32, bool> &remoteRankPortMap, bool isUseDefault, const std::map<u32, bool> &isChangeLinkMap,
2080 : bool isCurTag)
2081 : {
2082 0 : bool isSendRecv = opId.isSendRecv;
2083 :
2084 : // bsr判断当前故障的send、recv是否remoterank是否相同的情况，如果是相同只操作故障op，如果不同都操作
2085 0 : u32 sendRemoteRank = userRank_ == opId.bsrInfo[HCCL_SEND].detRank ? opId.bsrInfo[HCCL_SEND].srcRank : opId.bsrInfo[HCCL_SEND].detRank;
2086 0 : u32 recvRemoteRank = userRank_ == opId.bsrInfo[HCCL_RECV].detRank ? opId.bsrInfo[HCCL_RECV].srcRank : opId.bsrInfo[HCCL_RECV].detRank;
2087 0 : bool isQpnSatify = (opId.bsrInfo[HCCL_RECV].tpQpn != 0) && (opId.bsrInfo[HCCL_SEND].tpQpn != 0);
2088 0 : HCCL_INFO("[SetBsrTransportStatusImpl]SendremoteRank[%u], RecvremoteRank[%u], isUseDefault[%d], isQpnSatify[%d]",
2089 : sendRemoteRank, recvRemoteRank, isUseDefault, isQpnSatify);
2090 0 : if (opId.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV && sendRemoteRank == recvRemoteRank && isQpnSatify)
2091 : {
2092 0 : CHK_RET(SetBsrTransportStatusImplforchange(opCommTransport, opId, remoteRank, remoteRankPortMap, isUseDefault,
2093 : isChangeLinkMap, isCurTag));
2094 0 : return HCCL_SUCCESS;
2095 : }
2096 :
2097 : // 借轨的resume阶段
2098 0 : for (auto &levelNSubCommTransport : opCommTransport)
2099 : {
2100 0 : for (auto &singleSubCommTransport : levelNSubCommTransport)
2101 : {
2102 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++)
2103 : {
2104 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
2105 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
2106 0 : bool isRankSatisfy = (!isSendRecv || (isSendRecv && remoteRank == transportRemoteRank));
2107 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
2108 0 : bool isPortSatisfy = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end() &&
2109 0 : remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
2110 0 : bool isChangeLink = (isChangeLinkMap.find(transportRemoteRank) != isChangeLinkMap.end() &&
2111 0 : isChangeLinkMap.find(transportRemoteRank)->second);
2112 0 : HCCL_INFO("[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], isRankSatisfy[%d], "
2113 : "isPortSatisfy[%d], isChangeLink[%d], isCurTag[%d]",
2114 : transportRemoteRank, isUseDefault, isValid, isRankSatisfy, isPortSatisfy, isChangeLink, isCurTag);
2115 0 : if (isValid && isRankSatisfy && isPortSatisfy && (isChangeLink || isCurTag))
2116 : {
2117 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, false));
2118 : }
2119 : }
2120 : }
2121 : }
2122 0 : return HCCL_SUCCESS;
2123 : }
2124 :
2125 0 : HcclResult HcclCommunicator::SetTransportStatus(const HcclOpIdentifier &opId, bool statusStop,
2126 : const std::map<u32, bool> &remoteRankPortMap, const std::map<u32, bool> &isChangeLinkMap, bool isChangeLinkFlag)
2127 : {
2128 0 : std::string newTag(reinterpret_cast<const char *>(opId.newTag));
2129 0 : u32 remoteRank = userRank_ == opId.detRank ? opId.srcRank : opId.detRank;
2130 :
2131 0 : if (resMap_.find(newTag) == resMap_.end())
2132 : {
2133 0 : HCCL_ERROR("HcclCommunicator SetTransportStatus failed: newTag[%s] is not in resMap", newTag.c_str());
2134 0 : return HCCL_E_INTERNAL;
2135 : }
2136 :
2137 0 : if (statusStop)
2138 : {
2139 0 : CHK_RET(SetTransportStatusImpl(resMap_[newTag].opTransportResponse, statusStop, opId, remoteRank,
2140 : remoteRankPortMap, true));
2141 0 : CHK_RET(SetTransportStatusImpl(resMap_[newTag].opTransportResponseBackUp, statusStop, opId,
2142 : remoteRank, remoteRankPortMap, false));
2143 : }
2144 : else
2145 : {
2146 0 : if (isChangeLinkFlag)
2147 : {
2148 : // 借轨场景
2149 0 : for (auto &resMapIt : resMap_)
2150 : {
2151 0 : bool isCurTag = false;
2152 0 : if (resMapIt.first == newTag)
2153 : {
2154 0 : isCurTag = true;
2155 : }
2156 0 : if (hostResMap_.find(resMapIt.first) != hostResMap_.end())
2157 : {
2158 : // 若当前tag未进行aicpu展开，则不重新build资源
2159 0 : continue;
2160 : }
2161 :
2162 0 : if ((HcclCMDType::HCCL_CMD_BATCH_SEND_RECV == opId.opType && !isCurTag) ||
2163 0 : (HcclCMDType::HCCL_CMD_BATCH_SEND_RECV != opId.opType &&
2164 0 : resMapIt.first.find("BatchSendRecv") != std::string::npos))
2165 : {
2166 0 : continue;
2167 : }
2168 0 : CHK_RET(SetTransportStatusImplForChange(resMapIt.second.opTransportResponse, opId, remoteRank,
2169 : remoteRankPortMap, true, isChangeLinkMap, isCurTag));
2170 0 : CHK_RET(SetTransportStatusImplForChange(resMapIt.second.opTransportResponseBackUp, opId,
2171 : remoteRank, remoteRankPortMap, false, isChangeLinkMap, isCurTag));
2172 :
2173 0 : std::string tag;
2174 0 : CHK_RET(GetTagFromNewTag(resMapIt.first, tag));
2175 0 : CHK_RET(ReAllocTransports(tag, resMapIt.first));
2176 0 : CHK_RET(BuildOpRemoteResParam(resMapIt.second, resMapIt.first, opId.opType, true));
2177 0 : HCCL_RUN_INFO("[%s]success to set status of [%s] resume", __func__, resMapIt.first.c_str());
2178 0 : }
2179 0 : CHK_RET(CopyHostOpResToDeviceParam(newTag));
2180 : }
2181 : else
2182 : {
2183 : // 原地重执行
2184 0 : CHK_RET(SetTransportStatusImpl(resMap_[newTag].opTransportResponse, statusStop, opId,
2185 : remoteRank, remoteRankPortMap, true));
2186 0 : CHK_RET(SetTransportStatusImpl(resMap_[newTag].opTransportResponseBackUp, statusStop, opId,
2187 : remoteRank, remoteRankPortMap, false));
2188 0 : std::string tag(reinterpret_cast<const char *>(opId.tag));
2189 0 : CHK_RET(ReAllocTransports(tag, newTag));
2190 0 : CHK_RET(BuildOpRemoteResParam(resMap_[newTag], newTag, opId.opType, true));
2191 0 : CHK_RET(CopyHostOpResToDeviceParam(newTag));
2192 0 : }
2193 : }
2194 0 : return HCCL_SUCCESS;
2195 0 : }
2196 :
2197 0 : HcclResult HcclCommunicator::SetTransportResumeStatus(const std::map<u32, bool> &remoteRankPortMap, const std::map<u32, bool> &isChangeLinkMap, bool isChangeLinkFlag, bool statusStop)
2198 : {
2199 0 : HCCL_INFO("[SetTransportResumeStatus]isChangeLinkFlag[%d], rank[%u], group[%s]", isChangeLinkFlag, userRank_, identifier_.c_str());
2200 :
2201 0 : if (statusStop)
2202 : {
2203 0 : for (auto &resMapIt: resMap_)
2204 : {
2205 0 : CHK_RET(ResumeTransportsImpl(resMapIt.second.opTransportResponse, remoteRankPortMap, true, statusStop));
2206 0 : CHK_RET(ResumeTransportsImpl(resMapIt.second.opTransportResponseBackUp, remoteRankPortMap, false, statusStop));
2207 : }
2208 : }
2209 : else
2210 : {
2211 0 : if (isChangeLinkFlag)
2212 : {
2213 0 : for (auto &resMapIt: resMap_)
2214 : {
2215 0 : if (hostResMap_.find(resMapIt.first) != hostResMap_.end())
2216 : {
2217 0 : continue;
2218 : }
2219 0 : CHK_RET(ResumeTransportsImplForChange(resMapIt.second.opTransportResponse,
2220 : remoteRankPortMap, isChangeLinkMap, true));
2221 0 : CHK_RET(ResumeTransportsImplForChange(resMapIt.second.opTransportResponseBackUp,
2222 : remoteRankPortMap, isChangeLinkMap, false));
2223 :
2224 0 : std::string tag;
2225 0 : CHK_RET(GetTagFromNewTag(resMapIt.first, tag));
2226 0 : CHK_RET(ReAllocTransports(tag, resMapIt.first));
2227 0 : CHK_RET(BuildOpRemoteResParam(resMapIt.second, resMapIt.first, HcclCMDType::HCCL_CMD_ALL, true));
2228 0 : CHK_RET(CopyHostOpResToDeviceParam(resMapIt.first));
2229 0 : }
2230 : }
2231 : else
2232 : {
2233 0 : for (auto &resMapIt: resMap_)
2234 : {
2235 0 : CHK_RET(ResumeTransportsImpl(resMapIt.second.opTransportResponse, remoteRankPortMap, true, statusStop));
2236 0 : CHK_RET(ResumeTransportsImpl(resMapIt.second.opTransportResponseBackUp, remoteRankPortMap, false, statusStop));
2237 0 : std::string tag;
2238 0 : CHK_RET(GetTagFromNewTag(resMapIt.first, tag));
2239 0 : CHK_RET(ReAllocTransports(tag, resMapIt.first));
2240 0 : CHK_RET(BuildOpRemoteResParam(resMapIt.second, resMapIt.first, HcclCMDType::HCCL_CMD_ALL, true));
2241 0 : CHK_RET(CopyHostOpResToDeviceParam(resMapIt.first));
2242 0 : }
2243 : }
2244 : }
2245 0 : return HCCL_SUCCESS;
2246 : }
2247 :
2248 0 : HcclResult HcclCommunicator::ResumeTransportsImplForChange(OpCommTransport &opCommTransport, const std::map<u32, bool> &remoteRankPortMap,
2249 : const std::map<u32, bool> &isChangeLinkMap, bool isUseDefault)
2250 : {
2251 : // 借轨的resume阶段
2252 0 : for (auto &levelNSubCommTransport: opCommTransport)
2253 : {
2254 0 : for (auto &singleSubCommTransport: levelNSubCommTransport)
2255 : {
2256 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++)
2257 : {
2258 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
2259 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
2260 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
2261 0 : bool isPortSatisfy = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end() &&
2262 0 : remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
2263 0 : bool isChangeLink = (isChangeLinkMap.find(transportRemoteRank) != isChangeLinkMap.end());
2264 0 : HCCL_INFO("[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], "
2265 : "isPortSatisfy[%d], isChangeLink[%d]",
2266 : transportRemoteRank, isUseDefault, isValid, isPortSatisfy, isChangeLink);
2267 0 : if (isValid && isPortSatisfy && isChangeLink)
2268 : {
2269 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, false));
2270 : }
2271 : }
2272 : }
2273 : }
2274 0 : return HCCL_SUCCESS;
2275 : }
2276 :
2277 0 : HcclResult HcclCommunicator::ResumeTransportsImpl(OpCommTransport &opCommTransport,
2278 : const std::map<u32, bool> &remoteRankPortMap, bool isUseDefault, bool statusStop)
2279 : {
2280 0 : for (auto &levelNSubCommTransport: opCommTransport) {
2281 0 : for (auto &singleSubCommTransport: levelNSubCommTransport) {
2282 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
2283 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
2284 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
2285 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
2286 0 : bool isPortSatisfy = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end() &&
2287 0 : remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
2288 0 : HCCL_INFO("[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], isPortSatisfy[%d]",
2289 : transportRemoteRank, isUseDefault, isValid, isPortSatisfy);
2290 0 : if (isValid && isPortSatisfy) {
2291 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, statusStop));
2292 : }
2293 : }
2294 : }
2295 : }
2296 0 : return HCCL_SUCCESS;
2297 : }
2298 :
2299 0 : HcclResult HcclCommunicator::ReAllocTransports(const std::string &tag, const std::string &newTag)
2300 : {
2301 0 : HcclResult ret = HCCL_SUCCESS;
2302 0 : HCCL_INFO("[%s] alloc tag[%s] transports", __func__, newTag.c_str());
2303 :
2304 0 : AlgResourceResponse &algResResponse = resMap_[newTag];
2305 0 : DeviceMem expMem = cclBufferManager_.GetCommCCLBuffer();
2306 :
2307 0 : TransportIOMem transMem{algResResponse.cclInputMem, algResResponse.cclOutputMem,
2308 0 : algResResponse.paramInputMem, algResResponse.paramOutputMem, algResResponse.scratchMem,
2309 0 : algResResponse.aivInputMem, algResResponse.aivOutputMem, expMem, DeviceMem()};
2310 :
2311 : {
2312 : // Transport资源 重建链, 一定是AICPU展开，所以 isAicpuModeEn=true
2313 0 : StateGuard<HcclCommunicator, HcclCommState> guard(this, HcclCommState::BUILDING);
2314 0 : ret = transportManager_->Alloc(tag, transMem, algResResponse.opTransportResponse, true);
2315 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]Realloc transports failed, tag[%s]", __func__, newTag.c_str()),
2316 : ret);
2317 0 : }
2318 :
2319 0 : if (IsEnableBackupLink())
2320 : {
2321 0 : HCCL_INFO("[%s] alloc tag[%s] backup transports", __func__, newTag.c_str());
2322 : // 超节点 && level2支持重执行 && Aicpu：备用Transport资源 重建链
2323 0 : StateGuard<HcclCommunicator, HcclCommState> guard(this, HcclCommState::BUILDING);
2324 0 : ret = transportManager_->Alloc(tag, transMem, algResResponse.opTransportResponseBackUp, true, true);
2325 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
2326 : HCCL_ERROR("[%s]Alloc backup transports failed, tag[%s]", __func__, newTag.c_str()), ret);
2327 0 : }
2328 0 : SaveLinkRes(algResResponse.opTransportResponse);
2329 0 : SaveLinkRes(algResResponse.opTransportResponseBackUp);
2330 0 : return HCCL_SUCCESS;
2331 0 : }
2332 :
2333 6 : HcclResult HcclCommunicator::Stop()
2334 : {
2335 6 : HcclUs startut = TIME_NOW();
2336 6 : isSuspending = true;
2337 6 : HCCL_DEBUG("HcclCommunicator Stop begin.");
2338 6 : for (auto &it : tagCommInfo_)
2339 : {
2340 0 : CHK_RET(TraverseLinkVector(it.second.commLevel1, true));
2341 0 : CHK_RET(TraverseLinkVector(it.second.commLevel0, true));
2342 0 : CHK_RET(TraverseLinkVector(it.second.commLevel2, true));
2343 0 : CHK_RET(TraverseLinkVector(it.second.commP2P, true));
2344 0 : if (it.second.commIntraServer)
2345 : {
2346 0 : const std::vector<LINK> &ret = it.second.commIntraServer->TransportInfo();
2347 0 : CHK_RET(MigrateLinkVectorToStopOrResume(ret, true));
2348 : }
2349 : }
2350 6 : CHK_RET(TraverseAlgResourceResponse(true));
2351 6 : HcclUs endut = TIME_NOW();
2352 6 : HCCL_RUN_INFO("HcclCommunicator::Stop, Stop take time:[%lld]us",
2353 : DURATION_US(endut - startut).count());
2354 6 : return HCCL_SUCCESS;
2355 : }
2356 :
2357 1 : HcclResult HcclCommunicator::HostMC2EnvResume()
2358 : {
2359 1 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine())
2360 : {
2361 0 : HCCL_DEBUG("[NsRecovery]reset the suspending flag");
2362 0 : KfcExecControl controlCmd;
2363 0 : controlCmd.kfcCmd = KfcCommand::kNone;
2364 0 : controlCmd.bgCmd = BackgroundCommand::kNone;
2365 0 : controlCmd.suspendingStatus = HcclComSuspendingFlag::isResume;
2366 0 : CHK_RET(kfcControlTransferH2D_->Put(0, sizeof(KfcExecControl), reinterpret_cast<uint8_t *>(&controlCmd)));
2367 0 : if (IsEnableCustom())
2368 : {
2369 0 : CHK_RET(customControlTransferH2D_->Put(0, sizeof(KfcExecControl), reinterpret_cast<uint8_t *>(&controlCmd)));
2370 : }
2371 : }
2372 1 : return HCCL_SUCCESS;
2373 : }
2374 :
2375 1 : HcclResult HcclCommunicator::ClearWinBuffer()
2376 : {
2377 1 : DeviceMem winBuffer = cclBufferManager_.GetCommExpBuffer();
2378 1 : if (winBuffer.ptr() != nullptr)
2379 : {
2380 0 : HCCL_INFO("HcclCommunicator::Resume, start to clear win buffer");
2381 0 : CHK_RET(hrtMemSet(static_cast<u8 *>(winBuffer.ptr()), EXP_BUFFER_SIZE, EXP_BUFFER_SIZE));
2382 : }
2383 1 : return HCCL_SUCCESS;
2384 1 : }
2385 :
2386 1 : HcclResult HcclCommunicator::AivResume()
2387 : {
2388 1 : if (GetExternalInputHcclAivMode())
2389 : {
2390 0 : HCCL_DEBUG("AivResume begin.");
2391 :
2392 0 : CHK_RET(cclBufferManager_.ClearCommAIVbuffer());
2393 0 : HCCL_INFO("[AIV][AivResumeClearSyncBuf] clear aiv buffer done");
2394 :
2395 0 : aivOpbaseTag_ = TAG_INIT_VALUE;
2396 0 : aivOffloadTag_ = TAG_INIT_VALUE;
2397 0 : HCCL_INFO("[AIV][AivResume] clear aiv tag done");
2398 : }
2399 1 : return HCCL_SUCCESS;
2400 : }
2401 :
2402 1 : HcclResult HcclCommunicator::Resume()
2403 : {
2404 1 : HcclUs startut = TIME_NOW();
2405 1 : bool isChangedLink = false;
2406 1 : HCCL_RUN_INFO("HcclCommunicator Resume begin, group[%s].", identifier_.c_str());
2407 : // 发生N秒快恢, 头尾计数可能不对，需要将头尾计数清零
2408 1 : CHK_RET(ClearOpCounterMem());
2409 1 : for (auto &it : tagCommInfo_)
2410 : {
2411 0 : CHK_RET(TraverseLinkVector(it.second.commLevel1, false));
2412 0 : CHK_RET(TraverseLinkVector(it.second.commLevel0, false));
2413 0 : CHK_RET(TraverseLinkVector(it.second.commLevel2, false));
2414 0 : CHK_RET(TraverseLinkVector(it.second.commP2P, false));
2415 0 : if (it.second.commIntraServer)
2416 : {
2417 0 : const std::vector<LINK> &ret = it.second.commIntraServer->TransportInfo();
2418 0 : CHK_RET(MigrateLinkVectorToStopOrResume(ret, false));
2419 : }
2420 : }
2421 :
2422 1 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine()) {
2423 0 : CHK_RET(CheckExitWaitResumeState(isChangedLink));
2424 : }
2425 :
2426 1 : if (!isChangedLink) {
2427 1 : CHK_RET(TraverseAlgResourceResponse(false));
2428 : }
2429 1 : HcclUs cleanNotifyStart = TIME_NOW();
2430 1 : CHK_RET(hrtResourceClean());
2431 1 : HcclUs cleanNotifyEnd = TIME_NOW();
2432 1 : HCCL_RUN_INFO("HcclCommunicator::Resume, hrtResourceClean notify take time:[%lld]us",
2433 : DURATION_US(cleanNotifyEnd - cleanNotifyStart).count());
2434 1 : CHK_RET(HostMC2EnvResume());
2435 1 : CHK_RET(ClearWinBuffer());
2436 1 : CHK_RET(AivResume());
2437 1 : isSuspending = false;
2438 :
2439 1 : HcclUs endut = TIME_NOW();
2440 1 : HCCL_RUN_INFO("HcclCommunicator::Resume, Resume take time:[%lld]us, group[%s]",
2441 : DURATION_US(endut - startut).count(), identifier_.c_str());
2442 :
2443 1 : return HCCL_SUCCESS;
2444 : }
2445 :
2446 134 : HcclResult HcclCommunicator::CheckSuspendingStatus()
2447 : {
2448 134 : if (isSuspending)
2449 : {
2450 0 : return HCCL_E_SUSPENDING;
2451 : }
2452 134 : return HCCL_SUCCESS;
2453 : }
2454 :
2455 0 : HcclResult HcclCommunicator::SwitchNic(uint32_t nRanks, uint32_t *ranks, bool *useBackup)
2456 : {
2457 0 : CHK_RET(SwitchNic(nRanks, ranks, useBackup, kfcControlTransferH2D_, kfcStatusTransferD2H_));
2458 0 : if (IsEnableCustom())
2459 : {
2460 0 : CHK_RET(SwitchNic(nRanks, ranks, useBackup, customControlTransferH2D_, customStatusTransferD2H_));
2461 : }
2462 0 : return HCCL_SUCCESS;
2463 : }
2464 :
2465 122 : void HcclCommunicator::SaveLinkRes(const OpCommTransport &opTransportResponse)
2466 : {
2467 122 : std::lock_guard<std::mutex> commLock(linkResMapMutex_);
2468 1066 : for (auto &opCommTransport : opTransportResponse)
2469 : {
2470 1044 : for (auto &transports : opCommTransport)
2471 : {
2472 375 : for (u32 i = 0; i < transports.transportRequests.size(); i++)
2473 : {
2474 275 : if (transports.links[i] != nullptr &&
2475 0 : transports.links[i]->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP)
2476 : {
2477 0 : auto remoteRank = transports.transportRequests[i].remoteUserRank;
2478 0 : std::string localServerId = rankInfoList_.size() > userRank_ ? rankInfoList_[userRank_].serverId : "";
2479 0 : s32 localDevicePhyId = rankInfoList_.size() > userRank_ ? rankInfoList_[userRank_].devicePhyId : -1;
2480 0 : std::string remoteServerId = rankInfoList_.size() > remoteRank ? rankInfoList_[remoteRank].serverId : "";
2481 0 : s32 remoteDevicePhyId = rankInfoList_.size() > remoteRank ? rankInfoList_[remoteRank].devicePhyId : -1;
2482 0 : LinkInfo linkInfo(identifier_, userRank_, localServerId, localDevicePhyId, remoteRank, remoteServerId, remoteDevicePhyId);
2483 0 : linkResMap_.emplace(transports.links[i].get(), linkInfo);
2484 0 : }
2485 : }
2486 : }
2487 : }
2488 244 : return;
2489 122 : }
2490 :
2491 15 : HcclResult HcclCommunicator::GetTransportCqeErrors(const HcclNetDevCtx netDevCtx,
2492 : std::vector<ErrCqeInfo> &infos, u32 &num)
2493 : {
2494 15 : if (netDevCtx == nullptr)
2495 : {
2496 0 : return HCCL_SUCCESS;
2497 : }
2498 15 : HcclIpAddress localIp;
2499 15 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
2500 :
2501 15 : u32 qpn = 0;
2502 15 : std::vector<std::pair<Transport *, CqeInfo>> infolist;
2503 15 : Transport::GetTransportErrorCqe(netDevCtx, infolist, num);
2504 15 : std::lock_guard<std::mutex> commLock(linkResMapMutex_);
2505 16 : for (auto &info : infolist)
2506 : {
2507 1 : auto iter = linkResMap_.find(info.first);
2508 1 : if (iter != linkResMap_.end())
2509 : {
2510 0 : CHK_RET((info.first)->GetTransportId(qpn));
2511 0 : infos.push_back(ErrCqeInfo(info.second, iter->second, qpn));
2512 : }
2513 : else
2514 : {
2515 1 : LinkInfo linkInfo;
2516 1 : linkInfo.localDevicePhyId = -1;
2517 1 : linkInfo.remoteDevicePhyId = -1;
2518 1 : infos.push_back(ErrCqeInfo(info.second, linkInfo, qpn));
2519 1 : HCCL_RUN_WARNING("[GetTransportCqeErrors]Transport linkInfo was not saved, only print CqeInfo");
2520 1 : }
2521 : }
2522 15 : num = infos.size();
2523 15 : return HCCL_SUCCESS;
2524 15 : }
2525 :
2526 112 : void HcclCommunicator::ClearOpTransportResponseLinks(OpCommTransport &opTransportResponse)
2527 : {
2528 1850 : for (auto &levelNSubCommTransport : opTransportResponse)
2529 : {
2530 2056 : for (auto &singleSubCommTransport : levelNSubCommTransport)
2531 : {
2532 311 : u32 size = singleSubCommTransport.transportRequests.size();
2533 311 : singleSubCommTransport.links.resize(size, nullptr);
2534 309 : singleSubCommTransport.status.resize(size, TransportStatus::INIT);
2535 306 : HCCL_INFO("[%s] size[%u], linksSize[%d]", __func__, size, singleSubCommTransport.links.size());
2536 : }
2537 : }
2538 108 : }
2539 :
2540 0 : HcclResult HcclCommunicator::SetDevIbverbsData(CommBase *comm, bool isSupportNormalQP, u64 commBufferSize,
2541 : void *commInPtr, void *commOutPtr)
2542 : {
2543 0 : const u32 curRankId = comm->Rank();
2544 0 : const u32 rankSize = comm->RankSize();
2545 :
2546 0 : CHK_RET(AllocAndClearHostMem(sizeof(TransportDeviceNormalData) * rankSize, transDevIbverbsDataMem_));
2547 0 : TransportDeviceNormalData *transDevIbverbsData = reinterpret_cast<TransportDeviceNormalData*>(transDevIbverbsDataMem_->ptr());
2548 :
2549 0 : for (u32 i = 0; i < rankSize; i++)
2550 : {
2551 0 : auto &data = transDevIbverbsData[i];
2552 0 : if (i != curRankId)
2553 : {
2554 : // 对端link的信息
2555 0 : const auto transport = comm->GetTransportByRank(i);
2556 0 : void *bufferIn = nullptr;
2557 0 : void *bufferOut = nullptr;
2558 0 : u32 remoteInMemKey = 0;
2559 0 : u32 remoteOutMemKey = 0;
2560 0 : CHK_RET(transport->GetRemoteMem(UserMemType::INPUT_MEM, &bufferIn));
2561 0 : CHK_RET(transport->GetRemoteMem(UserMemType::OUTPUT_MEM, &bufferOut));
2562 0 : data.remoteInputMem.addr = reinterpret_cast<uint64_t>(bufferIn);
2563 0 : data.remoteOutputMem.addr = reinterpret_cast<uint64_t>(bufferOut);
2564 0 : CHK_RET(transport->GetRemoteMemSize(UserMemType::INPUT_MEM, data.remoteInputMem.size));
2565 0 : CHK_RET(transport->GetRemoteMemSize(UserMemType::OUTPUT_MEM, data.remoteOutputMem.size));
2566 : // IBV链路需要的资源
2567 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP)
2568 : {
2569 0 : CHK_RET(transport->GetRemoteMemKey(UserMemType::INPUT_MEM, &remoteInMemKey));
2570 0 : CHK_RET(transport->GetRemoteMemKey(UserMemType::OUTPUT_MEM, &remoteOutMemKey));
2571 0 : data.remoteInputMem.key = remoteInMemKey;
2572 0 : data.remoteOutputMem.key = remoteOutMemKey;
2573 0 : CHK_RET(transport->GetLocalMemDetails(UserMemType::INPUT_MEM, data.localInputMem));
2574 0 : CHK_RET(transport->GetLocalMemDetails(UserMemType::OUTPUT_MEM, data.localOutputMem));
2575 0 : std::vector<HcclQpInfoV2> qpInfos;
2576 0 : CHK_RET(transport->GetAiQpInfo(qpInfos));
2577 0 : data.qpInfo = qpInfos[0];
2578 0 : }
2579 0 : }
2580 : else
2581 : {
2582 : // 本rank的信息
2583 0 : data.localInputMem.addr = reinterpret_cast<uint64_t>(commInPtr);
2584 0 : data.localInputMem.size = commBufferSize;
2585 0 : data.localOutputMem.addr = reinterpret_cast<uint64_t>(commOutPtr);
2586 0 : data.localOutputMem.size = commBufferSize;
2587 : }
2588 :
2589 0 : if (isSupportNormalQP)
2590 : {
2591 0 : data.qpMode = QPMode::NORMAL;
2592 : }
2593 : // Debugging info
2594 0 : data.Print();
2595 : }
2596 0 : return HCCL_SUCCESS;
2597 : }
2598 :
2599 0 : HcclResult HcclCommunicator::GetTransportLocalMem(const std::shared_ptr<Transport> &transport,
2600 : UserMemType memType, MemDetails &detail)
2601 : {
2602 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP)
2603 : {
2604 0 : CHK_RET(transport->GetLocalMemDetails(memType, detail));
2605 : }
2606 0 : return HCCL_SUCCESS;
2607 : }
2608 :
2609 0 : HcclResult HcclCommunicator::GetTransportRemoteMem(const std::shared_ptr<Transport> &transport,
2610 : UserMemType memType, MemDetails &detail)
2611 : {
2612 0 : void *addr = nullptr;
2613 0 : CHK_RET(transport->GetRemoteMem(memType, &addr));
2614 :
2615 0 : detail.addr = reinterpret_cast<uint64_t>(addr);
2616 0 : CHK_RET(transport->GetRemoteMemSize(memType, detail.size));
2617 :
2618 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP)
2619 : {
2620 0 : CHK_RET(transport->GetRemoteMemKey(memType, &detail.key));
2621 : }
2622 :
2623 0 : return HCCL_SUCCESS;
2624 : }
2625 :
2626 0 : HcclResult HcclCommunicator::GenAiRMAInfo(CommBase *comm)
2627 : {
2628 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2629 0 : HcclAiRMAInfo *aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2630 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2631 :
2632 0 : const std::string &tag = comm->Tag();
2633 0 : aiRMAInfoPtr->curRankId = comm->Rank();
2634 0 : aiRMAInfoPtr->rankNum = comm->RankSize();
2635 :
2636 0 : CHK_RET(GetAIVNormalQPInfo(comm, tag));
2637 :
2638 0 : u32 tmpQueueSize = aiRMAInfoPtr->rankNum * aiRMAInfoPtr->qpNum;
2639 0 : u32 tmpMemSize = aiRMAInfoPtr->rankNum;
2640 0 : u32 tmpMemDetailSize = aiRMAInfoPtr->rankNum * AiMemMaxNum;
2641 :
2642 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiSqMem_));
2643 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiScqMem_));
2644 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiRqMem_));
2645 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiRcqMem_));
2646 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAMemInfo) * tmpMemSize, aiMemMem_));
2647 0 : HcclAiRMAMemInfo *aiMemHost = reinterpret_cast<HcclAiRMAMemInfo*>(aiMemMem_->ptr());
2648 :
2649 0 : CHK_RET(AllocAndClearHostMem(sizeof(MemDetails) * tmpMemDetailSize, aiMemDetailsMem_));
2650 0 : MemDetails *aiMemDetailsHost = reinterpret_cast<MemDetails*>(aiMemDetailsMem_->ptr());
2651 0 : CHK_RET(DeviceMem::alloc(aiMemDetailsDev_, aiMemDetailsMem_->size()));
2652 0 : u64 memBase = reinterpret_cast<uint64_t>(aiMemDetailsDev_.ptr());
2653 :
2654 0 : for (u32 i = 0; i < aiRMAInfoPtr->rankNum; i++)
2655 : {
2656 0 : MemDetails &remoteIn = aiMemDetailsHost[i * AiMemMaxNum +
2657 0 : GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_INPUT)];
2658 0 : MemDetails &remoteOut = aiMemDetailsHost[i * AiMemMaxNum +
2659 0 : GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_OUTPUT)];
2660 0 : MemDetails &localIn = aiMemDetailsHost[i * AiMemMaxNum +
2661 0 : GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_INPUT)];
2662 0 : MemDetails &localOut = aiMemDetailsHost[i * AiMemMaxNum +
2663 0 : GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_OUTPUT)];
2664 :
2665 0 : if (i != aiRMAInfoPtr->curRankId)
2666 : {
2667 : // link rank info
2668 0 : const auto transport = comm->GetTransportByRank(i);
2669 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::INPUT_MEM, remoteIn));
2670 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::OUTPUT_MEM, remoteOut));
2671 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::INPUT_MEM, localIn));
2672 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::OUTPUT_MEM, localOut));
2673 :
2674 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP)
2675 : {
2676 0 : CHK_RET(GenIbvAiRMAInfo(i, transport, tag, aiRMAInfoPtr));
2677 : }
2678 0 : }
2679 : else
2680 : {
2681 0 : void *commInPtr = nullptr;
2682 0 : void *commOutPtr = nullptr;
2683 : u64 commInSize;
2684 : u64 commOutSize;
2685 0 : CHK_RET(cclBufferManager_.GetInCCLbuffer(commInPtr, commInSize));
2686 0 : CHK_RET(cclBufferManager_.GetOutCCLbuffer(commOutPtr, commOutSize));
2687 0 : localIn.addr = reinterpret_cast<uint64_t>(commInPtr);
2688 0 : localIn.size = commInSize;
2689 0 : localOut.addr = reinterpret_cast<uint64_t>(commOutPtr);
2690 0 : localOut.size = commOutSize;
2691 : }
2692 :
2693 0 : aiMemHost[i].memMaxNum = AiMemMaxNum;
2694 0 : aiMemHost[i].sizeOfMemDetails = static_cast<u32>(sizeof(MemDetails));
2695 0 : aiMemHost[i].memDetailPtr = memBase + i * AiMemMaxNum * aiMemHost[i].sizeOfMemDetails;
2696 :
2697 0 : HCCL_DEBUG("[%s] tag[%s] curRankId[%u] dstRankId[%u] rankNum[%u] qpNum[%u] memMaxNum[%u] sizeOfMemDetails[%u] "
2698 : "memDetailPtr[%p] remoteInAddr[%p] remoteInSize[%llu] remoteInKey[%u] remoteOutAddr[%p] "
2699 : "remoteOutSize[%llu] remoteOutKey[%u] localInAddr[%p] localInSize[%llu] localInKey[%u] "
2700 : "localOutAddr[%p] localOutSize[%llu] localOutKey[%u]",
2701 : __func__, tag.c_str(), aiRMAInfoPtr->curRankId, i, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum,
2702 : aiMemHost[i].memMaxNum, aiMemHost[i].sizeOfMemDetails, aiMemHost[i].memDetailPtr,
2703 : remoteIn.addr, remoteIn.size, remoteIn.key, remoteOut.addr, remoteOut.size, remoteOut.key,
2704 : localIn.addr, localIn.size, localIn.key, localOut.addr, localOut.size, localOut.key);
2705 : }
2706 :
2707 0 : return HCCL_SUCCESS;
2708 : }
2709 :
2710 0 : HcclResult HcclCommunicator::GenAiRMAInfoV2(const std::string &tag)
2711 : {
2712 0 : CHK_PTR_NULL(rmaInfoMem_);
2713 0 : HcclRMAInfo *rmaInfoPtr = reinterpret_cast<HcclRMAInfo*>(rmaInfoMem_->ptr());
2714 0 : CHK_PTR_NULL(rmaInfoPtr);
2715 0 : rmaInfoPtr->curRankId = userRank_;;
2716 0 : rmaInfoPtr->rankNum = userRankSize_;
2717 0 : LevelNSubCommTransport& commTransport = resMap_[tag].opTransportResponse[COMM_COMBINE_ORDER];
2718 0 : CHK_PRT_RET(commTransport.size() <= 0, HCCL_ERROR("[%s] no LevelComm resource, please create comm first. "
2719 : "tag[%s], curRankId[%u] rankNum[%u]", __func__, tag.c_str(), rmaInfoPtr->curRankId,
2720 : rmaInfoPtr->rankNum), HCCL_E_INTERNAL);
2721 0 : std::vector<LINK>& links = commTransport[0].links;
2722 0 : CHK_PRT_RET(links.size() <= 0, HCCL_ERROR("[%s] no transport resource, please create links first. "
2723 : "tag[%s], curRankId[%u] rankNum[%u]", __func__, tag.c_str(), rmaInfoPtr->curRankId,
2724 : rmaInfoPtr->rankNum), HCCL_E_INTERNAL);
2725 0 : CHK_RET(GetAIVNormalQPInfoV2(links, tag));
2726 :
2727 0 : u32 tmpQueueSize = rmaInfoPtr->rankNum * rmaInfoPtr->qpNum;
2728 0 : u32 tmpMemSize = rmaInfoPtr->rankNum;
2729 0 : u32 tmpMemDetailSize = rmaInfoPtr->rankNum * AiMemMaxNum;
2730 :
2731 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiSqMem_));
2732 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiScqMem_));
2733 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiRqMem_));
2734 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiRcqMem_));
2735 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAMemInfo) * tmpMemSize, aiMemMem_));
2736 0 : HcclAiRMAMemInfo *aiMemHost = reinterpret_cast<HcclAiRMAMemInfo*>(aiMemMem_->ptr());
2737 :
2738 0 : CHK_RET(AllocAndClearHostMem(sizeof(MemDetails) * tmpMemDetailSize, aiMemDetailsMem_));
2739 0 : MemDetails *aiMemDetailsHost = reinterpret_cast<MemDetails*>(aiMemDetailsMem_->ptr());
2740 :
2741 0 : aiMemDetailsDev_ = DeviceMem::alloc(aiMemDetailsMem_->size());
2742 0 : u64 memBase = reinterpret_cast<uint64_t>(aiMemDetailsDev_.ptr());
2743 :
2744 0 : for (u32 i = 0; i < rmaInfoPtr->rankNum; i++) {
2745 0 : MemDetails &remoteIn = aiMemDetailsHost[i * AiMemMaxNum +
2746 0 : GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_INPUT)];
2747 0 : MemDetails &remoteOut = aiMemDetailsHost[i * AiMemMaxNum +
2748 0 : GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_OUTPUT)];
2749 0 : MemDetails &localIn = aiMemDetailsHost[i * AiMemMaxNum +
2750 0 : GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_INPUT)];
2751 0 : MemDetails &localOut = aiMemDetailsHost[i * AiMemMaxNum +
2752 0 : GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_OUTPUT)];
2753 :
2754 0 : if (i != rmaInfoPtr->curRankId) {
2755 : // link rank info
2756 0 : const auto transport = links[i];
2757 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::INPUT_MEM, remoteIn));
2758 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::OUTPUT_MEM, remoteOut));
2759 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::INPUT_MEM, localIn));
2760 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::OUTPUT_MEM, localOut));
2761 :
2762 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2763 0 : CHK_RET(GenIbvAiRMAInfo(i, transport, tag, rmaInfoPtr));
2764 : }
2765 0 : } else {
2766 0 : void *commInPtr = nullptr;
2767 0 : void *commOutPtr = nullptr;
2768 : u64 commInSize;
2769 : u64 commOutSize;
2770 0 : CHK_RET(cclBufferManager_.GetInCCLbuffer(commInPtr, commInSize));
2771 0 : CHK_RET(cclBufferManager_.GetOutCCLbuffer(commOutPtr, commOutSize));
2772 0 : localIn.addr = reinterpret_cast<uint64_t>(commInPtr);
2773 0 : localIn.size = commInSize;
2774 0 : localOut.addr = reinterpret_cast<uint64_t>(commOutPtr);
2775 0 : localOut.size = commOutSize;
2776 : }
2777 :
2778 0 : aiMemHost[i].memMaxNum = AiMemMaxNum;
2779 0 : aiMemHost[i].sizeOfMemDetails = static_cast<u32>(sizeof(MemDetails));
2780 0 : aiMemHost[i].memDetailPtr = memBase + i * AiMemMaxNum * aiMemHost[i].sizeOfMemDetails;
2781 :
2782 0 : HCCL_DEBUG("[%s] tag[%s] curRankId[%u] dstRankId[%u] rankNum[%u] qpNum[%u] memMaxNum[%u] sizeOfMemDetails[%u] "
2783 : "memDetailPtr[%p] remoteInAddr[%p] remoteInSize[%llu] remoteInKey[%u] remoteOutAddr[%p] "
2784 : "remoteOutSize[%llu] remoteOutKey[%u] localInAddr[%p] localInSize[%llu] localInKey[%u] "
2785 : "localOutAddr[%p] localOutSize[%llu] localOutKey[%u]",
2786 : __func__, tag.c_str(), rmaInfoPtr->curRankId, i, rmaInfoPtr->rankNum, rmaInfoPtr->qpNum,
2787 : aiMemHost[i].memMaxNum, aiMemHost[i].sizeOfMemDetails, aiMemHost[i].memDetailPtr,
2788 : remoteIn.addr, remoteIn.size, remoteIn.key, remoteOut.addr, remoteOut.size, remoteOut.key,
2789 : localIn.addr, localIn.size, localIn.key, localOut.addr, localOut.size, localOut.key);
2790 : }
2791 :
2792 0 : return HCCL_SUCCESS;
2793 : }
2794 :
2795 0 : HcclResult HcclCommunicator::H2DAiRMAInfo(const std::string &tag, rtStream_t aiCpuStream)
2796 : {
2797 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2798 0 : HcclAiRMAInfo *aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2799 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2800 :
2801 0 : CHK_PTR_NULL(combinOparaMem_);
2802 0 : HcclCombinOpParam *combinOparaPtr = reinterpret_cast<HcclCombinOpParam*>(combinOparaMem_->ptr());
2803 0 : CHK_PTR_NULL(combinOparaPtr);
2804 :
2805 0 : aiRMAInfoPtr->sizeOfAiRMAWQ = static_cast<u32>(sizeof(HcclAiRMAWQ));
2806 0 : aiRMAInfoPtr->sizeOfAiRMACQ = static_cast<u32>(sizeof(HcclAiRMACQ));
2807 0 : aiRMAInfoPtr->sizeOfAiRMAMem = static_cast<u32>(sizeof(HcclAiRMAMemInfo));
2808 :
2809 0 : CHK_RET(DeviceMem::alloc(aiSqDev_, aiSqMem_->size()));
2810 0 : aiRMAInfoPtr->sqPtr = aiSqDev_.ptr();
2811 0 : CHK_RET(hrtMemAsyncCopy(aiSqDev_.ptr(), aiSqDev_.size(), aiSqMem_->ptr(), aiSqDev_.size(),
2812 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2813 :
2814 0 : CHK_RET(DeviceMem::alloc(aiScqDev_, aiScqMem_->size()));
2815 0 : aiRMAInfoPtr->scqPtr = aiScqDev_.ptr();
2816 0 : CHK_RET(hrtMemAsyncCopy(aiScqDev_.ptr(), aiScqDev_.size(), aiScqMem_->ptr(), aiScqDev_.size(),
2817 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2818 :
2819 0 : CHK_RET(DeviceMem::alloc(aiRqDev_, aiRqMem_->size()));
2820 0 : aiRMAInfoPtr->rqPtr = aiRqDev_.ptr();
2821 0 : CHK_RET(hrtMemAsyncCopy(aiRqDev_.ptr(), aiRqDev_.size(), aiRqMem_->ptr(), aiRqDev_.size(),
2822 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2823 :
2824 0 : CHK_RET(DeviceMem::alloc(aiRcqDev_, aiRcqMem_->size()));
2825 0 : aiRMAInfoPtr->rcqPtr = aiRcqDev_.ptr();
2826 0 : CHK_RET(hrtMemAsyncCopy(aiRcqDev_.ptr(), aiRcqDev_.size(), aiRcqMem_->ptr(), aiRcqDev_.size(),
2827 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2828 :
2829 0 : CHK_RET(hrtMemAsyncCopy(aiMemDetailsDev_.ptr(), aiMemDetailsDev_.size(), aiMemDetailsMem_->ptr(),
2830 : aiMemDetailsDev_.size(), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2831 :
2832 0 : CHK_RET(DeviceMem::alloc(aiMemDev_, aiMemMem_->size()));
2833 0 : aiRMAInfoPtr->memPtr = aiMemDev_.ptr();
2834 0 : CHK_RET(hrtMemAsyncCopy(aiMemDev_.ptr(), aiMemDev_.size(), aiMemMem_->ptr(), aiMemDev_.size(),
2835 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2836 :
2837 0 : combinOparaPtr->sizeOfAiRMAInfo = static_cast<u64>(sizeof(HcclAiRMAInfo));
2838 0 : CHK_RET(DeviceMem::alloc(aiRMAInfoDev_, combinOparaPtr->sizeOfAiRMAInfo));
2839 0 : combinOparaPtr->aiRMAInfo = aiRMAInfoDev_.ptr();
2840 0 : CHK_RET(hrtMemAsyncCopy(aiRMAInfoDev_.ptr(), aiRMAInfoDev_.size(), aiRMAInfoMem_->ptr(), aiRMAInfoDev_.size(),
2841 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2842 :
2843 0 : HCCL_INFO("[%s] tag[%s] curRankId[%u] rankNum[%u] qpNum[%u] aiRMAInfo[%p] sizeOfAiRMAInfo[%llu] "
2844 : "sizeOfAiRMAWQ[%u] sizeOfAiRMACQ[%u] sizeOfAiRMAMem[%u] sqPtr[%p] sqSize[%llu] sqCount[%zu] "
2845 : "scqPtr[%p] scqSize[%llu] scqCount[%zu] rqPtr[%p] rqSize[%llu] rqCount[%zu] rcqPtr[%p] "
2846 : "rcqSize[%llu] rcqCount[%zu] memPtr[%p] memSize[%llu] memCount[%zu] memDetailCount[%zu]",
2847 : __func__, tag.c_str(), aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum,
2848 : combinOparaPtr->aiRMAInfo, combinOparaPtr->sizeOfAiRMAInfo, aiRMAInfoPtr->sizeOfAiRMAWQ,
2849 : aiRMAInfoPtr->sizeOfAiRMACQ, aiRMAInfoPtr->sizeOfAiRMAMem, aiRMAInfoPtr->sqPtr,
2850 : aiSqDev_.size(), aiSqMem_->size(), aiRMAInfoPtr->scqPtr, aiScqDev_.size(), aiScqMem_->size(),
2851 : aiRMAInfoPtr->rqPtr, aiRqDev_.size(), aiRqMem_->size(), aiRMAInfoPtr->rcqPtr, aiRcqDev_.size(),
2852 : aiRcqMem_->size(), aiRMAInfoPtr->memPtr, aiMemDev_.size(), aiMemMem_->size(), aiMemDetailsMem_->size());
2853 :
2854 0 : return HCCL_SUCCESS;
2855 : }
2856 :
2857 0 : HcclResult HcclCommunicator::H2DAiRMAInfoV2(const std::string &tag, rtStream_t aiCpuStream)
2858 : {
2859 0 : CHK_PTR_NULL(rmaInfoMem_);
2860 0 : HcclRMAInfo *rmaInfoPtr = reinterpret_cast<HcclRMAInfo*>(rmaInfoMem_->ptr());
2861 0 : CHK_PTR_NULL(rmaInfoPtr);
2862 :
2863 0 : CHK_PTR_NULL(combinOparaMem_);
2864 0 : HcclCombinOpParam *combinOparaPtr = reinterpret_cast<HcclCombinOpParam*>(combinOparaMem_->ptr());
2865 0 : CHK_PTR_NULL(combinOparaPtr);
2866 :
2867 0 : rmaInfoPtr->sizeOfRMAWQ = static_cast<u32>(sizeof(HcclAiRMAWQ));
2868 0 : rmaInfoPtr->sizeOfRMACQ = static_cast<u32>(sizeof(HcclAiRMACQ));
2869 0 : rmaInfoPtr->sizeOfRMAMem = static_cast<u32>(sizeof(HcclAiRMAMemInfo));
2870 :
2871 0 : CHK_RET(DeviceMem::alloc(aiSqDev_, aiSqMem_->size()));
2872 0 : rmaInfoPtr->sqPtr = reinterpret_cast<uintptr_t>(aiSqDev_.ptr());
2873 0 : CHK_RET(hrtMemAsyncCopy(aiSqDev_.ptr(), aiSqDev_.size(), aiSqMem_->ptr(), aiSqDev_.size(),
2874 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2875 :
2876 0 : CHK_RET(DeviceMem::alloc(aiScqDev_, aiScqMem_->size()));
2877 0 : rmaInfoPtr->scqPtr = reinterpret_cast<uintptr_t>(aiScqDev_.ptr());
2878 0 : CHK_RET(hrtMemAsyncCopy(aiScqDev_.ptr(), aiScqDev_.size(), aiScqMem_->ptr(), aiScqDev_.size(),
2879 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2880 :
2881 0 : CHK_RET(DeviceMem::alloc(aiRqDev_, aiRqMem_->size()));
2882 0 : rmaInfoPtr->rqPtr = reinterpret_cast<uintptr_t>(aiRqDev_.ptr());
2883 0 : CHK_RET(hrtMemAsyncCopy(aiRqDev_.ptr(), aiRqDev_.size(), aiRqMem_->ptr(), aiRqDev_.size(),
2884 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2885 :
2886 0 : CHK_RET(DeviceMem::alloc(aiRcqDev_, aiRcqMem_->size()));
2887 0 : rmaInfoPtr->rcqPtr = reinterpret_cast<uintptr_t>(aiRcqDev_.ptr());
2888 0 : CHK_RET(hrtMemAsyncCopy(aiRcqDev_.ptr(), aiRcqDev_.size(), aiRcqMem_->ptr(), aiRcqDev_.size(),
2889 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2890 :
2891 0 : CHK_RET(hrtMemAsyncCopy(aiMemDetailsDev_.ptr(), aiMemDetailsDev_.size(), aiMemDetailsMem_->ptr(),
2892 : aiMemDetailsDev_.size(), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2893 :
2894 0 : CHK_RET(DeviceMem::alloc(aiMemDev_, aiMemMem_->size()));
2895 0 : rmaInfoPtr->memPtr = reinterpret_cast<uintptr_t>(aiMemDev_.ptr());
2896 0 : CHK_RET(hrtMemAsyncCopy(aiMemDev_.ptr(), aiMemDev_.size(), aiMemMem_->ptr(), aiMemDev_.size(),
2897 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2898 :
2899 0 : combinOparaPtr->sizeOfAiRMAInfo = static_cast<u64>(sizeof(HcclAiRMAInfo));
2900 0 : CHK_RET(DeviceMem::alloc(aiRMAInfoDev_, combinOparaPtr->sizeOfAiRMAInfo));
2901 0 : combinOparaPtr->aiRMAInfo = aiRMAInfoDev_.ptr();
2902 0 : CHK_RET(hrtMemAsyncCopy(aiRMAInfoDev_.ptr(), aiRMAInfoDev_.size(), rmaInfoMem_->ptr(), aiRMAInfoDev_.size(),
2903 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2904 :
2905 0 : HCCL_INFO("[%s] tag[%s] curRankId[%u] rankNum[%u] qpNum[%u] aiRMAInfo[%p] sizeOfAiRMAInfo[%llu] "
2906 : "sizeOfAiRMAWQ[%u] sizeOfAiRMACQ[%u] sizeOfAiRMAMem[%u] sqPtr[%p] sqSize[%llu] sqCount[%zu] "
2907 : "scqPtr[%p] scqSize[%llu] scqCount[%zu] rqPtr[%p] rqSize[%llu] rqCount[%zu] rcqPtr[%p] "
2908 : "rcqSize[%llu] rcqCount[%zu] memPtr[%p] memSize[%llu] memCount[%zu] memDetailCount[%zu]",
2909 : __func__, tag.c_str(), rmaInfoPtr->curRankId, rmaInfoPtr->rankNum, rmaInfoPtr->qpNum,
2910 : combinOparaPtr->aiRMAInfo, combinOparaPtr->sizeOfAiRMAInfo, rmaInfoPtr->sizeOfRMAWQ,
2911 : rmaInfoPtr->sizeOfRMACQ, rmaInfoPtr->sizeOfRMAMem, rmaInfoPtr->sqPtr,
2912 : aiSqDev_.size(), aiSqMem_->size(), rmaInfoPtr->scqPtr, aiScqDev_.size(), aiScqMem_->size(),
2913 : rmaInfoPtr->rqPtr, aiRqDev_.size(), aiRqMem_->size(), rmaInfoPtr->rcqPtr, aiRcqDev_.size(),
2914 : aiRcqMem_->size(), rmaInfoPtr->memPtr, aiMemDev_.size(), aiMemMem_->size(), aiMemDetailsMem_->size());
2915 :
2916 0 : return HCCL_SUCCESS;
2917 : }
2918 :
2919 0 : HcclResult HcclCommunicator::GetAIVNormalQPInfo(CommBase *comm, const std::string &tag)
2920 : {
2921 : // 获取 Transport QP 数量
2922 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2923 0 : HcclAiRMAInfo *aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2924 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2925 :
2926 0 : aiRMAInfoPtr->qpNum = HCCL_QPS_PER_CONNECTION_DEFAULT;
2927 0 : for (u32 i = 0; i < aiRMAInfoPtr->rankNum; i++)
2928 : {
2929 0 : if (i != aiRMAInfoPtr->curRankId)
2930 : {
2931 0 : const auto transport = comm->GetTransportByRank(i);
2932 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP)
2933 : {
2934 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2935 0 : CHK_RET(transport->GetAiRMAQueueInfo(aiQpVec));
2936 0 : aiRMAInfoPtr->qpNum = static_cast<u32>(aiQpVec.size());
2937 0 : }
2938 0 : }
2939 : }
2940 :
2941 0 : CHK_PRT_RET(aiRMAInfoPtr->qpNum <= 0,
2942 : HCCL_ERROR("[%s] invalid qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u]",
2943 : __func__, tag.c_str(), aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum),
2944 : HCCL_E_INTERNAL);
2945 :
2946 0 : return HCCL_SUCCESS;
2947 : }
2948 :
2949 0 : HcclResult HcclCommunicator::GetAIVNormalQPInfoV2(std::vector<LINK>& links, const std::string &tag)
2950 : {
2951 : // 获取 Transport QP 数量
2952 0 : CHK_PTR_NULL(rmaInfoMem_);
2953 0 : HcclRMAInfo *rmaInfoPtr = reinterpret_cast<HcclRMAInfo*>(rmaInfoMem_->ptr());
2954 0 : CHK_PTR_NULL(rmaInfoPtr);
2955 : // 获取 Transport QP 数量（暂时只支持单QP）
2956 0 : rmaInfoPtr->qpNum = HCCL_QPS_PER_CONNECTION_DEFAULT;
2957 0 : for (u32 i = 0; i < links.size(); i++) {
2958 0 : if (i != rmaInfoPtr->curRankId) {
2959 0 : if (links[i]->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2960 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2961 0 : CHK_RET(links[i]->GetAiRMAQueueInfo(aiQpVec));
2962 0 : rmaInfoPtr->qpNum = static_cast<u32>(aiQpVec.size());
2963 0 : break;
2964 0 : }
2965 : }
2966 : }
2967 :
2968 0 : CHK_PRT_RET(rmaInfoPtr->qpNum <= 0,
2969 : HCCL_ERROR("[%s] invalid qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u]", __func__,
2970 : tag.c_str(), rmaInfoPtr->curRankId, rmaInfoPtr->rankNum, rmaInfoPtr->qpNum),
2971 : HCCL_E_INTERNAL);
2972 :
2973 0 : return HCCL_SUCCESS;
2974 : }
2975 :
2976 : template<typename T>
2977 0 : HcclResult HcclCommunicator::GenIbvAiRMAInfo(u32 rankid, const std::shared_ptr<Transport> &transport,
2978 : const std::string &tag, T* aiRMAInfoPtr)
2979 : {
2980 0 : HCCL_INFO("[HcclCommunicator][%s] Start prepare.", __func__);
2981 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2982 0 : CHK_RET(transport->GetAiRMAQueueInfo(aiQpVec));
2983 :
2984 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2985 0 : CHK_PRT_RET(aiQpVec.size() != aiRMAInfoPtr->qpNum,
2986 : HCCL_ERROR("[%s] different qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u] qpVecNum[%u]",
2987 : __func__, tag.c_str(), aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum,
2988 : aiRMAInfoPtr->qpNum, aiQpVec.size()),
2989 : HCCL_E_INTERNAL);
2990 :
2991 0 : HcclAiRMAWQ *aiSqHost = reinterpret_cast<HcclAiRMAWQ*>(aiSqMem_->ptr());
2992 0 : HcclAiRMACQ *aiScqHost = reinterpret_cast<HcclAiRMACQ*>(aiScqMem_->ptr());
2993 0 : HcclAiRMAWQ *aiRqHost = reinterpret_cast<HcclAiRMAWQ*>(aiRqMem_->ptr());
2994 0 : HcclAiRMACQ *aiRcqHost = reinterpret_cast<HcclAiRMACQ*>(aiRcqMem_->ptr());
2995 :
2996 0 : for (u32 j = 0; j < aiRMAInfoPtr->qpNum; j++)
2997 : {
2998 0 : const auto &aiQpInfo = aiQpVec[j];
2999 0 : u32 qpIndex = rankid * aiRMAInfoPtr->qpNum + j;
3000 0 : aiSqHost[qpIndex] = aiQpInfo.sq;
3001 0 : aiScqHost[qpIndex] = aiQpInfo.scq;
3002 0 : aiRqHost[qpIndex] = aiQpInfo.rq;
3003 0 : aiRcqHost[qpIndex] = aiQpInfo.rcq;
3004 : }
3005 0 : return HCCL_SUCCESS;
3006 0 : }
3007 :
3008 302 : HcclResult HcclCommunicator::SetAivCoreLimit(u32 aivCoreLimit)
3009 : {
3010 302 : numBlocks_ = aivCoreLimit;
3011 302 : return HCCL_SUCCESS;
3012 : }
3013 :
3014 0 : HcclResult HcclCommunicator::GetAlgInfo(const std::string &algConfig, const std::string &tag, std::string &algName)
3015 : {
3016 0 : CHK_PRT_RET((ALGCFG_TO_NAME.find(algConfig) == ALGCFG_TO_NAME.end()),
3017 : HCCL_ERROR("[%s] invalid algConfig=[%s]", __func__, algConfig.c_str()),
3018 : HCCL_E_PARA);
3019 :
3020 0 : algName = ALGCFG_TO_NAME[algConfig];
3021 0 : HCCL_INFO("[%s] tag=[%s], algName=[%s]",
3022 : __func__, tag.c_str(), algName.c_str());
3023 0 : return HCCL_SUCCESS;
3024 : }
3025 :
3026 0 : HcclResult HcclCommunicator::SetGroupMode(bool isGroup)
3027 : {
3028 0 : isGroupMode_ = isGroup;
3029 0 : CHK_RET(transportManager_->SetGroupMode(isGroup));
3030 0 : return HCCL_SUCCESS;
3031 : }
3032 :
3033 0 : bool HcclCommunicator::GetGroupMode()
3034 : {
3035 0 : return isGroupMode_;
3036 : }
3037 :
3038 0 : HcclResult HcclCommunicator::GetCommUserMemSize(uint64_t &size)
3039 : {
3040 0 : if (!isUserMemRegisted_ || userMemMap_.empty()) {
3041 0 : HCCL_INFO("[HcclCommunicator][%s] get comm user mem size failed", __func__);
3042 0 : return HCCL_E_NOT_FOUND;
3043 : }
3044 0 : size = userMemMap_.begin()->second->size();
3045 0 : return HCCL_SUCCESS;
3046 : }
3047 :
3048 0 : HcclResult HcclCommunicator::GetAivQPInfoV2(std::vector<LINK>& links, const std::string &tag)
3049 : {
3050 0 : HCCL_DEBUG("[HcclCommunicator][%s] Start prepare.", __func__);
3051 : // 获取 Transport QP 数量
3052 0 : CHK_PTR_NULL(aiRMAInfoMem_);
3053 0 : HcclAiRMAInfo *aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
3054 0 : CHK_PTR_NULL(aiRMAInfoPtr);
3055 : // 获取 Transport QP 数量（暂时只支持单QP）
3056 0 : aiRMAInfoPtr->qpNum = HCCL_QPS_PER_CONNECTION_DEFAULT;
3057 0 : for (u32 i = 0; i < links.size(); i++) { //server num
3058 0 : if (i != (aiRMAInfoPtr->curRankId / meshAggregationRankSize_)) { //判断server
3059 0 : if (links[i]->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
3060 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
3061 0 : CHK_RET(links[i]->GetAiRMAQueueInfo(aiQpVec));
3062 0 : aiRMAInfoPtr->qpNum = static_cast<u32>(aiQpVec.size());
3063 0 : break;
3064 0 : }
3065 : }
3066 : }
3067 0 : CHK_PRT_RET(aiRMAInfoPtr->qpNum <= 0,
3068 : HCCL_ERROR("[%s] invalid qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u]", __func__,
3069 : tag.c_str(), aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum),
3070 : HCCL_E_INTERNAL);
3071 :
3072 0 : return HCCL_SUCCESS;
3073 : }
3074 : }
|
