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