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