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 "bind_relation.h"
12 : #include "common/bqs_log.h"
13 : #include "fsm/try_push_state.h"
14 : #include "profile_manager.h"
15 : #include "state_manager.h"
16 :
17 : namespace dgw {
18 28 : FsmStatus TryPushState::PreProcess(Entity &entity)
19 : {
20 28 : DGW_LOG_INFO("[FSM] Entity id:[%u] type:[%s] state:[%s] desc:[%s].",
21 : entity.GetId(), entity.GetTypeDesc().c_str(),
22 : entity.GetStateDesc(FsmState::FSM_TRY_PUSH_STATE).c_str(), entity.ToString().c_str());
23 :
24 : // use entity to find routes
25 28 : bqs::OptionalArg args = {};
26 28 : args.eType = entity.GetType();
27 28 : args.queueType = entity.GetQueueType();
28 28 : bqs::EntityInfo entityInfo(entity.GetId(), entity.GetDeviceId(), &args);
29 : // get ordered subscribe queue
30 56 : auto &srcToDstRelation = (entity.GetResIndex() == 0U) ? bqs::BindRelation::GetInstance().GetSrcToDstRelation() :
31 28 : bqs::BindRelation::GetInstance().GetSrcToDstExtraRelation();
32 28 : bqs::ProfileManager::GetInstance(entity.GetResIndex()).
33 28 : SetSrcQueueNum(static_cast<uint32_t>(srcToDstRelation.size()));
34 28 : const auto iter = srcToDstRelation.find(entityInfo);
35 28 : if (iter == srcToDstRelation.end()) {
36 1 : DGW_LOG_WARN("Can't find dst entities for entity:%u", entity.GetId());
37 1 : return PostProcess(entity);
38 : }
39 :
40 27 : std::vector<Entity*> dstEntitiesCanPush;
41 27 : std::vector<Entity*> reprocessDstEntities;
42 27 : std::vector<Entity*> abnormalDstEntities;
43 57 : for (auto &dst : iter->second) {
44 30 : const auto dstEntity = dst.GetEntity();
45 30 : if (dstEntity == nullptr) {
46 0 : DGW_LOG_WARN("[FSM] Recv entity is nullptr, id:[%u].", dst.GetId());
47 0 : continue;
48 : }
49 30 : DGW_LOG_INFO("Find dst entity, id:[%u] type:[%s].", dstEntity->GetId(), dstEntity->GetTypeDesc().c_str());
50 30 : dstEntity->SelectDstEntities(entity.GetTransId() + entity.GetRouteLabel(), dstEntitiesCanPush,
51 : reprocessDstEntities, abnormalDstEntities);
52 30 : }
53 :
54 27 : Mbuf *const mbuf = entity.GetMbuf();
55 : // check if current data is the first one to process{entity.sendObject size == 0}
56 27 : const bool firstData = (entity.GetSendDataObjs().size() == 0U);
57 27 : auto dataObj = DataObjManager::Instance().CreateDataObj(&entity, mbuf);
58 27 : if (dataObj == nullptr) {
59 0 : DGW_LOG_WARN("Can't alloc dataObj for entity:[%s].", entity.ToString().c_str());
60 0 : return FsmStatus::FSM_SUCCESS;
61 : }
62 27 : std::vector<Entity*> dstEntitiesToPush;
63 53 : for (auto canPushDstEntity: dstEntitiesCanPush) {
64 26 : dataObj->AddRecvEntity(canPushDstEntity);
65 26 : if (!firstData) {
66 : // 前序数据未完成发送，当前数据不能发送
67 1 : continue;
68 : }
69 25 : (void)dstEntitiesToPush.emplace_back(canPushDstEntity);
70 : }
71 :
72 27 : std::shared_ptr<DynamicSchedMgr::RequestInfo> dynamicRequest = nullptr;
73 27 : uint32_t schedCfgKey = 0U;
74 30 : for (auto reprocessEntity : reprocessDstEntities) {
75 3 : dataObj->AddRecvEntity(reprocessEntity);
76 3 : reprocessEntity->ReprocessInTryPush(entity, dynamicRequest, schedCfgKey);
77 : }
78 27 : const auto sendRet = SendRequestForDynamicGroup(dynamicRequest, schedCfgKey, entity);
79 27 : if (sendRet != FsmStatus::FSM_SUCCESS) {
80 1 : return sendRet;
81 : }
82 26 : if (dataObj->GetRecvEntitySize() > 0U) {
83 24 : entity.AddDataObjToSendList(dataObj);
84 : }
85 :
86 : // if current data is not the first one to process, dstEntitiesToPush will be empty
87 26 : InnerMessage msg;
88 26 : msg.msgType = InnerMsgType::INNER_MSG_PUSH;
89 26 : FsmStatus processRet = FsmStatus::FSM_SUCCESS;
90 50 : for (auto entityToPush: dstEntitiesToPush) {
91 : // schedule each receiving entity
92 24 : (void)entityToPush->AddDataObjToRecvList(dataObj);
93 24 : if (entityToPush->ProcessMessage(msg) == FsmStatus::FSM_ERROR) {
94 1 : processRet = FsmStatus::FSM_ERROR;
95 1 : entity.RemoveRecvEntityFromSendList(entityToPush);
96 : };
97 : }
98 :
99 27 : for (auto abnormalEntity: abnormalDstEntities) {
100 1 : if (abnormalEntity->AbProcessInTryPush() == FsmStatus::FSM_ERROR) {
101 1 : processRet = FsmStatus::FSM_ERROR;
102 : }
103 : }
104 :
105 26 : if (processRet == FsmStatus::FSM_ERROR) {
106 2 : (void)entity.ChangeState(FsmState::FSM_IDLE_STATE);
107 2 : return processRet;
108 : }
109 :
110 24 : return PostProcess(entity);
111 28 : }
112 :
113 27 : FsmStatus TryPushState::SendRequestForDynamicGroup(const DynamicRequestPtr dynamicRequest, const uint32_t schedCfgKey,
114 : Entity &entity) const
115 : {
116 27 : if (dynamicRequest == nullptr) {
117 24 : return FsmStatus::FSM_SUCCESS;
118 : }
119 3 : std::vector<DynamicSchedMgr::RequestInfo> dynamicRequests;
120 3 : dynamicRequests.emplace_back(*dynamicRequest);
121 3 : const auto requestRet = DynamicSchedMgr::GetInstance(entity.GetResIndex()).
122 3 : SendRequest(schedCfgKey, dynamicRequests);
123 3 : if (requestRet != FsmStatus::FSM_SUCCESS) {
124 1 : DGW_LOG_WARN("Entity:[%s] sendRequest fail, ret is %d.",
125 : entity.ToString().c_str(), static_cast<int32_t>(requestRet));
126 1 : return requestRet;
127 : }
128 2 : DGW_LOG_INFO("Entity[%s] SetWaitDecisionState to true", entity.ToString().c_str());
129 2 : entity.SetDynamicReqTime(DynamicSchedMgr::GetInstance(entity.GetResIndex()).DynamicSchedNow());
130 2 : entity.SetWaitDecisionState(true);
131 2 : return FsmStatus::FSM_SUCCESS;
132 3 : }
133 :
134 0 : FsmStatus TryPushState::ProcessMessage(Entity &entity, const InnerMessage &msg)
135 : {
136 : (void)msg;
137 0 : return PreProcess(entity);
138 : }
139 :
140 25 : FsmStatus TryPushState::PostProcess(Entity &entity)
141 : {
142 25 : return entity.ChangeState(FsmState::FSM_PEEK_STATE);
143 : }
144 :
145 : REGISTER_STATE(FSM_TRY_PUSH_STATE, ENTITY_QUEUE, TryPushState);
146 : REGISTER_STATE(FSM_TRY_PUSH_STATE, ENTITY_TAG, TryPushState);
147 : REGISTER_STATE(FSM_TRY_PUSH_STATE, ENTITY_GROUP, TryPushState);
148 : } // namespace dgw
|
