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 "operator_kernel_dequeue_base.h"
12 :
13 : #include "aicpusd_profiler.h"
14 : #include "aicpusd_drv_manager.h"
15 : #include "aicpusd_model_execute.h"
16 : #include "aicpusd_resource_manager.h"
17 : #include "operator_kernel_common.h"
18 :
19 :
20 : namespace AicpuSchedule {
21 74 : int32_t OperatorKernelDequeueBase::DequeueTask(BufEnQueueInfo &bufInfo, const RunContext &taskContext,
22 : const bool needPending) const
23 : {
24 74 : void *taskMBuf = nullptr;
25 74 : Mbuf ** const mBufPptr = reinterpret_cast<Mbuf **>(static_cast<uintptr_t>(bufInfo.mBufPtr));
26 74 : if (mBufPptr == nullptr) {
27 1 : aicpusd_err("param mBufPptr is null.");
28 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
29 : }
30 73 : const auto queueId = bufInfo.queueID;
31 73 : const auto streamId = taskContext.streamId;
32 73 : const auto deviceId = AicpuDrvManager::GetInstance().GetDeviceId();
33 : // clear unused eventState
34 73 : EventWaitManager::QueueNotEmptyWaitManager().ResetEventState(static_cast<size_t>(queueId));
35 73 : int32_t ret = AICPU_SCHEDULE_OK;
36 73 : g_aicpuProfiler.SetQueueId(queueId);
37 : do {
38 73 : ret = halQueueDeQueue(deviceId, queueId, &taskMBuf);
39 73 : if (ret == DRV_ERROR_NONE) {
40 : // guard taskMBuf
41 67 : const auto guardRet = BufManager::GetInstance().GuardBuf(reinterpret_cast<Mbuf *>(taskMBuf),
42 67 : taskContext.modelId);
43 67 : if (guardRet != AICPU_SCHEDULE_OK) {
44 1 : aicpusd_err("BufManager guard dequeue failed, modelId[%u], ret[%d].", taskContext.modelId, guardRet);
45 1 : return guardRet;
46 : }
47 66 : break;
48 : }
49 6 : if (ret == DRV_ERROR_QUEUE_EMPTY) {
50 4 : aicpusd_run_info("Dequeue empty on queueId[%u], ret[%d].", queueId, ret);
51 4 : if (needPending) {
52 2 : bool needWait = false;
53 : // if exist NotEmptyEvent, needWait return true and not record wait stream
54 2 : EventWaitManager::QueueNotEmptyWaitManager().
55 2 : WaitEvent(static_cast<size_t>(queueId), streamId, needWait);
56 2 : if (needWait) {
57 2 : aicpusd_run_info("%s pending, queueId:%u, streamId:%u.", __func__, queueId, streamId);
58 2 : bool *pending = const_cast<bool *>(&taskContext.pending);
59 2 : *pending = true;
60 2 : return AICPU_SCHEDULE_OK;
61 : }
62 : } else {
63 2 : aicpusd_info("no need pending");
64 2 : return AICPU_SCHEDULE_OK;
65 : }
66 : } else {
67 2 : aicpusd_err("Failed to dequeue on queueId[%u], ret[%d].", queueId, ret);
68 2 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
69 : }
70 0 : } while (true);
71 66 : *mBufPptr = PtrToPtr<void, Mbuf>(taskMBuf);
72 :
73 66 : uint32_t headSize = 0U;
74 66 : void *headBuf = nullptr;
75 66 : const auto drvRet = halMbufGetPrivInfo(*mBufPptr, &headBuf, &headSize);
76 66 : if (drvRet != DRV_ERROR_NONE) {
77 1 : aicpusd_err("Failed to get head info in input information, ret[%d].", drvRet);
78 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
79 : }
80 :
81 65 : (void)ProcessMbufHeadInDequeueTask(taskContext.modelId, headBuf, headSize);
82 65 : (void)SetModelEndOfSequence(taskContext.modelId, headBuf, headSize);
83 :
84 65 : g_aicpuProfiler.SetMbufHead(headBuf);
85 65 : OperatorKernelCommon::TraceQueueData(taskContext, headBuf, headSize, "Dequeued");
86 65 : return AICPU_SCHEDULE_OK;
87 : }
88 :
89 66 : void OperatorKernelDequeueBase::ProcessMbufHeadInDequeueTask(const uint32_t modelId, void * const headBuf,
90 : const uint32_t headSize) const
91 : {
92 66 : if ((headBuf == nullptr) || (static_cast<size_t>(headSize) < sizeof(MbufHeadMsg))) {
93 56 : aicpusd_debug("Skip process mbuf head msg. modelId=%u, headSize=%u, baseSize=%lu",
94 : modelId, headSize, sizeof(MbufHeadMsg));
95 56 : return;
96 : }
97 :
98 10 : MbufHeadMsg * const msg = PtrToPtr<uint8_t, MbufHeadMsg>(PtrAdd<uint8_t>(PtrToPtr<void, uint8_t>(headBuf),
99 10 : MBUF_HEAD_MAX_SIZE, static_cast<size_t>(headSize) - sizeof(MbufHeadMsg)));
100 :
101 10 : SetModelNullData(modelId, msg);
102 10 : SetModelRetCode(modelId, msg);
103 10 : SetMbufStepId(modelId, msg);
104 :
105 10 : return;
106 : }
107 :
108 66 : void OperatorKernelDequeueBase::SetModelEndOfSequence(const uint32_t modelId, void * const headBuf,
109 : const uint32_t headSize) const
110 : {
111 66 : const auto model = AicpuModelManager::GetInstance().GetModel(modelId);
112 66 : if (model == nullptr) {
113 1 : return;
114 : }
115 :
116 65 : if ((headBuf != nullptr) && (headSize > MBUF_HEAD_END_OF_SEQUENCE_POS)) {
117 1 : const uint8_t * const endOfSequence = PtrAdd<uint8_t>(PtrToPtr<void, uint8_t>(headBuf), MBUF_HEAD_MAX_SIZE,
118 : static_cast<size_t>(MBUF_HEAD_END_OF_SEQUENCE_POS));
119 1 : if (*endOfSequence == END_OF_SEQUENCE_FLAG) {
120 1 : model->SetModelEndOfSequence();
121 1 : aicpusd_info("Set model end of sequence success.");
122 : }
123 : }
124 : }
125 :
126 12 : void OperatorKernelDequeueBase::SetModelNullData(const uint32_t modelId, const MbufHeadMsg * const headMsg) const
127 : {
128 12 : if (!FeatureCtrl::ShouldSetModuleNullData()) {
129 0 : aicpusd_info("skip SetModelNullData");
130 0 : return;
131 : }
132 :
133 12 : aicpusd_info("Mbuf head msg, flags=%u, dataFlag=%u.", headMsg->flags, headMsg->dataFlag);
134 12 : if ((headMsg->dataFlag & MBUF_HEAD_DATA_FLAG_MASK) == static_cast<uint8_t>(DataFlag::DFLOW_NULL_DATA_FLAG)) {
135 2 : const auto model = AicpuModelManager::GetInstance().GetModel(modelId);
136 2 : if (model == nullptr) {
137 1 : aicpusd_debug("Skip process mbuf head msg, model is null. modelId=%u", modelId);
138 1 : return;
139 : }
140 :
141 1 : model->SetNullDataFlag(true);
142 : }
143 :
144 11 : return;
145 : }
146 :
147 13 : void OperatorKernelDequeueBase::SetModelRetCode(const uint32_t modelId, const MbufHeadMsg * const headMsg) const
148 : {
149 13 : const auto model = AicpuModelManager::GetInstance().GetModel(modelId);
150 13 : if (model == nullptr) {
151 1 : aicpusd_debug("Skip process mbuf head msg, model is null. modelId=%u", modelId);
152 1 : return;
153 : }
154 :
155 12 : if (!model->AbnormalEnabled()) {
156 : // The input mbuf may not be initialized. Set retcode when abnormal enabled
157 10 : return;
158 : }
159 :
160 2 : if ((headMsg->retCode != 0) && (model->GetModelRetCode() == 0)) {
161 1 : model->SetModelRetCode(headMsg->retCode);
162 1 : aicpusd_info("Set model ret code success, modelId=%u, retCode=%d", model->GetId(), headMsg->retCode);
163 : }
164 :
165 2 : return;
166 : }
167 :
168 13 : void OperatorKernelDequeueBase::SetMbufStepId(const uint32_t modelId, MbufHeadMsg * const headMsg) const
169 : {
170 13 : const auto model = AicpuModelManager::GetInstance().GetModel(modelId);
171 13 : if (model == nullptr) {
172 1 : aicpusd_debug("Skip process mbuf head msg, model is null. modelId=%u", modelId);
173 1 : return;
174 : }
175 :
176 12 : const StepIdInfo info = model->GetStepIdInfo();
177 12 : if (model->GetHeadNodeFlag()) {
178 : // head node set step id to mbuf head
179 1 : aicpusd_debug("set mbuf head. modelId=%u, before=%u, after=%u", modelId, headMsg->stepId, info.stepId);
180 1 : headMsg->stepId = info.stepId;
181 : } else {
182 : // get step id from mbuf head and refresh global step id
183 11 : if (info.stepIdAddr != nullptr) {
184 1 : aicpusd_debug("get mbuf head. modelId=%u, before=%lu, after=%u",
185 : modelId, *info.stepIdAddr, headMsg->stepId);
186 1 : if (*info.stepIdAddr <= headMsg->stepId) {
187 1 : *info.stepIdAddr = headMsg->stepId;
188 : }
189 : }
190 : }
191 :
192 12 : return;
193 : }
194 :
195 3 : int32_t OperatorKernelDequeueBase::AlignTimestamp(BatchDequeueInfo &batchDeqInfo, const RunContext &taskContext,
196 : uint32_t &maxAlignTimestamp, uint32_t &minAlignTimestamp,
197 : uint32_t &minTimestampIndex)
198 : {
199 3 : uint32_t minTimestamp = UINT32_MAX;
200 4 : for (uint32_t i = 0U; i < batchDeqInfo.inputNums; ++i) {
201 : // mbuf and mbufHead has been checked in dequeue, not nullptr
202 3 : Mbuf ** const mbufpPtr = PtrToPtr<void, Mbuf*>(ValueToPtr(batchDeqInfo.mbufAddrs[i]));
203 3 : uint32_t headSize = 0U;
204 3 : void *headBuf = nullptr;
205 3 : (void) halMbufGetPrivInfo(*mbufpPtr, &headBuf, &headSize);
206 3 : if (headBuf == nullptr || static_cast<size_t>(headSize) < sizeof(MbufHeadMsg)) {
207 2 : aicpusd_err("Mbuf head is error headSize[%u]", headSize);
208 2 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
209 : }
210 1 : const MbufHeadMsg * const curHeadInfo = PtrToPtr<uint8_t, MbufHeadMsg>(PtrAdd<uint8_t>(
211 1 : PtrToPtr<void, uint8_t>(headBuf), MBUF_HEAD_MAX_SIZE, static_cast<size_t>(headSize) - sizeof(MbufHeadMsg)));
212 1 : if (curHeadInfo->startTime != curHeadInfo->endTime) {
213 0 : aicpusd_err("Mbuf head starttime[%llu] and endtime[%llu] not equal",
214 : curHeadInfo->startTime, curHeadInfo->endTime);
215 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
216 : }
217 1 : const uint32_t curTimeStamp = static_cast<uint32_t>(curHeadInfo->startTime);
218 1 : if (curTimeStamp < batchDeqInfo.alignOffsets[i]) {
219 0 : aicpusd_err("Mbuf head timestamp[%u] < alignOffset[%u], modelId[%u], streamId[%u]",
220 : curTimeStamp, batchDeqInfo.alignOffsets[i], taskContext.modelId, taskContext.streamId);
221 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
222 : }
223 1 : const uint32_t timeAlign = curTimeStamp - batchDeqInfo.alignOffsets[i];
224 1 : if (curTimeStamp < minTimestamp) {
225 1 : minTimestamp = curTimeStamp;
226 1 : minTimestampIndex = i;
227 : }
228 1 : if (timeAlign > maxAlignTimestamp) {
229 1 : maxAlignTimestamp = timeAlign;
230 : }
231 1 : if (timeAlign < minAlignTimestamp) {
232 1 : minAlignTimestamp = timeAlign;
233 : }
234 : }
235 1 : return AICPU_SCHEDULE_OK;
236 : }
237 :
238 : } // namespace AicpuSchedule
|
