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