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_gather_dequeue.h"
12 :
13 : #include "aicpusd_event_process.h"
14 : #include "aicpusd_status.h"
15 : #include "aicpusd_context.h"
16 : #include "aicpusd_model_execute.h"
17 : #include "aicpusd_msg_send.h"
18 : #include "aicpusd_resource_manager.h"
19 : #include "operator_kernel_common.h"
20 :
21 :
22 : namespace AicpuSchedule {
23 : namespace {
24 : const std::string KERNEL_GATHER_DEQUEUE = "gatherDequeue";
25 : } // namespace
26 :
27 10 : int32_t OperatorKernelGatherDequeue::Compute(const AicpuTaskInfo &kernelTaskInfo, const RunContext &taskContext)
28 : {
29 10 : aicpusd_info("Start ModelGatherDeque. modelId=%u, streamId=%u, taskId=%u",
30 : taskContext.modelId, kernelTaskInfo.streamID, kernelTaskInfo.taskID);
31 10 : if (kernelTaskInfo.paraBase == 0UL) {
32 1 : aicpusd_err("kernelTaskInfo.paraBase is null");
33 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
34 : }
35 :
36 9 : const GatherDequeParam * const batchDeqInfo = PtrToPtr<void, GatherDequeParam>(ValueToPtr(kernelTaskInfo.paraBase));
37 9 : if ((batchDeqInfo->inputNums == 0U) || (batchDeqInfo->queueIdsAddr == 0U) || (batchDeqInfo->mbufAddrsAddr == 0U)) {
38 1 : aicpusd_err("inputNums or queueIdsAddr or mbufAddrsAddr is invalid");
39 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
40 : }
41 :
42 8 : const auto model = AicpuModelManager::GetInstance().GetModel(taskContext.modelId);
43 8 : if (model == nullptr) {
44 1 : aicpusd_err("Cannot get model by modelId:[%u], streamId[%u], taskId[%u].",
45 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
46 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
47 : }
48 :
49 7 : int32_t gatherRet = AICPU_SCHEDULE_OK;
50 7 : if (SelectMbuf(batchDeqInfo, taskContext, model, gatherRet)) {
51 3 : return gatherRet;
52 : }
53 :
54 4 : EventWaitManager::AnyQueNotEmptyWaitManager().ResetEventState(static_cast<size_t>(taskContext.modelId));
55 : // if not gathered, then pending; if pending fail, try gather again
56 4 : bool blockOnClientQ = false;
57 4 : const auto tempContext = taskContext;
58 : // DequeAndCheckIfReady with tempContext to avoid taskContext being modified
59 4 : while (!DequeAndCheckIfReady(batchDeqInfo, gatherRet, model, tempContext, blockOnClientQ)) {
60 3 : bool needWait = false;
61 3 : aicpusd_info("Batch queue is not gathered, modelId[%u], streamId[%u], taskId[%u].",
62 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
63 3 : EventWaitManager::AnyQueNotEmptyWaitManager().WaitEvent(static_cast<size_t>(taskContext.modelId),
64 3 : taskContext.streamId, needWait);
65 3 : if (needWait) {
66 3 : bool * const pending = const_cast<bool *>(&taskContext.pending);
67 3 : *pending = true;
68 3 : aicpusd_info("ModelGatherDeque pending, modelId[%u], streamId[%u], taskId[%u].",
69 : taskContext.modelId, taskContext.streamId, kernelTaskInfo.taskID);
70 3 : if (blockOnClientQ && !model->GetModelDestroyStatus()) {
71 1 : aicpusd_info("Submit supply enque event for model[%u] blocked on client queue.", taskContext.modelId);
72 1 : AICPUSubEventInfo subEventInfo = {};
73 1 : subEventInfo.modelId = taskContext.modelId;
74 1 : (void) AicpuMsgSend::SendAICPUSubEvent(
75 : PtrToPtr<AICPUSubEventInfo, const char_t>(&subEventInfo),
76 : static_cast<uint32_t>(sizeof(AICPUSubEventInfo)), AICPU_SUB_EVENT_SUPPLY_ENQUEUE,
77 : CP_DEFAULT_GROUP_ID, true);
78 : }
79 3 : break;
80 : }
81 : }
82 4 : return (gatherRet == AICPU_SCHEDULE_ERROR_MODEL_UNLOAD) ? AICPU_SCHEDULE_OK : gatherRet;
83 : }
84 :
85 16 : bool OperatorKernelGatherDequeue::SelectMbuf(const GatherDequeParam * const batchDeqInfo, const RunContext &taskContext,
86 : void *const modelPtr, int32_t &gatherRet) const
87 : {
88 16 : AicpuModel* const model = PtrToPtr<void, AicpuModel>(modelPtr);
89 16 : Mbuf ***mbufPptr = PtrToPtr<void, Mbuf**>(ValueToPtr(batchDeqInfo->mbufAddrsAddr));
90 : const auto selectRes =
91 16 : model->SelectGatheredMbuf(mbufPptr, batchDeqInfo->inputsAlignTimeout, batchDeqInfo->inputsAlignMaxCacheNum);
92 16 : if (selectRes == GatherResult::UN_SELECTED) {
93 11 : return false;
94 : }
95 :
96 5 : if (selectRes == GatherResult::SELECTED) {
97 8 : for (size_t i = 0U; i < static_cast<size_t>(batchDeqInfo->inputNums); ++i) {
98 6 : (void) BufManager::GetInstance().GuardBuf(*mbufPptr[i], taskContext.modelId);
99 : }
100 2 : gatherRet = AICPU_SCHEDULE_OK;
101 2 : return true;
102 : }
103 :
104 : // fake selected
105 : // allow drop
106 3 : if (batchDeqInfo->inputsAlignDropout != 0U) {
107 4 : for (size_t i = 0U; i < static_cast<size_t>(batchDeqInfo->inputNums); ++i) {
108 3 : if (*mbufPptr[i] != nullptr) {
109 2 : aicpusd_info("free the [%zu]th mbuf", i);
110 2 : (void) halMbufFree(*mbufPptr[i]);
111 : }
112 : }
113 1 : return false;
114 : }
115 : // not allow drop
116 2 : Mbuf *const stubMbuf = MakeUpPassedMbuf(taskContext.modelId);
117 8 : for (size_t i = 0U; i < static_cast<size_t>(batchDeqInfo->inputNums); ++i) {
118 6 : if (*mbufPptr[i] != nullptr) {
119 2 : (void) BufManager::GetInstance().GuardBuf(*mbufPptr[i], taskContext.modelId);
120 : } else {
121 : // when FAKE_SELECTED, there's some null mbuf, then we should alloc
122 4 : aicpusd_info("stub mbuf for the [%zu]th input", i);
123 4 : *mbufPptr[i] = stubMbuf;
124 : }
125 : }
126 2 : gatherRet = AICPU_SCHEDULE_ERROR_DISCARD_DATA;
127 2 : return true;
128 : }
129 :
130 6 : Mbuf *OperatorKernelGatherDequeue::MakeUpPassedMbuf(const uint32_t modelId) const
131 : {
132 6 : Mbuf *stubMbuf = BufManager::GetInstance().MallocAndGuardBufU64(
133 : static_cast<uint64_t>(sizeof(RuntimeTensorDesc)), modelId);
134 6 : if (stubMbuf == nullptr) {
135 1 : aicpusd_err("Failed to alloc stubMbuf");
136 1 : return nullptr;
137 : }
138 :
139 5 : MbufHeadMsg * const headMsg = GetMbufHeadMsg(stubMbuf);
140 5 : if (headMsg == nullptr) {
141 1 : aicpusd_err("null head");
142 1 : return nullptr;
143 : }
144 :
145 4 : headMsg->retCode = INNER_ERROR_BASE + AICPU_SCHEDULE_ERROR_DISCARD_DATA;
146 :
147 4 : uint64_t mbufLen = 0UL;
148 4 : void *dataPtr = nullptr;
149 4 : const auto bufInfoRet = OperatorKernelCommon::GetMbufAddrAndSize(stubMbuf, &dataPtr, &mbufLen, modelId, true);
150 4 : if (bufInfoRet != AICPU_SCHEDULE_OK) {
151 1 : return nullptr;
152 : }
153 3 : RuntimeTensorDesc *const tensorDesc = PtrToPtr<void, RuntimeTensorDesc>(dataPtr);
154 3 : tensorDesc->shape[0U] = 1;
155 3 : tensorDesc->shape[1U] = 0;
156 3 : tensorDesc->originalShape[0U] = 1;
157 3 : tensorDesc->originalShape[1U] = 0;
158 3 : tensorDesc->dataSize = 0U;
159 :
160 3 : return stubMbuf;
161 : }
162 :
163 14 : MbufHeadMsg *OperatorKernelGatherDequeue::GetMbufHeadMsg(Mbuf *const mbuf) const
164 : {
165 14 : void *headBuf = nullptr;
166 14 : uint32_t headSize = 0U;
167 14 : const auto drvRet = halMbufGetPrivInfo(mbuf, &headBuf, &headSize);
168 14 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) ||
169 14 : (headBuf == nullptr) || (static_cast<size_t>(headSize) < sizeof(MbufHeadMsg))) {
170 1 : aicpusd_err("Failed to get head info in input information, ret[%d].", drvRet);
171 1 : return nullptr;
172 : }
173 13 : return PtrToPtr<char_t, MbufHeadMsg>(PtrToPtr<void, char_t>(headBuf) + headSize - sizeof(MbufHeadMsg));
174 : }
175 :
176 5 : bool OperatorKernelGatherDequeue::DequeAndCheckIfReady(const GatherDequeParam * const batchDeqInfo, int32_t &gatherRet,
177 : void *const modelPtr, const RunContext &taskContext,
178 : bool &blockOnClientQ) const
179 : {
180 5 : AicpuModel* const model = PtrToPtr<void, AicpuModel>(modelPtr);
181 5 : uint32_t * const queueIds = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqInfo->queueIdsAddr));
182 5 : uint32_t *const deviceTypes = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqInfo->deviceTypeAddr));
183 5 : uint32_t *const deviceIds = PtrToPtr<void, uint32_t>(ValueToPtr(batchDeqInfo->deviceIdAddr));
184 5 : DeployContext deployCtx = DeployContext::DEVICE;
185 5 : (void) GetAicpuDeployContext(deployCtx);
186 5 : uint64_t loopCnt = 0UL;
187 : while (true) {
188 13 : loopCnt++;
189 13 : const size_t queueIndex = model->GetCurDequeIndex(static_cast<size_t>(batchDeqInfo->inputNums));
190 13 : aicpusd_info("start [%llu]th dequeue queue[%zu]", loopCnt, queueIndex);
191 13 : Mbuf *mbuf = nullptr;
192 : // host cpusd with device queue, so the queue is clientQ
193 13 : if ((deployCtx != DeployContext::DEVICE) && (deviceTypes[queueIndex] == 0U)) {
194 2 : blockOnClientQ = true;
195 : BufEnQueueBuffInfo queueInfo = {
196 2 : queueIds[queueIndex], static_cast<int32_t>(deviceIds[queueIndex]), PtrToValue(&mbuf)};
197 2 : gatherRet = ModelAttachAndDequeueBuff(queueInfo, taskContext, true);
198 2 : } else {
199 11 : blockOnClientQ = false;
200 11 : BufEnQueueInfo queueInfo = { queueIds[queueIndex], PtrToValue(&mbuf) };
201 11 : gatherRet = DequeueTask(queueInfo, taskContext, false);
202 : }
203 13 : if ((gatherRet != static_cast<int32_t>(AICPU_SCHEDULE_OK)) || (mbuf == nullptr)) {
204 4 : aicpusd_warn("Failed to deque from queue[%u], gatherRet[%d].", queueIds[queueIndex], gatherRet);
205 5 : return false;
206 : }
207 :
208 9 : if (!StoreMbufIntoModel(mbuf, queueIndex, batchDeqInfo->inputNums, modelPtr)) {
209 0 : continue;
210 : }
211 :
212 9 : if (SelectMbuf(batchDeqInfo, taskContext, modelPtr, gatherRet)) {
213 1 : return true;
214 : }
215 8 : }
216 : return false;
217 : }
218 :
219 9 : bool OperatorKernelGatherDequeue::StoreMbufIntoModel(Mbuf *const mbuf, const size_t index, const uint32_t capacity,
220 : void *const modelPtr) const
221 : {
222 9 : AicpuModel* const model = PtrToPtr<void, AicpuModel>(modelPtr);
223 9 : const MbufHeadMsg * const headMsg = GetMbufHeadMsg(mbuf);
224 9 : if (headMsg == nullptr) {
225 0 : return false;
226 : }
227 9 : const auto result = model->StoreDequedMbuf(headMsg->transId, headMsg->dataLabel, index, mbuf, capacity);
228 9 : if (result == StoreResult::FAIL_STORE) {
229 0 : return false;
230 : }
231 :
232 9 : const auto guardRet = model->UnGardModelBuf(mbuf);
233 9 : if (guardRet != AICPU_SCHEDULE_OK) {
234 0 : aicpusd_warn("BufManager unguard mbuf failed, drvRet[%d].", guardRet);
235 : }
236 9 : return (result == StoreResult::SUCCESS_STORE);
237 : }
238 :
239 :
240 6 : REGISTER_OPERATOR_KERNEL(KERNEL_GATHER_DEQUEUE, OperatorKernelGatherDequeue);
241 : } // namespace AicpuSchedule
|
