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 "entity.h"
12 : #include "state_manager.h"
13 :
14 : namespace dgw {
15 :
16 : namespace {
17 : constexpr uint32_t SCHEDULE_THRESHOLD = 100U;
18 : constexpr uint32_t DYNAMIC_SCHEDULE_THRESHOLD = 100U;
19 : }
20 :
21 366 : Entity::Entity(const EntityMaterial &material, const uint32_t resIndex)
22 366 : : type_(material.eType),
23 366 : id_(material.id),
24 366 : deviceId_(material.resId),
25 366 : hostGroupId_(material.hostGroupId),
26 366 : globalId_(material.globalId),
27 366 : uuId_(material.uuId),
28 366 : schedCfgKey_(material.schedCfgKey),
29 366 : resIndex_(resIndex),
30 366 : queueType_(material.queueType),
31 366 : subscribeStatus_(SubscribeStatus::SUBSCRIBE_INVALID),
32 366 : scheduleCount_(0U),
33 366 : curState_(FsmState::FSM_IDLE_STATE),
34 366 : mbuf_(nullptr),
35 366 : transId_(0UL),
36 366 : refCount_(0U),
37 366 : direction_(EntityDirection::DIRECTION_SEND),
38 366 : needTransId_(false),
39 366 : msgType_(InnerMsgType::INNER_MSG_INVALID),
40 366 : routeLabel_(0U),
41 366 : waitingDecision_(false),
42 366 : dynamicReqTime_(0UL)
43 : {
44 366 : (void)entityDesc_.append("qid:").append(std::to_string(id_))
45 732 : .append(", type:").append(std::to_string(static_cast<int32_t>(type_)))
46 732 : .append(", globalId:").append(std::to_string(globalId_))
47 732 : .append(", schedCfgKey:").append(std::to_string(schedCfgKey_))
48 732 : .append(", hostGrpId:").append(std::to_string(hostGroupId_))
49 732 : .append(", deviceId:").append(std::to_string(deviceId_))
50 732 : .append(", resIndex:").append(std::to_string(resIndex_))
51 366 : .append(", queue type:").append(std::to_string(queueType_));
52 366 : }
53 :
54 300 : FsmStatus Entity::Init(const FsmState state, const EntityDirection direction)
55 : {
56 300 : curState_ = state;
57 300 : direction_ = direction;
58 300 : entityDesc_.append(", direction:").append(std::to_string(static_cast<int32_t>(direction_)));
59 300 : return FsmStatus::FSM_SUCCESS;
60 : }
61 :
62 17 : FsmStatus Entity::Uninit()
63 : {
64 17 : return FsmStatus::FSM_SUCCESS;
65 : }
66 :
67 684 : uint32_t Entity::GetQueueId() const
68 : {
69 684 : return id_;
70 : }
71 :
72 82 : FsmStatus Entity::AllowDeque()
73 : {
74 82 : if (scheduleCount_ >= SCHEDULE_THRESHOLD) {
75 1 : return FsmStatus::FSM_FAILED;
76 : }
77 :
78 : // The destination receiver has not finished scheduling, keep peek state
79 : // if waitingDecision, maybe some decision is comming, then return fail,
80 : // so fsm can transfer to idle to process decision
81 140 : if ((!waitingDecision_ && !sendDataObjs_.empty()) ||
82 59 : (sendDataObjs_.size() > DYNAMIC_SCHEDULE_THRESHOLD)) {
83 22 : DGW_LOG_DEBUG("[FSM] Entity:[%s] state:[%s] not finish count:[%zu].",
84 : entityDesc_.c_str(), GetStateDesc(FsmState::FSM_PEEK_STATE).c_str(), sendDataObjs_.size());
85 22 : return waitingDecision_ ? FsmStatus::FSM_FAILED : FsmStatus::FSM_KEEP_STATE;
86 : };
87 :
88 59 : return FsmStatus::FSM_SUCCESS;
89 : }
90 :
91 1 : FsmStatus Entity::ResetSrcState()
92 : {
93 1 : return FsmStatus::FSM_SUCCESS;
94 : }
95 :
96 13 : void Entity::ResetSrcSubState()
97 : {
98 13 : return;
99 : }
100 :
101 1 : void Entity::ReprocessInTryPush(const Entity &srcEntity, DynamicRequestPtr &dynamicRequest, uint32_t &schedCfgKey)
102 : {
103 : (void)srcEntity;
104 : (void)dynamicRequest;
105 : (void)schedCfgKey;
106 1 : return;
107 : }
108 :
109 1 : FsmStatus Entity::AbProcessInTryPush()
110 : {
111 1 : return FsmStatus::FSM_SUCCESS;
112 : }
113 :
114 1 : FsmStatus Entity::SendData(const DataObjPtr dataObj)
115 : {
116 : (void)dataObj;
117 1 : return FsmStatus::FSM_SUCCESS;
118 : }
119 :
120 118 : FsmStatus Entity::ProcessMessage(const InnerMessage &msg)
121 : {
122 118 : DGW_LOG_DEBUG("[FSM] Entity qid:[%u] type:[%s] state:[%s] ProcessMessage.",
123 : id_, GetTypeDesc().c_str(), GetStateDesc(curState_).c_str());
124 : // if cur_state is not full, then there's no need to process f2nf msg
125 118 : if ((msg.msgType == dgw::InnerMsgType::INNER_MSG_F2NF) && (curState_ != FsmState::FSM_FULL_STATE)) {
126 1 : return FsmStatus::FSM_SUCCESS;
127 : }
128 :
129 117 : auto const state = StateManager::Instance().GetState(curState_, type_);
130 117 : if (state != nullptr) {
131 116 : return state->ProcessMessage(*this, msg);
132 : }
133 1 : DGW_LOG_ERROR("[FSM] Entity qid:[%u] type:[%s] state:[%s] get state failed.",
134 : id_, GetTypeDesc().c_str(), GetStateDesc(curState_).c_str());
135 1 : return FsmStatus::FSM_FAILED;
136 : }
137 :
138 268 : FsmStatus Entity::ChangeState(const FsmState nextState)
139 : {
140 268 : DGW_LOG_DEBUG("[FSM] Entity qid:[%u] type:[%s] change from state:[%s] to state:[%s].",
141 : id_, GetTypeDesc().c_str(), GetStateDesc(curState_).c_str(), GetStateDesc(nextState).c_str());
142 268 : curState_ = nextState;
143 268 : auto const state = StateManager::Instance().GetState(nextState, type_);
144 268 : if (state != nullptr) {
145 267 : return state->PreProcess(*this);
146 : }
147 1 : return FsmStatus::FSM_FAILED;
148 : }
149 :
150 30 : FsmStatus Entity::AddDataObjToSendList(const DataObjPtr &dataObj)
151 : {
152 30 : sendDataObjs_.push_back(dataObj);
153 30 : return FsmStatus::FSM_SUCCESS;
154 : }
155 :
156 36 : FsmStatus Entity::AddDataObjToRecvList(const DataObjPtr &dataObj)
157 : {
158 39 : for (const auto &recvDataObj : recvDataObjs_) {
159 4 : if (recvDataObj == dataObj) {
160 1 : DGW_LOG_INFO("Skip AddDataObjToRecvList");
161 1 : return FsmStatus::FSM_SUCCESS;
162 : }
163 : }
164 35 : recvDataObjs_.push_back(dataObj);
165 35 : return FsmStatus::FSM_SUCCESS;
166 : }
167 :
168 3 : FsmStatus Entity::RemoveDataObjFromSendList(const DataObjPtr &dataObj)
169 : {
170 3 : if (!sendDataObjs_.empty()) {
171 : // what's intension?
172 3 : if ((dataObj != nullptr) && (dataObj->GetSendEntity() != nullptr)) {
173 3 : DGW_LOG_INFO("[FSM] id:[%u] type:[%s] state:[%s], remove id:[%u].",
174 : id_, GetTypeDesc().c_str(), GetStateDesc(curState_).c_str(), dataObj->GetSendEntity()->GetId());
175 : }
176 3 : sendDataObjs_.pop_front();
177 3 : DGW_LOG_INFO("Entity[%s] remove one sendDataObj", entityDesc_.c_str());
178 : }
179 3 : return FsmStatus::FSM_SUCCESS;
180 : }
181 :
182 12 : void Entity::RemoveRecvEntityFromSendList(const Entity* const recvEntityPtr)
183 : {
184 12 : if (!sendDataObjs_.empty()) {
185 10 : const auto sendDataObj = sendDataObjs_.front();
186 10 : sendDataObj->RemoveRecvEntity(recvEntityPtr);
187 10 : if (sendDataObj->GetRecvEntitySize() == 0U) {
188 9 : sendDataObjs_.pop_front();
189 : }
190 10 : }
191 :
192 12 : if (sendDataObjs_.empty()) {
193 11 : waitingDecision_ = false;
194 : }
195 12 : }
196 :
197 1372 : const std::string &Entity::GetTypeDesc() const
198 : {
199 1372 : return StateManager::Instance().GetTypeDesc(type_);
200 : }
201 :
202 1869 : const std::string &Entity::GetStateDesc(const FsmState id) const
203 : {
204 1869 : return StateManager::Instance().GetStateDesc(id, type_);
205 : }
206 :
207 14 : bool Entity::Equal(const Entity * const recvEntityPtr) const
208 : {
209 41 : return ((type_ == recvEntityPtr->GetType()) && (id_ == recvEntityPtr->GetId()) &&
210 41 : (deviceId_ == recvEntityPtr->GetDeviceId()) && (queueType_ == recvEntityPtr->GetQueueType()));
211 : }
212 :
213 2 : bool Entity::UpdateSendObject(const EntityPtr group, const EntityPtr elem)
214 : {
215 2 : for (auto sendDataObj : sendDataObjs_) {
216 1 : DGW_LOG_INFO("Try to replace group[%s] with elem[%s] in Entity[%s].",
217 : group->ToString().c_str(), elem->ToString().c_str(), ToString().c_str());
218 1 : if (sendDataObj->UpdateRecvEntities(group, elem)) {
219 1 : DGW_LOG_INFO("replace group[%s] with elem[%s] in Entity[%s].",
220 : group->ToString().c_str(), elem->ToString().c_str(), ToString().c_str());
221 1 : return true;
222 : }
223 1 : }
224 1 : DGW_LOG_ERROR("Invalid UpdateSendObject for dst[%s] with elem[%s] in Entity[%s].",
225 : group->ToString().c_str(), elem->ToString().c_str(), entityDesc_.c_str());
226 1 : return false;
227 : }
228 :
229 1 : FsmStatus Entity::MakeSureOutputCompletion()
230 : {
231 1 : return FsmStatus::FSM_SUCCESS;
232 : }
233 :
234 1 : bool Entity::IsDataPeeked() const
235 : {
236 1 : return false;
237 : }
238 :
239 20 : uint32_t Entity::GetMbufDeviceId() const
240 : {
241 20 : return deviceId_;
242 : }
243 :
244 20 : uint32_t Entity::GetMbufQueueType() const
245 : {
246 20 : return queueType_;
247 : }
248 : }
|
