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