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 DGW_ENTITY_H
12 : #define DGW_ENTITY_H
13 :
14 : #include <list>
15 : #include <vector>
16 : #include "data_obj_manager.h"
17 : #include "dynamic_sched_mgr.hpp"
18 : #include "fsm/state_define.h"
19 : #include "statistic_manager.h"
20 :
21 : #include "dgw_client.h"
22 : #include "hccl/comm_channel_manager.h"
23 : #include "bqs_util.h"
24 : namespace dgw {
25 :
26 : // invalid group id
27 : constexpr int32_t INVALID_GROUP_ID = -1;
28 : struct EntityMaterial {
29 : EntityType eType;
30 : EntityDirection direction;
31 : uint32_t id;
32 : uint32_t globalId;
33 : uint32_t uuId;
34 : uint32_t schedCfgKey;
35 : uint32_t resId; // deviceId
36 : int32_t hostGroupId = INVALID_GROUP_ID;
37 : const CommChannel* channel = nullptr;
38 : bqs::GroupPolicy groupPolicy = bqs::GroupPolicy::HASH;
39 : uint32_t peerInstanceNum = 1U;
40 : uint32_t localInstanceIndex = 0U;
41 : uint32_t queueType = bqs::LOCAL_Q;
42 : };
43 :
44 : using DynamicRequestPtr = std::shared_ptr<DynamicSchedMgr::RequestInfo>;
45 :
46 : constexpr uint32_t MBUF_HEAD_MAX_SIZE = 256U;
47 :
48 : class Entity {
49 : public:
50 : explicit Entity(const EntityMaterial &material, const uint32_t resIndex);
51 366 : virtual ~Entity() = default;
52 : Entity(const Entity &) = delete;
53 : Entity(const Entity &&) = delete;
54 : Entity &operator = (const Entity &) = delete;
55 : Entity &operator = (Entity &&) = delete;
56 :
57 : virtual FsmStatus Dequeue() = 0;
58 : virtual void SelectDstEntities(const uint64_t key, std::vector<Entity*> &toPushDstEntities,
59 : std::vector<Entity*> &reprocessDstEntities, std::vector<Entity*> &abnormalDstEntities) = 0;
60 : virtual FsmStatus ClearQueue() = 0;
61 : virtual FsmStatus PauseSubscribe(const Entity &fullEntity) = 0;
62 : virtual FsmStatus ResumeSubscribe(const Entity ¬FullEntity) = 0;
63 :
64 : virtual FsmStatus Init(const FsmState state, const EntityDirection direction);
65 : virtual FsmStatus Uninit();
66 : virtual uint32_t GetQueueId() const;
67 : virtual FsmStatus ResetSrcState();
68 : virtual void ResetSrcSubState();
69 : virtual void ReprocessInTryPush(const Entity &srcEntity, DynamicRequestPtr &dynamicRequest, uint32_t &schedCfgKey);
70 : virtual FsmStatus AbProcessInTryPush();
71 : virtual FsmStatus SendData(const DataObjPtr dataObj);
72 : virtual bool IsDataPeeked() const;
73 : virtual FsmStatus MakeSureOutputCompletion();
74 : virtual uint32_t GetMbufDeviceId() const;
75 : virtual uint32_t GetMbufQueueType() const;
76 :
77 : FsmStatus AllowDeque();
78 : FsmStatus ProcessMessage(const InnerMessage &msg);
79 : FsmStatus ChangeState(const FsmState nextState);
80 : FsmStatus RemoveDataObjFromSendList(const DataObjPtr &dataObj);
81 : void RemoveRecvEntityFromSendList(const Entity* const recvEntityPtr);
82 : FsmStatus AddDataObjToSendList(const DataObjPtr &dataObj);
83 : FsmStatus AddDataObjToRecvList(const DataObjPtr &dataObj);
84 : const std::string &GetTypeDesc() const;
85 : const std::string &GetStateDesc(const FsmState id) const;
86 : bool Equal(const Entity* const recvEntityPtr) const;
87 : bool UpdateSendObject(const EntityPtr group, const EntityPtr elem);
88 :
89 15 : inline void AddScheduleCount()
90 : {
91 15 : scheduleCount_++;
92 15 : }
93 50 : inline void ResetScheduleCount()
94 : {
95 50 : scheduleCount_ = 0U;
96 50 : }
97 24 : inline uint32_t GetScheduleCount() const
98 : {
99 24 : return scheduleCount_;
100 : }
101 56 : inline EntityType GetType() const
102 : {
103 56 : return type_;
104 : }
105 440 : inline uint32_t GetId() const
106 : {
107 440 : return id_;
108 : }
109 26 : inline Mbuf *GetMbuf() const
110 : {
111 26 : return mbuf_;
112 : }
113 2 : inline void SetMbuf(Mbuf * const mbuf)
114 : {
115 2 : mbuf_ = mbuf;
116 2 : }
117 793 : inline uint32_t GetDeviceId() const
118 : {
119 793 : return deviceId_;
120 : }
121 61 : inline DataObjList &GetRecvDataObjs()
122 : {
123 61 : return recvDataObjs_;
124 : }
125 34 : inline const DataObjList &GetSendDataObjs() const
126 : {
127 34 : return sendDataObjs_;
128 : }
129 279 : inline uint32_t GetRefCount() const
130 : {
131 279 : return refCount_;
132 : }
133 279 : inline void DecreaseRefCount()
134 : {
135 279 : refCount_--;
136 279 : }
137 294 : inline void IncreaseRefCount()
138 : {
139 294 : refCount_++;
140 294 : }
141 1802 : inline EntityDirection GetDirection() const
142 : {
143 1802 : return direction_;
144 : }
145 944 : inline FsmState GetCurState() const
146 : {
147 944 : return curState_;
148 : }
149 51 : inline uint64_t GetTransId() const
150 : {
151 51 : return transId_;
152 : }
153 4 : inline void SetTransId(const uint64_t transId)
154 : {
155 4 : transId_ = transId;
156 4 : }
157 42 : inline uint32_t GetRouteLabel() const
158 : {
159 42 : return routeLabel_;
160 : }
161 4 : inline void SetRouteLabel(const uint32_t routeLabel)
162 : {
163 4 : routeLabel_ = routeLabel;
164 4 : }
165 :
166 57 : inline uint32_t GetQueueType() const
167 : {
168 57 : return queueType_;
169 : }
170 :
171 80 : inline int32_t GetHostGroupId() const
172 : {
173 80 : return hostGroupId_;
174 : }
175 49 : inline void SetNeedTransId(const bool needTransId)
176 : {
177 49 : needTransId_ = needTransId;
178 49 : }
179 160 : inline bool IsNeedTransId() const
180 : {
181 160 : return needTransId_;
182 : }
183 2 : inline void SetMessageType(const InnerMsgType msgType)
184 : {
185 2 : msgType_ = msgType;
186 2 : }
187 10 : inline InnerMsgType GetMessageType() const
188 : {
189 10 : return msgType_;
190 : }
191 :
192 1629 : inline const std::string &ToString() const
193 : {
194 1629 : return entityDesc_;
195 : }
196 : inline bqs::EntityStatisticInfo &GetStatisticInfo()
197 : {
198 : return statInfo_;
199 : }
200 3 : inline void SetWaitDecisionState(const bool waitState)
201 : {
202 3 : waitingDecision_ = waitState;
203 3 : }
204 51 : inline bool GetWaitDecisionState() const
205 : {
206 51 : return waitingDecision_;
207 : }
208 289 : inline uint32_t GetGlobalId() const
209 : {
210 289 : return globalId_;
211 : }
212 :
213 3 : inline uint32_t GetUuId() const
214 : {
215 3 : return uuId_;
216 : }
217 :
218 286 : inline uint32_t GetSchedCfgKey() const
219 : {
220 286 : return schedCfgKey_;
221 : }
222 :
223 114 : inline uint32_t GetResIndex() const
224 : {
225 114 : return resIndex_;
226 : }
227 :
228 2 : inline void SetDynamicReqTime(const uint64_t dynamicReqTime)
229 : {
230 2 : dynamicReqTime_ = dynamicReqTime;
231 2 : }
232 :
233 1 : inline uint64_t GetDynamicReqTime() const
234 : {
235 1 : return dynamicReqTime_;
236 : }
237 :
238 : protected:
239 : EntityType type_;
240 : uint32_t id_;
241 : uint32_t deviceId_;
242 : // the group which this entity belong to
243 : int32_t hostGroupId_;
244 : uint32_t globalId_;
245 : uint32_t uuId_;
246 : uint32_t schedCfgKey_;
247 : uint32_t resIndex_;
248 : uint32_t queueType_;
249 : SubscribeStatus subscribeStatus_;
250 : uint32_t scheduleCount_;
251 : FsmState curState_;
252 : Mbuf *mbuf_;
253 : uint64_t transId_;
254 : uint32_t refCount_;
255 : EntityDirection direction_;
256 : // whether src entity need get transId from mbuf head
257 : bool needTransId_;
258 : InnerMsgType msgType_;
259 : uint32_t routeLabel_;
260 : bool waitingDecision_;
261 : uint64_t dynamicReqTime_;
262 : // entity desc
263 : std::string entityDesc_;
264 : // statistic info
265 : bqs::EntityStatisticInfo statInfo_;
266 : DataObjList recvDataObjs_;
267 : DataObjList sendDataObjs_;
268 : };
269 : }
270 : #endif
|
