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 "bind_relation.h"
12 : #include <atomic>
13 : #include <numeric>
14 : #include <queue>
15 : #include <algorithm>
16 : #include "common/bqs_log.h"
17 : #include "subscribe_manager.h"
18 : #include "statistic_manager.h"
19 : #include "entity_manager.h"
20 : #include "schedule_config.h"
21 : #include "hccl/comm_channel_manager.h"
22 :
23 : namespace bqs {
24 : namespace {
25 : // RELATION_UPPER_BOUND: 64 * 1024
26 : constexpr const uint32_t RELATION_UPPER_BOUND = 65536U;
27 : } // namespace
28 :
29 580 : BindRelation &BindRelation::GetInstance()
30 : {
31 580 : static BindRelation instance;
32 580 : return instance;
33 : }
34 :
35 159 : BqsStatus BindRelation::CheckMultiLayerBind(const EntityInfo& srcEntity, const EntityInfo& dstEntity,
36 : const uint32_t index) const
37 : {
38 159 : const MapEnitityInfoToInfoSet &srcToDstRelation = (index == 0) ? srcToDstRelation_ : srcToDstRelationExtra_;
39 159 : const MapEnitityInfoToInfoSet &dstToSrcRelation = (index == 0) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
40 :
41 159 : const auto srcToDstIter = srcToDstRelation.find(dstEntity);
42 159 : if (srcToDstIter != srcToDstRelation.end()) {
43 2 : auto dstSets = srcToDstIter->second;
44 4 : for (auto itDst = dstSets.begin(); itDst != dstSets.end(); ++itDst) {
45 3 : if ((*itDst).GetType() == dgw::EntityType::ENTITY_TAG) {
46 1 : BQS_LOG_WARN("Bind relation[%s->%s] ignore check multi layer bind.",
47 : (srcToDstIter->first).ToString().c_str(), (*itDst).ToString().c_str());
48 1 : return BQS_STATUS_OK;
49 : }
50 : }
51 :
52 1 : BQS_LOG_ERROR("Bind relation[%s->*] already exists, can't add relation [%s->%s], suggest to add "
53 : "relation[%s->*] directly.",
54 : (srcToDstIter->first).ToString().c_str(), srcEntity.ToString().c_str(),
55 : dstEntity.ToString().c_str(), srcEntity.ToString().c_str());
56 1 : return BQS_STATUS_PARAM_INVALID;
57 2 : }
58 :
59 157 : const auto dstToSrcIter = dstToSrcRelation.find(srcEntity);
60 157 : if (dstToSrcIter != dstToSrcRelation.end()) {
61 1 : BQS_LOG_ERROR("Bind relation[*->%s] already exists, can't add relation [%s->%s], suggest to add "
62 : "relation[*->%s] directly.",
63 : (dstToSrcIter->first).ToString().c_str(), srcEntity.ToString().c_str(),
64 : dstEntity.ToString().c_str(), dstEntity.ToString().c_str());
65 1 : return BQS_STATUS_PARAM_INVALID;
66 : }
67 156 : return BQS_STATUS_OK;
68 : }
69 :
70 :
71 160 : BqsStatus BindRelation::CheckEntityExistInGroup(const EntityInfo& src, const EntityInfo& dst,
72 : const uint32_t resIndex) const
73 : {
74 160 : if (src.GetType() == dgw::EntityType::ENTITY_GROUP) {
75 10 : const auto &entitiesInGroup = GetEntitiesInGroup(src.GetId());
76 25 : for (auto iter = entitiesInGroup.begin(); iter != entitiesInGroup.end(); ++iter) {
77 16 : const auto &element = (*(*iter));
78 16 : if (dst == element) {
79 0 : BQS_LOG_ERROR("dst entity[%s] has exist in src group entity[%s].",
80 : dst.ToString().c_str(), src.ToString().c_str());
81 1 : return BQS_STATUS_PARAM_INVALID;
82 : }
83 16 : const auto existSrcEntity = dgw::EntityManager::Instance(resIndex).GetEntityById(
84 : element.GetQueueType(), element.GetDeviceId(), element.GetType(),
85 16 : element.GetId(), dgw::EntityDirection::DIRECTION_SEND);
86 17 : if ((existSrcEntity != nullptr) &&
87 1 : (existSrcEntity->GetHostGroupId() != static_cast<int32_t>(src.GetId()))) {
88 1 : BQS_LOG_ERROR("Entity[%s] in group[%s] exists in other src side.",
89 : element.ToString().c_str(), src.ToString().c_str());
90 1 : return BQS_STATUS_PARAM_INVALID;
91 : }
92 16 : }
93 : } else {
94 150 : const auto srcEntityPtr = dgw::EntityManager::Instance(resIndex).
95 : GetEntityById(src.GetQueueType(), src.GetDeviceId(), src.GetType(),
96 150 : src.GetId(), dgw::EntityDirection::DIRECTION_SEND);
97 150 : if (srcEntityPtr != nullptr) {
98 57 : const int32_t groupId = srcEntityPtr->GetHostGroupId();
99 57 : if (groupId != dgw::INVALID_GROUP_ID) {
100 1 : BQS_LOG_ERROR("Entity[%s] has exist in group[%d].", src.ToString().c_str(), groupId);
101 1 : return BQS_STATUS_PARAM_INVALID;
102 : }
103 56 : BQS_LOG_INFO("Entity[%s] has been created.", src.ToString().c_str());
104 : }
105 150 : }
106 :
107 158 : if (dst.GetType() == dgw::EntityType::ENTITY_GROUP) {
108 12 : const auto &entitiesInGroup = GetEntitiesInGroup(dst.GetId());
109 32 : for (auto iter = entitiesInGroup.begin(); iter != entitiesInGroup.end(); ++iter) {
110 22 : const auto &element = (*(*iter));
111 22 : if (src == element) {
112 1 : BQS_LOG_ERROR("src entity[%s] has exist in dst group entity[%s].",
113 : src.ToString().c_str(), dst.ToString().c_str());
114 2 : return BQS_STATUS_PARAM_INVALID;
115 : }
116 21 : const auto existDstEntity = dgw::EntityManager::Instance(resIndex).
117 : GetEntityById(element.GetQueueType(), element.GetDeviceId(), element.GetType(),
118 21 : element.GetId(), dgw::EntityDirection::DIRECTION_RECV);
119 22 : if ((existDstEntity != nullptr) &&
120 1 : (existDstEntity->GetHostGroupId() != static_cast<int32_t>(dst.GetId()))) {
121 1 : BQS_LOG_ERROR("Entity[%s] in group[%s] exists in other dst side.",
122 : element.ToString().c_str(), dst.ToString().c_str());
123 1 : return BQS_STATUS_PARAM_INVALID;
124 : }
125 21 : }
126 : } else {
127 146 : const auto dstEntityPtr = dgw::EntityManager::Instance(resIndex).
128 : GetEntityById(dst.GetQueueType(), dst.GetDeviceId(), dst.GetType(),
129 146 : dst.GetId(), dgw::EntityDirection::DIRECTION_RECV);
130 146 : if (dstEntityPtr != nullptr) {
131 7 : const int32_t groupId = dstEntityPtr->GetHostGroupId();
132 7 : if (groupId != dgw::INVALID_GROUP_ID) {
133 1 : BQS_LOG_ERROR("Entity[%s] has exist in group[%d].", dst.ToString().c_str(), groupId);
134 1 : return BQS_STATUS_PARAM_INVALID;
135 : }
136 6 : BQS_LOG_INFO("Entity[%s] has been created.", dst.ToString().c_str());
137 : }
138 146 : }
139 155 : return BQS_STATUS_OK;
140 : }
141 :
142 163 : BqsStatus BindRelation::CheckBind(const EntityInfo& srcEntity, const EntityInfo& dstEntity,
143 : const uint32_t resIndex, uint32_t &index) const
144 : {
145 163 : if (srcEntity == dstEntity) {
146 1 : BQS_LOG_ERROR("Bind relation[%s->%s] failed, as can't bind to self.", srcEntity.ToString().c_str(),
147 : dstEntity.ToString().c_str());
148 1 : return BQS_STATUS_PARAM_INVALID;
149 : }
150 :
151 162 : if (GetBindRelationIndex(srcEntity, dstEntity, index) != BQS_STATUS_OK) {
152 1 : BQS_LOG_ERROR("GetBindRelationIndex error");
153 1 : return BQS_STATUS_PARAM_INVALID;
154 : }
155 :
156 161 : if (resIndex != index) {
157 1 : BQS_LOG_INFO("relation[%s->%s] should be processed by threads[%u] while current threads[%u]",
158 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), index, resIndex);
159 1 : return BQS_STATUS_OK;
160 : }
161 :
162 160 : const MapEnitityInfoToInfoSet &dstToSrcRelation = (index == 0) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
163 160 : if (dstToSrcRelation.size() >= RELATION_UPPER_BOUND) {
164 0 : BQS_LOG_ERROR("Bind relation[%u->%u] failed, as the maximum number of relation supported is %u, "
165 : "current number is %zu.",
166 : srcEntity.GetId(), dstEntity.GetId(), RELATION_UPPER_BOUND, dstToSrcRelation.size());
167 0 : return BQS_STATUS_INNER_ERROR;
168 : }
169 160 : const auto dstToSrcIter = dstToSrcRelation.find(dstEntity);
170 160 : if ((dstToSrcIter != dstToSrcRelation.end()) && (dstToSrcIter->second.count(srcEntity) != 0U)) {
171 1 : BQS_LOG_WARN("Bind relation[%s->%s] already exists, no need bind.", srcEntity.ToString().c_str(),
172 : dstEntity.ToString().c_str());
173 1 : (void)UpdateSubscribeEvent(srcEntity, EventType::ENQUEUE, index);
174 1 : (void)UpdateSubscribeEvent(dstEntity, EventType::F2NF, index);
175 1 : return BQS_STATUS_OK;
176 : }
177 :
178 159 : const auto &abnormalDstToSrcIter = abnormalDstToSrc_.find(dstEntity);
179 159 : if ((abnormalDstToSrcIter != abnormalDstToSrc_.end()) && (abnormalDstToSrcIter->second.count(srcEntity) != 0U)) {
180 1 : BQS_LOG_WARN("Bind relation[%s->%s] already exists in abnormal bind relations, cannot bind.",
181 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
182 1 : return BQS_STATUS_OK;
183 : }
184 :
185 158 : auto ret = CheckMultiLayerBind(srcEntity, dstEntity, index);
186 158 : if (ret != BQS_STATUS_OK) {
187 2 : return ret;
188 : }
189 :
190 : // check whether entity exist in group
191 156 : ret = CheckEntityExistInGroup(srcEntity, dstEntity, index);
192 156 : if (ret != BQS_STATUS_OK) {
193 1 : return ret;
194 : }
195 155 : return BQS_STATUS_OK;
196 : }
197 :
198 632 : BqsStatus BindRelation::SetEntityPtr(EntityInfo &entityInfo, const dgw::EntityDirection direction,
199 : const uint32_t index) const
200 : {
201 632 : if (entityInfo.GetEntity() != nullptr) {
202 372 : return BQS_STATUS_OK;
203 : }
204 :
205 260 : const auto entity = dgw::EntityManager::Instance(index).GetEntityById(
206 260 : entityInfo.GetQueueType(), entityInfo.GetDeviceId(), entityInfo.GetType(), entityInfo.GetId(), direction);
207 260 : if (entity == nullptr) {
208 2 : BQS_LOG_ERROR("Missing entity [%s].", entityInfo.ToString().c_str());
209 2 : return BQS_STATUS_INNER_ERROR;
210 : }
211 258 : entityInfo.SetEntity(entity);
212 258 : return BQS_STATUS_OK;
213 260 : }
214 :
215 159 : BqsStatus BindRelation::AddSrcToDst(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t index)
216 : {
217 159 : BqsStatus ret = BQS_STATUS_OK;
218 317 : if ((SetEntityPtr(srcEntity, dgw::EntityDirection::DIRECTION_SEND, index) != BQS_STATUS_OK) ||
219 158 : (SetEntityPtr(dstEntity, dgw::EntityDirection::DIRECTION_RECV, index) != BQS_STATUS_OK)) {
220 1 : BQS_LOG_ERROR("Bind relation add [%s->%s] failed becuause of missing entity.",
221 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
222 1 : return BQS_STATUS_INNER_ERROR;
223 : }
224 :
225 158 : MapEnitityInfoToInfoSet &srcToDstRelation = (index == 0) ? srcToDstRelation_ : srcToDstRelationExtra_;
226 158 : const auto iter = srcToDstRelation.find(srcEntity);
227 158 : if (iter == srcToDstRelation.end()) {
228 103 : ret = SubscribeEvent(srcEntity, EventType::ENQUEUE, index);
229 103 : if (ret != BQS_STATUS_OK) {
230 1 : BQS_LOG_ERROR("Bind relation add [%s->%s] failed, as subscribe failed, ret=%d.",
231 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), static_cast<int32_t>(ret));
232 1 : return ret;
233 : } else {
234 102 : (void)srcToDstRelation.emplace(
235 408 : std::make_pair(srcEntity, EntityInfoSet{ dstEntity }));
236 : }
237 : } else {
238 55 : (void)iter->second.emplace(dstEntity);
239 : }
240 :
241 157 : return ret;
242 102 : }
243 :
244 158 : BqsStatus BindRelation::AddDstToSrc(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t index)
245 : {
246 158 : BqsStatus ret = BQS_STATUS_OK;
247 :
248 315 : if ((SetEntityPtr(srcEntity, dgw::EntityDirection::DIRECTION_SEND, index) != BQS_STATUS_OK) ||
249 157 : (SetEntityPtr(dstEntity, dgw::EntityDirection::DIRECTION_RECV, index) != BQS_STATUS_OK)) {
250 1 : BQS_LOG_ERROR("Bind relation add [%s->%s] failed becuause of missing entity.",
251 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
252 1 : return BQS_STATUS_INNER_ERROR;
253 : }
254 :
255 157 : MapEnitityInfoToInfoSet &dstToSrcRelation = (index == 0) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
256 157 : const auto iter = dstToSrcRelation.find(dstEntity);
257 157 : if (iter == dstToSrcRelation.end()) {
258 152 : ret = SubscribeEvent(dstEntity, EventType::F2NF, index);
259 152 : if (ret != BQS_STATUS_OK) {
260 0 : BQS_LOG_ERROR("Bind relation add [%s->%s] failed, as subscribe f2nf failed, ret=%d.",
261 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), static_cast<int32_t>(ret));
262 : } else {
263 152 : (void)dstToSrcRelation.emplace(
264 608 : std::make_pair(dstEntity, EntityInfoSet{ srcEntity }));
265 : }
266 : } else {
267 5 : (void)iter->second.emplace(srcEntity);
268 : }
269 157 : return ret;
270 152 : }
271 :
272 149 : BqsStatus BindRelation::DelSrcToDst(const EntityInfo& srcEntity, const EntityInfo& dstEntity, const uint32_t index)
273 : {
274 149 : auto ret = BQS_STATUS_OK;
275 149 : MapEnitityInfoToInfoSet &srcToDstRelation = (index == 0) ? srcToDstRelation_ : srcToDstRelationExtra_;
276 149 : const auto iter = srcToDstRelation.find(srcEntity);
277 149 : if (iter == srcToDstRelation.end()) {
278 7 : BQS_LOG_WARN("Bind relation[%s->%s] dst doesn't exist, no need unbind.", srcEntity.ToString().c_str(),
279 : dstEntity.ToString().c_str());
280 7 : return BQS_STATUS_OK;
281 : }
282 :
283 142 : (void)iter->second.erase(dstEntity);
284 142 : if (iter->second.empty()) {
285 88 : (void)UnsubscribeEvent(srcEntity, EventType::ENQUEUE, index);
286 88 : BQS_LOG_INFO("delete route [%s->*]", srcEntity.ToString().c_str());
287 88 : (void)srcToDstRelation.erase(iter);
288 : }
289 :
290 142 : return ret;
291 : }
292 :
293 148 : BqsStatus BindRelation::DelDstToSrc(const EntityInfo& srcEntity, const EntityInfo& dstEntity, const uint32_t index)
294 : {
295 148 : MapEnitityInfoToInfoSet &dstToSrcRelation = (index == 0) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
296 148 : auto ret = BQS_STATUS_OK;
297 148 : const auto iter = dstToSrcRelation.find(dstEntity);
298 148 : if (iter == dstToSrcRelation.end()) {
299 8 : BQS_LOG_WARN("Bind relation[%s->%s] dst doesn't exist, no need unbind.", srcEntity.ToString().c_str(),
300 : dstEntity.ToString().c_str());
301 8 : return BQS_STATUS_OK;
302 : }
303 :
304 140 : (void)iter->second.erase(srcEntity);
305 140 : if (iter->second.empty()) {
306 138 : (void)UnsubscribeEvent(dstEntity, EventType::F2NF, index);
307 138 : (void)dstToSrcRelation.erase(iter);
308 : }
309 :
310 140 : return ret;
311 : }
312 :
313 164 : BqsStatus BindRelation::Bind(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t resIndex)
314 : {
315 164 : uint32_t index = 0;
316 164 : auto ret = CheckBind(srcEntity, dstEntity, resIndex, index);
317 164 : if (ret != BQS_STATUS_OK) {
318 4 : return ret;
319 : }
320 :
321 160 : if (index != resIndex) {
322 1 : return BQS_STATUS_RETRY;
323 : }
324 :
325 159 : ret = CreateEntity(srcEntity, dstEntity, index);
326 159 : if (ret != BQS_STATUS_OK) {
327 0 : return ret;
328 : }
329 :
330 159 : ret = AddSrcToDst(srcEntity, dstEntity, index);
331 159 : if (ret != BQS_STATUS_OK) {
332 1 : (void) DeleteEntity(srcEntity, true, index);
333 1 : (void) DeleteEntity(dstEntity, false, index);
334 1 : return ret;
335 : }
336 :
337 158 : ret = AddDstToSrc(srcEntity, dstEntity, index);
338 158 : if (ret != BQS_STATUS_OK) {
339 : // roll back
340 1 : (void) DelSrcToDst(srcEntity, dstEntity, index);
341 1 : (void) DeleteEntity(srcEntity, true, index);
342 1 : (void) DeleteEntity(dstEntity, false, index);
343 1 : BQS_LOG_ERROR("Bind relation add [%s->%s] failed, as subscribe f2nf failed, ret=%d.",
344 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), static_cast<int32_t>(ret));
345 1 : return ret;
346 : }
347 :
348 157 : BQS_LOG_RUN_INFO("Bind relation add {src[%s]->dst[%s]} success on resIndex[%u].",
349 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), resIndex);
350 157 : return BQS_STATUS_OK;
351 : }
352 :
353 144 : BqsStatus BindRelation::UnBind(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t resIndex)
354 : {
355 : // check unbind for src and dst entity
356 144 : auto ret = CheckUnBind(srcEntity);
357 144 : if (ret != BQS_STATUS_OK) {
358 1 : return ret;
359 : }
360 143 : ret = CheckUnBind(dstEntity);
361 143 : if (ret != BQS_STATUS_OK) {
362 1 : return ret;
363 : }
364 :
365 142 : uint32_t index = 0U;
366 142 : if (GetBindIndexBySrc(srcEntity, index) != BQS_STATUS_OK) {
367 1 : BQS_LOG_WARN("GetBindIndexBySrc failed");
368 1 : return BQS_STATUS_OK;
369 : }
370 141 : if (index != resIndex) {
371 1 : return BQS_STATUS_RETRY;
372 : }
373 :
374 140 : ret = DelSrcToDst(srcEntity, dstEntity, resIndex);
375 140 : if (ret == BQS_STATUS_OK) {
376 139 : ret = DelDstToSrc(srcEntity, dstEntity, resIndex);
377 139 : if (ret != BQS_STATUS_OK) {
378 : // roll back
379 0 : (void)AddSrcToDst(srcEntity, dstEntity, resIndex);
380 : }
381 : }
382 140 : if (ret != BQS_STATUS_OK) {
383 1 : BQS_LOG_ERROR("Bind relation del [%s->%s] failed, bqsStatus=%d.",
384 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), static_cast<int32_t>(ret));
385 1 : return ret;
386 : }
387 :
388 : // no roll back if delete entity failed
389 : // check src entity whether exist multi bind
390 139 : auto &srcToDstRelation = (resIndex == 0U) ? srcToDstRelation_ : srcToDstRelationExtra_;
391 139 : auto &dstToSrcRelation = (resIndex == 0U) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
392 139 : const auto srcIter = srcToDstRelation.find(srcEntity);
393 139 : if (srcIter == srcToDstRelation.end()) {
394 87 : (void)DeleteEntity(srcEntity, true, resIndex);
395 : }
396 : // check dst entity whether exist multi bind
397 139 : const auto dstIter = dstToSrcRelation.find(dstEntity);
398 139 : if (dstIter == dstToSrcRelation.end()) {
399 137 : (void)DeleteEntity(dstEntity, false, resIndex);
400 : }
401 :
402 : // delete abnormal bind relation
403 139 : DelAbnormalSrcToDst(srcEntity, dstEntity);
404 139 : DelAbnormalDstToSrc(srcEntity, dstEntity);
405 :
406 139 : return BQS_STATUS_OK;
407 : }
408 :
409 5 : BqsStatus BindRelation::UnBindBySrc(const EntityInfo& srcEntity)
410 : {
411 5 : UnBindAbnormalRelationBySrc(srcEntity);
412 5 : return UnBindRelationBySrc(srcEntity);
413 : }
414 :
415 14 : BqsStatus BindRelation::UnBindRelationBySrc(const EntityInfo& srcEntity)
416 : {
417 14 : uint32_t index = 0;
418 14 : if (GetBindIndexBySrc(srcEntity, index) != BQS_STATUS_OK) {
419 0 : BQS_LOG_WARN("GetBindIndexBySrc failed");
420 0 : return BQS_STATUS_OK;
421 : }
422 :
423 14 : MapEnitityInfoToInfoSet &srcToDstRelation = (index == 0) ? srcToDstRelation_ : srcToDstRelationExtra_;
424 14 : const auto srcToDstIter = srcToDstRelation.find(srcEntity);
425 14 : if (srcToDstIter == srcToDstRelation.end()) {
426 5 : BQS_LOG_WARN("No relation [%s->*] exists, no need unbind", srcEntity.ToString().c_str());
427 5 : return BQS_STATUS_OK;
428 : }
429 :
430 9 : const auto ret = UnsubscribeEvent(srcEntity, EventType::ENQUEUE, index);
431 9 : if (ret != BQS_STATUS_OK) {
432 1 : BQS_LOG_ERROR("Unsubscribe queue[%s] failed, bqsStatus=%d.", srcEntity.ToString().c_str(),
433 : static_cast<int32_t>(ret));
434 1 : return BQS_STATUS_DRIVER_ERROR;
435 : }
436 :
437 8 : MapEnitityInfoToInfoSet &dstToSrcRelation = (index == 0) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
438 17 : for (const auto &dstEntity : srcToDstIter->second) {
439 9 : (void)DelDstToSrc(srcEntity, dstEntity, index);
440 : // delete dst entity
441 9 : const auto dstIter = dstToSrcRelation.find(dstEntity);
442 9 : if (dstIter == dstToSrcRelation.end()) {
443 9 : (void)DeleteEntity(dstEntity, false, index);
444 : }
445 : }
446 :
447 8 : (void)srcToDstRelation.erase(srcToDstIter);
448 : // delete src entity
449 8 : (void)DeleteEntity(srcEntity, true, index);
450 8 : return BQS_STATUS_OK;
451 : }
452 :
453 7 : BqsStatus BindRelation::UnBindByDst(const EntityInfo& dstEntity)
454 : {
455 7 : UnBindAbnormalRelationByDst(dstEntity);
456 7 : return UnBindRelationByDst(dstEntity);
457 : }
458 :
459 11 : BqsStatus BindRelation::UnBindRelationByDst(const EntityInfo& dstEntity)
460 : {
461 11 : uint32_t index = 0;
462 11 : if (GetBindIndexByDst(dstEntity, index) != BQS_STATUS_OK) {
463 2 : BQS_LOG_WARN("GetBindIndexByDst failed");
464 2 : return BQS_STATUS_OK;
465 : }
466 :
467 9 : MapEnitityInfoToInfoSet &dstToSrcRelation = (index == 0) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
468 9 : const auto dstToSrcIter = dstToSrcRelation.find(dstEntity);
469 9 : if (dstToSrcIter == dstToSrcRelation.end()) {
470 2 : BQS_LOG_WARN("No bind relation[*->%s] exists, no need unbind.", dstEntity.ToString().c_str());
471 2 : return BQS_STATUS_OK;
472 : }
473 :
474 7 : auto ret = UnsubscribeEvent(dstEntity, EventType::F2NF, index);
475 7 : if (ret != BQS_STATUS_OK) {
476 0 : BQS_LOG_ERROR("UnsubscribeFullToNotFull queue[%s] failed, bqsStatus=%d.", dstEntity.ToString().c_str(),
477 : static_cast<int32_t>(ret));
478 0 : return BQS_STATUS_DRIVER_ERROR;
479 : }
480 :
481 7 : MapEnitityInfoToInfoSet &srcToDstRelation = (index == 0) ? srcToDstRelation_ : srcToDstRelationExtra_;
482 16 : for (const auto &srcEntity : dstToSrcIter->second) {
483 9 : ret = DelSrcToDst(srcEntity, dstEntity, index);
484 9 : if (ret != BQS_STATUS_OK) {
485 0 : break;
486 : }
487 : // delete src entity
488 9 : const auto srcIter = srcToDstRelation.find(srcEntity);
489 9 : if (srcIter == srcToDstRelation.end()) {
490 7 : (void)DeleteEntity(srcEntity, true, index);
491 : }
492 : }
493 :
494 7 : if (ret == BQS_STATUS_OK) {
495 7 : (void)dstToSrcRelation.erase(dstToSrcIter);
496 : // delete dst entity
497 7 : (void)DeleteEntity(dstEntity, false, index);
498 : } else {
499 0 : BQS_LOG_ERROR("Bind relation del [*->%s] failed, bqsStatus=%d.", dstEntity.ToString().c_str(),
500 : static_cast<int32_t>(ret));
501 : }
502 :
503 7 : return ret;
504 : }
505 :
506 : // topsort
507 136 : void BindRelation::Order(const uint32_t index)
508 : {
509 136 : StatisticManager::GetInstance().BindNum(CountBinds());
510 136 : StatisticManager::GetInstance().AbnormalBindNum(CountAbnormalBinds());
511 272 : StatisticManager::GetInstance().SubscribeNum(
512 136 : static_cast<uint32_t>(srcToDstRelation_.size() + srcToDstRelationExtra_.size()));
513 136 : if (index == 0U) {
514 134 : OrderOneTable(orderedSubscribeQueueId_, srcToDstRelation_, dstToSrcRelation_);
515 : } else {
516 2 : OrderOneTable(orderedSubscribeQueueIdExtra_, srcToDstRelationExtra_, dstToSrcRelationExtra_);
517 : }
518 136 : }
519 :
520 136 : void BindRelation::OrderOneTable(std::vector<EntityInfo> &orderedSubscribeQueueId,
521 : const MapEnitityInfoToInfoSet &srcToDstRelation, const MapEnitityInfoToInfoSet &dstToSrcRelation)
522 : {
523 136 : isHasLoop_ = false;
524 136 : orderedSubscribeQueueId.clear();
525 136 : orderedSubscribeQueueId.reserve(srcToDstRelation.size());
526 136 : std::unordered_map<EntityInfo, uint32_t, EntityInfoHash> inDegrees;
527 :
528 : // as queue input edge is only one, so we can use order by traverse
529 136 : std::queue<EntityInfo> subscribeQueue;
530 229 : for (auto &iter : srcToDstRelation) {
531 : // No input queue is head queue
532 93 : if (dstToSrcRelation.count(iter.first) == 0U) {
533 92 : (void)subscribeQueue.emplace(iter.first);
534 : }
535 191 : for (const auto &dstQ : iter.second) {
536 98 : const auto degIter = inDegrees.find(dstQ);
537 98 : if (degIter != inDegrees.end()) {
538 2 : degIter->second++;
539 : } else {
540 96 : inDegrees[dstQ] = 1U;
541 : }
542 : }
543 : }
544 :
545 : // protect for loop
546 324 : while (!subscribeQueue.empty()) {
547 188 : auto queueId = subscribeQueue.front();
548 188 : subscribeQueue.pop();
549 188 : const auto &dstQueueIter = srcToDstRelation.find(queueId);
550 188 : if (dstQueueIter == srcToDstRelation.end()) {
551 95 : continue;
552 : }
553 :
554 93 : orderedSubscribeQueueId.emplace_back(dstQueueIter->first);
555 191 : for (auto dstQueueId : dstQueueIter->second) {
556 98 : --(inDegrees[dstQueueId]);
557 98 : if (inDegrees[dstQueueId] == 0U) {
558 96 : (void)subscribeQueue.emplace(dstQueueId);
559 : }
560 98 : }
561 188 : }
562 :
563 136 : if (orderedSubscribeQueueId.size() != srcToDstRelation.size()) {
564 0 : BQS_LOG_ERROR("orderedSubscribeQueueId.size is [%zu] is not equal to srcToDstRelation.size[%zu], "
565 : "may be with loop in bind relation, use unordered instead.",
566 : orderedSubscribeQueueId.size(), srcToDstRelation.size());
567 0 : isHasLoop_ = true;
568 0 : orderedSubscribeQueueId.clear();
569 0 : for (auto &srcToDstIter : srcToDstRelation) {
570 0 : (void)orderedSubscribeQueueId.emplace_back(srcToDstIter.first);
571 : }
572 : }
573 136 : }
574 :
575 92 : const MapEnitityInfoToInfoSet &BindRelation::GetSrcToDstRelation() const
576 : {
577 92 : return srcToDstRelation_;
578 : }
579 :
580 189 : const MapEnitityInfoToInfoSet &BindRelation::GetDstToSrcRelation() const
581 : {
582 189 : return dstToSrcRelation_;
583 : }
584 :
585 19 : const MapEnitityInfoToInfoSet &BindRelation::GetAbnormalSrcToDstRelation() const
586 : {
587 19 : return abnormalSrcToDst_;
588 : }
589 :
590 9 : const MapEnitityInfoToInfoSet &BindRelation::GetAbnormalDstToSrcRelation() const
591 : {
592 9 : return abnormalDstToSrc_;
593 : }
594 :
595 29 : const std::vector<EntityInfo> &BindRelation::GetOrderedSubscribeQueueId() const
596 : {
597 29 : return orderedSubscribeQueueId_;
598 : }
599 :
600 136 : uint32_t BindRelation::CountBinds() const
601 : {
602 408 : uint32_t bindCount = std::accumulate(std::begin(srcToDstRelation_), std::end(srcToDstRelation_), 0U,
603 94 : [](const uint32_t previous,
604 : const std::pair<EntityInfo, EntityInfoSet> &dstQueueId) {
605 94 : return previous + static_cast<uint32_t>(dstQueueId.second.size());
606 : });
607 136 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
608 30 : bindCount += std::accumulate(std::begin(srcToDstRelationExtra_), std::end(srcToDstRelationExtra_), 0U,
609 2 : [](const uint32_t previous,
610 : const std::pair<EntityInfo, EntityInfoSet> &dstQueueId) {
611 2 : return previous + static_cast<uint32_t>(dstQueueId.second.size());
612 : });
613 : }
614 136 : return bindCount;
615 : }
616 :
617 136 : uint32_t BindRelation::CountAbnormalBinds() const
618 : {
619 408 : const uint32_t bindCount = std::accumulate(std::begin(abnormalSrcToDst_), std::end(abnormalSrcToDst_), 0U,
620 18 : [](const uint32_t previous,
621 : const std::pair<EntityInfo, EntityInfoSet> &abnormalBinds) {
622 18 : return previous + static_cast<uint32_t>(abnormalBinds.second.size());
623 : });
624 136 : return bindCount;
625 : }
626 :
627 26 : BqsStatus BindRelation::CreateGroup(const std::vector<EntityInfoPtr> &entities, uint32_t &groupId)
628 : {
629 26 : if (entities.empty()) {
630 0 : BQS_LOG_ERROR("entity is empty.");
631 0 : return BQS_STATUS_PARAM_INVALID;
632 : }
633 :
634 : // generate group id
635 26 : groupId = GenerateGroupId();
636 : // save to allGroupConfig
637 26 : const auto ret = allGroupConfig_.emplace(std::make_pair(groupId, entities));
638 26 : if (!ret.second) {
639 0 : BQS_LOG_ERROR("create group [%u] failed.", groupId);
640 0 : return BQS_STATUS_GROUP_HAS_EXIST;
641 : }
642 26 : return BQS_STATUS_OK;
643 : }
644 :
645 9 : BqsStatus BindRelation::DeleteGroup(const uint32_t groupId)
646 : {
647 9 : const auto indexIter = group2ResIndex_.find(groupId);
648 9 : if (indexIter != group2ResIndex_.end()) {
649 : const MapEnitityInfoToInfoSet &srcToDstRelation =
650 4 : (indexIter->second.first == 0U) ? srcToDstRelation_ : srcToDstRelationExtra_;
651 : const MapEnitityInfoToInfoSet &dstToSrcRelation =
652 4 : (indexIter->second.first == 0U) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
653 4 : OptionalArg args = {};
654 4 : args.eType = dgw::EntityType::ENTITY_GROUP;
655 4 : const EntityInfo group(groupId, indexIter->second.second, &args);
656 8 : if ((srcToDstRelation.find(group) != srcToDstRelation.end()) ||
657 8 : (dstToSrcRelation.find(group) != dstToSrcRelation.end())) {
658 1 : BQS_LOG_ERROR("group[%u] still exist in routes. Please delete route first.", groupId);
659 1 : return BQS_STATUS_GROUP_EXIST_IN_ROUTE;
660 : }
661 4 : }
662 :
663 8 : const auto iter = allGroupConfig_.find(groupId);
664 8 : if (iter == allGroupConfig_.end()) {
665 1 : BQS_LOG_RUN_INFO("group %u does not exist.", groupId);
666 1 : return BQS_STATUS_OK;
667 : }
668 24 : for (auto infoPtr : iter->second) {
669 17 : if (infoPtr->GetType() == dgw::EntityType::ENTITY_TAG) {
670 2 : (void)dgw::CommChannelManager::GetInstance().DeleteCommChannel(*(infoPtr->GetCommChannel()));
671 : }
672 17 : }
673 7 : (void)allGroupConfig_.erase(iter);
674 7 : if (indexIter != group2ResIndex_.end()) {
675 3 : (void) group2ResIndex_.erase(indexIter);
676 : }
677 7 : return BQS_STATUS_OK;
678 : }
679 :
680 161 : BqsStatus BindRelation::CreateEntity(const EntityInfo &src, const EntityInfo &dst, const uint32_t resIndex)
681 : {
682 : // when one src entity bind with multi dst entities, src entity may has been created
683 161 : const auto srcRet = CreateEntity(src, true, resIndex);
684 161 : if ((srcRet != BQS_STATUS_OK) && (srcRet != BQS_STATUS_ENTITY_EXIST)) {
685 : // roolback
686 0 : (void) DeleteEntity(src, true, resIndex);
687 0 : return srcRet;
688 : }
689 161 : const auto dstRet = CreateEntity(dst, false, resIndex);
690 161 : if ((dstRet != BQS_STATUS_OK) && (dstRet != BQS_STATUS_ENTITY_EXIST)) {
691 : // roolback
692 0 : (void) DeleteEntity(dst, false, resIndex);
693 0 : if (srcRet == BQS_STATUS_ENTITY_EXIST) {
694 0 : (void) DeleteEntity(src, true, resIndex);
695 : }
696 0 : return dstRet;
697 : }
698 :
699 : // set needTransId for entity: no need check nullptr
700 161 : const auto srcEntity = dgw::EntityManager::Instance(resIndex).GetEntityById(
701 161 : src.GetQueueType(), src.GetDeviceId(), src.GetType(), src.GetId(), dgw::EntityDirection::DIRECTION_SEND);
702 161 : if (srcEntity == nullptr) {
703 1 : BQS_LOG_ERROR("Missing entity [%s].", src.ToString().c_str());
704 1 : return BQS_STATUS_INNER_ERROR;
705 : }
706 160 : if (srcEntity->IsNeedTransId()) {
707 0 : return BQS_STATUS_OK;
708 : }
709 : // if dst entity is group, src entity need get transId
710 160 : if (dst.GetType() == dgw::EntityType::ENTITY_GROUP) {
711 10 : srcEntity->SetNeedTransId(true);
712 10 : BQS_LOG_INFO("entity[%s] need get transId when scheduled because dst entity[%s].",
713 : src.ToString().c_str(), dst.ToString().c_str());
714 10 : return BQS_STATUS_OK;
715 : }
716 150 : return BQS_STATUS_OK;
717 161 : }
718 :
719 322 : BqsStatus BindRelation::CreateEntity(const EntityInfo &info, const bool isSrc, const uint32_t resIndex)
720 : {
721 : // check entity exist
722 322 : const dgw::EntityDirection direction = isSrc ? dgw::EntityDirection::DIRECTION_SEND :
723 : dgw::EntityDirection::DIRECTION_RECV;
724 322 : const auto entity = dgw::EntityManager::Instance(resIndex).GetEntityById(
725 322 : info.GetQueueType(), info.GetDeviceId(), info.GetType(), info.GetId(), direction);
726 322 : if (entity != nullptr) {
727 66 : BQS_LOG_INFO("Entity[%s] has been created, no need created again.", info.ToString().c_str());
728 66 : return BQS_STATUS_ENTITY_EXIST;
729 : }
730 :
731 : // create entity for [group]
732 256 : if (info.GetType() == dgw::EntityType::ENTITY_GROUP) {
733 19 : const auto ret = CreateEntityForGroup(info, isSrc, resIndex);
734 19 : return ret;
735 : }
736 : // create entify for [queue or channel]
737 237 : dgw::EntityMaterial material = {};
738 237 : material.eType = info.GetType();
739 237 : material.direction = direction;
740 237 : material.id = info.GetId();
741 237 : material.globalId = info.GetGlobalId();
742 237 : material.uuId = info.GetUuId();
743 237 : material.schedCfgKey = info.GetSchedCfgKey();
744 237 : material.resId = info.GetDeviceId();
745 237 : material.channel = info.GetCommChannel();
746 237 : material.queueType = info.GetQueueType();
747 237 : if (nullptr == dgw::EntityManager::Instance(resIndex).CreateEntity(material)) {
748 0 : BQS_LOG_ERROR("Create entityPtr for entity[%s] failed.", info.ToString().c_str());
749 0 : return BQS_STATUS_INNER_ERROR;
750 : }
751 237 : return BQS_STATUS_OK;
752 322 : }
753 :
754 20 : BqsStatus BindRelation::CreateEntityForGroup(const EntityInfo &groupEntity, const bool isSrc,
755 : const uint32_t resIndex)
756 : {
757 20 : const uint32_t groupId = groupEntity.GetId();
758 20 : BQS_LOG_INFO("Begin to create entityPtr for group[%u].", groupId);
759 20 : const std::vector<EntityInfoPtr> &entities = GetEntitiesInGroup(groupId);
760 20 : if (entities.empty()) {
761 1 : BQS_LOG_ERROR("group %u does not exist.", groupId);
762 1 : return BQS_STATUS_INNER_ERROR;
763 : }
764 :
765 19 : const dgw::EntityDirection direction = isSrc ? dgw::EntityDirection::DIRECTION_SEND :
766 : dgw::EntityDirection::DIRECTION_RECV;
767 19 : std::vector<dgw::EntityPtr> entityPtrVec;
768 : // create entity in group
769 54 : for (auto &info : entities) {
770 35 : dgw::EntityMaterial material = {};
771 35 : material.eType = info->GetType();
772 35 : material.direction = direction;
773 35 : material.id = info->GetId();
774 35 : material.globalId = info->GetGlobalId();
775 35 : material.uuId = info->GetUuId();
776 35 : material.schedCfgKey = info->GetSchedCfgKey();
777 35 : material.resId = info->GetDeviceId();
778 35 : material.channel = info->GetCommChannel();
779 35 : material.hostGroupId = static_cast<int32_t>(groupId);
780 35 : material.queueType = info->GetQueueType();
781 35 : auto entityPtr = dgw::EntityManager::Instance(resIndex).CreateEntity(material);
782 35 : if (entityPtr == nullptr) {
783 0 : BQS_LOG_ERROR("Create entityPtr for entity[%s] failed.", info->ToString().c_str());
784 0 : return BQS_STATUS_INNER_ERROR;
785 : }
786 35 : entityPtrVec.emplace_back(entityPtr);
787 : // src entity in group need transId
788 35 : entityPtr->SetNeedTransId(true);
789 35 : BQS_LOG_RUN_INFO("Entity group[%u] add element[%s] success.", groupId, info->ToString().c_str());
790 35 : }
791 : // save group
792 19 : const dgw::FsmStatus ret = dgw::EntityManager::Instance(resIndex).CreateGroup(groupId, entityPtrVec);
793 19 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
794 0 : BQS_LOG_ERROR("Save group[%u] failed.", groupId);
795 0 : return BQS_STATUS_INNER_ERROR;
796 : }
797 19 : group2ResIndex_[groupId] = std::make_pair(resIndex, groupEntity.GetDeviceId());
798 :
799 : // create group entity
800 19 : dgw::EntityMaterial material = {};
801 19 : material.eType = dgw::EntityType::ENTITY_GROUP;
802 19 : material.direction = direction;
803 19 : material.id = groupId;
804 19 : material.globalId = groupEntity.GetGlobalId();
805 19 : material.uuId = groupEntity.GetUuId();
806 19 : material.schedCfgKey = groupEntity.GetSchedCfgKey();
807 19 : material.resId = groupEntity.GetDeviceId();
808 19 : material.groupPolicy = groupEntity.GetGroupPolicy();
809 19 : material.peerInstanceNum = groupEntity.GetPeerInstanceNum();
810 19 : material.localInstanceIndex = groupEntity.GetLocalInstanceIndex();
811 19 : material.queueType = groupEntity.GetQueueType();
812 19 : const auto groupEntityPtr = dgw::EntityManager::Instance(resIndex).CreateEntity(material);
813 19 : if (groupEntityPtr == nullptr) {
814 0 : BQS_LOG_ERROR("Create entityPtr for group[%u] failed.", groupId);
815 0 : return BQS_STATUS_INNER_ERROR;
816 : }
817 19 : return BQS_STATUS_OK;
818 19 : }
819 :
820 260 : BqsStatus BindRelation::DeleteEntity(const EntityInfo &info, const bool isSrc, const uint32_t resIndex) const
821 : {
822 260 : BQS_LOG_INFO("DeleteEntity: %s", info.ToString().c_str());
823 260 : const dgw::EntityDirection direction = isSrc ? dgw::EntityDirection::DIRECTION_SEND :
824 : dgw::EntityDirection::DIRECTION_RECV;
825 : // group
826 260 : if (info.GetType() == dgw::EntityType::ENTITY_GROUP) {
827 20 : return DeleteEntityForGroup(info.GetQueueType(), info.GetDeviceId(), info.GetId(), direction, resIndex);
828 : }
829 : // queue or tag
830 240 : const auto ret = dgw::EntityManager::Instance(resIndex).DeleteEntity(
831 : info.GetQueueType(), info.GetDeviceId(), info.GetType(), info.GetId(), direction);
832 240 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
833 0 : BQS_LOG_ERROR("Delete entity[%s] failed.", info.ToString().c_str());
834 0 : return BQS_STATUS_INNER_ERROR;
835 : }
836 240 : return BQS_STATUS_OK;
837 : }
838 :
839 20 : BqsStatus BindRelation::DeleteEntityForGroup(
840 : const uint32_t queueType, const uint32_t deviceId, const uint32_t groupId, const dgw::EntityDirection direction,
841 : const uint32_t resIndex) const
842 : {
843 : // delete group entity
844 20 : dgw::FsmStatus ret = dgw::EntityManager::Instance(resIndex).
845 20 : DeleteEntity(queueType, deviceId, dgw::EntityType::ENTITY_GROUP, groupId, direction);
846 20 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
847 0 : BQS_LOG_ERROR("Delete group entity[%u] failed.", groupId);
848 0 : return BQS_STATUS_INNER_ERROR;
849 : }
850 : // delete group
851 20 : ret = dgw::EntityManager::Instance(resIndex).DeleteGroup(groupId);
852 20 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
853 0 : BQS_LOG_ERROR("Delete group[%u] failed.", groupId);
854 0 : return BQS_STATUS_INNER_ERROR;
855 : }
856 : // delete entity in group
857 20 : const std::vector<EntityInfoPtr> &entities = GetEntitiesInGroup(groupId);
858 55 : for (auto &info : entities) {
859 35 : ret = dgw::EntityManager::Instance(resIndex).DeleteEntity(
860 : info->GetQueueType(), info->GetDeviceId(), info->GetType(), info->GetId(), direction);
861 35 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
862 0 : BQS_LOG_ERROR("delete entityPtr for entity[%s] failed.", info->ToString().c_str());
863 0 : return BQS_STATUS_INNER_ERROR;
864 : }
865 : }
866 20 : return BQS_STATUS_OK;
867 : }
868 :
869 26 : uint32_t BindRelation::GenerateGroupId()
870 : {
871 : static uint32_t groupId = 0U;
872 : uint32_t currGroupId;
873 26 : lockForGroup_.Lock();
874 26 : ++groupId;
875 26 : currGroupId = groupId;
876 26 : lockForGroup_.Unlock();
877 26 : return currGroupId;
878 : }
879 :
880 122 : const std::vector<EntityInfoPtr> &BindRelation::GetEntitiesInGroup(const uint32_t groupId) const
881 : {
882 122 : static const std::vector<EntityInfoPtr> emptyVec;
883 122 : const auto iter = allGroupConfig_.find(groupId);
884 122 : if (iter != allGroupConfig_.end()) {
885 117 : return iter->second;
886 : }
887 5 : return emptyVec;
888 : }
889 :
890 255 : BqsStatus BindRelation::SubscribeEvent(const EntityInfo &subscribeEntity, const EventType eventType,
891 : const uint32_t index) const
892 : {
893 255 : if (subscribeEntity.GetType() == dgw::EntityType::ENTITY_QUEUE) {
894 210 : const uint32_t SubQueueType = subscribeEntity.GetQueueType();
895 : const auto subscribeManager =
896 210 : Subscribers::GetInstance().GetSubscribeManager(index, subscribeEntity.GetDeviceId());
897 210 : if (subscribeManager == nullptr) {
898 0 : DGW_LOG_ERROR("Failed to find subscribeManager for isHost:%d, resIndex: %u, device: %u",
899 : SubQueueType == bqs::LOCAL_Q, index, subscribeEntity.GetDeviceId());
900 0 : return BQS_STATUS_INNER_ERROR;
901 : }
902 210 : return (eventType == EventType::ENQUEUE) ?
903 89 : subscribeManager->Subscribe(subscribeEntity.GetId()) :
904 210 : subscribeManager->SubscribeFullToNotFull(subscribeEntity.GetId());
905 : }
906 45 : if (subscribeEntity.GetType() == dgw::EntityType::ENTITY_GROUP) {
907 19 : const auto entitiesInGroup = GetEntitiesInGroup(subscribeEntity.GetId());
908 54 : for (const auto &entity : entitiesInGroup) {
909 35 : if (entity->GetType() != dgw::EntityType::ENTITY_QUEUE) {
910 3 : continue;
911 : }
912 32 : const uint32_t queuType = entity->GetQueueType();
913 : const auto subscribeManager =
914 32 : Subscribers::GetInstance().GetSubscribeManager(index, entity->GetDeviceId());
915 32 : if (subscribeManager == nullptr) {
916 0 : DGW_LOG_ERROR("Failed to find subscribeManager for ishost: %d, resIndex: %u, device: %u",
917 : queuType == bqs::LOCAL_Q, index, entity->GetDeviceId());
918 0 : return BQS_STATUS_INNER_ERROR;
919 : }
920 32 : const auto ret = (eventType == EventType::ENQUEUE) ?
921 15 : subscribeManager->Subscribe(entity->GetId()) :
922 17 : subscribeManager->SubscribeFullToNotFull(entity->GetId());
923 32 : if (ret != BQS_STATUS_OK) {
924 0 : BQS_LOG_ERROR("Subscribe queue[%u] in group[%u] failed.", entity->GetId(), subscribeEntity.GetId());
925 0 : return ret;
926 : }
927 : }
928 19 : return BQS_STATUS_OK;
929 19 : }
930 26 : return BQS_STATUS_OK;
931 : }
932 :
933 242 : BqsStatus BindRelation::UnsubscribeEvent(const EntityInfo &subscribeEntity, const EventType eventType,
934 : const uint32_t index) const
935 : {
936 242 : if (subscribeEntity.GetType() == dgw::EntityType::ENTITY_QUEUE) {
937 199 : const uint32_t SubQueueType = subscribeEntity.GetQueueType();
938 : const auto subscribeManager =
939 199 : Subscribers::GetInstance().GetSubscribeManager(index, subscribeEntity.GetDeviceId());
940 199 : if (subscribeManager == nullptr) {
941 0 : DGW_LOG_ERROR("Failed to find subscribeManager for SubQueueType: %u resIndex: %u, device: %u",
942 : SubQueueType, index, subscribeEntity.GetDeviceId());
943 0 : return BQS_STATUS_INNER_ERROR;
944 : }
945 199 : return (eventType == EventType::ENQUEUE) ?
946 84 : subscribeManager->Unsubscribe(subscribeEntity.GetId()) :
947 199 : subscribeManager->UnsubscribeFullToNotFull(subscribeEntity.GetId());
948 : }
949 43 : auto result = BQS_STATUS_OK;
950 43 : if (subscribeEntity.GetType() == dgw::EntityType::ENTITY_GROUP) {
951 17 : const auto entitiesInGroup = GetEntitiesInGroup(subscribeEntity.GetId());
952 48 : for (const auto &entity : entitiesInGroup) {
953 31 : if (entity->GetType() != dgw::EntityType::ENTITY_QUEUE) {
954 3 : continue;
955 : }
956 : const auto subscribeManager =
957 28 : Subscribers::GetInstance().GetSubscribeManager(index, entity->GetDeviceId());
958 28 : if (subscribeManager == nullptr) {
959 0 : DGW_LOG_ERROR("Failed to find subscribeManager for resIndex: %u, device: %u",
960 : index, entity->GetDeviceId());
961 0 : return BQS_STATUS_INNER_ERROR;
962 : }
963 28 : const auto ret = (eventType == EventType::ENQUEUE) ?
964 13 : subscribeManager->Unsubscribe(entity->GetId()) :
965 15 : subscribeManager->UnsubscribeFullToNotFull(entity->GetId());
966 28 : if (ret != BQS_STATUS_OK) {
967 0 : result = ret;
968 0 : BQS_LOG_ERROR("Unsubscribe queue[%u] in group[%u] failed.", entity->GetId(), subscribeEntity.GetId());
969 : }
970 : }
971 17 : }
972 43 : return result;
973 : }
974 :
975 4 : BqsStatus BindRelation::UpdateSubscribeEvent(const EntityInfo &subscribeEntity, const EventType eventType,
976 : const uint32_t index) const
977 : {
978 4 : if (subscribeEntity.GetType() == dgw::EntityType::ENTITY_QUEUE) {
979 : const auto subscribeManager =
980 2 : Subscribers::GetInstance().GetSubscribeManager(index, subscribeEntity.GetDeviceId());
981 2 : if (subscribeManager == nullptr) {
982 0 : DGW_LOG_ERROR("Failed to find subscribeManager for resIndex: %u, device: %u",
983 : index, subscribeEntity.GetDeviceId());
984 0 : return BQS_STATUS_INNER_ERROR;
985 : }
986 2 : return (eventType == EventType::ENQUEUE) ?
987 1 : subscribeManager->UpdateSubscribe(subscribeEntity.GetId()) :
988 2 : subscribeManager->UpdateSubscribeFullToNotFull(subscribeEntity.GetId());
989 : }
990 2 : auto result = BQS_STATUS_OK;
991 2 : if (subscribeEntity.GetType() == dgw::EntityType::ENTITY_GROUP) {
992 2 : const auto entitiesInGroup = GetEntitiesInGroup(subscribeEntity.GetId());
993 6 : for (const auto &entity : entitiesInGroup) {
994 4 : if (entity->GetType() != dgw::EntityType::ENTITY_QUEUE) {
995 0 : continue;
996 : }
997 : const auto subscribeManager =
998 4 : Subscribers::GetInstance().GetSubscribeManager(index, entity->GetDeviceId());
999 4 : if (subscribeManager == nullptr) {
1000 0 : DGW_LOG_ERROR("Failed to find subscribeManager for resIndex: %u, device: %u",
1001 : index, entity->GetDeviceId());
1002 0 : return BQS_STATUS_INNER_ERROR;
1003 : }
1004 4 : const auto ret = (eventType == EventType::ENQUEUE) ?
1005 2 : subscribeManager->UpdateSubscribe(entity->GetId()) :
1006 2 : subscribeManager->UpdateSubscribeFullToNotFull(entity->GetId());
1007 4 : if (ret != BQS_STATUS_OK) {
1008 0 : result = ret;
1009 0 : BQS_LOG_ERROR("Subscribe queue[%u] in group[%u] failed.",
1010 : entity->GetId(), subscribeEntity.GetId());
1011 : }
1012 : }
1013 2 : }
1014 2 : return result;
1015 : }
1016 :
1017 278 : BqsStatus BindRelation::CheckUnBind(const EntityInfo &entity) const
1018 : {
1019 : (void)entity;
1020 278 : return BQS_STATUS_OK;
1021 : }
1022 :
1023 9 : void BindRelation::MarkAbnormalSrc(const EntityInfo &srcEntity)
1024 : {
1025 9 : const auto &iter = srcToDstRelation_.find(srcEntity);
1026 9 : if (iter == srcToDstRelation_.end()) {
1027 4 : BQS_LOG_WARN("No relation [%s->*] exists, no need mark", srcEntity.ToString().c_str());
1028 4 : return;
1029 : } else {
1030 5 : auto abnormalSrc = iter->first;
1031 5 : abnormalSrc.SetEntity(nullptr);
1032 :
1033 5 : auto &abnormalDstSet = iter->second;
1034 10 : for (auto abnormalDst : abnormalDstSet) {
1035 5 : abnormalDst.SetEntity(nullptr);
1036 5 : abnormalSrcToDst_[abnormalSrc].emplace(abnormalDst);
1037 5 : abnormalDstToSrc_[abnormalDst].emplace(abnormalSrc);
1038 5 : }
1039 5 : BQS_LOG_RUN_INFO("Mark abnormal relation [%s->*]", abnormalSrc.ToString().c_str());
1040 5 : }
1041 : }
1042 :
1043 4 : void BindRelation::MarkAbnormalDst(const EntityInfo &dstEntity)
1044 : {
1045 4 : const auto &iter = dstToSrcRelation_.find(dstEntity);
1046 4 : if (iter == dstToSrcRelation_.end()) {
1047 1 : BQS_LOG_WARN("No relation [*->%s] exists, no need mark", dstEntity.ToString().c_str());
1048 1 : return;
1049 : } else {
1050 3 : auto abnormalDst = iter->first;
1051 3 : abnormalDst.SetEntity(nullptr);
1052 :
1053 3 : const auto &abnormalSrcSet = iter->second;
1054 6 : for (auto abnormalSrc : abnormalSrcSet) {
1055 3 : abnormalSrc.SetEntity(nullptr);
1056 3 : abnormalDstToSrc_[abnormalDst].emplace(abnormalSrc);
1057 3 : abnormalSrcToDst_[abnormalSrc].emplace(abnormalDst);
1058 3 : }
1059 3 : BQS_LOG_RUN_INFO("Mark abnormal relation [*->%s]", abnormalDst.ToString().c_str());
1060 3 : }
1061 : }
1062 :
1063 139 : void BindRelation::DelAbnormalSrcToDst(const EntityInfo &srcEntity, const EntityInfo &dstEntity)
1064 : {
1065 139 : const auto iter = abnormalSrcToDst_.find(srcEntity);
1066 139 : if (iter == abnormalSrcToDst_.end()) {
1067 133 : BQS_LOG_WARN("Bind relation[%s->%s] dst doesn't exist in abnormal bind relations, no need unbind.",
1068 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
1069 133 : return;
1070 : }
1071 :
1072 6 : (void)iter->second.erase(dstEntity);
1073 6 : if (iter->second.empty()) {
1074 6 : (void)abnormalSrcToDst_.erase(iter);
1075 : }
1076 : }
1077 :
1078 139 : void BindRelation::DelAbnormalDstToSrc(const EntityInfo &srcEntity, const EntityInfo &dstEntity)
1079 : {
1080 139 : const auto iter = abnormalDstToSrc_.find(dstEntity);
1081 139 : if (iter == abnormalDstToSrc_.end()) {
1082 133 : BQS_LOG_WARN("Bind relation[%s->%s] dst doesn't exist in abnormal bind relations, no need unbind.",
1083 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
1084 133 : return;
1085 : }
1086 :
1087 6 : (void)iter->second.erase(srcEntity);
1088 6 : if (iter->second.empty()) {
1089 6 : (void)abnormalDstToSrc_.erase(iter);
1090 : }
1091 : }
1092 :
1093 5 : void BindRelation::UnBindAbnormalRelationBySrc(const EntityInfo &srcEntity)
1094 : {
1095 5 : const auto srcToDstIter = abnormalSrcToDst_.find(srcEntity);
1096 5 : if (srcToDstIter == abnormalSrcToDst_.end()) {
1097 5 : BQS_LOG_WARN("No relation [%s->*] exists in abnormal bind relation, no need unbind",
1098 : srcEntity.ToString().c_str());
1099 5 : return;
1100 : }
1101 :
1102 0 : for (const auto &dstEntity : srcToDstIter->second) {
1103 0 : DelAbnormalDstToSrc(srcEntity, dstEntity);
1104 : }
1105 :
1106 0 : (void)abnormalSrcToDst_.erase(srcToDstIter);
1107 : }
1108 :
1109 7 : void BindRelation::UnBindAbnormalRelationByDst(const EntityInfo &dstEntity)
1110 : {
1111 7 : const auto dstToSrcIter = abnormalDstToSrc_.find(dstEntity);
1112 7 : if (dstToSrcIter == abnormalDstToSrc_.end()) {
1113 7 : BQS_LOG_WARN("No bind relation[*->%s] exists in abnormal bind relations, no need unbind.",
1114 : dstEntity.ToString().c_str());
1115 7 : return;
1116 : }
1117 :
1118 0 : for (const auto &srcEntity : dstToSrcIter->second) {
1119 0 : DelAbnormalSrcToDst(srcEntity, dstEntity);
1120 : }
1121 :
1122 0 : (void)abnormalDstToSrc_.erase(dstToSrcIter);
1123 : }
1124 :
1125 2 : BqsStatus BindRelation::ClearInputQueue(const uint32_t index, const std::unordered_set<uint32_t>& keySet)
1126 : {
1127 2 : const auto &inputQueues = (index == 0U) ? orderedSubscribeQueueId_ : orderedSubscribeQueueIdExtra_;
1128 2 : for (const auto &info: inputQueues) {
1129 0 : if (keySet.count(info.GetSchedCfgKey()) == 0U) {
1130 0 : continue;
1131 : }
1132 0 : const auto entity = info.GetEntity();
1133 0 : if (entity == nullptr) {
1134 0 : BQS_LOG_ERROR("[%s] has no entity, this should not happen.", info.ToString().c_str());
1135 0 : return BQS_STATUS_INNER_ERROR;
1136 : }
1137 0 : const auto ret = entity->ClearQueue();
1138 0 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
1139 0 : return BQS_STATUS_INNER_ERROR;
1140 : }
1141 0 : }
1142 2 : return BQS_STATUS_OK;
1143 : }
1144 :
1145 2 : BqsStatus BindRelation::MakeSureOutputCompletion(const uint32_t index, const std::unordered_set<uint32_t>& keySet)
1146 : {
1147 2 : const auto &dstToSrcRelation = (index == 0U) ? dstToSrcRelation_ : dstToSrcRelationExtra_;
1148 2 : for (const auto &dstItem: dstToSrcRelation) {
1149 1 : auto &dst = dstItem.first;
1150 1 : if (keySet.count(dst.GetSchedCfgKey()) == 0U) {
1151 0 : continue;
1152 : }
1153 1 : const auto entity = dst.GetEntity();
1154 1 : if (entity == nullptr) {
1155 0 : BQS_LOG_ERROR("[%s] has no entity, this should not happen.", dst.ToString().c_str());
1156 0 : return BQS_STATUS_INNER_ERROR;
1157 : }
1158 1 : const auto ret = entity->MakeSureOutputCompletion();
1159 1 : if (ret != dgw::FsmStatus::FSM_SUCCESS) {
1160 1 : return BQS_STATUS_INNER_ERROR;
1161 : }
1162 1 : }
1163 1 : return BQS_STATUS_OK;
1164 : }
1165 :
1166 160 : BqsStatus BindRelation::GetBindRelationIndex(const EntityInfo &srcEntity, const EntityInfo &dstEntity,
1167 : uint32_t &index) const
1168 : {
1169 160 : if (!GlobalCfg::GetInstance().GetNumaFlag()) {
1170 150 : index = 0U;
1171 150 : return BQS_STATUS_OK;
1172 : }
1173 10 : const auto srcToDstIter = srcToDstRelation_.find(srcEntity);
1174 10 : const auto dstToSrcIter = dstToSrcRelation_.find(dstEntity);
1175 10 : const auto srcToDstIterExtra = srcToDstRelationExtra_.find(srcEntity);
1176 10 : const auto dstToSrcIterExtra = dstToSrcRelationExtra_.find(dstEntity);
1177 16 : if ((srcToDstIter != srcToDstRelation_.end() || dstToSrcIter != dstToSrcRelation_.end()) &&
1178 16 : (srcToDstIterExtra == srcToDstRelationExtra_.end() && dstToSrcIterExtra == dstToSrcRelationExtra_.end())) {
1179 3 : index = 0;
1180 3 : return BQS_STATUS_OK;
1181 : }
1182 :
1183 7 : if ((srcToDstIterExtra != srcToDstRelationExtra_.end() || dstToSrcIterExtra != dstToSrcRelationExtra_.end()) &&
1184 7 : (srcToDstIter == srcToDstRelation_.end() && dstToSrcIter == dstToSrcRelation_.end())) {
1185 0 : index = 1;
1186 0 : return BQS_STATUS_OK;
1187 : }
1188 :
1189 21 : if ((srcToDstIterExtra == srcToDstRelationExtra_.end() || dstToSrcIterExtra == dstToSrcRelationExtra_.end()) &&
1190 21 : (srcToDstIter == srcToDstRelation_.end() && dstToSrcIter == dstToSrcRelation_.end())) {
1191 7 : index = GlobalCfg::GetInstance().GetResIndexByDeviceId(srcEntity.GetDeviceId());
1192 7 : return BQS_STATUS_OK;
1193 : }
1194 :
1195 0 : return BQS_STATUS_PARAM_INVALID;
1196 : }
1197 :
1198 153 : BqsStatus BindRelation::GetBindIndexBySrc(const EntityInfo &srcEntity, uint32_t &index) const
1199 : {
1200 153 : if (!GlobalCfg::GetInstance().GetNumaFlag()) {
1201 142 : index = 0U;
1202 142 : return BQS_STATUS_OK;
1203 : }
1204 11 : const auto srcToDstIter = srcToDstRelation_.find(srcEntity);
1205 11 : if (srcToDstIter != srcToDstRelation_.end()) {
1206 10 : index = 0;
1207 10 : return BQS_STATUS_OK;
1208 : }
1209 :
1210 1 : const auto srcToDstIterExtra = srcToDstRelationExtra_.find(srcEntity);
1211 1 : if (srcToDstIterExtra != srcToDstRelationExtra_.end()) {
1212 1 : index = 1;
1213 1 : return BQS_STATUS_OK;
1214 : }
1215 :
1216 0 : return BQS_STATUS_PARAM_INVALID;
1217 : }
1218 :
1219 11 : BqsStatus BindRelation::GetBindIndexByDst(const EntityInfo &srcEntity, uint32_t &index) const
1220 : {
1221 11 : if (!GlobalCfg::GetInstance().GetNumaFlag()) {
1222 9 : index = 0U;
1223 9 : return BQS_STATUS_OK;
1224 : }
1225 2 : const auto dstToSrcIter = dstToSrcRelation_.find(srcEntity);
1226 2 : if (dstToSrcIter != dstToSrcRelation_.end()) {
1227 0 : index = 0;
1228 0 : return BQS_STATUS_OK;
1229 : }
1230 :
1231 2 : const auto dstToSrcIterExtra = dstToSrcRelationExtra_.find(srcEntity);
1232 2 : if (dstToSrcIterExtra != dstToSrcRelationExtra_.end()) {
1233 0 : index = 1;
1234 0 : return BQS_STATUS_OK;
1235 : }
1236 :
1237 2 : return BQS_STATUS_PARAM_INVALID;
1238 : }
1239 :
1240 12 : const MapEnitityInfoToInfoSet &BindRelation::GetSrcToDstExtraRelation() const
1241 : {
1242 12 : return srcToDstRelationExtra_;
1243 : }
1244 :
1245 7 : const MapEnitityInfoToInfoSet &BindRelation::GetDstToSrcExtraRelation() const
1246 : {
1247 7 : return dstToSrcRelationExtra_;
1248 : }
1249 :
1250 0 : const std::vector<EntityInfo> &BindRelation::GetOrderedSubscribeQueueIdExtra() const
1251 : {
1252 0 : return orderedSubscribeQueueIdExtra_;
1253 : }
1254 :
1255 13 : void BindRelation::AppendAbnormalEntity(const EntityInfo &info, const dgw::EntityDirection direction,
1256 : const uint32_t index) {
1257 13 : if (index == 0) {
1258 13 : if (direction == dgw::EntityDirection::DIRECTION_SEND) {
1259 9 : abnormalSrc_.emplace_back(info);
1260 : } else {
1261 4 : abnormalDst_.emplace_back(info);
1262 : }
1263 : }
1264 13 : }
1265 :
1266 28 : void BindRelation::ClearAbnormalEntityInfo(const uint32_t index) {
1267 28 : if (index == 0) {
1268 28 : abnormalSrc_.clear();
1269 28 : abnormalDst_.clear();
1270 : }
1271 28 : }
1272 :
1273 34 : void BindRelation::UpdateRelation(const uint32_t index)
1274 : {
1275 34 : if (index == 0) {
1276 34 : if (abnormalSrc_.empty() && abnormalDst_.empty()) {
1277 26 : return;
1278 : }
1279 :
1280 17 : for (const auto &abnormalSrc : abnormalSrc_) {
1281 9 : MarkAbnormalSrc(abnormalSrc);
1282 9 : UnBindRelationBySrc(abnormalSrc);
1283 : }
1284 :
1285 12 : for (const auto &abnormalDst : abnormalDst_) {
1286 4 : MarkAbnormalDst(abnormalDst);
1287 4 : UnBindRelationByDst(abnormalDst);
1288 : }
1289 :
1290 8 : Order(index);
1291 :
1292 8 : abnormalSrc_.clear();
1293 8 : abnormalDst_.clear();
1294 : }
1295 : }
1296 : } // namespace bqs
|
