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 364 : 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 1881 : bool HcclCommunicator::IsEnableCustom() const { 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 496 : HcclResult HcclCommunicator::InitRankInfo(const RankTable_t& rankTable)
117 : {
118 496 : CHK_RET(InitTcpMode(rankTable));
119 498 : SetAttrs();
120 498 : localRank_ = attrCollector_.GetLocalRank();
121 498 : deviceLogicId_ = attrCollector_.GetDeviceLogicId();
122 : // 按通信域配置是否使用算子级重执行
123 498 : 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 994 : SetRetryEnable(
127 498 : deviceType_, superPodNum_, serverNum_, deviceNumPerAggregation_, isDiffDeviceType_, isAivMode, serverIp,
128 498 : localIp, retryEnable_, commConfig_.GetConfigInterServerRetryEnable(),
129 497 : 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 1483 : nicRanksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT
153 0 : || rankInfo.deviceInfo.port < SINGLE_PROCESS_MIN_PORT
154 1483 : || 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 523 : 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 523 : CHK_SMART_PTR_NULL(queueNotifyManager_);
236 524 : CHK_RET(queueNotifyManager_->Init());
237 524 : queueNotifyManagerRefac_.reset(new (std::nothrow) QueueNotifyManager());
238 523 : CHK_SMART_PTR_NULL(queueNotifyManagerRefac_);
239 523 : 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 523 : CHK_SMART_PTR_NULL(opStreamManager_);
296 523 : 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 529 : void HcclCommunicator::SetAttrs()
364 : {
365 529 : 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 532 : deviceNumPerAggregation_ = attrCollector_.GetDeviceNumPerAggregation();
376 531 : deviceNumPerServer_ = attrCollector_.GetDeviceNumPerServer();
377 531 : isHaveCpuRank_ = attrCollector_.GetHaveCpuRank();
378 : // TransformRankInfoByServerId
379 531 : servRankInfo_ = attrCollector_.GetServRankInfo();
380 : // GetModuleInfo
381 530 : isDiffDeviceModule_ = attrCollector_.GetDiffDeviceModule();
382 532 : isDiffDeviceType_ = attrCollector_.GetDiffDeviceType();
383 532 : gcdDeviceNumPerAggregation_ = attrCollector_.GetGcdDeviceNumPerAggregation();
384 530 : moduleNum_ = attrCollector_.GetModuleNum();
385 531 : multiModuleDiffDeviceNumMode_ = attrCollector_.GetMultiModuleDiffDeviceNumMode();
386 532 : multiSuperPodDiffServerNumMode_ = attrCollector_.GetMultiSuperPodDiffServerNumMode();
387 532 : 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 531 : hccsPortNum_ = attrCollector_.GetHccsPortNum();
398 532 : attrCollector_.GetPairLinkCounter(pairLinkCounter_);
399 531 : attrCollector_.GetPairLinkInfo(pairLinkInfo_);
400 : // SetInterModeInSuperPod
401 532 : isUsedInterHccsMode_ = attrCollector_.GetUsedInterHccsMode();
402 : // GetRankInfoList
403 532 : rankInfoList_ = attrCollector_.GetRankInfoList();
404 : // Localinfo
405 531 : devIpAddr_ = attrCollector_.GetDevIpAddr();
406 531 : devBackupIpAddr_ = attrCollector_.GetDevBackupIpAddr();
407 530 : devBackupPort_ = attrCollector_.GetBackupDevPort();
408 532 : devBackupPort_ = devBackupPort_ == HCCL_INVALID_PORT ? AICPU_RETRY_BACKUP_PORT : devBackupPort_;
409 532 : devicePhyId_ = attrCollector_.GetDevicePhyId();
410 532 : hostIp_ = attrCollector_.GetHostIp();
411 531 : hostPort_ = attrCollector_.GetHostPort();
412 :
413 532 : interServer_ = attrCollector_.GetInterServe();
414 531 : nicDeployment_ = attrCollector_.GetNicDeployment();
415 532 : }
416 :
417 76 : 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() const { return ranktableCrc_; }
493 :
494 0 : u32 HcclCommunicator::GetServerNum() const { return serverNum_; }
495 :
496 0 : u32 HcclCommunicator::GetRealUserRank() const { return realUserRank_; }
497 :
498 1 : u32 HcclCommunicator::GetModuleNum() const { 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 85 : }
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 603 : HcclResult HcclCommunicator::ReleasePreemptSocket()
625 : {
626 603 : 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 602 : 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 602 : 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 602 : HCCL_INFO(
674 : "[HcclCommunicator][ReleasePreemptSocket] release all preempt socket success, comm id[%s].",
675 : identifier_.c_str());
676 :
677 601 : 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 recovering, "
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 811 : HcclResult HcclCommunicator::UnRegisterBackGroundThread()
703 : {
704 811 : CHK_PRT_RET(
705 : isInvalidComm_,
706 : HCCL_ERROR(
707 : "[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recovering, "
708 : "this comm is invalid.",
709 : __func__, identifier_.c_str(), userRank_, deviceLogicId_),
710 : HCCL_E_UNAVAIL);
711 811 : CHK_RET(UnRegisterBackGroundThread(kfcControlTransferH2D_, kfcStatusTransferD2H_));
712 811 : if (IsEnableCustom()) {
713 0 : CHK_RET(UnRegisterBackGroundThread(customControlTransferH2D_, customStatusTransferD2H_));
714 : }
715 811 : return HCCL_SUCCESS;
716 : }
717 :
718 811 : HcclResult HcclCommunicator::UnRegisterBackGroundThread(
719 : std::shared_ptr<HDCommunicate>& controlH2D, std::shared_ptr<HDCommunicate>& statusD2H)
720 : {
721 811 : HCCL_INFO("start to stop the backGround Thread");
722 811 : if (deviceType_ == DevType::DEV_TYPE_910 || (deviceType_ == DevType::DEV_TYPE_910B && !GetAicpuUnfoldFlag())) {
723 491 : if (GetMC2EnvFlag()) {
724 0 : if (controlH2D != nullptr) {
725 1 : BackgroundCommand request = BackgroundCommand::kStop;
726 1 : 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 811 : return HCCL_SUCCESS;
748 : }
749 :
750 811 : HcclResult HcclCommunicator::DestroyAicpuComm()
751 : {
752 811 : CHK_PRT_RET(
753 : isInvalidComm_,
754 : HCCL_ERROR(
755 : "[HcclCommunicator][%s] comm[%s], rank[%u], devId[%d], snapshot recovering, "
756 : "this comm is invalid.",
757 : __func__, identifier_.c_str(), userRank_, deviceLogicId_),
758 : HCCL_E_UNAVAIL);
759 811 : CHK_RET(DestroyAicpuComm(kfcControlTransferH2D_, kfcStatusTransferD2H_));
760 811 : if (IsEnableCustom()) {
761 0 : CHK_RET(DestroyAicpuComm(customControlTransferH2D_, customStatusTransferD2H_));
762 : }
763 811 : return HCCL_SUCCESS;
764 : }
765 :
766 811 : HcclResult HcclCommunicator::DestroyAicpuComm(
767 : std::shared_ptr<HDCommunicate>& controlH2D, std::shared_ptr<HDCommunicate>& statusD2H)
768 : {
769 811 : HCCL_INFO("[HcclCommunicator][%s]start to destroy the aicpu comm, group[%s].", __func__, identifier_.c_str());
770 765 : if (deviceType_ != DevType::DEV_TYPE_910_93 && !(deviceType_ == DevType::DEV_TYPE_910B && GetAicpuUnfoldFlag())
771 1576 : && (myRankConnectMode_ == 0)) {
772 765 : HCCL_INFO("[HcclCommunicator][%s]Device type[%d] no needs to destroy the aicpu comm.", __func__, deviceType_);
773 765 : 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 945 : 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 945 : } else if (serverId_ == rankInfo.serverId) {
835 702 : if (rankInfo.devicePhyId != HOST_DEVICE_ID) {
836 702 : 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() const { return deviceType_; }
889 :
890 0 : u32 HcclCommunicator::NslbGetServerNum() const { 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, const 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() const { return isNsRecovery_; }
982 :
983 12 : bool HcclCommunicator::GetAicpuCommEngine() const { 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 346 : bool HcclCommunicator::GetAicpuUnfoldFlag() const { 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() const { 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 recovering, "
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 recovering, "
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 5 : && (opType == HcclCMDType::HCCL_CMD_ALLTOALL || opType == HcclCMDType::HCCL_CMD_ALLTOALLV
1309 5 : || opType == HcclCMDType::HCCL_CMD_ALLTOALLVC || opParam.isZeroCopy)) {
1310 0 : return true;
1311 : }
1312 :
1313 62 : return false;
1314 : }
1315 :
1316 48 : HcclResult HcclCommunicator::AllocOpBaseModeScratchMem(
1317 : HcclCMDType opType, const OpParam& opParam, AlgResourceRequest& resRequest, AlgResourceResponse& algResResponse)
1318 : {
1319 48 : if (resRequest.scratchMemSize == 0) {
1320 43 : return HCCL_SUCCESS;
1321 : }
1322 :
1323 5 : 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 5 : if (opType == HcclCMDType::HCCL_CMD_REDUCE_SCATTER) {
1339 8 : DeviceMem tmpBuffer;
1340 8 : 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 2 : HcclResult HcclCommunicator::AicpuInitOpTilingDataFromOpParam(
1521 : const OpParam& opParam, const HcclCMDType& opType, struct OpTilingData* opTilingData)
1522 : {
1523 2 : opTilingData->workflowMode = (IsForceAicpuOpBaseMode(opParam, opType) && !opParam.isZeroCopy) ?
1524 : static_cast<u8>(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) :
1525 2 : static_cast<u8>(GetWorkflowMode());
1526 2 : opTilingData->inputPtr = reinterpret_cast<u64>(opParam.inputPtr);
1527 2 : opTilingData->outputPtr = reinterpret_cast<u64>(opParam.outputPtr);
1528 2 : opTilingData->reduceType = static_cast<u8>(opParam.reduceType);
1529 2 : opTilingData->syncMode = static_cast<u8>(opParam.syncMode);
1530 2 : opTilingData->root = opParam.root;
1531 2 : opTilingData->dstRank = opParam.dstRank;
1532 2 : opTilingData->srcRank = opParam.srcRank;
1533 2 : opTilingData->opType = static_cast<u8>(opType);
1534 2 : opTilingData->inplaceSupportRetry = static_cast<u8>(inplaceSupportRetry_);
1535 2 : opTilingData->retryEnable = static_cast<u8>(retryEnable_);
1536 2 : opTilingData->inPlaceSupportRetryStatus = static_cast<u8>(inPlaceSupportRetryStatus_);
1537 2 : opTilingData->isInplacePreSync = static_cast<u8>(isInplacePreSync_);
1538 2 : opTilingData->isPostSync = static_cast<u8>(isPostSync_);
1539 2 : opTilingData->userStreamId = opParam.stream.id();
1540 2 : opTilingData->inputSymWindow = reinterpret_cast<u64>(opParam.inputSymWindow);
1541 2 : opTilingData->inputOffset = opParam.inputOffset;
1542 2 : opTilingData->outputSymWindow = reinterpret_cast<u64>(opParam.outputSymWindow);
1543 2 : opTilingData->outputOffset = opParam.outputOffset;
1544 2 : opTilingData->aicpuUnfoldMode = static_cast<u8>(opParam.aicpuUnfoldMode);
1545 2 : return HCCL_SUCCESS;
1546 : }
1547 :
1548 1 : HcclResult HcclCommunicator::KernelLaunchChooseAicpuOrCustom(
1549 : void* inputPtr, void* outputPtr, const rtStream_t stm, u64 addr, void* tilingDataPtr, u32 tilingDataSize,
1550 : const std::string& kernelName, HcclWorkflowMode mode, const std::string& tag, bool isCustom)
1551 : {
1552 1 : return AicpuUnfoldKernelLaunchV2(
1553 1 : inputPtr, outputPtr, stm, addr, tilingDataPtr, tilingDataSize, kernelName, mode, tag, isCustom);
1554 : }
1555 :
1556 633 : HcclResult HcclCommunicator::SaveTraceInfo(std::string& logInfo)
1557 : {
1558 633 : opBaseAtraceInfo_->SaveTraceInfo(logInfo, AtraceOption::Opbasekey);
1559 633 : return HCCL_SUCCESS;
1560 : }
1561 :
1562 0 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>> HcclCommunicator::GetPhyIdNicInfo()
1563 : {
1564 0 : return rankDevicePhyIdNicInfoMap_;
1565 : }
1566 :
1567 1 : vector<u32> HcclCommunicator::GetRanksPort() { return nicRanksPort_; }
1568 :
1569 0 : vector<RankInfo> HcclCommunicator::GetRanksList() { return rankInfoList_; }
1570 :
1571 292 : std::string HcclCommunicator::GetUniqueId(void)
1572 : {
1573 : static std::atomic<u32> idCounter(0);
1574 :
1575 292 : std::string uniqueId("");
1576 292 : uniqueId += std::to_string(SalGetPid());
1577 292 : uniqueId += '-';
1578 292 : uniqueId += std::to_string(idCounter.fetch_add(1));
1579 292 : uniqueId += '-';
1580 292 : uniqueId += std::to_string(SalGetSysTime());
1581 :
1582 292 : return uniqueId;
1583 0 : }
1584 :
1585 1 : u8 HcclCommunicator::GetDeterministicConfig() const
1586 : {
1587 1 : CHK_SMART_PTR_NULL(implAlg_);
1588 1 : return implAlg_->GetDeterministicConfig();
1589 : }
1590 :
1591 235 : HcclResult HcclCommunicator::SetDeterministicConfig(const u8 deterministic)
1592 : {
1593 235 : CHK_SMART_PTR_NULL(implAlg_);
1594 235 : CHK_RET(implAlg_->SetDeterministicConfig(deterministic));
1595 235 : return HCCL_SUCCESS;
1596 : }
1597 :
1598 0 : HcclResult HcclCommunicator::MigrateLinkToStopOrResume(LINK& link, bool isStop)
1599 : {
1600 0 : if (isStop) {
1601 0 : return link->Stop();
1602 : }
1603 0 : return link->Resume();
1604 : }
1605 :
1606 0 : HcclResult HcclCommunicator::MigrateLinkVectorToStopOrResume(const std::vector<LINK>& links, bool isStop)
1607 : {
1608 0 : for (auto it : links) {
1609 0 : if (it) {
1610 0 : CHK_RET(MigrateLinkToStopOrResume(it, isStop));
1611 : }
1612 0 : }
1613 0 : return HCCL_SUCCESS;
1614 : }
1615 :
1616 0 : HcclResult HcclCommunicator::TraverseLinkVector(std::vector<std::unique_ptr<CommBase>>& commBaseVector, bool isStop)
1617 : {
1618 0 : for (unsigned int i = 0; i < commBaseVector.size(); i++) {
1619 0 : auto commBase = commBaseVector[i].get();
1620 0 : if (commBase == nullptr) {
1621 0 : continue;
1622 : }
1623 0 : const std::vector<LINK>& ret = commBase->TransportInfo();
1624 0 : CHK_RET(MigrateLinkVectorToStopOrResume(ret, isStop));
1625 : }
1626 0 : return HCCL_SUCCESS;
1627 : }
1628 :
1629 0 : HcclResult HcclCommunicator::TraverseSingleSubCommTransport(SingleSubCommTransport& commTransport, bool isStop)
1630 : {
1631 0 : for (unsigned int i = 0; i < commTransport.transportRequests.size(); i++) {
1632 0 : if (!commTransport.transportRequests[i].isValid) {
1633 0 : continue;
1634 : }
1635 0 : if (commTransport.links[i] == nullptr) {
1636 0 : continue;
1637 : }
1638 :
1639 0 : if (isStop) {
1640 0 : CHK_RET(commTransport.links[i]->Stop());
1641 : } else {
1642 0 : CHK_RET(commTransport.links[i]->Resume());
1643 : }
1644 : }
1645 0 : return HCCL_SUCCESS;
1646 : }
1647 :
1648 0 : HcclResult HcclCommunicator::TraverseLevelNSubCommTransport(LevelNSubCommTransport& levelNSubCommTransport, bool isStop)
1649 : {
1650 0 : for (unsigned int jj = 0; jj < levelNSubCommTransport.size(); jj++) {
1651 0 : CHK_RET(TraverseSingleSubCommTransport(levelNSubCommTransport[jj], isStop));
1652 : }
1653 0 : return HCCL_SUCCESS;
1654 : }
1655 :
1656 0 : HcclResult HcclCommunicator::TraverseOpCommTransport(OpCommTransport& opCommTransport, bool isStop)
1657 : {
1658 0 : for (unsigned int ii = 0; ii < opCommTransport.size(); ii++) {
1659 0 : CHK_RET(TraverseLevelNSubCommTransport(opCommTransport[ii], isStop));
1660 : }
1661 0 : return HCCL_SUCCESS;
1662 : }
1663 :
1664 7 : HcclResult HcclCommunicator::TraverseAlgResourceResponse(bool isStop)
1665 : {
1666 7 : for (auto& it : resMap_) {
1667 0 : CHK_RET(TraverseOpCommTransport(it.second.opTransportResponse, isStop));
1668 : }
1669 7 : return HCCL_SUCCESS;
1670 : }
1671 :
1672 0 : HcclResult HcclCommunicator::ResetNotify()
1673 : {
1674 0 : CHK_SMART_PTR_NULL(notifyPool_);
1675 0 : CHK_SMART_PTR_NULL(queueNotifyManagerRefac_);
1676 0 : notifyPool_->ResetNotify();
1677 0 : queueNotifyManagerRefac_->ResetNotify();
1678 0 : return HCCL_SUCCESS;
1679 : }
1680 :
1681 0 : HcclResult HcclCommunicator::ResetNotifyForDestRank(s64 destRank)
1682 : {
1683 0 : CHK_SMART_PTR_NULL(notifyPool_);
1684 0 : CHK_SMART_PTR_NULL(queueNotifyManagerRefac_);
1685 0 : notifyPool_->ResetNotifyForDestRank(destRank);
1686 0 : return HCCL_SUCCESS;
1687 : }
1688 :
1689 79 : void HcclCommunicator::InsertNewTagToTagMap(std::string& newTag, std::string& tag)
1690 : {
1691 79 : const auto& mapIt = newTagToTagMap_.find(newTag);
1692 77 : if (mapIt == newTagToTagMap_.end()) {
1693 78 : newTagToTagMap_.insert({newTag, tag});
1694 : } else {
1695 1 : mapIt->second = tag;
1696 : }
1697 158 : return;
1698 : }
1699 :
1700 0 : HcclResult HcclCommunicator::GetTagFromNewTag(const std::string& newTag, std::string& tag)
1701 : {
1702 0 : const auto& mapIt = newTagToTagMap_.find(newTag);
1703 0 : if (mapIt == newTagToTagMap_.end()) {
1704 0 : HCCL_ERROR("[OpRetry]newTag[%s] is not in newTagToTagMap_", newTag.c_str());
1705 0 : return HCCL_E_INTERNAL;
1706 : } else {
1707 0 : tag = mapIt->second;
1708 : }
1709 0 : return HCCL_SUCCESS;
1710 : }
1711 :
1712 : HcclResult
1713 0 : HcclCommunicator::SetSignalTransport(SingleSubCommTransport& singleSubCommTransport, u32 linkIdx, bool statusStop)
1714 : {
1715 0 : RankId loc = singleSubCommTransport.transportRequests[linkIdx].localUserRank;
1716 0 : RankId rmt = singleSubCommTransport.transportRequests[linkIdx].remoteUserRank;
1717 0 : if (statusStop) {
1718 0 : if (singleSubCommTransport.links[linkIdx]
1719 0 : && singleSubCommTransport.links[linkIdx]->GetLinkType() == LinkType::LINK_ROCE) {
1720 0 : CHK_RET(singleSubCommTransport.links[linkIdx]->Stop());
1721 0 : singleSubCommTransport.status[linkIdx] = TransportStatus::STOP;
1722 0 : HCCL_INFO("[SetTransportStatus]set transport status to stop, loc[%u], rmt[%u]", loc, rmt);
1723 : }
1724 : } else {
1725 0 : if (singleSubCommTransport.links[linkIdx] && singleSubCommTransport.status[linkIdx] == TransportStatus::STOP) {
1726 0 : HCCL_INFO("[SetTransportStatus]set transport status to resume, loc[%u], rmt[%u]", loc, rmt);
1727 0 : CHK_RET(singleSubCommTransport.links[linkIdx]->DeInit());
1728 0 : singleSubCommTransport.links[linkIdx] = nullptr; // 赋值为nullptr, 供后面重新建链
1729 0 : singleSubCommTransport.status[linkIdx] = TransportStatus::INIT;
1730 : }
1731 : }
1732 0 : return HCCL_SUCCESS;
1733 : }
1734 :
1735 0 : HcclResult HcclCommunicator::SetBsrTransportStatusImpl(
1736 : OpCommTransport& opCommTransport, bool statusStop, const HcclOpIdentifier& opId, u32 remoteRank)
1737 : {
1738 0 : u32 commIndex = 0;
1739 0 : if ((userRank_ == opId.detRank && remoteRank > userRank_)
1740 0 : || (userRank_ == opId.srcRank && remoteRank < userRank_)) {
1741 0 : commIndex = COMM_INDEX_0;
1742 : } else {
1743 0 : commIndex = COMM_INDEX_1;
1744 : }
1745 0 : CHK_PRT_RET(
1746 : commIndex >= opCommTransport[COMM_COMBINE_ORDER].size(),
1747 : HCCL_ERROR(
1748 : "[SetBsrTransportStatusImpl] batchsendrecv op commIndex[%u] is larger than "
1749 : "opTransportResponse size[%zu]",
1750 : remoteRank, opCommTransport[COMM_COMBINE_ORDER].size()),
1751 : HCCL_E_PARA);
1752 : SingleSubCommTransport& commCombined
1753 0 : = const_cast<SingleSubCommTransport&>(opCommTransport[COMM_COMBINE_ORDER][commIndex]);
1754 0 : u32 Rank = commCombined.userRank2subCommRank[remoteRank];
1755 0 : CHK_PRT_RET(
1756 : Rank >= commCombined.links.size(),
1757 : HCCL_ERROR(
1758 : "[SetBsrTransportStatusImpl] batchsendrecv op remoteRank[%u], get Rank[%u],"
1759 : "the size of combinedComm links is [%zu]",
1760 : remoteRank, Rank, commCombined.links.size()),
1761 : HCCL_E_PARA);
1762 0 : CHK_SMART_PTR_NULL(commCombined.links[Rank]);
1763 :
1764 0 : RankId loc = commCombined.transportRequests[Rank].localUserRank;
1765 0 : RankId rmt = commCombined.transportRequests[Rank].remoteUserRank;
1766 0 : if (!commCombined.transportRequests[Rank].isValid) {
1767 0 : return HCCL_SUCCESS;
1768 : }
1769 0 : if (statusStop) {
1770 0 : if (commCombined.links[Rank]->GetLinkType() == LinkType::LINK_ROCE) {
1771 0 : CHK_RET(commCombined.links[Rank]->Stop());
1772 0 : commCombined.status[Rank] = TransportStatus::STOP;
1773 0 : HCCL_INFO(
1774 : "[SetBsrTransportStatusImpl]set bsr transport status to stop, comindex[%u] loc[%u], rmt[%u]", commIndex,
1775 : loc, rmt);
1776 : }
1777 : } else {
1778 0 : if (commCombined.status[Rank] == TransportStatus::STOP) {
1779 0 : HCCL_INFO(
1780 : "[SetBsrTransportStatusImpl]set bsr transport status to resume, comindex[%u] loc[%u], rmt[%u]",
1781 : commIndex, loc, rmt);
1782 0 : CHK_RET(commCombined.links[Rank]->DeInit());
1783 0 : commCombined.links[Rank] = nullptr; // 赋值为nullptr, 供后面重新建链
1784 0 : commCombined.status[Rank] = TransportStatus::INIT;
1785 : }
1786 : }
1787 0 : return HCCL_SUCCESS;
1788 : }
1789 :
1790 0 : HcclResult HcclCommunicator::SetBsrTransportStatusImplforchange(
1791 : OpCommTransport& opCommTransport, const HcclOpIdentifier& opId, u32 remoteRank,
1792 : const std::map<u32, bool>& remoteRankPortMap, bool isUseDefault, const std::map<u32, bool>& isChangeLinkMap,
1793 : bool isCurTag)
1794 : {
1795 : bool isPortSatisfy
1796 0 : = (remoteRankPortMap.find(remoteRank) != remoteRankPortMap.end()
1797 0 : && remoteRankPortMap.find(remoteRank)->second == isUseDefault);
1798 : bool isChangeLink
1799 0 : = (isChangeLinkMap.find(remoteRank) != isChangeLinkMap.end() && isChangeLinkMap.find(remoteRank)->second);
1800 0 : HCCL_INFO(
1801 : "[SetBsrTransportStatusImplforchange]remoteRank[%u], isUseDefault[%d], "
1802 : "isPortSatisfy[%d], isChangeLink[%d], isCurTag[%d]",
1803 : remoteRank, isUseDefault, isPortSatisfy, isChangeLink, isCurTag);
1804 0 : if (!isPortSatisfy || !(isChangeLink || isCurTag)) {
1805 0 : return HCCL_SUCCESS;
1806 : }
1807 :
1808 0 : u32 commIndex = 0;
1809 0 : if ((userRank_ == opId.detRank && remoteRank > userRank_)
1810 0 : || (userRank_ == opId.srcRank && remoteRank < userRank_)) {
1811 0 : commIndex = COMM_INDEX_0;
1812 : } else {
1813 0 : commIndex = COMM_INDEX_1;
1814 : }
1815 0 : CHK_PRT_RET(
1816 : commIndex >= opCommTransport[COMM_COMBINE_ORDER].size(),
1817 : HCCL_ERROR(
1818 : "[SetBsrTransportStatusImplforchange] batchsendrecv op commIndex[%u] is larger than "
1819 : "opTransportResponse size[%zu]",
1820 : commIndex, opCommTransport[COMM_COMBINE_ORDER].size()),
1821 : HCCL_E_PARA);
1822 : SingleSubCommTransport& commCombined
1823 0 : = static_cast<SingleSubCommTransport&>(opCommTransport[COMM_COMBINE_ORDER][commIndex]);
1824 0 : u32 rank = commCombined.userRank2subCommRank[remoteRank];
1825 0 : CHK_PRT_RET(
1826 : rank >= commCombined.links.size(),
1827 : HCCL_ERROR(
1828 : "[SetBsrTransportStatusImplforchange] batchsendrecv op remoteRank[%u], get Rank[%u],"
1829 : "the size of combinedComm links is [%zu]",
1830 : remoteRank, rank, commCombined.links.size()),
1831 : HCCL_E_PARA);
1832 0 : CHK_SMART_PTR_NULL(commCombined.links[rank]);
1833 :
1834 0 : RankId loc = commCombined.transportRequests[rank].localUserRank;
1835 0 : RankId rmt = commCombined.transportRequests[rank].remoteUserRank;
1836 0 : if (!commCombined.transportRequests[rank].isValid) {
1837 0 : return HCCL_SUCCESS;
1838 : }
1839 :
1840 0 : if (commCombined.status[rank] == TransportStatus::STOP) {
1841 0 : HCCL_INFO(
1842 : "[SetBsrTransportStatusImplforchange]set bsr transport status to resume, comindex[%u] loc[%u], rmt[%u]",
1843 : commIndex, loc, rmt);
1844 0 : CHK_RET(commCombined.links[rank]->DeInit());
1845 0 : commCombined.links[rank] = nullptr; // 赋值为nullptr, 供后面重新建链
1846 0 : commCombined.status[rank] = TransportStatus::INIT;
1847 : }
1848 0 : return HCCL_SUCCESS;
1849 : }
1850 :
1851 0 : HcclResult HcclCommunicator::SetTransportStatusImpl(
1852 : OpCommTransport& opCommTransport, bool statusStop, const HcclOpIdentifier& opId, u32 remoteRank,
1853 : const std::map<u32, bool>& remoteRankPortMap, bool isUseDefault)
1854 : {
1855 0 : bool isSendRecv = opId.isSendRecv;
1856 :
1857 : // stop阶段及原地重执行的resume阶段
1858 : // bsr判断当前故障的send、recv是否remoterank是否相同的情况，如果是相同只操作故障op，如果不同都操作
1859 0 : u32 sendRemoteRank = userRank_ == opId.bsrInfo[HCCL_SEND].detRank ? opId.bsrInfo[HCCL_SEND].srcRank :
1860 : opId.bsrInfo[HCCL_SEND].detRank;
1861 0 : u32 recvRemoteRank = userRank_ == opId.bsrInfo[HCCL_RECV].detRank ? opId.bsrInfo[HCCL_RECV].srcRank :
1862 : opId.bsrInfo[HCCL_RECV].detRank;
1863 : bool isBsrPortSatisfy
1864 0 : = (remoteRankPortMap.find(remoteRank) != remoteRankPortMap.end()
1865 0 : && remoteRankPortMap.find(remoteRank)->second == isUseDefault);
1866 0 : bool isQpnSatify = (opId.bsrInfo[HCCL_RECV].tpQpn != 0) && (opId.bsrInfo[HCCL_SEND].tpQpn != 0);
1867 0 : HCCL_INFO(
1868 : "[SetBsrTransportStatusImpl]SendremoteRank[%u], RecvremoteRank[%u], isUseDefault[%d], isQpnSatisfy[%d], "
1869 : "isBsrPortSatisfy[%d]",
1870 : sendRemoteRank, recvRemoteRank, isUseDefault, isQpnSatify, isBsrPortSatisfy);
1871 0 : if (opId.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV && sendRemoteRank == recvRemoteRank && isBsrPortSatisfy
1872 0 : && isQpnSatify) {
1873 0 : CHK_RET(SetBsrTransportStatusImpl(opCommTransport, statusStop, opId, remoteRank));
1874 0 : return HCCL_SUCCESS;
1875 : }
1876 :
1877 0 : for (auto& levelNSubCommTransport : opCommTransport) {
1878 0 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
1879 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
1880 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
1881 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
1882 0 : bool isRankSatisfy = ((!isSendRecv) || (isSendRecv && remoteRank == transportRemoteRank));
1883 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
1884 : bool isPortSatisfy
1885 0 : = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end()
1886 0 : && remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
1887 0 : HCCL_INFO(
1888 : "[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], isRankSatisfy[%d], "
1889 : "isPortSatisfy[%d]",
1890 : transportRemoteRank, isUseDefault, isValid, isRankSatisfy, isPortSatisfy);
1891 0 : if (isValid && isRankSatisfy && isPortSatisfy) {
1892 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, statusStop));
1893 : }
1894 : }
1895 : }
1896 : }
1897 0 : return HCCL_SUCCESS;
1898 : }
1899 :
1900 0 : HcclResult HcclCommunicator::SetTransportStatusImplForChange(
1901 : OpCommTransport& opCommTransport, const HcclOpIdentifier& opId, u32 remoteRank,
1902 : const std::map<u32, bool>& remoteRankPortMap, bool isUseDefault, const std::map<u32, bool>& isChangeLinkMap,
1903 : bool isCurTag)
1904 : {
1905 0 : bool isSendRecv = opId.isSendRecv;
1906 :
1907 : // bsr判断当前故障的send、recv是否remoterank是否相同的情况，如果是相同只操作故障op，如果不同都操作
1908 0 : u32 sendRemoteRank = userRank_ == opId.bsrInfo[HCCL_SEND].detRank ? opId.bsrInfo[HCCL_SEND].srcRank :
1909 : opId.bsrInfo[HCCL_SEND].detRank;
1910 0 : u32 recvRemoteRank = userRank_ == opId.bsrInfo[HCCL_RECV].detRank ? opId.bsrInfo[HCCL_RECV].srcRank :
1911 : opId.bsrInfo[HCCL_RECV].detRank;
1912 0 : bool isQpnSatify = (opId.bsrInfo[HCCL_RECV].tpQpn != 0) && (opId.bsrInfo[HCCL_SEND].tpQpn != 0);
1913 0 : HCCL_INFO(
1914 : "[SetBsrTransportStatusImpl]SendremoteRank[%u], RecvremoteRank[%u], isUseDefault[%d], isQpnSatisfy[%d]",
1915 : sendRemoteRank, recvRemoteRank, isUseDefault, isQpnSatify);
1916 0 : if (opId.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV && sendRemoteRank == recvRemoteRank && isQpnSatify) {
1917 0 : CHK_RET(SetBsrTransportStatusImplforchange(
1918 : opCommTransport, opId, remoteRank, remoteRankPortMap, isUseDefault, isChangeLinkMap, isCurTag));
1919 0 : return HCCL_SUCCESS;
1920 : }
1921 :
1922 : // 借轨的resume阶段
1923 0 : for (auto& levelNSubCommTransport : opCommTransport) {
1924 0 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
1925 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
1926 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
1927 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
1928 0 : bool isRankSatisfy = (!isSendRecv || (isSendRecv && remoteRank == transportRemoteRank));
1929 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
1930 : bool isPortSatisfy
1931 0 : = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end()
1932 0 : && remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
1933 : bool isChangeLink
1934 0 : = (isChangeLinkMap.find(transportRemoteRank) != isChangeLinkMap.end()
1935 0 : && isChangeLinkMap.find(transportRemoteRank)->second);
1936 0 : HCCL_INFO(
1937 : "[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], isRankSatisfy[%d], "
1938 : "isPortSatisfy[%d], isChangeLink[%d], isCurTag[%d]",
1939 : transportRemoteRank, isUseDefault, isValid, isRankSatisfy, isPortSatisfy, isChangeLink, isCurTag);
1940 0 : if (isValid && isRankSatisfy && isPortSatisfy && (isChangeLink || isCurTag)) {
1941 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, false));
1942 : }
1943 : }
1944 : }
1945 : }
1946 0 : return HCCL_SUCCESS;
1947 : }
1948 :
1949 0 : HcclResult HcclCommunicator::SetTransportStatus(
1950 : const HcclOpIdentifier& opId, bool statusStop, const std::map<u32, bool>& remoteRankPortMap,
1951 : const std::map<u32, bool>& isChangeLinkMap, bool isChangeLinkFlag)
1952 : {
1953 0 : std::string newTag(reinterpret_cast<const char*>(opId.newTag));
1954 0 : u32 remoteRank = userRank_ == opId.detRank ? opId.srcRank : opId.detRank;
1955 :
1956 0 : if (resMap_.find(newTag) == resMap_.end()) {
1957 0 : HCCL_ERROR("HcclCommunicator SetTransportStatus failed: newTag[%s] is not in resMap", newTag.c_str());
1958 0 : return HCCL_E_INTERNAL;
1959 : }
1960 :
1961 0 : if (statusStop) {
1962 0 : CHK_RET(SetTransportStatusImpl(
1963 : resMap_[newTag].opTransportResponse, statusStop, opId, remoteRank, remoteRankPortMap, true));
1964 0 : CHK_RET(SetTransportStatusImpl(
1965 : resMap_[newTag].opTransportResponseBackUp, statusStop, opId, remoteRank, remoteRankPortMap, false));
1966 : } else {
1967 0 : if (isChangeLinkFlag) {
1968 : // 借轨场景
1969 0 : for (auto& resMapIt : resMap_) {
1970 0 : bool isCurTag = false;
1971 0 : if (resMapIt.first == newTag) {
1972 0 : isCurTag = true;
1973 : }
1974 0 : if (hostResMap_.find(resMapIt.first) != hostResMap_.end()) {
1975 : // 若当前tag未进行aicpu展开，则不重新build资源
1976 0 : continue;
1977 : }
1978 :
1979 0 : if ((HcclCMDType::HCCL_CMD_BATCH_SEND_RECV == opId.opType && !isCurTag)
1980 0 : || (HcclCMDType::HCCL_CMD_BATCH_SEND_RECV != opId.opType
1981 0 : && resMapIt.first.find("BatchSendRecv") != std::string::npos)) {
1982 0 : continue;
1983 : }
1984 0 : CHK_RET(SetTransportStatusImplForChange(
1985 : resMapIt.second.opTransportResponse, opId, remoteRank, remoteRankPortMap, true, isChangeLinkMap,
1986 : isCurTag));
1987 0 : CHK_RET(SetTransportStatusImplForChange(
1988 : resMapIt.second.opTransportResponseBackUp, opId, remoteRank, remoteRankPortMap, false,
1989 : isChangeLinkMap, isCurTag));
1990 :
1991 0 : std::string tag;
1992 0 : CHK_RET(GetTagFromNewTag(resMapIt.first, tag));
1993 0 : CHK_RET(ReAllocTransports(tag, resMapIt.first));
1994 0 : CHK_RET(BuildOpRemoteResParam(resMapIt.second, resMapIt.first, opId.opType, true));
1995 0 : HCCL_RUN_INFO("[%s]success to set status of [%s] resume", __func__, resMapIt.first.c_str());
1996 0 : }
1997 0 : CHK_RET(CopyHostOpResToDeviceParam(newTag));
1998 : } else {
1999 : // 原地重执行
2000 0 : CHK_RET(SetTransportStatusImpl(
2001 : resMap_[newTag].opTransportResponse, statusStop, opId, remoteRank, remoteRankPortMap, true));
2002 0 : CHK_RET(SetTransportStatusImpl(
2003 : resMap_[newTag].opTransportResponseBackUp, statusStop, opId, remoteRank, remoteRankPortMap, false));
2004 0 : std::string tag(reinterpret_cast<const char*>(opId.tag));
2005 0 : CHK_RET(ReAllocTransports(tag, newTag));
2006 0 : CHK_RET(BuildOpRemoteResParam(resMap_[newTag], newTag, opId.opType, true));
2007 0 : CHK_RET(CopyHostOpResToDeviceParam(newTag));
2008 0 : }
2009 : }
2010 0 : return HCCL_SUCCESS;
2011 0 : }
2012 :
2013 0 : HcclResult HcclCommunicator::SetTransportResumeStatus(
2014 : const std::map<u32, bool>& remoteRankPortMap, const std::map<u32, bool>& isChangeLinkMap, bool isChangeLinkFlag,
2015 : bool statusStop)
2016 : {
2017 0 : HCCL_INFO(
2018 : "[SetTransportResumeStatus]isChangeLinkFlag[%d], rank[%u], group[%s]", isChangeLinkFlag, userRank_,
2019 : identifier_.c_str());
2020 :
2021 0 : if (statusStop) {
2022 0 : for (auto& resMapIt : resMap_) {
2023 0 : CHK_RET(ResumeTransportsImpl(resMapIt.second.opTransportResponse, remoteRankPortMap, true, statusStop));
2024 0 : CHK_RET(
2025 : ResumeTransportsImpl(resMapIt.second.opTransportResponseBackUp, remoteRankPortMap, false, statusStop));
2026 : }
2027 : } else {
2028 0 : if (isChangeLinkFlag) {
2029 0 : for (auto& resMapIt : resMap_) {
2030 0 : if (hostResMap_.find(resMapIt.first) != hostResMap_.end()) {
2031 0 : continue;
2032 : }
2033 0 : CHK_RET(ResumeTransportsImplForChange(
2034 : resMapIt.second.opTransportResponse, remoteRankPortMap, isChangeLinkMap, true));
2035 0 : CHK_RET(ResumeTransportsImplForChange(
2036 : resMapIt.second.opTransportResponseBackUp, remoteRankPortMap, isChangeLinkMap, false));
2037 :
2038 0 : std::string tag;
2039 0 : CHK_RET(GetTagFromNewTag(resMapIt.first, tag));
2040 0 : CHK_RET(ReAllocTransports(tag, resMapIt.first));
2041 0 : CHK_RET(BuildOpRemoteResParam(resMapIt.second, resMapIt.first, HcclCMDType::HCCL_CMD_ALL, true));
2042 0 : CHK_RET(CopyHostOpResToDeviceParam(resMapIt.first));
2043 0 : }
2044 : } else {
2045 0 : for (auto& resMapIt : resMap_) {
2046 0 : CHK_RET(ResumeTransportsImpl(resMapIt.second.opTransportResponse, remoteRankPortMap, true, statusStop));
2047 0 : CHK_RET(ResumeTransportsImpl(
2048 : resMapIt.second.opTransportResponseBackUp, remoteRankPortMap, false, statusStop));
2049 0 : std::string tag;
2050 0 : CHK_RET(GetTagFromNewTag(resMapIt.first, tag));
2051 0 : CHK_RET(ReAllocTransports(tag, resMapIt.first));
2052 0 : CHK_RET(BuildOpRemoteResParam(resMapIt.second, resMapIt.first, HcclCMDType::HCCL_CMD_ALL, true));
2053 0 : CHK_RET(CopyHostOpResToDeviceParam(resMapIt.first));
2054 0 : }
2055 : }
2056 : }
2057 0 : return HCCL_SUCCESS;
2058 : }
2059 :
2060 0 : HcclResult HcclCommunicator::ResumeTransportsImplForChange(
2061 : OpCommTransport& opCommTransport, const std::map<u32, bool>& remoteRankPortMap,
2062 : const std::map<u32, bool>& isChangeLinkMap, bool isUseDefault)
2063 : {
2064 : // 借轨的resume阶段
2065 0 : for (auto& levelNSubCommTransport : opCommTransport) {
2066 0 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
2067 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
2068 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
2069 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
2070 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
2071 : bool isPortSatisfy
2072 0 : = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end()
2073 0 : && remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
2074 0 : bool isChangeLink = (isChangeLinkMap.find(transportRemoteRank) != isChangeLinkMap.end());
2075 0 : HCCL_INFO(
2076 : "[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], "
2077 : "isPortSatisfy[%d], isChangeLink[%d]",
2078 : transportRemoteRank, isUseDefault, isValid, isPortSatisfy, isChangeLink);
2079 0 : if (isValid && isPortSatisfy && isChangeLink) {
2080 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, false));
2081 : }
2082 : }
2083 : }
2084 : }
2085 0 : return HCCL_SUCCESS;
2086 : }
2087 :
2088 0 : HcclResult HcclCommunicator::ResumeTransportsImpl(
2089 : OpCommTransport& opCommTransport, const std::map<u32, bool>& remoteRankPortMap, bool isUseDefault, bool statusStop)
2090 : {
2091 0 : for (auto& levelNSubCommTransport : opCommTransport) {
2092 0 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
2093 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
2094 0 : u32 transportRemoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
2095 0 : bool isValid = singleSubCommTransport.transportRequests[i].isValid;
2096 : // isPortSatisfy表示当前对端使用的主备网口是否和changeLinkInfo一致
2097 : bool isPortSatisfy
2098 0 : = (remoteRankPortMap.find(transportRemoteRank) != remoteRankPortMap.end()
2099 0 : && remoteRankPortMap.find(transportRemoteRank)->second == isUseDefault);
2100 0 : HCCL_INFO(
2101 : "[SetTransportStatus]remoteRank[%u], isUseDefault[%d], isValid[%d], isPortSatisfy[%d]",
2102 : transportRemoteRank, isUseDefault, isValid, isPortSatisfy);
2103 0 : if (isValid && isPortSatisfy) {
2104 0 : CHK_RET(SetSignalTransport(singleSubCommTransport, i, statusStop));
2105 : }
2106 : }
2107 : }
2108 : }
2109 0 : return HCCL_SUCCESS;
2110 : }
2111 :
2112 0 : HcclResult HcclCommunicator::ReAllocTransports(const std::string& tag, const std::string& newTag)
2113 : {
2114 0 : HcclResult ret = HCCL_SUCCESS;
2115 0 : HCCL_INFO("[%s] alloc tag[%s] transports", __func__, newTag.c_str());
2116 :
2117 0 : AlgResourceResponse& algResResponse = resMap_[newTag];
2118 0 : DeviceMem expMem = cclBufferManager_.GetCommCCLBuffer();
2119 :
2120 : TransportIOMem transMem{
2121 0 : algResResponse.cclInputMem,
2122 0 : algResResponse.cclOutputMem,
2123 0 : algResResponse.paramInputMem,
2124 0 : algResResponse.paramOutputMem,
2125 0 : algResResponse.scratchMem,
2126 0 : algResResponse.aivInputMem,
2127 0 : algResResponse.aivOutputMem,
2128 : expMem,
2129 : DeviceMem(),
2130 0 : {}};
2131 :
2132 : {
2133 : // Transport资源 重建链, 一定是AICPU展开，所以 isAicpuModeEn=true
2134 0 : StateGuard<HcclCommunicator, HcclCommState> guard(this, HcclCommState::BUILDING);
2135 0 : ret = transportManager_->Alloc(tag, transMem, algResResponse.opTransportResponse, true);
2136 0 : CHK_PRT_RET(
2137 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s]Realloc transports failed, tag[%s]", __func__, newTag.c_str()), ret);
2138 0 : }
2139 :
2140 0 : if (IsEnableBackupLink()) {
2141 0 : HCCL_INFO("[%s] alloc tag[%s] backup transports", __func__, newTag.c_str());
2142 : // 超节点 && level2支持重执行 && Aicpu：备用Transport资源 重建链
2143 0 : StateGuard<HcclCommunicator, HcclCommState> guard(this, HcclCommState::BUILDING);
2144 0 : ret = transportManager_->Alloc(tag, transMem, algResResponse.opTransportResponseBackUp, true, true);
2145 0 : CHK_PRT_RET(
2146 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s]Alloc backup transports failed, tag[%s]", __func__, newTag.c_str()),
2147 : ret);
2148 0 : }
2149 0 : SaveLinkRes(algResResponse.opTransportResponse);
2150 0 : SaveLinkRes(algResResponse.opTransportResponseBackUp);
2151 0 : return HCCL_SUCCESS;
2152 0 : }
2153 :
2154 6 : HcclResult HcclCommunicator::Stop()
2155 : {
2156 6 : HcclUs startut = TIME_NOW();
2157 6 : isSuspending = true;
2158 6 : HCCL_DEBUG("HcclCommunicator Stop begin.");
2159 6 : for (auto& it : tagCommInfo_) {
2160 0 : CHK_RET(TraverseLinkVector(it.second.commLevel1, true));
2161 0 : CHK_RET(TraverseLinkVector(it.second.commLevel0, true));
2162 0 : CHK_RET(TraverseLinkVector(it.second.commLevel2, true));
2163 0 : CHK_RET(TraverseLinkVector(it.second.commP2P, true));
2164 0 : if (it.second.commIntraServer) {
2165 0 : const std::vector<LINK>& ret = it.second.commIntraServer->TransportInfo();
2166 0 : CHK_RET(MigrateLinkVectorToStopOrResume(ret, true));
2167 : }
2168 : }
2169 6 : CHK_RET(TraverseAlgResourceResponse(true));
2170 6 : HcclUs endut = TIME_NOW();
2171 6 : HCCL_RUN_INFO("HcclCommunicator::Stop, Stop take time:[%lld]us", DURATION_US(endut - startut).count());
2172 6 : return HCCL_SUCCESS;
2173 : }
2174 :
2175 1 : HcclResult HcclCommunicator::HostMC2EnvResume()
2176 : {
2177 1 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine()) {
2178 0 : HCCL_DEBUG("[NsRecovery]reset the suspending flag");
2179 0 : KfcExecControl controlCmd;
2180 0 : controlCmd.kfcCmd = KfcCommand::kNone;
2181 0 : controlCmd.bgCmd = BackgroundCommand::kNone;
2182 0 : controlCmd.suspendingStatus = HcclComSuspendingFlag::isResume;
2183 0 : CHK_RET(kfcControlTransferH2D_->Put(0, sizeof(KfcExecControl), reinterpret_cast<uint8_t*>(&controlCmd)));
2184 0 : if (IsEnableCustom()) {
2185 0 : CHK_RET(customControlTransferH2D_->Put(0, sizeof(KfcExecControl), reinterpret_cast<uint8_t*>(&controlCmd)));
2186 : }
2187 : }
2188 1 : return HCCL_SUCCESS;
2189 : }
2190 :
2191 1 : HcclResult HcclCommunicator::ClearWinBuffer()
2192 : {
2193 1 : DeviceMem winBuffer = cclBufferManager_.GetCommExpBuffer();
2194 1 : if (winBuffer.ptr() != nullptr) {
2195 0 : HCCL_INFO("HcclCommunicator::Resume, start to clear win buffer");
2196 0 : CHK_RET(hrtMemSet(static_cast<u8*>(winBuffer.ptr()), EXP_BUFFER_SIZE, EXP_BUFFER_SIZE));
2197 : }
2198 1 : return HCCL_SUCCESS;
2199 1 : }
2200 :
2201 1 : HcclResult HcclCommunicator::AivResume()
2202 : {
2203 1 : if (GetExternalInputHcclAivMode()) {
2204 0 : HCCL_DEBUG("AivResume begin.");
2205 :
2206 0 : CHK_RET(cclBufferManager_.ClearCommAIVbuffer());
2207 0 : HCCL_INFO("[AIV][AivResumeClearSyncBuf] clear aiv buffer done");
2208 :
2209 0 : aivOpbaseTag_ = TAG_INIT_VALUE;
2210 0 : aivOffloadTag_ = TAG_INIT_VALUE;
2211 0 : HCCL_INFO("[AIV][AivResume] clear aiv tag done");
2212 : }
2213 1 : return HCCL_SUCCESS;
2214 : }
2215 :
2216 1 : HcclResult HcclCommunicator::Resume()
2217 : {
2218 1 : HcclUs startut = TIME_NOW();
2219 1 : bool isChangedLink = false;
2220 1 : HCCL_RUN_INFO("HcclCommunicator Resume begin, group[%s].", identifier_.c_str());
2221 : // 发生N秒快恢, 头尾计数可能不对，需要将头尾计数清零
2222 1 : CHK_RET(ClearOpCounterMem());
2223 1 : for (auto& it : tagCommInfo_) {
2224 0 : CHK_RET(TraverseLinkVector(it.second.commLevel1, false));
2225 0 : CHK_RET(TraverseLinkVector(it.second.commLevel0, false));
2226 0 : CHK_RET(TraverseLinkVector(it.second.commLevel2, false));
2227 0 : CHK_RET(TraverseLinkVector(it.second.commP2P, false));
2228 0 : if (it.second.commIntraServer) {
2229 0 : const std::vector<LINK>& ret = it.second.commIntraServer->TransportInfo();
2230 0 : CHK_RET(MigrateLinkVectorToStopOrResume(ret, false));
2231 : }
2232 : }
2233 :
2234 1 : if (GetAicpuUnfoldFlag() || GetAicpuCommEngine()) {
2235 0 : CHK_RET(CheckExitWaitResumeState(isChangedLink));
2236 : }
2237 :
2238 1 : if (!isChangedLink) {
2239 1 : CHK_RET(TraverseAlgResourceResponse(false));
2240 : }
2241 1 : HcclUs cleanNotifyStart = TIME_NOW();
2242 1 : CHK_RET(hrtResourceClean());
2243 1 : HcclUs cleanNotifyEnd = TIME_NOW();
2244 1 : HCCL_RUN_INFO(
2245 : "HcclCommunicator::Resume, hrtResourceClean notify take time:[%lld]us",
2246 : DURATION_US(cleanNotifyEnd - cleanNotifyStart).count());
2247 1 : CHK_RET(HostMC2EnvResume());
2248 1 : CHK_RET(ClearWinBuffer());
2249 1 : CHK_RET(AivResume());
2250 1 : isSuspending = false;
2251 :
2252 1 : HcclUs endut = TIME_NOW();
2253 1 : HCCL_RUN_INFO(
2254 : "HcclCommunicator::Resume, Resume take time:[%lld]us, group[%s]", DURATION_US(endut - startut).count(),
2255 : identifier_.c_str());
2256 :
2257 1 : return HCCL_SUCCESS;
2258 : }
2259 :
2260 123 : HcclResult HcclCommunicator::CheckSuspendingStatus()
2261 : {
2262 123 : if (isSuspending) {
2263 0 : return HCCL_E_SUSPENDING;
2264 : }
2265 123 : return HCCL_SUCCESS;
2266 : }
2267 :
2268 0 : HcclResult HcclCommunicator::SwitchNic(uint32_t nRanks, uint32_t* ranks, bool* useBackup)
2269 : {
2270 0 : CHK_RET(SwitchNic(nRanks, ranks, useBackup, kfcControlTransferH2D_, kfcStatusTransferD2H_));
2271 0 : if (IsEnableCustom()) {
2272 0 : CHK_RET(SwitchNic(nRanks, ranks, useBackup, customControlTransferH2D_, customStatusTransferD2H_));
2273 : }
2274 0 : return HCCL_SUCCESS;
2275 : }
2276 :
2277 122 : void HcclCommunicator::SaveLinkRes(const OpCommTransport& opTransportResponse)
2278 : {
2279 122 : std::lock_guard<std::mutex> commLock(linkResMapMutex_);
2280 1066 : for (auto& opCommTransport : opTransportResponse) {
2281 1044 : for (auto& transports : opCommTransport) {
2282 375 : for (u32 i = 0; i < transports.transportRequests.size(); i++) {
2283 275 : if (transports.links[i] != nullptr
2284 275 : && transports.links[i]->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2285 0 : auto remoteRank = transports.transportRequests[i].remoteUserRank;
2286 : std::string localServerId
2287 0 : = rankInfoList_.size() > userRank_ ? rankInfoList_[userRank_].serverId : "";
2288 0 : s32 localDevicePhyId = rankInfoList_.size() > userRank_ ? rankInfoList_[userRank_].devicePhyId : -1;
2289 : std::string remoteServerId
2290 0 : = rankInfoList_.size() > remoteRank ? rankInfoList_[remoteRank].serverId : "";
2291 : s32 remoteDevicePhyId
2292 0 : = rankInfoList_.size() > remoteRank ? rankInfoList_[remoteRank].devicePhyId : -1;
2293 : LinkInfo linkInfo(
2294 0 : identifier_, userRank_, localServerId, localDevicePhyId, remoteRank, remoteServerId,
2295 0 : remoteDevicePhyId);
2296 0 : linkResMap_.emplace(transports.links[i].get(), linkInfo);
2297 0 : }
2298 : }
2299 : }
2300 : }
2301 244 : return;
2302 122 : }
2303 :
2304 : HcclResult
2305 15 : HcclCommunicator::GetTransportCqeErrors(const HcclNetDevCtx netDevCtx, std::vector<ErrCqeInfo>& infos, u32& num)
2306 : {
2307 15 : if (netDevCtx == nullptr) {
2308 0 : return HCCL_SUCCESS;
2309 : }
2310 15 : HcclIpAddress localIp;
2311 15 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
2312 :
2313 15 : u32 qpn = 0;
2314 15 : std::vector<std::pair<Transport*, CqeInfo>> infolist;
2315 15 : Transport::GetTransportErrorCqe(netDevCtx, infolist, num);
2316 15 : std::lock_guard<std::mutex> commLock(linkResMapMutex_);
2317 16 : for (auto& info : infolist) {
2318 1 : auto iter = linkResMap_.find(info.first);
2319 1 : if (iter != linkResMap_.end()) {
2320 0 : CHK_RET((info.first)->GetTransportId(qpn));
2321 0 : infos.push_back(ErrCqeInfo(info.second, iter->second, qpn));
2322 : } else {
2323 1 : LinkInfo linkInfo;
2324 1 : linkInfo.localDevicePhyId = -1;
2325 1 : linkInfo.remoteDevicePhyId = -1;
2326 1 : infos.push_back(ErrCqeInfo(info.second, linkInfo, qpn));
2327 1 : HCCL_RUN_WARNING("[GetTransportCqeErrors]Transport linkInfo was not saved, only print CqeInfo");
2328 1 : }
2329 : }
2330 15 : num = infos.size();
2331 15 : return HCCL_SUCCESS;
2332 15 : }
2333 :
2334 111 : void HcclCommunicator::ClearOpTransportResponseLinks(OpCommTransport& opTransportResponse)
2335 : {
2336 1860 : for (auto& levelNSubCommTransport : opTransportResponse) {
2337 2062 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
2338 310 : u32 size = singleSubCommTransport.transportRequests.size();
2339 310 : singleSubCommTransport.links.resize(size, nullptr);
2340 309 : singleSubCommTransport.status.resize(size, TransportStatus::INIT);
2341 308 : HCCL_INFO("[%s] size[%u], linksSize[%d]", __func__, size, singleSubCommTransport.links.size());
2342 : }
2343 : }
2344 109 : }
2345 :
2346 0 : HcclResult HcclCommunicator::SetDevIbverbsData(
2347 : CommBase* comm, bool isSupportNormalQP, u64 commBufferSize, void* commInPtr, void* commOutPtr)
2348 : {
2349 0 : const u32 curRankId = comm->Rank();
2350 0 : const u32 rankSize = comm->RankSize();
2351 :
2352 0 : CHK_RET(AllocAndClearHostMem(sizeof(TransportDeviceNormalData) * rankSize, transDevIbverbsDataMem_));
2353 : TransportDeviceNormalData* transDevIbverbsData
2354 0 : = static_cast<TransportDeviceNormalData*>(transDevIbverbsDataMem_->ptr());
2355 :
2356 0 : for (u32 i = 0; i < rankSize; i++) {
2357 0 : auto& data = transDevIbverbsData[i];
2358 0 : if (i != curRankId) {
2359 : // 对端link的信息
2360 0 : const auto transport = comm->GetTransportByRank(i);
2361 0 : CHK_PTR_NULL(transport);
2362 0 : void* bufferIn = nullptr;
2363 0 : void* bufferOut = nullptr;
2364 0 : u32 remoteInMemKey = 0;
2365 0 : u32 remoteOutMemKey = 0;
2366 0 : CHK_RET(transport->GetRemoteMem(UserMemType::INPUT_MEM, &bufferIn));
2367 0 : CHK_RET(transport->GetRemoteMem(UserMemType::OUTPUT_MEM, &bufferOut));
2368 0 : data.remoteInputMem.addr = reinterpret_cast<uint64_t>(bufferIn);
2369 0 : data.remoteOutputMem.addr = reinterpret_cast<uint64_t>(bufferOut);
2370 0 : CHK_RET(transport->GetRemoteMemSize(UserMemType::INPUT_MEM, data.remoteInputMem.size));
2371 0 : CHK_RET(transport->GetRemoteMemSize(UserMemType::OUTPUT_MEM, data.remoteOutputMem.size));
2372 : // IBV链路需要的资源
2373 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2374 0 : CHK_RET(transport->GetRemoteMemKey(UserMemType::INPUT_MEM, &remoteInMemKey));
2375 0 : CHK_RET(transport->GetRemoteMemKey(UserMemType::OUTPUT_MEM, &remoteOutMemKey));
2376 0 : data.remoteInputMem.key = remoteInMemKey;
2377 0 : data.remoteOutputMem.key = remoteOutMemKey;
2378 0 : CHK_RET(transport->GetLocalMemDetails(UserMemType::INPUT_MEM, data.localInputMem));
2379 0 : CHK_RET(transport->GetLocalMemDetails(UserMemType::OUTPUT_MEM, data.localOutputMem));
2380 0 : std::vector<HcclQpInfoV2> qpInfos;
2381 0 : CHK_RET(transport->GetAiQpInfo(qpInfos));
2382 0 : data.qpInfo = qpInfos[0];
2383 0 : }
2384 0 : } else {
2385 : // 本rank的信息
2386 0 : data.localInputMem.addr = reinterpret_cast<uint64_t>(commInPtr);
2387 0 : data.localInputMem.size = commBufferSize;
2388 0 : data.localOutputMem.addr = reinterpret_cast<uint64_t>(commOutPtr);
2389 0 : data.localOutputMem.size = commBufferSize;
2390 : }
2391 :
2392 0 : if (isSupportNormalQP) {
2393 0 : data.qpMode = QPMode::NORMAL;
2394 : }
2395 : // Debugging info
2396 0 : data.Print();
2397 : }
2398 0 : return HCCL_SUCCESS;
2399 : }
2400 :
2401 0 : HcclResult HcclCommunicator::GetTransportLocalMem(
2402 : const std::shared_ptr<Transport>& transport, UserMemType memType, MemDetails& detail) const
2403 : {
2404 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2405 0 : CHK_RET(transport->GetLocalMemDetails(memType, detail));
2406 : }
2407 0 : return HCCL_SUCCESS;
2408 : }
2409 :
2410 0 : HcclResult HcclCommunicator::GetTransportRemoteMem(
2411 : const std::shared_ptr<Transport>& transport, UserMemType memType, MemDetails& detail) const
2412 : {
2413 0 : void* addr = nullptr;
2414 0 : CHK_RET(transport->GetRemoteMem(memType, &addr));
2415 :
2416 0 : detail.addr = reinterpret_cast<uint64_t>(addr);
2417 0 : CHK_RET(transport->GetRemoteMemSize(memType, detail.size));
2418 :
2419 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2420 0 : CHK_RET(transport->GetRemoteMemKey(memType, &detail.key));
2421 : }
2422 :
2423 0 : return HCCL_SUCCESS;
2424 : }
2425 :
2426 0 : HcclResult HcclCommunicator::GenAiRMAInfo(CommBase* comm)
2427 : {
2428 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2429 0 : HcclAiRMAInfo* aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2430 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2431 :
2432 0 : const std::string& tag = comm->Tag();
2433 0 : aiRMAInfoPtr->curRankId = comm->Rank();
2434 0 : aiRMAInfoPtr->rankNum = comm->RankSize();
2435 :
2436 0 : CHK_RET(GetAIVNormalQPInfo(comm, tag));
2437 :
2438 0 : u32 tmpQueueSize = aiRMAInfoPtr->rankNum * aiRMAInfoPtr->qpNum;
2439 0 : u32 tmpMemSize = aiRMAInfoPtr->rankNum;
2440 0 : u32 tmpMemDetailSize = aiRMAInfoPtr->rankNum * AiMemMaxNum;
2441 :
2442 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiSqMem_));
2443 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiScqMem_));
2444 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiRqMem_));
2445 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiRcqMem_));
2446 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAMemInfo) * tmpMemSize, aiMemMem_));
2447 0 : HcclAiRMAMemInfo* aiMemHost = reinterpret_cast<HcclAiRMAMemInfo*>(aiMemMem_->ptr());
2448 :
2449 0 : CHK_RET(AllocAndClearHostMem(sizeof(MemDetails) * tmpMemDetailSize, aiMemDetailsMem_));
2450 0 : MemDetails* aiMemDetailsHost = reinterpret_cast<MemDetails*>(aiMemDetailsMem_->ptr());
2451 0 : CHK_RET(DeviceMem::alloc(aiMemDetailsDev_, aiMemDetailsMem_->size()));
2452 0 : u64 memBase = reinterpret_cast<uint64_t>(aiMemDetailsDev_.ptr());
2453 :
2454 0 : for (u32 i = 0; i < aiRMAInfoPtr->rankNum; i++) {
2455 0 : MemDetails& remoteIn = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_INPUT)];
2456 0 : MemDetails& remoteOut = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_OUTPUT)];
2457 0 : MemDetails& localIn = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_INPUT)];
2458 0 : MemDetails& localOut = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_OUTPUT)];
2459 :
2460 0 : if (i != aiRMAInfoPtr->curRankId) {
2461 : // link rank info
2462 0 : const auto transport = comm->GetTransportByRank(i);
2463 0 : CHK_PTR_NULL(transport);
2464 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::INPUT_MEM, remoteIn));
2465 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::OUTPUT_MEM, remoteOut));
2466 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::INPUT_MEM, localIn));
2467 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::OUTPUT_MEM, localOut));
2468 :
2469 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2470 0 : CHK_RET(GenIbvAiRMAInfo(i, transport, tag, aiRMAInfoPtr));
2471 : }
2472 0 : } else {
2473 0 : void* commInPtr = nullptr;
2474 0 : void* commOutPtr = nullptr;
2475 : u64 commInSize;
2476 : u64 commOutSize;
2477 0 : CHK_RET(cclBufferManager_.GetInCCLbuffer(commInPtr, commInSize));
2478 0 : CHK_RET(cclBufferManager_.GetOutCCLbuffer(commOutPtr, commOutSize));
2479 0 : localIn.addr = reinterpret_cast<uint64_t>(commInPtr);
2480 0 : localIn.size = commInSize;
2481 0 : localOut.addr = reinterpret_cast<uint64_t>(commOutPtr);
2482 0 : localOut.size = commOutSize;
2483 : }
2484 :
2485 0 : aiMemHost[i].memMaxNum = AiMemMaxNum;
2486 0 : aiMemHost[i].sizeOfMemDetails = static_cast<u32>(sizeof(MemDetails));
2487 0 : aiMemHost[i].memDetailPtr = memBase + i * AiMemMaxNum * aiMemHost[i].sizeOfMemDetails;
2488 :
2489 0 : HCCL_DEBUG(
2490 : "[%s] tag[%s] curRankId[%u] dstRankId[%u] rankNum[%u] qpNum[%u] memMaxNum[%u] sizeOfMemDetails[%u] "
2491 : "memDetailPtr[%p] remoteInAddr[%p] remoteInSize[%llu] remoteInKey[%u] remoteOutAddr[%p] "
2492 : "remoteOutSize[%llu] remoteOutKey[%u] localInAddr[%p] localInSize[%llu] localInKey[%u] "
2493 : "localOutAddr[%p] localOutSize[%llu] localOutKey[%u]",
2494 : __func__, tag.c_str(), aiRMAInfoPtr->curRankId, i, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum,
2495 : aiMemHost[i].memMaxNum, aiMemHost[i].sizeOfMemDetails, aiMemHost[i].memDetailPtr, remoteIn.addr,
2496 : remoteIn.size, remoteIn.key, remoteOut.addr, remoteOut.size, remoteOut.key, localIn.addr, localIn.size,
2497 : localIn.key, localOut.addr, localOut.size, localOut.key);
2498 : }
2499 :
2500 0 : return HCCL_SUCCESS;
2501 : }
2502 :
2503 0 : HcclResult HcclCommunicator::GenAiRMAInfoV2(const std::string& tag)
2504 : {
2505 0 : CHK_PTR_NULL(rmaInfoMem_);
2506 0 : HcclRMAInfo* rmaInfoPtr = reinterpret_cast<HcclRMAInfo*>(rmaInfoMem_->ptr());
2507 0 : CHK_PTR_NULL(rmaInfoPtr);
2508 0 : rmaInfoPtr->curRankId = userRank_;
2509 : ;
2510 0 : rmaInfoPtr->rankNum = userRankSize_;
2511 0 : LevelNSubCommTransport& commTransport = resMap_[tag].opTransportResponse[COMM_COMBINE_ORDER];
2512 0 : CHK_PRT_RET(
2513 : commTransport.size() <= 0,
2514 : HCCL_ERROR(
2515 : "[%s] no LevelComm resource, please create comm first. "
2516 : "tag[%s], curRankId[%u] rankNum[%u]",
2517 : __func__, tag.c_str(), rmaInfoPtr->curRankId, rmaInfoPtr->rankNum),
2518 : HCCL_E_INTERNAL);
2519 0 : std::vector<LINK>& links = commTransport[0].links;
2520 0 : CHK_PRT_RET(
2521 : links.size() <= 0,
2522 : HCCL_ERROR(
2523 : "[%s] no transport resource, please create links first. "
2524 : "tag[%s], curRankId[%u] rankNum[%u]",
2525 : __func__, tag.c_str(), rmaInfoPtr->curRankId, rmaInfoPtr->rankNum),
2526 : HCCL_E_INTERNAL);
2527 0 : CHK_RET(GetAIVNormalQPInfoV2(links, tag));
2528 :
2529 0 : u32 tmpQueueSize = rmaInfoPtr->rankNum * rmaInfoPtr->qpNum;
2530 0 : u32 tmpMemSize = rmaInfoPtr->rankNum;
2531 0 : u32 tmpMemDetailSize = rmaInfoPtr->rankNum * AiMemMaxNum;
2532 :
2533 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiSqMem_));
2534 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiScqMem_));
2535 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAWQ) * tmpQueueSize, aiRqMem_));
2536 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMACQ) * tmpQueueSize, aiRcqMem_));
2537 0 : CHK_RET(AllocAndClearHostMem(sizeof(HcclAiRMAMemInfo) * tmpMemSize, aiMemMem_));
2538 0 : HcclAiRMAMemInfo* aiMemHost = reinterpret_cast<HcclAiRMAMemInfo*>(aiMemMem_->ptr());
2539 :
2540 0 : CHK_RET(AllocAndClearHostMem(sizeof(MemDetails) * tmpMemDetailSize, aiMemDetailsMem_));
2541 0 : MemDetails* aiMemDetailsHost = reinterpret_cast<MemDetails*>(aiMemDetailsMem_->ptr());
2542 :
2543 0 : aiMemDetailsDev_ = DeviceMem::alloc(aiMemDetailsMem_->size());
2544 0 : u64 memBase = reinterpret_cast<uint64_t>(aiMemDetailsDev_.ptr());
2545 :
2546 0 : for (u32 i = 0; i < rmaInfoPtr->rankNum; i++) {
2547 0 : MemDetails& remoteIn = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_INPUT)];
2548 0 : MemDetails& remoteOut = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::REMOTE_OUTPUT)];
2549 0 : MemDetails& localIn = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_INPUT)];
2550 0 : MemDetails& localOut = aiMemDetailsHost[i * AiMemMaxNum + GetAiMemTypeVal(HcclAiRMAMemType::LOCAL_OUTPUT)];
2551 :
2552 0 : if (i != rmaInfoPtr->curRankId) {
2553 : // link rank info
2554 0 : const auto transport = links[i];
2555 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::INPUT_MEM, remoteIn));
2556 0 : CHK_RET(GetTransportRemoteMem(transport, UserMemType::OUTPUT_MEM, remoteOut));
2557 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::INPUT_MEM, localIn));
2558 0 : CHK_RET(GetTransportLocalMem(transport, UserMemType::OUTPUT_MEM, localOut));
2559 :
2560 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2561 0 : CHK_RET(GenIbvAiRMAInfo(i, transport, tag, rmaInfoPtr));
2562 : }
2563 0 : } else {
2564 0 : void* commInPtr = nullptr;
2565 0 : void* commOutPtr = nullptr;
2566 : u64 commInSize;
2567 : u64 commOutSize;
2568 0 : CHK_RET(cclBufferManager_.GetInCCLbuffer(commInPtr, commInSize));
2569 0 : CHK_RET(cclBufferManager_.GetOutCCLbuffer(commOutPtr, commOutSize));
2570 0 : localIn.addr = reinterpret_cast<uint64_t>(commInPtr);
2571 0 : localIn.size = commInSize;
2572 0 : localOut.addr = reinterpret_cast<uint64_t>(commOutPtr);
2573 0 : localOut.size = commOutSize;
2574 : }
2575 :
2576 0 : aiMemHost[i].memMaxNum = AiMemMaxNum;
2577 0 : aiMemHost[i].sizeOfMemDetails = static_cast<u32>(sizeof(MemDetails));
2578 0 : aiMemHost[i].memDetailPtr = memBase + i * AiMemMaxNum * aiMemHost[i].sizeOfMemDetails;
2579 :
2580 0 : HCCL_DEBUG(
2581 : "[%s] tag[%s] curRankId[%u] dstRankId[%u] rankNum[%u] qpNum[%u] memMaxNum[%u] sizeOfMemDetails[%u] "
2582 : "memDetailPtr[%p] remoteInAddr[%p] remoteInSize[%llu] remoteInKey[%u] remoteOutAddr[%p] "
2583 : "remoteOutSize[%llu] remoteOutKey[%u] localInAddr[%p] localInSize[%llu] localInKey[%u] "
2584 : "localOutAddr[%p] localOutSize[%llu] localOutKey[%u]",
2585 : __func__, tag.c_str(), rmaInfoPtr->curRankId, i, rmaInfoPtr->rankNum, rmaInfoPtr->qpNum,
2586 : aiMemHost[i].memMaxNum, aiMemHost[i].sizeOfMemDetails, aiMemHost[i].memDetailPtr, remoteIn.addr,
2587 : remoteIn.size, remoteIn.key, remoteOut.addr, remoteOut.size, remoteOut.key, localIn.addr, localIn.size,
2588 : localIn.key, localOut.addr, localOut.size, localOut.key);
2589 : }
2590 :
2591 0 : return HCCL_SUCCESS;
2592 : }
2593 :
2594 0 : HcclResult HcclCommunicator::H2DAiRMAInfo(const std::string& tag, rtStream_t aiCpuStream)
2595 : {
2596 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2597 0 : HcclAiRMAInfo* aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2598 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2599 :
2600 0 : CHK_PTR_NULL(combinOparaMem_);
2601 0 : HcclCombinOpParam* combinOparaPtr = reinterpret_cast<HcclCombinOpParam*>(combinOparaMem_->ptr());
2602 0 : CHK_PTR_NULL(combinOparaPtr);
2603 :
2604 0 : aiRMAInfoPtr->sizeOfAiRMAWQ = static_cast<u32>(sizeof(HcclAiRMAWQ));
2605 0 : aiRMAInfoPtr->sizeOfAiRMACQ = static_cast<u32>(sizeof(HcclAiRMACQ));
2606 0 : aiRMAInfoPtr->sizeOfAiRMAMem = static_cast<u32>(sizeof(HcclAiRMAMemInfo));
2607 :
2608 0 : CHK_RET(DeviceMem::alloc(aiSqDev_, aiSqMem_->size()));
2609 0 : aiRMAInfoPtr->sqPtr = aiSqDev_.ptr();
2610 0 : CHK_RET(hrtMemAsyncCopy(
2611 : aiSqDev_.ptr(), aiSqDev_.size(), aiSqMem_->ptr(), aiSqDev_.size(),
2612 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2613 :
2614 0 : CHK_RET(DeviceMem::alloc(aiScqDev_, aiScqMem_->size()));
2615 0 : aiRMAInfoPtr->scqPtr = aiScqDev_.ptr();
2616 0 : CHK_RET(hrtMemAsyncCopy(
2617 : aiScqDev_.ptr(), aiScqDev_.size(), aiScqMem_->ptr(), aiScqDev_.size(),
2618 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2619 :
2620 0 : CHK_RET(DeviceMem::alloc(aiRqDev_, aiRqMem_->size()));
2621 0 : aiRMAInfoPtr->rqPtr = aiRqDev_.ptr();
2622 0 : CHK_RET(hrtMemAsyncCopy(
2623 : aiRqDev_.ptr(), aiRqDev_.size(), aiRqMem_->ptr(), aiRqDev_.size(),
2624 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2625 :
2626 0 : CHK_RET(DeviceMem::alloc(aiRcqDev_, aiRcqMem_->size()));
2627 0 : aiRMAInfoPtr->rcqPtr = aiRcqDev_.ptr();
2628 0 : CHK_RET(hrtMemAsyncCopy(
2629 : aiRcqDev_.ptr(), aiRcqDev_.size(), aiRcqMem_->ptr(), aiRcqDev_.size(),
2630 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2631 :
2632 0 : CHK_RET(hrtMemAsyncCopy(
2633 : aiMemDetailsDev_.ptr(), aiMemDetailsDev_.size(), aiMemDetailsMem_->ptr(), aiMemDetailsDev_.size(),
2634 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2635 :
2636 0 : CHK_RET(DeviceMem::alloc(aiMemDev_, aiMemMem_->size()));
2637 0 : aiRMAInfoPtr->memPtr = aiMemDev_.ptr();
2638 0 : CHK_RET(hrtMemAsyncCopy(
2639 : aiMemDev_.ptr(), aiMemDev_.size(), aiMemMem_->ptr(), aiMemDev_.size(),
2640 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2641 :
2642 0 : combinOparaPtr->sizeOfAiRMAInfo = static_cast<u64>(sizeof(HcclAiRMAInfo));
2643 0 : CHK_RET(DeviceMem::alloc(aiRMAInfoDev_, combinOparaPtr->sizeOfAiRMAInfo));
2644 0 : combinOparaPtr->aiRMAInfo = aiRMAInfoDev_.ptr();
2645 0 : CHK_RET(hrtMemAsyncCopy(
2646 : aiRMAInfoDev_.ptr(), aiRMAInfoDev_.size(), aiRMAInfoMem_->ptr(), aiRMAInfoDev_.size(),
2647 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2648 :
2649 0 : HCCL_INFO(
2650 : "[%s] tag[%s] curRankId[%u] rankNum[%u] qpNum[%u] aiRMAInfo[%p] sizeOfAiRMAInfo[%llu] "
2651 : "sizeOfAiRMAWQ[%u] sizeOfAiRMACQ[%u] sizeOfAiRMAMem[%u] sqPtr[%p] sqSize[%llu] sqCount[%zu] "
2652 : "scqPtr[%p] scqSize[%llu] scqCount[%zu] rqPtr[%p] rqSize[%llu] rqCount[%zu] rcqPtr[%p] "
2653 : "rcqSize[%llu] rcqCount[%zu] memPtr[%p] memSize[%llu] memCount[%zu] memDetailCount[%zu]",
2654 : __func__, tag.c_str(), aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum,
2655 : combinOparaPtr->aiRMAInfo, combinOparaPtr->sizeOfAiRMAInfo, aiRMAInfoPtr->sizeOfAiRMAWQ,
2656 : aiRMAInfoPtr->sizeOfAiRMACQ, aiRMAInfoPtr->sizeOfAiRMAMem, aiRMAInfoPtr->sqPtr, aiSqDev_.size(),
2657 : aiSqMem_->size(), aiRMAInfoPtr->scqPtr, aiScqDev_.size(), aiScqMem_->size(), aiRMAInfoPtr->rqPtr,
2658 : aiRqDev_.size(), aiRqMem_->size(), aiRMAInfoPtr->rcqPtr, aiRcqDev_.size(), aiRcqMem_->size(),
2659 : aiRMAInfoPtr->memPtr, aiMemDev_.size(), aiMemMem_->size(), aiMemDetailsMem_->size());
2660 :
2661 0 : return HCCL_SUCCESS;
2662 : }
2663 :
2664 0 : HcclResult HcclCommunicator::H2DAiRMAInfoV2(const std::string& tag, rtStream_t aiCpuStream)
2665 : {
2666 0 : CHK_PTR_NULL(rmaInfoMem_);
2667 0 : HcclRMAInfo* rmaInfoPtr = reinterpret_cast<HcclRMAInfo*>(rmaInfoMem_->ptr());
2668 0 : CHK_PTR_NULL(rmaInfoPtr);
2669 :
2670 0 : CHK_PTR_NULL(combinOparaMem_);
2671 0 : HcclCombinOpParam* combinOparaPtr = reinterpret_cast<HcclCombinOpParam*>(combinOparaMem_->ptr());
2672 0 : CHK_PTR_NULL(combinOparaPtr);
2673 :
2674 0 : rmaInfoPtr->sizeOfRMAWQ = static_cast<u32>(sizeof(HcclAiRMAWQ));
2675 0 : rmaInfoPtr->sizeOfRMACQ = static_cast<u32>(sizeof(HcclAiRMACQ));
2676 0 : rmaInfoPtr->sizeOfRMAMem = static_cast<u32>(sizeof(HcclAiRMAMemInfo));
2677 :
2678 0 : CHK_RET(DeviceMem::alloc(aiSqDev_, aiSqMem_->size()));
2679 0 : rmaInfoPtr->sqPtr = reinterpret_cast<uintptr_t>(aiSqDev_.ptr());
2680 0 : CHK_RET(hrtMemAsyncCopy(
2681 : aiSqDev_.ptr(), aiSqDev_.size(), aiSqMem_->ptr(), aiSqDev_.size(),
2682 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2683 :
2684 0 : CHK_RET(DeviceMem::alloc(aiScqDev_, aiScqMem_->size()));
2685 0 : rmaInfoPtr->scqPtr = reinterpret_cast<uintptr_t>(aiScqDev_.ptr());
2686 0 : CHK_RET(hrtMemAsyncCopy(
2687 : aiScqDev_.ptr(), aiScqDev_.size(), aiScqMem_->ptr(), aiScqDev_.size(),
2688 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2689 :
2690 0 : CHK_RET(DeviceMem::alloc(aiRqDev_, aiRqMem_->size()));
2691 0 : rmaInfoPtr->rqPtr = reinterpret_cast<uintptr_t>(aiRqDev_.ptr());
2692 0 : CHK_RET(hrtMemAsyncCopy(
2693 : aiRqDev_.ptr(), aiRqDev_.size(), aiRqMem_->ptr(), aiRqDev_.size(),
2694 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2695 :
2696 0 : CHK_RET(DeviceMem::alloc(aiRcqDev_, aiRcqMem_->size()));
2697 0 : rmaInfoPtr->rcqPtr = reinterpret_cast<uintptr_t>(aiRcqDev_.ptr());
2698 0 : CHK_RET(hrtMemAsyncCopy(
2699 : aiRcqDev_.ptr(), aiRcqDev_.size(), aiRcqMem_->ptr(), aiRcqDev_.size(),
2700 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2701 :
2702 0 : CHK_RET(hrtMemAsyncCopy(
2703 : aiMemDetailsDev_.ptr(), aiMemDetailsDev_.size(), aiMemDetailsMem_->ptr(), aiMemDetailsDev_.size(),
2704 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2705 :
2706 0 : CHK_RET(DeviceMem::alloc(aiMemDev_, aiMemMem_->size()));
2707 0 : rmaInfoPtr->memPtr = reinterpret_cast<uintptr_t>(aiMemDev_.ptr());
2708 0 : CHK_RET(hrtMemAsyncCopy(
2709 : aiMemDev_.ptr(), aiMemDev_.size(), aiMemMem_->ptr(), aiMemDev_.size(),
2710 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2711 :
2712 0 : combinOparaPtr->sizeOfAiRMAInfo = static_cast<u64>(sizeof(HcclAiRMAInfo));
2713 0 : CHK_RET(DeviceMem::alloc(aiRMAInfoDev_, combinOparaPtr->sizeOfAiRMAInfo));
2714 0 : combinOparaPtr->aiRMAInfo = aiRMAInfoDev_.ptr();
2715 0 : CHK_RET(hrtMemAsyncCopy(
2716 : aiRMAInfoDev_.ptr(), aiRMAInfoDev_.size(), rmaInfoMem_->ptr(), aiRMAInfoDev_.size(),
2717 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, aiCpuStream));
2718 :
2719 0 : HCCL_INFO(
2720 : "[%s] tag[%s] curRankId[%u] rankNum[%u] qpNum[%u] aiRMAInfo[%p] sizeOfAiRMAInfo[%llu] "
2721 : "sizeOfAiRMAWQ[%u] sizeOfAiRMACQ[%u] sizeOfAiRMAMem[%u] sqPtr[%p] sqSize[%llu] sqCount[%zu] "
2722 : "scqPtr[%p] scqSize[%llu] scqCount[%zu] rqPtr[%p] rqSize[%llu] rqCount[%zu] rcqPtr[%p] "
2723 : "rcqSize[%llu] rcqCount[%zu] memPtr[%p] memSize[%llu] memCount[%zu] memDetailCount[%zu]",
2724 : __func__, tag.c_str(), rmaInfoPtr->curRankId, rmaInfoPtr->rankNum, rmaInfoPtr->qpNum, combinOparaPtr->aiRMAInfo,
2725 : combinOparaPtr->sizeOfAiRMAInfo, rmaInfoPtr->sizeOfRMAWQ, rmaInfoPtr->sizeOfRMACQ, rmaInfoPtr->sizeOfRMAMem,
2726 : rmaInfoPtr->sqPtr, aiSqDev_.size(), aiSqMem_->size(), rmaInfoPtr->scqPtr, aiScqDev_.size(), aiScqMem_->size(),
2727 : rmaInfoPtr->rqPtr, aiRqDev_.size(), aiRqMem_->size(), rmaInfoPtr->rcqPtr, aiRcqDev_.size(), aiRcqMem_->size(),
2728 : rmaInfoPtr->memPtr, aiMemDev_.size(), aiMemMem_->size(), aiMemDetailsMem_->size());
2729 :
2730 0 : return HCCL_SUCCESS;
2731 : }
2732 :
2733 0 : HcclResult HcclCommunicator::GetAIVNormalQPInfo(CommBase* comm, const std::string& tag)
2734 : {
2735 : // 获取 Transport QP 数量
2736 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2737 0 : HcclAiRMAInfo* aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2738 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2739 :
2740 0 : aiRMAInfoPtr->qpNum = HCCL_QPS_PER_CONNECTION_DEFAULT;
2741 0 : for (u32 i = 0; i < aiRMAInfoPtr->rankNum; i++) {
2742 0 : if (i != aiRMAInfoPtr->curRankId) {
2743 0 : const auto transport = comm->GetTransportByRank(i);
2744 0 : CHK_PTR_NULL(transport);
2745 0 : if (transport->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2746 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2747 0 : CHK_RET(transport->GetAiRMAQueueInfo(aiQpVec));
2748 0 : aiRMAInfoPtr->qpNum = static_cast<u32>(aiQpVec.size());
2749 0 : }
2750 0 : }
2751 : }
2752 :
2753 0 : CHK_PRT_RET(
2754 : aiRMAInfoPtr->qpNum <= 0,
2755 : HCCL_ERROR(
2756 : "[%s] invalid qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u]", __func__, tag.c_str(),
2757 : aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum),
2758 : HCCL_E_INTERNAL);
2759 :
2760 0 : return HCCL_SUCCESS;
2761 : }
2762 :
2763 0 : HcclResult HcclCommunicator::GetAIVNormalQPInfoV2(std::vector<LINK>& links, const std::string& tag)
2764 : {
2765 : // 获取 Transport QP 数量
2766 0 : CHK_PTR_NULL(rmaInfoMem_);
2767 0 : HcclRMAInfo* rmaInfoPtr = reinterpret_cast<HcclRMAInfo*>(rmaInfoMem_->ptr());
2768 0 : CHK_PTR_NULL(rmaInfoPtr);
2769 : // 获取 Transport QP 数量（暂时只支持单QP）
2770 0 : rmaInfoPtr->qpNum = HCCL_QPS_PER_CONNECTION_DEFAULT;
2771 0 : for (u32 i = 0; i < links.size(); i++) {
2772 0 : if (i != rmaInfoPtr->curRankId) {
2773 0 : if (links[i]->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2774 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2775 0 : CHK_RET(links[i]->GetAiRMAQueueInfo(aiQpVec));
2776 0 : rmaInfoPtr->qpNum = static_cast<u32>(aiQpVec.size());
2777 0 : break;
2778 0 : }
2779 : }
2780 : }
2781 :
2782 0 : CHK_PRT_RET(
2783 : rmaInfoPtr->qpNum <= 0,
2784 : HCCL_ERROR(
2785 : "[%s] invalid qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u]", __func__, tag.c_str(),
2786 : rmaInfoPtr->curRankId, rmaInfoPtr->rankNum, rmaInfoPtr->qpNum),
2787 : HCCL_E_INTERNAL);
2788 :
2789 0 : return HCCL_SUCCESS;
2790 : }
2791 :
2792 : template <typename T>
2793 0 : HcclResult HcclCommunicator::GenIbvAiRMAInfo(
2794 : u32 rankid, const std::shared_ptr<Transport>& transport, const std::string& tag, T* aiRMAInfoPtr)
2795 : {
2796 0 : HCCL_INFO("[HcclCommunicator][%s] Start prepare.", __func__);
2797 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2798 0 : CHK_RET(transport->GetAiRMAQueueInfo(aiQpVec));
2799 :
2800 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2801 0 : CHK_PRT_RET(
2802 : aiQpVec.size() != aiRMAInfoPtr->qpNum,
2803 : HCCL_ERROR(
2804 : "[%s] different qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u] qpVecNum[%u]", __func__, tag.c_str(),
2805 : aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum, aiQpVec.size()),
2806 : HCCL_E_INTERNAL);
2807 :
2808 0 : HcclAiRMAWQ* aiSqHost = reinterpret_cast<HcclAiRMAWQ*>(aiSqMem_->ptr());
2809 0 : HcclAiRMACQ* aiScqHost = reinterpret_cast<HcclAiRMACQ*>(aiScqMem_->ptr());
2810 0 : HcclAiRMAWQ* aiRqHost = reinterpret_cast<HcclAiRMAWQ*>(aiRqMem_->ptr());
2811 0 : HcclAiRMACQ* aiRcqHost = reinterpret_cast<HcclAiRMACQ*>(aiRcqMem_->ptr());
2812 :
2813 0 : for (u32 j = 0; j < aiRMAInfoPtr->qpNum; j++) {
2814 0 : const auto& aiQpInfo = aiQpVec[j];
2815 0 : u32 qpIndex = rankid * aiRMAInfoPtr->qpNum + j;
2816 0 : aiSqHost[qpIndex] = aiQpInfo.sq;
2817 0 : aiScqHost[qpIndex] = aiQpInfo.scq;
2818 0 : aiRqHost[qpIndex] = aiQpInfo.rq;
2819 0 : aiRcqHost[qpIndex] = aiQpInfo.rcq;
2820 : }
2821 0 : return HCCL_SUCCESS;
2822 0 : }
2823 :
2824 302 : HcclResult HcclCommunicator::SetAivCoreLimit(u32 aivCoreLimit)
2825 : {
2826 302 : numBlocks_ = aivCoreLimit;
2827 302 : return HCCL_SUCCESS;
2828 : }
2829 :
2830 0 : HcclResult HcclCommunicator::GetAlgInfo(const std::string& algConfig, const std::string& tag, std::string& algName)
2831 : {
2832 0 : CHK_PRT_RET(
2833 : (ALGCFG_TO_NAME.find(algConfig) == ALGCFG_TO_NAME.end()),
2834 : HCCL_ERROR("[%s] invalid algConfig=[%s]", __func__, algConfig.c_str()), HCCL_E_PARA);
2835 :
2836 0 : algName = ALGCFG_TO_NAME[algConfig];
2837 0 : HCCL_INFO("[%s] tag=[%s], algName=[%s]", __func__, tag.c_str(), algName.c_str());
2838 0 : return HCCL_SUCCESS;
2839 : }
2840 :
2841 0 : HcclResult HcclCommunicator::SetGroupMode(bool isGroup)
2842 : {
2843 0 : isGroupMode_ = isGroup;
2844 0 : CHK_RET(transportManager_->SetGroupMode(isGroup));
2845 0 : return HCCL_SUCCESS;
2846 : }
2847 :
2848 0 : bool HcclCommunicator::GetGroupMode() const { return isGroupMode_; }
2849 :
2850 0 : HcclResult HcclCommunicator::GetCommUserMemSize(uint64_t& size)
2851 : {
2852 0 : if (!isUserMemRegisted_ || userMemMap_.empty()) {
2853 0 : HCCL_INFO("[HcclCommunicator][%s] get comm user mem size failed", __func__);
2854 0 : return HCCL_E_NOT_FOUND;
2855 : }
2856 0 : size = userMemMap_.begin()->second->size();
2857 0 : return HCCL_SUCCESS;
2858 : }
2859 :
2860 0 : HcclResult HcclCommunicator::GetAivQPInfoV2(std::vector<LINK>& links, const std::string& tag)
2861 : {
2862 0 : HCCL_DEBUG("[HcclCommunicator][%s] Start prepare.", __func__);
2863 : // 获取 Transport QP 数量
2864 0 : CHK_PTR_NULL(aiRMAInfoMem_);
2865 0 : HcclAiRMAInfo* aiRMAInfoPtr = reinterpret_cast<HcclAiRMAInfo*>(aiRMAInfoMem_->ptr());
2866 0 : CHK_PTR_NULL(aiRMAInfoPtr);
2867 : // 获取 Transport QP 数量（暂时只支持单QP）
2868 0 : aiRMAInfoPtr->qpNum = HCCL_QPS_PER_CONNECTION_DEFAULT;
2869 0 : for (u32 i = 0; i < links.size(); i++) { // server num
2870 0 : if (i != (aiRMAInfoPtr->curRankId / meshAggregationRankSize_)) { // 判断server
2871 0 : if (links[i]->GetTransportType() == TransportType::TRANS_TYPE_IBV_EXP) {
2872 0 : std::vector<HcclAiRMAQueueInfo> aiQpVec;
2873 0 : CHK_RET(links[i]->GetAiRMAQueueInfo(aiQpVec));
2874 0 : aiRMAInfoPtr->qpNum = static_cast<u32>(aiQpVec.size());
2875 0 : break;
2876 0 : }
2877 : }
2878 : }
2879 0 : CHK_PRT_RET(
2880 : aiRMAInfoPtr->qpNum <= 0,
2881 : HCCL_ERROR(
2882 : "[%s] invalid qpNum. tag[%s] curRankId[%u] rankNum[%u] qpNum[%u]", __func__, tag.c_str(),
2883 : aiRMAInfoPtr->curRankId, aiRMAInfoPtr->rankNum, aiRMAInfoPtr->qpNum),
2884 : HCCL_E_INTERNAL);
2885 :
2886 0 : return HCCL_SUCCESS;
2887 : }
2888 : } // namespace hccl
|
