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_manager.h"
12 : #include <algorithm>
13 : #include "bqs_util.h"
14 : #include "client_entity.h"
15 : #include "common/bqs_log.h"
16 : #include "driver/ascend_hal_external.h"
17 : #include "group_entity.h"
18 : #include "queue_manager.h"
19 : #include "schedule_config.h"
20 : #include "simple_entity.h"
21 : #include "state_manager.h"
22 :
23 : namespace dgw {
24 : namespace {
25 : const uint32_t SHIFT32 = 32U;
26 : // default channel request capacity
27 : constexpr size_t DEFAULT_CHANNEL_CAPACITY = 32UL;
28 : }
29 :
30 7 : EntityManager::~EntityManager()
31 : {
32 7 : (void)pthread_rwlock_destroy(&srcCommChannels_.lock);
33 7 : (void)pthread_rwlock_destroy(&dstCommChannels_.lock);
34 7 : }
35 :
36 1929 : EntityManager &EntityManager::Instance(const uint32_t resIndex)
37 : {
38 1929 : if (resIndex != 0U) {
39 20 : static EntityManager instanceExtra(resIndex);
40 20 : return instanceExtra;
41 : }
42 :
43 1909 : static EntityManager instance(resIndex);
44 1909 : return instance;
45 : }
46 :
47 1731 : EntityPtr EntityManager::DoGetEntity(const uint32_t queueType, const uint32_t deviceId, const EntityType eType,
48 : const uint32_t id, const EntityDirection direction)
49 : {
50 1731 : const auto iterDevId = idToEntity_.find(deviceId);
51 1731 : if (iterDevId == idToEntity_.end()) {
52 370 : return nullptr;
53 : }
54 :
55 1361 : const auto qTypeIter = iterDevId->second.find(queueType);
56 1361 : if (qTypeIter == iterDevId->second.end()) {
57 44 : return nullptr;
58 : }
59 :
60 1317 : const auto iterType = qTypeIter->second.find(eType);
61 1317 : if (iterType == qTypeIter->second.end()) {
62 81 : return nullptr;
63 : }
64 :
65 1236 : const auto itEntityIter = iterType->second.find(id);
66 1236 : if (itEntityIter == iterType->second.end()) {
67 341 : return nullptr;
68 : }
69 :
70 895 : EntityPtr entity = nullptr;
71 1793 : for (const auto &elem: itEntityIter->second) {
72 898 : if (elem->GetDirection() == direction) {
73 891 : entity = elem;
74 : }
75 : }
76 895 : return entity;
77 895 : }
78 :
79 1433 : EntityPtr EntityManager::GetEntityById(const uint32_t queueType, const uint32_t deviceId, const EntityType eType,
80 : const uint32_t id, const EntityDirection direction)
81 : {
82 1433 : EntityPtr entity = DoGetEntity(queueType, deviceId, eType, id, direction);
83 1433 : if (entity != nullptr) {
84 612 : DGW_LOG_DEBUG("Direction:[%s] type:[%s] id:[%u] state:[%s] queue id:[%u] "
85 : "device id:[%u] owner deviceid:[%u] is found",
86 : GetDirectionDesc(entity->GetDirection()), entity->GetTypeDesc().c_str(), id,
87 : entity->GetStateDesc(entity->GetCurState()).c_str(),
88 : entity->GetQueueId(), deviceId, entity->GetDeviceId());
89 : } else {
90 821 : DGW_LOG_WARN("Type:[%s] id:[%u], device id:[%u], queueType:[%u], direction[%d] does not exist",
91 : StateManager::Instance().GetTypeDesc(eType).c_str(), id, deviceId,
92 : queueType, static_cast<int32_t>(direction));
93 : }
94 1433 : return entity;
95 0 : }
96 :
97 1 : EntityPtr EntityManager::GetSrcEntityByGlobalId(const uint32_t key, const uint32_t globalId) const
98 : {
99 1 : EntityPtr entity = nullptr;
100 1 : uint64_t uniqKey = key;
101 1 : uniqKey = (uniqKey << SHIFT32) | static_cast<uint64_t>(globalId);
102 1 : const auto iter = globalIdToSrcEntity_.find(uniqKey);
103 1 : if (iter != globalIdToSrcEntity_.end()) {
104 1 : entity = iter->second;
105 : }
106 1 : return entity;
107 0 : }
108 :
109 1 : EntityPtr EntityManager::GetDstEntityByGlobalId(const uint32_t key, const uint32_t globalId) const
110 : {
111 1 : EntityPtr entity = nullptr;
112 1 : uint64_t uniqKey = key;
113 1 : uniqKey = (uniqKey << SHIFT32) | static_cast<uint64_t>(globalId);
114 1 : const auto iter = globalIdToDstEntity_.find(uniqKey);
115 1 : if (iter != globalIdToDstEntity_.end()) {
116 1 : entity = iter->second;
117 : }
118 1 : return entity;
119 0 : }
120 :
121 28 : const std::vector<EntityPtr> &EntityManager::GetEntitiesInGroup(const uint32_t groupId)
122 : {
123 28 : static const std::vector<EntityPtr> emptyVec;
124 28 : const auto iter = groupEntityMap_.find(groupId);
125 28 : if (iter != groupEntityMap_.end()) {
126 23 : return iter->second;
127 : }
128 5 : return emptyVec;
129 : }
130 :
131 26 : FsmStatus EntityManager::CreateGroup(const uint32_t groupId, std::vector<EntityPtr>& entities)
132 : {
133 26 : const auto iter = groupEntityMap_.find(groupId);
134 : // when group src bind to multi dst
135 26 : if (iter != groupEntityMap_.end()) {
136 1 : DGW_LOG_WARN("groupId[%u] has exist.", groupId);
137 1 : return FsmStatus::FSM_SUCCESS;
138 : }
139 25 : (void) groupEntityMap_.insert(std::make_pair(groupId, entities));
140 25 : return FsmStatus::FSM_SUCCESS;
141 : }
142 :
143 25 : FsmStatus EntityManager::DeleteGroup(const uint32_t groupId)
144 : {
145 25 : const auto iter = groupEntityMap_.find(groupId);
146 25 : if (iter == groupEntityMap_.end()) {
147 3 : DGW_LOG_WARN("groupId[%u] does not exist.", groupId);
148 3 : return FsmStatus::FSM_SUCCESS;
149 : }
150 22 : (void) groupEntityMap_.erase(iter);
151 22 : return FsmStatus::FSM_SUCCESS;
152 : }
153 :
154 296 : EntityPtr EntityManager::CreateEntity(const EntityMaterial &material)
155 : {
156 296 : auto entity = GetEntityById(material.queueType, material.resId, material.eType, material.id, material.direction);
157 296 : if (entity != nullptr) {
158 1 : DGW_LOG_INFO("Type:[%s] id:[%u], direction:[%d] is existed", entity->GetTypeDesc().c_str(), material.id,
159 : static_cast<int32_t>(material.direction));
160 1 : entity->IncreaseRefCount();
161 1 : return entity;
162 : }
163 :
164 : // alloc entity
165 295 : entity = AllocEntity(material);
166 295 : if (entity == nullptr) {
167 1 : DGW_LOG_ERROR("Create [%s] entity failed type:[%s] id:[%u].", GetDirectionDesc(material.direction),
168 : StateManager::Instance().GetTypeDesc(material.eType).c_str(), material.id);
169 1 : return nullptr;
170 : }
171 :
172 : // init and save entity
173 294 : FsmState state = FsmState::FSM_IDLE_STATE;
174 294 : if (material.direction == EntityDirection::DIRECTION_RECV) {
175 174 : state = FsmState::FSM_WAIT_PUSH_STATE;
176 : }
177 294 : const auto ret = entity->Init(state, material.direction);
178 294 : if (ret != FsmStatus::FSM_SUCCESS) {
179 1 : DGW_LOG_ERROR("Init entity failed, direction:[%s].", entity->ToString().c_str());
180 1 : return nullptr;
181 : }
182 293 : entity->IncreaseRefCount();
183 293 : idToEntity_[material.resId][material.queueType][material.eType][material.id].emplace_back(entity);
184 :
185 293 : uint64_t uniqKey = material.schedCfgKey;
186 293 : uniqKey = (uniqKey << SHIFT32) | static_cast<uint64_t>(material.globalId);
187 293 : if (material.direction == EntityDirection::DIRECTION_RECV) {
188 174 : DGW_LOG_INFO("add globalIdToDstEntity_ for %u:%u", material.schedCfgKey, material.globalId);
189 174 : globalIdToDstEntity_[uniqKey] = entity;
190 : } else {
191 119 : DGW_LOG_INFO("add globalIdToSrcEntity_ for %u:%u", material.schedCfgKey, material.globalId);
192 119 : globalIdToSrcEntity_[uniqKey] = entity;
193 : }
194 293 : DGW_LOG_INFO("Create entity[%s] success, state:[%s]",
195 : entity->ToString().c_str(), entity->GetStateDesc(entity->GetCurState()).c_str());
196 293 : bqs::StatisticManager::GetInstance().SetExistEntityFlag(true);
197 :
198 293 : if (material.eType == dgw::EntityType::ENTITY_TAG) {
199 29 : const ChannelEntityPtr channelEntity = std::dynamic_pointer_cast<ChannelEntity>(entity);
200 29 : (void)InsertCommChannel(channelEntity, material.direction == EntityDirection::DIRECTION_SEND);
201 29 : }
202 293 : return entity;
203 296 : }
204 :
205 298 : FsmStatus EntityManager::DeleteEntity(const uint32_t queueType, const uint32_t deviceId, const EntityType eType,
206 : const uint32_t id, const EntityDirection direction) {
207 298 : EntityPtr entity = DoGetEntity(queueType, deviceId, eType, id, direction);
208 298 : if (entity == nullptr) {
209 19 : DGW_LOG_WARN("Failed to find entity for queueType[%u], deviceId[%u], type[%s], id:[%u], direction:[%d]",
210 : queueType, deviceId, StateManager::Instance().GetTypeDesc(eType).c_str(), id,
211 : static_cast<int32_t>(direction));
212 19 : return FsmStatus::FSM_SUCCESS;
213 : }
214 :
215 : // delete entity
216 279 : entity->DecreaseRefCount();
217 279 : if (entity->GetRefCount() != 0U) {
218 1 : return FsmStatus::FSM_SUCCESS;
219 : }
220 278 : auto &entityVec = idToEntity_[deviceId][queueType][eType][id];
221 278 : auto entityVectorIter = entityVec.begin();
222 279 : while (entityVectorIter != entityVec.end()) {
223 279 : if ((*entityVectorIter)->GetDirection() == direction) {
224 278 : DGW_LOG_INFO("Delete entity[%s] from entityMap", entity->ToString().c_str());
225 278 : entityVec.erase(entityVectorIter);
226 278 : break;
227 : }
228 1 : ++entityVectorIter;
229 : }
230 278 : CleanEntityMap(queueType, deviceId, eType, id);
231 :
232 278 : uint64_t uniqKey = entity->GetSchedCfgKey();
233 278 : uniqKey = (uniqKey << SHIFT32) | static_cast<uint64_t>(entity->GetGlobalId());
234 278 : if (direction == EntityDirection::DIRECTION_RECV) {
235 166 : globalIdToDstEntity_.erase(uniqKey);
236 : } else {
237 112 : globalIdToSrcEntity_.erase(uniqKey);
238 : }
239 :
240 278 : DGW_LOG_INFO("Success to delete entity[%s], direction:[%s].",
241 : entity->ToString().c_str(), GetDirectionDesc(direction));
242 284 : if (!dgw::ScheduleConfig::GetInstance().GetSchedKeys().empty() &&
243 6 : (eType == dgw::EntityType::ENTITY_QUEUE)) {
244 5 : dgw::DynamicSchedMgr::GetInstance(resIndex_).DeleteQueue(entity->GetGlobalId(), entity->GetSchedCfgKey());
245 : }
246 :
247 278 : if (eType == dgw::EntityType::ENTITY_TAG) {
248 29 : (void)EraseCommChannel(entity, direction == EntityDirection::DIRECTION_SEND);
249 : }
250 278 : DGW_LOG_INFO("Finish to delete entity[%s], direction:[%s].",
251 : entity->ToString().c_str(), GetDirectionDesc(direction));
252 278 : (void)entity->Uninit();
253 278 : return FsmStatus::FSM_SUCCESS;
254 298 : }
255 :
256 278 : void EntityManager::CleanEntityMap(const uint32_t queueType, const uint32_t deviceId, const EntityType eType,
257 : const uint32_t id)
258 : {
259 278 : if (!idToEntity_[deviceId][queueType][eType][id].empty()) {
260 1 : return;
261 : }
262 :
263 277 : DGW_LOG_INFO("Erase id[%u] in entityMap[%u:%u:%d].", id, deviceId, queueType, static_cast<int32_t>(eType));
264 277 : idToEntity_[deviceId][queueType][eType].erase(id);
265 277 : if (!idToEntity_[deviceId][queueType][eType].empty()) {
266 127 : return;
267 : }
268 :
269 150 : DGW_LOG_INFO("Erase eType[%u] in entityMap[%u:%u].", static_cast<int32_t>(eType), deviceId, queueType);
270 150 : idToEntity_[deviceId][queueType].erase(eType);
271 150 : if (!idToEntity_[deviceId][queueType].empty()) {
272 42 : return;
273 : }
274 :
275 108 : DGW_LOG_INFO("Erase queueType[%u] in entityMap[%u].", queueType, deviceId);
276 108 : idToEntity_[deviceId].erase(queueType);
277 108 : if (!idToEntity_[deviceId].empty()) {
278 22 : return;
279 : }
280 :
281 86 : DGW_LOG_INFO("Erase deviceId[%u] in entityMap.", deviceId);
282 86 : idToEntity_.erase(deviceId);
283 86 : if (idToEntity_.empty()) {
284 84 : bqs::StatisticManager::GetInstance().SetExistEntityFlag(false);
285 : }
286 : }
287 :
288 5 : FsmStatus EntityManager::ProbeSrcCommChannel(
289 : const std::function<FsmStatus(const ChannelEntityPtr &, uint32_t &)> procFunc)
290 : {
291 5 : bool emptySched = true;
292 5 : (void)pthread_rwlock_rdlock(&srcCommChannels_.lock);
293 5 : auto &entities = srcCommChannels_.entities;
294 5 : DGW_LOG_INFO("EntityManager[%u] srcCommChannels_'s size is %zu", resIndex_, entities.size());
295 10 : for (auto iter = entities.begin(); iter != entities.end(); ++iter) {
296 5 : uint32_t probeCount = 0U;
297 5 : (void)procFunc(*iter, probeCount);
298 6 : emptySched = (probeCount != 0U) ? false : emptySched;
299 : }
300 5 : (void)pthread_rwlock_unlock(&srcCommChannels_.lock);
301 5 : if (emptySched) {
302 1 : bqs::StatisticManager::GetInstance().HcclMpiRecvReqEmptySchedStat();
303 : }
304 5 : return FsmStatus::FSM_SUCCESS;
305 1 : }
306 :
307 14 : FsmStatus EntityManager::SupplyRecvRequestEvent()
308 : {
309 14 : return SupplyEventForRecvRequest(EVENT_RECV_REQUEST_MSG);
310 : }
311 :
312 3 : FsmStatus EntityManager::SupplyOneTrackEvent()
313 : {
314 3 : return SupplyEventForRecvRequest(EVENT_RECV_COMPLETION_MSG);
315 : }
316 :
317 18 : FsmStatus EntityManager::SupplyEventForRecvRequest(uint32_t msgType)
318 : {
319 6 : const auto checkFunc = [](const ChannelEntityPtr entity)->bool {
320 6 : return entity->CheckRecvReqEventContinue();
321 : };
322 :
323 18 : bool supplyEventFlag = false;
324 18 : (void)pthread_rwlock_rdlock(&srcCommChannels_.lock);
325 18 : const auto &entities = srcCommChannels_.entities;
326 18 : supplyEventFlag = std::any_of(entities.begin(), entities.end(), checkFunc);
327 18 : (void)pthread_rwlock_unlock(&srcCommChannels_.lock);
328 :
329 18 : if (supplyEventFlag) {
330 1 : (void)SupplyEvent(msgType);
331 1 : bqs::StatisticManager::GetInstance().RecvReqEventSupplyStat();
332 1 : DGW_LOG_DEBUG("Supply receive request event[%u].", msgType);
333 : }
334 18 : return FsmStatus::FSM_SUCCESS;
335 : }
336 :
337 11 : FsmStatus EntityManager::TestSomeCommChannels(
338 : const std::function<FsmStatus(CommChannels &, uint32_t &, uint32_t &)> procFunc, const bool isSrc)
339 : {
340 11 : uint32_t totalCompCount = 0U;
341 11 : CommChannels &channels = isSrc ? srcCommChannels_ : dstCommChannels_;
342 11 : auto ret = FsmStatus::FSM_SUCCESS;
343 11 : (void)pthread_rwlock_rdlock(&channels.lock);
344 11 : ret = procFunc(channels, totalCompCount, resIndex_);
345 11 : (void)pthread_rwlock_unlock(&channels.lock);
346 11 : if (totalCompCount == 0U) {
347 5 : isSrc ? bqs::StatisticManager::GetInstance().HcclMpiRecvCompEmptySchedStat() :
348 3 : bqs::StatisticManager::GetInstance().HcclMpiSendCompEmptySchedStat();
349 : }
350 11 : return ret;
351 : }
352 :
353 33 : FsmStatus EntityManager::EraseCommChannel(const EntityPtr &entity, const bool isSrc)
354 : {
355 33 : CommChannels &channels = isSrc ? srcCommChannels_ : dstCommChannels_;
356 33 : (void)pthread_rwlock_wrlock(&channels.lock);
357 33 : auto &entities = channels.entities;
358 :
359 33 : const auto iter = std::find(entities.begin(), entities.end(), entity);
360 33 : if (iter != entities.end()) {
361 33 : (void)entities.erase(iter);
362 33 : DGW_LOG_INFO("Success to erase entity:[%s]", entity->ToString().c_str());
363 : }
364 33 : (void)pthread_rwlock_unlock(&channels.lock);
365 :
366 33 : DGW_LOG_INFO("Success to erase comm channel entity, entity:[%s], isSrc:[%d] comm channels size:[%zu].",
367 : entity->ToString().c_str(), static_cast<int32_t>(isSrc), channels.entities.size());
368 33 : return FsmStatus::FSM_SUCCESS;
369 : }
370 :
371 29 : FsmStatus EntityManager::InsertCommChannel(const ChannelEntityPtr &entity, const bool isSrc)
372 : {
373 29 : CommChannels &channels = isSrc ? srcCommChannels_ : dstCommChannels_;
374 29 : (void)pthread_rwlock_wrlock(&channels.lock);
375 29 : (void)channels.entities.push_back(entity);
376 29 : const size_t curCapacity = channels.requests.capacity();
377 29 : if (curCapacity < channels.entities.size()) {
378 2 : const size_t newCapacity = curCapacity + DEFAULT_CHANNEL_CAPACITY;
379 2 : channels.requests.reserve(newCapacity);
380 2 : channels.compIndices.reserve(newCapacity);
381 2 : channels.compStatus.reserve(newCapacity);
382 2 : DGW_LOG_INFO("Success to reserve requests capacity, primary:[%zu], current:[%zu]", curCapacity, newCapacity);
383 : }
384 29 : (void)pthread_rwlock_unlock(&channels.lock);
385 :
386 29 : DGW_LOG_INFO("Success to save comm channel entity:[%s], isSrc:[%d] comm channels size:[%zu].",
387 : entity->ToString().c_str(), static_cast<int32_t>(isSrc), channels.entities.size());
388 29 : return FsmStatus::FSM_SUCCESS;
389 : }
390 :
391 294 : EntityPtr EntityManager::AllocEntity(const EntityMaterial &material) const
392 : {
393 294 : EntityPtr entity = nullptr;
394 : try {
395 294 : DoAllocEntity(material, entity);
396 0 : } catch (std::exception &e) {
397 0 : DGW_LOG_ERROR("catch %s", e.what());
398 0 : }
399 294 : return entity;
400 0 : }
401 :
402 294 : void EntityManager::DoAllocEntity(const EntityMaterial &material, EntityPtr &entity) const
403 : {
404 294 : if (material.eType == dgw::EntityType::ENTITY_TAG) {
405 30 : entity = std::make_shared<ChannelEntity>(material, resIndex_);
406 264 : } else if (material.eType == dgw::EntityType::ENTITY_GROUP) {
407 19 : entity = std::make_shared<GroupEntity>(material, resIndex_);
408 : } else {
409 245 : if (material.queueType == bqs::CLIENT_Q) {
410 22 : entity = std::make_shared<ClientEntity>(material, resIndex_);
411 : } else {
412 223 : entity = std::make_shared<SimpleEntity>(material, resIndex_);
413 : }
414 : }
415 294 : }
416 :
417 6 : CommChannels &EntityManager::GetCommChannels(const bool isSrc)
418 : {
419 6 : return isSrc ? srcCommChannels_ : dstCommChannels_;
420 : }
421 :
422 6 : FsmStatus EntityManager::SupplyEvent(const uint32_t eventId) const
423 : {
424 6 : if (resIndex_ != 0U) {
425 2 : return SupplyEvent(eventId, bqs::QueueManager::GetInstance().GetExtraDeviceId(),
426 2 : static_cast<uint32_t>(bqs::EventGroupId::ENQUEUE_GROUP_ID_EXTRA));
427 : }
428 4 : return SupplyEvent(eventId, bqs::QueueManager::GetInstance().GetDeviceId(),
429 4 : static_cast<uint32_t>(bqs::EventGroupId::ENQUEUE_GROUP_ID));
430 : }
431 :
432 6 : FsmStatus EntityManager::SupplyEvent(const uint32_t eventId, const uint32_t deviceId, const uint32_t groupId) const
433 : {
434 6 : uint32_t submitDeviceId = deviceId;
435 6 : uint32_t submitGroupId = groupId;
436 9 : if (!bqs::GlobalCfg::GetInstance().GetNumaFlag() &&
437 3 : ((eventId == static_cast<uint32_t>(EVENT_RECV_REQUEST_MSG)) ||
438 3 : (eventId == static_cast<uint32_t>(EVENT_SEND_COMPLETION_MSG)) ||
439 : (eventId == static_cast<uint32_t>(EVENT_RECV_COMPLETION_MSG)))) {
440 3 : submitGroupId = static_cast<uint32_t>(bqs::EventGroupId::F2NF_GROUP_ID);
441 3 : submitDeviceId = bqs::QueueManager::GetInstance().GetDeviceId();
442 : }
443 6 : event_summary sched = { };
444 6 : sched.pid = getpid();
445 6 : sched.grp_id = submitGroupId;
446 6 : sched.event_id = static_cast<EVENT_ID>(eventId);
447 6 : sched.dst_engine = ACPU_LOCAL;
448 6 : const auto ret = halEschedSubmitEvent(submitDeviceId, &sched);
449 6 : if (ret != DRV_ERROR_NONE) {
450 1 : DGW_LOG_ERROR("Call halEschedSumbmitEvent failed, event:[%u], deviceId[%u], groupId[%u], ret:[%d].",
451 : eventId, submitDeviceId, submitGroupId, static_cast<int32_t>(ret));
452 1 : return FsmStatus::FSM_FAILED;
453 : }
454 5 : return FsmStatus::FSM_SUCCESS;
455 : }
456 :
457 7 : FsmStatus EntityManager::CheckLinkStatus()
458 : {
459 7 : (void)pthread_rwlock_rdlock(&srcCommChannels_.lock);
460 7 : auto &srcEntities = srcCommChannels_.entities;
461 8 : for (auto iter = srcEntities.begin(); iter != srcEntities.end(); ++iter) {
462 2 : if ((*iter)->linkStatus_ == dgw::ChannelLinkStatus::UNCONNECTED) {
463 1 : DGW_LOG_INFO("CheckLinkStatus srcCommChannels queueId[%u] unconnected", (*iter)->GetQueueId());
464 1 : (void)pthread_rwlock_unlock(&srcCommChannels_.lock);
465 1 : return FsmStatus::FSM_FAILED;
466 : }
467 : }
468 6 : (void)pthread_rwlock_unlock(&srcCommChannels_.lock);
469 :
470 6 : (void)pthread_rwlock_rdlock(&dstCommChannels_.lock);
471 6 : auto &dstEntities = dstCommChannels_.entities;
472 7 : for (auto iter = dstEntities.begin(); iter != dstEntities.end(); ++iter) {
473 2 : if ((*iter)->linkStatus_ == dgw::ChannelLinkStatus::UNCONNECTED) {
474 1 : DGW_LOG_INFO("CheckLinkStatus dstCommChannels_ queueId[%u] unconnected", (*iter)->GetQueueId());
475 1 : (void)pthread_rwlock_unlock(&dstCommChannels_.lock);
476 1 : return FsmStatus::FSM_FAILED;
477 : }
478 : }
479 5 : (void)pthread_rwlock_unlock(&dstCommChannels_.lock);
480 5 : DGW_LOG_INFO("CheckLinkStatus connect success");
481 5 : return FsmStatus::FSM_SUCCESS;
482 : }
483 : } // namespace dgw
|
