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 "aicpusd_queue_event_process.h"
12 :
13 : #include "aicpusd_hal_interface_ref.h"
14 : #include "aicpusd_monitor.h"
15 : #include "aicpusd_feature_ctrl.h"
16 : #include "aicpusd_message_queue.h"
17 : #include "aicpusd_status.h"
18 : #include "securec.h"
19 : #include "type_def.h"
20 :
21 : namespace {
22 : const GroupShareAttr ALL_ATTR_FOR_GROUP = {1U, 1U, 1U, 1U, 0U}; // admin + read + write + alloc
23 : const uint16_t MAJOR_VERSION = 1U;
24 : } // namespace
25 :
26 : namespace AicpuSchedule {
27 302 : AicpuQueueEventProcess& AicpuQueueEventProcess::GetInstance()
28 : {
29 302 : static AicpuQueueEventProcess instance;
30 302 : return instance;
31 : }
32 :
33 16 : int32_t AicpuQueueEventProcess::DoProcessDrvMsg(const event_info& event, bool& needRes)
34 : {
35 16 : int32_t ret = AICPU_SCHEDULE_OK;
36 16 : switch (event.comm.subevent_id) {
37 5 : case DRV_SUBEVENT_GRANT_MSG: {
38 5 : ret = GrantQueue(event);
39 5 : needRes = true;
40 5 : break;
41 : }
42 6 : case DRV_SUBEVENT_ATTACH_MSG: {
43 6 : ret = AttachQueue(event);
44 6 : needRes = true;
45 6 : break;
46 : }
47 5 : default: {
48 5 : if (event.comm.subevent_id == DRV_SUBEVENT_QUEUE_INIT_MSG) {
49 1 : std::string groupName;
50 1 : ret = GetOrCreateGroup(groupName);
51 1 : if (ret != AICPU_SCHEDULE_OK) {
52 1 : needRes = true;
53 1 : return ret;
54 : }
55 1 : }
56 4 : event_info* const eventToDrv = const_cast<event_info*>(&event);
57 4 : const int32_t drvRet = halEventProc(AicpuDrvManager::GetInstance().GetDeviceId(), eventToDrv);
58 4 : if (drvRet != DRV_ERROR_NONE) {
59 1 : aicpusd_err("Call halEventProc failed, subevent_id=[%u],ret=[%d]", event.comm.subevent_id, drvRet);
60 1 : return drvRet;
61 : }
62 3 : break;
63 : }
64 : }
65 14 : return ret;
66 : }
67 :
68 7 : int32_t AicpuQueueEventProcess::ProcessDrvMsg(const event_info& event)
69 : {
70 7 : aicpusd_debug(
71 : "Begin to process drv event. eventId=%u, subEventId=%u, hostPid=%d, grpId=%u", event.comm.event_id,
72 : event.comm.subevent_id, event.comm.host_pid, event.comm.grp_id);
73 7 : bool needRes = false;
74 7 : int32_t ret = DoProcessDrvMsg(event, needRes);
75 7 : if (needRes) {
76 3 : event_proc_result rsp = {};
77 3 : rsp.ret = ret;
78 : const int32_t resRet =
79 3 : ResponseEvent(event, PtrToPtr<event_proc_result, const char_t>(&rsp), sizeof(event_proc_result));
80 3 : if (ret == AICPU_SCHEDULE_OK) {
81 2 : ret = resRet;
82 : }
83 : }
84 7 : aicpusd_debug(
85 : "End to process drv event. eventId=%u, subEventId=%u, hostPid=%d, grpId=%u", event.comm.event_id,
86 : event.comm.subevent_id, event.comm.host_pid, event.comm.grp_id);
87 7 : return ret;
88 : }
89 :
90 27 : int32_t AicpuQueueEventProcess::DoProcessQsMsg(
91 : const event_info& event, std::shared_ptr<CallbackMsg>& callback, const bqs::QsProcMsgRspDstAicpu*& qsProcMsgRsp,
92 : bool& isRes)
93 : {
94 27 : int32_t ret = AICPU_SCHEDULE_OK;
95 27 : switch (event.comm.subevent_id) {
96 1 : case bqs::ACL_BIND_QUEUE_INIT:
97 1 : ret = ProcessBindQueueInit(event);
98 1 : break;
99 4 : case bqs::AICPU_BIND_QUEUE_INIT_RES:
100 4 : ret = ProcessBindQueueInitRet(event, callback, &qsProcMsgRsp);
101 4 : isRes = true;
102 4 : break;
103 1 : case bqs::ACL_BIND_QUEUE:
104 1 : ret = ProcessQueueEventWithMbuf(event, bqs::AICPU_BIND_QUEUE);
105 1 : break;
106 1 : case bqs::AICPU_BIND_QUEUE_RES:
107 1 : ret = ProcessQsRetWithMbuf(event, callback, &qsProcMsgRsp);
108 1 : isRes = true;
109 1 : break;
110 1 : case bqs::ACL_UNBIND_QUEUE:
111 1 : ret = ProcessQueueEventWithMbuf(event, bqs::AICPU_UNBIND_QUEUE);
112 1 : break;
113 1 : case bqs::AICPU_UNBIND_QUEUE_RES:
114 1 : ret = ProcessQsRetWithMbuf(event, callback, &qsProcMsgRsp);
115 1 : isRes = true;
116 1 : break;
117 5 : case bqs::ACL_QUERY_QUEUE_NUM:
118 5 : ret = ProcessQueryQueueNum(event);
119 5 : break;
120 3 : case bqs::AICPU_QUERY_QUEUE_NUM_RES:
121 3 : ret = ProcessQsRet(event, callback, &qsProcMsgRsp);
122 3 : isRes = true;
123 3 : break;
124 8 : case bqs::ACL_QUERY_QUEUE:
125 8 : ret = ProcessQueueEventWithMbuf(event, bqs::AICPU_QUERY_QUEUE);
126 8 : break;
127 1 : case bqs::AICPU_QUERY_QUEUE_RES:
128 1 : ret = ProcessQsRetWithMbuf(event, callback, &qsProcMsgRsp);
129 1 : isRes = true;
130 1 : break;
131 1 : default:
132 1 : aicpusd_err("The queue event sub event id is not found, subevent id[%u]", event.comm.subevent_id);
133 1 : ret = AICPU_SCHEDULE_ERROR_NOT_FOUND_QUEUE_SUB_EVENT_ID;
134 1 : break;
135 : }
136 27 : return ret;
137 : }
138 :
139 27 : int32_t AicpuQueueEventProcess::ProcessQsMsg(const event_info& event)
140 : {
141 27 : aicpusd_info("Begin to ProcessQsMsg subevent_id[%u]", event.comm.subevent_id);
142 27 : bool isRes = false;
143 27 : const bqs::QsProcMsgRspDstAicpu* qsProcMsgRsp = nullptr;
144 27 : std::shared_ptr<CallbackMsg> callback = nullptr;
145 27 : const int32_t ret = DoProcessQsMsg(event, callback, qsProcMsgRsp, isRes);
146 27 : if (callback != nullptr) {
147 7 : return ResponseEvent(
148 7 : callback->event, PtrToPtr<const bqs::QsProcMsgRspDstAicpu, const char_t>(qsProcMsgRsp),
149 7 : sizeof(bqs::QsProcMsgRspDstAicpu));
150 : }
151 20 : if ((ret != AICPU_SCHEDULE_OK) && (!isRes)) {
152 12 : bqs::QsProcMsgRspDstAicpu msg = {};
153 12 : msg.retCode = ret;
154 12 : (void)ResponseEvent(
155 : event, PtrToPtr<bqs::QsProcMsgRspDstAicpu, const char_t>(&msg), sizeof(bqs::QsProcMsgRspDstAicpu));
156 12 : return ret;
157 : }
158 8 : return ret;
159 27 : }
160 :
161 18 : int32_t AicpuQueueEventProcess::AddCallback(uint64_t userData, std::shared_ptr<CallbackMsg>& callback)
162 : {
163 18 : bool ret = false;
164 : {
165 18 : const std::lock_guard<std::mutex> guard(lockCallback_);
166 18 : const auto msg = callbacks_.emplace(userData, callback);
167 18 : ret = msg.second;
168 18 : }
169 18 : if (!ret) {
170 1 : aicpusd_err(
171 : "Save event callback failed, subevent_id[%u], userData[%llu]", callback->event.comm.subevent_id, userData);
172 1 : return AICPU_SCHEDULE_ERROR_ADD_CALLBACK_FAILED;
173 : }
174 17 : aicpusd_info(
175 : "Successfully added callback, subevent_id[%d], userData[%llu].", callback->event.comm.subevent_id, userData);
176 17 : return AICPU_SCHEDULE_OK;
177 : }
178 :
179 11 : int32_t AicpuQueueEventProcess::GetAndDeleteCallback(const uint64_t userData, std::shared_ptr<CallbackMsg>& callback)
180 : {
181 11 : const std::lock_guard<std::mutex> guard(lockCallback_);
182 11 : const auto iter = callbacks_.find(userData);
183 11 : if (iter == callbacks_.end()) {
184 2 : aicpusd_err("Get event callback failed, userData[%llu]", userData);
185 2 : return AICPU_SCHEDULE_ERROR_GET_CALLBACK_FAILED;
186 : }
187 9 : callback = iter->second;
188 9 : (void)callbacks_.erase(iter);
189 9 : aicpusd_info(
190 : "Successfully got and deleted callback, subevent_id[%u], userData[%llu].", callback->event.comm.subevent_id,
191 : userData);
192 9 : return AICPU_SCHEDULE_OK;
193 11 : }
194 :
195 17 : int32_t AicpuQueueEventProcess::CreateAndAddCallbackMsg(
196 : const event_info& event, Mbuf* const buff, const uint64_t userData, std::shared_ptr<CallbackMsg>& callback)
197 : {
198 : try {
199 17 : callback = std::make_shared<CallbackMsg>();
200 0 : } catch (std::bad_alloc&) {
201 0 : aicpusd_err("Create callback msg failed, subevent_id[%u].", event.comm.subevent_id);
202 0 : if (buff != nullptr) {
203 0 : (void)halMbufFree(buff);
204 : }
205 0 : return AICPU_SCHEDULE_ERROR_CREATE_CALLBACK_FAILED;
206 0 : }
207 17 : callback->event = event;
208 17 : callback->buff = buff;
209 17 : return AddCallback(userData, callback);
210 : }
211 :
212 9 : int32_t AicpuQueueEventProcess::ProcessBindQueueInit(const event_info& event)
213 : {
214 9 : aicpusd_info("Begin to ProcessBindQueueInit.");
215 9 : if (initPipeline_ != BindQueueInitStatus::UNINIT) {
216 1 : aicpusd_err("Already call bind queue init, don't call repeatedly.");
217 1 : return AICPU_SCHEDULE_ERROR_REPEATED_BIND_QUEUE_INIT;
218 : }
219 16 : if ((lockInit_.test_and_set()) || (initPipeline_ != BindQueueInitStatus::UNINIT)) {
220 1 : aicpusd_err("Already call bind queue init, don't call repeatedly.");
221 1 : return AICPU_SCHEDULE_ERROR_REPEATED_BIND_QUEUE_INIT;
222 : }
223 14 : const ScopeGuard lockGuard([this]() { lockInit_.clear(); });
224 7 : initPipeline_ = BindQueueInitStatus::INITING;
225 :
226 : // check event params
227 7 : const char_t* qsBindInitMsg = nullptr;
228 7 : int32_t ret = ParseQueueEventMessage(event, qsBindInitMsg, sizeof(bqs::QsBindInit));
229 7 : if (ret != AICPU_SCHEDULE_OK) {
230 1 : return ret;
231 : }
232 :
233 : // query qs pid
234 6 : ret = QueryQsPid();
235 6 : if (ret != AICPU_SCHEDULE_OK) {
236 1 : return ret;
237 : }
238 :
239 : // add qs to cp shape group
240 5 : std::string groupName;
241 5 : ret = GetOrCreateGroup(groupName);
242 5 : if (ret != AICPU_SCHEDULE_OK) {
243 1 : return ret;
244 : }
245 4 : ret = ShareGroupWithProcess(groupName, qsPid_);
246 4 : if (ret != AICPU_SCHEDULE_OK) {
247 1 : return ret;
248 : }
249 :
250 : // add callback
251 3 : std::shared_ptr<CallbackMsg> callback = nullptr;
252 3 : const uint64_t userData = PtrToValue(PtrToPtr<const char_t, const void>(qsBindInitMsg));
253 3 : ret = CreateAndAddCallbackMsg(event, nullptr, userData, callback);
254 3 : if (ret != AICPU_SCHEDULE_OK) {
255 1 : return ret;
256 : }
257 :
258 : // send init pipeline event to qs
259 2 : bqs::QsBindInit msg = {};
260 2 : msg.syncEventHead = userData;
261 2 : msg.pid = curPid_;
262 2 : msg.grpId = AicpuDrvManager::GetInstance().GetGroupId();
263 2 : msg.majorVersion = MAJOR_VERSION;
264 2 : ret = SendEventToQs(PtrToPtr<bqs::QsBindInit, char_t>(&msg), sizeof(bqs::QsBindInit), bqs::AICPU_BIND_QUEUE_INIT);
265 2 : if (ret != AICPU_SCHEDULE_OK) {
266 1 : (void)GetAndDeleteCallback(userData, callback);
267 1 : return ret;
268 : }
269 1 : aicpusd_info("Successfully processed bind queue init.");
270 1 : return AICPU_SCHEDULE_OK;
271 7 : }
272 :
273 4 : int32_t AicpuQueueEventProcess::ProcessBindQueueInitRet(
274 : const event_info& event, std::shared_ptr<CallbackMsg>& callback,
275 : const bqs::QsProcMsgRspDstAicpu** const qsProcMsgRsp)
276 : {
277 4 : aicpusd_info("Begin to ProcessBindQueueInitRet.");
278 4 : const int32_t ret = ProcessQsRet(event, callback, qsProcMsgRsp);
279 4 : if (ret != AICPU_SCHEDULE_OK) {
280 1 : return ret;
281 : }
282 3 : pipelineQueueId_ = (*qsProcMsgRsp)->retValue;
283 3 : auto drvRet = halQueueInit(AicpuDrvManager::GetInstance().GetDeviceId());
284 3 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
285 1 : aicpusd_err("halQueueInit error, deviceId[%u], ret[%d]", AicpuDrvManager::GetInstance().GetDeviceId(), drvRet);
286 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
287 : }
288 2 : drvRet = halQueueAttach(AicpuDrvManager::GetInstance().GetDeviceId(), pipelineQueueId_, 0);
289 2 : if (drvRet != DRV_ERROR_NONE) {
290 1 : aicpusd_err("Cp attach Qs queue failed, queueId[%u] ret[%d]", pipelineQueueId_, static_cast<int32_t>(drvRet));
291 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
292 : }
293 1 : initPipeline_ = BindQueueInitStatus::INITED;
294 1 : aicpusd_info("Successfully processed bind queue init ret, queueId[%u].", pipelineQueueId_);
295 1 : return AICPU_SCHEDULE_OK;
296 : }
297 :
298 5 : int32_t AicpuQueueEventProcess::ProcessQueryQueueNum(const event_info& event)
299 : {
300 5 : aicpusd_info("Begin to ProcessQueryQueueNum.");
301 5 : if (initPipeline_ != BindQueueInitStatus::INITED) {
302 1 : aicpusd_err("Need call bind queue init before bind queue.");
303 1 : return AICPU_SCHEDULE_ERROR_CP_QS_PIPELINE_NOT_INIT;
304 : }
305 4 : const char_t* msg = nullptr;
306 4 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(bqs::QueueRouteQuery));
307 4 : if (ret != AICPU_SCHEDULE_OK) {
308 1 : return ret;
309 : }
310 3 : const bqs::QueueRouteQuery* const queryMsg = PtrToPtr<const char_t, const bqs::QueueRouteQuery>(msg);
311 : // add callback
312 3 : std::shared_ptr<CallbackMsg> callback = nullptr;
313 3 : const uint64_t userData = PtrToValue(queryMsg);
314 3 : ret = CreateAndAddCallbackMsg(event, nullptr, userData, callback);
315 3 : if (ret != AICPU_SCHEDULE_OK) {
316 1 : return ret;
317 : }
318 :
319 : // send event to qs
320 2 : bqs::QueueRouteQuery sendMsg = *queryMsg;
321 2 : sendMsg.syncEventHead = userData;
322 2 : ret = SendEventToQs(
323 : PtrToPtr<bqs::QueueRouteQuery, char_t>(&sendMsg), sizeof(bqs::QueueRouteQuery), bqs::AICPU_QUERY_QUEUE_NUM);
324 2 : if (ret != AICPU_SCHEDULE_OK) {
325 1 : (void)GetAndDeleteCallback(userData, callback);
326 1 : return ret;
327 : }
328 :
329 1 : aicpusd_info("Successfully processed query queue num.");
330 1 : return AICPU_SCHEDULE_OK;
331 3 : }
332 :
333 9 : int32_t AicpuQueueEventProcess::ProcessQsRet(
334 : const event_info& event, std::shared_ptr<CallbackMsg>& callback,
335 : const bqs::QsProcMsgRspDstAicpu** const qsProcMsgRsp)
336 : {
337 9 : const char_t* msg = nullptr;
338 9 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(bqs::QsProcMsgRspDstAicpu));
339 9 : if (ret != AICPU_SCHEDULE_OK) {
340 1 : return ret;
341 : }
342 8 : *qsProcMsgRsp = PtrToPtr<const char_t, const bqs::QsProcMsgRspDstAicpu>(msg);
343 8 : ret = GetAndDeleteCallback((*qsProcMsgRsp)->syncEventHead, callback);
344 8 : if (ret != AICPU_SCHEDULE_OK) {
345 1 : return ret;
346 : }
347 7 : return AICPU_SCHEDULE_OK;
348 : }
349 :
350 12 : int32_t AicpuQueueEventProcess::CheckAndInitParamWithMbuf(
351 : const event_info& event, const bqs::QueueRouteList*& msg, bqs::QsRouteHead*& routeHead) const
352 : {
353 12 : if (initPipeline_ != BindQueueInitStatus::INITED) {
354 1 : aicpusd_err("Need call bind queue init first.");
355 1 : return AICPU_SCHEDULE_ERROR_CP_QS_PIPELINE_NOT_INIT;
356 : }
357 11 : const char_t* routeListMsg = nullptr;
358 11 : const int32_t ret = ParseQueueEventMessage(event, routeListMsg, sizeof(bqs::QueueRouteList));
359 11 : if (ret != AICPU_SCHEDULE_OK) {
360 1 : return ret;
361 : }
362 10 : msg = PtrToPtr<const char_t, const bqs::QueueRouteList>(routeListMsg);
363 10 : routeHead = PtrToPtr<void, bqs::QsRouteHead>(ValueToPtr(msg->routeListMsgAddr));
364 10 : if (routeHead == nullptr) {
365 1 : aicpusd_err("The event_info msg QsRouteHead is nullptr.");
366 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
367 : }
368 :
369 9 : if (routeHead->routeNum == 0U) {
370 1 : aicpusd_err("The route num is 0.");
371 1 : return AICPU_SCHEDULE_ERROR_ROUTE_NUM_IS_ZERO;
372 : }
373 8 : const uint32_t routeSize =
374 8 : static_cast<uint32_t>(sizeof(bqs::QsRouteHead) + (routeHead->routeNum * sizeof(bqs::QueueRoute)));
375 8 : if (routeHead->length < routeSize) {
376 1 : aicpusd_err("The event_info msg QsRouteHead.length[%u] < routeSize[%u].", routeHead->length, routeSize);
377 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
378 : }
379 7 : return AICPU_SCHEDULE_OK;
380 : }
381 :
382 11 : int32_t AicpuQueueEventProcess::ProcessQueueEventWithMbuf(
383 : const event_info& event, const bqs::QueueSubEventType drvSubeventId)
384 : {
385 11 : const bqs::QueueRouteList* msg = nullptr;
386 11 : bqs::QsRouteHead* routeHead = nullptr;
387 11 : int32_t ret = CheckAndInitParamWithMbuf(event, msg, routeHead);
388 11 : if (ret != AICPU_SCHEDULE_OK) {
389 4 : return ret;
390 : }
391 :
392 7 : if (drvSubeventId == bqs::AICPU_BIND_QUEUE) {
393 : const bqs::QueueRoute* const queueRoute =
394 2 : PtrToPtr<void, bqs::QueueRoute>(ValueToPtr(msg->routeListMsgAddr + sizeof(bqs::QsRouteHead)));
395 2 : ret = AddQueueAuthToQs(queueRoute, routeHead->routeNum);
396 2 : if (ret != AICPU_SCHEDULE_OK) {
397 1 : return ret;
398 : }
399 : }
400 :
401 6 : routeHead->subEventId = drvSubeventId;
402 6 : const uint64_t userData = PtrToValue(PtrToPtr<const bqs::QueueRouteList, const void>(msg));
403 6 : routeHead->userData = userData;
404 :
405 : // alloc mbuf
406 6 : Mbuf* buff = nullptr;
407 6 : ret = AllocMbufAndEnqueue(routeHead, static_cast<size_t>(routeHead->length), &buff);
408 6 : if (ret != AICPU_SCHEDULE_OK) {
409 1 : if (buff != nullptr) {
410 1 : (void)halMbufFree(buff);
411 : }
412 1 : return ret;
413 : }
414 :
415 : // add callback
416 5 : std::shared_ptr<CallbackMsg> callback = nullptr;
417 5 : ret = CreateAndAddCallbackMsg(event, buff, userData, callback);
418 5 : if (ret != AICPU_SCHEDULE_OK) {
419 1 : AicpuMonitor::GetInstance().SendKillMsgToTsd();
420 1 : return ret;
421 : }
422 :
423 4 : bqs::QueueRouteList msgToQs = *msg;
424 4 : msgToQs.syncEventHead = userData;
425 4 : ret = SendEventToQs(
426 : PtrToPtr<bqs::QueueRouteList, char_t>(&msgToQs), sizeof(bqs::QueueRouteList),
427 : bqs::AICPU_QUEUE_RELATION_PROCESS);
428 4 : if (ret != AICPU_SCHEDULE_OK) {
429 1 : AicpuMonitor::GetInstance().SendKillMsgToTsd();
430 1 : return ret;
431 : }
432 3 : return AICPU_SCHEDULE_OK;
433 5 : }
434 :
435 9 : int32_t AicpuQueueEventProcess::SendEventToQs(
436 : char_t* const msg, const size_t msgLen, const bqs::QueueSubEventType drvSubeventId) const
437 : {
438 9 : event_summary sched = {};
439 9 : sched.pid = qsPid_;
440 9 : sched.grp_id = 7U; // default: 7 is qs event group
441 9 : sched.event_id = EVENT_QS_MSG;
442 9 : sched.subevent_id = drvSubeventId;
443 9 : sched.msg_len = static_cast<uint32_t>(msgLen);
444 9 : sched.msg = msg;
445 9 : sched.dst_engine = CCPU_DEVICE;
446 :
447 9 : const int32_t drvRet = halEschedSubmitEvent(AicpuDrvManager::GetInstance().GetDeviceId(), &sched);
448 9 : if (drvRet != DRV_ERROR_NONE) {
449 4 : aicpusd_err("Failed to submit event to qs, ret=[%d].", drvRet);
450 4 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
451 : }
452 5 : return AICPU_SCHEDULE_OK;
453 : }
454 :
455 75 : int32_t AicpuQueueEventProcess::ParseQueueEventMessage(
456 : const event_info& event, const char_t*& msg, const size_t msgSize, const bool isSyncEvent) const
457 : {
458 75 : size_t offset = 0U;
459 75 : if (isSyncEvent) {
460 34 : offset = sizeof(event_sync_msg);
461 : }
462 75 : if (event.priv.msg_len != (msgSize + offset)) {
463 10 : aicpusd_err("The len[%u] is not equal to sizeof(msg)[%zu] + [%zu].", event.priv.msg_len, msgSize, offset);
464 10 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
465 : }
466 65 : msg = &(event.priv.msg[offset]);
467 65 : if (msg == nullptr) {
468 0 : aicpusd_err("The event_info msg is nullptr.");
469 0 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
470 : }
471 65 : return AICPU_SCHEDULE_OK;
472 : }
473 :
474 23 : int32_t AicpuQueueEventProcess::ResponseEvent(const event_info& event, const char_t* const msg, const size_t len) const
475 : {
476 23 : aicpusd_info("Begin to response event event_id[%u], subevent_id[%u].", event.comm.event_id, event.comm.subevent_id);
477 23 : event_summary response = {};
478 23 : auto const msgHead = PtrToPtr<const char_t, const event_sync_msg>(event.priv.msg);
479 23 : response.dst_engine = msgHead->dst_engine;
480 23 : response.policy = ONLY;
481 23 : response.pid = msgHead->pid;
482 23 : response.grp_id = msgHead->gid;
483 23 : response.event_id = static_cast<EVENT_ID>(msgHead->event_id);
484 23 : response.subevent_id = msgHead->subevent_id;
485 23 : response.msg_len = static_cast<uint32_t>(len);
486 23 : response.msg = const_cast<char_t*>(msg);
487 23 : const int32_t drvRet = halEschedSubmitEvent(AicpuDrvManager::GetInstance().GetDeviceId(), &response);
488 23 : if (drvRet != DRV_ERROR_NONE) {
489 2 : aicpusd_err(
490 : "Failed to response event to acl event_id[%u], subevent_id[%u].", event.comm.event_id,
491 : event.comm.subevent_id);
492 2 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
493 : }
494 21 : aicpusd_info(
495 : "Successfully responded to event, event_id[%u], subevent_id[%u].", event.comm.event_id, event.comm.subevent_id);
496 21 : return AICPU_SCHEDULE_OK;
497 : }
498 :
499 3 : int32_t AicpuQueueEventProcess::ProcessQsRetWithMbuf(
500 : const event_info& event, std::shared_ptr<CallbackMsg>& callback,
501 : const bqs::QsProcMsgRspDstAicpu** const qsProcMsgRsp)
502 : {
503 3 : int32_t ret = ProcessQsRet(event, callback, qsProcMsgRsp);
504 3 : if (ret != AICPU_SCHEDULE_OK) {
505 0 : return ret;
506 : }
507 3 : Mbuf* resultMbuf = nullptr;
508 6 : const auto drvRet = halQueueDeQueue(
509 3 : AicpuDrvManager::GetInstance().GetDeviceId(), pipelineQueueId_, PtrToPtr<Mbuf*, void*>(&resultMbuf));
510 3 : if ((drvRet != DRV_ERROR_NONE) || (resultMbuf == nullptr)) {
511 0 : aicpusd_err("Dequeue from pipelineQ[%u] fail, ret is %d", pipelineQueueId_, static_cast<int32_t>(drvRet));
512 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
513 : }
514 3 : callback->buff = resultMbuf;
515 3 : ret = CopyResult(callback);
516 3 : const auto drvFreeRet = halMbufFree(callback->buff);
517 3 : if (drvFreeRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
518 0 : aicpusd_err("Failed to free mbuf, ret[%d].", drvFreeRet);
519 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
520 : }
521 3 : callback->buff = nullptr;
522 :
523 3 : return ret;
524 : }
525 :
526 12 : int32_t AicpuQueueEventProcess::CopyResult(const std::shared_ptr<CallbackMsg>& callback) const
527 : {
528 12 : const char_t* msg = nullptr;
529 12 : const int32_t ret = ParseQueueEventMessage(callback->event, msg, sizeof(bqs::QueueRouteList));
530 12 : if (ret != AICPU_SCHEDULE_OK) {
531 1 : return ret;
532 : }
533 :
534 11 : bqs::QsRouteHead* const routeHead = PtrToPtr<void, bqs::QsRouteHead>(
535 11 : ValueToPtr(PtrToPtr<const char_t, const bqs::QueueRouteList>(msg)->routeListMsgAddr));
536 11 : if (routeHead == nullptr) {
537 1 : aicpusd_err("QueueRoute is nullptr.");
538 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
539 : }
540 :
541 10 : if (callback->buff == nullptr) {
542 1 : aicpusd_err("Mbuf is nullptr.");
543 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
544 : }
545 9 : uint8_t* mbufData = nullptr;
546 9 : int32_t drvRet = halMbufGetBuffAddr(callback->buff, PtrToPtr<uint8_t*, void*>(&mbufData));
547 9 : if ((drvRet != DRV_ERROR_NONE) || (mbufData == nullptr)) {
548 1 : aicpusd_err("Failed to get mbuf data, ret[%d].", drvRet);
549 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
550 : }
551 :
552 8 : uint64_t mbufSize = 0U;
553 8 : drvRet = halMbufGetDataLen(callback->buff, &mbufSize);
554 8 : if (drvRet != DRV_ERROR_NONE) {
555 1 : aicpusd_err("Failed to get mbuf data size, ret[%d].", drvRet);
556 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
557 : }
558 :
559 7 : if (static_cast<uint64_t>(routeHead->length) != mbufSize) {
560 5 : aicpusd_err("RouteHead size[%u], but mbuf size[%lu].", routeHead->length, mbufSize);
561 5 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
562 : }
563 :
564 : const errno_t memRet =
565 2 : memcpy_s(routeHead, static_cast<size_t>(routeHead->length), mbufData, static_cast<size_t>(mbufSize));
566 2 : if (memRet != EOK) {
567 1 : aicpusd_err("Memcpy size[%u] failed, ret[%d]", routeHead->length, memRet);
568 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
569 : }
570 :
571 1 : return AICPU_SCHEDULE_OK;
572 : }
573 :
574 4 : int32_t AicpuQueueEventProcess::AddQueueAuthToQs(const bqs::QueueRoute* const queueRoute, const uint32_t routeNum)
575 : {
576 4 : std::set<uint32_t> srcQueuesId;
577 4 : std::set<uint32_t> dstQueuesId;
578 9 : for (uint32_t i = 0U; i < routeNum; ++i) {
579 5 : auto iter = grantedSrcQueueSet_.find(queueRoute[i].srcId);
580 5 : if (iter == grantedSrcQueueSet_.end()) {
581 4 : (void)srcQueuesId.insert(queueRoute[i].srcId);
582 : } else {
583 1 : aicpusd_info("aicpusd has already grant queueId[%u] read authority to qs", queueRoute[i].srcId);
584 : }
585 5 : iter = grantedDstQueueSet_.find(queueRoute[i].dstId);
586 5 : if (iter == grantedDstQueueSet_.end()) {
587 4 : (void)dstQueuesId.insert(queueRoute[i].dstId);
588 : } else {
589 1 : aicpusd_info("aicpusd has already grant queueId[%u] write authority to qs", queueRoute[i].dstId);
590 : }
591 : }
592 :
593 7 : for (auto iter = srcQueuesId.begin(); iter != srcQueuesId.end(); ++iter) {
594 4 : QueueShareAttr attr = {};
595 4 : attr.read = 1U;
596 : const int32_t drvRet =
597 4 : halQueueGrant(AicpuDrvManager::GetInstance().GetDeviceId(), static_cast<int32_t>(*iter), qsPid_, attr);
598 4 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
599 1 : aicpusd_err("Call halQueueGrant read failed, queueid[%u], ret[%d].", *iter, drvRet);
600 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
601 : }
602 3 : (void)grantedSrcQueueSet_.insert(*iter);
603 : }
604 :
605 5 : for (auto iter = dstQueuesId.begin(); iter != dstQueuesId.end(); ++iter) {
606 3 : QueueShareAttr attr = {};
607 3 : attr.write = 1U;
608 : const int32_t drvRet =
609 3 : halQueueGrant(AicpuDrvManager::GetInstance().GetDeviceId(), static_cast<int32_t>(*iter), qsPid_, attr);
610 3 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
611 1 : aicpusd_err("Call halQueueGrant write failed, queueid[%u], ret[%d].", *iter, drvRet);
612 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
613 : }
614 2 : (void)grantedDstQueueSet_.insert(*iter);
615 : }
616 2 : return AICPU_SCHEDULE_OK;
617 4 : }
618 :
619 6 : int32_t AicpuQueueEventProcess::AllocMbufAndEnqueue(
620 : const bqs::QsRouteHead* const data, const size_t size, Mbuf** const buff)
621 : {
622 6 : int32_t drvRet = halMbufAlloc(size, buff);
623 6 : if ((drvRet != DRV_ERROR_NONE) || (*buff == nullptr)) {
624 0 : aicpusd_err("Failed to alloc mbuf, size[%u], ret[%d].", size, drvRet);
625 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
626 : }
627 6 : drvRet = halMbufSetDataLen(*buff, static_cast<uint64_t>(size));
628 6 : if (drvRet != DRV_ERROR_NONE) {
629 1 : aicpusd_err("Failed to set data len[%zu] for mbuf, ret[%d]", size, drvRet);
630 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
631 : }
632 :
633 5 : bqs::QsRouteHead* mbufData = nullptr;
634 5 : drvRet = halMbufGetBuffAddr(*buff, PtrToPtr<bqs::QsRouteHead*, void*>(&mbufData));
635 5 : if ((drvRet != DRV_ERROR_NONE) || (mbufData == nullptr)) {
636 0 : aicpusd_err("Failed to get mbuf data, ret[%d]", drvRet);
637 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
638 : }
639 :
640 5 : const errno_t ret = memcpy_s(mbufData, size, data, size);
641 5 : if (ret != EOK) {
642 0 : aicpusd_err("Memcpy size=[%zu] failed, ret=[%d]", size, ret);
643 0 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
644 : }
645 :
646 : // lock for enqueue
647 5 : lockEnqueue_.Lock();
648 5 : drvRet = halQueueEnQueue(AicpuDrvManager::GetInstance().GetDeviceId(), pipelineQueueId_, *buff);
649 5 : lockEnqueue_.Unlock();
650 5 : if (drvRet != DRV_ERROR_NONE) {
651 0 : aicpusd_err("Call halQueueEnQueue error, queue id[%u], ret=[%d]", pipelineQueueId_, drvRet);
652 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
653 : }
654 5 : aicpusd_info("Call halQueueEnQueue success, queue id[%u], ret=[%d]", pipelineQueueId_, drvRet);
655 5 : return AICPU_SCHEDULE_OK;
656 : }
657 :
658 6 : int32_t AicpuQueueEventProcess::QueryQsPid()
659 : {
660 6 : aicpusd_info("Begin to QueryQsPid.");
661 : // query qs pid
662 6 : halQueryDevpidInfo para = {};
663 6 : para.hostpid = AicpuDrvManager::GetInstance().GetHostPid();
664 6 : para.proc_type = DEVDRV_PROCESS_QS;
665 6 : para.vfid = AicpuDrvManager::GetInstance().GetVfId();
666 6 : para.devid = AicpuDrvManager::GetInstance().GetDeviceId();
667 6 : const int32_t drvRet = halQueryDevpid(para, &qsPid_);
668 6 : if (drvRet != DRV_ERROR_NONE) {
669 1 : aicpusd_err("Query qs pid failed, ret=[%d]", drvRet);
670 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
671 : }
672 5 : aicpusd_info("Successfully queried QS pid.");
673 5 : return AICPU_SCHEDULE_OK;
674 : }
675 :
676 3 : int32_t AicpuQueueEventProcess::CreateGroupForMaster(
677 : std::string& outGroupName, const char_t* const inGroupName, const uint64_t size, const uint32_t allocFlag)
678 : {
679 3 : aicpusd_info("Create new group for master aicpusd[%d].", curPid_);
680 3 : std::string groupName;
681 3 : if (inGroupName == nullptr) {
682 3 : groupName = "Aicpusd_" + std::to_string(curPid_);
683 : } else {
684 0 : groupName = std::string(inGroupName);
685 : }
686 3 : GroupCfg groupConf = {};
687 3 : groupConf.maxMemSize = size;
688 3 : groupConf.privMbufFlag = BUFF_ENABLE_PRIVATE_MBUF;
689 3 : groupConf.cacheAllocFlag = allocFlag;
690 3 : int32_t drvRet = halGrpCreate(groupName.c_str(), &groupConf);
691 3 : if (drvRet != DRV_ERROR_NONE) {
692 0 : aicpusd_err("Create group failed in aicpusd[%d], result[%d]", curPid_, drvRet);
693 0 : return drvRet;
694 : }
695 3 : drvRet = halGrpAddProc(groupName.c_str(), curPid_, ALL_ATTR_FOR_GROUP);
696 3 : if (drvRet != DRV_ERROR_NONE) {
697 0 : aicpusd_err("Add group[%s] for master aicpusd[%d] failed, ret[%d]", groupName.c_str(), curPid_, drvRet);
698 0 : return drvRet;
699 : }
700 3 : drvRet = halGrpAttach(groupName.c_str(), 0);
701 3 : if (drvRet != DRV_ERROR_NONE) {
702 0 : aicpusd_err("Attach group[%s] for master aicpusd[%d] failed, ret[%d]", groupName.c_str(), curPid_, drvRet);
703 0 : return drvRet;
704 : }
705 3 : outGroupName = groupName;
706 3 : grpName_ = groupName;
707 3 : type_ = CpType::MASTER;
708 3 : BuffCfg buffConfig = {};
709 3 : drvRet = halBuffInit(&buffConfig);
710 3 : if (drvRet != DRV_ERROR_NONE) {
711 0 : aicpusd_err("Buffer initial failed for master aicpusd[%d], ret[%d]", curPid_, drvRet);
712 0 : return drvRet;
713 : }
714 3 : aicpusd_info("Create new group[%s] for master aicpusd[%d] success", groupName.c_str(), curPid_);
715 3 : return AICPU_SCHEDULE_OK;
716 3 : }
717 :
718 2 : int32_t AicpuQueueEventProcess::AttachGroupForSlave(
719 : const std::map<std::string, GroupShareAttr>& grpInfos, std::string& outGroupName)
720 : {
721 : // only one group
722 2 : const uint32_t groupNum = grpInfos.size();
723 2 : if (groupNum != 1U) {
724 0 : aicpusd_err("Slave aicpusd[%d] should own only one group rather than [%u]", curPid_, groupNum);
725 0 : return AICPU_SCHEDULE_ERROR_MULTI_GRP_ERROR;
726 : }
727 :
728 : // attach and initial process
729 2 : const std::string groupName = grpInfos.begin()->first;
730 2 : auto drvRet = halGrpAttach(groupName.c_str(), 0);
731 2 : if (drvRet != DRV_ERROR_NONE) {
732 1 : aicpusd_err("Group[%s] attach failed for slave aicpusd[%d] ret[%d]", groupName.c_str(), curPid_, drvRet);
733 1 : return drvRet;
734 : }
735 1 : outGroupName = groupName;
736 1 : type_ = CpType::SLAVE;
737 1 : grpName_ = outGroupName;
738 1 : BuffCfg buffConfig = {};
739 1 : drvRet = halBuffInit(&buffConfig);
740 1 : if (drvRet != DRV_ERROR_NONE) {
741 0 : aicpusd_err("Buffer initial failed for slave aicpusd[%d] ret[%d", curPid_, drvRet);
742 0 : return drvRet;
743 : }
744 1 : aicpusd_info("Attach group[%s] for slave aicpusd[%d] success", outGroupName.c_str(), curPid_);
745 1 : return AICPU_SCHEDULE_OK;
746 2 : }
747 :
748 9 : int32_t AicpuQueueEventProcess::GetOrCreateGroup(std::string& outGroupName)
749 : {
750 : // check group is exists
751 9 : if (!grpName_.empty()) {
752 4 : aicpusd_info("Aicpusd[%d] already created group[%s].", curPid_, grpName_.c_str());
753 4 : outGroupName = grpName_;
754 4 : return AICPU_SCHEDULE_OK;
755 : }
756 :
757 : // create or get group, need spinlock
758 5 : lockGroup_.Lock();
759 10 : const ScopeGuard lockGuard([this]() { lockGroup_.Unlock(); });
760 5 : if (!grpName_.empty()) {
761 0 : aicpusd_info("Aicpusd[%d] already created group[%s].", curPid_, grpName_.c_str());
762 0 : outGroupName = grpName_;
763 0 : return AICPU_SCHEDULE_OK;
764 : }
765 :
766 : // get group info for current process
767 5 : std::map<std::string, GroupShareAttr> grpInfos;
768 5 : const int32_t ret = AicpuDrvManager::GetInstance().QueryProcBuffInfo(curPid_, grpInfos);
769 5 : if (ret != AICPU_SCHEDULE_OK) {
770 1 : aicpusd_err("Fail to get group info of master aicpusd[%d]", curPid_);
771 1 : return ret;
772 : }
773 :
774 : // 0 group need to create group
775 4 : if (grpInfos.size() == 0U) {
776 3 : return CreateGroupForMaster(outGroupName);
777 : }
778 :
779 : // current process already in sharepool group, need attach group
780 1 : return AttachGroupForSlave(grpInfos, outGroupName);
781 5 : }
782 :
783 11 : int32_t AicpuQueueEventProcess::ShareGroupWithProcess(const std::string& groupName, const pid_t& pid) const
784 : {
785 11 : std::map<std::string, GroupShareAttr> grpInfos;
786 11 : const int32_t ret = AicpuDrvManager::GetInstance().QueryProcBuffInfo(pid, grpInfos);
787 11 : if (ret != AICPU_SCHEDULE_OK) {
788 2 : aicpusd_err("Fail to get group info of master aicpusd[%d]", pid);
789 2 : return ret;
790 : }
791 :
792 9 : for (const auto& iter : grpInfos) {
793 2 : const std::string queryGroupName(iter.first);
794 2 : if (queryGroupName != groupName) {
795 2 : if (iter.second.admin != 0U) {
796 1 : aicpusd_warn("Aicpusd already add group[%s] to slave process[%d]", groupName.c_str(), pid);
797 1 : return AICPU_SCHEDULE_OK;
798 : } else {
799 1 : aicpusd_err("Slave aicpusd[%d] already in group[%s], but doesn't has admin.", pid, groupName.c_str());
800 1 : return AICPU_SCHEDULE_ERROR_SLAVE_GRP_INVALID;
801 : }
802 : }
803 2 : }
804 7 : const int32_t drvRet = halGrpAddProc(groupName.c_str(), pid, ALL_ATTR_FOR_GROUP);
805 7 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) && (drvRet != static_cast<int32_t>(DRV_ERROR_REPEATED_INIT))) {
806 1 : aicpusd_err("Add group[%s] for slave process[%d] failed, result[%d]", groupName.c_str(), pid, drvRet);
807 1 : return drvRet;
808 : }
809 6 : return AICPU_SCHEDULE_OK;
810 11 : }
811 :
812 6 : int32_t AicpuQueueEventProcess::GrantQueue(const event_info& event)
813 : {
814 6 : std::string groupName;
815 6 : int32_t ret = GetOrCreateGroup(groupName);
816 6 : if (ret != AICPU_SCHEDULE_OK) {
817 1 : return ret;
818 : }
819 5 : if (type_ != CpType::MASTER) {
820 1 : aicpusd_err("Current process is not master, can't grant queue.");
821 1 : return AICPU_SCHEDULE_ERROR_GRANT_QUEUE_FAILED;
822 : }
823 4 : const char_t* msg = nullptr;
824 4 : ret = ParseQueueEventMessage(event, msg, sizeof(QueueGrantPara), true);
825 4 : if (ret != AICPU_SCHEDULE_OK) {
826 1 : return ret;
827 : }
828 3 : const QueueGrantPara* const grantPara = PtrToPtr<const char_t, const QueueGrantPara>(msg);
829 3 : ret = ShareGroupWithProcess(groupName, grantPara->pid);
830 3 : if (ret != AICPU_SCHEDULE_OK) {
831 1 : return ret;
832 : }
833 : const int32_t drvRet =
834 2 : halQueueGrant(grantPara->devid, static_cast<int32_t>(grantPara->qid), grantPara->pid, grantPara->attr);
835 2 : if (drvRet != DRV_ERROR_NONE) {
836 1 : aicpusd_err(
837 : "Fail to add queue[%d] authority for aicpusd[%d], result[%d].", grantPara->qid, grantPara->pid, drvRet);
838 1 : return drvRet;
839 : }
840 1 : aicpusd_info("Add queue[%d] auth for aicpusd[%d] success.", grantPara->qid, grantPara->pid);
841 1 : return AICPU_SCHEDULE_OK;
842 6 : }
843 :
844 6 : int32_t AicpuQueueEventProcess::AttachQueue(const event_info& event)
845 : {
846 6 : std::string groupName;
847 6 : int32_t ret = GetOrCreateGroup(groupName);
848 6 : if (ret != AICPU_SCHEDULE_OK) {
849 1 : return ret;
850 : }
851 5 : if (type_ != CpType::SLAVE) {
852 1 : aicpusd_warn("Current process is not slave, not need to attach queue.");
853 1 : return AICPU_SCHEDULE_OK;
854 : }
855 4 : const char_t* msg = nullptr;
856 4 : ret = ParseQueueEventMessage(event, msg, sizeof(QueueAttachPara), true);
857 4 : if (ret != AICPU_SCHEDULE_OK) {
858 1 : return ret;
859 : }
860 3 : const QueueAttachPara* const attachPara = PtrToPtr<const char_t, const QueueAttachPara>(msg);
861 3 : auto drvRet = halQueueInit(AicpuDrvManager::GetInstance().GetDeviceId());
862 3 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
863 1 : aicpusd_err("halQueueInit error, deviceId[%u], ret[%d]", AicpuDrvManager::GetInstance().GetDeviceId(), drvRet);
864 1 : return drvRet;
865 : }
866 2 : drvRet = halQueueAttach(AicpuDrvManager::GetInstance().GetDeviceId(), attachPara->qid, 0);
867 2 : if (drvRet != DRV_ERROR_NONE) {
868 1 : aicpusd_err("Fail to attach queue[%u], result[%d]", attachPara->qid, drvRet);
869 1 : return drvRet;
870 : }
871 1 : return AICPU_SCHEDULE_OK;
872 6 : }
873 :
874 1 : int32_t AicpuQueueEventProcess::ProcessProxyMsg(const event_info& event)
875 : {
876 1 : ProxyMsgRsp rsp = {};
877 1 : DoProcessProxyMsg(event, rsp);
878 1 : const int32_t resRet = ResponseEvent(event, PtrToPtr<ProxyMsgRsp, const char_t>(&rsp), sizeof(ProxyMsgRsp));
879 1 : return resRet;
880 : }
881 :
882 12 : void AicpuQueueEventProcess::DoProcessProxyMsg(const event_info& event, ProxyMsgRsp& rsp)
883 : {
884 12 : aicpusd_info("DoProcessProxyMsg, subevent: %u", event.comm.subevent_id);
885 12 : switch (event.comm.subevent_id) {
886 1 : case PROXY_SUBEVENT_CREATE_GROUP: {
887 1 : rsp.retCode = ProxyCreateGroup(event);
888 1 : break;
889 : }
890 1 : case PROXY_SUBEVENT_ALLOC_MBUF: {
891 1 : Mbuf* mbuf = nullptr;
892 1 : void* data = nullptr;
893 1 : rsp.retCode = ProxyAllocMbuf(event, &mbuf, &data);
894 1 : rsp.mbufAddr = PtrToValue(mbuf);
895 1 : rsp.dataAddr = PtrToValue(data);
896 1 : break;
897 : }
898 1 : case PROXY_SUBEVENT_FREE_MBUF: {
899 1 : rsp.retCode = ProxyFreeMbuf(event);
900 1 : break;
901 : }
902 1 : case PROXY_SUBEVENT_COPY_QMBUF: {
903 1 : rsp.retCode = ProxyCopyQMbuf(event);
904 1 : break;
905 : }
906 2 : case PROXY_SUBEVENT_ADD_GROUP: {
907 2 : rsp.retCode = ProxyAddGroup(event);
908 2 : break;
909 : }
910 4 : case PROXY_SUBEVENT_ALLOC_CACHE: {
911 4 : rsp.retCode = ProxyAllocCache(event);
912 4 : break;
913 : }
914 2 : default: {
915 2 : aicpusd_err("Unknown proxy subeventId: %u", event.comm.subevent_id);
916 2 : rsp.retCode = AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
917 2 : break;
918 : }
919 : }
920 12 : }
921 :
922 5 : int32_t AicpuQueueEventProcess::ProxyCreateGroup(const event_info& event)
923 : {
924 5 : const char_t* msg = nullptr;
925 5 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(ProxyMsgCreateGroup), true);
926 5 : if (ret != AICPU_SCHEDULE_OK) {
927 1 : return ret;
928 : }
929 4 : const ProxyMsgCreateGroup* const createGroupMsg = PtrToPtr<const char_t, const ProxyMsgCreateGroup>(msg);
930 4 : const char_t* const groupName = createGroupMsg->groupName;
931 4 : if ((!grpName_.empty())) {
932 1 : aicpusd_info("Aicpusd[%d] already created group[%s].", curPid_, grpName_.c_str());
933 1 : return AICPU_SCHEDULE_ERROR_INNER_ERROR;
934 : }
935 :
936 3 : aicpusd_info(
937 : "Proxy create group [%s], size[%lu]KB, allocSize[%ld].", groupName, createGroupMsg->size,
938 : createGroupMsg->allocSize);
939 :
940 3 : std::string outGroupName;
941 3 : ret = CreateGroupForMaster(outGroupName, groupName, createGroupMsg->size, 1U);
942 3 : if (ret != AICPU_SCHEDULE_OK) {
943 1 : aicpusd_err("Failed to create group [%s].", groupName);
944 1 : return ret;
945 : }
946 :
947 2 : if (createGroupMsg->allocSize >= 0) {
948 2 : GrpCacheAllocPara allocPar = {};
949 2 : allocPar.memSize =
950 2 : (createGroupMsg->allocSize == 0) ? createGroupMsg->size : static_cast<uint64_t>(createGroupMsg->allocSize);
951 2 : allocPar.memFlag = BUFF_SP_HUGEPAGE_ONLY;
952 2 : const auto allocRet = DoAllocCache(groupName, &allocPar);
953 2 : if (allocRet != AICPU_SCHEDULE_OK) {
954 1 : return allocRet;
955 : }
956 : }
957 1 : aicpusd_info("Proxy create group[%s] success", groupName);
958 1 : return AICPU_SCHEDULE_OK;
959 3 : }
960 :
961 4 : int32_t AicpuQueueEventProcess::ProxyAllocMbuf(const event_info& event, Mbuf** mbufPtr, void** dataPptr) const
962 : {
963 4 : const char_t* msg = nullptr;
964 4 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(ProxyMsgAllocMbuf), true);
965 4 : if (ret != AICPU_SCHEDULE_OK) {
966 1 : return ret;
967 : }
968 3 : const ProxyMsgAllocMbuf* const allocMbufMsg = PtrToPtr<const char_t, const ProxyMsgAllocMbuf>(msg);
969 3 : const uint64_t& mbufLen = allocMbufMsg->size;
970 3 : Mbuf* mbuf = nullptr;
971 3 : ret = halMbufAlloc(mbufLen, &mbuf);
972 3 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (mbuf == nullptr)) {
973 1 : aicpusd_err("Failed to alloc mbuf, size[%lu], ret=[%d].", mbufLen, ret);
974 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
975 : }
976 :
977 2 : ret = halMbufGetBuffAddr(mbuf, dataPptr);
978 2 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (*dataPptr == nullptr)) {
979 1 : aicpusd_err("Failed to get mbuf data, ret=[%d]", ret);
980 1 : (void)halMbufFree(mbuf);
981 1 : mbuf = nullptr;
982 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
983 : }
984 1 : *mbufPtr = mbuf;
985 1 : aicpusd_info("Proxy Alloc mbuf success, size[%lu].", mbufLen);
986 1 : return AICPU_SCHEDULE_OK;
987 : }
988 :
989 4 : int32_t AicpuQueueEventProcess::ProxyFreeMbuf(const event_info& event) const
990 : {
991 4 : const char_t* msg = nullptr;
992 4 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(ProxyMsgFreeMbuf), true);
993 4 : if (ret != AICPU_SCHEDULE_OK) {
994 1 : return ret;
995 : }
996 3 : const ProxyMsgFreeMbuf* const freeMbufMsg = PtrToPtr<const char_t, const ProxyMsgFreeMbuf>(msg);
997 3 : Mbuf* const mbuf = PtrToPtr<void, Mbuf>(ValueToPtr(freeMbufMsg->mbufAddr));
998 3 : if (mbuf == nullptr) {
999 1 : aicpusd_err("Null mbuf to free");
1000 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
1001 : }
1002 :
1003 2 : ret = halMbufFree(mbuf);
1004 2 : if (ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
1005 1 : aicpusd_err("Fail to free mbuf, ret is %d", ret);
1006 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1007 : }
1008 1 : aicpusd_info("Proxy free mbuf success.");
1009 1 : return AICPU_SCHEDULE_OK;
1010 : }
1011 :
1012 8 : int32_t AicpuQueueEventProcess::ProxyCopyQMbuf(const event_info& event) const
1013 : {
1014 8 : const char_t* msg = nullptr;
1015 8 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(ProxyMsgCopyQMbuf), true);
1016 8 : if (ret != AICPU_SCHEDULE_OK) {
1017 2 : return ret;
1018 : }
1019 6 : const ProxyMsgCopyQMbuf* const copyQMbufMsg = PtrToPtr<const char_t, const ProxyMsgCopyQMbuf>(msg);
1020 6 : void* destAddr = ValueToPtr(copyQMbufMsg->destAddr);
1021 6 : const auto& destLen = copyQMbufMsg->destLen;
1022 6 : const auto& queueId = copyQMbufMsg->queueId;
1023 :
1024 6 : Mbuf* mbuf = nullptr;
1025 : const auto drvRet =
1026 6 : halQueueDeQueue(AicpuDrvManager::GetInstance().GetDeviceId(), queueId, PtrToPtr<Mbuf*, void*>(&mbuf));
1027 6 : if ((drvRet != DRV_ERROR_NONE) || (mbuf == nullptr)) {
1028 1 : aicpusd_err("Fail to dequeue Mbuf from queue[%u], ret is %d", queueId, static_cast<int32_t>(drvRet));
1029 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1030 : }
1031 :
1032 0 : const ScopeGuard mbufGuard([&mbuf]() {
1033 5 : if (mbuf != nullptr) {
1034 5 : aicpusd_info("Guard to free mbuf");
1035 5 : const auto drvFreeRet = halMbufFree(mbuf);
1036 5 : if (drvFreeRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
1037 0 : aicpusd_err("Free mbuf failed, ret[%d].", drvFreeRet);
1038 : }
1039 5 : mbuf = nullptr;
1040 : }
1041 5 : });
1042 :
1043 5 : void* data = nullptr;
1044 5 : ret = halMbufGetBuffAddr(mbuf, &data);
1045 5 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (data == nullptr)) {
1046 1 : aicpusd_err("Failed to get mbuf data, ret=[%d]", ret);
1047 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1048 : }
1049 :
1050 4 : uint64_t dataLen = 0U;
1051 4 : ret = halMbufGetBuffSize(mbuf, &dataLen);
1052 4 : if (ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
1053 1 : aicpusd_err("Failed to get mbuf data size, ret=[%d]", ret);
1054 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1055 : }
1056 :
1057 3 : if (destLen < dataLen) {
1058 1 : aicpusd_err("Fail to copy for destLen[%u] < mbufLen[%u]", destLen, dataLen);
1059 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
1060 : }
1061 :
1062 2 : const errno_t eRet = memcpy_s(destAddr, destLen, data, dataLen);
1063 2 : if (eRet != EOK) {
1064 1 : aicpusd_err("Data copy failed. dstAddrLen[%lu], dataSize[%lu], ret[%d]", destLen, dataLen, eRet);
1065 1 : return AICPU_SCHEDULE_ERROR_SAFE_FUNCTION_ERR;
1066 : }
1067 :
1068 1 : aicpusd_info("Proxy copy queue[%u]'s mbuf success.", queueId);
1069 1 : return AICPU_SCHEDULE_OK;
1070 5 : }
1071 :
1072 3 : int32_t AicpuQueueEventProcess::ProxyAddGroup(const event_info& event) const
1073 : {
1074 3 : const char_t* msg = nullptr;
1075 3 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(ProxyMsgAddGroup), true);
1076 3 : if (ret != AICPU_SCHEDULE_OK) {
1077 1 : return ret;
1078 : }
1079 2 : const ProxyMsgAddGroup* const addGroupMsg = PtrToPtr<const char_t, const ProxyMsgAddGroup>(msg);
1080 2 : const char_t* const groupName = addGroupMsg->groupName;
1081 2 : const auto& pid = addGroupMsg->pid;
1082 2 : aicpusd_info("Proxy add group[%s] for pid[%d].", groupName, pid);
1083 :
1084 2 : const GroupShareAttr attrForGroup = {
1085 2 : addGroupMsg->admin, addGroupMsg->read, addGroupMsg->write, addGroupMsg->alloc, 0U};
1086 2 : ret = halGrpAddProc(groupName, pid, attrForGroup);
1087 2 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) && (ret != static_cast<int32_t>(DRV_ERROR_REPEATED_INIT))) {
1088 1 : aicpusd_err("Add group[%s] for proxy process [%d] failed, ret[%d]", groupName, pid, ret);
1089 1 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1090 : }
1091 1 : aicpusd_info("Proxy add group[%s] for pid[%d] success", groupName, pid);
1092 1 : return AICPU_SCHEDULE_OK;
1093 : }
1094 :
1095 4 : int32_t AicpuQueueEventProcess::ProxyAllocCache(const event_info& event) const
1096 : {
1097 4 : const char_t* msg = nullptr;
1098 4 : int32_t ret = ParseQueueEventMessage(event, msg, sizeof(ProxyMsgAllocCache), true);
1099 4 : if (ret != AICPU_SCHEDULE_OK) {
1100 1 : return ret;
1101 : }
1102 :
1103 3 : if (grpName_.empty()) {
1104 1 : aicpusd_err("Cannot alloc cache for group has not been created!");
1105 1 : return AICPU_SCHEDULE_ERROR_PARAMETER_NOT_VALID;
1106 : }
1107 2 : const ProxyMsgAllocCache* const allocCacheMsg = PtrToPtr<const char_t, const ProxyMsgAllocCache>(msg);
1108 2 : GrpCacheAllocPara allocPar = {};
1109 2 : allocPar.memSize = allocCacheMsg->memSize;
1110 2 : allocPar.memFlag = BUFF_SP_HUGEPAGE_ONLY;
1111 2 : allocPar.allocMaxSize = allocCacheMsg->allocMaxSize;
1112 2 : const auto allocRet = DoAllocCache(grpName_.c_str(), &allocPar);
1113 2 : if (allocRet != AICPU_SCHEDULE_OK) {
1114 1 : return allocRet;
1115 : }
1116 1 : aicpusd_info(
1117 : "Proxy alloc cache [%llu:%u] in group[%s] success", allocCacheMsg->memSize, allocCacheMsg->allocMaxSize,
1118 : grpName_.c_str());
1119 1 : return AICPU_SCHEDULE_OK;
1120 : }
1121 :
1122 4 : int32_t AicpuQueueEventProcess::DoAllocCache(const char_t* const groupName, GrpCacheAllocPara* const allocPar) const
1123 : {
1124 4 : if (&halGrpCacheAlloc == nullptr) {
1125 0 : aicpusd_err("halGrpCacheAlloc is not supported");
1126 0 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1127 : }
1128 4 : const auto allocRet = halGrpCacheAlloc(groupName, AicpuDrvManager::GetInstance().GetDeviceId(), allocPar);
1129 4 : if (allocRet != DRV_ERROR_NONE) {
1130 2 : aicpusd_err(
1131 : "Failed to allocate group [%s], size[%llu]KB, allocMaxSize[%u], ret is %d.", groupName, allocPar->memSize,
1132 : allocPar->allocMaxSize, static_cast<int32_t>(allocRet));
1133 2 : return AICPU_SCHEDULE_ERROR_DRV_ERR;
1134 : }
1135 2 : return AICPU_SCHEDULE_OK;
1136 : }
1137 : } // namespace AicpuSchedule
|
