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 <atomic>
12 : #include <chrono>
13 : #include <thread>
14 : #include <numeric>
15 : #include <sys/time.h>
16 : #include <dlog_pub.h>
17 : #include "sal.h"
18 :
19 : #include "adapter_prof.h"
20 : #include "dlprof_function.h"
21 : #include "dlrt_function.h"
22 : #include "externalinput_pub.h"
23 : #include "transport_heterog_event_roce.h"
24 : #include "transport_heterog_roce.h"
25 : #include "device_capacity.h"
26 : #include "rank_consistentcy_checker.h"
27 : #include "hccd_impl_pml.h"
28 :
29 : using namespace std;
30 :
31 : namespace hccl {
32 :
33 2 : HccdImplPml::HccdImplPml()
34 2 : : initializedFlag_(ATOMIC_FLAG_INIT),
35 2 : userRank_(INVALID_VALUE_RANKID), realUserRank_(INVALID_VALUE_RANKID), userRankSize_(INVALID_VALUE_RANKSIZE),
36 2 : devicePhyId_(INVALID_UINT),
37 2 : deviceLogicId_(-1),
38 2 : hcomGroupNicInit_(false),
39 2 : heterogRaInit_(false), hostRdmaInitFlag_(false),
40 2 : commHandle_(nullptr), mrManager_(nullptr),
41 2 : pMsgInfosMem_(nullptr), pReqInfosMem_(nullptr), memBlocksManager_(nullptr), pRecvWrInfosMem_(nullptr),
42 2 : transportResourceInfo_(mrManager_, pMsgInfosMem_, pReqInfosMem_, memBlocksManager_, pRecvWrInfosMem_),
43 2 : profilingInitiated_(false),
44 4 : mrManagerInit_(false), srqInit_(false)
45 : {
46 2 : }
47 :
48 2 : HccdImplPml::~HccdImplPml()
49 : {
50 : // 销毁异构通信资源
51 2 : DestroyHeterogTransport();
52 2 : DestroySrq();
53 2 : DeInitTransportMem();
54 2 : MrManagerDeInit();
55 :
56 : /* 网络资源销毁 */
57 2 : DeinitHeterogRaResource();
58 2 : }
59 :
60 0 : HcclResult HccdImplPml::Init(HcclCommParams ¶ms, const RankTable_t &rankTable)
61 : {
62 0 : CHK_RET(InitCommParams(params));
63 :
64 0 : CHK_RET(InitTcpMode(rankTable));
65 :
66 : // 获取serverId
67 0 : CHK_RET(GetServerId(rankTable));
68 :
69 : // 根据server整理rank信息
70 0 : CHK_RET(TransformRankInfoByServerId(rankTable.rankList, servRankInfo_));
71 :
72 : // 生成nicList
73 0 : for (auto iter : servRankInfo_[serverId_]) {
74 0 : if (((!iter.hostIp.IsInvalid()) || (!iter.deviceInfo.deviceIp[0].IsInvalid())) &&
75 0 : (iter.deviceInfo.devicePhyId != HOST_DEVICE_ID)) {
76 0 : nicList_.push_back(iter.deviceInfo.devicePhyId);
77 : }
78 0 : }
79 0 : std::sort(nicList_.begin(), nicList_.end());
80 :
81 : // 解析ranktable信息(生成rankInfoList_)，供给commfactory使用
82 0 : CHK_RET(GetRankInfoList(rankTable));
83 :
84 0 : for (u32 i = 0; i < rankInfoList_.size(); i++) {
85 0 : HCCL_DEBUG(" host ip: %s host port: %u dev phy id: %d", rankInfoList_[i].hostIp.GetReadableAddress(),
86 : rankInfoList_[i].hostPort, rankInfoList_[i].devicePhyId);
87 0 : if (rankInfoList_[i].userRank == userRank_) {
88 0 : devIpAddr_ = rankInfoList_[i].nicIp;
89 0 : devicePhyId_ = rankInfoList_[i].devicePhyId;
90 0 : break;
91 : }
92 : }
93 :
94 0 : ranksPort_.resize(userRankSize_, 0);
95 0 : for (auto rankInfo : rankTable.rankList) {
96 0 : ranksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT || rankInfo.deviceInfo.port == 0
97 0 : ? HETEROG_CCL_PORT : rankInfo.deviceInfo.port;
98 0 : }
99 :
100 : // 在确定 servRankInfo_ 和 serverId_ 信息后，就完成初始判断
101 :
102 0 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
103 0 : HCCL_INFO("init heterog comm, rank id[%u] device id[%u]", userRank_, devicePhyId_);
104 :
105 0 : HcclResult ret = InitPara(rankTable.collectiveId);
106 :
107 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
108 : HCCL_ERROR("[HcclImplBase][Init]errNo[0x%016llx] collectiveid[%s] parameter initialization failed",
109 : HCCL_ERROR_CODE(ret), rankTable.collectiveId.c_str()), ret);
110 :
111 0 : CHK_PRT_RET(devIpAddr_.empty(), HCCL_ERROR("[HcclImplBase][Init]devIpAddr_ size[%llu] "
112 : "should be greater than 0.", devIpAddr_.size()), HCCL_E_UNAVAIL);
113 :
114 0 : if (params.attr.mode != WorkMode::HCCL_MODE_AI_CPU && params.attr.mode != WorkMode::HCCL_MODE_PS) {
115 0 : CHK_RET(InitHeterogRaResource(rankTable));
116 0 : CHK_RET(InitHeterogRecvExecutor());
117 0 : CHK_RET(MrManagerInit());
118 0 : CHK_RET(InitRecvMsgAndRequestBuffer());
119 0 : CHK_RET(InitMemBlocksAndRecvWrMem());
120 0 : CHK_RET(CreateSrq());
121 : }
122 :
123 0 : return HCCL_SUCCESS;
124 : }
125 :
126 0 : HcclResult HccdImplPml::InitCommParams(HcclCommParams ¶ms)
127 : {
128 0 : commHandle_ = params.commHandle;
129 0 : userRank_ = params.rank;
130 0 : realUserRank_ = params.userRank;
131 0 : userRankSize_ = params.totalRanks;
132 0 : deviceLogicId_ = params.logicDevId;
133 0 : profilingOption_ = params.profilingOption;
134 0 : profilingInitiated_ = params.profilingInitiated;
135 0 : hcomGroupNicInit_ = params.hcomGroupNicInit;
136 0 : identifier_ = params.identifier;
137 0 : return HCCL_SUCCESS;
138 : }
139 :
140 0 : HcclResult HccdImplPml::GetServerId(const RankTable_t &rankTable)
141 : {
142 0 : for (u32 i = 0; i < rankTable.rankList.size(); i++) {
143 0 : if (rankTable.rankList[i].rankId == userRank_) {
144 0 : serverId_ = rankTable.rankList[i].serverId;
145 0 : break;
146 : }
147 : }
148 0 : if (serverId_.empty()) {
149 0 : HCCL_ERROR("[Get][ServerId]GetServerId fail");
150 0 : return HCCL_E_PARA;
151 : }
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : HcclResult HccdImplPml::TransformRankInfoByServerId(
156 : const std::vector<RankInfo_t> &rankList, ServRankInfo_t &servRankInfo) const
157 : {
158 : // 按server重新组织rank信息，便于后续校验及信息填写
159 0 : for (size_t index = 0; index < rankList.size(); ++index) {
160 0 : const RankInfo_t &rankInfo = rankList[index];
161 0 : std::string serverId = SalTrim(rankInfo.serverId);
162 : // 以serverID为索引，将server下的ranks放入vector
163 0 : ServRankInfo_t::iterator itr = servRankInfo.find(serverId);
164 0 : if (itr != servRankInfo.end()) {
165 0 : itr->second.push_back(rankInfo);
166 : } else {
167 0 : std::vector<RankInfo_t> rankInfoList;
168 0 : rankInfoList.push_back(rankInfo);
169 0 : std::pair<std::string, std::vector<RankInfo_t>> rankInfoPair(serverId, rankInfoList);
170 0 : servRankInfo.insert(rankInfoPair);
171 0 : }
172 0 : }
173 : // 每个server下的rank列表按设备Id从小到大的顺序排序
174 0 : for (auto &iter : servRankInfo) {
175 0 : std::sort(iter.second.begin(), iter.second.end(), CompareWithDevicePhyId);
176 : }
177 0 : return HCCL_SUCCESS;
178 : }
179 :
180 0 : bool HccdImplPml::CompareWithDevicePhyId(const RankInfo_t &left, const RankInfo_t &right)
181 : {
182 0 : return left.deviceInfo.devicePhyId < right.deviceInfo.devicePhyId;
183 : }
184 :
185 1 : HcclResult HccdImplPml::InitTcpMode(const RankTable_t &rankTable) const
186 : {
187 1 : bool isTcpMode = false;
188 1 : HCCL_INFO("[TcpMode][%u] [1:TCP, 2:RDMA, 3:RESERVED]", GetExternalInputProtocolType());
189 1 : if (GetExternalInputProtocolType() == ProtocolType::TCP) {
190 0 : isTcpMode = true;
191 1 : } else if (GetExternalInputProtocolType() == ProtocolType::RDMA) {
192 : // 通信协议选择RDMA
193 : } else {
194 1 : isTcpMode = (rankTable.nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST);
195 1 : HCCL_INFO("[Init][TcpMode]isTcpMode[%d] nicDeploy[%d]", isTcpMode, rankTable.nicDeploy);
196 : }
197 1 : SetTcpMode(isTcpMode);
198 1 : return HCCL_SUCCESS;
199 : }
200 :
201 0 : HcclResult HccdImplPml::GetRankInfoList(const RankTable_t &rankTable)
202 : {
203 : // 遍历rank table获取rank信息
204 0 : rankInfoList_.clear();
205 0 : for (auto iter = servRankInfo_.begin(); iter != servRankInfo_.end(); ++iter) {
206 0 : for (u32 index = 0; index < iter->second.size(); ++index) {
207 0 : const RankInfo_t &orgRankInfo = iter->second[index];
208 : // 构建comm 使用的rank 信息
209 0 : RankInfo rankInfo;
210 0 : rankInfo.userRank = orgRankInfo.rankId;
211 0 : rankInfo.worldRank = orgRankInfo.rankId;
212 0 : rankInfo.devicePhyId = orgRankInfo.deviceInfo.devicePhyId;
213 :
214 0 : rankInfo.serverId = orgRankInfo.serverId;
215 0 : rankInfo.serverIdx = orgRankInfo.serverIdx;
216 0 : rankInfo.hostIp = orgRankInfo.hostIp;
217 0 : rankInfo.hostPort = orgRankInfo.hostPort;
218 0 : rankInfo.localRank = orgRankInfo.localRank;
219 0 : rankInfo.superPodId = orgRankInfo.superPodId;
220 0 : CHK_RET(GetNicInfo(rankTable.nicDeploy, index, iter->second, rankInfo));
221 0 : rankInfo.nicIdx.assign(nicList_.begin(), nicList_.end());
222 0 : rankInfoList_.push_back(rankInfo);
223 0 : }
224 : }
225 : // 将rank id从小到大的顺序返回
226 0 : CHK_RET(SortRankInfoList());
227 :
228 0 : return HCCL_SUCCESS;
229 : }
230 :
231 0 : HcclResult HccdImplPml::GetNicInfo(const NICDeployment &nicDeploy, const u32 curRankIndex,
232 : const std::vector<RankInfo_t> &servRankList, RankInfo &rankInfo) const
233 : {
234 0 : CHK_PRT_RET(servRankList.empty(), HCCL_ERROR("[Get][NicInfo]errNo[0x%016llx] server rank list is empty",
235 : HCCL_ERROR_CODE(HCCL_E_PARA)), HCCL_E_PARA);
236 :
237 0 : rankInfo.nicDeploy = nicDeploy;
238 0 : if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
239 : // 检查网卡个数
240 : // 网卡挂载位置在host时，按rank index从网卡列表中获取
241 0 : const RankInfo_t &curRankInfo = servRankList[curRankIndex];
242 0 : rankInfo.nicIp.push_back(curRankInfo.hostIp);
243 : } else {
244 0 : CHK_PRT_RET(curRankIndex >= servRankList.size(), HCCL_ERROR("[Get][NicInfo]rankindex[%u] invalid,rank list "\
245 : "size is[%zu]", curRankIndex, servRankList.size()), HCCL_E_PARA);
246 :
247 0 : const RankInfo_t &curRankInfo = servRankList[curRankIndex];
248 0 : CHK_PRT_RET(curRankInfo.deviceInfo.deviceIp.size() == 0,
249 : HCCL_ERROR("[Get][NicInfo]rankindex[%u] invalid,deviceIp is zero", curRankIndex), HCCL_E_PARA);
250 0 : rankInfo.nicIp.push_back(curRankInfo.deviceInfo.deviceIp[0]);
251 : }
252 :
253 0 : return HCCL_SUCCESS;
254 : }
255 :
256 0 : HcclResult HccdImplPml::SortRankInfoList()
257 : {
258 : // 按rank id从小到大的顺序返回
259 0 : std::sort(rankInfoList_.begin(), rankInfoList_.end(), CompareWithUserRank);
260 :
261 0 : for (u32 index = 0; index < rankInfoList_.size(); ++index) {
262 0 : CHK_PRT_RET((index != rankInfoList_[index].userRank),
263 : HCCL_ERROR("[HcclImplBase][SortRankInfoList]errNo[0x%016llx] index[%u] != rankInfoList.userRank[%u]",
264 : HCCL_ERROR_CODE(HCCL_E_PARA), index, rankInfoList_[index].userRank), HCCL_E_PARA);
265 : }
266 0 : return HCCL_SUCCESS;
267 : }
268 :
269 0 : bool HccdImplPml::CompareWithUserRank(const RankInfo &left, const RankInfo &right)
270 : {
271 0 : return left.userRank < right.userRank;
272 : }
273 :
274 0 : HcclResult HccdImplPml::InitPara(const std::string &colectiveId)
275 : {
276 : // 检查当前user_rank 对应的devid和rt查到的一致
277 0 : for (u32 i = 0; i < rankInfoList_.size(); ++i) {
278 0 : if ((userRank_ == rankInfoList_[i].userRank) &&
279 0 : (static_cast<s32>(devicePhyId_) != rankInfoList_[i].devicePhyId)) {
280 0 : HCCL_ERROR("[Init][Para]errNo[0x%016llx] parameter check failed,userrank[%u] == rankInfoList.userrank[%u],"\
281 : "phyid[%d] != rankInfoList.devid[%d]", HCCL_ERROR_CODE(HCCL_E_PARA), userRank_,
282 : rankInfoList_[i].userRank, static_cast<s32>(devicePhyId_), rankInfoList_[i].devicePhyId);
283 0 : return HCCL_E_PARA;
284 : }
285 : }
286 0 : collectiveId_ = colectiveId;
287 :
288 0 : workSpaceRes_.reset(new (std::nothrow) WorkspaceResource(devicePhyId_, deviceLogicId_));
289 0 : CHK_SMART_PTR_NULL(workSpaceRes_);
290 :
291 0 : return HCCL_SUCCESS;
292 : }
293 :
294 0 : HcclResult HccdImplPml::InitHeterogRaResource(const RankTable_t &rankTable)
295 : {
296 0 : CHK_PRT_RET(rankTable.rankList.size() != userRankSize_, HCCL_ERROR("[Init][HeterogRaResourc] rank list size[%u]" \
297 : " is different from user rank size[%u]", rankTable.rankList.size(), userRankSize_), HCCL_E_PARA);
298 0 : ranksPort_.resize(userRankSize_, 0);
299 0 : for (auto rankInfo : rankTable.rankList) {
300 0 : ranksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT || rankInfo.deviceInfo.port == 0
301 0 : ? HETEROG_CCL_PORT : rankInfo.deviceInfo.port;
302 0 : }
303 :
304 0 : heterogRaInit_ = true;
305 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).HeterogInit(devicePhyId_, devIpAddr_[0],
306 : ranksPort_[userRank_]));
307 0 : if (!GetExternalInputHcclIsTcpMode()) {
308 0 : hostRdmaInitFlag_ = true;
309 : }
310 :
311 0 : return HCCL_SUCCESS;
312 : }
313 :
314 0 : HcclResult HccdImplPml::InitRecvMsgAndRequestBuffer()
315 : {
316 : // 拉远、下沉、推理场景(ps、worker)支持使用msg/request内存池
317 0 : if (pMsgInfosMem_ == nullptr) {
318 0 : pMsgInfosMem_.reset(new (std::nothrow) LocklessRingMemoryAllocate<HcclMessageInfo>(MEMORY_CAPACITY));
319 0 : CHK_SMART_PTR_NULL(pMsgInfosMem_);
320 0 : CHK_RET(pMsgInfosMem_->Init());
321 0 : HCCL_INFO("InitRecvMsgBuffer Success!");
322 : }
323 :
324 0 : if (pReqInfosMem_ == nullptr) {
325 0 : pReqInfosMem_.reset(new (std::nothrow) LocklessRingMemoryAllocate<HcclRequestInfo>(MEMORY_CAPACITY));
326 0 : CHK_SMART_PTR_NULL(pReqInfosMem_);
327 0 : CHK_RET(pReqInfosMem_->Init());
328 0 : HCCL_INFO("InitRequestBuffer Success!");
329 : }
330 :
331 0 : return HCCL_SUCCESS;
332 : }
333 :
334 0 : HcclResult HccdImplPml::InitMemBlocksAndRecvWrMem()
335 : {
336 0 : u32 memBlockNum = MEM_BLOCK_NUM;
337 0 : CHK_PRT(GetMemBlockNum(devicePhyId_, memBlockNum));
338 :
339 0 : if (!GetExternalInputHcclIsTcpMode()) {
340 : // 初始化信封内存
341 0 : memBlocksManager_.reset(new (std::nothrow) HeterogMemBlocksManager());
342 0 : CHK_SMART_PTR_NULL(memBlocksManager_);
343 0 : CHK_RET(memBlocksManager_->Init(memBlockNum));
344 :
345 : // 信封内存注册
346 0 : CHK_RET(mrManager_->GetKey(memBlocksManager_->GetMemAddr(), memBlocksManager_->GetMemSize(),
347 : transportResourceInfo_.lkey));
348 :
349 : // 初始化wr内存
350 0 : pRecvWrInfosMem_.reset(new (std::nothrow) LocklessRingMemoryAllocate<RecvWrInfo>(MEMORY_CAPACITY));
351 0 : CHK_SMART_PTR_NULL(pRecvWrInfosMem_);
352 0 : CHK_RET(pRecvWrInfosMem_->Init());
353 0 : HCCL_INFO("InitMemBlocksAndRecvWrMem Success!");
354 : }
355 :
356 0 : return HCCL_SUCCESS;
357 : }
358 :
359 0 : HcclResult HccdImplPml::CreateSrq()
360 : {
361 0 : u32 info = 0;
362 0 : CHK_RET(hrtDrvGetPlatformInfo(&info));
363 0 : if (info == 0) {
364 0 : std::string chipName;
365 0 : HcclResult ret = hrtHalGetChipInfo(devicePhyId_, chipName);
366 0 : if (ret == HCCL_SUCCESS) {
367 0 : if (chipName.find(SOC_NAME_910B) != std::string::npos) {
368 0 : HCCL_INFO("not support chip[%s] create srq", chipName.c_str());
369 0 : return HCCL_SUCCESS;
370 : }
371 : }
372 0 : }
373 :
374 0 : if (!srqInit_ && !GetExternalInputHcclIsTcpMode()) {
375 0 : RaResourceInfo raResourceInfo;
376 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
377 0 : void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;
378 :
379 : // 创建srq
380 0 : transportResourceInfo_.tagSrqInfo.srqEvent = HCCL_EVENT_RECV_REQUEST_MSG;
381 0 : transportResourceInfo_.dataSrqInfo.srqEvent = HCCL_EVENT_SEND_COMPLETION_MSG;
382 :
383 0 : transportResourceInfo_.tagSrqInfo.srqDepth = MAX_SRQ_DEPTH;
384 0 : transportResourceInfo_.dataSrqInfo.srqDepth = MAX_SRQ_DEPTH;
385 0 : CHK_RET(hrtRaCreateSrq(nicRdmaHandle, transportResourceInfo_.tagSrqInfo));
386 0 : CHK_RET(hrtRaCreateSrq(nicRdmaHandle, transportResourceInfo_.dataSrqInfo));
387 0 : HCCL_INFO("CreateSrq Success!");
388 :
389 0 : std::unique_ptr<TransportHeterogEventRoce> transportPtr;
390 0 : transportPtr.reset(new (std::nothrow) TransportHeterogEventRoce(transportResourceInfo_));
391 0 : CHK_SMART_PTR_NULL(transportPtr);
392 0 : CHK_RET(transportPtr->InitSrqRecvWqe());
393 :
394 0 : srqInit_ = true;
395 0 : }
396 :
397 0 : return HCCL_SUCCESS;
398 : }
399 :
400 0 : HcclResult HccdImplPml::AtomicInitSet()
401 : {
402 0 : CHK_PRT_RET(initializedFlag_.test_and_set(), HCCL_ERROR("[HcclImplBase][AtomicInitSet]errNo[0x%016llx] instance "\
403 : "already been initialized", HCCL_ERROR_CODE(HCCL_E_INTERNAL)), HCCL_E_INTERNAL);
404 0 : return HCCL_SUCCESS;
405 : }
406 :
407 0 : void HccdImplPml::AtomicInitClear()
408 : {
409 0 : initializedFlag_.clear();
410 0 : }
411 :
412 0 : HcclResult HccdImplPml::RegisterMemory(void* buffer, uint64_t size)
413 : {
414 : // 拉远、推理场景PS侧支持注册全局内存
415 0 : if (hostRdmaInitFlag_) {
416 0 : CHK_RET(mrManager_->RegGlobalMr(buffer, size));
417 : }
418 0 : return HCCL_SUCCESS;
419 : }
420 :
421 0 : HcclResult HccdImplPml::UnregisterMemory(void* buffer)
422 : {
423 0 : if (hostRdmaInitFlag_) {
424 0 : CHK_RET(mrManager_->DeRegGlobalMr(buffer));
425 : }
426 :
427 0 : return HCCL_SUCCESS;
428 : }
429 :
430 0 : HcclResult HccdImplPml::CheckCount(const u64 count) const
431 : {
432 0 : if (count > SYS_MAX_COUNT) {
433 0 : HCCL_ERROR("[Check][Count]errNo[0x%016llx] count[%llu] is invalid(bigger than MAX count[%llu])",
434 : HCCL_ERROR_CODE(HCCL_E_PARA), count, SYS_MAX_COUNT);
435 0 : return HCCL_E_PARA;
436 : }
437 0 : return HCCL_SUCCESS;
438 : }
439 :
440 4 : HcclResult HccdImplPml::CheckDataType(const HcclDataType dataType, bool needReduce)
441 : {
442 4 : if (needReduce) {
443 3 : if ((dataType == HCCL_DATA_TYPE_UINT64) ||
444 2 : (dataType == HCCL_DATA_TYPE_UINT8) || (dataType == HCCL_DATA_TYPE_UINT16) ||
445 1 : (dataType == HCCL_DATA_TYPE_UINT32) || (dataType == HCCL_DATA_TYPE_FP64) ||
446 : (dataType == HCCL_DATA_TYPE_RESERVED)) {
447 2 : HCCL_ERROR("[Check][DataType]errNo[0x%016llx] data type[%s] not supported",
448 : HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), GetDataTypeEnumStr(dataType).c_str());
449 2 : return HCCL_E_NOT_SUPPORT;
450 : }
451 : } else {
452 1 : if ((dataType >= HCCL_DATA_TYPE_RESERVED) || (dataType < HCCL_DATA_TYPE_INT8)) {
453 1 : HCCL_ERROR("[Check][DataType]errNo[0x%016llx] data type[%s] not supported",
454 : HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), GetDataTypeEnumStr(dataType).c_str());
455 1 : return HCCL_E_NOT_SUPPORT;
456 : }
457 : }
458 1 : return HCCL_SUCCESS;
459 : }
460 :
461 0 : HcclResult HccdImplPml::Isend(void *buffer, s32 count, HcclDataType dataType, u32 peerRank, s32 tag,
462 : HcclRequest &requestHandle, u32 userRequire)
463 : {
464 0 : if ((buffer == nullptr) && (count != 0)) {
465 0 : HCCL_ERROR("[Check][Buffer]errNo[0x%016llx] buffer[%p] or count[%d] is invalid",
466 : HCCL_ERROR_CODE(HCCL_E_PARA), buffer, count);
467 0 : return HCCL_E_PARA;
468 : }
469 0 : if (peerRank >= userRankSize_) {
470 0 : HCCL_ERROR("[Check][UserRank]errNo[0x%016llx] peerRank:[%u] is out of range[0 ~ %u]",
471 : HCCL_ERROR_CODE(HCCL_E_PARA), peerRank, userRankSize_);
472 0 : return HCCL_E_PARA;
473 : }
474 :
475 0 : TransportHandle transportHandle = nullptr;
476 0 : CHK_RET(BuildHeterogeneousTransport(0, peerRank, tag, transportHandle));
477 :
478 0 : TransportHeterog *transportPtr = reinterpret_cast<TransportHeterog *>(transportHandle);
479 0 : HcclRequestInfo* request = nullptr;
480 : TransData sendData(reinterpret_cast<u64>(buffer), reinterpret_cast<u64>(nullptr), count, dataType, false,
481 0 : userRequire);
482 0 : TransportEndPointInfo srcEp(0, userRank_, tag);
483 0 : TransportEndPointInfo dstEp(0, peerRank, tag);
484 0 : TransportEndPointParam epParam(srcEp, dstEp);
485 0 : CHK_RET(transportPtr->Isend(sendData, epParam, request));
486 0 : request->commHandle = commHandle_;
487 0 : requestHandle = request;
488 0 : return HCCL_SUCCESS;
489 : }
490 :
491 0 : HcclResult HccdImplPml::BuildHeterogeneousTransport(u32 commId, u32 peerRank, s32 tag,
492 : TransportHandle &transportHandle)
493 : {
494 0 : TransportEndPointInfo commRankTagKey(commId, peerRank, tag);
495 0 : std::unique_lock<SpinMutex> transportMapLock(transportMapSpinMutex_);
496 0 : std::unique_ptr<TransportHeterog>& transportInfo = transportStorage_[commRankTagKey];
497 0 : transportMapLock.unlock();
498 :
499 0 : if (transportInfo == nullptr) {
500 0 : std::string transTag;
501 0 : if (userRank_ > peerRank) {
502 0 : transTag = collectiveId_ + "_" + std::to_string(peerRank) + "_" + std::to_string(userRank_) + "_";
503 : } else {
504 0 : transTag = collectiveId_ + "_" + std::to_string(userRank_) + "_" + std::to_string(peerRank) + "_";
505 : }
506 0 : transTag += std::to_string(tag);
507 0 : std::unique_ptr<TransportHeterog> transportPtr;
508 : // 当前代码只保留TransportHeterogEventRoce
509 0 : transportPtr.reset(new (std::nothrow) TransportHeterogEventRoce(transTag, rankInfoList_[userRank_].nicIp[0],
510 0 : rankInfoList_[peerRank].nicIp[0], ranksPort_[peerRank], ranksPort_[userRank_], transportResourceInfo_));
511 0 : CHK_SMART_PTR_NULL(transportPtr);
512 0 : CHK_RET(transportPtr->SetDeviceIndex(deviceLogicId_));
513 0 : CHK_RET(transportPtr->Init());
514 0 : transportInfo = std::move(transportPtr);
515 0 : }
516 0 : transportHandle = transportInfo.get();
517 0 : CHK_PTR_NULL(transportHandle);
518 0 : return HCCL_SUCCESS;
519 0 : }
520 :
521 0 : HcclResult HccdImplPml::Improbe(u32 peerRank, s32 tag, s32 &flag, HcclMessage &msgHandle, HcclStatus &status)
522 : {
523 0 : if (peerRank >= userRankSize_) {
524 0 : HCCL_ERROR("[Check][UserRank]errNo[0x%016llx] peerRank:[%u] is out of range[0 ~ %u]",
525 : HCCL_ERROR_CODE(HCCL_E_PARA), peerRank, userRankSize_);
526 0 : return HCCL_E_PARA;
527 : }
528 :
529 0 : void* transportHandle = nullptr;
530 0 : CHK_RET(BuildHeterogeneousTransport(0, peerRank, tag, transportHandle));
531 :
532 0 : TransportHeterog *transportPtr = reinterpret_cast<TransportHeterog *>(transportHandle);
533 0 : TransportEndPointInfo srcEp(0, peerRank, tag);
534 0 : TransportEndPointInfo dstEp(0, userRank_, tag);
535 0 : TransportEndPointParam epParam(srcEp, dstEp);
536 0 : HcclMessageInfo *msg = nullptr;
537 0 : CHK_RET(transportPtr->Improbe(epParam, flag, msg, status));
538 0 : msgHandle = msg;
539 0 : return HCCL_SUCCESS;
540 : }
541 :
542 0 : HcclResult HccdImplPml::Imrecv(void* buffer, s32 count, HcclDataType dataType, HcclMessage msgHandle,
543 : HcclRequest &requestHandle)
544 : {
545 0 : HcclMessageInfo* msg = static_cast<HcclMessageInfo *>(msgHandle);
546 0 : CHK_PTR_NULL(msg);
547 0 : TransportHeterog *transportPtr = reinterpret_cast<TransportHeterog *>(msg->transportHandle);
548 0 : CHK_PTR_NULL(transportPtr);
549 :
550 0 : HcclRequestInfo* request = nullptr;
551 0 : TransData recvData(reinterpret_cast<u64>(nullptr), reinterpret_cast<u64>(buffer), count, dataType);
552 0 : CHK_RET(transportPtr->Imrecv(recvData, *msg, request));
553 0 : requestHandle = request;
554 0 : return HCCL_SUCCESS;
555 : }
556 :
557 0 : HcclResult HccdImplPml::HcclTest(HcclRequest requestHandle, s32 &flag, HcclStatus &compState)
558 : {
559 0 : HcclRequestInfo *request = reinterpret_cast<HcclRequestInfo *>(requestHandle);
560 0 : CHK_PTR_NULL(request->transportHandle);
561 :
562 0 : TransportHeterog *transportPtr = reinterpret_cast<TransportHeterog *>(request->transportHandle);
563 0 : return transportPtr->Test(*request, flag, compState);
564 : }
565 :
566 0 : u32 HccdImplPml::GetUserRank()
567 : {
568 0 : return realUserRank_;
569 : }
570 :
571 0 : u32 HccdImplPml::GetRankSize()
572 : {
573 0 : return userRankSize_;
574 : }
575 :
576 2 : void HccdImplPml::DestroyHeterogTransport()
577 : {
578 2 : std::unique_lock<SpinMutex> transportMapLock(transportMapSpinMutex_);
579 2 : transportStorage_.clear();
580 4 : return;
581 2 : }
582 :
583 2 : HcclResult HccdImplPml::DestroySrq()
584 : {
585 2 : if (srqInit_) {
586 0 : RaResourceInfo raResourceInfo;
587 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
588 0 : void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;
589 :
590 : // 销毁srq
591 0 : CHK_RET(hrtRaDestroySrq(nicRdmaHandle, transportResourceInfo_.tagSrqInfo));
592 0 : CHK_RET(hrtRaDestroySrq(nicRdmaHandle, transportResourceInfo_.dataSrqInfo));
593 0 : transportResourceInfo_.tagSrqInfo = SrqInfo();
594 0 : transportResourceInfo_.dataSrqInfo = SrqInfo();
595 0 : HCCL_INFO("DestroySrq Success!");
596 0 : srqInit_ = false;
597 0 : }
598 :
599 2 : return HCCL_SUCCESS;
600 : }
601 :
602 2 : HcclResult HccdImplPml::DeInitTransportMem()
603 : {
604 2 : if (memBlocksManager_ != nullptr) {
605 : // 解注册内存
606 0 : CHK_RET(mrManager_->ReleaseKey(memBlocksManager_->GetMemAddr(), memBlocksManager_->GetMemSize()));
607 0 : memBlocksManager_ = nullptr;
608 : }
609 :
610 2 : if (pMsgInfosMem_ != nullptr) {
611 0 : pMsgInfosMem_ = nullptr;
612 : }
613 :
614 2 : if (pReqInfosMem_ != nullptr) {
615 0 : pReqInfosMem_ = nullptr;
616 : }
617 :
618 2 : if (pRecvWrInfosMem_ != nullptr) {
619 0 : pRecvWrInfosMem_ = nullptr;
620 : }
621 :
622 2 : HCCL_INFO("DeInitTransportMem Success!");
623 2 : return HCCL_SUCCESS;
624 : }
625 :
626 0 : HcclResult HccdImplPml::MrManagerInit()
627 : {
628 : // 拉远、下沉、推理场景(ps、worker)支持使用mrManager
629 0 : if (!GetExternalInputHcclIsTcpMode()) {
630 0 : mrManager_.reset(new (std::nothrow) MrManager());
631 0 : CHK_SMART_PTR_NULL(mrManager_);
632 :
633 0 : RaResourceInfo raResourceInfo;
634 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
635 0 : void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;
636 :
637 0 : CHK_RET(mrManager_->Init(nicRdmaHandle));
638 0 : mrManagerInit_ = true;
639 0 : }
640 0 : return HCCL_SUCCESS;
641 : }
642 :
643 2 : HcclResult HccdImplPml::MrManagerDeInit()
644 : {
645 2 : if (mrManagerInit_) {
646 0 : RaResourceInfo raResourceInfo;
647 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
648 0 : void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;
649 :
650 0 : CHK_SMART_PTR_NULL(mrManager_);
651 0 : CHK_RET(mrManager_->DeInit(nicRdmaHandle));
652 0 : mrManager_ = nullptr;
653 0 : mrManagerInit_ = false;
654 0 : }
655 2 : return HCCL_SUCCESS;
656 : }
657 :
658 2 : HcclResult HccdImplPml::DeinitHeterogRaResource()
659 : {
660 2 : if (heterogRaInit_) {
661 0 : HCCL_INFO("deinit heterog ra resource!");
662 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).HeterogDeinit(devicePhyId_, devIpAddr_[0],
663 : ranksPort_[userRank_]));
664 0 : heterogRaInit_ = false;
665 : }
666 2 : return HCCL_SUCCESS;
667 : }
668 :
669 0 : HcclResult HccdImplPml::InitHeterogRecvExecutor() const
670 : {
671 0 : std::vector<SocketWlistInfoT> whiteList(userRankSize_);
672 0 : for (u32 i = 0; i < rankInfoList_.size(); i++) {
673 0 : whiteList[i].remoteIp.addr = rankInfoList_[i].nicIp[0].GetBinaryAddress().addr;
674 0 : whiteList[i].remoteIp.addr6 = rankInfoList_[i].nicIp[0].GetBinaryAddress().addr6;
675 0 : whiteList[i].connLimit = CONN_LIMIT;
676 : }
677 :
678 0 : RaResourceInfo raResourceInfo;
679 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
680 0 : SocketHandle nicSocketHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicSocketHandle;
681 :
682 0 : HCCL_DEBUG("ip[%s] device[%d]", devIpAddr_[0].GetReadableAddress(), deviceLogicId_);
683 :
684 0 : CHK_RET(hrtRaSocketWhiteListAdd(nicSocketHandle, whiteList.data(), userRankSize_));
685 :
686 0 : return HCCL_SUCCESS;
687 0 : }
688 :
689 0 : std::string HccdImplPml::GetUniqueId(void)
690 : {
691 : static std::atomic<u32> idCounter(0);
692 :
693 0 : std::string uniqueId("");
694 0 : uniqueId += std::to_string(SalGetPid());
695 0 : uniqueId += '-';
696 0 : uniqueId += std::to_string(idCounter.fetch_add(1));
697 0 : uniqueId += '-';
698 0 : uniqueId += std::to_string(SalGetSysTime());
699 :
700 0 : return uniqueId;
701 0 : }
702 :
703 : }
|
