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