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