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("[FSM] Entity id:[%u] type:[%s] state:[%s] desc:[%s].",
25 : 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("Begin to sendEntity:[%s] mbuf for entity:[%s].",
38 : dataObj->GetSendEntity()->ToString().c_str(), entity.ToString().c_str());
39 30 : bqs::ProfileManager::GetInstance(entity.GetResIndex()).AddEnqueueNum();
40 30 : ret = entity.SendData(dataObj);
41 30 : if (ret == FsmStatus::FSM_KEEP_STATE) {
42 10 : return PostProcess(entity);
43 : }
44 20 : if (ret != FsmStatus::FSM_SUCCESS) {
45 10 : DGW_LOG_WARN("Push to %s[%u] in device[%u] failed, error=[%d].",
46 : entity.GetTypeDesc().c_str(), entity.GetId(), entity.GetDeviceId(), static_cast<int32_t>(ret));
47 10 : bqs::StatisticManager::GetInstance().DataScheduleFailedStat();
48 10 : if (ret == FsmStatus::FSM_DEST_FULL) {
49 0 : return entity.ChangeState(FsmState::FSM_FULL_STATE);
50 : }
51 10 : if (ret == FsmStatus::FSM_ERROR_PENDING) {
52 2 : return entity.ChangeState(FsmState::FSM_ERROR_STATE);
53 : }
54 8 : return PostProcess(entity);
55 : }
56 :
57 10 : DGW_LOG_INFO("Push to %s[%u] in device[%u], owneredDevId[%u], ret is %d.",
58 : entity.GetTypeDesc().c_str(), entity.GetId(), entity.GetDeviceId(), entity.GetDeviceId(),
59 : static_cast<int32_t>(ret));
60 :
61 10 : if ((entity.GetMessageType() == InnerMsgType::INNER_MSG_F2NF) &&
62 0 : (dgw::EntityManager::Instance(entity.GetResIndex()).ShouldPauseSubscirpiton())) {
63 0 : (void)dataObj->GetSendEntity()->ResumeSubscribe(entity);
64 : }
65 :
66 10 : dataObj->GetSendEntity()->RemoveRecvEntityFromSendList(&entity);
67 10 : if ((dataObj->GetRecvEntitySize() == 0U) && !dataObj->ShouldMaintainMbuf()) {
68 2 : DGW_LOG_INFO("Entity[%s] free mbuf", entity.ToString().c_str());
69 2 : (void)halMbufFree(mbuf);
70 : }
71 10 : recvDataObjs.pop_front();
72 : }
73 36 : return PostProcess(entity);
74 : }
75 :
76 54 : FsmStatus PushState::PostProcess(Entity &entity)
77 : {
78 54 : return entity.ChangeState(FsmState::FSM_WAIT_PUSH_STATE);
79 : }
80 :
81 : REGISTER_STATE(FSM_PUSH_STATE, ENTITY_QUEUE, PushState);
82 : REGISTER_STATE(FSM_PUSH_STATE, ENTITY_TAG, PushState);
83 : } // namespace dgw
|