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 "simple_entity.h"
12 : #include "bqs_util.h"
13 : #include "dgw_client.h"
14 : #include "profile_manager.h"
15 : namespace dgw {
16 :
17 : namespace {
18 : constexpr uint32_t FLAG_DEVICEID_OFFSET = 32U;
19 : constexpr uint32_t MBUF_ALLOC_ALIGN = 64U;
20 : }
21 :
22 339 : SimpleEntity::SimpleEntity(const EntityMaterial &material, const uint32_t resIndex)
23 339 : : Entity(material, resIndex)
24 : {
25 339 : }
26 :
27 71 : FsmStatus SimpleEntity::Dequeue()
28 : {
29 71 : const auto dequeAllow = AllowDeque();
30 71 : if (dequeAllow != FsmStatus::FSM_SUCCESS) {
31 20 : return dequeAllow;
32 : }
33 :
34 51 : bqs::ProfileManager::GetInstance(resIndex_).AddDequeueNum();
35 51 : FsmStatus dequeRet = DoDequeue();
36 51 : if (dequeRet == FsmStatus::FSM_KEEP_STATE) {
37 10 : return dequeRet;
38 : }
39 41 : if (dequeRet == FsmStatus::FSM_SRC_EMPTY) {
40 13 : statInfo_.dequeueEmptyTimes++;
41 13 : DGW_LOG_DEBUG("QueueId[%u] on device[%u] has been dequeued to empty.", GetQueueId(), deviceId_);
42 13 : return FsmStatus::FSM_FAILED;
43 : }
44 28 : if (dequeRet != FsmStatus::FSM_SUCCESS) {
45 13 : statInfo_.dequeueFailTimes++;
46 13 : bqs::StatisticManager::GetInstance().DataScheduleFailedStat();
47 13 : return dequeRet;
48 : }
49 :
50 15 : DGW_LOG_INFO("Success to dequeue mbuf for entity[%s] in peek state.", entityDesc_.c_str());
51 15 : bqs::StatisticManager::GetInstance().DataDequeueStat();
52 15 : statInfo_.dequeueSuccTimes++;
53 : // PostDoDeque
54 15 : PostDeque();
55 :
56 : // save wait_scheduled mbuf to entity
57 15 : const auto refreshRet = RefreshWithData();
58 15 : if (refreshRet != FsmStatus::FSM_SUCCESS) {
59 1 : (void)halMbufFree(mbuf_);
60 1 : mbuf_ = nullptr;
61 1 : return refreshRet;
62 : }
63 14 : AddScheduleCount();
64 14 : return FsmStatus::FSM_SUCCESS;
65 : }
66 :
67 41 : FsmStatus SimpleEntity::DoDequeue()
68 : {
69 41 : return DoDequeueMbuf(PtrToPtr<Mbuf *, void*>(&mbuf_));
70 : }
71 :
72 48 : FsmStatus SimpleEntity::DoDequeueMbuf(void **mbufPtr) const
73 : {
74 48 : const uint32_t queueId = GetQueueId();
75 48 : const auto ret = halQueueDeQueue(deviceId_, queueId, mbufPtr);
76 48 : if (ret == DRV_ERROR_QUEUE_EMPTY) {
77 17 : DGW_LOG_DEBUG("QueueId[%u] on device[%u] has been dequeued to empty.", queueId, deviceId_);
78 17 : return FsmStatus::FSM_SRC_EMPTY;
79 : }
80 31 : if ((ret != DRV_ERROR_NONE) || (*mbufPtr == nullptr)) {
81 15 : DGW_LOG_ERROR("DeQueue from queueId[%u] on device[%u] failed, ret:[%d]",
82 : queueId, deviceId_, static_cast<int32_t>(ret));
83 15 : if (ret == DRV_ERROR_NOT_EXIST) {
84 2 : return FsmStatus::FSM_ERROR_PENDING;
85 : }
86 13 : return FsmStatus::FSM_FAILED;
87 : }
88 16 : return FsmStatus::FSM_SUCCESS;
89 : }
90 :
91 14 : void SimpleEntity::PostDeque()
92 : {
93 14 : return;
94 : }
95 :
96 16 : FsmStatus SimpleEntity::RefreshWithData()
97 : {
98 16 : if (!needTransId_) {
99 13 : return FsmStatus::FSM_SUCCESS;
100 : }
101 :
102 : // src entity is group entity or dst entities has group entity
103 : // get transId
104 3 : void *headBuf = nullptr;
105 3 : uint32_t headBufSize = 0U;
106 3 : const auto drvRet = halMbufGetPrivInfo(mbuf_, &headBuf, &headBufSize);
107 3 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
108 1 : DGW_LOG_ERROR("halMbufGetPrivInfo from mbuf in entity[%s] failed.", entityDesc_.c_str());
109 1 : return FsmStatus::FSM_FAILED;
110 : }
111 2 : if (headBufSize < MBUF_HEAD_MAX_SIZE) {
112 1 : DGW_LOG_ERROR("mbuf head size:%u in entity[%s] is invalid.", headBufSize, entityDesc_.c_str());
113 1 : return FsmStatus::FSM_FAILED;
114 : }
115 :
116 1 : const uint32_t offset = headBufSize - static_cast<uint32_t>(sizeof(bqs::IdentifyInfo));
117 1 : bqs::IdentifyInfo * const info = PtrToPtr<void, bqs::IdentifyInfo>(ValueToPtr(PtrToValue(headBuf) + offset));
118 1 : transId_ = info->transId;
119 1 : routeLabel_ = info->routeLabel;
120 1 : return FsmStatus::FSM_SUCCESS;
121 : }
122 :
123 2 : FsmStatus SimpleEntity::ResetSrcState()
124 : {
125 2 : return ChangeState(FsmState::FSM_IDLE_STATE);
126 : }
127 :
128 23 : void SimpleEntity::SelectDstEntities(const uint64_t key, std::vector<Entity*> &toPushDstEntities,
129 : std::vector<Entity*> &reprocessDstEntities, std::vector<Entity*> &abnormalDstEntities)
130 : {
131 : (void)key;
132 : (void)reprocessDstEntities;
133 : (void)abnormalDstEntities;
134 23 : toPushDstEntities.emplace_back(this);
135 23 : }
136 :
137 31 : FsmStatus SimpleEntity::SendData(const DataObjPtr dataObj)
138 : {
139 31 : Mbuf *const mbufToPush = PrepareMbufToPush(dataObj);
140 31 : if (mbufToPush == nullptr) {
141 7 : return FsmStatus::FSM_FAILED;
142 : }
143 24 : const auto sendRet = DoSendData(mbufToPush);
144 24 : if (sendRet == FsmStatus::FSM_SUCCESS) {
145 10 : statInfo_.enqueueSuccTimes++;
146 14 : } else if (sendRet != FsmStatus::FSM_KEEP_STATE) {
147 4 : Mbuf * const mbuf = const_cast<Mbuf *>(dataObj->GetMbuf());
148 4 : if (mbufToPush != mbuf) {
149 1 : (void)halMbufFree(mbufToPush);
150 : }
151 : }
152 24 : return sendRet;
153 : }
154 :
155 21 : Mbuf *SimpleEntity::PrepareMbufToPush(DataObjPtr dataObj) const
156 : {
157 21 : Entity *const sendEntity = dataObj->GetSendEntity();
158 21 : Mbuf * const mbuf = const_cast<Mbuf *>(dataObj->GetMbuf());
159 :
160 21 : if ((sendEntity->GetMbufQueueType() != bqs::CLIENT_Q) && (GetDeviceId() != sendEntity->GetMbufDeviceId())) {
161 2 : return SdmaCopy(mbuf);
162 : }
163 19 : if (dataObj->GetRecvEntitySize() > 1U) {
164 4 : Mbuf *copyMbuf = nullptr;
165 4 : auto &profileInstance = bqs::ProfileManager::GetInstance(resIndex_);
166 4 : const uint64_t copyBegin = profileInstance.GetCpuTick();
167 4 : const int32_t drvRet = halMbufCopyRef(mbuf, ©Mbuf);
168 4 : profileInstance.AddCopyTotalCost(profileInstance.GetCpuTick() - copyBegin);
169 4 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (copyMbuf == nullptr)) {
170 2 : BQS_LOG_RUN_WARN("MbufCopy failed when from queue[%u] to queue[%u] in device[%u], error=[%d].",
171 : sendEntity->GetId(), GetId(), GetDeviceId(), drvRet);
172 2 : bqs::StatisticManager::GetInstance().DataScheduleFailedStat();
173 : }
174 4 : return copyMbuf;
175 : }
176 :
177 15 : dataObj->MaintainMbuf();
178 15 : return mbuf;
179 : }
180 :
181 8 : FsmStatus SimpleEntity::DoSendData(Mbuf *const mbuf)
182 : {
183 8 : const auto ret = halQueueEnQueue(deviceId_, id_, mbuf);
184 8 : DGW_LOG_INFO("%s halQueueEnQueue queue id:[%u] device id:[%u] result:[%d].",
185 : entityDesc_.c_str(), id_, deviceId_, static_cast<int32_t>(ret));
186 8 : if (ret == DRV_ERROR_NONE) {
187 5 : bqs::StatisticManager::GetInstance().DataQueueEnqueueSuccStat();
188 5 : return FsmStatus::FSM_SUCCESS;
189 : }
190 :
191 3 : if (ret == DRV_ERROR_QUEUE_FULL) {
192 1 : bqs::StatisticManager::GetInstance().DataQueueEnqueueFullStat();
193 1 : DGW_LOG_INFO("[%s] halQueueEnQueue queue id:[%u] FULL!!!!.", GetTypeDesc().c_str(), id_);
194 1 : return FsmStatus::FSM_DEST_FULL;
195 : }
196 :
197 2 : bqs::StatisticManager::GetInstance().DataQueueEnqueueFailStat();
198 2 : DGW_LOG_ERROR("[%s] halQueueEnQueue queue id:[%u] FAILED!!!!.", GetTypeDesc().c_str(), id_);
199 2 : if (ret == DRV_ERROR_NOT_EXIST) {
200 1 : return FsmStatus::FSM_ERROR_PENDING;
201 : }
202 1 : return FsmStatus::FSM_FAILED;
203 : }
204 :
205 11 : Mbuf* SimpleEntity::SdmaCopy(Mbuf * const mbuf) const
206 : {
207 : // 调用sdma拷贝
208 11 : uint64_t desBufLen = 0;
209 11 : auto retHal = halMbufGetDataLen(mbuf, &desBufLen);
210 11 : if (retHal != static_cast<int32_t>(DRV_ERROR_NONE)) {
211 1 : BQS_LOG_ERROR("halMbufGetDataLen error ret=[%d]", retHal);
212 1 : return nullptr;
213 : }
214 10 : void *srcDataBuf = nullptr;
215 10 : auto drvRet = halMbufGetBuffAddr(mbuf, &srcDataBuf);
216 10 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (srcDataBuf == nullptr)) {
217 1 : DGW_LOG_ERROR("Fail to get buff addr for mbuf, ret=[%d]", drvRet);
218 1 : return nullptr;
219 : }
220 9 : void *srcHeadBuf = nullptr;
221 9 : uint32_t srcHeadBufSize = 0U;
222 9 : drvRet = halMbufGetPrivInfo(mbuf, &srcHeadBuf, &srcHeadBufSize);
223 9 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
224 1 : DGW_LOG_ERROR("halMbufGetPrivInfo fail, ret=[%d]", drvRet);
225 1 : return nullptr;
226 : }
227 :
228 8 : Mbuf *mbufPtr = AllocateMbuf(desBufLen);
229 8 : if (mbufPtr == nullptr) {
230 2 : BQS_LOG_ERROR("Allocate Mbuf fail.");
231 2 : return nullptr;
232 : }
233 :
234 6 : if (SdmaCopyData(srcDataBuf, desBufLen, mbufPtr) != FsmStatus::FSM_SUCCESS) {
235 1 : BQS_LOG_ERROR("Sdma copy data fail.");
236 1 : (void)halMbufFree(mbufPtr);
237 1 : return nullptr;
238 : }
239 :
240 5 : if (SdmaCopyHead(srcHeadBuf, srcHeadBufSize, mbufPtr) != FsmStatus::FSM_SUCCESS) {
241 2 : BQS_LOG_ERROR("Sdma copy head fail.");
242 2 : (void)halMbufFree(mbufPtr);
243 2 : return nullptr;
244 : }
245 :
246 3 : BQS_LOG_INFO("Success to sdma copy.");
247 3 : return mbufPtr;
248 : }
249 :
250 8 : Mbuf* SimpleEntity::AllocateMbuf(const uint64_t desBufLen) const
251 : {
252 8 : uint64_t flag = (static_cast<uint64_t>(deviceId_) << FLAG_DEVICEID_OFFSET) |
253 : static_cast<uint64_t>(BUFF_SP_NORMAL);
254 8 : int32_t memGroupId = 0;
255 8 : Mbuf *mbufPtr = nullptr;
256 : // alignSize use 64
257 8 : auto drvRet = halMbufAllocEx(desBufLen, MBUF_ALLOC_ALIGN, flag, memGroupId, &mbufPtr);
258 8 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) || mbufPtr == nullptr) {
259 1 : BQS_LOG_ERROR("halMbufAllocEx failed, drvRet=%d, dataSize=%lu, flag=%lu, groupId=%d.", drvRet, desBufLen,
260 : flag, memGroupId);
261 1 : return nullptr;
262 : }
263 7 : drvRet = halMbufSetDataLen(mbufPtr, desBufLen);
264 7 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
265 1 : BQS_LOG_ERROR("halMbufSetDataLen failed, ret=%d.", drvRet);
266 1 : (void)halMbufFree(mbufPtr);
267 1 : return nullptr;
268 : }
269 6 : return mbufPtr;
270 : }
271 :
272 7 : FsmStatus SimpleEntity::SdmaCopyData(void *const srcDataBuf, const uint64_t desBufLen, Mbuf *const mbufPtr) const
273 : {
274 7 : void *dstDataBuf = nullptr;
275 7 : auto drvRet = halMbufGetBuffAddr(mbufPtr, &dstDataBuf);
276 7 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (dstDataBuf == nullptr)) {
277 1 : DGW_LOG_ERROR("Fail to get buff addr for mbuf, ret=[%d].", drvRet);
278 1 : return FsmStatus::FSM_FAILED;
279 : }
280 6 : drvRet = halSdmaCopy(PtrToValue(dstDataBuf), desBufLen, PtrToValue(srcDataBuf), desBufLen);
281 6 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
282 1 : DGW_LOG_ERROR("halSdmaCopy error ret:%d.", drvRet);
283 1 : return FsmStatus::FSM_FAILED;
284 : }
285 5 : return FsmStatus::FSM_SUCCESS;
286 : }
287 :
288 6 : FsmStatus SimpleEntity::SdmaCopyHead(void *const srcHeadBuf, const uint32_t srcHeadBufSize, Mbuf *const mbufPtr) const
289 : {
290 6 : void *dstHeadBuf = nullptr;
291 6 : uint32_t dstHeadBufSize = 0U;
292 6 : auto drvRet = halMbufGetPrivInfo(mbufPtr, &dstHeadBuf, &dstHeadBufSize);
293 6 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
294 1 : DGW_LOG_ERROR("halMbufGetPrivInfo fail, ret=[%d]", drvRet);
295 1 : return FsmStatus::FSM_FAILED;
296 : }
297 5 : if ((srcHeadBuf == nullptr) || (dstHeadBuf == nullptr) || (dstHeadBufSize < srcHeadBufSize)) {
298 1 : DGW_LOG_ERROR("dstHeadBufSize[%u] vs srcHeadBufSize[%u]", dstHeadBufSize, srcHeadBufSize);
299 1 : return FsmStatus::FSM_FAILED;
300 : }
301 4 : drvRet = halSdmaCopy(PtrToValue(dstHeadBuf), dstHeadBufSize, PtrToValue(srcHeadBuf), srcHeadBufSize);
302 4 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
303 1 : DGW_LOG_ERROR("halSdmaCopy error ret:%d", drvRet);
304 1 : return FsmStatus::FSM_FAILED;
305 : }
306 3 : return FsmStatus::FSM_SUCCESS;
307 : }
308 :
309 30 : bqs::SubscribeManager *SimpleEntity::GetSubscriber() const
310 : {
311 : bqs::SubscribeManager * const subscribeManager =
312 30 : bqs::Subscribers::GetInstance().GetSubscribeManager(resIndex_, deviceId_);
313 30 : if (subscribeManager == nullptr) {
314 11 : DGW_LOG_ERROR("Failed to find subscribeManager for resIndex: %u, device: %u, queueType: %d",
315 : resIndex_, deviceId_, static_cast<int32_t>(queueType_));
316 : }
317 30 : return subscribeManager;
318 : }
319 :
320 5 : FsmStatus SimpleEntity::PauseSubscribe(const Entity &fullEntity)
321 : {
322 5 : if (subscribeStatus_ == SubscribeStatus::SUBSCRIBE_PAUSE) {
323 1 : DGW_LOG_WARN("No need to pause subscribe for entity[%s].", entityDesc_.c_str());
324 1 : return FsmStatus::FSM_SUCCESS;
325 : }
326 4 : DGW_LOG_INFO("[FSM] Pause subscribe src entity[%s] because dst entity[id:%u, type:%s] full.",
327 : entityDesc_.c_str(), fullEntity.GetId(), fullEntity.GetTypeDesc().c_str());
328 4 : subscribeStatus_ = SubscribeStatus::SUBSCRIBE_PAUSE;
329 :
330 4 : const auto subscriber = GetSubscriber();
331 4 : if (subscriber == nullptr) {
332 1 : return FsmStatus::FSM_FAILED;
333 : }
334 3 : subscriber->PauseSubscribe(GetQueueId(), fullEntity.GetId(), false);
335 3 : return FsmStatus::FSM_SUCCESS;
336 : }
337 :
338 5 : FsmStatus SimpleEntity::ResumeSubscribe(const Entity ¬FullEntity)
339 : {
340 5 : if (subscribeStatus_ == SubscribeStatus::SUBSCRIBE_RESUME) {
341 1 : DGW_LOG_WARN("No need to resume subscribe for entity[%s].", entityDesc_.c_str());
342 1 : return FsmStatus::FSM_SUCCESS;
343 : }
344 4 : DGW_LOG_INFO("[FSM] Resume subscribe src entity[%s] because dst entity[id:%u, type:%s] not full.",
345 : entityDesc_.c_str(), notFullEntity.GetId(), notFullEntity.GetTypeDesc().c_str());
346 4 : subscribeStatus_ = SubscribeStatus::SUBSCRIBE_RESUME;
347 4 : const auto subscriber = GetSubscriber();
348 4 : if (subscriber == nullptr) {
349 3 : return FsmStatus::FSM_FAILED;
350 : }
351 1 : subscriber->ResumeSubscribe(GetQueueId(), notFullEntity.GetId());
352 1 : return FsmStatus::FSM_SUCCESS;
353 : }
354 :
355 8 : FsmStatus SimpleEntity::ClearQueue()
356 : {
357 8 : DGW_LOG_INFO("Entity[%s] clear queue", entityDesc_.c_str());
358 : do {
359 9 : void *mbuf = nullptr;
360 9 : const auto dequeRet = DoDequeueMbuf(&mbuf);
361 9 : if (dequeRet == FsmStatus::FSM_SRC_EMPTY) {
362 5 : break;
363 : }
364 :
365 4 : if ((dequeRet != FsmStatus::FSM_SUCCESS) || (mbuf == nullptr)) {
366 3 : DGW_LOG_ERROR("DeQueue from queueId[%u] in device[%u] failed, ret:[%d]",
367 : GetQueueId(), deviceId_, static_cast<int32_t>(dequeRet));
368 3 : return FsmStatus::FSM_FAILED;
369 : }
370 1 : (void) halMbufFree(PtrToPtr<void, Mbuf>(mbuf));
371 1 : } while (true);
372 :
373 7 : for (auto sendDataObj : sendDataObjs_) {
374 4 : for (auto recvEntity : sendDataObj->GetRecvEntities()) {
375 2 : auto &recvObjs = recvEntity->GetRecvDataObjs();
376 5 : for (auto iter = recvObjs.begin(); iter != recvObjs.end();) {
377 3 : if (Equal((*iter)->GetSendEntity())) {
378 2 : auto tempIter = iter;
379 2 : iter++;
380 2 : recvObjs.erase(tempIter);
381 : } else {
382 1 : iter++;
383 : }
384 : }
385 : }
386 2 : }
387 5 : sendDataObjs_.clear();
388 5 : DGW_LOG_INFO("Entity[%s] clear queue finish", entityDesc_.c_str());
389 5 : return FsmStatus::FSM_SUCCESS;
390 : }
391 :
392 28 : bool SimpleEntity::IsDataPeeked() const
393 : {
394 28 : return (curState_ == FsmState::FSM_PEEK_STATE);
395 : }
396 :
397 232 : FsmStatus SimpleEntity::Uninit()
398 : {
399 232 : DGW_LOG_RUN_INFO("[%s] has dequeued[%lu], enqueued[%lu].", entityDesc_.c_str(), statInfo_.dequeueSuccTimes,
400 : statInfo_.enqueueSuccTimes);
401 232 : return FsmStatus::FSM_SUCCESS;
402 : }
403 :
404 : }
|
