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_aicpu_mgr.h"
12 : #include "ns_recovery/aicpu/ns_recovery_func_lite.h"
13 : #include "aicpu_daemon_service.h"
14 : #include "hcclCommTaskExceptionLite.h"
15 : #include "coll_comm_aicpu_destroy_func.h"
16 : #include "aicpu_indop_env.h"
17 : #include "unified_platform/pub_inc/config_plf_log.h"
18 : #include "dlhal_function_v2.h"
19 : #include "profiling_command_handle_lite.h"
20 : #include "adapter_hal_pub.h"
21 : #include "log.h"
22 : #include <chrono>
23 : #include <unistd.h>
24 :
25 : thread_local CollCommAicpu* CollCommAicpuMgr::currentComm_ = nullptr;
26 :
27 129758 : CollCommAicpuMgr& CollCommAicpuMgr::GetInstance()
28 : {
29 129758 : static CollCommAicpuMgr instance;
30 129758 : return instance;
31 : }
32 :
33 : // ==================== 通信域初始化 ====================
34 :
35 12 : HcclResult CollCommAicpuMgr::InitComm(CommAicpuParam* commAicpuParam)
36 : {
37 12 : CHK_PTR_NULL(commAicpuParam);
38 :
39 11 : std::string group = commAicpuParam->hcomId;
40 11 : CollCommAicpu* aicpuComm = nullptr;
41 11 : CHK_RET(AcquireAndCreateComm(group, &aicpuComm));
42 11 : if (aicpuComm == nullptr) {
43 0 : HCCL_ERROR("[CollCommAicpuMgr][InitComm] aicpuComm is null group[%s]", group.c_str());
44 0 : return HCCL_E_PTR;
45 : }
46 :
47 11 : HcclResult ret = aicpuComm->InitAicpuIndOp(commAicpuParam);
48 11 : CHK_PRT_RET(
49 : ret != HCCL_SUCCESS,
50 : HCCL_ERROR(
51 : "[CollCommAicpuMgr][%s]errNo[0x%016llx] Failed to init independent op comm group[%s]", __func__,
52 : HCCL_ERROR_CODE(ret), group.c_str()),
53 : ret);
54 :
55 : // 全局环境初始化 (call_once 保证只执行一次)
56 : static std::once_flag initBackGround;
57 11 : std::call_once(initBackGround, [aicpuComm, this]() {
58 3 : this->InitBackGroundThread(aicpuComm->GetDevId());
59 3 : });
60 :
61 : static std::once_flag initEnv;
62 11 : std::call_once(initEnv, [commAicpuParam, this]() {
63 3 : this->InitIndopEnv(commAicpuParam);
64 3 : });
65 :
66 11 : return HCCL_SUCCESS;
67 11 : }
68 :
69 24 : HcclResult CollCommAicpuMgr::AcquireAndCreateComm(const std::string& group, CollCommAicpu** outComm)
70 : {
71 24 : std::unique_lock<std::shared_mutex> rwlock(commMapMutex_);
72 24 : auto iter = commMap_.find(group);
73 24 : if (iter != commMap_.end()) {
74 : // 已存在 — 确保 CollCommAicpu 已创建
75 2 : if (iter->second.comm == nullptr) {
76 0 : EXCEPTION_CATCH(iter->second.comm = std::make_unique<CollCommAicpu>(), return HCCL_E_PTR);
77 : }
78 2 : *outComm = iter->second.comm.get();
79 2 : HCCL_INFO("[%s]Reuse existing comm group [%s]", __func__, group.c_str());
80 2 : return HCCL_SUCCESS;
81 : }
82 :
83 : // 未找到则创建新实例
84 22 : CommEntry entry;
85 22 : EXCEPTION_CATCH(entry.comm = std::make_unique<CollCommAicpu>(), return HCCL_E_PTR);
86 :
87 22 : *outComm = entry.comm.get();
88 22 : commMap_.insert({group, std::move(entry)});
89 22 : HCCL_RUN_INFO("[%s]Created new comm group [%s]", __func__, group.c_str());
90 22 : return HCCL_SUCCESS;
91 24 : }
92 :
93 : // ==================== 通信域注册表操作 ====================
94 :
95 35 : CollCommAicpu* CollCommAicpuMgr::AcquireCommForUse(const std::string& group)
96 : {
97 35 : HCCL_INFO("[CollCommAicpuMgr][%s]start, group[%s]", __func__, group.c_str());
98 35 : auto startTime = std::chrono::steady_clock::now();
99 35 : constexpr u32 pollIntervalUs = 10;
100 35 : constexpr u32 pollTimeoutMs = 10000;
101 35 : auto waitPollTimeOutMs = std::chrono::milliseconds(pollTimeoutMs);
102 :
103 : while (true) {
104 35 : std::unique_lock<std::shared_mutex> rwlock(commMapMutex_);
105 35 : auto iter = commMap_.find(group);
106 35 : if (iter == commMap_.end()) {
107 1 : HCCL_ERROR(
108 : "[CollCommAicpuMgr][%s] group[%s] not found, exist size[%zu]", __func__, group.c_str(),
109 : commMap_.size());
110 1 : auto curIter = commMap_.begin();
111 12 : while (curIter != commMap_.end()) {
112 11 : HCCL_ERROR("[CollCommAicpuMgr][%s] exist group [%s]", __func__, curIter->first.c_str());
113 11 : curIter++;
114 : }
115 1 : return nullptr;
116 : }
117 :
118 34 : if (iter->second.isUsed) {
119 0 : auto curTime = std::chrono::steady_clock::now();
120 0 : if ((curTime - startTime) >= waitPollTimeOutMs) {
121 0 : startTime = curTime;
122 0 : HCCL_RUN_INFO("[CollCommAicpuMgr][%s]wait, comm group [%s] has been used", __func__, group.c_str());
123 : }
124 0 : rwlock.unlock();
125 0 : usleep(pollIntervalUs);
126 0 : continue;
127 0 : }
128 34 : currentComm_ = iter->second.comm.get();
129 34 : iter->second.isUsed = true;
130 34 : HCCL_INFO("[CollCommAicpuMgr][%s]success, group[%s]", __func__, group.c_str());
131 34 : return iter->second.comm.get();
132 35 : }
133 : }
134 :
135 33 : void CollCommAicpuMgr::ReleaseComm(const std::string& group)
136 : {
137 33 : std::unique_lock<std::shared_mutex> rwlock(commMapMutex_);
138 33 : auto iter = commMap_.find(group);
139 33 : if (iter == commMap_.end()) {
140 1 : return;
141 : }
142 32 : currentComm_ = nullptr;
143 32 : iter->second.isUsed = false;
144 33 : }
145 :
146 28 : CollCommAicpu* CollCommAicpuMgr::FindCommByGroup(const std::string& group)
147 : {
148 28 : std::shared_lock<std::shared_mutex> lock(commMapMutex_);
149 28 : auto iter = commMap_.find(group);
150 28 : if (iter == commMap_.end()) {
151 0 : return nullptr;
152 : }
153 28 : return iter->second.comm.get();
154 28 : }
155 :
156 3 : CollCommAicpu* CollCommAicpuMgr::GetCurrentComm(const std::string& group)
157 : {
158 3 : if (group.empty()) {
159 1 : HCCL_ERROR("[CollCommAicpuMgr][%s] comm group is empty", __func__);
160 1 : return nullptr;
161 : }
162 2 : if (currentComm_ == nullptr) {
163 1 : HCCL_ERROR("[CollCommAicpuMgr][%s] currentComm_ is nullptr", __func__);
164 1 : return nullptr;
165 : }
166 1 : if (currentComm_->GetIdentifier() != group) {
167 0 : HCCL_ERROR("[CollCommAicpuMgr][%s] comm group[%s] is not current comm group", __func__, group.c_str());
168 0 : return nullptr;
169 : }
170 1 : return currentComm_;
171 : }
172 :
173 11 : HcclResult CollCommAicpuMgr::DestroyComm(const std::string& group)
174 : {
175 11 : std::unique_lock<std::shared_mutex> rwlock(commMapMutex_);
176 11 : auto iter = commMap_.find(group);
177 11 : if (iter == commMap_.end()) {
178 2 : HCCL_ERROR("[CollCommAicpuMgr][%s]group[%s] is not exist", __func__, group.c_str());
179 2 : return HCCL_E_PARA;
180 : }
181 :
182 9 : CollCommAicpu* aicpuComm = iter->second.comm.get();
183 9 : CHK_PTR_NULL(aicpuComm);
184 9 : aicpuComm->SetCommmStatus(HcclCommStatus::HCCL_COMM_STATUS_INVALID);
185 :
186 : // 正在使用中，不销毁，返回重试状态让调用方稍后再试
187 9 : if (iter->second.isUsed) {
188 1 : HCCL_RUN_WARNING("[CollCommAicpuMgr][%s]comm group [%s] has been used, skip erase", __func__, group.c_str());
189 1 : return HCCL_E_AGAIN;
190 : }
191 :
192 : // 防御性检查 legacy 通信域 busy 标记，避免 isUsed 与 legacy busy 不同步时误销毁
193 8 : if (aicpuComm->GetLegacy910CollComm() != nullptr && aicpuComm->IsLegacy910CollCommBusy()) {
194 0 : HCCL_RUN_WARNING("[CollCommAicpuMgr][%s]legacy comm group [%s] is busy, skip erase", __func__, group.c_str());
195 0 : return HCCL_E_AGAIN;
196 : }
197 :
198 8 : commMap_.erase(group);
199 8 : HCCL_RUN_INFO("[CollCommAicpuMgr][%s]Destroy comm group [%s] success.", __func__, group.c_str());
200 8 : return HCCL_SUCCESS;
201 11 : }
202 :
203 64812 : HcclResult CollCommAicpuMgr::GetAllComms(std::vector<std::pair<std::string, CollCommAicpu*>>& aicpuCommInfo)
204 : {
205 : // 调用方必须在外部持有 commMapMutex_ 共享锁（保护遍历+访问 comm 成员的完整临界区）
206 64824 : for (auto& kv : commMap_) {
207 12 : aicpuCommInfo.push_back({kv.first, kv.second.comm.get()});
208 : }
209 64812 : return HCCL_SUCCESS;
210 : }
211 :
212 64810 : std::shared_mutex& CollCommAicpuMgr::GetMutex() { return commMapMutex_; }
213 :
214 : // ==================== 全局环境初始化 ====================
215 :
216 4 : void CollCommAicpuMgr::InitIndopEnv(CommAicpuParam* commAicpuParam)
217 : {
218 4 : hcomm::SetTaskExceptionEnable(commAicpuParam->commConfig.taskExceptionEnable);
219 4 : Hccl::SetPlfDebugConfigValue(commAicpuParam->commConfig.plfDebugConfig);
220 4 : HCCL_RUN_INFO(
221 : "[%s]Env: taskExceptionEnable[%d], notifyWaitTimeout[%u s], plfDebugConfig[0x%llx]", __func__,
222 : commAicpuParam->commConfig.taskExceptionEnable, commAicpuParam->commConfig.notifyWaitTimeout,
223 : commAicpuParam->commConfig.plfDebugConfig);
224 4 : }
225 :
226 3 : void CollCommAicpuMgr::InitBackGroundThread(u32 devId)
227 : {
228 : static auto commandToBackGroud = Hccl::CommandToBackGroud::Default;
229 3 : static auto daemonServiceRun = [](void* info) {
230 3 : Hccl::AicpuDaemonService::GetInstance().ServiceRun(info);
231 3 : };
232 3 : static auto daemonServiceStop = [](void* info) {
233 3 : Hccl::AicpuDaemonService::GetInstance().ServiceStop(info);
234 3 : };
235 :
236 3 : hcomm::HcclCommTaskExceptionLite::GetInstance().Init(devId);
237 3 : Hccl::AicpuDaemonService::GetInstance().Register(&hcomm::HcclCommTaskExceptionLite::GetInstance());
238 3 : Hccl::AicpuDaemonService::GetInstance().Register(&hccl::CollCommAicpuDestroyFunc::GetInstance());
239 3 : Hccl::AicpuDaemonService::GetInstance().Register(&NsRecoveryFuncLite::GetInstance());
240 :
241 3 : if (Hccl::StartMC2MaintenanceThread != nullptr) {
242 3 : Hccl::StartMC2MaintenanceThread(daemonServiceRun, &commandToBackGroud, daemonServiceStop, &commandToBackGroud);
243 3 : HCCL_RUN_INFO("[%s]start BackGround thread success.", __func__);
244 : } else {
245 0 : HCCL_WARNING("[%s]StartMC2MaintenanceThread func is nullptr", __func__);
246 : }
247 3 : }
|
