Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "operator_kernel_model_batch_dequeue.h"
12 :
13 : #include "aicpusd_status.h"
14 : #include "aicpusd_profiler.h"
15 : #include "aicpusd_model_execute.h"
16 :
17 :
18 : namespace AicpuSchedule {
19 : namespace {
20 : const std::string KERNEL_MODEL_BATCH_DEQUEUE = "modelBatchDequeue";
21 : } // namespace
22 :
23 2 : int32_t OperatorKernelModelBatchDequeue::Compute(const AicpuTaskInfo &kernelTaskInfo, const RunContext &taskContext)
24 : {
25 2 : aicpusd_info("Begin to batch dequeue. modelId[%u].", taskContext.modelId);
26 2 : BatchDequeueInfo batchDeqInfo = {};
27 2 : auto ret = CheckAndParseBatchDequeueParams(kernelTaskInfo, taskContext, batchDeqInfo);
28 2 : if (ret != AICPU_SCHEDULE_OK) {
29 1 : return ret;
30 : }
31 1 : auto &inputsIsDequeue = AicpuModelManager::GetInstance().GetModel(taskContext.modelId)->MutableInputsIsDequeue();
32 1 : aicpusd_info("batch dequeue for %u queues.", batchDeqInfo.inputNums);
33 2 : for (uint32_t i = 0U; i < batchDeqInfo.inputNums; ++i) {
34 1 : if (inputsIsDequeue[i]) {
35 0 : aicpusd_info("the [%u]th queue has been dequed successfully", i);
36 0 : continue;
37 : }
38 1 : BufEnQueueInfo queueInfo = { batchDeqInfo.queueIds[i], batchDeqInfo.mbufAddrs[i] };
39 1 : ret = DoModelDequeue(queueInfo, taskContext);
40 1 : if (ret != AICPU_SCHEDULE_OK) {
41 0 : inputsIsDequeue.assign(inputsIsDequeue.size(), false);
42 0 : return ret;
43 : }
44 1 : if (taskContext.pending) {
45 0 : return AICPU_SCHEDULE_OK;
46 : }
47 1 : inputsIsDequeue[i] = true;
48 : }
49 1 : if (batchDeqInfo.alignOffsets != nullptr) {
50 0 : ret = AlignBatchDequeue(batchDeqInfo, taskContext);
51 0 : if ((ret != AICPU_SCHEDULE_OK) || (taskContext.pending)) {
52 0 : return ret;
53 : }
54 : }
55 :
56 1 : inputsIsDequeue.assign(inputsIsDequeue.size(), false);
57 1 : return ret;
58 : }
59 :
60 4 : int32_t OperatorKernelModelBatchDequeue::CheckAndParseBatchDequeueParams(const AicpuTaskInfo &kernelTaskInfo,
61 : const RunContext &taskContext, BatchDequeueInfo &batchDeqInfo) const
62 : {
63 : const BatchDequeueDesc *const batchDeqDesc =
64 4 : PtrToPtr<void, BatchDequeueDesc>(ValueToPtr(kernelTaskInfo.paraBase));
65 4 : if (batchDeqDesc == nullptr) {
66 1 : aicpusd_err("KernelTaskInfo paramBase is null, modelId[%u], streamId[%u], taskId[%u].",
67 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
68 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
69 : }
70 :
71 3 : const auto model = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
72 3 : if (model == nullptr) {
73 1 : aicpusd_err("Cannot get model by modelId:[%u], streamId[%u], taskId[%u].",
74 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
75 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
76 : }
77 2 : auto &inputsIsDequeue = model->MutableInputsIsDequeue();
78 2 : if (batchDeqDesc->inputNums != inputsIsDequeue.size()) {
79 1 : aicpusd_err("KernelTaskInfo inputNums[%u] is not equal model input queue size[%zu],"
80 : "modelId[%u], streamId[%u], taskId[%u]", batchDeqDesc->inputNums, inputsIsDequeue.size(),
81 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
82 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
83 : }
84 :
85 1 : batchDeqInfo.inputNums = batchDeqDesc->inputNums;
86 1 : batchDeqInfo.alignInterval = batchDeqDesc->alignInterval;
87 1 : batchDeqInfo.alignOffsets = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqDesc->alignOffsetsAddr));
88 1 : batchDeqInfo.queueIds = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqDesc->queueIdsAddr));
89 1 : if (batchDeqInfo.queueIds == nullptr) {
90 0 : aicpusd_err("KernelTaskInfo queueIds is null, modelId[%u], streamId[%u], taskId[%u]",
91 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
92 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
93 : }
94 1 : batchDeqInfo.mbufAddrs = PtrToPtr<void, uint64_t>(ValueToPtr(batchDeqDesc->mbufAddrsAddr));
95 1 : if (batchDeqInfo.mbufAddrs == nullptr) {
96 0 : aicpusd_err("KernelTaskInfo mbufAddrs is null, modelId[%u], streamId[%u], taskId[%u]",
97 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
98 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
99 : }
100 1 : return AICPU_SCHEDULE_OK;
101 : }
102 :
103 2 : int32_t OperatorKernelModelBatchDequeue::DoModelDequeue(BufEnQueueInfo &bufInfo, const RunContext &taskContext) const {
104 2 : return DequeueTask(bufInfo, taskContext, true);
105 : }
106 :
107 : // if max(inputs timestamp-alignOffset)-min(inputs timestamp-alignOffset) < alignInterval, return ok
108 : // else delete oldest data, then re-dequeue data until inputs timestamp alignment is satisfied or the queue is empty.
109 2 : int32_t OperatorKernelModelBatchDequeue::AlignBatchDequeue(BatchDequeueInfo &batchDeqInfo,
110 : const RunContext &taskContext)
111 : {
112 : // model has been checked, not nullptr
113 : auto &inputsIsDequeue =
114 2 : AicpuModelManager::GetInstance().GetModel(taskContext.modelId)->MutableInputsIsDequeue();
115 : while (true) {
116 2 : uint32_t maxAlignTimestamp = 0U;
117 2 : uint32_t minAlignTimestamp = UINT32_MAX;
118 2 : uint32_t minTimestampIndex = 0U;
119 2 : auto ret = AlignTimestamp(batchDeqInfo, taskContext, maxAlignTimestamp, minAlignTimestamp, minTimestampIndex);
120 2 : if (ret != AICPU_SCHEDULE_OK) {
121 2 : return ret;
122 : }
123 2 : if ((maxAlignTimestamp - minAlignTimestamp) <= batchDeqInfo.alignInterval) {
124 1 : return AICPU_SCHEDULE_OK;
125 : }
126 : BufEnQueueInfo queueInfo = {
127 1 : batchDeqInfo.queueIds[minTimestampIndex], batchDeqInfo.mbufAddrs[minTimestampIndex] };
128 1 : ret = DoModelDequeue(queueInfo, taskContext);
129 1 : if (ret != AICPU_SCHEDULE_OK) {
130 0 : return ret;
131 : }
132 1 : if (taskContext.pending) {
133 1 : inputsIsDequeue[minTimestampIndex] = false;
134 1 : return AICPU_SCHEDULE_OK;
135 : }
136 0 : }
137 : return AICPU_SCHEDULE_OK;
138 : }
139 :
140 6 : REGISTER_OPERATOR_KERNEL(KERNEL_MODEL_BATCH_DEQUEUE, OperatorKernelModelBatchDequeue);
141 : } // namespace AicpuSchedule
|
