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 "group_entity.h"
12 : #include "bqs_util.h"
13 : #include "common/bqs_log.h"
14 : #include "entity_manager.h"
15 : #include "strategy/strategy_manager.h"
16 :
17 : namespace dgw {
18 :
19 : namespace {
20 : constexpr int64_t GROUP_WAIT_TRANSID_TIMEOUT = -1L;
21 : }
22 :
23 27 : GroupEntity::GroupEntity(const EntityMaterial &material, const uint32_t resIndex)
24 27 : : Entity(material, resIndex), mbufDeviceId_(material.resId), mbufQueueType_(material.queueType)
25 : {
26 27 : groupInfo_.groupId = static_cast<int32_t>(material.id);
27 27 : groupInfo_.groupPolicy = material.groupPolicy;
28 27 : groupInfo_.timeout = GROUP_WAIT_TRANSID_TIMEOUT;
29 27 : groupInfo_.lastTransId = 0UL;
30 27 : groupInfo_.peerInstanceNum = material.peerInstanceNum;
31 27 : groupInfo_.localInstanceIndex = material.localInstanceIndex;
32 27 : groupInfo_.lastTimestamp = 0UL;
33 27 : }
34 :
35 10 : FsmStatus GroupEntity::Dequeue()
36 : {
37 10 : const auto dequeAllow = AllowDeque();
38 10 : if (dequeAllow != FsmStatus::FSM_SUCCESS) {
39 2 : return dequeAllow;
40 : }
41 :
42 : // select src entity in group
43 8 : FsmStatus selectStatus = FsmStatus::FSM_SUCCESS;
44 8 : const EntityPtr srcEntityPtr = SelectSrcEntity(selectStatus);
45 8 : if (selectStatus == FsmStatus::FSM_ERROR) {
46 1 : return FsmStatus::FSM_ERROR_PENDING;
47 : }
48 : // if select failed, group entity change to idle state
49 7 : if (srcEntityPtr == nullptr) {
50 6 : return FsmStatus::FSM_FAILED;
51 : }
52 :
53 : // select success
54 1 : DGW_LOG_INFO("Set srcEntity[%u] entity[%u] transId to %lu, routelabel to %u",
55 : srcEntityPtr->GetId(), id_, srcEntityPtr->GetTransId(), srcEntityPtr->GetRouteLabel());
56 1 : transId_ = srcEntityPtr->GetTransId();
57 1 : routeLabel_ = srcEntityPtr->GetRouteLabel();
58 1 : mbufDeviceId_ = srcEntityPtr->GetMbufDeviceId();
59 1 : mbufQueueType_ = srcEntityPtr->GetMbufQueueType();
60 1 : mbuf_ = srcEntityPtr->GetMbuf();
61 1 : srcEntityPtr->ResetSrcState();
62 : // set last trans id and last timestamp
63 1 : SetGroupInfo(GetTransId(), bqs::GetNowTime());
64 1 : AddScheduleCount();
65 1 : return FsmStatus::FSM_SUCCESS;
66 8 : }
67 :
68 13 : EntityPtr GroupEntity::SelectSrcEntity(FsmStatus &status)
69 : {
70 13 : const std::vector<EntityPtr> &entitiesInGroup = EntityManager::Instance(resIndex_).GetEntitiesInGroup(id_);
71 13 : if (entitiesInGroup.empty()) {
72 2 : DGW_LOG_ERROR("No entities in group, group id:%u", id_);
73 2 : return nullptr;
74 : }
75 :
76 : uint64_t waitTransId;
77 : // first schedule, set lastTimestamp
78 11 : if (groupInfo_.lastTransId == 0UL) {
79 9 : SetGroupInfo(0UL, bqs::GetNowTime());
80 18 : waitTransId = (groupInfo_.localInstanceIndex == 0U) ?
81 9 : groupInfo_.peerInstanceNum : groupInfo_.localInstanceIndex;
82 : } else {
83 2 : waitTransId = groupInfo_.lastTransId + groupInfo_.peerInstanceNum;
84 : }
85 :
86 : // mark whether all entity in group is peeked state
87 11 : bool allFinishPeekFlag = true;
88 : // mark whether no entity in group is peeked state
89 11 : bool noOneFinishPeekFlag = true;
90 : // peek every entity in group
91 21 : for (auto &entity : entitiesInGroup) {
92 14 : if (entity == nullptr) {
93 1 : DGW_LOG_ERROR("Entity is nullptr.");
94 4 : return nullptr;
95 : }
96 13 : DGW_LOG_INFO("[FSM] Begin to peek mbuf from entity[%s] in group:%u, waitTransId:%lu.",
97 : entity->ToString().c_str(), id_, waitTransId);
98 :
99 : // peek data from entity in group
100 13 : status = PeekFromEntityInGroup(*entity, waitTransId);
101 13 : if (status != FsmStatus::FSM_SUCCESS) {
102 10 : if (status == FsmStatus::FSM_ERROR) {
103 1 : DGW_LOG_ERROR("[FSM] Peek mbuf from entity[%s] failed.", entity->ToString().c_str());
104 1 : return nullptr;
105 : }
106 9 : allFinishPeekFlag = false;
107 9 : continue;
108 : }
109 3 : noOneFinishPeekFlag = false;
110 3 : if (Match(*entity, waitTransId, true)) {
111 2 : DGW_LOG_DEBUG("[FSM] Entity:[%s] state:[%s] transId:[%lu] has been selected from group entity:[%s].",
112 : entity->ToString().c_str(), entity->GetStateDesc(FsmState::FSM_PEEK_STATE).c_str(),
113 : waitTransId, ToString().c_str());
114 2 : return entity;
115 : }
116 : }
117 :
118 : // only when part of entities in group are peeked state, check timeout
119 7 : bool timeoutFlag = false;
120 7 : if ((!allFinishPeekFlag) && (!noOneFinishPeekFlag)) {
121 1 : timeoutFlag = CheckTimeout(waitTransId);
122 : }
123 : // if timeout or all entity in group finished peek, try to find min transId
124 7 : if (allFinishPeekFlag || timeoutFlag) {
125 1 : return SelectEntityWithMinTransId(entitiesInGroup);
126 : }
127 : // if not timeout && not all entity in group is peeked state && not exist waitTransId, wait next schedule
128 6 : return nullptr;
129 : }
130 :
131 7 : bool GroupEntity::Match(const Entity &entity, const uint64_t waitTransId, bool exactlyMatch) const {
132 7 : if ((entity.GetRouteLabel() != 0U) || (groupInfo_.groupPolicy == bqs::GroupPolicy::DYNAMIC)) {
133 2 : return true;
134 : }
135 5 : DGW_LOG_INFO("entity transid[%lu] vs waitTransId[%lu]", entity.GetTransId(), waitTransId);
136 5 : return exactlyMatch ? (entity.GetTransId() == waitTransId) : (entity.GetTransId() >= waitTransId);
137 : }
138 :
139 13 : FsmStatus GroupEntity::PeekFromEntityInGroup(Entity &entity, const uint64_t waitTransId) const
140 : {
141 13 : InnerMessage msg;
142 13 : msg.msgType = InnerMsgType::INNER_MSG_PUSH;
143 : do {
144 14 : if (entity.IsDataPeeked()) {
145 : // if match, return
146 4 : if (Match(entity, waitTransId, false)) {
147 3 : return FsmStatus::FSM_SUCCESS;
148 : };
149 : // not match, free mbuf
150 1 : Mbuf * const mbuf = entity.GetMbuf();
151 1 : if (mbuf != nullptr) {
152 1 : (void)halMbufFree(mbuf);
153 1 : entity.SetMbuf(nullptr);
154 1 : DGW_LOG_RUN_INFO("[FSM] Peek mbuf with transId[%lu] from entity[%s], "
155 : "but wait transId is [%lu], free mbuf!",
156 : entity.GetTransId(), entity.ToString().c_str(), waitTransId);
157 : }
158 : }
159 :
160 11 : DGW_LOG_DEBUG("[FSM] Entity[%s] try to peek data.", entity.ToString().c_str());
161 11 : if (entity.ProcessMessage(msg) == FsmStatus::FSM_ERROR) {
162 1 : return FsmStatus::FSM_ERROR;
163 : }
164 10 : } while (entity.IsDataPeeked());
165 :
166 9 : return FsmStatus::FSM_FAILED;
167 : }
168 :
169 5 : bool GroupEntity::CheckTimeout(const uint64_t waitTransId) const
170 : {
171 : (void)waitTransId;
172 5 : if (groupInfo_.timeout <= 0L) {
173 2 : DGW_LOG_INFO("[FSM] no need check timeout, timeoutInterval:%ld, waitTransId:%lu.",
174 : groupInfo_.timeout, waitTransId);
175 2 : return false;
176 : }
177 3 : const uint64_t currTimestamp = bqs::GetNowTime();
178 3 : if ((currTimestamp - groupInfo_.lastTimestamp) > static_cast<uint64_t>(groupInfo_.timeout)) {
179 2 : DGW_LOG_INFO("[FSM] timeout, currTimestamp:%lu, lasttimestamp:%lu, timeoutInterval:%ld, waitTransId:%lu.",
180 : currTimestamp, groupInfo_.lastTimestamp, groupInfo_.timeout, waitTransId);
181 2 : return true;
182 : }
183 1 : return false;
184 : }
185 :
186 2 : EntityPtr GroupEntity::SelectEntityWithMinTransId(const std::vector<EntityPtr> &entities) const
187 : {
188 2 : EntityPtr selectedEntity = nullptr;
189 2 : uint64_t minTransId = UINT64_MAX;
190 6 : for (auto &entity : entities) {
191 4 : if ((entity == nullptr) || !entity->IsDataPeeked()) {
192 2 : continue;
193 : }
194 2 : if (minTransId >= entity->GetTransId()) {
195 2 : minTransId = entity->GetTransId();
196 2 : selectedEntity = entity;
197 : }
198 : }
199 2 : if (selectedEntity != nullptr) {
200 2 : DGW_LOG_RUN_INFO("[FSM] Group entity[%s] get minTransId[%lu].", selectedEntity->ToString().c_str(), minTransId);
201 : }
202 2 : return selectedEntity;
203 0 : }
204 :
205 8 : void GroupEntity::SelectDstEntities(const uint64_t key, std::vector<Entity*> &toPushDstEntities,
206 : std::vector<Entity*> &reprocessDstEntities, std::vector<Entity*> &abnormalDstEntities)
207 : {
208 8 : if (groupInfo_.groupPolicy == bqs::GroupPolicy::DYNAMIC) {
209 3 : reprocessDstEntities.emplace_back(this);
210 4 : return;
211 : }
212 : // Group entity, use strategy to select dst entities
213 5 : Strategy *const strategy = StrategyManager::GetInstance().GetStrategy(groupInfo_.groupPolicy);
214 5 : if (strategy == nullptr) {
215 1 : DGW_LOG_ERROR("Strategy in group:%u with policy:%d is null. Please check!", id_,
216 : static_cast<int32_t>(groupInfo_.groupPolicy));
217 1 : return;
218 : }
219 4 : DGW_LOG_INFO("Get strategy:%s success.",
220 : StrategyManager::GetInstance().GetStrategyDesc(groupInfo_.groupPolicy).c_str());
221 4 : std::vector<EntityPtr> selEntities;
222 4 : (void) strategy->Search(id_, key, selEntities, resIndex_);
223 7 : for (auto selEntity : selEntities) {
224 4 : if (selEntity->GetCurState() == FsmState::FSM_ERROR_STATE) {
225 1 : abnormalDstEntities.emplace_back(this);
226 1 : break;
227 : }
228 3 : toPushDstEntities.emplace_back(selEntity.get());
229 4 : }
230 4 : }
231 :
232 4 : void GroupEntity::ReprocessInTryPush(const Entity &srcEntity, DynamicRequestPtr &dynamicRequest, uint32_t &schedCfgKey)
233 : {
234 4 : if (groupInfo_.groupPolicy != bqs::GroupPolicy::DYNAMIC) {
235 1 : return;
236 : }
237 :
238 3 : if (dynamicRequest == nullptr) {
239 3 : dynamicRequest = std::make_shared<DynamicSchedMgr::RequestInfo>();
240 3 : DGW_CHECK_RET_VOID((dynamicRequest != nullptr), "Fail to alloc dynamicRequest in group:%u.", id_);
241 3 : dynamicRequest->src.queueLogicId = srcEntity.GetGlobalId();
242 3 : dynamicRequest->src.modelUuid = srcEntity.GetUuId();
243 3 : dynamicRequest->src.queueId = srcEntity.GetId();
244 3 : DynamicSchedMgr::DecisionInfo decision = {};
245 3 : decision.transId = srcEntity.GetTransId();
246 3 : decision.routeLabel = srcEntity.GetRouteLabel();
247 3 : dynamicRequest->decisions.emplace_back(decision);
248 : }
249 3 : DynamicSchedMgr::DstGroupInfo dstGrp = {};
250 3 : dstGrp.logicGroupId = GetGlobalId();
251 3 : dynamicRequest->dsts.emplace_back(dstGrp);
252 3 : schedCfgKey = GetSchedCfgKey();
253 : }
254 :
255 1 : FsmStatus GroupEntity::AbProcessInTryPush()
256 : {
257 1 : return ChangeState(FsmState::FSM_ERROR_STATE);
258 : }
259 :
260 1 : FsmStatus GroupEntity::PauseSubscribe(const Entity &fullEntity)
261 : {
262 1 : const auto entities = EntityManager::Instance(resIndex_).GetEntitiesInGroup(id_);
263 3 : for (auto &entity : entities) {
264 2 : entity->PauseSubscribe(fullEntity);
265 : }
266 1 : return FsmStatus::FSM_SUCCESS;
267 1 : }
268 :
269 1 : FsmStatus GroupEntity::ResumeSubscribe(const Entity ¬FullEntity)
270 : {
271 1 : const auto entities = EntityManager::Instance(resIndex_).GetEntitiesInGroup(id_);
272 3 : for (auto &entity : entities) {
273 2 : (void) entity->ResumeSubscribe(notFullEntity);
274 : }
275 1 : return FsmStatus::FSM_SUCCESS;
276 1 : }
277 :
278 2 : FsmStatus GroupEntity::ClearQueue()
279 : {
280 2 : DGW_LOG_INFO("Entity[%s] clear queue", entityDesc_.c_str());
281 2 : const auto entities = EntityManager::Instance(resIndex_).GetEntitiesInGroup(id_);
282 4 : for (auto &entity : entities) {
283 3 : const auto ret = entity->ClearQueue();
284 3 : if (ret != FsmStatus::FSM_SUCCESS) {
285 1 : return ret;
286 : }
287 : }
288 1 : groupInfo_.lastTransId = 0U;
289 1 : return FsmStatus::FSM_SUCCESS;
290 2 : }
291 :
292 1 : FsmStatus GroupEntity::MakeSureOutputCompletion()
293 : {
294 1 : DGW_LOG_INFO("Entity[%s] MakeSureOutputCompletion", entityDesc_.c_str());
295 1 : FsmStatus ret = FsmStatus::FSM_SUCCESS;
296 1 : const auto entities = EntityManager::Instance(resIndex_).GetEntitiesInGroup(id_);
297 2 : for (auto &entity : entities) {
298 2 : ret = entity->MakeSureOutputCompletion();
299 2 : if (ret != FsmStatus::FSM_SUCCESS) {
300 1 : break;
301 : }
302 : }
303 1 : DGW_LOG_INFO("Entity[%s] Finish MakeSureOutputCompletion, ret is %d", entityDesc_.c_str(),
304 : static_cast<int32_t>(ret));
305 1 : return ret;
306 1 : }
307 :
308 10 : void GroupEntity::SetGroupInfo(const uint64_t lastTransId, const uint64_t lastTimestamp)
309 : {
310 10 : groupInfo_.lastTransId = lastTransId;
311 10 : groupInfo_.lastTimestamp = lastTimestamp;
312 10 : }
313 :
314 2 : uint32_t GroupEntity::GetMbufDeviceId() const
315 : {
316 2 : return mbufDeviceId_;
317 : }
318 :
319 2 : uint32_t GroupEntity::GetMbufQueueType() const
320 : {
321 2 : return mbufQueueType_;
322 : }
323 :
324 : }
|
