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