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_buff.h"
12 :
13 : #include "aicpusd_status.h"
14 : #include "aicpusd_monitor.h"
15 : #include "aicpusd_model_execute.h"
16 : #include "aicpusd_resource_manager.h"
17 : #include "operator_kernel_common.h"
18 :
19 :
20 : namespace AicpuSchedule {
21 : namespace {
22 : const std::string KERNEL_MODEL_BATCH_DEQUEUE_BUFF = "modelBatchDequeueBuff";
23 : } // namespace
24 :
25 4 : int32_t OperatorKernelModelBatchDequeueBuff::Compute(const AicpuTaskInfo &kernelTaskInfo, const RunContext &taskContext)
26 : {
27 4 : aicpusd_info("Begin to batch dequeue buff. modelId[%u].", taskContext.modelId);
28 : const BatchDequeueBuffDesc *const batchDeqBufDesc =
29 4 : PtrToPtr<void, BatchDequeueBuffDesc>(ValueToPtr(kernelTaskInfo.paraBase));
30 4 : if (batchDeqBufDesc == nullptr) {
31 1 : aicpusd_err("KernelTaskInfo paramBase is null, modelId[%u], streamId[%u], taskId[%u].",
32 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
33 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
34 : }
35 3 : BatchDequeueBuffInfo batchDeqBufInfo = {};
36 3 : auto ret = CheckAndParseBatchDeqBufParams(batchDeqBufDesc, kernelTaskInfo, taskContext, batchDeqBufInfo);
37 3 : if (ret != AICPU_SCHEDULE_OK) {
38 1 : return ret;
39 : }
40 :
41 2 : aicpusd_info("batch dequeue for %u queues", batchDeqBufInfo.inputNums);
42 3 : for (uint32_t i = 0U; i < batchDeqBufInfo.inputNums; ++i) {
43 : BufEnQueueBuffInfo queueInfo = {
44 2 : batchDeqBufInfo.queueIds[i], batchDeqBufInfo.deviceIds[i], batchDeqBufInfo.mbufAddrs[i]};
45 2 : const auto res = ModelAttachAndDequeueBuff(queueInfo, taskContext);
46 2 : if (res == AICPU_SCHEDULE_ERROR_MODEL_UNLOAD) {
47 1 : aicpusd_warn("model is destroy");
48 1 : bool * const pending = const_cast<bool *>(&taskContext.pending);
49 1 : *pending = true;
50 1 : return AICPU_SCHEDULE_OK;
51 : }
52 : }
53 :
54 1 : if (batchDeqBufInfo.alignOffsets != nullptr) {
55 0 : ret = AlignBatchDequeueBuff(batchDeqBufInfo, taskContext);
56 : }
57 1 : return ret;
58 : }
59 :
60 2 : int32_t OperatorKernelModelBatchDequeueBuff::CheckAndParseBatchDeqBufParams(
61 : const BatchDequeueBuffDesc *const batchDeqBufDesc, const AicpuTaskInfo &kernelTaskInfo,
62 : const RunContext &taskContext, BatchDequeueBuffInfo &batchDeqBufInfo) const
63 : {
64 2 : batchDeqBufInfo.inputNums = batchDeqBufDesc->inputNums;
65 2 : batchDeqBufInfo.alignInterval = batchDeqBufDesc->alignInterval;
66 2 : batchDeqBufInfo.alignOffsets = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqBufDesc->alignOffsetsAddr));
67 2 : batchDeqBufInfo.queueIds = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqBufDesc->queueIdsAddr));
68 2 : if (batchDeqBufInfo.queueIds == nullptr) {
69 0 : aicpusd_err("KernelTaskInfo queueIds is null, modelId[%u], streamId[%u], taskId[%u]",
70 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
71 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
72 : }
73 2 : batchDeqBufInfo.mbufAddrs = PtrToPtr<void, uint64_t>(ValueToPtr(batchDeqBufDesc->mbufAddrsAddr));
74 2 : if (batchDeqBufInfo.mbufAddrs == nullptr) {
75 0 : aicpusd_err("KernelTaskInfo mbufAddrs is null, modelId[%u], streamId[%u], taskId[%u]",
76 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
77 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
78 : }
79 2 : batchDeqBufInfo.deviceIds = PtrToPtr<void, int32_t>(ValueToPtr(batchDeqBufDesc->deviceIdAddr));
80 2 : if (batchDeqBufInfo.deviceIds == nullptr) {
81 0 : aicpusd_err("KernelTaskInfo deviceIds is null, modelId[%u], streamId[%u], taskId[%u]",
82 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
83 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
84 : }
85 2 : return AICPU_SCHEDULE_OK;
86 : }
87 :
88 3 : int32_t OperatorKernelModelBatchDequeueBuff::ModelAttachAndDequeueBuff(BufEnQueueBuffInfo &queueInfo,
89 : const RunContext &taskContext,
90 : bool tryOnce) const
91 : {
92 3 : const auto drvRet = halQueueAttach(static_cast<uint32_t>(queueInfo.deviceId), queueInfo.queueID, 0);
93 3 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
94 0 : aicpusd_err("Aicpusd attached queue[%u] failed ret[%d]", queueInfo.queueID, drvRet);
95 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
96 : }
97 3 : aicpusd_info("queue id %u device id %d.", queueInfo.queueID, queueInfo.deviceId);
98 3 : return ModelDequeueBuffTaskKernel(queueInfo, taskContext, tryOnce);
99 : }
100 :
101 6 : int32_t OperatorKernelModelBatchDequeueBuff::ModelDequeueBuffTaskKernel(BufEnQueueBuffInfo &bufInfo,
102 : const RunContext &taskContext,
103 : bool tryOnce) const
104 : {
105 6 : Mbuf ** const mBufPptr = PtrToPtr<void, Mbuf*>(ValueToPtr(bufInfo.mBufPtr));
106 6 : if (mBufPptr == nullptr) {
107 1 : aicpusd_err("param mBufPptr is null.");
108 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
109 : }
110 5 : const auto model = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
111 5 : if (model == nullptr) {
112 0 : aicpusd_err("Cannot get model by modelId:[%u], streamId[%u].",
113 : taskContext.modelId, taskContext.streamId);
114 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
115 : }
116 5 : const uint32_t queueId = bufInfo.queueID;
117 5 : const int32_t deviceId = bufInfo.deviceId;
118 5 : uint64_t respLen = 0U;
119 : while (true) {
120 5 : if (model->GetModelDestroyStatus()) {
121 2 : aicpusd_info("model is unloading, exit");
122 2 : return AICPU_SCHEDULE_ERROR_MODEL_UNLOAD;
123 : }
124 3 : const auto ret = halQueuePeek(static_cast<uint32_t>(deviceId), queueId, &respLen, 0);
125 3 : if ((ret == DRV_ERROR_NONE) && (respLen != 0U)) {
126 2 : aicpusd_info("halQueuePeek success");
127 2 : break;
128 : }
129 1 : if (tryOnce) {
130 1 : return AICPU_SCHEDULE_OK;
131 : }
132 0 : }
133 :
134 2 : const int32_t deqRet = DequeueBuff(bufInfo, taskContext, respLen, queueId, deviceId);
135 2 : if (deqRet != AICPU_SCHEDULE_OK) {
136 1 : aicpusd_warn("DequeueBuff failed from queue[%u] in device[%u] failed, deqRet[%d], respLen[%u]",
137 : queueId, deviceId, static_cast<int32_t>(deqRet), respLen);
138 : }
139 2 : return deqRet;
140 : }
141 :
142 2 : int32_t OperatorKernelModelBatchDequeueBuff::DequeueBuff(BufEnQueueBuffInfo &bufInfo, const RunContext &taskContext,
143 : const uint64_t respLen, const uint32_t queueId,
144 : const int32_t deviceId) const
145 : {
146 : // mBufPptr在之前已经做过判空处理，此处不在进行校验
147 2 : Mbuf ** const mBufPptr = reinterpret_cast<Mbuf **>(static_cast<uintptr_t>(bufInfo.mBufPtr));
148 2 : Mbuf *outMBuf = BufManager::GetInstance().MallocAndGuardBuf(respLen, taskContext.modelId);
149 2 : if (outMBuf == nullptr) {
150 1 : aicpusd_err("Failed to alloc mbuf, respLen[%u], modelId[%u].", respLen, taskContext.modelId);
151 1 : AicpuMonitor::GetInstance().SendKillMsgToTsd();
152 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
153 : }
154 1 : auto drvRet = halMbufSetDataLen(outMBuf, static_cast<uint64_t>(respLen));
155 1 : if (drvRet != DRV_ERROR_NONE) {
156 0 : aicpusd_err("halMbufSetDataLen error, queue id:%u, drvRet=%d", queueId, drvRet);
157 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
158 : }
159 :
160 1 : uint32_t headSize = 0U;
161 1 : void *headBuf = nullptr;
162 1 : drvRet = halMbufGetPrivInfo(outMBuf, &headBuf, &headSize);
163 1 : if (drvRet != DRV_ERROR_NONE) {
164 0 : aicpusd_err("Failed to get head info in input information, ret[%d].", drvRet);
165 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
166 : }
167 :
168 1 : void *dataAddrPtr = nullptr;
169 1 : drvRet = OperatorKernelCommon::GetMbufDataPtr(static_cast<uint64_t>(reinterpret_cast<uintptr_t>(&outMBuf)),
170 : &dataAddrPtr);
171 1 : if (drvRet != AICPU_SCHEDULE_OK) {
172 0 : aicpusd_err("Failed to get mbuf data addr. ret is [%d]", drvRet);
173 0 : return drvRet;
174 : }
175 :
176 1 : constexpr size_t totalLen = sizeof(struct buff_iovec) + sizeof(struct iovec_info);
177 1 : std::unique_ptr<char_t[]> vecUniquePtr(new (std::nothrow) char_t[totalLen], std::default_delete<char_t[]>());
178 1 : if (vecUniquePtr == nullptr) {
179 0 : aicpusd_err("failed to alloc memory for vecUniquePtr, size[%zu].", totalLen);
180 0 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
181 : }
182 1 : buff_iovec * const buffvec = reinterpret_cast<buff_iovec *>(vecUniquePtr.get());
183 1 : buffvec->context_base = headBuf;
184 1 : buffvec->context_len = headSize;
185 1 : buffvec->count = 1U;
186 1 : buffvec->ptr[0U].iovec_base = dataAddrPtr;
187 1 : buffvec->ptr[0U].len = respLen;
188 1 : drvRet = halQueueDeQueueBuff(static_cast<uint32_t>(deviceId), queueId, buffvec, -1);
189 1 : if (drvRet != DRV_ERROR_NONE) {
190 0 : aicpusd_err("halQueueDeQueueBuff to queue[%u] in device[%d] failed, error[%d]",
191 : queueId, deviceId, static_cast<int32_t>(drvRet));
192 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
193 : }
194 :
195 1 : *mBufPptr = PtrToPtr<void, Mbuf>(outMBuf);
196 :
197 1 : (void)ProcessMbufHeadInDequeueTask(taskContext.modelId, headBuf, headSize);
198 1 : (void)SetModelEndOfSequence(taskContext.modelId, headBuf, headSize);
199 :
200 1 : OperatorKernelCommon::TraceQueueData(taskContext, headBuf, headSize, "DequeuedBuff");
201 1 : return AICPU_SCHEDULE_OK;
202 1 : }
203 :
204 : // if max(inputs timestamp-alignOffset)-min(inputs timestamp-alignOffset) < alignInterval, return ok
205 : // else delete oldest data, then re-dequeue data until inputs timestamp alignment is satisfied or the queue is empty.
206 3 : int32_t OperatorKernelModelBatchDequeueBuff::AlignBatchDequeueBuff(BatchDequeueBuffInfo &batchDeqBufInfo,
207 : const RunContext &taskContext) const
208 : {
209 3 : BatchDequeueInfo batchDeqInfo = {batchDeqBufInfo.inputNums, batchDeqBufInfo.alignInterval,
210 3 : batchDeqBufInfo.alignOffsets, batchDeqBufInfo.queueIds, batchDeqBufInfo.mbufAddrs};
211 : while (true) {
212 3 : uint32_t maxAlignTimestamp = 0U;
213 3 : uint32_t minAlignTimestamp = UINT32_MAX;
214 3 : uint32_t minTimestampIndex = 0U;
215 3 : auto ret = AlignTimestamp(batchDeqInfo, taskContext, maxAlignTimestamp, minAlignTimestamp, minTimestampIndex);
216 3 : if (ret != AICPU_SCHEDULE_OK) {
217 0 : aicpusd_err("AlignTimestamp failed! ret[%ld]", ret);
218 3 : return ret;
219 : }
220 3 : if ((maxAlignTimestamp - minAlignTimestamp) <= batchDeqInfo.alignInterval) {
221 1 : return AICPU_SCHEDULE_OK;
222 : }
223 2 : BufEnQueueBuffInfo queBuffInfo = {batchDeqBufInfo.queueIds[minTimestampIndex],
224 2 : batchDeqBufInfo.deviceIds[minTimestampIndex], batchDeqBufInfo.mbufAddrs[minTimestampIndex]};
225 2 : ret = ModelDequeueBuffTaskKernel(queBuffInfo, taskContext);
226 2 : if (ret == AICPU_SCHEDULE_ERROR_MODEL_UNLOAD) {
227 1 : bool * const pending = const_cast<bool *>(&taskContext.pending);
228 1 : *pending = true;
229 1 : return AICPU_SCHEDULE_OK;
230 : }
231 1 : if (ret != AICPU_SCHEDULE_OK) {
232 1 : aicpusd_err("ModelDequeueBuffTaskKernel failed! ret[%ld]", ret);
233 1 : return ret;
234 : }
235 0 : }
236 : return AICPU_SCHEDULE_OK;
237 : }
238 :
239 6 : REGISTER_OPERATOR_KERNEL(KERNEL_MODEL_BATCH_DEQUEUE_BUFF, OperatorKernelModelBatchDequeueBuff);
240 : } // namespace AicpuSchedule
|
