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