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