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