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 : #ifndef QUEUE_SCHEDULE_BIND_RELATION_H
12 : #define QUEUE_SCHEDULE_BIND_RELATION_H
13 :
14 : #include <memory>
15 : #include <vector>
16 : #include <string>
17 : #include <unordered_map>
18 : #include <unordered_set>
19 : #include "hccl/hccl_types_in.h"
20 : #include "queue_schedule/dgw_client.h"
21 : #include "common/bqs_status.h"
22 : #include "common/bqs_util.h"
23 : #include "fsm/state_define.h"
24 : #include "hccl/comm_channel_manager.h"
25 : #include "data_obj_manager.h"
26 : #include "common/bqs_log.h"
27 :
28 : namespace bqs {
29 : enum class EventType : uint32_t {
30 : ENQUEUE,
31 : F2NF,
32 : };
33 :
34 : struct OptionalArg {
35 : const dgw::CommChannel* channelPtr = nullptr;
36 : dgw::EntityType eType = dgw::EntityType::ENTITY_QUEUE;
37 : bqs::GroupPolicy policy = bqs::GroupPolicy::HASH;
38 : uint32_t peerInstanceNum = 1U;
39 : uint32_t localInstanceIndex = 0U;
40 : uint32_t globalId = 0U;
41 : uint32_t uuId = 0U;
42 : uint32_t schedCfgKey = 0U;
43 : uint32_t queueType = bqs::LOCAL_Q;
44 : };
45 :
46 : class EntityInfo {
47 : public:
48 : EntityInfo() = delete;
49 3321 : ~EntityInfo() = default;
50 :
51 637 : explicit EntityInfo(const uint32_t id, const uint32_t deviceId, const OptionalArg *const args = nullptr)
52 637 : : id_(id), deviceId_(deviceId)
53 : {
54 637 : if (args != nullptr) {
55 261 : optionalArgs_ = *args;
56 : }
57 637 : (void)entityDesc_.append("qid:").append(std::to_string(id_))
58 1274 : .append(", type:").append(std::to_string(static_cast<int32_t>(optionalArgs_.eType)))
59 1274 : .append(", globalId:").append(std::to_string(optionalArgs_.globalId))
60 1274 : .append(", schedCfgKey:").append(std::to_string(optionalArgs_.schedCfgKey))
61 1274 : .append(", queue type:").append(std::to_string(optionalArgs_.queueType))
62 637 : .append(", deviceId:").append(std::to_string(deviceId_));
63 637 : if (optionalArgs_.channelPtr != nullptr) {
64 31 : (void)entityDesc_.append(", ").append(optionalArgs_.channelPtr->ToString());
65 : }
66 637 : }
67 :
68 1346 : bool operator==(const EntityInfo &entityInfo) const
69 : {
70 1346 : if (id_ != entityInfo.id_) {
71 198 : return false;
72 : }
73 1148 : if (optionalArgs_.eType != entityInfo.optionalArgs_.eType) {
74 1 : return false;
75 : }
76 1147 : if (optionalArgs_.queueType != entityInfo.optionalArgs_.queueType) {
77 0 : return false;
78 : }
79 1147 : if (deviceId_ != entityInfo.deviceId_) {
80 0 : return false;
81 : }
82 1147 : return true;
83 : }
84 :
85 5024 : inline uint32_t GetId() const
86 : {
87 5024 : return id_;
88 : }
89 6089 : inline dgw::EntityType GetType() const
90 : {
91 6089 : return optionalArgs_.eType;
92 : }
93 19 : inline bqs::GroupPolicy GetGroupPolicy() const
94 : {
95 19 : return optionalArgs_.policy;
96 : }
97 277 : inline const dgw::CommChannel *GetCommChannel() const
98 : {
99 277 : return optionalArgs_.channelPtr;
100 : }
101 :
102 725 : inline const dgw::EntityPtr &GetEntity() const
103 : {
104 725 : return entity_;
105 : }
106 :
107 274 : inline void SetEntity(const dgw::EntityPtr &entity)
108 : {
109 274 : entity_ = entity;
110 274 : }
111 :
112 1660 : inline const std::string &ToString() const
113 : {
114 1660 : return entityDesc_;
115 : }
116 :
117 19 : inline uint32_t GetPeerInstanceNum() const
118 : {
119 19 : return optionalArgs_.peerInstanceNum;
120 : }
121 :
122 19 : inline uint32_t GetLocalInstanceIndex() const
123 : {
124 19 : return optionalArgs_.localInstanceIndex;
125 : }
126 :
127 291 : inline uint32_t GetGlobalId() const
128 : {
129 291 : return optionalArgs_.globalId;
130 : }
131 :
132 291 : inline uint32_t GetUuId() const
133 : {
134 291 : return optionalArgs_.uuId;
135 : }
136 :
137 319 : inline uint32_t GetSchedCfgKey() const
138 : {
139 319 : return optionalArgs_.schedCfgKey;
140 : }
141 :
142 2223 : inline uint32_t GetDeviceId() const
143 : {
144 2223 : return deviceId_;
145 : }
146 :
147 2132 : inline uint32_t GetQueueType() const
148 : {
149 2132 : return optionalArgs_.queueType;
150 : }
151 :
152 : private:
153 : uint32_t id_;
154 : uint32_t deviceId_;
155 : OptionalArg optionalArgs_ = {};
156 : std::string entityDesc_;
157 : dgw::EntityPtr entity_ = nullptr;
158 : };
159 :
160 : class EntityInfoHash {
161 : public:
162 2525 : size_t operator()(const EntityInfo &info) const
163 : {
164 2525 : return std::hash<uint32_t>()(info.GetId()) ^ std::hash<uint16_t>()(static_cast<uint16_t>(info.GetType()));
165 : }
166 : };
167 :
168 : using EntityInfoPtr = std::shared_ptr<EntityInfo>;
169 : using EntityInfoSet = std::unordered_set<EntityInfo, EntityInfoHash>;
170 : using MapEnitityInfoToInfoSet = std::unordered_map<EntityInfo, EntityInfoSet, EntityInfoHash>;
171 :
172 : class BindRelation {
173 : public:
174 : static BindRelation &GetInstance();
175 :
176 20 : ~BindRelation() = default;
177 :
178 : BindRelation(const BindRelation &) = delete;
179 :
180 : BindRelation &operator=(const BindRelation &) = delete;
181 :
182 : BindRelation(BindRelation &&) = delete;
183 :
184 : BindRelation &operator=(BindRelation &&) = delete;
185 :
186 : /**
187 : * create relation srcEntity->dstEntity.
188 : * @param srcEntity src queue
189 : * @param dstEntity destination queue
190 : * @return BQS_STATUS_OK:success, other:failed
191 : */
192 : BqsStatus Bind(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t resIndex = 0U);
193 :
194 : /**
195 : * delete relation srcEntity->dstEntity.
196 : * @param srcEntity src queue id
197 : * @param dstEntity destination queue id
198 : * @return BQS_STATUS_OK:success, other:failed
199 : */
200 : BqsStatus UnBind(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t resIndex = 0U);
201 :
202 : /**
203 : * delete relation srcEntity->*.
204 : * @param srcEntity src queue id
205 : * @return BQS_STATUS_OK:success, other:failed
206 : */
207 : BqsStatus UnBindBySrc(const EntityInfo& srcEntity);
208 :
209 : /**
210 : * delete relation *->dstEntity.
211 : * @param dstEntity destination queue id
212 : * @return BQS_STATUS_OK:success, other:failed
213 : */
214 : BqsStatus UnBindByDst(const EntityInfo& dstEntity);
215 :
216 : /**
217 : * order subscribe queue id by topology.
218 : */
219 : void Order(const uint32_t index = 0U);
220 :
221 : /**
222 : * create group.
223 : * @param entities entity in group
224 : * @param groupId group id
225 : * @return BQS_STATUS_OK:success, other:failed
226 : */
227 : BqsStatus CreateGroup(const std::vector<EntityInfoPtr> &entities, uint32_t &groupId);
228 :
229 : /**
230 : * delete group.
231 : * @param groupId group id
232 : * @return BQS_STATUS_OK:success, other:failed
233 : */
234 : BqsStatus DeleteGroup(const uint32_t groupId);
235 :
236 : /**
237 : * get src to destination relation.
238 : * @return src to destination relation
239 : */
240 : const MapEnitityInfoToInfoSet &GetSrcToDstRelation() const;
241 :
242 : /* *
243 : * get destination to src relation.
244 : * @return destination to src relation
245 : */
246 : const MapEnitityInfoToInfoSet &GetDstToSrcRelation() const;
247 :
248 : /* *
249 : * Get ordered subscribe queue.
250 : * @return ordered subscribe queue id
251 : */
252 : const std::vector<EntityInfo> &GetOrderedSubscribeQueueId() const;
253 :
254 : /**
255 : * Get bind size.
256 : * @return size of all binds
257 : */
258 : uint32_t CountBinds() const;
259 :
260 : /**
261 : * Get abnormal bind size.
262 : * @return size of all binds
263 : */
264 : uint32_t CountAbnormalBinds() const;
265 :
266 : /**
267 : * Get group config.
268 : * @return endpoint entity info in current group
269 : */
270 : const std::vector<EntityInfoPtr> &GetEntitiesInGroup(const uint32_t groupId) const;
271 :
272 : BqsStatus UnBindRelationBySrc(const EntityInfo &srcEntity);
273 :
274 : BqsStatus UnBindRelationByDst(const EntityInfo &dstEntity);
275 :
276 : void MarkAbnormalSrc(const EntityInfo &srcEntity);
277 :
278 : void MarkAbnormalDst(const EntityInfo &dstEntity);
279 :
280 : const MapEnitityInfoToInfoSet &GetAbnormalSrcToDstRelation() const;
281 :
282 : const MapEnitityInfoToInfoSet &GetAbnormalDstToSrcRelation() const;
283 :
284 : BqsStatus GetBindRelationIndex(const EntityInfo &srcEntity, const EntityInfo &dstEntity, uint32_t &index) const;
285 :
286 : const MapEnitityInfoToInfoSet &GetSrcToDstExtraRelation() const;
287 :
288 : const MapEnitityInfoToInfoSet &GetDstToSrcExtraRelation() const;
289 :
290 : const std::vector<EntityInfo> &GetOrderedSubscribeQueueIdExtra() const;
291 :
292 : BqsStatus GetBindIndexBySrc(const EntityInfo &srcEntity, uint32_t &index) const;
293 :
294 : BqsStatus ClearInputQueue(const uint32_t index, const std::unordered_set<uint32_t>& keySet);
295 :
296 : BqsStatus MakeSureOutputCompletion(const uint32_t index, const std::unordered_set<uint32_t>& keySet);
297 :
298 : void ClearAbnormalEntityInfo(const uint32_t index);
299 :
300 : void AppendAbnormalEntity(const EntityInfo &info, const dgw::EntityDirection direction, const uint32_t index);
301 :
302 : void UpdateRelation(const uint32_t index);
303 :
304 : private:
305 20 : BindRelation() = default;
306 :
307 : /**
308 : * check whether can add relation srcEntity->dstEntity.
309 : * @param srcEntity src queue id
310 : * @param dstEntity destination queue id
311 : * @return BQS_STATUS_OK:success, other:failed
312 : */
313 : BqsStatus CheckBind(const EntityInfo& srcEntity, const EntityInfo& dstEntity,
314 : const uint32_t resIndex, uint32_t &index) const;
315 :
316 : /**
317 : * check whether exist multi-layer bind relation.
318 : * @param srcEntity src entity
319 : * @param dstEntity dst entity
320 : * @return BQS_STATUS_OK:success, other:exist
321 : */
322 : BqsStatus CheckMultiLayerBind(const EntityInfo& srcEntity, const EntityInfo& dstEntity, const uint32_t index) const;
323 :
324 : /**
325 : * check whether entity exist in group.
326 : * @param src src entity
327 : * @param dst dst entity
328 : * @return BQS_STATUS_OK:success, other:exist
329 : */
330 : BqsStatus CheckEntityExistInGroup(const EntityInfo& src, const EntityInfo& dst, const uint32_t resIndex = 0U) const;
331 :
332 : /**
333 : * del relation from srcToDstRelation_
334 : * @param srcEntity src queue id
335 : * @param dstEntity destination queue id
336 : * @return BQS_STATUS_OK:success, other:exist
337 : */
338 : BqsStatus DelSrcToDst(const EntityInfo& srcEntity, const EntityInfo& dstEntity, const uint32_t index = 0U);
339 :
340 : /**
341 : * add relation to srcToDstRelation_
342 : * @param srcEntity src queue id
343 : * @param dstEntity destination queue id
344 : * @return BQS_STATUS_OK:success, other:exist
345 : */
346 : BqsStatus AddSrcToDst(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t index = 0U);
347 :
348 : /**
349 : * del relation from dstToSrcRelation_
350 : * @param srcEntity src queue id
351 : * @param dstEntity destination queue id
352 : * @return BQS_STATUS_OK:success, other:exist
353 : */
354 : BqsStatus DelDstToSrc(const EntityInfo& srcEntity, const EntityInfo& dstEntity, const uint32_t index);
355 :
356 : /**
357 : * del relation to dstToSrcRelation_
358 : * @param srcEntity src queue id
359 : * @param dstEntity destination queue id
360 : * @return BQS_STATUS_OK:success, other:exist
361 : */
362 : BqsStatus AddDstToSrc(EntityInfo &srcEntity, EntityInfo &dstEntity, const uint32_t index = 0U);
363 :
364 : /**
365 : * generate group id
366 : * @return group id
367 : */
368 : uint32_t GenerateGroupId();
369 :
370 : /**
371 : * Create entity
372 : * @param src src entity info
373 : * @param dst dst entity info
374 : * @return BQS_STATUS_OK:success, other:exist
375 : */
376 : BqsStatus CreateEntity(const EntityInfo &src, const EntityInfo &dst, const uint32_t resIndex = 0U);
377 :
378 : /**
379 : * Create entity
380 : * @param info entity info
381 : * @param isSrc is src
382 : * @return BQS_STATUS_OK:success, other:exist
383 : */
384 : BqsStatus CreateEntity(const EntityInfo &info, const bool isSrc, const uint32_t resIndex);
385 :
386 : /**
387 : * create entity for group
388 : * @param groupEntity group entity
389 : * @param isSrc is src
390 : * @return BQS_STATUS_OK:success, other:exist
391 : */
392 : BqsStatus CreateEntityForGroup(const EntityInfo &groupEntity, const bool isSrc, const uint32_t resIndex);
393 :
394 : /**
395 : * Delete entity
396 : * @param info entity info
397 : * @return BQS_STATUS_OK:success, other:exist
398 : */
399 : BqsStatus DeleteEntity(const EntityInfo &info, const bool isSrc, const uint32_t resIndex = 0U) const;
400 :
401 : /**
402 : * delete entity for group
403 : * @param groupId group id
404 : * @return BQS_STATUS_OK:success, other:exist
405 : */
406 : BqsStatus DeleteEntityForGroup(const uint32_t queueType, const uint32_t deviceId,
407 : const uint32_t groupId, const dgw::EntityDirection direction, const uint32_t resIndex) const;
408 :
409 : /**
410 : * subscribe event
411 : * @param subscribeEntity entity info
412 : * @param eventType event type
413 : * @return BQS_STATUS_OK: success, other: failed
414 : */
415 : BqsStatus SubscribeEvent(const EntityInfo &subscribeEntity, const EventType eventType,
416 : const uint32_t index) const;
417 :
418 : /**
419 : * unsubscribe event
420 : * @param subscribeEntity entity info
421 : * @param eventType event type
422 : * @return BQS_STATUS_OK: success, other: failed
423 : */
424 : BqsStatus UnsubscribeEvent(const EntityInfo &subscribeEntity, const EventType eventType,
425 : const uint32_t index) const;
426 :
427 : /**
428 : * update subscribe event
429 : * @param subscribeEntity entity info
430 : * @param eventType event type
431 : * @return BQS_STATUS_OK: success, other: failed
432 : */
433 : BqsStatus UpdateSubscribeEvent(const EntityInfo &subscribeEntity, const EventType eventType,
434 : const uint32_t index = 0U) const;
435 :
436 : /**
437 : * check whether can execute unbind for entity
438 : * @param entity entity info
439 : * @return BQS_STATUS_OK: success, other: failed
440 : */
441 : BqsStatus CheckUnBind(const EntityInfo &entity) const;
442 :
443 : /**
444 : * set entityptr for entityInfo
445 : * @param entitiInfo to expand
446 : * @return BQS_STATUS_OK: success, other: failed
447 : */
448 : BqsStatus SetEntityPtr(EntityInfo &entityInfo, const dgw::EntityDirection direction, const uint32_t index) const;
449 :
450 : void UnBindAbnormalRelationBySrc(const EntityInfo &srcEntity);
451 :
452 : void UnBindAbnormalRelationByDst(const EntityInfo &dstEntity);
453 :
454 : void DelAbnormalSrcToDst(const EntityInfo &srcEntity, const EntityInfo &dstEntity);
455 :
456 : void DelAbnormalDstToSrc(const EntityInfo &srcEntity, const EntityInfo &dstEntity);
457 :
458 : BqsStatus GetBindIndexByDst(const EntityInfo &srcEntity, uint32_t &index) const;
459 :
460 : void OrderOneTable(std::vector<EntityInfo> &orderedSubscribeQueueId,
461 : const MapEnitityInfoToInfoSet &srcToDstRelation,
462 : const MapEnitityInfoToInfoSet &dstToSrcRelation);
463 :
464 : // a queue can be bound to multi queue.
465 : MapEnitityInfoToInfoSet srcToDstRelation_;
466 :
467 : // a queue can bind to only one queue
468 : MapEnitityInfoToInfoSet dstToSrcRelation_;
469 :
470 : // abnormal route src->dst.
471 : MapEnitityInfoToInfoSet abnormalSrcToDst_;
472 :
473 : // abnormal route dst->src
474 : MapEnitityInfoToInfoSet abnormalDstToSrc_;
475 :
476 : // subscribed queue ordered by topology
477 : std::vector<EntityInfo> orderedSubscribeQueueId_;
478 :
479 : // group config key: group id; value: endpoint entify in group
480 : std::unordered_map<uint32_t, std::vector<EntityInfoPtr>> allGroupConfig_;
481 : std::unordered_map<uint32_t, std::pair<uint32_t, uint32_t>> group2ResIndex_;
482 :
483 : // whether has loop in relation
484 : bool isHasLoop_ = false;
485 :
486 : // spin lock for generate group id
487 : SpinLock lockForGroup_;
488 :
489 : // a queue can be bound to multi queue.
490 : MapEnitityInfoToInfoSet srcToDstRelationExtra_;
491 :
492 : // a queue can bind to only one queue
493 : MapEnitityInfoToInfoSet dstToSrcRelationExtra_;
494 :
495 : std::vector<EntityInfo> orderedSubscribeQueueIdExtra_;
496 :
497 : std::vector<EntityInfo> abnormalSrc_;
498 : std::vector<EntityInfo> abnormalDst_;
499 : };
500 : } // namespace bqs
501 : #endif // QUEUE_SCHEDULE_BIND_RELATION_H
|
