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 : #include "queue_process_ccpu.h"
11 : #include "log_inner.h"
12 : #include "runtime/rt_mem_queue.h"
13 : #include "runtime/dev.h"
14 : #include "queue_schedule/qs_client.h"
15 :
16 : namespace acl {
17 0 : aclError QueueProcessorCcpu::acltdtCreateQueue(const acltdtQueueAttr *const attr, uint32_t *const qid)
18 : {
19 0 : ACL_LOG_INFO("Start to create queue");
20 0 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(qid);
21 0 : ACL_REQUIRES_OK(acltdtCreateGroup());
22 0 : constexpr int32_t deviceId = 0;
23 : static bool isQueueIint = false;
24 0 : if (!isQueueIint) {
25 0 : ACL_LOG_INFO("need to init queue once");
26 0 : const rtError_t ret = rtMemQueueInit(deviceId);
27 0 : if ((ret != ACL_RT_SUCCESS) && (ret != ACL_ERROR_RT_REPEATED_INIT)) {
28 0 : return ret;
29 : }
30 0 : isQueueIint = true;
31 : }
32 0 : ACL_REQUIRES_OK(acltdtCreateQueueWithAttr(deviceId, attr, qid));
33 0 : ACL_LOG_INFO("Successfully to execute create queue, qid is %u", *qid);
34 0 : return ACL_SUCCESS;
35 : }
36 :
37 1 : aclError QueueProcessorCcpu::acltdtCreateGroup()
38 : {
39 : static bool isGroupCreate = false;
40 1 : if (isGroupCreate) {
41 0 : return ACL_SUCCESS;
42 : }
43 1 : ACL_LOG_INFO("Start to create group");
44 1 : size_t grpNum = 0U;
45 1 : std::string grpName;
46 1 : const int32_t pid = mmGetPid();
47 1 : const std::lock_guard<std::mutex> groupLock(muForCreateGroup_);
48 1 : if (isGroupCreate) {
49 0 : return ACL_SUCCESS;
50 : }
51 1 : ACL_REQUIRES_OK(QueryGroup(pid, grpNum, grpName));
52 1 : if (grpNum == 0U) {
53 1 : ACL_LOG_INFO("need to create group");
54 1 : const rtMemGrpConfig_t grpConfig = {};
55 1 : const std::string gName = "acltdt" + std::to_string(pid);
56 1 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemGrpCreate(gName.c_str(), &grpConfig), rtMemGrpCreate);
57 :
58 1 : rtMemGrpShareAttr_t shareAttr = {};
59 1 : shareAttr.admin = 1;
60 1 : shareAttr.read = 1;
61 1 : shareAttr.write = 1;
62 1 : shareAttr.alloc = 1;
63 1 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemGrpAddProc(gName.c_str(), pid, &shareAttr), rtMemGrpAddProc);
64 1 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemGrpAttach(gName.c_str(), 0), rtMemGrpAttach);
65 1 : ACL_REQUIRES_OK(QueryGroupId(gName));
66 1 : } else {
67 0 : ACL_REQUIRES_OK(QueryGroupId(grpName));
68 : }
69 1 : ACL_REQUIRES_OK(MbufInit());
70 1 : isGroupCreate = true;
71 1 : return ACL_SUCCESS;
72 1 : }
73 :
74 1 : aclError QueueProcessorCcpu::acltdtDestroyQueue(const uint32_t qid)
75 : {
76 1 : return acltdtDestroyQueueOndevice(qid, true);
77 : }
78 :
79 1 : aclError QueueProcessorCcpu::acltdtBindQueueRoutes(acltdtQueueRouteList *const qRouteList)
80 : {
81 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(qRouteList);
82 1 : ACL_LOG_INFO("Start to acltdtBindQueueRoutes, queue route is %zu", qRouteList->routeList.size());
83 : // qs is thread mode, so no need to grant queue to qs
84 1 : constexpr int32_t deviceId = 0;
85 1 : ACL_REQUIRES_OK(InitQueueSchedule(deviceId));
86 : // get dst id
87 1 : const int32_t dstPid = mmGetPid();
88 1 : rtEschedEventSummary_t eventSum = {0, 0U, 0, 0U, 0U, nullptr, 0U, 0};
89 1 : rtEschedEventReply_t ack = {nullptr, 0U, 0U};
90 1 : bqs::QsProcMsgRsp qsRsp = {0UL, 0, 0U, 0U, 0U, {0}};
91 1 : eventSum.pid = dstPid;
92 1 : eventSum.grpId = bqs::BIND_QUEUE_GROUP_ID;
93 1 : eventSum.eventId = RT_MQ_SCHED_EVENT_QS_MSG;
94 1 : eventSum.dstEngine = static_cast<uint32_t>(RT_MQ_DST_ENGINE_CCPU_DEVICE);
95 1 : ack.buf = reinterpret_cast<char_t *>(&qsRsp);
96 1 : ack.bufLen = sizeof(qsRsp);
97 1 : const std::lock_guard<std::recursive_mutex> lk(muForQueueCtrl_);
98 1 : if (!isQsInit_) {
99 1 : ACL_REQUIRES_OK(SendConnectQsMsg(deviceId, eventSum, ack));
100 1 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemQueueAttach(deviceId, qsContactId_, 0), rtMemQueueAttach);
101 1 : isQsInit_ = true;
102 : }
103 1 : ACL_REQUIRES_OK(SendBindUnbindMsgOnDevice(qRouteList, true, eventSum, ack));
104 1 : ACL_LOG_INFO("Successfully to execute acltdtBindQueueRoutes, queue route is %zu",
105 : qRouteList->routeList.size());
106 1 : return ACL_SUCCESS;
107 1 : }
108 :
109 1 : aclError QueueProcessorCcpu::acltdtUnbindQueueRoutes(acltdtQueueRouteList *const qRouteList)
110 : {
111 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(qRouteList);
112 1 : ACL_LOG_INFO("Start to acltdtUnBindQueueRoutes, queue route is %zu", qRouteList->routeList.size());
113 : // get dst id
114 1 : const int32_t dstPid = mmGetPid();
115 1 : rtEschedEventSummary_t eventSum = {0, 0U, 0, 0U, 0U, nullptr, 0U, 0};
116 1 : rtEschedEventReply_t ack = {nullptr, 0U, 0U};
117 1 : bqs::QsProcMsgRsp qsRsp = {0UL, 0, 0U, 0U, 0U, {0}};
118 1 : eventSum.pid = dstPid;
119 1 : eventSum.grpId = bqs::BIND_QUEUE_GROUP_ID;
120 1 : eventSum.eventId = RT_MQ_SCHED_EVENT_QS_MSG;
121 1 : eventSum.dstEngine = static_cast<uint32_t>(RT_MQ_DST_ENGINE_CCPU_DEVICE);
122 1 : ack.buf = reinterpret_cast<char_t *>(&qsRsp);
123 1 : ack.bufLen = sizeof(qsRsp);
124 1 : const std::lock_guard<std::recursive_mutex> lk(muForQueueCtrl_);
125 1 : ACL_REQUIRES_OK(SendBindUnbindMsgOnDevice(qRouteList, false, eventSum, ack));
126 1 : ACL_LOG_INFO("Successfully to execute acltdtUnBindQueueRoutes, queue route is %zu",
127 : qRouteList->routeList.size());
128 1 : return ACL_SUCCESS;
129 1 : }
130 :
131 1 : aclError QueueProcessorCcpu::acltdtQueryQueueRoutes(const acltdtQueueRouteQueryInfo *const queryInfo,
132 : acltdtQueueRouteList *const qRouteList)
133 : {
134 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(queryInfo);
135 1 : ACL_REQUIRES_NOT_NULL_WITH_INPUT_REPORT(qRouteList);
136 1 : ACL_LOG_INFO("Start to acltdtQueryQueueRoutes");
137 1 : constexpr int32_t deviceId = 0;
138 : // get dst id
139 1 : const int32_t dstPid = mmGetPid();
140 1 : rtEschedEventSummary_t eventSum = {0, 0U, 0, 0U, 0U, nullptr, 0U, 0};
141 1 : rtEschedEventReply_t ack = {nullptr, 0U, 0U};
142 1 : bqs::QsProcMsgRsp qsRsp = {0UL, 0, 0U, 0U, 0U, {0}};
143 1 : eventSum.pid = dstPid;
144 1 : eventSum.grpId = bqs::BIND_QUEUE_GROUP_ID;
145 1 : eventSum.eventId = RT_MQ_SCHED_EVENT_QS_MSG;
146 1 : eventSum.dstEngine = static_cast<uint32_t>(RT_MQ_DST_ENGINE_CCPU_DEVICE);
147 1 : ack.buf = reinterpret_cast<char_t *>(&qsRsp);
148 1 : ack.bufLen = sizeof(qsRsp);
149 1 : const std::lock_guard<std::recursive_mutex> lk(muForQueueCtrl_);
150 1 : size_t routeNum = 0UL;
151 1 : ACL_REQUIRES_OK(GetQueueRouteNum(queryInfo, deviceId, eventSum, ack, routeNum));
152 1 : ACL_REQUIRES_OK(QueryQueueRoutesOnDevice(queryInfo, routeNum, eventSum, ack, qRouteList));
153 1 : return ACL_SUCCESS;
154 1 : }
155 :
156 1 : aclError QueueProcessorCcpu::QueryGroup(const int32_t pid, size_t &grpNum, std::string &grpName) const
157 : {
158 1 : rtMemGrpQueryInput_t input = {};
159 1 : input.cmd = RT_MEM_GRP_QUERY_GROUPS_OF_PROCESS;
160 1 : input.grpQueryByProc.pid = pid;
161 1 : rtMemGrpQueryOutput_t output = {};
162 1 : rtMemGrpOfProc_t outputInfo[QUERY_BUFF_GRP_MAX_NUM] = {{}};
163 1 : output.groupsOfProc = outputInfo;
164 1 : output.maxNum = QUERY_BUFF_GRP_MAX_NUM;
165 :
166 1 : ACL_REQUIRES_RTS_OK_WARN_NOT_SUPPORT(rtMemGrpQuery(&input, &output), rtMemGrpQuery);
167 1 : grpNum = output.resultNum;
168 1 : if (grpNum > 0) {
169 0 : grpName = std::string(output.groupsOfProc->groupName);
170 : }
171 1 : ACL_LOG_INFO("This proc [%d] has [%zu] group, name is %s", input.grpQueryByProc.pid,
172 : grpNum, grpName.c_str());
173 :
174 1 : return ACL_SUCCESS;
175 : }
176 :
177 1 : aclError QueueProcessorCcpu::MbufInit() const
178 : {
179 : static bool isMbufInit = false;
180 1 : if (!isMbufInit) {
181 1 : rtMemBuffCfg_t cfg = {{}};
182 1 : const rtError_t ret = rtMbufInit(&cfg);
183 1 : if ((ret != ACL_RT_SUCCESS) && (ret != ACL_ERROR_RT_REPEATED_INIT)) {
184 0 : return ret;
185 : }
186 1 : isMbufInit = true;
187 : }
188 1 : return ACL_SUCCESS;
189 : }
190 :
191 1 : aclError QueueProcessorCcpu::acltdtAllocBuf(const size_t size, const uint32_t type, acltdtBuf *const buf)
192 : {
193 1 : ACL_REQUIRES_OK(acltdtCreateGroup());
194 1 : ACL_REQUIRES_OK(acltdtAllocBufData(size, type, buf));
195 1 : return ACL_SUCCESS;
196 : }
197 : }
|
