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 "entity/llm_comm_entity_mgr.h"
12 : #include <memory>
13 : #include "securec.h"
14 : #include "ascend_hal_define.h"
15 : #include "ascend_hal.h"
16 : #include "fsm/state_define.h"
17 : #include "llm_common/hccl_proxy.h"
18 : #include "llm_common/llm_common.h"
19 : #include "common/scope_guard.h"
20 :
21 : namespace FlowFunc {
22 : namespace {
23 : constexpr size_t kDefaultMultiRequestCount = 1024UL;
24 : constexpr uint64_t kCheckTimeoutLoopCount = 1000UL;
25 : constexpr uint64_t kProcessTimeout = 1000000UL; // 1s
26 : constexpr uint64_t kResetProfilingTimeInterval = 60UL * 60UL; // one hour
27 : } // namespace
28 :
29 : LlmCommEntityMgr &LlmCommEntityMgr::GetInstance() {
30 : static LlmCommEntityMgr manager;
31 : return manager;
32 : }
33 :
34 : LlmCommEntityMgr::LlmCommEntityMgr() : listen_conn_(nullptr), listen_hccl_addr_({}) {
35 : comp_indices_.resize(kDefaultMultiRequestCount);
36 : comp_status_.resize(kDefaultMultiRequestCount);
37 : }
38 :
39 : LlmCommEntityMgr::~LlmCommEntityMgr() {
40 : ClearEntities();
41 : }
42 :
43 : EntityPtr LlmCommEntityMgr::GetEntityByConn(HcclConn conn) {
44 : std::lock_guard<std::mutex> lock(entity_mutex_);
45 : auto iter = server_entity_map_.find(conn);
46 : if (iter != server_entity_map_.end()) {
47 : return iter->second;
48 : }
49 : UDF_LOG_INFO("Not exist entity, conn:%p.", conn);
50 : return nullptr;
51 : }
52 :
53 0 : HcclConn LlmCommEntityMgr::GetEntityByIp(uint32_t ip) {
54 0 : std::lock_guard<std::mutex> lock(entity_mutex_);
55 : auto iter = ip_to_conns_.find(ip);
56 : if (iter != ip_to_conns_.end()) {
57 : return iter->second;
58 : }
59 : UDF_RUN_LOG_WARN("Entity is not exist for remote ip:%u.", ip);
60 : return nullptr;
61 : }
62 :
63 : size_t LlmCommEntityMgr::GetEntityMapSize() {
64 : mgr_need_use_mtx_.store(true, std::memory_order_relaxed);
65 : ScopeGuard guard([this] { mgr_need_use_mtx_.store(false, std::memory_order_relaxed); });
66 : std::lock_guard<std::mutex> lock(entity_mutex_);
67 : return server_entity_map_.size();
68 : }
69 :
70 : EntityPtr LlmCommEntityMgr::GetEntityByRemoteClusterId(uint64_t remote_cluster_id) {
71 : mgr_need_use_mtx_.store(true, std::memory_order_relaxed);
72 : ScopeGuard guard([this] { mgr_need_use_mtx_.store(false, std::memory_order_relaxed); });
73 : std::lock_guard<std::mutex> lock(entity_mutex_);
74 : auto iter = client_entity_map_.find(remote_cluster_id);
75 : if ((iter != client_entity_map_.end()) && (iter->second->GetCurState() != FsmState::kFsmDestroyState)) {
76 : return iter->second;
77 : }
78 : UDF_RUN_LOG_WARN("Not exist entity, remote_cluster_id:%lu.", remote_cluster_id);
79 : return nullptr;
80 : }
81 :
82 : EntityPtr LlmCommEntityMgr::CreateEntity(EntityType type, HcclConn conn, HcclAddr &local_hccl_addr,
83 : HcclAddr &remote_hccl_addr, uint64_t remote_cluster_id) {
84 : EntityPtr entity;
85 : try {
86 : entity = std::make_shared<LlmCommEntity>(type, conn, local_hccl_addr, remote_hccl_addr);
87 : } catch (const std::bad_alloc &) {
88 : UDF_LOG_ERROR("Make shared failed");
89 : return nullptr;
90 : }
91 : mgr_need_use_mtx_.store(true, std::memory_order_relaxed);
92 : ScopeGuard guard([this] { mgr_need_use_mtx_.store(false, std::memory_order_relaxed); });
93 : UDF_LOG_INFO("Set high priority flag.");
94 : std::lock_guard<std::mutex> lock(entity_mutex_);
95 : if (type == EntityType::kEntityServer) {
96 : server_entity_map_[conn] = entity;
97 : (void)ip_to_conns_.emplace(remote_hccl_addr.info.tcp.ipv4Addr, conn);
98 : } else {
99 : client_entity_map_[remote_cluster_id] = entity;
100 : }
101 : UDF_LOG_INFO("Success to create entity:%s.", entity->GetDesc().c_str());
102 : return entity;
103 : }
104 :
105 : void LlmCommEntityMgr::AddClientEntityMap(uint64_t remote_cluster_id, EntityPtr entity) {
106 : std::lock_guard<std::mutex> lock(entity_mutex_);
107 : entity->SetRemoteClusterId(remote_cluster_id);
108 : UDF_LOG_INFO("Add client entity for cluster:%lu", remote_cluster_id);
109 : (void)client_entity_map_.emplace(remote_cluster_id, entity);
110 : }
111 :
112 : FsmStatus LlmCommEntityMgr::DeleteEntityByRemoteClusterId(uint64_t remote_cluster_id) {
113 : mgr_need_use_mtx_.store(true, std::memory_order_relaxed);
114 : ScopeGuard guard([this] { mgr_need_use_mtx_.store(false, std::memory_order_relaxed); });
115 : std::lock_guard<std::mutex> lock(entity_mutex_);
116 : auto iter = client_entity_map_.find(remote_cluster_id);
117 : if (iter == client_entity_map_.end()) {
118 : UDF_LOG_INFO("Not exist remote_cluster_id:%lu.", remote_cluster_id);
119 : return FsmStatus::kFsmSuccess;
120 : }
121 : UDF_LOG_INFO("Delete entity:%s.", iter->second->GetDesc().c_str());
122 : (void)client_entity_map_.erase(iter);
123 : return FsmStatus::kFsmSuccess;
124 : }
125 :
126 : std::vector<int32_t> &LlmCommEntityMgr::GetCompIndices(size_t req_size) {
127 : if (comp_indices_.size() < req_size) {
128 : comp_indices_.resize(req_size);
129 : }
130 : return comp_indices_;
131 : }
132 :
133 : std::vector<HcclStatus> &LlmCommEntityMgr::GetCompStatus(size_t req_size) {
134 : if (comp_status_.size() < req_size) {
135 : comp_status_.resize(req_size);
136 : }
137 : return comp_status_;
138 : }
139 :
140 : FsmStatus LlmCommEntityMgr::InitServerConn(uint32_t ip, uint16_t port, bool need_lock) {
141 : server_conn_inited_ = false;
142 : uint64_t start_tick = StatisticManager::GetInstance().GetCpuTick();
143 : if (need_lock) {
144 : std::lock_guard<std::mutex> lock(switch_mutex_);
145 : }
146 : HcclResult ret = HcclRawOpen(&listen_conn_);
147 : if (ret != HcclResult::HCCL_SUCCESS) {
148 : UDF_LOG_ERROR("Call HcclRawOpen failed, ret:%d.", ret);
149 : return FsmStatus::kFsmHcclFailed;
150 : }
151 : listen_hccl_addr_.type = HcclAddrType::HCCL_ADDR_TYPE_ROCE;
152 : listen_hccl_addr_.info.tcp.ipv4Addr = ip;
153 : listen_hccl_addr_.info.tcp.port = port;
154 : ret = HcclRawBind(listen_conn_, &listen_hccl_addr_);
155 : if (ret != HcclResult::HCCL_SUCCESS) {
156 : UDF_LOG_ERROR("Bind server conn failed, ret:%d.", ret);
157 : return FsmStatus::kFsmHcclFailed;
158 : }
159 : ret = HcclRawListen(listen_conn_, 1);
160 : if (ret != HcclResult::HCCL_SUCCESS) {
161 : UDF_LOG_ERROR("Listen server conn failed, ret:%d.", ret);
162 : return FsmStatus::kFsmHcclFailed;
163 : }
164 : initialized_ = true;
165 : UDF_LOG_INFO("Init server conn time cost:%.2f us.",
166 : StatisticManager::GetInstance().GetTimeCost(StatisticManager::GetInstance().GetCpuTick() - start_tick));
167 : server_ip_ = ip;
168 : server_port_ = port;
169 : server_conn_inited_ = true;
170 : return FsmStatus::kFsmSuccess;
171 : }
172 :
173 : void LlmCommEntityMgr::ReopenServerConn() {
174 : if (listen_conn_ != nullptr) {
175 : auto ret = HcclRawForceClose(listen_conn_);
176 : if (ret != HCCL_SUCCESS) {
177 : UDF_LOG_ERROR("Close conn failed, ret:%d.", ret);
178 : }
179 : listen_conn_ = nullptr;
180 : }
181 : auto init_ret = InitServerConn(server_ip_, server_port_, false);
182 : if (init_ret != FsmStatus::kFsmSuccess) {
183 : UDF_LOG_ERROR("Init server conn failed, ret:%d.", static_cast<int32_t>(init_ret));
184 : }
185 : }
186 :
187 : FsmStatus LlmCommEntityMgr::InitClientConn(HcclAddr &local_hccl_addr, HcclConn &hccl_conn) {
188 : HcclResult ret = HcclRawOpen(&hccl_conn);
189 : if (ret != HcclResult::HCCL_SUCCESS) {
190 : UDF_LOG_ERROR("Fail to create local conn, local_hccl_addr:%s, ret:%d.", ToDesc(local_hccl_addr).c_str(), ret);
191 : return FsmStatus::kFsmHcclFailed;
192 : }
193 : return FsmStatus::kFsmSuccess;
194 : }
195 :
196 : void LlmCommEntityMgr::PromptHandleReq() {
197 : HandleLinkRequest();
198 : // process all conn entities: no need lock
199 : for (auto iter = server_entity_map_.begin(); iter != server_entity_map_.end();) {
200 : if (iter->second->GetCurState() == FsmState::kFsmErrorState) {
201 : iter++;
202 : continue;
203 : }
204 : if (iter->second->GetReqIsUsing().load(std::memory_order_relaxed) && !iter->second->GetEntityOccupied().load()) {
205 : iter++;
206 : continue;
207 : }
208 : auto &mutex = iter->second->GetMutex();
209 : if (mutex.try_lock()) {
210 : std::lock_guard<std::mutex> lock(mutex, std::adopt_lock);
211 : if (iter->second->GetCurState() == FsmState::kFsmDestroyState) {
212 : std::lock_guard<std::mutex> mapLock(entity_mutex_);
213 : UDF_LOG_INFO("start erase entity:%lu.", iter->second->GetRemoteClusterId());
214 : EraseIpToConnMap(iter->second->GetRemoteIp(), iter->second->GetConn());
215 : EraseClientMapByClusterId(iter->second->GetRemoteClusterId());
216 : iter = server_entity_map_.erase(iter);
217 : continue;
218 : }
219 : bool is_init_or_link = (iter->second->GetCurState() == FsmState::kFsmInitState) ||
220 : (iter->second->GetCurState() == FsmState::kFsmLinkState);
221 : FsmStatus status = iter->second->ProcessState();
222 : if (is_init_or_link && status == FsmStatus::kFsmEstablishLinkSuc) {
223 : AddClientEntityMap(iter->second->GetClientClusterInfo().cluster_id, iter->second);
224 : } else if ((status != FsmStatus::kFsmSuccess) && (status != FsmStatus::kFsmKeepState)) {
225 : (void)iter->second->ChangeState(FsmState::kFsmErrorState);
226 : }
227 : }
228 : iter++;
229 : }
230 : }
231 :
232 : void LlmCommEntityMgr::DecoderHandleReq() {
233 : if (mgr_need_use_mtx_.load(std::memory_order_relaxed)) {
234 : return;
235 : }
236 : std::lock_guard<std::mutex> mapLock(entity_mutex_);
237 : for (auto iter = client_entity_map_.begin(); iter != client_entity_map_.end();) {
238 : if (iter->second->GetCurState() == FsmState::kFsmErrorState) {
239 : iter++;
240 : continue;
241 : }
242 : if (iter->second->GetReqIsUsing().load(std::memory_order_relaxed) && !iter->second->GetEntityOccupied().load()) {
243 : iter++;
244 : continue;
245 : }
246 : auto &mutex = iter->second->GetMutex();
247 : if (!iter->second->GetIsUnlinking().load(std::memory_order_relaxed) && mutex.try_lock()) {
248 : std::lock_guard<std::mutex> lock(mutex, std::adopt_lock);
249 : FsmStatus status = iter->second->ProcessState();
250 : if ((status != FsmStatus::kFsmSuccess) && (status != FsmStatus::kFsmKeepState)) {
251 : (void)iter->second->ChangeState(FsmState::kFsmErrorState);
252 : }
253 : }
254 : iter++;
255 : }
256 : UDF_LOG_DEBUG("DecoderHandleReq free lock.");
257 : }
258 :
259 : void LlmCommEntityMgr::HandleRequest(bool is_prompt) {
260 : static uint64_t func_execute_count = 0UL;
261 : func_execute_count++;
262 : uint64_t start_tick = StatisticManager::GetInstance().GetCpuTick();
263 : uint32_t loop_count = 0;
264 : UDF_LOG_DEBUG("Enter HandleRequest.");
265 : while ((loop_count < kCheckTimeoutLoopCount) ||
266 : (CheckTimeout(start_tick, kProcessTimeout) != FsmStatus::kFsmTimeout)) {
267 : loop_count++;
268 : if (loop_count > kCheckTimeoutLoopCount) {
269 : loop_count = 0U;
270 : }
271 : if (is_prompt) {
272 : if (!initialized_.load()) {
273 : continue;
274 : }
275 : std::lock_guard<std::mutex> lk(switch_mutex_);
276 : PromptHandleReq();
277 : } else {
278 : DecoderHandleReq();
279 : }
280 : }
281 : // one hour check statistic info
282 : if (func_execute_count % kResetProfilingTimeInterval == 0UL) {
283 : StatisticManager::GetInstance().ResetProfilingTrack();
284 : }
285 : }
286 :
287 : void LlmCommEntityMgr::HandleLinkRequest() {
288 : if (!server_conn_inited_) {
289 : ReopenServerConn();
290 : return;
291 : }
292 : HcclConn hccl_conn = nullptr;
293 : HcclAddr remote_hccl_addr{};
294 : HcclResult accept_ret = HcclRawAccept(listen_conn_, &remote_hccl_addr, &hccl_conn);
295 : if ((accept_ret != HCCL_SUCCESS) && (accept_ret != HCCL_E_AGAIN)) {
296 : UDF_LOG_ERROR("Fail to call HcclRawAccept, ret:%d.", accept_ret);
297 : ReopenServerConn();
298 : return;
299 : }
300 : if (hccl_conn == nullptr) {
301 : return;
302 : }
303 : // accept new link
304 : const uint32_t remote_ip = remote_hccl_addr.info.tcp.ipv4Addr;
305 12 : bool cleared_residual = false;
306 12 : auto entity = FindServerEntityByIp(remote_ip, cleared_residual);
307 12 : if (entity != nullptr) {
308 1 : (void)HcclRawForceClose(entity->GetConn());
309 1 : entity->SetConn(hccl_conn);
310 1 : entity->SetLinkEstablished(false);
311 1 : entity->SetProbeLinkClusterInfoFlag(false);
312 1 : entity->ClearResource();
313 1 : entity->ChangeState(FsmState::kFsmLinkState);
314 1 : UDF_RUN_LOG_INFO("Success to accept new force link, remote hccl addr:%s.", ToDesc(remote_hccl_addr).c_str());
315 1 : return;
316 : }
317 11 : if (cleared_residual) {
318 0 : UDF_RUN_LOG_INFO("Success to accept new link with residual data in ip_to_conns map, remote hccl addr:%s.",
319 : ToDesc(remote_hccl_addr).c_str());
320 : }
321 11 : entity = this->CreateEntity(EntityType::kEntityServer, hccl_conn, listen_hccl_addr_, remote_hccl_addr);
322 : if (entity == nullptr) {
323 : UDF_LOG_ERROR("failed to create server comm entity.");
324 : return;
325 : }
326 : entity->GetServerTickRecord().link_start_tick = StatisticManager::GetInstance().GetCpuTick();
327 : UDF_LOG_INFO("Success to accept new link, remote hccl addr:%s.", ToDesc(remote_hccl_addr).c_str());
328 : }
329 :
330 : FsmStatus LlmCommEntityMgr::RegisterHcclMr(uint32_t dev_id, std::vector<uint64_t> &mem_addrs) {
331 : GroupQueryInput drv_input;
332 : error_t ret = memset_s(&drv_input, sizeof(drv_input), 0, sizeof(drv_input));
333 : if (ret != EOK) {
334 : UDF_LOG_ERROR("Memset failed, ret=%d!", ret);
335 : return FsmStatus::kFsmFailed;
336 : }
337 : char *grp_name_ptr = drv_input.grpQueryGroupAddrPara.grpName;
338 : FsmStatus query_grp_ret = QueryCurMemGrp(&grp_name_ptr);
339 : if (query_grp_ret != FsmStatus::kFsmSuccess) {
340 : return query_grp_ret;
341 : }
342 : UDF_LOG_INFO("Current group name:%s", grp_name_ptr);
343 : drv_input.grpQueryGroupAddrPara.devId = dev_id;
344 : const auto k_drv_input_len = static_cast<uint32_t>(sizeof(drv_input));
345 : const std::unique_ptr<GroupQueryOutput> k_drv_output_ptr(new (std::nothrow) GroupQueryOutput());
346 : if (k_drv_output_ptr == nullptr) {
347 : UDF_LOG_ERROR("Malloc failed.");
348 : return FsmStatus::kFsmFailed;
349 : }
350 : uint32_t drv_output_len = 0U;
351 : GroupQueryOutput *drv_output = k_drv_output_ptr.get();
352 : int32_t drv_ret = halGrpQuery(GRP_QUERY_GROUP_ADDR_INFO, &drv_input, k_drv_input_len, drv_output, &drv_output_len);
353 : if (drv_ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
354 : UDF_LOG_ERROR("Call halGrpQuery failed, ret=%d.", drv_ret);
355 : return FsmStatus::kFsmDrvFailed;
356 : }
357 : size_t output_num = (static_cast<size_t>(drv_output_len) / sizeof(GrpQueryGroupAddrInfo));
358 : for (size_t i = 0; i < output_num; ++i) {
359 : void *addr_ptr = reinterpret_cast<void *>(static_cast<uintptr_t>(drv_output->grpQueryGroupAddrInfo[i].addr));
360 : HcclResult hccl_ret = HcclRegisterGlobalMemory(addr_ptr, drv_output->grpQueryGroupAddrInfo[i].size);
361 : if (hccl_ret != HcclResult::HCCL_SUCCESS) {
362 : UDF_LOG_ERROR("Call HcclRegisterGlobalMemory failed, ret:%d.", hccl_ret);
363 : return FsmStatus::kFsmHcclFailed;
364 : }
365 : UDF_LOG_INFO("Register mr success, addr:%llu, size:%llu.", drv_output->grpQueryGroupAddrInfo[i].addr,
366 : drv_output->grpQueryGroupAddrInfo[i].size);
367 : mem_addrs.emplace_back(drv_output->grpQueryGroupAddrInfo[i].addr);
368 : }
369 : return FsmStatus::kFsmSuccess;
370 : }
371 :
372 : FsmStatus LlmCommEntityMgr::QueryCurMemGrp(char **group_name) {
373 : const std::unique_ptr<GroupQueryOutput> k_drv_output_ptr(new (std::nothrow) GroupQueryOutput());
374 : if (k_drv_output_ptr == nullptr) {
375 : UDF_LOG_ERROR("Malloc failed.");
376 : return FsmStatus::kFsmFailed;
377 : }
378 : GroupQueryOutput *drv_output = k_drv_output_ptr.get();
379 : uint32_t drv_output_len = 0U;
380 : auto cur_pid = drvDeviceGetBareTgid();
381 : auto drv_ret = halGrpQuery(GRP_QUERY_GROUPS_OF_PROCESS, &cur_pid, static_cast<uint32_t>(sizeof(cur_pid)), drv_output,
382 : &drv_output_len);
383 : if (drv_ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
384 : UDF_LOG_ERROR("Call halGrpQuery failed, ret=%d.", drv_ret);
385 : return FsmStatus::kFsmDrvFailed;
386 : }
387 : if (drv_output_len == 0U) {
388 : UDF_LOG_ERROR("Query current mem group failed, size is zero.");
389 : return FsmStatus::kFsmFailed;
390 : }
391 : if ((drv_output_len / sizeof(drv_output->grpQueryGroupsOfProcInfo[0])) > 1) {
392 : UDF_LOG_WARN("Query current mem group not expected over 1, size:%u.", drv_output_len);
393 : }
394 : errno_t ret = strcpy_s(*group_name, sizeof(drv_output->grpQueryGroupsOfProcInfo[0].groupName),
395 : drv_output->grpQueryGroupsOfProcInfo[0].groupName);
396 : if (ret != EOK) {
397 : UDF_LOG_ERROR("Copy group name failed, ret=%d!", ret);
398 : return FsmStatus::kFsmFailed;
399 : }
400 : return FsmStatus::kFsmSuccess;
401 : }
402 :
403 : FsmStatus LlmCommEntityMgr::UnRegisterHcclMr(std::vector<uint64_t> &mem_addrs) {
404 : for (const auto &k_mem_addr : mem_addrs) {
405 : void *addr_ptr = reinterpret_cast<void *>(static_cast<uintptr_t>(k_mem_addr));
406 : HcclResult ret = HcclUnregisterGlobalMemory(addr_ptr);
407 : if (ret != HcclResult::HCCL_SUCCESS) {
408 : UDF_LOG_ERROR("Unregister mr failed, ret:%d, addr:%lu", ret, k_mem_addr);
409 : return FsmStatus::kFsmHcclFailed;
410 : }
411 : UDF_LOG_INFO("Unregister mr success, addr:%lu", k_mem_addr);
412 : }
413 : return FsmStatus::kFsmSuccess;
414 : }
415 :
416 : void LlmCommEntityMgr::ClearEntities() {
417 : std::lock_guard<std::mutex> lock(entity_mutex_);
418 64 : ip_to_conns_.clear();
419 : server_entity_map_.clear();
420 : client_entity_map_.clear();
421 : }
422 :
423 12 : EntityPtr LlmCommEntityMgr::FindServerEntityByIp(uint32_t ip, bool &cleared_residual) {
424 12 : cleared_residual = false;
425 12 : std::lock_guard<std::mutex> lock(entity_mutex_);
426 12 : auto iter = ip_to_conns_.find(ip);
427 12 : if (iter == ip_to_conns_.end()) {
428 11 : return nullptr;
429 : }
430 1 : const HcclConn conn = iter->second;
431 1 : auto entity_iter = server_entity_map_.find(conn);
432 1 : if (entity_iter == server_entity_map_.end()) {
433 0 : EraseIpToConnMap(ip, conn);
434 0 : cleared_residual = true;
435 0 : return nullptr;
436 : }
437 1 : return entity_iter->second;
438 12 : }
439 :
440 : void LlmCommEntityMgr::EraseIpToConnMap(uint32_t ip, const HcclConn conn) {
441 : for (auto multi_iter = ip_to_conns_.find(ip); multi_iter != ip_to_conns_.end(); multi_iter++) {
442 : if (multi_iter->second == conn) {
443 : (void)ip_to_conns_.erase(multi_iter);
444 : return;
445 : }
446 : }
447 : }
448 :
449 : void LlmCommEntityMgr::EraseClientMapByClusterId(uint64_t remote_cluster_id) {
450 : auto iter = client_entity_map_.find(remote_cluster_id);
451 : if (iter != client_entity_map_.end()) {
452 : client_entity_map_.erase(iter);
453 : }
454 : }
455 :
456 : void LlmCommEntityMgr::DumpServerEntities() {
457 : std::unordered_map<HcclConn, EntityPtr> entity_map;
458 : {
459 : std::lock_guard<std::mutex> lock(entity_mutex_);
460 : entity_map = server_entity_map_;
461 : }
462 : for (auto &iter : entity_map) {
463 : iter.second->Dump();
464 : }
465 : }
466 :
467 : void LlmCommEntityMgr::DumpClientEntities() {
468 : std::unordered_map<uint64_t, EntityPtr> entity_map;
469 : {
470 : std::lock_guard<std::mutex> lock(entity_mutex_);
471 : entity_map = client_entity_map_;
472 : }
473 : for (auto &iter : entity_map) {
474 : iter.second->Dump();
475 : }
476 : }
477 :
478 : bool LlmCommEntityMgr::HasAnyLink() {
479 : mgr_need_use_mtx_.store(true, std::memory_order_relaxed);
480 : ScopeGuard guard([this] { mgr_need_use_mtx_.store(false, std::memory_order_relaxed); });
481 : std::lock_guard<std::mutex> lock(entity_mutex_);
482 : UDF_LOG_INFO("server_entity_map_.size=%zu, client_entity_map_.size=%zu", server_entity_map_.size(),
483 : client_entity_map_.size());
484 : return (!server_entity_map_.empty()) || (!client_entity_map_.empty());
485 : }
486 :
487 : void LlmCommEntityMgr::FinalizeServerConn() {
488 : initialized_.store(false);
489 : std::lock_guard<std::mutex> lock(switch_mutex_);
490 : if (listen_conn_ != nullptr) {
491 : (void)HcclRawClose(listen_conn_);
492 : listen_conn_ = nullptr;
493 : UDF_LOG_INFO("server listen conn closed");
494 : }
495 : }
496 :
497 : size_t LlmCommEntityMgr::QueryLinkNum() const {
498 : return client_entity_map_.size();
499 : }
500 : } // namespace FlowFunc
|
