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(
21 : "[FSM] Entity id:[%u] type:[%s] state:[%s] desc:[%s].", 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(
51 30 : entity.GetTransId() + entity.GetRouteLabel(), dstEntitiesCanPush, reprocessDstEntities,
52 : abnormalDstEntities);
53 30 : }
54 :
55 27 : Mbuf* const mbuf = entity.GetMbuf();
56 : // check if current data is the first one to process{entity.sendObject size == 0}
57 27 : const bool firstData = (entity.GetSendDataObjs().size() == 0U);
58 27 : auto dataObj = DataObjManager::Instance().CreateDataObj(&entity, mbuf);
59 27 : if (dataObj == nullptr) {
60 0 : DGW_LOG_WARN("Can't alloc dataObj for entity:[%s].", entity.ToString().c_str());
61 0 : return FsmStatus::FSM_SUCCESS;
62 : }
63 27 : std::vector<Entity*> dstEntitiesToPush;
64 53 : for (auto canPushDstEntity : dstEntitiesCanPush) {
65 26 : dataObj->AddRecvEntity(canPushDstEntity);
66 26 : if (!firstData) {
67 : // 前序数据未完成发送，当前数据不能发送
68 1 : continue;
69 : }
70 25 : (void)dstEntitiesToPush.emplace_back(canPushDstEntity);
71 : }
72 :
73 27 : std::shared_ptr<DynamicSchedMgr::RequestInfo> dynamicRequest = nullptr;
74 27 : uint32_t schedCfgKey = 0U;
75 30 : for (auto reprocessEntity : reprocessDstEntities) {
76 3 : dataObj->AddRecvEntity(reprocessEntity);
77 3 : reprocessEntity->ReprocessInTryPush(entity, dynamicRequest, schedCfgKey);
78 : }
79 27 : const auto sendRet = SendRequestForDynamicGroup(dynamicRequest, schedCfgKey, entity);
80 27 : if (sendRet != FsmStatus::FSM_SUCCESS) {
81 1 : return sendRet;
82 : }
83 26 : if (dataObj->GetRecvEntitySize() > 0U) {
84 24 : entity.AddDataObjToSendList(dataObj);
85 : }
86 :
87 : // if current data is not the first one to process, dstEntitiesToPush will be empty
88 26 : InnerMessage msg;
89 26 : msg.msgType = InnerMsgType::INNER_MSG_PUSH;
90 26 : FsmStatus processRet = FsmStatus::FSM_SUCCESS;
91 50 : for (auto entityToPush : dstEntitiesToPush) {
92 : // schedule each receiving entity
93 24 : (void)entityToPush->AddDataObjToRecvList(dataObj);
94 24 : if (entityToPush->ProcessMessage(msg) == FsmStatus::FSM_ERROR) {
95 1 : processRet = FsmStatus::FSM_ERROR;
96 1 : entity.RemoveRecvEntityFromSendList(entityToPush);
97 : };
98 : }
99 :
100 27 : for (auto abnormalEntity : abnormalDstEntities) {
101 1 : if (abnormalEntity->AbProcessInTryPush() == FsmStatus::FSM_ERROR) {
102 1 : processRet = FsmStatus::FSM_ERROR;
103 : }
104 : }
105 :
106 26 : if (processRet == FsmStatus::FSM_ERROR) {
107 2 : (void)entity.ChangeState(FsmState::FSM_IDLE_STATE);
108 2 : return processRet;
109 : }
110 :
111 24 : return PostProcess(entity);
112 28 : }
113 :
114 27 : FsmStatus TryPushState::SendRequestForDynamicGroup(
115 : const DynamicRequestPtr dynamicRequest, const uint32_t schedCfgKey, Entity& entity) const
116 : {
117 27 : if (dynamicRequest == nullptr) {
118 24 : return FsmStatus::FSM_SUCCESS;
119 : }
120 3 : std::vector<DynamicSchedMgr::RequestInfo> dynamicRequests;
121 3 : dynamicRequests.emplace_back(*dynamicRequest);
122 : const auto requestRet =
123 3 : DynamicSchedMgr::GetInstance(entity.GetResIndex()).SendRequest(schedCfgKey, dynamicRequests);
124 3 : if (requestRet != FsmStatus::FSM_SUCCESS) {
125 1 : DGW_LOG_WARN(
126 : "Entity:[%s] sendRequest fail, ret is %d.", entity.ToString().c_str(), static_cast<int32_t>(requestRet));
127 1 : return requestRet;
128 : }
129 2 : DGW_LOG_INFO("Entity[%s] SetWaitDecisionState to true", entity.ToString().c_str());
130 2 : entity.SetDynamicReqTime(DynamicSchedMgr::GetInstance(entity.GetResIndex()).DynamicSchedNow());
131 2 : entity.SetWaitDecisionState(true);
132 2 : return FsmStatus::FSM_SUCCESS;
133 3 : }
134 :
135 0 : FsmStatus TryPushState::ProcessMessage(Entity& entity, const InnerMessage& msg)
136 : {
137 : (void)msg;
138 0 : return PreProcess(entity);
139 : }
140 :
141 25 : FsmStatus TryPushState::PostProcess(Entity& entity) { return entity.ChangeState(FsmState::FSM_PEEK_STATE); }
142 :
143 : REGISTER_STATE(FSM_TRY_PUSH_STATE, ENTITY_QUEUE, TryPushState);
144 : REGISTER_STATE(FSM_TRY_PUSH_STATE, ENTITY_TAG, TryPushState);
145 : REGISTER_STATE(FSM_TRY_PUSH_STATE, ENTITY_GROUP, TryPushState);
146 : } // namespace dgw
|
