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_prepare_dynamic_input_output.h"
12 :
13 : #include "aicpusd_status.h"
14 : #include "aicpusd_model_execute.h"
15 : #include "aicpusd_resource_manager.h"
16 : #include "operator_kernel_common.h"
17 :
18 :
19 : namespace AicpuSchedule {
20 : namespace {
21 : const std::string KERNEL_PREPARE_DYNAMIC_INPUT_OUTPUT = "prepareDynamicInputOutput";
22 : const std::string KERNEL_PREPARE_DYNAMIC_INPUT_OUTPUT_V2 = "prepareDynamicInputOutputV2";
23 : } // namespace
24 :
25 4 : int32_t PrepareDynamicInputOutputBase::PrepareDynamicInputOutput(const AicpuTaskInfo &kernelTaskInfo,
26 : const RunContext &taskContext,
27 : const bool hostAllocDynamicOutput) const
28 : {
29 4 : aicpusd_info("Start ModelPrepareDynamicInputOutput. modelId=%u, streamId=%u, taskId=%u.",
30 : taskContext.modelId, kernelTaskInfo.streamID, kernelTaskInfo.taskID);
31 4 : if (kernelTaskInfo.paraBase == 0UL) {
32 1 : aicpusd_err("kernelTaskInfo.paraBase is null");
33 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
34 : }
35 :
36 : const PrepareDynamicInputOutputKernelArgs * const param =
37 3 : PtrToPtr<void, PrepareDynamicInputOutputKernelArgs>(ValueToPtr(kernelTaskInfo.paraBase));
38 3 : if (((param->inputsNum != 0U) && ((param->inputDynamicFlagsAddr == 0U) || (param->inputMbufAddrsAddr == 0U))) ||
39 3 : ((param->outputsNum != 0U) && ((param->outputTensorSizesAddr == 0U) || (param->outputMbufAddrsAddr == 0U))) ||
40 3 : (param->reqMsgMbufAddr == 0U)) {
41 1 : aicpusd_err("input or output invalid, input: {inputNums[%u], inputDynamicFlagsAddr[%u], inputMbufAddr[%u]}, "
42 : "output: {outputNums[%u], outputTensorSizeAddr[%u], outputMbufAddr[%u]}, reqMsgMbufAddr[%u].",
43 : param->inputsNum, param->inputDynamicFlagsAddr, param->inputMbufAddrsAddr,
44 : param->outputsNum, param->outputTensorSizesAddr, param->outputMbufAddrsAddr,
45 : param->reqMsgMbufAddr);
46 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
47 : }
48 :
49 2 : std::vector<Mbuf *> mbufsToFree;
50 0 : const ScopeGuard mbufGuard([&mbufsToFree, &taskContext]() {
51 2 : AicpuModel *modelPtr = nullptr;
52 2 : if (!mbufsToFree.empty()) {
53 0 : modelPtr = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
54 0 : if (modelPtr == nullptr) {
55 0 : aicpusd_err("cannot get aicpuModel by modelId:[%u]!", taskContext.modelId);
56 : }
57 : }
58 2 : for (const auto mbuf : mbufsToFree) {
59 0 : if (modelPtr != nullptr) {
60 0 : (void) modelPtr->UnGardModelBuf(mbuf);
61 : }
62 0 : (void) halMbufFree(mbuf);
63 : }
64 4 : });
65 :
66 2 : int32_t ret = AllocateAndInitOutput(param, taskContext, mbufsToFree, hostAllocDynamicOutput);
67 2 : if (ret != AICPU_SCHEDULE_OK) {
68 0 : return ret;
69 : }
70 :
71 2 : ret = PrepareReqMsg(param, taskContext, mbufsToFree, hostAllocDynamicOutput);
72 2 : if (ret != AICPU_SCHEDULE_OK) {
73 0 : return ret;
74 : }
75 :
76 2 : mbufsToFree.clear();
77 2 : return AICPU_SCHEDULE_OK;
78 2 : }
79 :
80 7 : int32_t PrepareDynamicInputOutputBase::AllocateAndInitOutput(const PrepareDynamicInputOutputKernelArgs * const param,
81 : const RunContext &taskContext,
82 : std::vector<Mbuf *> &mbufsToFree,
83 : const bool hostAllocDynamicOutput) const
84 : {
85 7 : if (param->outputsNum == 0U) {
86 1 : aicpusd_info("Zero outputs");
87 1 : return AICPU_SCHEDULE_OK;
88 : }
89 6 : if (param->inputsNum == 0U) {
90 1 : aicpusd_err("Zero inputs");
91 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
92 : }
93 :
94 5 : const uint64_t * const inputPptrs = PtrToPtr<void, uint64_t>(ValueToPtr(param->inputMbufAddrsAddr));
95 5 : Mbuf *const inputMbuf = *(reinterpret_cast<Mbuf **>(inputPptrs[0U]));
96 5 : void *customBuf = nullptr;
97 5 : uint32_t customBufSize = 0U;
98 5 : const auto ret = halMbufGetPrivInfo(inputMbuf, &customBuf, &customBufSize);
99 5 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (customBuf == nullptr)) {
100 1 : aicpusd_err("Failed to get customBuf, ret[%d].", ret);
101 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
102 : }
103 :
104 4 : const int64_t * const outputTensorSizes = PtrToPtr<void, int64_t>(ValueToPtr(param->outputTensorSizesAddr));
105 4 : uint64_t *const outputPptrs = PtrToPtr<void, uint64_t>(ValueToPtr(param->outputMbufAddrsAddr));
106 10 : for (size_t outputIndex = 0U; outputIndex < static_cast<size_t>(param->outputsNum); ++outputIndex) {
107 7 : if ((outputTensorSizes[outputIndex] < 0) || (outputPptrs[outputIndex] == 0U)) {
108 1 : aicpusd_err("Invalid outputTensorSizes[%zu]:%ld, or invalid pptr.",
109 : outputIndex, outputTensorSizes[outputIndex]);
110 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
111 : }
112 :
113 6 : if (hostAllocDynamicOutput && (outputTensorSizes[outputIndex] == 0)) {
114 1 : aicpusd_info("Skip allocate mbuf for [%zu]th output for its size is 0.", outputIndex);
115 1 : *(reinterpret_cast<Mbuf **>(outputPptrs[outputIndex])) = nullptr;
116 1 : continue;
117 : }
118 5 : const size_t allocSize = static_cast<size_t>(outputTensorSizes[outputIndex]) + sizeof(RuntimeTensorDesc);
119 5 : Mbuf *mbuf = BufManager::GetInstance().MallocAndGuardBufU64(static_cast<uint64_t>(allocSize),
120 5 : taskContext.modelId);
121 5 : if (mbuf == nullptr) {
122 0 : aicpusd_err("model[%u] alloc mbuf fail, size: %zu.", taskContext.modelId, allocSize);
123 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
124 : }
125 5 : mbufsToFree.emplace_back(mbuf);
126 :
127 5 : const auto copyRet = OperatorKernelCommon::CopyMbufHeadInfo(customBuf, customBufSize, mbuf);
128 5 : if (copyRet != AICPU_SCHEDULE_OK) {
129 0 : aicpusd_err("model[%u] copy head fail for [%zu]th output.", taskContext.modelId, outputIndex);
130 0 : return copyRet;
131 : }
132 5 : *(reinterpret_cast<Mbuf **>(outputPptrs[outputIndex])) = mbuf;
133 : }
134 3 : return AICPU_SCHEDULE_OK;
135 : }
136 :
137 5 : int32_t PrepareDynamicInputOutputBase::PrepareReqMsg(const PrepareDynamicInputOutputKernelArgs * const param,
138 : const RunContext &taskContext, std::vector<Mbuf *> &mbufsToFree,
139 : const bool hostAllocDynamicOutput) const
140 : {
141 : // CalculateReqMsgSize
142 5 : size_t reqMsgSize = 0U;
143 5 : const uint32_t * const inputDynamicFlags = PtrToPtr<void, uint32_t>(ValueToPtr(param->inputDynamicFlagsAddr));
144 15 : for (size_t i = 0U; i < static_cast<size_t>(param->inputsNum); ++i) {
145 10 : aicpusd_info("inputDynamicFlags[%zu] is %u.", i, inputDynamicFlags[i]);
146 10 : if (inputDynamicFlags[i] > 0U) {
147 5 : reqMsgSize += sizeof(RuntimeTensorDesc);
148 : } else {
149 5 : reqMsgSize += sizeof(uint64_t);
150 : }
151 : }
152 5 : reqMsgSize += sizeof(uint64_t) * static_cast<size_t>(param->outputsNum);
153 : // AllocReqMbuf
154 5 : Mbuf *reqMbuf = BufManager::GetInstance().MallocAndGuardBufU64(static_cast<uint64_t>(reqMsgSize),
155 5 : taskContext.modelId);
156 5 : if (reqMbuf == nullptr) {
157 0 : aicpusd_err("model[%u] alloc mbuf fail, size: %zu.", taskContext.modelId, reqMsgSize);
158 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
159 : }
160 5 : mbufsToFree.emplace_back(reqMbuf);
161 :
162 5 : void *reqDataPtr = nullptr;
163 5 : const auto reqRet = halMbufGetBuffAddr(reqMbuf, &reqDataPtr);
164 5 : if (reqRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
165 0 : aicpusd_err("model[%u] failed to get reqData ptr, ret[%d].", taskContext.modelId, reqRet);
166 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
167 : }
168 5 : char_t *reqCursor = PtrToPtr<void, char_t>(reqDataPtr);
169 :
170 5 : const uint64_t * const inputPptrs = PtrToPtr<void, uint64_t>(ValueToPtr(param->inputMbufAddrsAddr));
171 5 : if (param->inputsNum > 0U) {
172 5 : Mbuf *const firstInputMbuf = *(reinterpret_cast<Mbuf **>(inputPptrs[0U]));
173 5 : void *customBuf = nullptr;
174 5 : uint32_t customBufSize = 0U;
175 5 : const auto ret = halMbufGetPrivInfo(firstInputMbuf, &customBuf, &customBufSize);
176 5 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (customBuf == nullptr)) {
177 1 : aicpusd_err("Failed to get customBuf, ret[%d].", ret);
178 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
179 : }
180 :
181 4 : const auto copyRet = OperatorKernelCommon::CopyMbufHeadInfo(customBuf, customBufSize, reqMbuf);
182 4 : if (copyRet != AICPU_SCHEDULE_OK) {
183 0 : aicpusd_err("Copy head fail for reqMbuf, ret is %d.", copyRet);
184 0 : return copyRet;
185 : }
186 4 : (void)UpdateReqMsgHead(reqMbuf, inputPptrs, param->inputsNum, taskContext);
187 : }
188 :
189 4 : const uint32_t * const srcFusionOffsets = PtrToPtr<void, uint32_t>(ValueToPtr(param->inputFusionOffsetsAddr));
190 11 : for (size_t i = 0U; i < static_cast<size_t>(param->inputsNum); ++i) {
191 8 : Mbuf * const inputMbuf = *(reinterpret_cast<Mbuf **>(inputPptrs[i]));
192 8 : void *dataPtr = nullptr;
193 8 : const auto dataRet = halMbufGetBuffAddr(inputMbuf, &dataPtr);
194 8 : if (dataRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
195 0 : aicpusd_err("model[%u] failed to get data ptr, ret[%d].", taskContext.modelId, dataRet);
196 1 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
197 : }
198 :
199 8 : if ((srcFusionOffsets != nullptr) && (srcFusionOffsets[i] > 0)) {
200 2 : uint64_t totalOffset = 0UL;
201 2 : const auto ret = OperatorKernelCommon::UpdateDataPtr(PtrToValue(&inputMbuf),
202 2 : static_cast<int32_t>(srcFusionOffsets[i]),
203 : dataPtr, totalOffset);
204 2 : if (ret != AICPU_SCHEDULE_OK) {
205 1 : aicpusd_err("Failed to update the[%zu]th data addr. fusion offset = %d.", i, srcFusionOffsets[i]);
206 1 : return ret;
207 : }
208 1 : aicpusd_info("Successfully updated the[%zu]th data addr. fusion offset = %d.", i, srcFusionOffsets[i]);
209 : }
210 :
211 7 : if (inputDynamicFlags[i] > 0U) {
212 3 : RuntimeTensorDesc *const tensorDesc = PtrToPtr<char_t, RuntimeTensorDesc>(reqCursor);
213 3 : const errno_t eRet = memcpy_s(tensorDesc, sizeof(RuntimeTensorDesc), dataPtr, sizeof(RuntimeTensorDesc));
214 3 : if (eRet != EOK) {
215 0 : aicpusd_err("model[%u] Data copy failed, ret[%d].", taskContext.modelId, eRet);
216 0 : return AICPU_SCHEDULE_ERROR_SAFE_FUNCTION_ERR;
217 : }
218 3 : tensorDesc->dataAddr = PtrToValue(dataPtr) + static_cast<uint64_t>(sizeof(RuntimeTensorDesc));
219 3 : reqCursor += sizeof(RuntimeTensorDesc);
220 : } else {
221 4 : uint64_t *const dataAddr = PtrToPtr<char_t, uint64_t>(reqCursor);
222 4 : *dataAddr = static_cast<uint64_t>(PtrToValue(dataPtr) + sizeof(RuntimeTensorDesc));
223 4 : reqCursor += sizeof(uint64_t);
224 : }
225 : }
226 :
227 3 : const int64_t * const outputTensorSizes = PtrToPtr<void, int64_t>(ValueToPtr(param->outputTensorSizesAddr));
228 3 : uint64_t *const outputPptrs = PtrToPtr<void, uint64_t>(ValueToPtr(param->outputMbufAddrsAddr));
229 8 : for (size_t i = 0U; i < static_cast<size_t>(param->outputsNum); ++i) {
230 5 : uint64_t *const dataAddr = PtrToPtr<char_t, uint64_t>(reqCursor);
231 5 : if (hostAllocDynamicOutput && (outputTensorSizes[i] == 0)) {
232 1 : *dataAddr = 0U;
233 : } else {
234 4 : Mbuf *const outputMbuf = *(reinterpret_cast<Mbuf **>(outputPptrs[i]));
235 4 : void *dataPtr = nullptr;
236 4 : const auto dataRet = halMbufGetBuffAddr(outputMbuf, &dataPtr);
237 4 : if (dataRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
238 0 : aicpusd_err("model[%u] failed to get data ptr, ret[%d].", taskContext.modelId, dataRet);
239 0 : return AICPU_SCHEDULE_ERROR_FROM_DRV;
240 : }
241 4 : *dataAddr = static_cast<uint64_t>(PtrToValue(dataPtr) + sizeof(RuntimeTensorDesc));
242 : }
243 5 : reqCursor += sizeof(uint64_t);
244 : }
245 :
246 3 : Mbuf **const reqMbufPptr = PtrToPtr<void, Mbuf*>(ValueToPtr(param->reqMsgMbufAddr));
247 3 : *reqMbufPptr = reqMbuf;
248 3 : const auto model = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
249 3 : if (model != nullptr) {
250 2 : aicpusd_info("Reset model. modelId=%u", taskContext.modelId);
251 2 : model->SetModelRetCode(0);
252 2 : model->SetNullDataFlag(false);
253 2 : model->ReSetModelEndOfSequence();
254 : }
255 3 : return AICPU_SCHEDULE_OK;
256 : }
257 :
258 5 : int32_t PrepareDynamicInputOutputBase::UpdateReqMsgHead(Mbuf *const reqMbuf, const uint64_t * const inputPptrs,
259 : const uint32_t inputNum, const RunContext &taskContext) const
260 : {
261 5 : void *headBuf = nullptr;
262 5 : uint32_t headSize = 0U;
263 5 : const auto ret = halMbufGetPrivInfo(reqMbuf, &headBuf, &headSize);
264 5 : if ((ret != DRV_ERROR_NONE) || (headBuf == nullptr) || (static_cast<size_t>(headSize) < sizeof(MbufHeadMsg))) {
265 1 : aicpusd_err("Skip %s. modelId=%u, headSize=%u, baseSize=%lu",
266 : __func__, taskContext.modelId, headSize, sizeof(MbufHeadMsg));
267 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
268 : }
269 4 : MbufHeadMsg * const msg = PtrToPtr<uint8_t, MbufHeadMsg>(PtrAdd<uint8_t>(PtrToPtr<void, uint8_t>(headBuf),
270 4 : MBUF_HEAD_MAX_SIZE, static_cast<size_t>(headSize) - sizeof(MbufHeadMsg)));
271 4 : int32_t retCode = 0;
272 4 : bool nullDataFlag = false;
273 4 : bool isEndofSequence = false;
274 4 : uint32_t i = 1U;
275 8 : while ((i < inputNum) && ((retCode == 0) || !nullDataFlag || !isEndofSequence)) {
276 4 : Mbuf *const inputMbuf = *(reinterpret_cast<Mbuf **>(inputPptrs[i++]));
277 4 : ExtractHeadInfo(inputMbuf, retCode, nullDataFlag, isEndofSequence);
278 : }
279 :
280 4 : if (retCode != 0 && (msg->retCode == 0)) {
281 2 : aicpusd_info("update reqMsgHead's ret code for model[%u].", taskContext.modelId);
282 2 : msg->retCode = retCode;
283 : }
284 4 : if (nullDataFlag) {
285 2 : aicpusd_info("update reqMsgHead's nullDataFlag for model[%u].", taskContext.modelId);
286 2 : msg->dataFlag |= MBUF_HEAD_DATA_FLAG_MASK;
287 : }
288 :
289 4 : if (isEndofSequence) {
290 2 : aicpusd_info("update reqMsgHead's endofSequence for model[%u].", taskContext.modelId);
291 2 : uint8_t * const res = PtrAdd<uint8_t>(PtrToPtr<void, uint8_t>(headBuf), MBUF_HEAD_MAX_SIZE,
292 : static_cast<size_t>(MBUF_HEAD_END_OF_SEQUENCE_POS));
293 2 : *res = END_OF_SEQUENCE_FLAG;
294 : }
295 4 : aicpusd_info("reqmsg's retcode is %d, nullflag is %d", msg->retCode, static_cast<int32_t>(msg->dataFlag));
296 4 : return AICPU_SCHEDULE_OK;
297 : }
298 :
299 4 : void PrepareDynamicInputOutputBase::ExtractHeadInfo(Mbuf *const mbuf, int32_t &retCode, bool &nullDataFlag,
300 : bool &isEndofSequence) const
301 : {
302 4 : void *customBuf = nullptr;
303 4 : uint32_t customBufSize = 0U;
304 4 : const auto ret = halMbufGetPrivInfo(mbuf, &customBuf, &customBufSize);
305 4 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (customBuf == nullptr) ||
306 4 : (customBufSize < MBUF_HEAD_MAX_SIZE)) {
307 0 : aicpusd_err("Failed to get customBuf, ret[%d].", ret);
308 0 : return;
309 : }
310 4 : MbufHeadMsg * const inputMsg = PtrToPtr<uint8_t, MbufHeadMsg>(
311 : PtrAdd<uint8_t>(PtrToPtr<void, uint8_t>(customBuf),
312 4 : MBUF_HEAD_MAX_SIZE, static_cast<size_t>(customBufSize) - sizeof(MbufHeadMsg)));
313 4 : if ((retCode == 0) && (inputMsg->retCode != 0)) {
314 2 : retCode = inputMsg->retCode;
315 : }
316 :
317 4 : if (!nullDataFlag &&
318 4 : (inputMsg->dataFlag & MBUF_HEAD_DATA_FLAG_MASK) == static_cast<uint8_t>(DataFlag::DFLOW_NULL_DATA_FLAG)) {
319 2 : nullDataFlag = true;
320 : }
321 4 : if (!isEndofSequence) {
322 4 : const uint8_t * const endOfSequence = PtrAdd<uint8_t>(
323 : PtrToPtr<void, uint8_t>(customBuf), MBUF_HEAD_MAX_SIZE,
324 : static_cast<size_t>(MBUF_HEAD_END_OF_SEQUENCE_POS));
325 4 : if (*endOfSequence == END_OF_SEQUENCE_FLAG) {
326 2 : isEndofSequence = true;
327 : }
328 : }
329 : }
330 :
331 3 : int32_t OperatorKernelPrepareDynamicInputOutput::Compute(const AicpuTaskInfo &kernelTaskInfo,
332 : const RunContext &taskContext)
333 : {
334 3 : return PrepareDynamicInputOutput(kernelTaskInfo, taskContext, false);
335 : }
336 :
337 1 : int32_t OperatorKernelPrepareDynamicInputOutputV2::Compute(const AicpuTaskInfo &kernelTaskInfo,
338 : const RunContext &taskContext)
339 : {
340 1 : return PrepareDynamicInputOutput(kernelTaskInfo, taskContext, true);
341 : }
342 :
343 :
344 6 : REGISTER_OPERATOR_KERNEL(KERNEL_PREPARE_DYNAMIC_INPUT_OUTPUT, OperatorKernelPrepareDynamicInputOutput);
345 6 : REGISTER_OPERATOR_KERNEL(KERNEL_PREPARE_DYNAMIC_INPUT_OUTPUT_V2, OperatorKernelPrepareDynamicInputOutputV2);
346 : } // namespace AicpuSchedule
|
