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