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 "aicpusd_model_execute.h"
12 :
13 : #include <algorithm>
14 : #include <stack>
15 : #include <cstring>
16 :
17 : #include "aicpusd_resource_manager.h"
18 : #include "aicpusd_drv_manager.h"
19 : #include "aicpusd_profiler.h"
20 : #include "aicpusd_monitor.h"
21 : #include "aicpusd_msg_send.h"
22 : #include "aicpusd_info.h"
23 : #include "aicpu_event_struct.h"
24 : #include "type_def.h"
25 : #include "profiling_adp.h"
26 : #include "aicpusd_event_process.h"
27 : #include "aicpusd_context.h"
28 : #include "dump_task.h"
29 : #include "operator_kernel_register.h"
30 : #include "aicpusd_model_err_process.h"
31 :
32 : namespace {
33 : // model operate values
34 : const std::string OPERATE_VALUE[static_cast<int32_t>(AicpuSchedule::AicpuModelOperate::MODEL_OPERATE_MAX)] = {
35 : "LOAD", "EXECUTE", "ABORT", "TASK_REPORT", "END_GRAPH", "RUN_TASK", "DESTROY", "STOP", "RESTART", "CLEAR_INPUT"};
36 :
37 : // model status values
38 : const std::string STATUS_VALUE[static_cast<int32_t>(AicpuSchedule::AicpuModelStatus::MODEL_STATUS_MAX)] = {
39 : "UNINIT", "IDLE", "LOADING", "RUNNING", "ERROR", "ABORT", "STOPPED"};
40 : constexpr uint32_t HCCL_QUERY_PREPARE_INTERVAL = 20000U; // us
41 : constexpr uint32_t HCCL_QUERY_LOG_INTERVAL = 500U;
42 : constexpr char_t const* REPORT_STATUS_TASK_NAME = "modelReportStatus";
43 : constexpr int32_t TIMES_MS_TO_NS = 1000;
44 : constexpr uint32_t ASYNC_TASK_INTERVAL = 2000U; // us
45 : constexpr uint32_t AICPU_STREAM_TASK_DUMP_MASK = 0x01U;
46 : } // namespace
47 :
48 : namespace AicpuSchedule {
49 : constexpr int32_t INVALID_TASK_INDEX = -1;
50 : // attention: if change AicpuModelStatus or AicpuModelOperate define, pls update this table together.
51 : const bool AicpuModel::modelOperatePermission[static_cast<int32_t>(
52 : AicpuModelStatus::MODEL_STATUS_MAX)][static_cast<int32_t>(AicpuModelOperate::MODEL_OPERATE_MAX)] = {
53 : // load, execute, abort, taskReport, endGraph, runTask, destroy, stop, restart, clearInput
54 : {true, false, false, false, false, false, true, true, false, false}, // uninit
55 : {false, true, false, false, true, true, true, true, false, false}, // idle
56 : {false, true, false, false, false, false, true, true, false, false}, // loading
57 : {false, false, true, true, true, true, true, true, false, false}, // running
58 : {false, true, true, false, false, false, true, true, false, false}, // error
59 : {false, true, false, false, false, false, true, true, false, false}, // abort
60 : {false, false, false, false, false, false, true, false, true, true} // stopped
61 : };
62 :
63 : // attention: if change AicpuModelStatus or AicpuModelOperate define, pls update this table together.
64 : const AicpuModelStatus AicpuModel::operateNextStatus[static_cast<int32_t>(AicpuModelOperate::MODEL_OPERATE_MAX)] = {
65 : AicpuModelStatus::MODEL_STATUS_LOADING, // load
66 : AicpuModelStatus::MODEL_STATUS_RUNNING, // execute
67 : AicpuModelStatus::MODEL_STATUS_ABORT, // abort
68 : AicpuModelStatus::MODEL_STATUS_ERROR, // taskReport
69 : AicpuModelStatus::MODEL_STATUS_IDLE, // endGraph
70 : AicpuModelStatus::MODEL_STATUS_MAX, // runTask
71 : AicpuModelStatus::MODEL_STATUS_UNINIT, // destroy
72 : AicpuModelStatus::MODEL_STATUS_STOPPED, // stop
73 : AicpuModelStatus::MODEL_STATUS_IDLE, // restart
74 : AicpuModelStatus::MODEL_STATUS_STOPPED, // clearInput
75 : };
76 :
77 17 : void AicpuStream::InitAicpuStream(const uint32_t streamId, const std::vector<const AicpuTaskInfo*>& tasks)
78 : {
79 17 : const std::unique_lock<std::mutex> lockForStream(mutexForStream_);
80 17 : streamId_ = streamId;
81 17 : nextTaskIndex_ = 0LU;
82 17 : (void)std::transform(
83 17 : tasks.begin(), tasks.end(), std::back_inserter(tasks_),
84 97 : [](const AicpuTaskInfo* const taskPtr) { return *taskPtr; });
85 17 : }
86 :
87 7 : void AicpuStream::ResetToStart()
88 : {
89 7 : const std::unique_lock<std::mutex> lockForStream(mutexForStream_);
90 7 : nextTaskIndex_ = 0LU;
91 7 : }
92 :
93 10 : void AicpuStream::ResetTasks()
94 : {
95 10 : aicpusd_info("Stream[%u] clear load info begin.", streamId_);
96 10 : const std::unique_lock<std::mutex> lockForStream(mutexForStream_);
97 10 : tasks_.clear();
98 10 : }
99 :
100 10 : void AicpuStream::ShowProgress()
101 : {
102 10 : if (nextTaskIndex_ >= tasks_.size()) {
103 7 : aicpusd_run_info("Stream[%u] is finished", streamId_);
104 7 : return;
105 : }
106 3 : const auto kernelName = PtrToPtr<void, char_t>(ValueToPtr(tasks_[nextTaskIndex_].kernelName));
107 3 : if (kernelName != nullptr) {
108 2 : aicpusd_run_info("Stream[%u] is at %zu task[%s] now.", streamId_, nextTaskIndex_, kernelName);
109 : } else {
110 1 : aicpusd_run_info("Stream[%u] is at %zu task[invalid] now.", streamId_, nextTaskIndex_);
111 : }
112 : }
113 :
114 93 : int32_t AicpuStream::ExecuteNextTask(const RunContext& runContext, bool& streamEnd)
115 : {
116 : // need lock all execute period
117 93 : const std::unique_lock<std::mutex> lockForSteam(mutexForStream_);
118 93 : if (nextTaskIndex_ >= tasks_.size()) {
119 2 : aicpusd_err(
120 : "Model[%u] stream[%u] has reach task end, total taskNum[%zu], nextTaskIndex[%zu].", runContext.modelId,
121 : streamId_, tasks_.size(), nextTaskIndex_);
122 2 : streamEnd = true;
123 2 : return AICPU_SCHEDULE_OK;
124 : }
125 91 : const auto& taskInfo = tasks_[nextTaskIndex_];
126 91 : aicpusd_info(
127 : "Model[%u] stream[%u] begin to execute %zuth task, taskId[%u], kernelType[%u].", runContext.modelId, streamId_,
128 : nextTaskIndex_, taskInfo.taskID, taskInfo.kernelType);
129 91 : *(const_cast<int32_t*>(&runContext.gotoTaskIndex)) = INVALID_TASK_INDEX;
130 91 : const int32_t ret = ExecuteTask(taskInfo, runContext);
131 91 : if (ret != AICPU_SCHEDULE_OK) {
132 18 : const auto model = AicpuModelManager::GetInstance().GetModel(runContext.modelId);
133 18 : uint64_t transId = UINT64_MAX;
134 18 : if (model != nullptr) {
135 3 : transId = model->GetModelTransId();
136 : }
137 18 : aicpusd_err(
138 : "Model[%u] stream[%u] execute %zuth task failed, taskId[%u], ret[%d], transId[%lu].", runContext.modelId,
139 : streamId_, nextTaskIndex_, taskInfo.taskID, ret, transId);
140 18 : nextTaskIndex_++;
141 18 : return ret;
142 : }
143 73 : if (runContext.pending) {
144 1 : aicpusd_info(
145 : "Model[%u] stream[%u] pending on %zuth task, taskId[%u].", runContext.modelId, streamId_, nextTaskIndex_,
146 : taskInfo.taskID);
147 72 : } else if (runContext.gotoTaskIndex != INVALID_TASK_INDEX) {
148 2 : nextTaskIndex_ = static_cast<size_t>(runContext.gotoTaskIndex);
149 2 : aicpusd_info(
150 : "Model[%u] stream[%u] goto %zuth task, taskId[%u].", runContext.modelId, streamId_, nextTaskIndex_,
151 : taskInfo.taskID);
152 : } else {
153 70 : aicpusd_info(
154 : "Model[%u] stream[%u] execute %zuth task success, taskId[%u].", runContext.modelId, streamId_,
155 : nextTaskIndex_, taskInfo.taskID);
156 70 : nextTaskIndex_++;
157 : }
158 73 : streamEnd = nextTaskIndex_ >= tasks_.size();
159 73 : return AICPU_SCHEDULE_OK;
160 93 : }
161 :
162 91 : int32_t AicpuStream::ExecuteTask(const AicpuTaskInfo& kernelTaskInfo, const RunContext& taskContext)
163 : {
164 91 : if (((kernelTaskInfo.kernelType == static_cast<uint32_t>(AicpuKernelType::CCE_KERNEL)) ||
165 66 : (kernelTaskInfo.kernelType == static_cast<uint32_t>(AicpuKernelType::CCE_KERNEL_HWTS))) &&
166 25 : (kernelTaskInfo.kernelSo == 0UL)) {
167 25 : const int32_t ret = OperatorKernelRegister::Instance().RunOperatorKernel(kernelTaskInfo, taskContext);
168 25 : if (ret != AICPU_SCHEDULE_OK) {
169 17 : AicpuModelErrProc::GetInstance().RecordAicpuOpErrLog(taskContext, kernelTaskInfo, ret);
170 : }
171 :
172 25 : return ret;
173 : }
174 :
175 66 : aicpu::HwtsTsKernel aicpufwKernelInfo = {};
176 66 : const int32_t ret = ConvertToTsKernel(kernelTaskInfo, aicpufwKernelInfo);
177 66 : if (ret != AICPU_SCHEDULE_OK) {
178 1 : return ret;
179 : }
180 :
181 65 : AicpuMonitor::GetInstance().SetAicpuStreamTaskStartTime(taskContext.modelId);
182 65 : const int32_t retAicpu = aeCallInterface(&aicpufwKernelInfo);
183 65 : AicpuMonitor::GetInstance().SetAicpuStreamTaskEndTime(taskContext.modelId);
184 65 : if (retAicpu != AE_STATUS_SUCCESS) {
185 1 : aicpusd_err("Aicpu engine process failed, result[%d].", retAicpu);
186 1 : return retAicpu;
187 : }
188 64 : aicpusd_info("Aicpu engine process success.");
189 : // dump mask.
190 64 : if ((kernelTaskInfo.taskFlag & AICPU_STREAM_TASK_DUMP_MASK) != 0U) {
191 1 : OpDumpTaskManager& opDumpTaskMgr = OpDumpTaskManager::GetInstance();
192 1 : TaskInfoExt dumpTaskInfo(taskContext.streamId, kernelTaskInfo.taskID);
193 1 : DumpFileName dumpFileName(taskContext.streamId, kernelTaskInfo.taskID);
194 1 : if (opDumpTaskMgr.DumpOpInfo(dumpTaskInfo, dumpFileName) != AICPU_SCHEDULE_OK) {
195 1 : aicpusd_warn(
196 : "dump op info failed, modelId[%u], streamId[%u], taskID[%u].", taskContext.modelId,
197 : taskContext.streamId, kernelTaskInfo.taskID);
198 : }
199 : }
200 64 : return AICPU_SCHEDULE_OK;
201 : }
202 :
203 67 : int32_t AicpuStream::ConvertToTsKernel(const AicpuTaskInfo& kernelTaskInfo, aicpu::HwtsTsKernel& aicpufwKernelInfo)
204 : {
205 67 : aicpufwKernelInfo.kernelType = kernelTaskInfo.kernelType;
206 67 : if (kernelTaskInfo.kernelType == static_cast<uint32_t>(AicpuKernelType::FWK_KERNEL)) {
207 65 : aicpufwKernelInfo.kernelBase.fwkKernel.kernel = kernelTaskInfo.paraBase;
208 2 : } else if (
209 2 : (kernelTaskInfo.kernelType == static_cast<uint32_t>(AicpuKernelType::CCE_KERNEL)) ||
210 1 : (kernelTaskInfo.kernelType == static_cast<uint32_t>(AicpuKernelType::CCE_KERNEL_HWTS))) {
211 1 : aicpufwKernelInfo.kernelBase.cceKernel.kernelName = kernelTaskInfo.kernelName;
212 1 : aicpufwKernelInfo.kernelBase.cceKernel.kernelSo = kernelTaskInfo.kernelSo;
213 1 : aicpufwKernelInfo.kernelBase.cceKernel.paramBase = kernelTaskInfo.paraBase;
214 : } else {
215 1 : aicpusd_err("task[%u] kernelType[%u] is invalid.", kernelTaskInfo.taskID, kernelTaskInfo.kernelType);
216 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
217 : }
218 66 : return AICPU_SCHEDULE_OK;
219 : }
220 :
221 14 : int32_t AicpuStream::AttachReportStatusQueue()
222 : {
223 14 : const uint32_t deviceId = AicpuDrvManager::GetInstance().GetDeviceId();
224 96 : for (auto& taskInfo : tasks_) {
225 86 : const auto kernelName = PtrToPtr<const void, const char_t>(ValueToPtr(taskInfo.kernelName));
226 86 : if (kernelName == nullptr) {
227 2 : aicpusd_err("the name of kernel is nullptr.");
228 4 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
229 : }
230 168 : if (std::string(kernelName) == REPORT_STATUS_TASK_NAME) {
231 : const ReportStatusInfo* const bufInfo =
232 3 : PtrToPtr<void, ReportStatusInfo>(ValueToPtr(static_cast<uintptr_t>(taskInfo.paraBase)));
233 3 : if (bufInfo == nullptr) {
234 1 : aicpusd_err(
235 : "ModelReportStatus kernelTaskInfo paramBase is null, streamId[%u], taskId[%u]", taskInfo.streamID,
236 : taskInfo.taskID);
237 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
238 : }
239 2 : const uint32_t queueId = bufInfo->statusOutputQueue.queueId;
240 2 : const auto drvRet = halQueueAttach(deviceId, queueId, 0);
241 2 : if (drvRet != DRV_ERROR_NONE) {
242 1 : aicpusd_err("Aicpusd attach report status queue[%u] failed, ret[%d]", queueId, drvRet);
243 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
244 : }
245 1 : aicpusd_info("Attach report status queue[%u] success.", queueId);
246 : }
247 : }
248 10 : return AICPU_SCHEDULE_OK;
249 : }
250 :
251 21 : int32_t AicpuModel::ModelLoad(const AicpuModelInfo* const modelInfo, const ModelCfgInfo* const cfgInfo)
252 : {
253 : (void)cfgInfo;
254 21 : if (modelInfo == nullptr) {
255 1 : aicpusd_err("Model load failed, as param is null.");
256 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
257 : }
258 20 : const auto modelId = modelInfo->moduleID;
259 : // need lock all load period
260 20 : std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
261 20 : if (!lockForModel.owns_lock()) {
262 1 : aicpusd_err("Model[%u] load failed, as get lock failed.", modelId);
263 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
264 : }
265 :
266 19 : modelId_ = modelId;
267 19 : modelTsId_ = modelInfo->tsId;
268 19 : int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_LOAD);
269 19 : if (ret != AICPU_SCHEDULE_OK) {
270 1 : aicpusd_err("Model[%u] checkOperateAndUpdateStatus failed, ret[%d].", modelId, ret);
271 1 : return ret;
272 : }
273 18 : aicpusd_info("Model[%u] load begin.", modelId);
274 18 : ret = LoadStreamAndTask(modelInfo);
275 18 : if (ret != AICPU_SCHEDULE_OK) {
276 4 : lockForModel.unlock();
277 4 : aicpusd_err("Model[%u] load stream and task failed , ret[%d].", modelId, ret);
278 4 : const int32_t destroyRet = ModelDestroy();
279 4 : if (destroyRet != AICPU_SCHEDULE_OK) {
280 1 : aicpusd_err(
281 : "Model[%u] load stream and task failed and rollback failed, rollback ret[%d].", modelId, destroyRet);
282 : }
283 4 : return ret;
284 : }
285 :
286 : // load queue info
287 14 : ret = LoadQueueInfo(modelInfo);
288 14 : if (ret != AICPU_SCHEDULE_OK) {
289 2 : lockForModel.unlock();
290 2 : aicpusd_err("Model[%u] load queue info failed, ret[%d].", modelId, ret);
291 2 : const int32_t destroyRet = ModelDestroy();
292 2 : if (destroyRet != AICPU_SCHEDULE_OK) {
293 1 : aicpusd_err("Model[%u] load queue info failed and rollback failed, rollback ret[%d].", modelId, destroyRet);
294 : }
295 2 : aicpusd_info("Model[%u] rollback end.", modelId);
296 2 : return ret;
297 : }
298 :
299 : // attach report status queue
300 12 : ret = AttachReportStatusQueue();
301 12 : if (ret != AICPU_SCHEDULE_OK) {
302 2 : lockForModel.unlock();
303 2 : aicpusd_err("Model[%u] attach report status queue failed, ret[%d].", modelId, ret);
304 2 : const int32_t destroyRet = ModelDestroy();
305 2 : if (destroyRet != AICPU_SCHEDULE_OK) {
306 1 : aicpusd_err(
307 : "Model[%u] attach report status queue failed and rollback failed, rollback ret[%d].", modelId,
308 : destroyRet);
309 : }
310 2 : aicpusd_info("Model[%u] rollback end.", modelId);
311 2 : return ret;
312 : }
313 :
314 : // load success set isValid true;
315 10 : SetExtModelId(AicpuModelManager::GetInstance().GetExtModelId(modelId));
316 10 : isValid = true;
317 10 : endOfSequence_ = false;
318 10 : abnormalBreak_ = AicpuModelManager::GetInstance().AbnormalBreak(modelId);
319 10 : abnormalEnqueue_ = AicpuModelManager::GetInstance().AbnormalEnqueue(modelId);
320 10 : abnormalEnabled_ = AicpuModelManager::GetInstance().AbnormalEnabled(modelId);
321 10 : aicpusd_info(
322 : "Model[%u] load success, abnormal break[%d], abnormal enqueue[%d], abnormal enabled[%d].", modelId,
323 : static_cast<int32_t>(abnormalBreak_), static_cast<int32_t>(abnormalEnqueue_),
324 : static_cast<int32_t>(abnormalEnabled_));
325 10 : int32_t msgQId = 0;
326 10 : if (AicpuModelManager::GetInstance().GetModelMsgQueues(modelId, true, msgQId) == AICPU_SCHEDULE_OK) {
327 1 : aicpusd_info("Model[%u]'s input msg queue is %d", modelId, msgQId);
328 1 : inputMsgQueueIds_.insert(static_cast<size_t>(msgQId));
329 : }
330 10 : if (AicpuModelManager::GetInstance().GetModelMsgQueues(modelId, false, msgQId) == AICPU_SCHEDULE_OK) {
331 1 : aicpusd_info("Model[%u]'s output msg queue is %d", modelId, msgQId);
332 1 : outputMsgQueueIds_.insert(static_cast<size_t>(msgQId));
333 : }
334 10 : return AICPU_SCHEDULE_OK;
335 20 : }
336 :
337 13 : int32_t AicpuModel::ModelExecute()
338 : {
339 : {
340 13 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
341 13 : if (!lockForModel.owns_lock()) {
342 1 : aicpusd_err("Model[%u] execute failed, as get lock failed.", modelId_);
343 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
344 : }
345 12 : const int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_EXECUTE);
346 12 : if (ret != AICPU_SCHEDULE_OK) {
347 1 : aicpusd_err("Model[%u] checkOperateAndUpdateStatus failed, ret[%d].", modelId_, ret);
348 1 : return ret;
349 : }
350 13 : }
351 :
352 11 : const int32_t ret = ResetModelForExecute();
353 11 : if (ret != AICPU_SCHEDULE_OK) {
354 2 : aicpusd_err("model[%u] ResetModelForExecute fail", modelId_);
355 2 : return ret;
356 : }
357 9 : const bool isProfOpen = (&(aicpu::IsModelProfOpen) == nullptr) ? false : aicpu::IsModelProfOpen();
358 9 : aicpusd_info("Model[%u] execute begin, prof model[%d].", modelId_, isProfOpen);
359 9 : const bool hasThread = AicpuDrvManager::GetInstance().HasThread();
360 : // if model id is invalid, do not report profiling
361 9 : if ((!isProfOpen) || (extModelId_ == INVALID_NUMBER)) {
362 8 : ActiveOtherAicpuStreams();
363 8 : return ExecuteStream(modelS0Stream_, !hasThread);
364 : }
365 : // report start tag
366 1 : aicpu::ProfModelMessage profModelMessage("AICPU_MODEL");
367 : (void)profModelMessage.SetAicpuModelId(extModelId_)
368 : ->SetDataTagId(MSPROF_AICPU_MODEL_TAG)
369 1 : ->SetAicpuModelIterId(static_cast<uint16_t>(iteratorCount_))
370 : ->SetAicpuModelTimeStamp(aicpu::GetSystemTick())
371 : ->SetAicpuTagId(aicpu::MODEL_EXECUTE_START)
372 : ->SetEventId(static_cast<uint16_t>(0U))
373 1 : ->SetDeviceId(AicpuDrvManager::GetInstance().GetDeviceId());
374 1 : (void)profModelMessage.ReportProfModelMessage();
375 : // has thread means no need execute inline
376 1 : ActiveOtherAicpuStreams();
377 1 : return ExecuteStream(modelS0Stream_, !hasThread);
378 1 : }
379 :
380 11 : void AicpuModel::ActiveOtherAicpuStreams()
381 : {
382 11 : const bool syncSendFlag = GetCpuMode();
383 13 : for (const auto aicpustreamId : otherAicpuStreams_) {
384 2 : AICPUSubEventInfo subEventInfo = {};
385 2 : subEventInfo.modelId = modelId_;
386 2 : subEventInfo.para.streamInfo.streamId = aicpustreamId;
387 :
388 2 : const int32_t ret = AicpuMsgSend::SendAICPUSubEvent(
389 : PtrToPtr<AICPUSubEventInfo, const char_t>(&subEventInfo), static_cast<uint32_t>(sizeof(AICPUSubEventInfo)),
390 : AICPU_SUB_EVENT_ACTIVE_STREAM, CP_DEFAULT_GROUP_ID, syncSendFlag);
391 2 : if (ret != AICPU_SCHEDULE_OK) {
392 1 : aicpusd_err("Send aicpu subevent failed. Event modelId is %d, streamId: %u", modelId_, aicpustreamId);
393 : };
394 : }
395 11 : }
396 :
397 8 : int32_t AicpuModel::TaskReport()
398 : {
399 8 : const int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_TASK_REPORT);
400 8 : if (ret != AICPU_SCHEDULE_OK) {
401 5 : aicpusd_err("Model[%u] taskReport failed, as CheckOperateAndUpdateStatus failed ret[%d].", modelId_, ret);
402 5 : return ret;
403 : }
404 :
405 3 : aicpusd_info("Model[%u] task report success", modelId_);
406 3 : return AICPU_SCHEDULE_OK;
407 : }
408 :
409 7 : int32_t AicpuModel::ModelAbort()
410 : {
411 : // need lock all abort period
412 7 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
413 7 : if (!lockForModel.owns_lock()) {
414 1 : aicpusd_err("Model[%u] abort failed, as get lock failed.", modelId_);
415 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
416 : }
417 6 : int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_ABORT);
418 6 : if (ret != AICPU_SCHEDULE_OK) {
419 4 : aicpusd_err("Model[%u] abort failed, as CheckOperateAndUpdateStatus failed ret[%d].", modelId_, ret);
420 4 : return ret;
421 : }
422 :
423 2 : aicpusd_info("Model[%u] abort begin.", modelId_);
424 2 : ret = ReleaseModelResource();
425 2 : if (ret != AICPU_SCHEDULE_OK) {
426 1 : aicpusd_err("Model[%u] abort failed, as ReleaseModelResource failed ret[%d].", modelId_, ret);
427 1 : return ret;
428 : }
429 : {
430 : // use write lock to wait stream execute end, don't remove it.
431 1 : (void)pthread_rwlock_wrlock(&rwlockForStream_);
432 1 : (void)pthread_rwlock_unlock(&rwlockForStream_);
433 : }
434 :
435 1 : aicpusd_info("Model[%u] abort success.", modelId_);
436 1 : return AICPU_SCHEDULE_OK;
437 7 : }
438 :
439 17 : int32_t AicpuModel::ModelDestroy()
440 : {
441 : // need lock all destroy period
442 17 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
443 17 : if (!lockForModel.owns_lock()) {
444 1 : aicpusd_err("Model[%u] destroy failed, as get lock failed.", modelId_);
445 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
446 : }
447 16 : int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_DESTROY);
448 16 : if (ret != AICPU_SCHEDULE_OK) {
449 1 : aicpusd_err("Model[%u] destroy failed, as CheckOperateAndUpdateStatus failed ret[%d].", modelId_, ret);
450 1 : return ret;
451 : }
452 :
453 15 : aicpusd_info("Model[%u] destroy begin, isValid=%d", modelId_, isValid);
454 15 : isValid = false;
455 : // destroy all resource.
456 15 : ret = ReleaseModelResource();
457 15 : if (ret != AICPU_SCHEDULE_OK) {
458 1 : aicpusd_err("Model[%u] destroy failed, as ReleaseModelResource failed ret[%d].", modelId_, ret);
459 1 : return ret;
460 : }
461 14 : ClearLoadInfo();
462 14 : aicpusd_info("Model[%u] destroy success", modelId_);
463 14 : return AICPU_SCHEDULE_OK;
464 17 : }
465 :
466 4 : int32_t AicpuModel::ModelStop()
467 : {
468 4 : aicpusd_info("Model[%u] Begin to stop", modelId_);
469 : // need lock all destroy period
470 4 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
471 4 : if (!lockForModel.owns_lock()) {
472 1 : aicpusd_err("Model[%u] destroy failed, as get lock failed.", modelId_);
473 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
474 : }
475 : // update status to stop
476 3 : int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_STOP);
477 3 : if (ret != AICPU_SCHEDULE_OK) {
478 1 : aicpusd_err("Model[%u] stop failed, as CheckOperateAndUpdateStatus failed, ret[%d].", modelId_, ret);
479 1 : return ret;
480 : }
481 :
482 2 : aicpusd_info("Model[%u] destroy begin, isValid=%d", modelId_, isValid);
483 : // destroy all resource.
484 2 : ret = ReleaseModelResource();
485 2 : if (ret != AICPU_SCHEDULE_OK) {
486 1 : aicpusd_err("Model[%u] stop failed, as ReleaseModelResource failed, ret[%d].", modelId_, ret);
487 1 : return ret;
488 : }
489 :
490 : // if getlock, then all streams in model is not running, so we can ensure model is stopped
491 1 : aicpusd_run_info("Model[%u] try to get stream lock", modelId_);
492 1 : (void)pthread_rwlock_wrlock(&rwlockForStream_);
493 1 : (void)pthread_rwlock_unlock(&rwlockForStream_);
494 1 : aicpusd_run_info("Model[%u] stop success", modelId_);
495 1 : return AICPU_SCHEDULE_OK;
496 4 : }
497 :
498 4 : int32_t AicpuModel::ModelRestart()
499 : {
500 4 : aicpusd_info("Model[%u] begin to restart", modelId_);
501 : {
502 : // need lock
503 4 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
504 4 : if (!lockForModel.owns_lock()) {
505 1 : aicpusd_err("Model[%u] restart failed, as get lock failed.", modelId_);
506 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
507 : }
508 : // update status to recover
509 3 : const int32_t updateRet = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_RESTART);
510 3 : if (updateRet != AICPU_SCHEDULE_OK) {
511 1 : aicpusd_err(
512 : "Model[%u] restart failed, as CheckOperateAndUpdateStatus failed, ret[%d].", modelId_, updateRet);
513 1 : return updateRet;
514 : }
515 2 : aicpusd_info("Model[%u] change status success", modelId_);
516 4 : }
517 :
518 2 : AICPUSubEventInfo subEventInfo = {};
519 2 : subEventInfo.modelId = modelId_;
520 2 : const auto ret = AicpuMsgSend::SendAICPUSubEvent(
521 : PtrToPtr<AICPUSubEventInfo, const char_t>(&subEventInfo), static_cast<uint32_t>(sizeof(AICPUSubEventInfo)),
522 : AICPU_SUB_EVENT_REPEAT_MODEL, CP_DEFAULT_GROUP_ID, true);
523 2 : if (ret != AICPU_SCHEDULE_OK) {
524 1 : aicpusd_err("Send aicpu subevent failed. Event modelId is %d", modelId_);
525 1 : return ret;
526 : }
527 1 : aicpusd_info("Model[%u] restart success", modelId_);
528 1 : return AICPU_SCHEDULE_OK;
529 : }
530 :
531 6 : int32_t AicpuModel::ModelClearInput()
532 : {
533 6 : aicpusd_info("Model[%u] begin to clearInput", modelId_);
534 : // need lock
535 6 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
536 6 : if (!lockForModel.owns_lock()) {
537 1 : aicpusd_err("Model[%u] clear input failed, as get lock failed.", modelId_);
538 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
539 : }
540 : // update status to recover
541 5 : const int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_CLEAR_INPUT);
542 5 : if (ret != AICPU_SCHEDULE_OK) {
543 1 : aicpusd_err("Model[%u] clear input failed, as CheckOperateAndUpdateStatus failed, ret[%d].", modelId_, ret);
544 1 : return ret;
545 : }
546 :
547 4 : const uint32_t deviceId = AicpuDrvManager::GetInstance().GetDeviceId();
548 4 : auto clearRet = ModelClearInputQueues(inputQueueIds_, deviceId);
549 4 : if (clearRet != AICPU_SCHEDULE_OK) {
550 2 : aicpusd_err("Failed to clear input queues");
551 2 : return clearRet;
552 : }
553 :
554 2 : clearRet = ModelClearInputQueues(inputMsgQueueIds_, deviceId);
555 2 : if (clearRet != AICPU_SCHEDULE_OK) {
556 1 : aicpusd_err("Failed to clear input msg queues");
557 1 : return clearRet;
558 : }
559 :
560 1 : aicpusd_info("Model[%u] clearInput success", modelId_);
561 1 : return AICPU_SCHEDULE_OK;
562 6 : }
563 :
564 5 : int32_t AicpuModel::ModelClearInputQueues(const std::unordered_set<size_t>& queueIds, const uint32_t deviceId) const
565 : {
566 8 : for (const auto queueId : queueIds) {
567 : do {
568 6 : void* mbuf = nullptr;
569 6 : const auto ret = halQueueDeQueue(deviceId, queueId, &mbuf);
570 6 : if (ret == DRV_ERROR_QUEUE_EMPTY) {
571 3 : break;
572 : }
573 3 : if ((ret != DRV_ERROR_NONE) || (mbuf == nullptr)) {
574 2 : aicpusd_err(
575 : "Dequeue from queueId[%u], deviceId[%u] fail, ret is %d", queueId, deviceId,
576 : static_cast<int32_t>(ret));
577 2 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
578 : }
579 1 : (void)halMbufFree(PtrToPtr<void, Mbuf>(mbuf));
580 1 : } while (true);
581 : }
582 3 : return AICPU_SCHEDULE_OK;
583 : }
584 :
585 21506 : int32_t AicpuModel::Exit()
586 : {
587 21506 : if (IsValid()) {
588 2 : isValid = false;
589 2 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_);
590 : // destroy all resource.
591 2 : const int32_t ret = ReleaseModelResource();
592 2 : ClearLoadInfo();
593 2 : if (ret != AICPU_SCHEDULE_OK) {
594 1 : aicpusd_err("Model[%u] Exit failed, as ReleaseModelResource failed ret[%d].", modelId_, ret);
595 1 : return ret;
596 : }
597 1 : aicpusd_info("Model[%u] exit", modelId_);
598 2 : }
599 21505 : return AICPU_SCHEDULE_OK;
600 : }
601 :
602 3 : int32_t AicpuModel::EndGraph()
603 : {
604 3 : const std::unique_lock<std::mutex> lockForModel(mutexForModel_, std::try_to_lock);
605 3 : if (!lockForModel.owns_lock()) {
606 1 : aicpusd_err("Model[%u] end graph failed, as get lock failed.", modelId_);
607 1 : return AICPU_SCHEDULE_ERROR_IN_WORKING;
608 : }
609 2 : const int32_t ret = CheckOperateAndUpdateStatus(AicpuModelOperate::MODEL_OPERATE_END_GRAPH);
610 2 : if (ret != AICPU_SCHEDULE_OK) {
611 1 : aicpusd_err("Model[%u] end graph failed, as CheckOperateAndUpdateStatus failed ret[%d].", modelId_, ret);
612 1 : return ret;
613 : }
614 1 : aicpusd_info("Model[%u] end graph.", modelId_);
615 1 : return AICPU_SCHEDULE_OK;
616 3 : }
617 :
618 1 : int32_t AicpuModel::ActiveStream(const uint32_t streamId)
619 : {
620 1 : aicpusd_info("Model[%u] active stream[%u].", modelId_, streamId);
621 1 : return ExecuteStream(streamId, false);
622 : }
623 :
624 1 : int32_t AicpuModel::RecoverStream(const uint32_t streamId)
625 : {
626 1 : aicpusd_info("Model[%u] recover stream[%u].", modelId_, streamId);
627 1 : return ExecuteStream(streamId, false);
628 : }
629 :
630 16 : int32_t AicpuModel::ExecuteStream(const uint32_t streamId, const bool executeInline)
631 : {
632 16 : aicpusd_info("Model[%u] execute stream[%u] begin.", modelId_, streamId);
633 16 : uint32_t currentStreamId = streamId;
634 16 : (void)pthread_rwlock_rdlock(&rwlockForStream_);
635 16 : AicpuStream* stream = GetStreamByStreamId(currentStreamId);
636 16 : if (stream == nullptr) {
637 4 : (void)pthread_rwlock_unlock(&rwlockForStream_);
638 4 : aicpusd_err("Model[%u] execute stream[%u] failed, stream not found.", modelId_, currentStreamId);
639 4 : return AICPU_SCHEDULE_ERROR_STREAM_NOT_FOUND;
640 : }
641 12 : RunContext runContext = {
642 12 : .modelId = modelId_,
643 12 : .modelTsId = modelTsId_,
644 : .streamId = currentStreamId,
645 : .pending = false,
646 : .executeInline = executeInline,
647 12 : .gotoTaskIndex = 0};
648 12 : g_aicpuProfiler.SetModelId(modelId_);
649 12 : g_aicpuProfiler.SetStreamId(streamId);
650 12 : bool streamEnd = false;
651 12 : std::stack<uint32_t> streamStack;
652 12 : int32_t ret = AICPU_SCHEDULE_OK;
653 : while (true) {
654 94 : ret = CheckOperate(AicpuModelOperate::MODEL_OPERATE_RUN_TASK);
655 94 : if (ret != AICPU_SCHEDULE_OK) {
656 1 : break;
657 : }
658 93 : ret = stream->ExecuteNextTask(runContext, streamEnd);
659 : // check if pending or if error
660 93 : if (((ret != AICPU_SCHEDULE_OK) && (!abnormalEnabled_.load())) ||
661 : (ret == AICPU_SCHEDULE_ERROR_TASK_EXECUTE_FAILED)) {
662 2 : break;
663 : }
664 :
665 : // update model ret code
666 91 : UpdateModelRetCode(ret);
667 91 : if (executeInline) {
668 74 : uint32_t switchStreamId = INVALID_NUMBER;
669 74 : if (currentStreamId != runContext.streamId) {
670 2 : streamStack.emplace(currentStreamId);
671 2 : switchStreamId = runContext.streamId;
672 72 : } else if (streamEnd) {
673 9 : if (!streamStack.empty()) {
674 1 : switchStreamId = streamStack.top();
675 1 : streamStack.pop();
676 : }
677 : } else {
678 : // do nothing
679 : }
680 74 : if (switchStreamId != INVALID_NUMBER) {
681 3 : currentStreamId = switchStreamId;
682 3 : runContext.streamId = currentStreamId;
683 3 : stream = GetStreamByStreamId(currentStreamId);
684 3 : if (stream == nullptr) {
685 1 : aicpusd_err(
686 : "Model[%u] execute stream[%u] failed as stream[%u] not found.", modelId_, streamId,
687 : currentStreamId);
688 1 : ret = AICPU_SCHEDULE_ERROR_STREAM_NOT_FOUND;
689 1 : break;
690 : }
691 2 : continue;
692 : }
693 : }
694 : // not inline, when pending or stream end, exit.
695 88 : if (runContext.pending || streamEnd) {
696 : break;
697 : }
698 82 : }
699 12 : (void)pthread_rwlock_unlock(&rwlockForStream_);
700 12 : if (ret != AICPU_SCHEDULE_OK) {
701 4 : aicpusd_err(
702 : "Model[%u] execute stream[%u] failed currentStreamId[%u], ret[%d].", modelId_, streamId, currentStreamId,
703 : ret);
704 : // execute failed, need taskReport.
705 4 : (void)TaskReport();
706 4 : ProcessModelException(modelId_);
707 4 : return ret;
708 : }
709 8 : aicpusd_info("Model[%u] execute stream[%u] success.", modelId_, streamId);
710 8 : return AICPU_SCHEDULE_OK;
711 12 : }
712 :
713 91 : void AicpuModel::UpdateModelRetCode(const int32_t retCode)
714 : {
715 107 : if ((retCode != 0) && (abnormalEnabled_.load()) && (retCode_.load() == 0)) {
716 1 : retCode_ = retCode + INNER_ERROR_BASE;
717 2 : aicpusd_run_info("Update Model[%u] retCode to %d", modelId_, retCode_.load());
718 : }
719 91 : }
720 :
721 12 : int32_t AicpuModel::AttachReportStatusQueue()
722 : {
723 12 : int32_t ret = AICPU_SCHEDULE_OK;
724 12 : (void)pthread_rwlock_wrlock(&rwlockForStream_);
725 22 : for (auto& it : aicpuStreams_) {
726 12 : ret = it.second.AttachReportStatusQueue();
727 12 : if (ret != AICPU_SCHEDULE_OK) {
728 2 : break;
729 : }
730 : }
731 12 : (void)pthread_rwlock_unlock(&rwlockForStream_);
732 12 : return ret;
733 : }
734 :
735 : // attention: must get rwlockForStream_ lock out side.
736 13 : AicpuStream* AicpuModel::GetStreamByStreamId(const uint32_t streamId)
737 : {
738 13 : const auto iter = aicpuStreams_.find(streamId);
739 13 : if (iter != aicpuStreams_.end()) {
740 10 : return &(iter->second);
741 : }
742 3 : return nullptr;
743 : }
744 :
745 2 : int32_t AicpuModel::ModelRepeat()
746 : {
747 2 : aicpusd_info("Model[%u] ExtModel[%u] repeat begin.", modelId_, extModelId_);
748 2 : const bool isProfOpen = (&(aicpu::IsModelProfOpen) == nullptr) ? false : aicpu::IsModelProfOpen();
749 2 : if (isProfOpen && (extModelId_ != INVALID_NUMBER)) {
750 : // report end tag
751 1 : aicpu::ProfModelMessage profModelMessage("AICPU_MODEL");
752 : (void)profModelMessage.SetAicpuModelId(extModelId_)
753 : ->SetDataTagId(MSPROF_AICPU_MODEL_TAG)
754 1 : ->SetAicpuModelIterId(static_cast<uint16_t>(iteratorCount_))
755 : ->SetAicpuModelTimeStamp(aicpu::GetSystemTick())
756 : ->SetAicpuTagId(aicpu::MODEL_EXECUTE_END)
757 : ->SetEventId(static_cast<uint16_t>(0U))
758 1 : ->SetDeviceId(AicpuDrvManager::GetInstance().GetDeviceId());
759 1 : (void)profModelMessage.ReportProfModelMessage();
760 1 : aicpusd_info("ReportProfModelMessage done, iterateId=%llu", iteratorCount_);
761 1 : }
762 2 : iteratorCount_++;
763 2 : return ModelExecute();
764 : }
765 :
766 6 : void AicpuModel::ProcessModelException(const uint32_t modelId) const
767 : {
768 6 : if (AbnormalNeedBreak()) {
769 2 : return;
770 : }
771 : uint32_t runMode;
772 5 : const aicpu::status_t ret = aicpu::GetAicpuRunMode(runMode);
773 5 : if (ret != aicpu::AICPU_ERROR_NONE) {
774 1 : aicpusd_err("GetAicpuRunMode returned [%u]", ret);
775 1 : return;
776 : }
777 : // continue to execute the model.
778 4 : if (runMode == aicpu::AicpuRunMode::PROCESS_SOCKET_MODE) {
779 4 : AICPUSubEventInfo subEventInfo = {};
780 4 : subEventInfo.modelId = modelId;
781 4 : const int32_t result = AicpuMsgSend::SendAICPUSubEvent(
782 : PtrToPtr<AICPUSubEventInfo, const char_t>(&subEventInfo), static_cast<uint32_t>(sizeof(AICPUSubEventInfo)),
783 : AICPU_SUB_EVENT_REPEAT_MODEL, CP_DEFAULT_GROUP_ID, false);
784 4 : if (result != AICPU_SCHEDULE_OK) {
785 1 : aicpusd_err("Send aicpu subevent failed. Event modelId is %d", modelId);
786 : }
787 : }
788 : }
789 :
790 11 : int32_t AicpuModel::ResetModelForExecute()
791 : {
792 11 : const int32_t ret = ReleaseModelResource();
793 11 : if (ret != AICPU_SCHEDULE_OK) {
794 3 : return ret;
795 : }
796 :
797 8 : SetModelRetCode(0);
798 8 : SetModelTransId(UINT64_MAX);
799 :
800 8 : (void)pthread_rwlock_wrlock(&rwlockForStream_);
801 15 : for (auto& it : aicpuStreams_) {
802 7 : it.second.ResetToStart();
803 : }
804 8 : (void)pthread_rwlock_unlock(&rwlockForStream_);
805 8 : return AICPU_SCHEDULE_OK;
806 : }
807 :
808 47 : int32_t AicpuModel::CheckOperateAndUpdateStatus(const AicpuModelOperate operate)
809 : {
810 47 : const std::unique_lock<std::mutex> lockForModelStatus(mutexForModelStatus_);
811 47 : const int32_t operateIndex = static_cast<int32_t>(operate);
812 47 : if (!modelOperatePermission[static_cast<int32_t>(modelStatus_)][operateIndex]) {
813 7 : aicpusd_err(
814 : "Model[%u] status[%s] is not allow operate[%s].", modelId_,
815 : STATUS_VALUE[static_cast<int32_t>(modelStatus_)].c_str(), OPERATE_VALUE[operateIndex].c_str());
816 7 : return AICPU_SCHEDULE_ERROR_MODEL_STATUS_NOT_ALLOW_OPERATE;
817 : }
818 40 : const AicpuModelStatus dstStatus = operateNextStatus[operateIndex];
819 40 : if ((dstStatus < AicpuModelStatus::MODEL_STATUS_MAX) && (dstStatus != modelStatus_)) {
820 37 : aicpusd_info(
821 : "Model[%u] status change from [%s] to [%s] as operate[%s].", modelId_,
822 : STATUS_VALUE[static_cast<int32_t>(modelStatus_)].c_str(),
823 : STATUS_VALUE[static_cast<int32_t>(dstStatus)].c_str(), OPERATE_VALUE[operateIndex].c_str());
824 37 : modelStatus_ = dstStatus;
825 : }
826 40 : return AICPU_SCHEDULE_OK;
827 47 : }
828 :
829 73 : int32_t AicpuModel::CheckOperate(const AicpuSchedule::AicpuModelOperate operate)
830 : {
831 73 : const std::unique_lock<std::mutex> lockForModelStatus(mutexForModelStatus_);
832 73 : if (modelOperatePermission[static_cast<size_t>(modelStatus_)][static_cast<size_t>(operate)]) {
833 71 : return AICPU_SCHEDULE_OK;
834 : }
835 2 : aicpusd_err(
836 : "Model[%u] status[%s] is not allow operate[%s].", modelId_,
837 : STATUS_VALUE[static_cast<size_t>(modelStatus_)].c_str(), OPERATE_VALUE[static_cast<size_t>(operate)].c_str());
838 2 : return AICPU_SCHEDULE_ERROR_MODEL_STATUS_NOT_ALLOW_OPERATE;
839 73 : }
840 :
841 29 : int32_t AicpuModel::UnGardModelBuf(Mbuf* const mbuf)
842 : {
843 29 : const std::unique_lock<std::mutex> lockForBufFree(mutexForMbuffList_);
844 58 : return BufManager::GetInstance().UnGuardBuf(modelId_, mbuf);
845 29 : }
846 :
847 28 : int32_t AicpuModel::ReleaseModelResource()
848 : {
849 28 : prepareData_.dequeueIndex = 0U;
850 28 : prepareData_.lastInputMbuflistPtr = nullptr;
851 28 : inputDataPtrs_.clear();
852 28 : postpareData_.enqueueIndex = 0U;
853 28 : endOfSequence_ = false;
854 28 : retCode_ = 0;
855 28 : nullDataFlag_ = false;
856 : {
857 28 : const std::unique_lock<std::mutex> lockForBufFree(mutexForMbuffList_);
858 28 : BufManager::GetInstance().FreeBuf(modelId_);
859 28 : }
860 56 : (void)EventWaitManager::EndGraphWaitManager().ClearBatch({modelId_});
861 56 : (void)EventWaitManager::PrepareMemWaitManager().ClearBatch({modelId_});
862 56 : (void)EventWaitManager::AnyQueNotEmptyWaitManager().ClearBatch({modelId_});
863 56 : (void)EventWaitManager::TableUnlockWaitManager().ClearBatch({modelId_});
864 : {
865 28 : const std::unique_lock<std::mutex> lockForModelNotifyId(mutexForModelNotifyId_);
866 28 : (void)EventWaitManager::NotifyWaitManager().ClearBatch(modelNotifyId_);
867 28 : }
868 : {
869 28 : const std::unique_lock<std::mutex> lockForQueue(mutexForQueueEventSubscribed_);
870 28 : int32_t ret = EventWaitManager::QueueNotEmptyWaitManager().ClearBatch(inputQueueIds_);
871 28 : if (ret != AICPU_SCHEDULE_OK) {
872 1 : return ret;
873 : }
874 27 : ret = EventWaitManager::QueueNotFullWaitManager().ClearBatch(outputQueueIds_);
875 27 : if (ret != AICPU_SCHEDULE_OK) {
876 1 : return ret;
877 : }
878 28 : }
879 :
880 26 : int32_t msgRet = EventWaitManager::QueueNotEmptyWaitManager().ClearBatch(inputMsgQueueIds_);
881 26 : if (msgRet != AICPU_SCHEDULE_OK) {
882 1 : return msgRet;
883 : }
884 25 : msgRet = EventWaitManager::QueueNotFullWaitManager().ClearBatch(outputMsgQueueIds_);
885 25 : if (msgRet != AICPU_SCHEDULE_OK) {
886 1 : return msgRet;
887 : }
888 :
889 24 : ClearAllLockedTable();
890 24 : return AICPU_SCHEDULE_OK;
891 : }
892 :
893 23 : int32_t AicpuModel::LoadStreamAndTask(const AicpuModelInfo* const modelInfo)
894 : {
895 23 : if (static_cast<int32_t>(modelInfo->streamInfoNum) == 0) {
896 3 : aicpusd_err("Load model[%u] failed as stream num is 0.", modelId_);
897 3 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
898 : }
899 :
900 20 : const auto infoPtr = reinterpret_cast<const StreamInfo*>(static_cast<uintptr_t>(modelInfo->streamInfoPtr));
901 20 : if (infoPtr == nullptr) {
902 1 : aicpusd_err("Load model[%u] failed as streamInfoPtr is null.", modelId_);
903 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
904 : }
905 :
906 19 : (void)pthread_rwlock_wrlock(&rwlockForStream_);
907 19 : modelS0Stream_ = INVALID_NUMBER;
908 : // collect all stream information in the model.
909 110 : for (uint32_t i = 0U; i < static_cast<uint32_t>(modelInfo->streamInfoNum); i++) {
910 92 : auto& streamInfo = infoPtr[i];
911 92 : allStreams_.emplace_back(streamInfo);
912 91 : aicpusd_info("Load model[%u] stream[%u]", modelId_, streamInfo.streamID);
913 92 : if ((streamInfo.streamFlag & AICPU_STREAM_INDEX) != 0U) {
914 17 : if ((streamInfo.streamFlag & HEAD_STREAM_INDEX) == 0U) {
915 0 : otherAicpuStreams_.emplace_back(streamInfo.streamID);
916 0 : continue;
917 : }
918 17 : if (modelS0Stream_ != INVALID_NUMBER) {
919 1 : (void)pthread_rwlock_unlock(&rwlockForStream_);
920 1 : aicpusd_err(
921 : "load model[%u] failed as stream[%u] and stream[%u] are both s0 stream.", modelId_, modelS0Stream_,
922 : streamInfo.streamID);
923 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
924 : }
925 16 : aicpusd_info("Load model[%u] modelS0Stream[%u]", modelId_, streamInfo.streamID);
926 16 : modelS0Stream_ = streamInfo.streamID;
927 : } else {
928 75 : if (reportStream_ == INVALID_NUMBER) {
929 6 : aicpusd_info("Load model[%u] reportStream[%u].", modelId_, streamInfo.streamID);
930 6 : reportStream_ = streamInfo.streamID;
931 : }
932 : }
933 : }
934 18 : if (modelS0Stream_ == INVALID_NUMBER) {
935 3 : (void)pthread_rwlock_unlock(&rwlockForStream_);
936 3 : aicpusd_err("load model[%u] failed as no s0 stream found.", modelId_);
937 3 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
938 : }
939 :
940 15 : ModelStreamManager::GetInstance().Reg(modelId_, allStreams_);
941 15 : const auto aicpuTaskNum = modelInfo->aicpuTaskNum;
942 15 : if (static_cast<int32_t>(aicpuTaskNum) == 0) {
943 1 : (void)pthread_rwlock_unlock(&rwlockForStream_);
944 1 : aicpusd_err("load model[%u] failed as aicpuTaskNum is 0.", modelId_);
945 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
946 : }
947 14 : const auto taskPtr = PtrToPtr<void, AicpuTaskInfo>(ValueToPtr(modelInfo->aicpuTaskPtr));
948 14 : if (taskPtr == nullptr) {
949 1 : (void)pthread_rwlock_unlock(&rwlockForStream_);
950 1 : aicpusd_err("Load model[%u] failed as aicpuTaskPtr is null.", modelId_);
951 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
952 : }
953 :
954 13 : std::unordered_map<uint32_t, std::vector<const AicpuTaskInfo*>> streamTasks;
955 98 : for (size_t i = 0U; i < aicpuTaskNum; i++) {
956 85 : const uint32_t streamId = taskPtr[i].streamID;
957 85 : LoadWaitNotifyId(taskPtr[i], modelNotifyId_);
958 85 : streamTasks[streamId].emplace_back(&taskPtr[i]);
959 85 : aicpusd_info(
960 : "Task: taskID[%u], streamId[%u], kernelType[%u], "
961 : "kernelName[%lu], kernelSo[%lu], taskFlag[%u].",
962 : taskPtr[i].taskID, taskPtr[i].streamID, taskPtr[i].kernelType, taskPtr[i].kernelName, taskPtr[i].kernelSo,
963 : taskPtr[i].taskFlag);
964 : }
965 :
966 26 : for (auto& streamTask : streamTasks) {
967 13 : aicpuStreams_[streamTask.first].InitAicpuStream(streamTask.first, streamTask.second);
968 13 : aicpusd_info("Model[%u] aicpu stream[%u] has %zu tasks.", modelId_, streamTask.first, streamTask.second.size());
969 : }
970 13 : (void)pthread_rwlock_unlock(&rwlockForStream_);
971 13 : return AICPU_SCHEDULE_OK;
972 13 : }
973 :
974 88 : void AicpuModel::LoadWaitNotifyId(const AicpuTaskInfo& aicpuTaskInfo, std::unordered_set<size_t>& waitNotifyIdSet) const
975 : {
976 88 : if ((aicpuTaskInfo.kernelSo != 0U) || ((aicpuTaskInfo.kernelType != aicpu::KERNEL_TYPE_CCE) &&
977 28 : (aicpuTaskInfo.kernelType != aicpu::KERNEL_TYPE_AICPU))) {
978 74 : return;
979 : }
980 14 : const auto kernelName = PtrToPtr<void, char_t>(ValueToPtr(aicpuTaskInfo.kernelName));
981 38 : if ((kernelName == nullptr) || (std::string(kernelName) != "waitNotify")) {
982 12 : return;
983 : }
984 2 : const auto notifyId = PtrToPtr<void, uint32_t>(ValueToPtr(aicpuTaskInfo.paraBase));
985 2 : if (notifyId == nullptr) {
986 1 : aicpusd_warn(
987 : "Stream[%u] taskId[%u] is waitNotify, but paraBase is null.", aicpuTaskInfo.streamID, aicpuTaskInfo.taskID);
988 1 : return;
989 : }
990 1 : (void)waitNotifyIdSet.emplace(*notifyId);
991 1 : aicpusd_info(
992 : "Model[%u] stream[%u] taskId[%u] use notifyId[%u]", modelId_, aicpuTaskInfo.streamID, aicpuTaskInfo.taskID,
993 : *notifyId);
994 : }
995 :
996 28 : int32_t AicpuModel::LoadQueueInfo(const AicpuModelInfo* const modelInfo)
997 : {
998 28 : aicpusd_info("LoadQueueInfo for model[%u] begin.", modelId_);
999 28 : if (static_cast<int32_t>(modelInfo->queueSize) == 0) {
1000 11 : aicpusd_info("Model[%u] queueSize is 0, no need load queue info.", modelId_);
1001 11 : return AICPU_SCHEDULE_OK;
1002 : }
1003 17 : aicpusd_info("Task queue size[%u].", modelInfo->queueSize);
1004 17 : const auto infoPtr = PtrToPtr<void, QueInfo>(ValueToPtr(modelInfo->queueInfoPtr));
1005 17 : if (infoPtr == nullptr) {
1006 3 : aicpusd_info(
1007 : "Model[%u] load was not successful, as queueSize[%u] but queueInfoPtr is null.", modelId_,
1008 : modelInfo->queueSize);
1009 3 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
1010 : }
1011 :
1012 14 : const std::unique_lock<std::mutex> lockForQueue(mutexForQueueEventSubscribed_);
1013 14 : const uint32_t deviceId = AicpuDrvManager::GetInstance().GetDeviceId();
1014 14 : auto drvRet = halQueueInit(deviceId);
1015 14 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
1016 1 : aicpusd_err("halQueueInit error, deviceId[%u], ret[%d]", deviceId, drvRet);
1017 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
1018 : }
1019 13 : int32_t ret = AICPU_SCHEDULE_OK;
1020 21 : for (int32_t i = 0; i < static_cast<int32_t>(modelInfo->queueSize); i++) {
1021 13 : if (infoPtr[i].flag == static_cast<uint32_t>(QueueDirectionFlag::QUEUE_CLIENT_INPUT_FLAG) ||
1022 12 : infoPtr[i].flag == static_cast<uint32_t>(QueueDirectionFlag::QUEUE_CLIENT_OUTPUT_FLAG)) {
1023 1 : aicpusd_info("it is qlient qs flag[%d] = [%u]", i, infoPtr[i].flag);
1024 1 : continue;
1025 : }
1026 12 : if (infoPtr[i].flag == static_cast<uint32_t>(QueueDirectionFlag::QUEUE_INPUT_FLAG)) {
1027 8 : QueueSetInputPara inPutParam = {};
1028 8 : QueueSetInput inPut = {};
1029 8 : inPut.queSetWorkMode.qid = infoPtr[i].queueID;
1030 8 : inPut.queSetWorkMode.workMode = QUEUE_MODE_PULL;
1031 8 : inPutParam.inBuff = static_cast<void*>(&inPut);
1032 8 : inPutParam.inLen = static_cast<uint32_t>(sizeof(QueueSetInput));
1033 8 : drvRet = halQueueAttach(deviceId, infoPtr[i].queueID, 0);
1034 8 : if (drvRet != DRV_ERROR_NONE) {
1035 1 : aicpusd_err("Aicpusd attached queue[%u] failed ret[%d]", infoPtr[i].queueID, drvRet);
1036 2 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
1037 : }
1038 7 : drvRet = halQueueSet(0U, QUEUE_SET_WORK_MODE, &inPutParam);
1039 7 : if (drvRet != DRV_ERROR_NONE) {
1040 1 : aicpusd_err(
1041 : "Aicpusd set work mode for queue[%u] failed ret[%d]", infoPtr[i].queueID,
1042 : static_cast<int32_t>(drvRet));
1043 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
1044 : }
1045 6 : ret = AicpuDrvManager::GetInstance().SubscribeQueueNotEmptyEvent(infoPtr[i].queueID);
1046 6 : (void)inputQueueIds_.insert(infoPtr[i].queueID);
1047 6 : aicpusd_run_info("Load model, model[%u], input queue[%u].", modelId_, infoPtr[i].queueID);
1048 4 : } else if (infoPtr[i].flag == static_cast<uint32_t>(QueueDirectionFlag::QUEUE_OUTPUT_FLAG)) {
1049 3 : drvRet = halQueueAttach(deviceId, infoPtr[i].queueID, 0);
1050 3 : if (drvRet != DRV_ERROR_NONE) {
1051 1 : aicpusd_err("Aicpusd attached queue[%u] failed ret[%d]", infoPtr[i].queueID, drvRet);
1052 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
1053 : }
1054 2 : ret = AicpuDrvManager::GetInstance().SubscribeQueueNotFullEvent(infoPtr[i].queueID);
1055 2 : (void)outputQueueIds_.insert(infoPtr[i].queueID);
1056 2 : aicpusd_run_info("Load model, model[%u], output queue[%u].", modelId_, infoPtr[i].queueID);
1057 : } else {
1058 1 : aicpusd_err("queue[%u] flag[%u] is unknown.", infoPtr[i].queueID, infoPtr[i].flag);
1059 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
1060 : }
1061 :
1062 8 : if (ret != AICPU_SCHEDULE_OK) {
1063 1 : aicpusd_err("Failed to subscribe the event of queue, ret[%d].", ret);
1064 1 : AicpuMonitor::GetInstance().SendKillMsgToTsd();
1065 1 : return ret;
1066 : }
1067 7 : queueEventSubscribedInfo_.emplace_back(infoPtr[i]);
1068 : }
1069 8 : inputsIsDequeue_.resize(inputQueueIds_.size(), false);
1070 8 : aicpusd_info("LoadQueueInfo for model[%u] end.", modelId_);
1071 8 : return AICPU_SCHEDULE_OK;
1072 14 : }
1073 :
1074 20 : void AicpuModel::ClearLoadInfo()
1075 : {
1076 20 : aicpusd_info("Model[%u] clear load info begin.", modelId_);
1077 20 : isDestroyModel_ = true;
1078 20 : (void)pthread_rwlock_wrlock(&rwlockForStream_);
1079 20 : ModelStreamManager::GetInstance().UnReg(modelId_, allStreams_);
1080 20 : allStreams_.clear();
1081 30 : for (auto& it : aicpuStreams_) {
1082 10 : it.second.ShowProgress();
1083 10 : it.second.ResetTasks();
1084 : }
1085 20 : aicpuStreams_.clear();
1086 20 : otherAicpuStreams_.clear();
1087 :
1088 20 : (void)pthread_rwlock_unlock(&rwlockForStream_);
1089 : {
1090 20 : const std::unique_lock<std::mutex> lockForQueue(mutexForQueueEventSubscribed_);
1091 24 : for (QueInfo& itQueueInfo : queueEventSubscribedInfo_) {
1092 4 : if (itQueueInfo.flag == static_cast<uint32_t>(QueueDirectionFlag::QUEUE_INPUT_FLAG)) {
1093 2 : (void)AicpuDrvManager::GetInstance().UnSubscribeQueueNotEmptyEvent(itQueueInfo.queueID);
1094 : QueueSetInputPara inPutParam;
1095 : QueueSetInput inPut;
1096 2 : inPut.queSetWorkMode.qid = itQueueInfo.queueID;
1097 2 : inPut.queSetWorkMode.workMode = QUEUE_MODE_PUSH;
1098 2 : inPutParam.inBuff = static_cast<void*>(&inPut);
1099 2 : inPutParam.inLen = static_cast<uint32_t>(sizeof(QueueSetInput));
1100 2 : (void)halQueueSet(0U, QUEUE_SET_WORK_MODE, &inPutParam);
1101 2 : } else if (itQueueInfo.flag == static_cast<uint32_t>(QueueDirectionFlag::QUEUE_OUTPUT_FLAG)) {
1102 1 : (void)AicpuDrvManager::GetInstance().UnSubscribeQueueNotFullEvent(itQueueInfo.queueID);
1103 : } else {
1104 1 : aicpusd_err("queue[%u] flag[%u] is unknown.", itQueueInfo.queueID, itQueueInfo.flag);
1105 : }
1106 : }
1107 20 : queueEventSubscribedInfo_.clear();
1108 20 : inputQueueIds_.clear();
1109 20 : outputQueueIds_.clear();
1110 20 : }
1111 : {
1112 20 : const std::unique_lock<std::mutex> lockForModelNotifyId(mutexForModelNotifyId_);
1113 20 : modelNotifyId_.clear();
1114 20 : }
1115 :
1116 20 : ClearGatheredMbuf();
1117 :
1118 21 : for (const auto inputMsgQ : inputMsgQueueIds_) {
1119 1 : (void)AicpuDrvManager::GetInstance().UnSubscribeQueueNotEmptyEvent(static_cast<uint32_t>(inputMsgQ));
1120 : }
1121 20 : inputMsgQueueIds_.clear();
1122 21 : for (const auto outputMsgQ : outputMsgQueueIds_) {
1123 1 : (void)AicpuDrvManager::GetInstance().UnSubscribeQueueNotFullEvent(static_cast<uint32_t>(outputMsgQ));
1124 : }
1125 20 : outputMsgQueueIds_.clear();
1126 20 : iteratorCount_ = 0UL;
1127 20 : aicpusd_info("Model[%u] clear load info end.", modelId_);
1128 20 : }
1129 :
1130 16 : StoreResult AicpuModel::StoreDequedMbuf(
1131 : const uint64_t transId, const uint32_t routeLabel, const size_t qIndex, Mbuf* const mbuf, const uint32_t queueCount)
1132 : {
1133 : // first get exception transids, then discard mbuf which belong to exption transid
1134 : // if mbuf is not belong to exption transid, then store
1135 16 : if (IsTransIdException(transId)) {
1136 1 : aicpusd_info("Discard mbuf[%u:%u] in model[%u] for it was exception.", transId, routeLabel, modelId_);
1137 1 : halMbufFree(mbuf);
1138 1 : return StoreResult::ABORT_STORE;
1139 : }
1140 30 : if (gatheredMbuf_[transId][routeLabel].Init(queueCount) &&
1141 15 : gatheredMbuf_[transId][routeLabel].Store(qIndex, mbuf, gatheredMbufCntList_)) {
1142 14 : aicpusd_info(
1143 : "Model[%u] Store mbuf[%u:%u] into [%u]th queue, queuesize[%llu].", modelId_, transId, routeLabel, qIndex,
1144 : gatheredMbufCntList_[qIndex]);
1145 14 : return StoreResult::SUCCESS_STORE;
1146 : }
1147 :
1148 1 : aicpusd_err("Failed to store mbuf in Model[%u] transId:%llu, routeLabel:%u", modelId_, transId, routeLabel);
1149 1 : return StoreResult::FAIL_STORE;
1150 : }
1151 :
1152 15 : void AicpuModel::ClearExceptionStore()
1153 : {
1154 15 : std::vector<int64_t> transIdsToClear;
1155 15 : GetExcptionTransIdsToClear(transIdsToClear);
1156 17 : for (const auto transId : transIdsToClear) {
1157 2 : auto transIdIter = gatheredMbuf_.find(transId);
1158 2 : if (transIdIter == gatheredMbuf_.end()) {
1159 1 : continue;
1160 : }
1161 2 : for (auto& routeLabelIter : transIdIter->second) {
1162 1 : auto& mbufStore = routeLabelIter.second;
1163 1 : aicpusd_info("clear exception for transid[%lu], routelabel[%u]", transIdIter->first, routeLabelIter.first);
1164 1 : mbufStore.Free(&gatheredMbufCntList_);
1165 : }
1166 1 : gatheredMbuf_.erase(transId);
1167 : }
1168 15 : UpdateExcptionTransIdsStatus(transIdsToClear);
1169 15 : }
1170 :
1171 13 : GatherResult AicpuModel::SelectGatheredMbuf(Mbuf*** const mbufPptr, const int32_t timeOut, const uint32_t cacheNum)
1172 : {
1173 13 : std::pair<uint64_t, uint32_t> mbufKey = {UINT64_MAX, UINT32_MAX};
1174 : // first get transids which need to clear, clear them, then mark transid to cleared
1175 13 : ClearExceptionStore();
1176 13 : const auto ret = GatherDequedMbuf(mbufPptr, mbufKey, timeOut, cacheNum);
1177 13 : if (ret != GatherResult::UN_SELECTED) {
1178 3 : aicpusd_info(
1179 : "[%u:%u] is selected in Model[%u], ret is %d", mbufKey.first, mbufKey.second, modelId_,
1180 : static_cast<int32_t>(ret));
1181 3 : ClearDequedMbuf(mbufKey.first, mbufKey.second);
1182 : }
1183 13 : return ret;
1184 : }
1185 :
1186 14 : GatherResult AicpuModel::GatherDequedMbuf(
1187 : Mbuf*** const mbufPptr, std::pair<uint64_t, uint32_t>& mbufKey, const int32_t timeOutMs, const uint32_t cacheNum)
1188 : {
1189 14 : const uint32_t timeOutNs = static_cast<uint32_t>(timeOutMs * TIMES_MS_TO_NS);
1190 14 : AICPUSD_CHECK(timeOutNs < UINT_MAX, GatherResult::UN_SELECTED, "Invalid timeout value %ld ms.", timeOutMs);
1191 14 : const uint64_t minBirthDay = GetCurrentTime() - timeOutNs;
1192 14 : uint64_t count = 0U;
1193 14 : bool timeoutCandidate = false;
1194 14 : bool countoutCandidate = false;
1195 14 : std::pair<uint64_t, uint32_t> timeOutKey = {UINT64_MAX, UINT32_MAX};
1196 14 : std::pair<uint64_t, uint32_t> countOutKey = {UINT64_MAX, UINT32_MAX};
1197 41 : for (auto& transIdIter : gatheredMbuf_) {
1198 28 : auto& routeLabelMap = transIdIter.second;
1199 63 : for (auto& routeLabelIter : routeLabelMap) {
1200 36 : count++;
1201 36 : auto& mbufStore = routeLabelIter.second;
1202 36 : if (!mbufStore.IsReady()) {
1203 35 : if ((timeOutMs > 0) && (mbufStore.BirthDay() < minBirthDay)) {
1204 1 : timeOutKey.first = transIdIter.first;
1205 1 : timeOutKey.second = routeLabelIter.first;
1206 1 : timeoutCandidate = true;
1207 : }
1208 :
1209 35 : if ((cacheNum > 0U) &&
1210 5 : ((transIdIter.first <= countOutKey.first) && (routeLabelIter.first < countOutKey.second))) {
1211 5 : countOutKey.first = transIdIter.first;
1212 5 : countOutKey.second = routeLabelIter.first;
1213 5 : countoutCandidate = true;
1214 : }
1215 35 : continue;
1216 : }
1217 :
1218 1 : mbufKey.first = transIdIter.first;
1219 1 : mbufKey.second = routeLabelIter.first;
1220 1 : if (mbufStore.Consume(mbufPptr, gatheredMbufCntList_)) {
1221 1 : return GatherResult::SELECTED;
1222 : }
1223 : }
1224 : }
1225 :
1226 13 : if ((countoutCandidate && (count > cacheNum))) {
1227 2 : mbufKey.first = countOutKey.first;
1228 2 : mbufKey.second = countOutKey.second;
1229 2 : if (gatheredMbuf_[mbufKey.first][mbufKey.second].Consume(mbufPptr, gatheredMbufCntList_)) {
1230 2 : aicpusd_info("pass [%u:%u] for count[%u] > cache[%u]", mbufKey.first, mbufKey.second, count, cacheNum);
1231 2 : return GatherResult::FAKE_SELECTED;
1232 : }
1233 : }
1234 11 : if (timeoutCandidate) {
1235 1 : mbufKey.first = timeOutKey.first;
1236 1 : mbufKey.second = timeOutKey.second;
1237 1 : if (gatheredMbuf_[mbufKey.first][mbufKey.second].Consume(mbufPptr, gatheredMbufCntList_)) {
1238 1 : aicpusd_info("pass [%u:%u] for timeout[%d] ms", mbufKey.first, mbufKey.second, timeOutMs);
1239 1 : return GatherResult::FAKE_SELECTED;
1240 : }
1241 : }
1242 :
1243 10 : return GatherResult::UN_SELECTED;
1244 : }
1245 :
1246 3 : void AicpuModel::ClearDequedMbuf(const uint64_t transId, const uint32_t routeLabel)
1247 : {
1248 3 : if (gatheredMbuf_[transId][routeLabel].IsEmpty()) {
1249 3 : (void)gatheredMbuf_[transId].erase(routeLabel);
1250 : }
1251 3 : if (gatheredMbuf_[transId].empty()) {
1252 2 : (void)gatheredMbuf_.erase(transId);
1253 : }
1254 3 : }
1255 :
1256 21 : void AicpuModel::ClearGatheredMbuf()
1257 : {
1258 22 : for (auto& transIdIter : gatheredMbuf_) {
1259 2 : for (auto& routeLabelIter : transIdIter.second) {
1260 1 : routeLabelIter.second.Free(&gatheredMbufCntList_);
1261 : }
1262 : }
1263 21 : gatheredMbuf_.clear();
1264 21 : gatheredMbufCntList_.clear();
1265 21 : }
1266 :
1267 5 : void AicpuModel::RecordLockedTable(const uint32_t tableId)
1268 : {
1269 5 : if (tableLocked_.find(tableId) == tableLocked_.end()) {
1270 4 : tableLocked_[tableId] = 1U;
1271 : } else {
1272 1 : tableLocked_[tableId]++;
1273 : }
1274 5 : }
1275 :
1276 2 : void AicpuModel::ClearLockedTable(const uint32_t tableId)
1277 : {
1278 2 : if ((tableLocked_.find(tableId) == tableLocked_.end()) || (tableLocked_[tableId] == 0U)) {
1279 1 : return;
1280 : }
1281 :
1282 1 : if (--tableLocked_[tableId] == 0U) {
1283 1 : (void)tableLocked_.erase(tableId);
1284 : }
1285 : }
1286 :
1287 2 : bool AicpuModel::IsTableLocked(const uint32_t tableId) { return tableLocked_.find(tableId) != tableLocked_.end(); }
1288 :
1289 26 : void AicpuModel::ClearAllLockedTable()
1290 : {
1291 28 : for (auto iter = tableLocked_.begin(); iter != tableLocked_.end(); iter++) {
1292 2 : aicpusd_run_info("Model[%u] has locked table[%u] %u times.", modelId_, iter->first, iter->second);
1293 204 : for (uint32_t i = 0U; i < iter->second; i++) {
1294 202 : TableLockManager::GetInstance().UnLockTable(iter->first);
1295 : }
1296 : }
1297 26 : tableLocked_.clear();
1298 26 : if (tableTryLock_ != INVALID_TABLE_ID) {
1299 1 : aicpusd_run_info("Model[%u] was trying to lock table[%d]", modelId_, tableTryLock_);
1300 1 : tableTryLock_ = INVALID_TABLE_ID;
1301 : }
1302 26 : }
1303 :
1304 7 : uint32_t& AicpuModel::GetInputConsumeNumRef() { return inputConsumeNum_; }
1305 :
1306 20 : size_t AicpuModel::GetCurDequeIndex(const size_t qCnt)
1307 : {
1308 20 : if (gatheredMbufCntList_.empty()) {
1309 5 : aicpusd_info("store list empty use 0th index");
1310 5 : return 0UL;
1311 : }
1312 15 : size_t qIndex = gatheredMbufCntList_.begin()->first;
1313 15 : size_t minCnt = gatheredMbufCntList_.begin()->second;
1314 67 : for (size_t index = 0UL; index < qCnt; index++) {
1315 52 : size_t curCnt = 0UL;
1316 52 : auto iter = gatheredMbufCntList_.find(index);
1317 52 : if (iter != gatheredMbufCntList_.end()) {
1318 44 : curCnt = iter->second;
1319 : }
1320 52 : if (curCnt < minCnt) {
1321 11 : qIndex = index;
1322 11 : minCnt = curCnt;
1323 41 : } else if ((curCnt == minCnt) && (index < qIndex)) {
1324 1 : qIndex = index;
1325 1 : minCnt = curCnt;
1326 : }
1327 52 : aicpusd_info(
1328 : "update modelId:%u, qIndex:%zu, cnt:%llu, index:%zu, curCnt:%zu", modelId_, qIndex, minCnt, index, curCnt);
1329 : }
1330 15 : return qIndex;
1331 : }
1332 :
1333 13 : int32_t AicpuModel::ProcessDataException(const uint64_t transId, const uint32_t type)
1334 : {
1335 13 : std::unique_lock<std::mutex> lockForAsyncTask(mutexForExceptionTrans_);
1336 : // exception occur
1337 13 : if (type == static_cast<uint32_t>(ExceptionAction::ADD)) {
1338 8 : if (exceptionTranses_.find(transId) != exceptionTranses_.end()) {
1339 2 : aicpusd_warn("Transid[%lu] is already in exception list.", transId);
1340 2 : return AICPU_SCHEDULE_OK;
1341 : }
1342 6 : aicpusd_info("Add new exception transid[%lu] for Model[%u].", transId, modelId_);
1343 6 : exceptionTranses_[transId] = false;
1344 6 : AICPUSubEventInfo subEventInfo = {};
1345 6 : subEventInfo.modelId = modelId_;
1346 6 : const auto ret = AicpuMsgSend::SendAICPUSubEvent(
1347 : PtrToPtr<AICPUSubEventInfo, const char_t>(&subEventInfo), static_cast<uint32_t>(sizeof(AICPUSubEventInfo)),
1348 : AICPU_SUB_EVENT_SUPPLY_ENQUEUE, CP_DEFAULT_GROUP_ID, true);
1349 6 : if (ret != AICPU_SCHEDULE_OK) {
1350 1 : aicpusd_warn("Send aicpu subevent failed. Event modelId is %d.", modelId_);
1351 : };
1352 6 : return AICPU_SCHEDULE_OK;
1353 : }
1354 :
1355 5 : if (type == static_cast<uint32_t>(ExceptionAction::EXPIRE)) {
1356 4 : if (exceptionTranses_.find(transId) == exceptionTranses_.end()) {
1357 2 : aicpusd_warn("Transid[%lu] is not in exception list in model[%u], no need to expire.", transId, modelId_);
1358 : } else {
1359 2 : aicpusd_info("Transid[%lu] is expired in model[%u]", transId, modelId_);
1360 2 : exceptionTranses_.erase(transId);
1361 : }
1362 4 : return AICPU_SCHEDULE_OK;
1363 : }
1364 :
1365 1 : aicpusd_err(
1366 : "Model[%u] failed to process data exception of transid[%lu] for type[%u] is invalid.", modelId_, transId, type);
1367 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
1368 13 : }
1369 :
1370 15 : bool AicpuModel::IsTransIdException(const uint64_t transId)
1371 : {
1372 15 : std::unique_lock<std::mutex> lockForAsyncTask(mutexForExceptionTrans_);
1373 30 : return (exceptionTranses_.find(transId) != exceptionTranses_.end());
1374 15 : }
1375 :
1376 15 : void AicpuModel::GetExcptionTransIdsToClear(std::vector<int64_t>& excptionTransIdsToClear)
1377 : {
1378 15 : std::unique_lock<std::mutex> lockForAsyncTask(mutexForExceptionTrans_);
1379 17 : for (const auto& elem : exceptionTranses_) {
1380 2 : if (!elem.second) {
1381 2 : excptionTransIdsToClear.emplace_back(elem.first);
1382 : }
1383 : }
1384 15 : }
1385 :
1386 15 : void AicpuModel::UpdateExcptionTransIdsStatus(const std::vector<int64_t> excptionTransIdsCleared)
1387 : {
1388 15 : std::unique_lock<std::mutex> lockForAsyncTask(mutexForExceptionTrans_);
1389 17 : for (const auto transId : excptionTransIdsCleared) {
1390 2 : auto iter = exceptionTranses_.find(transId);
1391 2 : if (iter != exceptionTranses_.end()) {
1392 2 : aicpusd_info("update transId[%lu] to cleared in model[%u]", transId, modelId_);
1393 2 : iter->second = true;
1394 : }
1395 : }
1396 15 : }
1397 :
1398 20 : bool QueueMbufStore::Init(const size_t queueNum)
1399 : {
1400 20 : if (!queuesLists_.empty()) {
1401 7 : return true;
1402 : }
1403 13 : AICPUSD_EXCEPTION_CATCH(queuesLists_.resize(queueNum), return false);
1404 13 : birthTimeStamp_ = GetCurrentTime();
1405 13 : return true;
1406 : }
1407 :
1408 17 : bool QueueMbufStore::Store(const size_t qIndex, Mbuf* const mbuf, std::map<size_t, uint64_t>& gCntList)
1409 : {
1410 17 : if (queuesLists_.empty()) {
1411 1 : aicpusd_err("QueueMbufStore has not been initialized");
1412 1 : return false;
1413 : }
1414 16 : if (qIndex >= queuesLists_.size()) {
1415 1 : aicpusd_err("qIndex [%zu] invalid", qIndex);
1416 1 : return false;
1417 : }
1418 15 : queuesLists_[qIndex].push_back(mbuf);
1419 15 : gCntList[qIndex]++;
1420 15 : aicpusd_info(
1421 : "store mbuf index:%zu, queue list size:%zu, buf cnt map cnt:%llu", qIndex, queuesLists_[qIndex].size(),
1422 : gCntList[qIndex]);
1423 15 : return true;
1424 : }
1425 :
1426 5 : bool QueueMbufStore::Consume(Mbuf*** mbufPptr, std::map<size_t, uint64_t>& gCntList)
1427 : {
1428 5 : if (queuesLists_.empty()) {
1429 1 : aicpusd_err("QueueMbufStore has not been initialized");
1430 1 : return false;
1431 : }
1432 15 : for (size_t arrIndex = 0U; arrIndex < queuesLists_.size(); ++arrIndex) {
1433 11 : if (!queuesLists_[arrIndex].empty()) {
1434 8 : *mbufPptr[arrIndex] = queuesLists_[arrIndex].front();
1435 8 : queuesLists_[arrIndex].pop_front();
1436 8 : if (gCntList[arrIndex] > 0) {
1437 8 : gCntList[arrIndex]--;
1438 : }
1439 8 : aicpusd_info(
1440 : "consume mbuf index:%zu, queue list size:%zu, buf cnt map cnt:%llu", arrIndex,
1441 : queuesLists_[arrIndex].size(), gCntList[arrIndex]);
1442 : } else {
1443 3 : *mbufPptr[arrIndex] = nullptr;
1444 : }
1445 : }
1446 4 : return true;
1447 : }
1448 :
1449 17 : void QueueMbufStore::Free(std::map<size_t, uint64_t>* const gCntList)
1450 : {
1451 49 : for (size_t i = 0UL; i < queuesLists_.size(); i++) {
1452 71 : while (!queuesLists_[i].empty()) {
1453 7 : Mbuf* tmpBuf = queuesLists_[i].front();
1454 7 : if (tmpBuf != nullptr) {
1455 5 : (void)halMbufFree(tmpBuf);
1456 : }
1457 7 : queuesLists_[i].pop_front();
1458 7 : if ((gCntList != nullptr) && ((*gCntList)[i] > 0)) {
1459 3 : (*gCntList)[i]--;
1460 : }
1461 : }
1462 32 : if (gCntList != nullptr) {
1463 3 : aicpusd_info("free store buffer index:%zu, buffer map count:%llu", i, (*gCntList)[i]);
1464 : }
1465 : }
1466 17 : }
1467 :
1468 39 : bool QueueMbufStore::IsReady() const
1469 : {
1470 39 : if (queuesLists_.empty()) {
1471 1 : return false;
1472 : }
1473 74 : for (size_t index = 0UL; index < queuesLists_.size(); index++) {
1474 73 : if (queuesLists_[index].empty()) {
1475 37 : return false;
1476 : }
1477 : }
1478 1 : return true;
1479 : }
1480 15 : QueueMbufStore::~QueueMbufStore() { Free(); }
1481 :
1482 6 : bool QueueMbufStore::IsEmpty() const
1483 : {
1484 6 : if (queuesLists_.empty()) {
1485 1 : return true;
1486 : }
1487 23 : for (size_t index = 0UL; index < queuesLists_.size(); index++) {
1488 19 : if (!queuesLists_[index].empty()) {
1489 1 : return false;
1490 : }
1491 : }
1492 4 : return true;
1493 : }
1494 : } // namespace AicpuSchedule
|
