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 "coll_comm.h"
12 : #include "exception_handler.h"
13 : #include "rank_graph_v2.h"
14 : #include "kfc.h"
15 : #include "dlhal_function.h"
16 : #include "hcclCommTaskException.h"
17 : #include "symmetric_memory/symmetric_memory.h"
18 : #include "hccl_team_mgr.h"
19 :
20 : #include <cstdint>
21 : #include <exception>
22 : #include "launch_aicpu.h"
23 : #include "launch_device.h"
24 :
25 : namespace hccl {
26 166 : void SymmetricMemoryDeleter::operator()(SymmetricMemory* ptr) const { delete ptr; }
27 :
28 243 : CollComm::CollComm(
29 : void* comm, uint32_t rankId, const std::string& commName, const ManagerCallbacks& callbacks,
30 243 : CollCommInitMode initMode)
31 243 : : comm_(comm),
32 243 : rankId_(rankId),
33 243 : commId_(commName),
34 243 : config_(commName),
35 243 : callbacks_(callbacks),
36 486 : initMode_(initMode)
37 : {
38 243 : groupScheduleMgr = std::make_shared<GroupScheduleMgr>();
39 243 : }
40 :
41 448 : CollComm::~CollComm()
42 : {
43 243 : if (!IsFullMode()) { // SimpleMode是简化版collComm，只初始化了myrank、rankgraph未初始化以下资源，不需要析构处理
44 38 : return;
45 : }
46 :
47 : // 先注销TaskException，再销毁通信域资源，防止通信域资源销毁后rts回调TaskException
48 205 : hcomm::TaskExceptionHost* handler = hcomm::TaskExceptionHost::GetInstance(deviceLogicId_);
49 205 : if (handler != nullptr) {
50 205 : (void)handler->UnRegister(reinterpret_cast<u64>(this));
51 : }
52 :
53 205 : CHK_PRT(HcclBinaryUnLoad());
54 :
55 : // 兜底释放所有team的syncMem本地内存
56 205 : HcclTeamMgr::GetInstance().ClearByCollComm(this);
57 205 : HCCL_INFO("[CollComm][~CollComm] collComm deinit");
58 205 : (void)DestroyAicpuComm();
59 205 : HCCL_RUN_INFO("[CollComm][~CollComm] cclBuffer free, commId[%s].", commId_.c_str());
60 813 : }
61 :
62 165 : HcclResult CollComm::Init(void* rankGraph, aclrtBinHandle binHandle, HcclMem cclBuffer, uint32_t opExpansionMode)
63 : {
64 165 : if (IsFullMode()) { // A5和下一代
65 165 : return InitFullMode(rankGraph, binHandle, cclBuffer, opExpansionMode);
66 : } else { // A2/A3使用简化版CollComm
67 0 : return InitSimpleMode(rankGraph, binHandle, cclBuffer, opExpansionMode);
68 : }
69 : }
70 :
71 : HcclResult
72 0 : CollComm::InitSimpleMode(void* rankGraph, aclrtBinHandle binHandle, HcclMem cclBuffer, uint32_t opExpansionMode)
73 : {
74 0 : CHK_PTR_NULL(rankGraph);
75 :
76 : EXCEPTION_HANDLE_BEGIN
77 :
78 0 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
79 :
80 : // SimpleMode: A2/A3的RankGraph是保存在hccl::Communicator中的静态对象裸指针，CollComm不负责释放
81 0 : rankgraph_ = static_cast<RankGraph*>(rankGraph);
82 :
83 0 : uint32_t rankNum = 0;
84 0 : CHK_PTR_NULL(rankgraph_);
85 0 : CHK_RET(rankgraph_->GetRankSize(&rankNum));
86 :
87 0 : EXCEPTION_CATCH(
88 : myRank_ = std::make_shared<MyRank>(binHandle, rankId_, config_, callbacks_, rankgraph_, rankIpPortMap_),
89 : return HCCL_E_PTR);
90 :
91 0 : CHK_RET(myRank_->Init(cclBuffer, opExpansionMode, rankNum));
92 :
93 0 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
94 :
95 0 : EXCEPTION_HANDLE_END
96 0 : return HCCL_SUCCESS;
97 : }
98 :
99 : HcclResult
100 165 : CollComm::InitFullMode(void* rankGraph, aclrtBinHandle binHandle, HcclMem cclBuffer, uint32_t opExpansionMode)
101 : {
102 165 : CHK_PTR_NULL(rankGraph);
103 :
104 : EXCEPTION_HANDLE_BEGIN
105 :
106 164 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
107 164 : rankGraphOwner_ = std::make_unique<RankGraphV2>(rankGraph);
108 164 : rankgraph_ = rankGraphOwner_.get();
109 164 : uint32_t rankNum = 0;
110 164 : CHK_PTR_NULL(rankgraph_);
111 164 : CHK_RET(rankgraph_->GetRankSize(&rankNum));
112 164 : CHK_RET(GetRankIpPortMap());
113 :
114 164 : u32 threadNum = 0xffffffff;
115 164 : u32 notifyNumPerThread = 0xffffffff;
116 164 : if (!commEngineResMgr_) {
117 164 : EXCEPTION_CATCH(commEngineResMgr_ = std::make_unique<CommEngineResMgr>(), return HCCL_E_PTR);
118 164 : CHK_PRT(commEngineResMgr_->Init(threadNum, notifyNumPerThread, commId_, binHandle, callbacks_));
119 : }
120 :
121 164 : if (!contextMgr_) {
122 164 : EXCEPTION_CATCH(contextMgr_ = std::make_unique<ContextManager>(), return HCCL_E_PTR);
123 : }
124 :
125 164 : EXCEPTION_CATCH(
126 : myRank_ = std::make_shared<MyRank>(binHandle, rankId_, config_, callbacks_, rankgraph_, rankIpPortMap_),
127 : return HCCL_E_PTR);
128 164 : CHK_RET(myRank_->Init(cclBuffer, opExpansionMode, rankNum));
129 164 : CHK_RET(hrtGetDevice(&deviceLogicId_));
130 164 : CHK_RET(InitSymmetricMemory());
131 :
132 164 : CHK_RET(InitHDCommunicate());
133 :
134 164 : if (!hcclCommDfx_) {
135 164 : EXCEPTION_CATCH(hcclCommDfx_ = std::make_unique<HcclCommDfx>(), return HCCL_E_PTR);
136 : }
137 164 : CHK_RET(hcclCommDfx_->Init(deviceLogicId_, commId_, rankId_));
138 164 : CHK_RET(InitTaskExceptionHandler());
139 :
140 164 : CHK_RET(InitKfcAndRegisterCollComm());
141 :
142 164 : Hccl::HcclCommunicator* comV2 = static_cast<Hccl::HcclCommunicator*>(comm_);
143 164 : CHK_PTR_NULL(comV2);
144 164 : CHK_RET(comV2->GetCclBufferSharedPtr(cclBuffer_));
145 :
146 0 : EXCEPTION_HANDLE_END
147 164 : return HCCL_SUCCESS;
148 : }
149 :
150 164 : HcclResult CollComm::InitSymmetricMemory()
151 : {
152 164 : uint32_t rankSize = GetRankSize();
153 164 : HCCL_RUN_INFO(
154 : "[CollComm][InitSymmetricMemory] commId[%s], rank[%u], rankSize[%u].", commId_.c_str(), rankId_, rankSize);
155 :
156 164 : EXCEPTION_CATCH(
157 : symmetricMemory_.reset(new SymmetricMemory(rankId_, rankSize, 0, SymmetricMemoryMode::URMA)),
158 : return HCCL_E_PTR);
159 164 : CHK_SMART_PTR_NULL(symmetricMemory_);
160 164 : return HCCL_SUCCESS;
161 : }
162 :
163 1 : HcclResult CollComm::RegisterSymmetricMemoryResource(void* ptr, size_t size, SymmetricMemoryResource& resource)
164 : {
165 1 : CHK_PTR_NULL(ptr);
166 1 : CHK_PRT_RET(
167 : size == 0, HCCL_ERROR("[CollComm][RegisterSymmetricMemoryResource] invalid symmetric memory size 0."),
168 : HCCL_E_PARA);
169 1 : CHK_SMART_PTR_NULL(myRank_);
170 :
171 1 : CommMems* commMems = myRank_->GetCommMems();
172 1 : CHK_PTR_NULL(commMems);
173 :
174 1 : CommMem commMem{};
175 1 : commMem.type = COMM_MEM_TYPE_DEVICE;
176 1 : commMem.addr = ptr;
177 1 : commMem.size = static_cast<uint64_t>(size);
178 3 : resource.memTag = std::string(HCCL_SYMMETRIC_MEMORY_TAG_PREFIX) + commId_ + "_addr_"
179 3 : + std::to_string(reinterpret_cast<uintptr_t>(ptr)) + "_size_" + std::to_string(size);
180 1 : HcclResult ret = commMems->CommRegMem(resource.memTag, commMem, &resource.memHandle);
181 1 : CHK_PRT_RET(
182 : ret != HCCL_SUCCESS,
183 : HCCL_ERROR(
184 : "[CollComm][RegisterSymmetricMemoryResource] CommRegMem failed, tag[%s], ptr[%p], "
185 : "size[%zu], ret[%d].",
186 : resource.memTag.c_str(), ptr, size, ret),
187 : ret);
188 :
189 1 : HCCL_RUN_INFO(
190 : "[CollComm][RegisterSymmetricMemoryResource] register symmetric memory success, group[%s], "
191 : "tag[%s], ptr[%p], size[%zu], memHandle[%p].",
192 : commId_.c_str(), resource.memTag.c_str(), ptr, size, resource.memHandle);
193 1 : return HCCL_SUCCESS;
194 : }
195 :
196 0 : void CollComm::UnregisterSymmetricMemoryResource(const SymmetricMemoryResource& resource)
197 : {
198 0 : if (resource.memHandle == nullptr || resource.memTag.empty()) {
199 0 : HCCL_WARNING(
200 : "[CollComm][UnregisterSymmetricMemoryResource] invalid resource, tag[%s], memHandle[%p].",
201 : resource.memTag.c_str(), resource.memHandle);
202 0 : return;
203 : }
204 0 : if (myRank_ == nullptr) {
205 0 : HCCL_WARNING(
206 : "[CollComm][UnregisterSymmetricMemoryResource] myRank is null, skip CommUnregMem, "
207 : "tag[%s], memHandle[%p].",
208 : resource.memTag.c_str(), resource.memHandle);
209 0 : return;
210 : }
211 0 : CommMems* commMems = myRank_->GetCommMems();
212 0 : if (commMems == nullptr) {
213 0 : HCCL_WARNING(
214 : "[CollComm][UnregisterSymmetricMemoryResource] commMems is null, skip CommUnregMem, "
215 : "tag[%s], memHandle[%p].",
216 : resource.memTag.c_str(), resource.memHandle);
217 0 : return;
218 : }
219 0 : HcclResult ret = commMems->CommUnregMem(resource.memTag, resource.memHandle);
220 0 : if (ret != HCCL_SUCCESS) {
221 0 : HCCL_WARNING(
222 : "[CollComm][UnregisterSymmetricMemoryResource] CommUnregMem failed, tag[%s], "
223 : "memHandle[%p], ret[%d].",
224 : resource.memTag.c_str(), resource.memHandle, ret);
225 : }
226 0 : ret = myRank_->UnregMemByTag(resource.memTag);
227 0 : if (ret != HCCL_SUCCESS) {
228 0 : HCCL_ERROR(
229 : "[CollComm][UnregisterSymmetricMemoryResource] UnregMemByTag failed, tag[%s], ret[%d].",
230 : resource.memTag.c_str(), ret);
231 0 : return;
232 : }
233 0 : HCCL_INFO(
234 : "[CollComm][UnregisterSymmetricMemoryResource] unregister symmetric memory success, "
235 : "tag[%s], memHandle[%p].",
236 : resource.memTag.c_str(), resource.memHandle);
237 : }
238 :
239 2 : HcclResult CollComm::RegisterWindow(void* ptr, size_t size, HcclCommSymWindow* winHandle)
240 : {
241 2 : CHK_SMART_PTR_NULL(symmetricMemory_);
242 2 : return symmetricMemory_->RegisterUrmaSymmetricMem(ptr, size, winHandle);
243 : }
244 :
245 0 : HcclResult CollComm::DeregisterWindow(HcclCommSymWindow winHandle)
246 : {
247 0 : CHK_SMART_PTR_NULL(symmetricMemory_);
248 0 : SymmetricMemoryResource resource;
249 0 : HcclResult getResourceRet = symmetricMemory_->GetRegisteredMemoryResource(winHandle, resource);
250 0 : CHK_PRT_RET(
251 : getResourceRet != HCCL_SUCCESS && getResourceRet != HCCL_E_NOT_FOUND,
252 : HCCL_ERROR(
253 : "[CollComm][DeregisterWindow] get registered symmetric memory resource failed, "
254 : "winHandle[%p], ret[%d].",
255 : winHandle, getResourceRet),
256 : getResourceRet);
257 :
258 0 : HcclResult ret = symmetricMemory_->DeregisterUrmaSymmetricMem(winHandle);
259 0 : if (ret == HCCL_SUCCESS && getResourceRet == HCCL_SUCCESS) {
260 0 : UnregisterSymmetricMemoryResource(resource);
261 : }
262 0 : return ret;
263 0 : }
264 :
265 0 : HcclResult CollComm::GetCommSymWin(void* ptr, size_t size, HcclCommSymWindow* winHandle, size_t* offset)
266 : {
267 0 : CHK_SMART_PTR_NULL(symmetricMemory_);
268 0 : return symmetricMemory_->FindUrmaSymmetricWindow(ptr, size, winHandle, offset);
269 : }
270 :
271 2 : HcclResult CollComm::RegisterPendingSymmetricMemHandles(std::vector<HcclMemHandle>& memHandles)
272 : {
273 2 : memHandles.clear();
274 2 : if (symmetricMemory_ == nullptr) {
275 0 : return HCCL_SUCCESS;
276 : }
277 :
278 2 : std::vector<SymmetricMemoryRegisterInfo> registerInfos;
279 : // HcclCommSymWinRegister只记录窗口，真正CommRegMem延迟到ChannelAcquire阶段执行。
280 2 : CHK_RET(symmetricMemory_->GetPendingRegisterInfos(registerInfos));
281 2 : if (registerInfos.empty()) {
282 1 : return HCCL_SUCCESS;
283 : }
284 :
285 1 : std::vector<std::pair<void*, SymmetricMemoryResource>> registeredResources;
286 2 : for (const SymmetricMemoryRegisterInfo& registerInfo : registerInfos) {
287 1 : SymmetricMemoryResource resource;
288 1 : HcclResult ret = RegisterSymmetricMemoryResource(registerInfo.userVa, registerInfo.userSize, resource);
289 1 : if (ret != HCCL_SUCCESS) {
290 0 : HCCL_ERROR(
291 : "[CollComm][RegisterPendingSymmetricMemHandles] register symmetric memory failed, "
292 : "win[%p], userVa[%p], size[%zu], ret[%d].",
293 : registerInfo.devWin, registerInfo.userVa, registerInfo.userSize, ret);
294 0 : for (const auto& registeredResource : registeredResources) {
295 0 : symmetricMemory_->RemoveRegisteredMemoryResource(registeredResource.first);
296 0 : UnregisterSymmetricMemoryResource(registeredResource.second);
297 : }
298 0 : return ret;
299 : }
300 :
301 1 : ret = symmetricMemory_->SetRegisteredMemoryResource(registerInfo.devWin, resource);
302 1 : if (ret != HCCL_SUCCESS) {
303 0 : HCCL_ERROR(
304 : "[CollComm][RegisterPendingSymmetricMemHandles] save symmetric memory resource failed, "
305 : "win[%p], userVa[%p], size[%zu], ret[%d].",
306 : registerInfo.devWin, registerInfo.userVa, registerInfo.userSize, ret);
307 0 : UnregisterSymmetricMemoryResource(resource);
308 0 : for (const auto& registeredResource : registeredResources) {
309 0 : symmetricMemory_->RemoveRegisteredMemoryResource(registeredResource.first);
310 0 : UnregisterSymmetricMemoryResource(registeredResource.second);
311 : }
312 0 : return ret;
313 : }
314 1 : registeredResources.emplace_back(registerInfo.devWin, resource);
315 : // 仅返回本次新注册的memHandle，避免普通URMA重复携带历史对称内存句柄。
316 1 : memHandles.emplace_back(static_cast<HcclMemHandle>(resource.memHandle));
317 1 : }
318 :
319 1 : return HCCL_SUCCESS;
320 2 : }
321 :
322 1 : HcclResult CollComm::UpdateSymmetricRemoteMem(
323 : uint32_t remoteRank, const CommMem* remoteMems, const std::vector<std::string>& memTags)
324 : {
325 1 : if (symmetricMemory_ == nullptr) {
326 0 : return HCCL_SUCCESS;
327 : }
328 1 : return symmetricMemory_->UpdateRemoteMem(remoteRank, remoteMems, memTags);
329 : }
330 :
331 164 : HcclResult CollComm::InitKfcAndRegisterCollComm()
332 : {
333 164 : myRank_->SetKfcControlTransfer(kfcControlTransferH2D_, kfcStatusTransferD2H_);
334 164 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
335 164 : return HCCL_SUCCESS;
336 : }
337 :
338 205 : HcclResult CollComm::DestroyAicpuComm()
339 : {
340 205 : CHK_PTR_NULL(callbacks_.getAicpuCommState);
341 176 : if (callbacks_.getAicpuCommState()) {
342 7 : CHK_SMART_PTR_NULL(kfcControlTransferH2D_);
343 3 : CHK_SMART_PTR_NULL(kfcStatusTransferD2H_);
344 :
345 3 : Hccl::KfcCommand opCmd = Hccl::KfcCommand::DESTROY_AICPU_COMM;
346 3 : CHK_RET(kfcControlTransferH2D_->Put(0, sizeof(Hccl::KfcCommand), reinterpret_cast<uint8_t*>(&opCmd)));
347 3 : HCCL_RUN_INFO(
348 : "[%s]group[%s] send Hccl::KfcCommand[%d] success", __func__, commId_.c_str(), static_cast<int>(opCmd));
349 :
350 3 : Hccl::KfcExecStatus opInfo;
351 3 : constexpr u32 WAIT_CMD_TIMEOUT = 10 * 1000; // 最大等待10秒
352 3 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
353 3 : auto startTime = std::chrono::steady_clock::now();
354 :
355 : while (true) {
356 2988 : CHK_RET(kfcStatusTransferD2H_->Get(0, sizeof(Hccl::KfcExecStatus), reinterpret_cast<uint8_t*>(&opInfo)));
357 2988 : if (opInfo.kfcStatus == Hccl::KfcStatus::DESTROY_AICPU_COMM_DONE) {
358 0 : HCCL_RUN_INFO("[%s]get Hccl::KfcStatus[%d] success", __func__, static_cast<int>(opInfo.kfcStatus));
359 0 : return HCCL_SUCCESS;
360 2988 : } else if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
361 3 : HCCL_ERROR(
362 : "[%s]timeout, maxTime[%u ms] and get the opExecStatus is [%s].", __func__, WAIT_CMD_TIMEOUT,
363 : opInfo.kfcStatus.Describe().c_str());
364 3 : return HCCL_E_TIMEOUT;
365 : }
366 2985 : usleep(TEN_MILLISECOND_OF_USLEEP);
367 2985 : }
368 : }
369 169 : return HCCL_SUCCESS;
370 : }
371 :
372 193 : uint32_t CollComm::GetMyRankId() const { return rankId_; }
373 :
374 6 : HcclResult CollComm::GetParentRankId(u32& parentRankId) const
375 : {
376 6 : Hccl::HcclCommunicator* comV2 = static_cast<Hccl::HcclCommunicator*>(comm_);
377 6 : CHK_PTR_NULL(comV2);
378 6 : parentRankId = comV2->GetRankInParentComm();
379 6 : return HCCL_SUCCESS;
380 : }
381 :
382 164 : HcclResult CollComm::InitHDCommunicate()
383 : {
384 : // 初始化aicpu进程 host-device 共享内存
385 164 : EXCEPTION_CATCH(
386 : (kfcControlTransferH2D_
387 : = std::make_shared<hccl::HDCommunicate>(deviceLogicId_, HCCL_HDC_TYPE_H2D, sizeof(Hccl::KfcCommand))),
388 : return HCCL_E_PTR);
389 164 : CHK_RET(kfcControlTransferH2D_->InitHost());
390 :
391 164 : EXCEPTION_CATCH(
392 : (kfcStatusTransferD2H_
393 : = std::make_shared<hccl::HDCommunicate>(deviceLogicId_, HCCL_HDC_TYPE_D2H, sizeof(Hccl::KfcExecStatus))),
394 : return HCCL_E_PTR);
395 164 : CHK_RET(kfcStatusTransferD2H_->InitHost());
396 :
397 164 : return HCCL_SUCCESS;
398 : }
399 :
400 164 : HcclResult CollComm::GetHDCommunicate(
401 : HDCommunicateParams& kfcControlTransferH2DParams, HDCommunicateParams& kfcStatusTransferD2HParams)
402 : {
403 164 : CHK_SMART_PTR_NULL(kfcControlTransferH2D_);
404 164 : CHK_SMART_PTR_NULL(kfcStatusTransferD2H_);
405 164 : kfcControlTransferH2DParams = kfcControlTransferH2D_->GetCommunicateParams();
406 164 : kfcStatusTransferD2HParams = kfcStatusTransferD2H_->GetCommunicateParams();
407 164 : HCCL_INFO("%s success, group[%s]", __func__, commId_.c_str());
408 164 : return HCCL_SUCCESS;
409 : }
410 :
411 2 : HcclCommStatus CollComm::GetCommStatus() const
412 : {
413 2 : std::lock_guard<std::mutex> lock(commMutex_);
414 2 : return commStatus_;
415 2 : }
416 :
417 2 : HcclResult CollComm::Suspend()
418 : {
419 2 : HCCL_RUN_INFO("[CollComm][Suspend] commId[%s] start to suspend.", commId_.c_str());
420 : {
421 2 : std::lock_guard<std::mutex> lock(commMutex_);
422 2 : if (commStatus_ == HcclCommStatus::HCCL_COMM_STATUS_SUSPENDING) {
423 1 : HCCL_WARNING("[CollComm][Suspend] The current communication has been suspended, no need to suspend again.");
424 1 : return HcclResult::HCCL_SUCCESS;
425 : }
426 :
427 1 : CHK_SMART_PTR_NULL(myRank_);
428 :
429 1 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_SUSPENDING;
430 2 : }
431 :
432 1 : return myRank_->StopLaunch();
433 : }
434 :
435 3 : HcclResult CollComm::Clean()
436 : {
437 3 : HCCL_RUN_INFO("[CollComm][Clean] commId[%s] start to clean.", commId_.c_str());
438 : {
439 3 : std::lock_guard<std::mutex> lock(commMutex_);
440 3 : if (commStatus_ != HcclCommStatus::HCCL_COMM_STATUS_SUSPENDING) {
441 1 : HCCL_ERROR(
442 : "[CollComm][Clean] The current communication is not suspended, cannot clean, status is [%u]",
443 : static_cast<uint32_t>(commStatus_));
444 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
445 : }
446 2 : if (isCleaned_) {
447 1 : HCCL_WARNING("[CollComm][Clean] The current communication has been cleaned, no need to clean again.");
448 1 : return HcclResult::HCCL_SUCCESS;
449 : }
450 :
451 1 : CHK_SMART_PTR_NULL(myRank_);
452 :
453 1 : isCleaned_ = true;
454 3 : }
455 :
456 : // 先清理Host
457 1 : return myRank_->Clean();
458 : }
459 :
460 2 : HcclResult CollComm::Resume()
461 : {
462 : {
463 2 : std::lock_guard<std::mutex> lock(commMutex_);
464 2 : if (commStatus_ == HcclCommStatus::HCCL_COMM_STATUS_INVALID) {
465 1 : HCCL_ERROR("[CollComm][Resume] Comm has been error, can not resume now!");
466 1 : return HcclResult::HCCL_E_INTERNAL;
467 : }
468 1 : if (commStatus_ != HcclCommStatus::HCCL_COMM_STATUS_SUSPENDING) {
469 0 : HCCL_WARNING(
470 : "[CollComm][Resume] The current communication is normal, no need to resume, status is [%u]",
471 : static_cast<uint32_t>(commStatus_));
472 0 : return HcclResult::HCCL_SUCCESS;
473 : }
474 :
475 1 : HCCL_INFO("[CollComm][Resume] start to Resume.");
476 1 : CHK_SMART_PTR_NULL(myRank_);
477 1 : auto ret = myRank_->Resume();
478 1 : if (ret != HcclResult::HCCL_SUCCESS) {
479 0 : HCCL_ERROR("[CollComm][Resume] %s failed, ret = 0x%016llx", __func__, HCCL_ERROR_CODE(ret));
480 0 : return ret;
481 : }
482 :
483 1 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
484 1 : isCleaned_ = false;
485 2 : }
486 1 : HCCL_INFO("[CollComm][Resume] commId[%s] resume success.", commId_.c_str());
487 1 : return HcclResult::HCCL_SUCCESS;
488 : }
489 :
490 164 : HcclResult CollComm::InitTaskExceptionHandler()
491 : {
492 164 : hcomm::TaskExceptionHost* handler = hcomm::TaskExceptionHost::GetInstance(deviceLogicId_);
493 164 : CHK_PTR_NULL(handler);
494 164 : CHK_RET(handler->Register(reinterpret_cast<u64>(this)));
495 164 : return HCCL_SUCCESS;
496 : }
497 :
498 0 : Hccl::ErrorMessageReport CollComm::GetAicpuTaskException()
499 : {
500 0 : Hccl::ErrorMessageReport errorMessage;
501 0 : CHK_PRT_RET(kfcStatusTransferD2H_ == nullptr, HCCL_ERROR("[%s]fail, d2h is nullptr", __func__), errorMessage);
502 :
503 0 : HcclResult ret = kfcStatusTransferD2H_->Get(
504 : sizeof(Hccl::KfcStatus) + sizeof(Hccl::KfcErrType), sizeof(errorMessage),
505 : reinterpret_cast<uint8_t*>(&errorMessage));
506 :
507 0 : CHK_PRT_RET(
508 : ret != HCCL_SUCCESS,
509 : HCCL_ERROR("[%s]fail, group [%s], ret[%d]", __func__, commId_.c_str(), static_cast<int>(ret)), errorMessage);
510 0 : HCCL_INFO("[%s]group[%s] success", __func__, commId_.c_str());
511 0 : return errorMessage;
512 : }
513 :
514 0 : uint32_t CollComm::UpdateIndex() { return index_ += 1; }
515 :
516 164 : HcclResult CollComm::GetRankIpPortMap()
517 : {
518 164 : Hccl::HcclCommunicator* commV2 = static_cast<Hccl::HcclCommunicator*>(comm_);
519 164 : CHK_PTR_NULL(commV2);
520 164 : CHK_RET(commV2->GetRankIpPortMap(rankIpPortMap_));
521 164 : CHK_PTR_NULL(rankIpPortMap_);
522 : // rankIpPortMap_ 在单卡多进程场景下，用于保证端口不冲突
523 : // 该映射表记录了：Rank ID -> (IP地址 -> 已占用的端口号)
524 164 : return HCCL_SUCCESS;
525 : }
526 :
527 3 : HcclResult CollComm::GetHcclBinHandle(aclrtBinHandle& binHcclHandle)
528 : {
529 3 : std::lock_guard<std::mutex> lock(binHcclmutex_);
530 3 : HCCL_DEBUG("[%s] GetHcclBinHandle", __func__);
531 3 : if (binHcclHandle_ == nullptr) {
532 2 : std::string hcclJsonPath;
533 2 : CHK_RET(GetKernelFilePath(hcclJsonPath));
534 2 : hcclJsonPath += "libscatter_aicpu_kernel.json";
535 : HcclResult ret
536 2 : = LoadBinaryFromFile(hcclJsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0, binHcclHandle_);
537 2 : CHK_PRT_RET(
538 : ret != HCCL_SUCCESS,
539 : HCCL_ERROR(
540 : "[%s]errNo[0x%016llx]load aicpu file fail, path[%s] optionType[%u] cpuKernelMode[%u].", __func__,
541 : HCCL_ERROR_CODE(ret), hcclJsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0),
542 : ret);
543 :
544 2 : HCCL_INFO(
545 : "[%s]load aicpu file success, path[%s] optionType[%u] cpuKernelMode[%u].", __func__, hcclJsonPath.c_str(),
546 : ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0);
547 2 : }
548 3 : binHcclHandle = binHcclHandle_;
549 3 : return HCCL_SUCCESS;
550 3 : }
551 :
552 205 : HcclResult CollComm::HcclBinaryUnLoad()
553 : {
554 205 : std::lock_guard<std::mutex> lock(binHcclmutex_);
555 205 : if (binHcclHandle_ == nullptr) {
556 205 : HCCL_RUN_WARNING("[%s] binHcclHandle is nullptr", __func__);
557 205 : return HCCL_SUCCESS;
558 : }
559 :
560 0 : HCCL_DEBUG("[%s]aclrtBinaryUnLoad binHcclHandle", __func__);
561 0 : aclError ret = aclrtBinaryUnLoad(binHcclHandle_);
562 0 : binHcclHandle_ = nullptr;
563 0 : if (ret != 0) {
564 0 : HCCL_RUN_WARNING("[%s]aclrtBinaryUnLoad failed, aclRet[%d]", __func__, ret);
565 0 : return HCCL_E_INTERNAL;
566 : }
567 0 : return HCCL_SUCCESS;
568 205 : }
569 :
570 : } // namespace hccl
|
