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