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