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