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