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 "fsm/push_state.h"
12 : #include "common/bqs_log.h"
13 : #include "entity_manager.h"
14 : #include "data_obj_manager.h"
15 : #include "driver/ascend_hal.h"
16 : #include "profile_manager.h"
17 : #include "state_manager.h"
18 : #include "statistic_manager.h"
19 :
20 : namespace dgw {
21 :
22 56 : FsmStatus PushState::PreProcess(Entity& entity)
23 : {
24 56 : DGW_LOG_INFO(
25 : "[FSM] Entity id:[%u] type:[%s] state:[%s] desc:[%s].", entity.GetId(), entity.GetTypeDesc().c_str(),
26 : entity.GetStateDesc(FsmState::FSM_PUSH_STATE).c_str(), entity.ToString().c_str());
27 56 : auto& recvDataObjs = entity.GetRecvDataObjs();
28 56 : FsmStatus ret = FsmStatus::FSM_SUCCESS;
29 66 : while (!recvDataObjs.empty()) {
30 30 : const auto& dataObj = recvDataObjs.front();
31 30 : if ((dataObj == nullptr) || (dataObj->GetSendEntity() == nullptr)) {
32 0 : recvDataObjs.pop_front();
33 0 : continue;
34 : }
35 :
36 30 : Mbuf* const mbuf = const_cast<Mbuf*>(dataObj->GetMbuf());
37 30 : DGW_LOG_INFO(
38 : "Begin to sendEntity:[%s] mbuf for entity:[%s].", dataObj->GetSendEntity()->ToString().c_str(),
39 : entity.ToString().c_str());
40 30 : bqs::ProfileManager::GetInstance(entity.GetResIndex()).AddEnqueueNum();
41 30 : ret = entity.SendData(dataObj);
42 30 : if (ret == FsmStatus::FSM_KEEP_STATE) {
43 10 : return PostProcess(entity);
44 : }
45 20 : if (ret != FsmStatus::FSM_SUCCESS) {
46 10 : DGW_LOG_WARN(
47 : "Push to %s[%u] in device[%u] failed, error=[%d].", entity.GetTypeDesc().c_str(), entity.GetId(),
48 : entity.GetDeviceId(), static_cast<int32_t>(ret));
49 10 : bqs::StatisticManager::GetInstance().DataScheduleFailedStat();
50 10 : if (ret == FsmStatus::FSM_DEST_FULL) {
51 0 : return entity.ChangeState(FsmState::FSM_FULL_STATE);
52 : }
53 10 : if (ret == FsmStatus::FSM_ERROR_PENDING) {
54 2 : return entity.ChangeState(FsmState::FSM_ERROR_STATE);
55 : }
56 8 : return PostProcess(entity);
57 : }
58 :
59 10 : DGW_LOG_INFO(
60 : "Push to %s[%u] in device[%u], owneredDevId[%u], ret is %d.", entity.GetTypeDesc().c_str(), entity.GetId(),
61 : entity.GetDeviceId(), entity.GetDeviceId(), static_cast<int32_t>(ret));
62 :
63 10 : if ((entity.GetMessageType() == InnerMsgType::INNER_MSG_F2NF) &&
64 0 : (dgw::EntityManager::Instance(entity.GetResIndex()).ShouldPauseSubscirpiton())) {
65 0 : (void)dataObj->GetSendEntity()->ResumeSubscribe(entity);
66 : }
67 :
68 10 : dataObj->GetSendEntity()->RemoveRecvEntityFromSendList(&entity);
69 10 : if ((dataObj->GetRecvEntitySize() == 0U) && !dataObj->ShouldMaintainMbuf()) {
70 2 : DGW_LOG_INFO("Entity[%s] free mbuf", entity.ToString().c_str());
71 2 : (void)halMbufFree(mbuf);
72 : }
73 10 : recvDataObjs.pop_front();
74 : }
75 36 : return PostProcess(entity);
76 : }
77 :
78 54 : FsmStatus PushState::PostProcess(Entity& entity) { return entity.ChangeState(FsmState::FSM_WAIT_PUSH_STATE); }
79 :
80 : REGISTER_STATE(FSM_PUSH_STATE, ENTITY_QUEUE, PushState);
81 : REGISTER_STATE(FSM_PUSH_STATE, ENTITY_TAG, PushState);
82 : } // namespace dgw
|