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 : namespace AicpuSchedule {
20 74 : int32_t OperatorKernelDequeueBase::DequeueTask(
21 : BufEnQueueInfo& bufInfo, const RunContext& taskContext, const bool needPending) const
22 : {
23 74 : void* taskMBuf = nullptr;
24 74 : Mbuf** const mBufPptr = reinterpret_cast<Mbuf**>(static_cast<uintptr_t>(bufInfo.mBufPtr));
25 74 : if (mBufPptr == nullptr) {
26 1 : aicpusd_err("param mBufPptr is null.");
27 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
28 : }
29 73 : const auto queueId = bufInfo.queueID;
30 73 : const auto streamId = taskContext.streamId;
31 73 : const auto deviceId = AicpuDrvManager::GetInstance().GetDeviceId();
32 : // clear unused eventState
33 73 : EventWaitManager::QueueNotEmptyWaitManager().ResetEventState(static_cast<size_t>(queueId));
34 73 : int32_t ret = AICPU_SCHEDULE_OK;
35 73 : g_aicpuProfiler.SetQueueId(queueId);
36 : do {
37 73 : ret = halQueueDeQueue(deviceId, queueId, &taskMBuf);
38 73 : if (ret == DRV_ERROR_NONE) {
39 : // guard taskMBuf
40 : const auto guardRet =
41 67 : BufManager::GetInstance().GuardBuf(reinterpret_cast<Mbuf*>(taskMBuf), taskContext.modelId);
42 67 : if (guardRet != AICPU_SCHEDULE_OK) {
43 1 : aicpusd_err("BufManager guard dequeue failed, modelId[%u], ret[%d].", taskContext.modelId, guardRet);
44 1 : return guardRet;
45 : }
46 66 : break;
47 : }
48 6 : if (ret == DRV_ERROR_QUEUE_EMPTY) {
49 4 : aicpusd_run_info("Dequeue empty on queueId[%u], ret[%d].", queueId, ret);
50 4 : if (needPending) {
51 2 : bool needWait = false;
52 : // if exist NotEmptyEvent, needWait return true and not record wait stream
53 2 : EventWaitManager::QueueNotEmptyWaitManager().WaitEvent(
54 : static_cast<size_t>(queueId), streamId, needWait);
55 2 : if (needWait) {
56 2 : aicpusd_run_info("%s pending, queueId:%u, streamId:%u.", __func__, queueId, streamId);
57 2 : bool* pending = const_cast<bool*>(&taskContext.pending);
58 2 : *pending = true;
59 2 : return AICPU_SCHEDULE_OK;
60 : }
61 : } else {
62 2 : aicpusd_info("no need pending");
63 2 : return AICPU_SCHEDULE_OK;
64 : }
65 : } else {
66 2 : aicpusd_err("Failed to dequeue on queueId[%u], ret[%d].", queueId, ret);
67 2 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
68 : }
69 0 : } while (true);
70 66 : *mBufPptr = PtrToPtr<void, Mbuf>(taskMBuf);
71 :
72 66 : uint32_t headSize = 0U;
73 66 : void* headBuf = nullptr;
74 66 : const auto drvRet = halMbufGetPrivInfo(*mBufPptr, &headBuf, &headSize);
75 66 : if (drvRet != DRV_ERROR_NONE) {
76 1 : aicpusd_err("Failed to get head info in input information, ret[%d].", drvRet);
77 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
78 : }
79 :
80 65 : (void)ProcessMbufHeadInDequeueTask(taskContext.modelId, headBuf, headSize);
81 65 : (void)SetModelEndOfSequence(taskContext.modelId, headBuf, headSize);
82 :
83 65 : g_aicpuProfiler.SetMbufHead(headBuf);
84 65 : OperatorKernelCommon::TraceQueueData(taskContext, headBuf, headSize, "Dequeued");
85 65 : return AICPU_SCHEDULE_OK;
86 : }
87 :
88 66 : void OperatorKernelDequeueBase::ProcessMbufHeadInDequeueTask(
89 : const uint32_t modelId, void* const headBuf, const uint32_t headSize) const
90 : {
91 66 : if ((headBuf == nullptr) || (static_cast<size_t>(headSize) < sizeof(MbufHeadMsg))) {
92 56 : aicpusd_debug(
93 : "Skip process mbuf head msg. modelId=%u, headSize=%u, baseSize=%lu", modelId, headSize,
94 : sizeof(MbufHeadMsg));
95 56 : return;
96 : }
97 :
98 10 : MbufHeadMsg* const msg = PtrToPtr<uint8_t, MbufHeadMsg>(PtrAdd<uint8_t>(
99 10 : PtrToPtr<void, uint8_t>(headBuf), 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(
109 : const uint32_t modelId, void* const headBuf, 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>(
118 : PtrToPtr<void, uint8_t>(headBuf), MBUF_HEAD_MAX_SIZE, 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(
185 : "get mbuf head. modelId=%u, before=%lu, after=%u", 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(
196 : BatchDequeueInfo& batchDeqInfo, const RunContext& taskContext, uint32_t& maxAlignTimestamp,
197 : uint32_t& minAlignTimestamp, 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(
214 : "Mbuf head starttime[%llu] and endtime[%llu] not equal", 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(
220 : "Mbuf head timestamp[%u] < alignOffset[%u], modelId[%u], streamId[%u]", curTimeStamp,
221 : batchDeqInfo.alignOffsets[i], taskContext.modelId, taskContext.streamId);
222 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
223 : }
224 1 : const uint32_t timeAlign = curTimeStamp - batchDeqInfo.alignOffsets[i];
225 1 : if (curTimeStamp < minTimestamp) {
226 1 : minTimestamp = curTimeStamp;
227 1 : minTimestampIndex = i;
228 : }
229 1 : if (timeAlign > maxAlignTimestamp) {
230 1 : maxAlignTimestamp = timeAlign;
231 : }
232 1 : if (timeAlign < minAlignTimestamp) {
233 1 : minAlignTimestamp = timeAlign;
234 : }
235 : }
236 1 : return AICPU_SCHEDULE_OK;
237 : }
238 :
239 : } // namespace AicpuSchedule
|
