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