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