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 : namespace AicpuSchedule {
18 : namespace {
19 : const std::string KERNEL_MODEL_BATCH_DEQUEUE = "modelBatchDequeue";
20 : } // namespace
21 :
22 2 : int32_t OperatorKernelModelBatchDequeue::Compute(const AicpuTaskInfo& kernelTaskInfo, const RunContext& taskContext)
23 : {
24 2 : aicpusd_info("Begin to batch dequeue. modelId[%u].", taskContext.modelId);
25 2 : BatchDequeueInfo batchDeqInfo = {};
26 2 : auto ret = CheckAndParseBatchDequeueParams(kernelTaskInfo, taskContext, batchDeqInfo);
27 2 : if (ret != AICPU_SCHEDULE_OK) {
28 1 : return ret;
29 : }
30 1 : auto& inputsIsDequeue = AicpuModelManager::GetInstance().GetModel(taskContext.modelId)->MutableInputsIsDequeue();
31 1 : aicpusd_info("batch dequeue for %u queues.", batchDeqInfo.inputNums);
32 2 : for (uint32_t i = 0U; i < batchDeqInfo.inputNums; ++i) {
33 1 : if (inputsIsDequeue[i]) {
34 0 : aicpusd_info("the [%u]th queue has been dequed successfully", i);
35 0 : continue;
36 : }
37 1 : BufEnQueueInfo queueInfo = {batchDeqInfo.queueIds[i], batchDeqInfo.mbufAddrs[i]};
38 1 : ret = DoModelDequeue(queueInfo, taskContext);
39 1 : if (ret != AICPU_SCHEDULE_OK) {
40 0 : inputsIsDequeue.assign(inputsIsDequeue.size(), false);
41 0 : return ret;
42 : }
43 1 : if (taskContext.pending) {
44 0 : return AICPU_SCHEDULE_OK;
45 : }
46 1 : inputsIsDequeue[i] = true;
47 : }
48 1 : if (batchDeqInfo.alignOffsets != nullptr) {
49 0 : ret = AlignBatchDequeue(batchDeqInfo, taskContext);
50 0 : if ((ret != AICPU_SCHEDULE_OK) || (taskContext.pending)) {
51 0 : return ret;
52 : }
53 : }
54 :
55 1 : inputsIsDequeue.assign(inputsIsDequeue.size(), false);
56 1 : return ret;
57 : }
58 :
59 4 : int32_t OperatorKernelModelBatchDequeue::CheckAndParseBatchDequeueParams(
60 : const AicpuTaskInfo& kernelTaskInfo, const RunContext& taskContext, BatchDequeueInfo& batchDeqInfo) const
61 : {
62 4 : const BatchDequeueDesc* const batchDeqDesc = PtrToPtr<void, BatchDequeueDesc>(ValueToPtr(kernelTaskInfo.paraBase));
63 4 : if (batchDeqDesc == nullptr) {
64 1 : aicpusd_err(
65 : "KernelTaskInfo paramBase is null, modelId[%u], streamId[%u], taskId[%u].", taskContext.modelId,
66 : taskContext.streamId, kernelTaskInfo.taskID);
67 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
68 : }
69 :
70 3 : const auto model = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
71 3 : if (model == nullptr) {
72 1 : aicpusd_err(
73 : "Cannot get model by modelId:[%u], streamId[%u], taskId[%u].", taskContext.modelId, taskContext.streamId,
74 : 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(
80 : "KernelTaskInfo inputNums[%u] is not equal model input queue size[%zu],"
81 : "modelId[%u], streamId[%u], taskId[%u]",
82 : batchDeqDesc->inputNums, inputsIsDequeue.size(), taskContext.modelId, taskContext.streamId,
83 : kernelTaskInfo.taskID);
84 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
85 : }
86 :
87 1 : batchDeqInfo.inputNums = batchDeqDesc->inputNums;
88 1 : batchDeqInfo.alignInterval = batchDeqDesc->alignInterval;
89 1 : batchDeqInfo.alignOffsets = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqDesc->alignOffsetsAddr));
90 1 : batchDeqInfo.queueIds = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqDesc->queueIdsAddr));
91 1 : if (batchDeqInfo.queueIds == nullptr) {
92 0 : aicpusd_err(
93 : "KernelTaskInfo queueIds is null, modelId[%u], streamId[%u], taskId[%u]", taskContext.modelId,
94 : taskContext.streamId, kernelTaskInfo.taskID);
95 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
96 : }
97 1 : batchDeqInfo.mbufAddrs = PtrToPtr<void, uint64_t>(ValueToPtr(batchDeqDesc->mbufAddrsAddr));
98 1 : if (batchDeqInfo.mbufAddrs == nullptr) {
99 0 : aicpusd_err(
100 : "KernelTaskInfo mbufAddrs is null, modelId[%u], streamId[%u], taskId[%u]", taskContext.modelId,
101 : taskContext.streamId, kernelTaskInfo.taskID);
102 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
103 : }
104 1 : return AICPU_SCHEDULE_OK;
105 : }
106 :
107 2 : int32_t OperatorKernelModelBatchDequeue::DoModelDequeue(BufEnQueueInfo& bufInfo, const RunContext& taskContext) const
108 : {
109 2 : return DequeueTask(bufInfo, taskContext, true);
110 : }
111 :
112 : // if max(inputs timestamp-alignOffset)-min(inputs timestamp-alignOffset) < alignInterval, return ok
113 : // else delete oldest data, then re-dequeue data until inputs timestamp alignment is satisfied or the queue is empty.
114 2 : int32_t OperatorKernelModelBatchDequeue::AlignBatchDequeue(
115 : BatchDequeueInfo& batchDeqInfo, const RunContext& taskContext)
116 : {
117 : // model has been checked, not nullptr
118 2 : auto& inputsIsDequeue = AicpuModelManager::GetInstance().GetModel(taskContext.modelId)->MutableInputsIsDequeue();
119 : while (true) {
120 2 : uint32_t maxAlignTimestamp = 0U;
121 2 : uint32_t minAlignTimestamp = UINT32_MAX;
122 2 : uint32_t minTimestampIndex = 0U;
123 2 : auto ret = AlignTimestamp(batchDeqInfo, taskContext, maxAlignTimestamp, minAlignTimestamp, minTimestampIndex);
124 2 : if (ret != AICPU_SCHEDULE_OK) {
125 2 : return ret;
126 : }
127 2 : if ((maxAlignTimestamp - minAlignTimestamp) <= batchDeqInfo.alignInterval) {
128 1 : return AICPU_SCHEDULE_OK;
129 : }
130 : BufEnQueueInfo queueInfo = {
131 1 : batchDeqInfo.queueIds[minTimestampIndex], batchDeqInfo.mbufAddrs[minTimestampIndex]};
132 1 : ret = DoModelDequeue(queueInfo, taskContext);
133 1 : if (ret != AICPU_SCHEDULE_OK) {
134 0 : return ret;
135 : }
136 1 : if (taskContext.pending) {
137 1 : inputsIsDequeue[minTimestampIndex] = false;
138 1 : return AICPU_SCHEDULE_OK;
139 : }
140 0 : }
141 : return AICPU_SCHEDULE_OK;
142 : }
143 :
144 6 : REGISTER_OPERATOR_KERNEL(KERNEL_MODEL_BATCH_DEQUEUE, OperatorKernelModelBatchDequeue);
145 : } // namespace AicpuSchedule
|
