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 "client_entity.h"
12 :
13 : #include <thread>
14 :
15 : #include "entity_manager.h"
16 : #include "queue_manager.h"
17 :
18 : namespace dgw {
19 :
20 30 : ClientEntity::ClientEntity(const EntityMaterial& material, const uint32_t resIndex)
21 30 : : SimpleEntity(material, resIndex), asyncDataState_(AsyncDataState::FSM_ASYNC_DATA_INIT)
22 : {
23 30 : (void)entityDesc_.append(", data state:").append(std::to_string(static_cast<int32_t>(asyncDataState_)));
24 30 : }
25 :
26 13 : FsmStatus ClientEntity::DoDequeue()
27 : {
28 13 : if (asyncDataState_ == AsyncDataState::FSM_ASYNC_DATA_WAIT) {
29 1 : return FsmStatus::FSM_KEEP_STATE;
30 : }
31 12 : if (asyncDataState_ == AsyncDataState::FSM_ASYNC_DATA_SENT) {
32 1 : asyncDataState_ = AsyncDataState::FSM_ASYNC_ENTITY_DONE;
33 1 : return FsmStatus::FSM_SUCCESS;
34 : }
35 :
36 11 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_WAIT;
37 : try {
38 11 : InvokeDequeThread();
39 0 : } catch (std::exception& e) {
40 0 : DGW_LOG_ERROR("create async mem buff thread object failed, %s", e.what());
41 0 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_INIT;
42 0 : return FsmStatus::FSM_FAILED;
43 0 : }
44 11 : return FsmStatus::FSM_KEEP_STATE;
45 : }
46 :
47 11 : void ClientEntity::InvokeDequeThread()
48 : {
49 11 : std::thread th(&ClientEntity::DoClientDequeue, this);
50 11 : th.detach();
51 11 : }
52 :
53 13 : void ClientEntity::DoClientDequeue()
54 : {
55 13 : if (DoDequeueMbuf(PtrToPtr<Mbuf*, void*>(&mbuf_)) != FsmStatus::FSM_SUCCESS) {
56 12 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_INIT;
57 12 : return;
58 : }
59 :
60 1 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_SENT;
61 : // enable AsyncMemDequeue Queue flag
62 1 : bqs::QueueManager::GetInstance().enableAsyncMemDequeueFlag();
63 1 : const auto bqsRet = bqs::QueueManager::GetInstance().EnqueueAsynMemBuffEvent();
64 1 : if (bqsRet != bqs::BQS_STATUS_OK) {
65 1 : DGW_LOG_ERROR(
66 : "failed to EnqueueAsynMemBuffEvent for entity[%s], ret[%d].", entityDesc_.c_str(),
67 : static_cast<int32_t>(bqsRet));
68 : }
69 : }
70 :
71 23 : FsmStatus ClientEntity::DoDequeueMbuf(void** mbufPtr) const
72 : {
73 23 : DGW_LOG_INFO("Entity[%s] Start AsyncMemBuffDeQueueEvent", entityDesc_.c_str());
74 23 : const uint32_t deviceId = GetDeviceId();
75 23 : const uint32_t queueId = GetQueueId();
76 :
77 : // q empty idle
78 23 : int32_t srcStatus = static_cast<int32_t>(QUEUE_NORMAL);
79 : auto ret =
80 23 : halQueueGetStatus(deviceId, queueId, QUERY_QUEUE_STATUS, static_cast<uint32_t>(sizeof(uint32_t)), &srcStatus);
81 23 : if (ret != DRV_ERROR_NONE) {
82 4 : DGW_LOG_WARN("queue[%u] on device[%u] halQueueGetStatus ret=%d", queueId, deviceId, static_cast<int32_t>(ret));
83 4 : return (ret == DRV_ERROR_NOT_EXIST) ? FsmStatus::FSM_ERROR_PENDING : FsmStatus::FSM_FAILED;
84 : }
85 19 : if (srcStatus == static_cast<int32_t>(QUEUE_EMPTY)) {
86 1 : DGW_LOG_DEBUG("Entity[%s] has been dequeued to empty", entityDesc_.c_str());
87 1 : return FsmStatus::FSM_SRC_EMPTY;
88 : }
89 :
90 18 : DGW_LOG_INFO("Entity[%s] Begin halQueuePeek", entityDesc_.c_str());
91 : // get mbuf data
92 18 : uint64_t deqLen = 0U;
93 18 : ret = halQueuePeek(deviceId, queueId, &deqLen, -1);
94 18 : if ((ret != DRV_ERROR_NONE) || (deqLen == 0U)) {
95 2 : DGW_LOG_ERROR(
96 : "halQueuePeek from queue[%u] in device[%u] failed, ret[%d], deqLen[%lu]", queueId, deviceId,
97 : static_cast<int32_t>(ret), deqLen);
98 2 : return (ret == DRV_ERROR_NOT_EXIST) ? FsmStatus::FSM_ERROR_PENDING : FsmStatus::FSM_FAILED;
99 : }
100 16 : DGW_LOG_INFO("Entity[%s] Finish halQueuePeek", entityDesc_.c_str());
101 :
102 : // alloc mbuf
103 16 : Mbuf* mbuf = nullptr;
104 16 : int32_t retCode = halMbufAlloc(deqLen, &mbuf);
105 16 : if ((retCode != DRV_ERROR_NONE) || (mbuf == nullptr)) {
106 2 : DGW_LOG_ERROR("halMbufAlloc fail for entity[%s], size[%zu], ret=[%d].", entityDesc_.c_str(), deqLen, retCode);
107 2 : return FsmStatus::FSM_FAILED;
108 : }
109 :
110 14 : const auto dequeRet = FillMbufWithDeque(deqLen, mbuf);
111 14 : if (dequeRet != FsmStatus::FSM_SUCCESS) {
112 12 : (void)halMbufFree(mbuf);
113 12 : return dequeRet;
114 : }
115 2 : *mbufPtr = mbuf;
116 2 : return FsmStatus::FSM_SUCCESS;
117 : }
118 :
119 14 : FsmStatus ClientEntity::FillMbufWithDeque(const uint64_t deqLen, Mbuf* const mbuf) const
120 : {
121 : // setdatalen
122 14 : auto retCode = halMbufSetDataLen(mbuf, deqLen);
123 14 : if (retCode != static_cast<int32_t>(DRV_ERROR_NONE)) {
124 1 : DGW_LOG_ERROR("Failed to call halMbufSetDataLen for entity[%s], ret=[%d].", entityDesc_.c_str(), retCode);
125 1 : return FsmStatus::FSM_FAILED;
126 : }
127 :
128 : // get mbuf head
129 13 : void* headBuf = nullptr;
130 13 : uint32_t headBufSize = 0U;
131 13 : retCode = halMbufGetPrivInfo(mbuf, &headBuf, &headBufSize);
132 13 : if (retCode != static_cast<int32_t>(DRV_ERROR_NONE)) {
133 2 : DGW_LOG_ERROR("halMbufGetPrivInfo from mbuf for entity[%s] failed, ret is %d.", entityDesc_.c_str(), retCode);
134 2 : return FsmStatus::FSM_FAILED;
135 : }
136 11 : if (headBufSize < MBUF_HEAD_MAX_SIZE) {
137 1 : DGW_LOG_ERROR("mbuf head size:%u is invalid.", headBufSize);
138 1 : return FsmStatus::FSM_FAILED;
139 : }
140 :
141 : // get mbuf data
142 10 : void* dataPtr = nullptr;
143 10 : retCode = halMbufGetBuffAddr(mbuf, &dataPtr);
144 10 : if ((retCode != static_cast<int32_t>(DRV_ERROR_NONE)) || (dataPtr == nullptr)) {
145 7 : DGW_LOG_ERROR("Failed to get data or data is nullptr, retCode[%d].", retCode);
146 7 : return FsmStatus::FSM_FAILED;
147 : }
148 :
149 3 : const size_t totalLen = sizeof(struct buff_iovec) + sizeof(struct iovec_info);
150 3 : std::unique_ptr<char_t[]> vecUniquePtr(new (std::nothrow) char_t[totalLen], std::default_delete<char_t[]>());
151 3 : DGW_CHECK(
152 : (vecUniquePtr != nullptr), FsmStatus::FSM_FAILED, "failed to alloc memory for buffIovec, size[%zu].", totalLen);
153 :
154 3 : buff_iovec* const buffIovec = PtrToPtr<char_t, buff_iovec>(vecUniquePtr.get());
155 3 : buffIovec->context_base = headBuf;
156 3 : buffIovec->context_len = headBufSize;
157 3 : buffIovec->count = 1U;
158 3 : buffIovec->ptr[0U].iovec_base = dataPtr;
159 3 : buffIovec->ptr[0U].len = deqLen;
160 3 : const auto ret = halQueueDeQueueBuff(deviceId_, id_, buffIovec, -1);
161 3 : if (ret != DRV_ERROR_NONE) {
162 1 : DGW_LOG_ERROR(
163 : "halQueueDeQueueBuff queue[%u] on device[%u] fail, ret is %d.", id_, deviceId_, static_cast<int32_t>(ret));
164 1 : return FsmStatus::FSM_FAILED;
165 : }
166 :
167 2 : return FsmStatus::FSM_SUCCESS;
168 3 : }
169 :
170 1 : FsmStatus ClientEntity::ResetSrcState()
171 : {
172 1 : ResetSrcSubState();
173 1 : return ChangeState(FsmState::FSM_IDLE_STATE);
174 : }
175 :
176 1 : void ClientEntity::ResetSrcSubState() { asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_INIT; }
177 :
178 10 : Mbuf* ClientEntity::PrepareMbufToPush(DataObjPtr dataObj) const
179 : {
180 10 : Mbuf* const mbuf = const_cast<Mbuf*>(dataObj->GetMbuf());
181 10 : return mbuf;
182 : }
183 :
184 14 : FsmStatus ClientEntity::DoSendData(Mbuf* const mbuf)
185 : {
186 14 : if (asyncDataState_ == AsyncDataState::FSM_ASYNC_DATA_WAIT) {
187 1 : return FsmStatus::FSM_KEEP_STATE;
188 : }
189 13 : if (asyncDataState_ == AsyncDataState::FSM_ASYNC_DATA_SENT) {
190 2 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_INIT;
191 2 : return FsmStatus::FSM_SUCCESS;
192 : }
193 :
194 11 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_WAIT;
195 : try {
196 11 : InvokeEnqueThread(mbuf);
197 0 : } catch (std::exception& e) {
198 0 : DGW_LOG_ERROR("create aync mem buff thread object failed, %s", e.what());
199 0 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_INIT;
200 0 : return FsmStatus::FSM_FAILED;
201 0 : }
202 11 : return FsmStatus::FSM_KEEP_STATE;
203 : }
204 :
205 11 : void ClientEntity::InvokeEnqueThread(Mbuf* const mbuf)
206 : {
207 11 : std::thread th(&ClientEntity::DoClientEnqueue, this, mbuf);
208 11 : th.detach();
209 11 : }
210 :
211 18 : FsmStatus ClientEntity::DoClientEnqueue(Mbuf* const mbuf)
212 : {
213 18 : DGW_LOG_INFO("Entity[%s] Start AsyncMemBuffEnQueueEvent", entityDesc_.c_str());
214 18 : const auto ret = DoEnqueueMbuf(mbuf);
215 18 : if (ret != FsmStatus::FSM_SUCCESS) {
216 16 : if (ret == FsmStatus::FSM_DEST_FULL) {
217 2 : bqs::StatisticManager::GetInstance().DataQueueEnqueueFullStat();
218 : } else {
219 14 : bqs::StatisticManager::GetInstance().DataQueueEnqueueFailStat();
220 : }
221 16 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_INIT;
222 : } else {
223 2 : bqs::StatisticManager::GetInstance().DataQueueEnqueueSuccStat();
224 : // mask dst aysnc mem entity
225 2 : dgw::EntityManager::Instance().SetExistAsyncMemEntity();
226 : // enable AsyncMemEnqueue Queue flag
227 2 : bqs::QueueManager::GetInstance().enableAsyncMemEnqueueFlag();
228 2 : const auto bqsRet = bqs::QueueManager::GetInstance().EnqueueAsynMemBuffEvent();
229 2 : if (bqsRet != bqs::BQS_STATUS_OK) {
230 1 : DGW_LOG_ERROR("failed to EnqueueAsynMemBuffEvent, ret[%d].", static_cast<int32_t>(bqsRet));
231 : }
232 2 : asyncDataState_ = AsyncDataState::FSM_ASYNC_DATA_SENT;
233 : }
234 18 : return ret;
235 : }
236 :
237 18 : FsmStatus ClientEntity::DoEnqueueMbuf(Mbuf* const mbuf) const
238 : {
239 : // get mbuf head
240 18 : void* headBuf = nullptr;
241 18 : uint32_t headBufSize = 0U;
242 18 : auto ret = halMbufGetPrivInfo(mbuf, &headBuf, &headBufSize);
243 18 : if (ret != DRV_ERROR_NONE) {
244 3 : DGW_LOG_ERROR("halMbufGetPrivInfo from mbuf failed, ret is %d.", static_cast<int32_t>(ret));
245 3 : return FsmStatus::FSM_FAILED;
246 : }
247 15 : if (headBufSize < MBUF_HEAD_MAX_SIZE) {
248 1 : DGW_LOG_ERROR("mbuf head size:%u is invalid.", headBufSize);
249 1 : return FsmStatus::FSM_FAILED;
250 : }
251 :
252 : // get mbuf data length
253 14 : uint64_t dataLen = 0UL;
254 14 : auto retCode = halMbufGetDataLen(mbuf, &dataLen);
255 14 : if ((retCode != static_cast<int32_t>(DRV_ERROR_NONE)) || (dataLen == 0)) {
256 1 : DGW_LOG_ERROR("Fail to get buff size for mbuf, ret=[%d], dataLen[%lu]", retCode, dataLen);
257 1 : return FsmStatus::FSM_FAILED;
258 : }
259 :
260 : // get mbuf data
261 13 : void* dataPtr = nullptr;
262 13 : retCode = halMbufGetBuffAddr(mbuf, &dataPtr);
263 13 : if ((retCode != static_cast<int32_t>(DRV_ERROR_NONE)) || (dataPtr == nullptr)) {
264 6 : DGW_LOG_ERROR("Failed to get data or data is nullptr, ret[%d].", retCode);
265 6 : return FsmStatus::FSM_FAILED;
266 : }
267 :
268 7 : const size_t totalLen = sizeof(struct buff_iovec) + sizeof(struct iovec_info);
269 7 : std::unique_ptr<char_t[]> vecUniquePtr(new (std::nothrow) char_t[totalLen], std::default_delete<char_t[]>());
270 7 : DGW_CHECK(
271 : (vecUniquePtr != nullptr), FsmStatus::FSM_FAILED, "failed to alloc memory for buffIovec, size[%zu].", totalLen);
272 :
273 7 : buff_iovec* const buffIovec = PtrToPtr<char_t, buff_iovec>(vecUniquePtr.get());
274 7 : buffIovec->context_base = headBuf;
275 7 : buffIovec->context_len = headBufSize;
276 7 : buffIovec->count = 1U;
277 7 : buffIovec->ptr[0U].iovec_base = dataPtr;
278 7 : buffIovec->ptr[0U].len = dataLen;
279 7 : DGW_LOG_INFO("Entity[%s] Begin halQueueEnQueueBuff", entityDesc_.c_str());
280 7 : ret = halQueueEnQueueBuff(deviceId_, id_, buffIovec, -1);
281 7 : DGW_LOG_INFO(
282 : "entity:[%s] halQueueEnQueueBuff queue id:[%u] device id:[%u] result:[%d].", entityDesc_.c_str(), id_,
283 : deviceId_, static_cast<int32_t>(ret));
284 7 : if (ret == DRV_ERROR_QUEUE_FULL) {
285 2 : DGW_LOG_WARN("halQueueEnQueue queue id:[%u] on device:[%u] FULL!!!!.", id_, deviceId_);
286 2 : return FsmStatus::FSM_DEST_FULL;
287 : }
288 5 : if (ret != DRV_ERROR_NONE) {
289 3 : DGW_LOG_ERROR("halQueueEnQueueBuff queue id:[%u] on device:[%u] FAILED!!!!.", id_, deviceId_);
290 3 : return (ret == DRV_ERROR_NOT_EXIST) ? FsmStatus::FSM_ERROR_PENDING : FsmStatus::FSM_FAILED;
291 : }
292 2 : return FsmStatus::FSM_SUCCESS;
293 7 : }
294 :
295 0 : bool ClientEntity::IsDataPeeked() const { return (asyncDataState_ == AsyncDataState::FSM_ASYNC_ENTITY_DONE); }
296 :
297 : } // namespace dgw
|
