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 "config/config_info_operator.h"
12 :
13 : #include <set>
14 : #include <securec.h>
15 : #include <sstream>
16 :
17 : #include "hccl/hccl_ex.h"
18 : #include "hccl/comm_channel_manager.h"
19 : #include "common/type_def.h"
20 : #include "queue_schedule/dgw_client.h"
21 : #include "common/bqs_log.h"
22 : #include "queue_manager.h"
23 : #include "profile_manager.h"
24 : #include "statistic_manager.h"
25 : #include "subscribe_manager.h"
26 : #include "schedule_config.h"
27 : #include "dynamic_sched_mgr.hpp"
28 : #include "queue_schedule_hal_interface_ref.h"
29 :
30 : namespace bqs {
31 : namespace {
32 : // allowed max routes number
33 : constexpr size_t MAX_ROUTES_NUM = 8000UL;
34 : // allowed max endpoints number in one group
35 : constexpr uint32_t MAX_ENDPOINTS_NUM_IN_SINGLE_GROUP = 1000U;
36 : // max tag depth: hccl tag depth is 1024, so here is 1024/2
37 : constexpr uint32_t MAX_TAG_DEPTH = 512U;
38 : constexpr uint64_t INITIAL_MEMORY_SIZE = 5UL * 1024UL * 1024UL * 1024UL; // 5G
39 : constexpr uint16_t RESOURCE_ID_HOST_DEVICE_BIT_NUM = 14;
40 : constexpr uint16_t RESOURCE_ID_ENABLE_BIT_MASK = 0x8000;
41 : constexpr uint16_t ROUCE_ID_DEVICE_ID_DATA_MASK = 0x3FFF;
42 : const std::unordered_set<int32_t> CMD_PROCESSED_BY_ALL_RES = {
43 : static_cast<int32_t>(ConfigCmd::DGW_CFG_CMD_BIND_ROUTE),
44 : static_cast<int32_t>(ConfigCmd::DGW_CFG_CMD_UNBIND_ROUTE),
45 : static_cast<int32_t>(ConfigCmd::DGW_CFG_CMD_STOP_SCHEDULE),
46 : static_cast<int32_t>(ConfigCmd::DGW_CFG_CMD_CLEAR_AND_RESTART_SCHEDULE)};
47 : }
48 :
49 38 : ConfigInfoOperator::ConfigInfoOperator(const uint32_t deviceId, const std::string groupNames)
50 38 : : deviceId_(deviceId), groupNames_(groupNames), clientVersion_(0U)
51 38 : {}
52 :
53 88 : BqsStatus ConfigInfoOperator::ParseConfigEvent(const uint32_t subEventId, const uint32_t queueId, void *mbuf,
54 : const uint16_t clientVersion)
55 : {
56 88 : clientVersion_ = clientVersion;
57 : // get data buffer from mbuf
58 88 : void *mbufData = nullptr;
59 88 : auto getBuffRet = halMbufGetBuffAddr(PtrToPtr<void, Mbuf>(mbuf), &mbufData);
60 88 : if ((getBuffRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (mbufData == nullptr)) {
61 1 : BQS_LOG_ERROR("halMbufGetBuffAddr from queue[%u] in device[%u] failed, error[%d]",
62 : queueId, deviceId_, getBuffRet);
63 1 : return BQS_STATUS_DRIVER_ERROR;
64 : }
65 : // get mbuf len
66 87 : uint64_t dataLen = 0UL;
67 87 : getBuffRet = halMbufGetDataLen(PtrToPtr<void, Mbuf>(mbuf), &dataLen);
68 87 : if (getBuffRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
69 1 : BQS_LOG_ERROR("halMbufGetDataLen from queue[%u] in device[%u] failed, error[%d]",
70 : queueId, deviceId_, getBuffRet);
71 1 : return BQS_STATUS_DRIVER_ERROR;
72 : }
73 :
74 : // statistic info
75 86 : auto ret = BQS_STATUS_OK;
76 86 : const uintptr_t mbufDataAddr = PtrToValue(mbufData);
77 86 : const QueueSubEventType subEventType = static_cast<QueueSubEventType>(subEventId);
78 86 : switch (subEventType) {
79 40 : case QueueSubEventType::UPDATE_CONFIG: {
80 40 : ret = PreprocessUpdateCfgInfo(mbufDataAddr, static_cast<uint64_t>(dataLen));
81 40 : break;
82 : }
83 23 : case QueueSubEventType::QUERY_CONFIG_NUM: {
84 23 : ret = QueryConfigNum(mbufDataAddr, static_cast<uint64_t>(dataLen));
85 23 : break;
86 : }
87 9 : case QueueSubEventType::QUERY_CONFIG: {
88 9 : ret = QueryConfig(mbufDataAddr, static_cast<uint64_t>(dataLen));
89 9 : break;
90 : }
91 7 : case QueueSubEventType::DGW_CREATE_HCOM_HANDLE: {
92 7 : ret = CreateHcomHandle(mbufDataAddr, static_cast<uint64_t>(dataLen));
93 7 : break;
94 : }
95 6 : case QueueSubEventType::DGW_DESTORY_HCOM_HANDLE: {
96 6 : ret = DestroyHcomHandle(mbufDataAddr, static_cast<uint64_t>(dataLen));
97 6 : break;
98 : }
99 1 : default: {
100 1 : BQS_LOG_ERROR("Unsupport subEventId[%u] in bind relation procedure", subEventId);
101 1 : ret = BQS_STATUS_PARAM_INVALID;
102 1 : break;
103 : }
104 : }
105 86 : return ret;
106 : }
107 :
108 39 : BqsStatus ConfigInfoOperator::ProcessUpdateConfig(const uint32_t index)
109 : {
110 39 : BQS_LOG_INFO("Update config[add/del group, bind/unbind route], stage [server:process]");
111 : // no need to check cfgInfo and updateCfgInfo_ nullptr
112 39 : ConfigInfo * const cfgInfo = updateCfgInfo_->cfgInfo;
113 39 : if ((CMD_PROCESSED_BY_ALL_RES.count(static_cast<int32_t>(cfgInfo->cmd)) == 0U) && (index != 0U)) {
114 1 : BQS_LOG_INFO("Thread[%u] need not process cmd[%d]", index, static_cast<int32_t>(cfgInfo->cmd));
115 1 : return BQS_STATUS_OK;
116 : }
117 :
118 38 : auto ret = BQS_STATUS_OK;
119 38 : switch (cfgInfo->cmd) {
120 13 : case ConfigCmd::DGW_CFG_CMD_BIND_ROUTE:
121 : case ConfigCmd::DGW_CFG_CMD_UNBIND_ROUTE: {
122 13 : ret = ProcessUpdateRoutes(index);
123 13 : break;
124 : }
125 7 : case ConfigCmd::DGW_CFG_CMD_ADD_GROUP: {
126 7 : ret = ProcessAddGroup();
127 7 : break;
128 : }
129 8 : case ConfigCmd::DGW_CFG_CMD_DEL_GROUP: {
130 8 : ret = ProcessDelGroup();
131 8 : break;
132 : }
133 1 : case ConfigCmd::DGW_CFG_CMD_UPDATE_PROFILING: {
134 1 : ret = ProcessUpdateProfiling();
135 1 : break;
136 : }
137 2 : case ConfigCmd::DGW_CFG_CMD_SET_HCCL_PROTOCOL: {
138 2 : ret = ProcessUpdateHcclProtocol();
139 2 : break;
140 : }
141 4 : case ConfigCmd::DGW_CFG_CMD_INIT_DYNAMIC_SCHEDULE: {
142 4 : ret = ProcessInitDynamicSched();
143 4 : break;
144 : }
145 1 : case ConfigCmd::DGW_CFG_CMD_STOP_SCHEDULE: {
146 1 : ret = ProcessStopSchedule(index);
147 1 : break;
148 : }
149 1 : case ConfigCmd::DGW_CFG_CMD_CLEAR_AND_RESTART_SCHEDULE: {
150 1 : ret = ProcessRestartSchedule(index);
151 1 : break;
152 : }
153 1 : default: {
154 1 : ret = BQS_STATUS_PARAM_INVALID;
155 1 : BQS_LOG_WARN("cmd[%d] is invalid.", static_cast<int32_t>(cfgInfo->cmd));
156 1 : break;
157 : }
158 : }
159 38 : return ret;
160 : }
161 :
162 9 : BqsStatus ConfigInfoOperator::QueryConfig(const uintptr_t mbufData, const uint64_t dataLen) const
163 : {
164 9 : ConfigQuery * const cfgQry = PtrToPtr<void, ConfigQuery>(ValueToPtr(mbufData));
165 9 : if (cfgQry->mode == QueryMode::DGW_QUERY_MODE_GROUP) {
166 3 : return QueryGroup(mbufData, dataLen, false);
167 : }
168 6 : return QueryRoutes(mbufData, dataLen, false);
169 : }
170 :
171 23 : BqsStatus ConfigInfoOperator::QueryConfigNum(const uintptr_t mbufData, const uint64_t dataLen) const
172 : {
173 23 : ConfigQuery * const cfgQry = PtrToPtr<void, ConfigQuery>(ValueToPtr(mbufData));
174 23 : if (cfgQry->mode == QueryMode::DGW_QUERY_MODE_GROUP) {
175 8 : return QueryGroup(mbufData, dataLen, true);
176 : }
177 15 : return QueryRoutes(mbufData, dataLen, true);
178 : }
179 :
180 7 : void ConfigInfoOperator::SplitStringWithDelimeter(const std::string rawStr, const char_t delimeter,
181 : std::vector<std::string> &results) const
182 : {
183 7 : if (rawStr.empty()) {
184 1 : BQS_LOG_INFO("str to split is empty");
185 1 : return;
186 : }
187 6 : std::stringstream strStream(rawStr);
188 6 : std::string strElement;
189 18 : while (getline(strStream, strElement, delimeter)) {
190 12 : results.emplace_back(strElement);
191 : }
192 6 : }
193 :
194 11 : BqsStatus ConfigInfoOperator::QueryGroupAllocInfo()
195 : {
196 11 : if (!grpAllocInfos_.empty()) {
197 1 : BQS_LOG_INFO("grpAllocInfos_ has been inited");
198 1 : return BQS_STATUS_OK;
199 : }
200 10 : std::vector<std::string> groupNames;
201 10 : if (groupNames_.empty()) {
202 4 : const auto ret = QureySelfMemGroup(groupNames);
203 4 : if (ret != BQS_STATUS_OK) {
204 0 : return ret;
205 : }
206 : } else {
207 6 : SplitStringWithDelimeter(groupNames_, ',', groupNames);
208 : }
209 18 : for (const auto &groupName: groupNames) {
210 10 : GrpQueryGroupAddrPara queryPara = {};
211 10 : const errno_t eRet = memcpy_s(queryPara.grpName, BUFF_GRP_NAME_LEN,
212 10 : PtrToPtr<const char, const void>(groupName.c_str()) , strlen(groupName.c_str()) + 1);
213 10 : if (eRet != EOK) {
214 1 : BQS_LOG_ERROR("Failed to memcpy, ret[%d].", eRet);
215 1 : grpAllocInfos_.clear();
216 2 : return BQS_STATUS_INNER_ERROR;
217 : }
218 9 : queryPara.devId = deviceId_;
219 :
220 9 : GrpQueryGroupAddrInfo queryResults[BUFF_GROUP_ADDR_MAX_NUM] = {};
221 9 : uint32_t resultSize = 0U;
222 9 : const auto drvRet = halGrpQuery(GRP_QUERY_GROUP_ADDR_INFO, &queryPara,
223 : static_cast<uint32_t>(sizeof(queryPara)), &queryResults[0U], &resultSize);
224 9 : if ((drvRet != DRV_ERROR_NONE) || (static_cast<size_t>(resultSize) < sizeof(GrpQueryGroupAddrInfo))) {
225 1 : BQS_LOG_ERROR("Failed to halGrpQuery for group[%s], device[%u], ret[%d], resultSize[%u]",
226 : queryPara.grpName, queryPara.devId, static_cast<int32_t>(drvRet), resultSize);
227 1 : grpAllocInfos_.clear();
228 1 : return BQS_STATUS_DRIVER_ERROR;
229 : }
230 :
231 8 : const uint32_t resultLen = static_cast<uint32_t>(resultSize / sizeof(GrpQueryGroupAddrInfo));
232 16 : for (uint32_t index = 0U; index < resultLen; ++index) {
233 8 : grpAllocInfos_.emplace_back(queryResults[static_cast<size_t>(index)]);
234 : }
235 : }
236 8 : return BQS_STATUS_OK;
237 10 : }
238 :
239 7 : BqsStatus ConfigInfoOperator::QureySelfMemGroup(std::vector<std::string> &groupNames) const
240 : {
241 7 : std::unique_ptr<GroupQueryOutput> groupInfoPtr(new (std::nothrow) GroupQueryOutput());
242 7 : if (groupInfoPtr == nullptr) {
243 0 : BQS_LOG_ERROR("Fail to allocate GroupQueryOutput");
244 0 : return BQS_STATUS_INNER_ERROR;
245 : }
246 7 : GroupQueryOutput &groupInfo = *(groupInfoPtr.get());
247 7 : uint32_t groupInfoLen = 0U;
248 7 : pid_t curPid = drvDeviceGetBareTgid();
249 : // query group info for current qs process
250 7 : auto drvRet = halGrpQuery(GRP_QUERY_GROUPS_OF_PROCESS, &curPid, static_cast<uint32_t>(sizeof(curPid)),
251 : PtrToPtr<GroupQueryOutput, void>(&groupInfo), &groupInfoLen);
252 7 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
253 1 : BQS_LOG_ERROR("halGrpQuery of qs[%d] failed, ret[%d]", curPid, drvRet);
254 1 : return BQS_STATUS_DRIVER_ERROR;
255 : }
256 : // not in any group, cannot do attach process
257 6 : if (groupInfoLen == 0U) {
258 4 : BQS_LOG_WARN("QS has not been added to any memory group!");
259 4 : return BQS_STATUS_OK;
260 : }
261 2 : if ((groupInfoLen % sizeof(groupInfo.grpQueryGroupsOfProcInfo[0])) != 0U) {
262 1 : BQS_LOG_ERROR("Group info size[%d] is invalid", groupInfoLen);
263 1 : return BQS_STATUS_DRIVER_ERROR;
264 : }
265 1 : const uint32_t groupNum = static_cast<uint32_t>(groupInfoLen / sizeof(groupInfo.grpQueryGroupsOfProcInfo[0]));
266 1025 : for (uint32_t i = 0U; i < groupNum; ++i) {
267 1024 : std::string grpName(groupInfo.grpQueryGroupsOfProcInfo[i].groupName);
268 1024 : if (!IsSvmShareGrp(grpName)) {
269 1024 : groupNames.emplace_back(grpName);
270 : }
271 1024 : }
272 1 : return BQS_STATUS_OK;
273 7 : }
274 :
275 1024 : bool ConfigInfoOperator::IsSvmShareGrp(const std::string &grpName) const
276 : {
277 1024 : return grpName.find("svm_share_grp") != std::string::npos;
278 : }
279 :
280 8 : BqsStatus ConfigInfoOperator::CreateHcomHandle(const uintptr_t mbufData, const uint64_t dataLen)
281 : {
282 8 : const auto queryGroupRet = QueryGroupAllocInfo();
283 8 : if (queryGroupRet != BQS_STATUS_OK) {
284 0 : return queryGroupRet;
285 : }
286 8 : HcomHandleInfo * const info = PtrToPtr<void, HcomHandleInfo>(ValueToPtr(mbufData));
287 8 : if (info->rankTableLen == 0UL) {
288 0 : BQS_LOG_ERROR("Invalid rank table len[%lu].", info->rankTableLen);
289 0 : return BQS_STATUS_PARAM_INVALID;
290 : }
291 :
292 8 : uint32_t tempAddHcomTab = 0U;
293 8 : bool isOverflow = false;
294 8 : BqsCheckAssign32UAdd(static_cast<uint32_t>(sizeof(HcomHandleInfo)), info->rankTableLen, tempAddHcomTab, isOverflow);
295 8 : if (isOverflow) {
296 1 : BQS_LOG_ERROR("tempAddHcomTab[%u] is invalid.", tempAddHcomTab);
297 1 : return BQS_STATUS_PARAM_INVALID;
298 : }
299 :
300 7 : uint32_t cfgLen = 0U;
301 7 : BqsCheckAssign32UAdd(tempAddHcomTab, static_cast<uint32_t>(sizeof(CfgRetInfo)), cfgLen, isOverflow);
302 7 : if (isOverflow) {
303 0 : BQS_LOG_ERROR("cfgLen[%u] is invalid.", cfgLen);
304 0 : return BQS_STATUS_PARAM_INVALID;
305 : }
306 :
307 : // check dataLen
308 7 : if (dataLen < cfgLen) {
309 0 : BQS_LOG_ERROR("dataLen[%lu] is invalid, cfgLen is [%u].", dataLen, cfgLen);
310 0 : return BQS_STATUS_PARAM_INVALID;
311 : }
312 :
313 : // get rank table
314 7 : char_t * const rankTablePtr = PtrToPtr<void, char>(ValueToPtr(mbufData + sizeof(HcomHandleInfo)));
315 7 : const std::string rankTable(rankTablePtr, info->rankTableLen);
316 :
317 7 : auto result = BQS_STATUS_OK;
318 : // create hcom handle
319 7 : CommAttr attr = {};
320 7 : attr.deviceId = deviceId_;
321 7 : HcclComm hcomHandle = nullptr;
322 7 : const HcclResult hcclRet = HcclInitComm(rankTable.c_str(), static_cast<uint32_t>(info->rankId), &attr, &hcomHandle);
323 7 : if (hcclRet != HCCL_SUCCESS) {
324 0 : result = BQS_STATUS_HCCL_ERROR;
325 0 : BQS_LOG_ERROR("Failed to create hcom handle, hccl ret is[%d].", static_cast<int32_t>(hcclRet));
326 : } else {
327 13 : for (const auto &grpAllocInfo: grpAllocInfos_) {
328 7 : const uint64_t memorySize = (bqs::RunContext::HOST != bqs::GetRunContext()) ?
329 7 : static_cast<uint64_t>(grpAllocInfo.size) : INITIAL_MEMORY_SIZE;
330 7 : const auto registerRet = HcclRegisterMemory(hcomHandle,
331 7 : ValueToPtr(static_cast<uint64_t>(grpAllocInfo.addr)), memorySize);
332 7 : if (registerRet != HCCL_SUCCESS) {
333 1 : result = BQS_STATUS_HCCL_ERROR;
334 1 : BQS_LOG_ERROR("Failed to register memory, hccl ret is[%d].", static_cast<int32_t>(registerRet));
335 1 : HcclFinalizeComm(hcomHandle);
336 1 : break;
337 : }
338 6 : BQS_LOG_INFO("Register meomory size[%lu]", memorySize);
339 : }
340 7 : if (result == BQS_STATUS_OK) {
341 6 : info->hcomHandle = PtrToValue(hcomHandle);
342 6 : BQS_LOG_INFO("Success to create hcom handle[%lu]", info->hcomHandle);
343 : }
344 : }
345 :
346 : // write result to mbuf
347 7 : CfgRetInfo * const retInfo = PtrToPtr<void, CfgRetInfo>(
348 7 : ValueToPtr(mbufData + sizeof(HcomHandleInfo) + info->rankTableLen));
349 7 : retInfo->retCode = static_cast<int32_t>(result);
350 :
351 7 : return BQS_STATUS_OK;
352 7 : }
353 :
354 7 : BqsStatus ConfigInfoOperator::DestroyHcomHandle(const uintptr_t mbufData, const uint64_t dataLen) const
355 : {
356 7 : HcomHandleInfo * const info = PtrToPtr<void, HcomHandleInfo>(ValueToPtr(mbufData));
357 : // check dataLen
358 7 : bool overFlow = false;
359 7 : uint64_t cfgLen = BqsCheckAssign64UAdd(static_cast<uint64_t>(sizeof(HcomHandleInfo) + sizeof(CfgRetInfo)),
360 : info->rankTableLen, overFlow);
361 7 : if (overFlow || (dataLen < cfgLen)) {
362 1 : BQS_LOG_ERROR("dataLen[%lu] is invalid, cfgLen is [%lu].", dataLen, cfgLen);
363 1 : return BQS_STATUS_PARAM_INVALID;
364 : }
365 :
366 : // get hcom handle from mbuf
367 6 : const HcclComm hcomHandle = ValueToPtr(info->hcomHandle);
368 6 : if (hcomHandle == nullptr) {
369 0 : BQS_LOG_ERROR("Hcom handle is nullptr.");
370 0 : return BQS_STATUS_PARAM_INVALID;
371 : }
372 6 : BQS_LOG_RUN_INFO("Begin to Destroy hcom handle[%lu].", info->hcomHandle);
373 12 : for (const auto &grpAllocInfo: grpAllocInfos_) {
374 6 : const auto unRegisterRet = HcclUnregisterMemory(hcomHandle,
375 6 : ValueToPtr(static_cast<uint64_t>(grpAllocInfo.addr)));
376 6 : if (unRegisterRet != HCCL_SUCCESS) {
377 4 : BQS_LOG_ERROR("Failed to unRegister memory, hccl ret is[%d].", static_cast<int32_t>(unRegisterRet));
378 : }
379 : }
380 6 : BQS_LOG_RUN_INFO("After HcclUnregisterMemory when Destroy hcom handle[%lu].", info->hcomHandle);
381 :
382 : // destroy hcom handle
383 6 : auto result = BQS_STATUS_OK;
384 6 : const HcclResult hcclRet = HcclFinalizeComm(hcomHandle);
385 6 : if (hcclRet != HCCL_SUCCESS) {
386 0 : result = BQS_STATUS_HCCL_ERROR;
387 0 : BQS_LOG_ERROR("Failed to destroy hcom handle, hccl ret is[%d].", static_cast<int32_t>(hcclRet));
388 : } else {
389 6 : BQS_LOG_INFO("Success to destroy hcom handle[%lu]", info->hcomHandle);
390 : }
391 6 : BQS_LOG_RUN_INFO("After HcclFinalizeComm when Destroy hcom handle[%lu].", info->hcomHandle);
392 :
393 : // write result to mbuf
394 6 : CfgRetInfo * const retInfo = PtrToPtr<void, CfgRetInfo>(
395 6 : ValueToPtr(mbufData + sizeof(HcomHandleInfo) + info->rankTableLen));
396 6 : retInfo->retCode = static_cast<int32_t>(result);
397 :
398 6 : bqs::StatisticManager::GetInstance().ResetStatistic();
399 6 : bqs::ProfileManager::GetInstance(0U).ResetProfiling();
400 6 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
401 0 : bqs::ProfileManager::GetInstance(1U).ResetProfiling();
402 : }
403 6 : return BQS_STATUS_OK;
404 : }
405 :
406 13 : BqsStatus ConfigInfoOperator::QueryGroup(const uintptr_t mbufData, const uint64_t dataLen, const bool onlyQryNum) const
407 : {
408 13 : if (dataLen < sizeof(ConfigQuery)) {
409 1 : BQS_LOG_ERROR("dataLen[%lu] is invalid.", dataLen);
410 1 : return BQS_STATUS_PARAM_INVALID;
411 : }
412 : // no need check cfgQry nullptr
413 12 : ConfigQuery * const cfgQry = PtrToPtr<void, ConfigQuery>(ValueToPtr(mbufData));
414 12 : const uint32_t groupId = static_cast<uint32_t>(cfgQry->qry.groupQry.groupId);
415 12 : auto &entitiesInGroup = BindRelation::GetInstance().GetEntitiesInGroup(groupId);
416 :
417 12 : const size_t endpointNum = onlyQryNum ? 0UL : cfgQry->qry.routeQry.routeNum;
418 : // check total len
419 12 : const size_t totalLen = onlyQryNum ? (sizeof(ConfigQuery) + sizeof(CfgRetInfo)) :
420 3 : (sizeof(ConfigQuery) + sizeof(ConfigInfo) + (endpointNum * sizeof(Endpoint)) + sizeof(CfgRetInfo));
421 12 : if (dataLen != totalLen) {
422 1 : BQS_LOG_ERROR("mbuf dataLen[%lu] is not equal with totalLen[%zu].", dataLen, totalLen);
423 1 : return BQS_STATUS_PARAM_INVALID;
424 : }
425 :
426 11 : if (onlyQryNum) {
427 8 : cfgQry->qry.groupQry.endpointNum = static_cast<uint32_t>(entitiesInGroup.size());
428 8 : CfgRetInfo * const retInfo = PtrToPtr<void, CfgRetInfo>(ValueToPtr(mbufData + sizeof(ConfigQuery)));
429 8 : retInfo->retCode = (cfgQry->qry.groupQry.endpointNum == 0U) ?
430 : static_cast<int32_t>(BQS_STATUS_GROUP_NOT_EXIST) : static_cast<int32_t>(BQS_STATUS_OK);
431 8 : BQS_LOG_INFO("endpointNum is %u in group[%u].", cfgQry->qry.groupQry.endpointNum, groupId);
432 8 : return BQS_STATUS_OK;
433 : }
434 :
435 : // check number
436 3 : const uintptr_t results = mbufData + (totalLen - sizeof(CfgRetInfo));
437 3 : CfgRetInfo * const retInfo = PtrToPtr<void, CfgRetInfo>(ValueToPtr(results));
438 3 : if (endpointNum != entitiesInGroup.size()) {
439 0 : retInfo->retCode = static_cast<int32_t>(BQS_STATUS_PARAM_INVALID);
440 0 : BQS_LOG_ERROR("endpoint num in group[%u] info is [%zu], but searched endpoint num is [%zu].",
441 : groupId, endpointNum, entitiesInGroup.size());
442 0 : return BQS_STATUS_PARAM_INVALID;
443 : }
444 :
445 3 : BQS_LOG_INFO("Group [get], stage [server:process], relation [size:%zu]", entitiesInGroup.size());
446 : // convert and record route
447 : Endpoint * const endpoints =
448 3 : PtrToPtr<void, Endpoint>(ValueToPtr(mbufData + sizeof(ConfigQuery) + sizeof(ConfigInfo)));
449 3 : size_t idx = 0UL;
450 10 : for (auto &entity : entitiesInGroup) {
451 7 : Endpoint * const endpoint = PtrAdd<Endpoint>(endpoints, endpointNum, idx);
452 7 : (void)ConvertToEndpoint(*entity, *endpoint);
453 7 : idx++;
454 : }
455 3 : retInfo->retCode = static_cast<int32_t>(BQS_STATUS_OK);
456 3 : return BQS_STATUS_OK;
457 : }
458 :
459 23 : BqsStatus ConfigInfoOperator::QueryRoutes(const uintptr_t mbufData, const uint64_t dataLen, const bool onlyQryNum) const
460 : {
461 : // check data len
462 23 : if (dataLen < sizeof(ConfigQuery)) {
463 0 : BQS_LOG_ERROR("dataLen[%lu] is invalid.", dataLen);
464 0 : return BQS_STATUS_PARAM_INVALID;
465 : }
466 : // no need check cfgQry nullptr
467 23 : ConfigQuery * const cfgQry = PtrToPtr<void, ConfigQuery>(ValueToPtr(mbufData));
468 :
469 23 : const size_t routeNum = onlyQryNum ? 0UL : cfgQry->qry.routeQry.routeNum;
470 : // check total len
471 23 : const size_t totalLen = onlyQryNum ? (sizeof(ConfigQuery) + sizeof(CfgRetInfo)) :
472 6 : (sizeof(ConfigQuery) + sizeof(ConfigInfo) + (routeNum * sizeof(Route)) + sizeof(CfgRetInfo));
473 23 : if (dataLen != totalLen) {
474 1 : BQS_LOG_ERROR("mbuf dataLen[%lu] is not equal with totalLen[%zu].", dataLen, totalLen);
475 1 : return BQS_STATUS_PARAM_INVALID;
476 : }
477 :
478 22 : auto ret = BQS_STATUS_OK;
479 22 : switch (cfgQry->mode) {
480 4 : case QueryMode::DGW_QUERY_MODE_SRC_ROUTE: {
481 4 : const EntityInfoPtr src = CreateEntityInfo(cfgQry->qry.routeQry.src, true);
482 4 : ret = (src == nullptr) ? BQS_STATUS_FAILED : QueryRoutesBySrc(mbufData, *src, onlyQryNum);
483 4 : break;
484 4 : }
485 5 : case QueryMode::DGW_QUERY_MODE_DST_ROUTE: {
486 5 : const EntityInfoPtr dst = CreateEntityInfo(cfgQry->qry.routeQry.dst, true);
487 5 : ret = (dst == nullptr) ? BQS_STATUS_FAILED : QueryRoutesByDst(mbufData, *dst, onlyQryNum);
488 5 : break;
489 5 : }
490 8 : case QueryMode::DGW_QUERY_MODE_ALL_ROUTE: {
491 8 : ret = QueryAllRoutes(mbufData, onlyQryNum);
492 8 : break;
493 : }
494 4 : case QueryMode::DGW_QUERY_MODE_SRC_DST_ROUTE: {
495 4 : const EntityInfoPtr src = CreateEntityInfo(cfgQry->qry.routeQry.src, true);
496 4 : const EntityInfoPtr dst = CreateEntityInfo(cfgQry->qry.routeQry.dst, true);
497 4 : ret = ((src == nullptr) || (dst == nullptr)) ? BQS_STATUS_FAILED :
498 4 : QueryRoutesBySrcAndDst(mbufData, *src, *dst, onlyQryNum);
499 4 : break;
500 4 : }
501 1 : default: {
502 1 : ret = BQS_STATUS_PARAM_INVALID;
503 1 : BQS_LOG_ERROR("Unsupported query type{0:src, 1:dst, 2:src-and-dst 3:all}:%d",
504 : static_cast<int32_t>(cfgQry->mode));
505 1 : break;
506 : }
507 : }
508 22 : return ret;
509 : }
510 :
511 4 : BqsStatus ConfigInfoOperator::QueryRoutesBySrc(const uintptr_t mbufData, const EntityInfo &src,
512 : const bool onlyQryNum) const
513 : {
514 4 : std::list<std::pair<const EntityInfo *, const EntityInfo *>> routeList;
515 :
516 4 : QueryRoutesBySrcFromRelation(src, BindRelation::GetInstance().GetSrcToDstRelation(), routeList);
517 4 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
518 4 : QueryRoutesBySrcFromRelation(src, BindRelation::GetInstance().GetSrcToDstExtraRelation(), routeList);
519 : }
520 :
521 8 : return SaveQueryResult(routeList, mbufData, onlyQryNum);
522 4 : }
523 :
524 8 : void ConfigInfoOperator::QueryRoutesBySrcFromRelation(const EntityInfo &src,
525 : const MapEnitityInfoToInfoSet &srcToDstRelation,
526 : std::list<std::pair<const EntityInfo*, const EntityInfo*>> &routeList) const
527 : {
528 8 : const auto iter = srcToDstRelation.find(src);
529 8 : if (iter == srcToDstRelation.end()) {
530 5 : BQS_LOG_WARN("Record does not exist according to src Id:[%u] type:[%d]", src.GetId(),
531 : static_cast<int32_t>(src.GetType()));
532 : } else {
533 : // generate route list
534 3 : const auto &dstSet = iter->second;
535 8 : for (auto dstIter = dstSet.begin(); dstIter != dstSet.end(); ++dstIter) {
536 5 : routeList.emplace_back(std::make_pair(&src, &(*dstIter)));
537 : }
538 : }
539 8 : }
540 :
541 5 : BqsStatus ConfigInfoOperator::QueryRoutesByDst(const uintptr_t mbufData, const EntityInfo &dst,
542 : const bool onlyQryNum) const
543 : {
544 5 : std::list<std::pair<const EntityInfo *, const EntityInfo *>> routeList;
545 :
546 5 : QueryRoutesByDstFromRelation(dst, BindRelation::GetInstance().GetDstToSrcRelation(), routeList);
547 5 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
548 5 : QueryRoutesByDstFromRelation(dst, BindRelation::GetInstance().GetDstToSrcExtraRelation(), routeList);
549 : }
550 :
551 10 : return SaveQueryResult(routeList, mbufData, onlyQryNum);
552 5 : }
553 :
554 10 : void ConfigInfoOperator::QueryRoutesByDstFromRelation(const EntityInfo &dst,
555 : const MapEnitityInfoToInfoSet &dstToSrcRelation,
556 : std::list<std::pair<const EntityInfo*, const EntityInfo*>> &routeList) const
557 : {
558 10 : const auto iter = dstToSrcRelation.find(dst);
559 10 : if (iter == dstToSrcRelation.end()) {
560 7 : BQS_LOG_WARN("Record does not exist according to dst Id:[%u] type:[%d]", dst.GetId(),
561 : static_cast<int32_t>(dst.GetType()));
562 : } else {
563 3 : const auto &srcSet = iter->second;
564 : // generate route list
565 6 : for (auto srcIter = srcSet.begin(); srcIter != srcSet.end(); ++srcIter) {
566 3 : routeList.emplace_back(std::make_pair(&(*srcIter), &dst));
567 : }
568 : }
569 10 : }
570 :
571 4 : BqsStatus ConfigInfoOperator::QueryRoutesBySrcAndDst(const uintptr_t mbufData, const EntityInfo &src,
572 : const EntityInfo &dst, const bool onlyQryNum) const
573 : {
574 4 : uint32_t searchedRouteNum = 0U;
575 4 : auto &srcToDstRelation = BindRelation::GetInstance().GetSrcToDstRelation();
576 4 : const auto iter = srcToDstRelation.find(src);
577 4 : if ((iter != srcToDstRelation.end()) && (iter->second.count(dst) != 0UL)) {
578 3 : searchedRouteNum = 1U;
579 : } else {
580 1 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
581 1 : const auto &srcToDstRelationTmp = BindRelation::GetInstance().GetSrcToDstExtraRelation();
582 1 : const auto it = srcToDstRelationTmp.find(src);
583 1 : if ((it != srcToDstRelationTmp.end()) && (it->second.count(dst) != 0UL)) {
584 0 : searchedRouteNum = 1U;
585 : }
586 : }
587 : }
588 :
589 : // generate route list
590 4 : std::list<std::pair<const EntityInfo *, const EntityInfo *>> routeList;
591 4 : if (searchedRouteNum != 0U) {
592 3 : routeList.emplace_back(std::make_pair(&src, &dst));
593 : } else {
594 1 : BQS_LOG_WARN("Record does not exist according to src Id:[%u] type:[%d]", src.GetId(),
595 : static_cast<int32_t>(src.GetType()));
596 : }
597 8 : return SaveQueryResult(routeList, mbufData, onlyQryNum);
598 4 : }
599 :
600 8 : BqsStatus ConfigInfoOperator::QueryAllRoutes(const uintptr_t mbufData, const bool onlyQryNum) const
601 : {
602 : // gennerate route list
603 8 : std::list<std::pair<const EntityInfo *, const EntityInfo *>> routeList;
604 8 : auto &srcToDstRelation = BindRelation::GetInstance().GetSrcToDstRelation();
605 23 : for (auto iter = srcToDstRelation.begin(); iter != srcToDstRelation.end(); ++iter) {
606 15 : const auto &dstSet = iter->second;
607 33 : for (auto &dst : dstSet) {
608 18 : routeList.emplace_back(std::make_pair(&(iter->first), &dst));
609 : }
610 : }
611 :
612 8 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
613 4 : const auto &srcToDstRelationTmp = BindRelation::GetInstance().GetSrcToDstExtraRelation();
614 4 : for (auto iter = srcToDstRelationTmp.begin(); iter != srcToDstRelationTmp.end(); ++iter) {
615 0 : const auto &dstSet = iter->second;
616 0 : for (auto &dst : dstSet) {
617 0 : routeList.emplace_back(std::make_pair(&(iter->first), &dst));
618 : }
619 : }
620 : }
621 :
622 16 : return SaveQueryResult(routeList, mbufData, onlyQryNum);
623 8 : }
624 :
625 22 : BqsStatus ConfigInfoOperator::SaveQueryResult(std::list<std::pair<const EntityInfo *, const EntityInfo *>> &routeList,
626 : const uintptr_t mbufData, const bool onlyQryNum) const
627 : {
628 : // no need check cfgQry nullptr
629 22 : ConfigQuery * const cfgQry = PtrToPtr<void, ConfigQuery>(ValueToPtr(mbufData));
630 22 : const size_t totalRouteNum = routeList.size();
631 22 : if (onlyQryNum) {
632 15 : cfgQry->qry.routeQry.routeNum = static_cast<uint32_t>(totalRouteNum);
633 15 : CfgRetInfo * const retInfo = PtrToPtr<void, CfgRetInfo>(ValueToPtr(mbufData + sizeof(ConfigQuery)));
634 15 : retInfo->retCode = static_cast<int32_t>(BQS_STATUS_OK);
635 15 : return BQS_STATUS_OK;
636 : }
637 :
638 : // check number
639 7 : const size_t routeNum = static_cast<size_t>(cfgQry->qry.routeQry.routeNum);
640 7 : const uintptr_t results = mbufData + sizeof(ConfigQuery) + sizeof(ConfigInfo) + (routeNum * sizeof(Route));
641 7 : CfgRetInfo * const retInfo = PtrToPtr<void, CfgRetInfo>(ValueToPtr(results));
642 7 : if (routeNum != totalRouteNum) {
643 1 : retInfo->retCode = static_cast<int32_t>(BQS_STATUS_PARAM_INVALID);
644 1 : BQS_LOG_ERROR("Route num in query info is [%lu], but searched route num is [%lu].", routeNum, totalRouteNum);
645 1 : return BQS_STATUS_PARAM_INVALID;
646 : }
647 :
648 : // convert and record route
649 6 : Route * const routes = PtrToPtr<void, Route>(ValueToPtr(mbufData + sizeof(ConfigQuery) + sizeof(ConfigInfo)));
650 6 : size_t idx = 0UL;
651 17 : for (auto iter = routeList.begin(); iter != routeList.end(); ++iter) {
652 11 : Route * const route = PtrAdd<Route>(routes, routeNum, idx);
653 11 : (void)ConvertToRoute(*(iter->first), *(iter->second), *route);
654 11 : idx++;
655 : }
656 6 : retInfo->retCode = static_cast<int32_t>(BQS_STATUS_OK);
657 6 : return BQS_STATUS_OK;
658 : }
659 :
660 11 : BqsStatus ConfigInfoOperator::ConvertToRoute(const EntityInfo &src, const EntityInfo &dst, Route &route) const
661 : {
662 11 : route.status = RouteStatus::ACTIVE;
663 11 : (void)ConvertToEndpoint(src, route.src);
664 11 : (void)ConvertToEndpoint(dst, route.dst);
665 11 : return BQS_STATUS_OK;
666 : }
667 :
668 32 : BqsStatus ConfigInfoOperator::ConvertToEndpoint(const EntityInfo &entity, Endpoint &endpoint) const
669 : {
670 32 : auto ret = BQS_STATUS_OK;
671 32 : endpoint.status = EndpointStatus::AVAILABLE;
672 32 : switch (entity.GetType()) {
673 25 : case dgw::EntityType::ENTITY_QUEUE: {
674 25 : if (endpoint.type == EndpointType::MEM_QUEUE) {
675 2 : endpoint.attr.memQueueAttr.queueId = static_cast<int32_t>(entity.GetId());
676 : } else {
677 23 : endpoint.type = EndpointType::QUEUE;
678 23 : endpoint.attr.queueAttr.queueId = static_cast<int32_t>(entity.GetId());
679 : }
680 25 : break;
681 : }
682 3 : case dgw::EntityType::ENTITY_GROUP: {
683 3 : endpoint.type = EndpointType::GROUP;
684 3 : endpoint.attr.groupAttr.groupId = static_cast<int32_t>(entity.GetId());
685 : // group policy && endpoint num
686 3 : break;
687 : }
688 3 : case dgw::EntityType::ENTITY_TAG: {
689 3 : endpoint.type = EndpointType::COMM_CHANNEL;
690 3 : CommChannelAttr &attr = endpoint.attr.channelAttr;
691 3 : const dgw::CommChannel * const channel = entity.GetCommChannel();
692 3 : if (channel != nullptr) {
693 3 : attr.handle = PtrToValue(channel->GetHandle());
694 3 : attr.localTagId = channel->GetLocalTagId();
695 3 : attr.peerTagId = channel->GetPeerTagId();
696 3 : attr.localRankId = channel->GetLocalRankId();
697 3 : attr.peerRankId = channel->GetPeerRankId();
698 3 : attr.localTagDepth = channel->GetLocalTagDepth();
699 3 : attr.peerTagDepth = channel->GetPeerTagDepth();
700 : }
701 3 : break;
702 : }
703 1 : default: {
704 1 : BQS_LOG_ERROR("Unsupport entity type[%d].", static_cast<int32_t>(entity.GetType()));
705 1 : ret = BQS_STATUS_PARAM_INVALID;
706 1 : break;
707 : }
708 : }
709 32 : return ret;
710 : }
711 :
712 95 : EntityInfoPtr ConfigInfoOperator::CreateEntityInfo(const Endpoint &endpoint, const bool isQry) const
713 : {
714 95 : uint32_t id = 0U;
715 95 : uint32_t localDeviceId = deviceId_;
716 95 : OptionalArg args = {};
717 95 : args.eType = dgw::EntityType::ENTITY_INVALID;
718 95 : args.schedCfgKey = endpoint.rootModelId;
719 95 : args.globalId = endpoint.globalId;
720 95 : args.uuId = endpoint.modelId;
721 :
722 95 : dgw::EntityType &eType = args.eType;
723 95 : bqs::GroupPolicy &policy = args.policy;
724 95 : const dgw::CommChannel* &channelPtr = args.channelPtr;
725 95 : auto ret = BQS_STATUS_OK;
726 95 : switch (endpoint.type) {
727 76 : case EndpointType::QUEUE: {
728 76 : eType = dgw::EntityType::ENTITY_QUEUE;
729 76 : id = static_cast<uint32_t>(endpoint.attr.queueAttr.queueId);
730 76 : break;
731 : }
732 3 : case EndpointType::MEM_QUEUE: {
733 3 : eType = dgw::EntityType::ENTITY_QUEUE;
734 3 : id = static_cast<uint32_t>(endpoint.attr.memQueueAttr.queueId);
735 3 : break;
736 : }
737 8 : case EndpointType::COMM_CHANNEL: {
738 8 : eType = dgw::EntityType::ENTITY_TAG;
739 8 : const CommChannelAttr &attr = endpoint.attr.channelAttr;
740 8 : ret = CheckCommChannelAttr(attr, isQry);
741 8 : if (ret == BQS_STATUS_OK) {
742 5 : const dgw::CommChannel channel(ValueToPtr(attr.handle), attr.localTagId, attr.peerTagId,
743 5 : attr.localRankId, attr.peerRankId, attr.localTagDepth,
744 5 : attr.peerTagDepth);
745 5 : id = dgw::CommChannelManager::GetInstance().GetCommChannelId(channel, channelPtr);
746 5 : }
747 8 : break;
748 : }
749 7 : case EndpointType::GROUP: {
750 7 : eType = dgw::EntityType::ENTITY_GROUP;
751 7 : id = static_cast<uint32_t>(endpoint.attr.groupAttr.groupId);
752 7 : policy = endpoint.attr.groupAttr.policy;
753 7 : break;
754 : }
755 1 : default: {
756 1 : ret = BQS_STATUS_PARAM_INVALID;
757 1 : BQS_LOG_DEBUG("Unsupport endpoint type[%d].", static_cast<int32_t>(endpoint.type));
758 1 : break;
759 : }
760 : }
761 95 : if (ret != BQS_STATUS_OK) {
762 4 : return nullptr;
763 : }
764 :
765 91 : if (clientVersion_ >= 2U) {
766 0 : args.peerInstanceNum = endpoint.peerNum;
767 0 : args.localInstanceIndex = endpoint.localId;
768 : }
769 :
770 91 : if (GlobalCfg::GetInstance().GetNumaFlag() && ((endpoint.resId & RESOURCE_ID_ENABLE_BIT_MASK) != 0U)) {
771 4 : localDeviceId = (endpoint.resId & ROUCE_ID_DEVICE_ID_DATA_MASK);
772 : }
773 :
774 91 : uint32_t &queueType = args.queueType;
775 91 : if (endpoint.type == EndpointType::MEM_QUEUE) {
776 : // parse hostQ and device belonged
777 3 : bool isHostQueue = (((endpoint.resId >> RESOURCE_ID_HOST_DEVICE_BIT_NUM) & 1) != 0) ? true : false;
778 3 : uint32_t onwerDeviceId = ((endpoint.resId & RESOURCE_ID_ENABLE_BIT_MASK) != 0U) ?
779 2 : (endpoint.resId & ROUCE_ID_DEVICE_ID_DATA_MASK) : deviceId_;
780 3 : localDeviceId = onwerDeviceId;
781 3 : queueType = endpoint.attr.memQueueAttr.queueType;
782 3 : if ((bqs::GetRunContext() != bqs::RunContext::HOST) && (&drvGetLocalDevIDByHostDevID != nullptr)) {
783 1 : auto retCode = drvGetLocalDevIDByHostDevID(onwerDeviceId, &localDeviceId);
784 1 : if (retCode != static_cast<int32_t>(DRV_ERROR_NONE)) {
785 1 : BQS_LOG_INFO("host devid(%u) transform to local devid.", localDeviceId);
786 1 : localDeviceId = onwerDeviceId;
787 : }
788 : }
789 3 : BQS_LOG_INFO("[CreateEntityInfo] qid=%u, endpoint.resId=%u, isHostQueue=%d, "
790 : "onwerDeviceId=%u, localDeviceId=%u, queueType=%u",
791 : id, endpoint.resId, isHostQueue, onwerDeviceId, localDeviceId, queueType);
792 : }
793 :
794 : // create entity info ptr
795 91 : EntityInfoPtr entityPtr = nullptr;
796 : try {
797 91 : entityPtr = std::make_shared<EntityInfo>(id, localDeviceId, &args);
798 0 : } catch (...) {
799 0 : BQS_LOG_ERROR("Create entity info ptr failed, id[%u], type[%d].",
800 : id, static_cast<int32_t>(eType));
801 0 : }
802 :
803 91 : BQS_LOG_INFO("Create entity success: %s", entityPtr->ToString().c_str());
804 91 : return entityPtr;
805 : }
806 :
807 15 : BqsStatus ConfigInfoOperator::AttachAndCheckQueue(const EntityInfo& src, const EntityInfo& dst) const
808 : {
809 15 : auto srcRet = AttachQueue(src);
810 15 : auto dstRet = AttachQueue(dst);
811 15 : auto ret = (srcRet != BQS_STATUS_OK) ? srcRet : dstRet;
812 15 : if (ret != BQS_STATUS_OK) {
813 1 : BQS_LOG_ERROR("Fail to attach src[%s] or dst[%s], srcRet[%d], dstRet[%d].",
814 : src.ToString().c_str(), dst.ToString().c_str(), static_cast<int32_t>(srcRet), static_cast<int32_t>(dstRet));
815 1 : return ret;
816 : }
817 :
818 14 : srcRet = CheckQueueAuth(src, true);
819 14 : dstRet = CheckQueueAuth(dst, false);
820 14 : ret = (srcRet != BQS_STATUS_OK) ? srcRet : dstRet;
821 14 : if (ret != BQS_STATUS_OK) {
822 0 : BQS_LOG_ERROR("Fail to check src[%s] or dst[%s] queue auth, srcRet[%d], dstRet[%d].",
823 : src.ToString().c_str(), dst.ToString().c_str(), static_cast<int32_t>(srcRet), static_cast<int32_t>(dstRet));
824 0 : return ret;
825 : }
826 14 : return BQS_STATUS_OK;
827 : }
828 :
829 32 : BqsStatus ConfigInfoOperator::AttachQueue(const EntityInfo &info) const
830 : {
831 32 : auto ret = BQS_STATUS_OK;
832 32 : switch (info.GetType()) {
833 2 : case dgw::EntityType::ENTITY_TAG: {
834 2 : break;
835 : }
836 25 : case dgw::EntityType::ENTITY_QUEUE: {
837 25 : const auto drvRet = halQueueAttach(info.GetDeviceId(), info.GetId(), 0);
838 25 : if (drvRet != DRV_ERROR_NONE) {
839 3 : BQS_LOG_ERROR("Fail to attach queue[%s], device[%u], ret[%d]", info.ToString().c_str(),
840 : info.GetDeviceId(), static_cast<int32_t>(drvRet));
841 3 : ret = BQS_STATUS_DRIVER_ERROR;
842 : }
843 25 : break;
844 : }
845 4 : case dgw::EntityType::ENTITY_GROUP: {
846 4 : ret = AttachQueueInGroup(info.GetId());
847 4 : break;
848 : }
849 1 : default: {
850 1 : BQS_LOG_ERROR("Invalid entity[%s]", info.ToString().c_str());
851 1 : ret = BQS_STATUS_PARAM_INVALID;
852 1 : break;
853 : }
854 : }
855 32 : return ret;
856 : }
857 :
858 5 : BqsStatus ConfigInfoOperator::AttachQueueInGroup(const uint32_t groupId) const
859 : {
860 5 : auto &entitiesInGroup = BindRelation::GetInstance().GetEntitiesInGroup(groupId);
861 5 : if (entitiesInGroup.empty()) {
862 1 : BQS_LOG_ERROR("Group[%u] does not exist.", groupId);
863 1 : return BQS_STATUS_GROUP_NOT_EXIST;
864 : }
865 13 : for (const auto &info : entitiesInGroup) {
866 9 : if (info == nullptr) {
867 0 : BQS_LOG_ERROR("EntityInfo in Group[%u] is nullptr.", groupId);
868 0 : return BQS_STATUS_INNER_ERROR;
869 : }
870 : // endpoints in group is the same type
871 9 : if (info->GetType() == dgw::EntityType::ENTITY_QUEUE) {
872 5 : const auto drvRet = halQueueAttach(info->GetDeviceId(), info->GetId(), 0);
873 5 : if (drvRet != DRV_ERROR_NONE) {
874 0 : BQS_LOG_ERROR("Fail to attach queue[%s] in group[%u], ret[%d]",
875 : info->ToString().c_str(), groupId, static_cast<int32_t>(drvRet));
876 0 : return BQS_STATUS_DRIVER_ERROR;
877 : }
878 : }
879 : }
880 4 : return BQS_STATUS_OK;
881 : }
882 :
883 40 : BqsStatus ConfigInfoOperator::PreprocessUpdateCfgInfo(const uintptr_t mbufData, const uint64_t dataLen)
884 : {
885 : // check and record update config info
886 40 : auto ret = CheckAndRecordUpdateCfgInfo(mbufData, dataLen);
887 40 : if (ret != BQS_STATUS_OK) {
888 3 : BQS_LOG_ERROR("Record update config info failed.");
889 3 : return ret;
890 : }
891 :
892 : // attach queue and check queue auth
893 37 : ret = CheckFlowQueueAuth();
894 37 : if (ret != BQS_STATUS_OK) {
895 0 : BQS_LOG_ERROR("Check flow queue auth failed.");
896 0 : return ret;
897 : }
898 37 : return BQS_STATUS_WAIT;
899 : }
900 :
901 37 : BqsStatus ConfigInfoOperator::CheckFlowQueueAuth() const
902 : {
903 37 : ConfigInfo * const cfgInfo = updateCfgInfo_->cfgInfo;
904 : // create group no need attach queue and check queue auth
905 37 : if (cfgInfo->cmd == ConfigCmd::DGW_CFG_CMD_BIND_ROUTE) {
906 7 : size_t idx = 0UL;
907 21 : for (auto &entityPair : updateCfgInfo_->entitiesInRoutes) {
908 : // do attach queue and check src own read auth, dst own write auth
909 14 : const auto ret = AttachAndCheckQueue(*(entityPair.first), *(entityPair.second));
910 14 : if (ret != BQS_STATUS_OK) {
911 0 : BQS_LOG_ERROR("Src[%s] Dst[%s] do attach queue and check auth failed",
912 : entityPair.first->ToString().c_str(), entityPair.second->ToString().c_str());
913 : }
914 14 : updateCfgInfo_->results[idx]->retCode = static_cast<int32_t>(ret);
915 14 : idx++;
916 : }
917 : }
918 37 : return BQS_STATUS_OK;
919 : }
920 :
921 29 : BqsStatus ConfigInfoOperator::CheckQueueAuth(const EntityInfo &info, const bool isSrc) const
922 : {
923 29 : if (info.GetType() == dgw::EntityType::ENTITY_QUEUE) {
924 23 : if (info.GetQueueType() == bqs::CLIENT_Q) {
925 1 : return BQS_STATUS_OK;
926 : }
927 22 : return CheckQueueAuth(info.GetId(), info.GetDeviceId(), isSrc);
928 : }
929 6 : if (info.GetType() == dgw::EntityType::ENTITY_GROUP) {
930 4 : return CheckQueueAuthForGroup(info.GetId(), isSrc);
931 : }
932 : // else, ENTITY_TAG
933 2 : return BQS_STATUS_OK;
934 : }
935 :
936 5 : BqsStatus ConfigInfoOperator::CheckQueueAuthForGroup(const uint32_t groupId, const bool isSrc) const
937 : {
938 5 : auto &entityVec = BindRelation::GetInstance().GetEntitiesInGroup(groupId);
939 14 : for (const auto &entityInfoPtr : entityVec) {
940 10 : if (entityInfoPtr == nullptr) {
941 0 : BQS_LOG_ERROR("EntityInfo in Group[%u] is nullptr.", groupId);
942 1 : return BQS_STATUS_INNER_ERROR;
943 : }
944 10 : if (entityInfoPtr->GetType() == dgw::EntityType::ENTITY_QUEUE) {
945 6 : if (entityInfoPtr->GetQueueType() == bqs::CLIENT_Q) {
946 1 : return BQS_STATUS_OK;
947 : }
948 5 : const auto ret = CheckQueueAuth(entityInfoPtr->GetId(), entityInfoPtr->GetDeviceId(), isSrc);
949 5 : if (ret != BQS_STATUS_OK) {
950 0 : return ret;
951 : }
952 : }
953 : }
954 4 : return BQS_STATUS_OK;
955 : }
956 :
957 27 : BqsStatus ConfigInfoOperator::CheckQueueAuth(const uint32_t queueId, const uint32_t resId, const bool isSrc) const
958 : {
959 27 : std::unique_ptr<QueueQueryOutput> output(new (std::nothrow) QueueQueryOutput());
960 27 : if (output == nullptr) {
961 0 : BQS_LOG_ERROR("Malloc memory for output failed.");
962 0 : return BQS_STATUS_INNER_ERROR;
963 : }
964 27 : QueueQueryOutputPara outputPara = {output.get(), static_cast<uint32_t>(sizeof(QueueQueryOutput))};
965 27 : QueQueryQueueAttr queAttr = {static_cast<int32_t>(queueId)};
966 27 : QueueQueryInputPara inputPara = {&queAttr, static_cast<uint32_t>(sizeof(queAttr))};
967 27 : const auto drvRet = halQueueQuery(resId, QUEUE_QUERY_QUE_ATTR_OF_CUR_PROC, &inputPara, &outputPara);
968 27 : if (drvRet != DRV_ERROR_NONE) {
969 0 : BQS_LOG_ERROR("Fail to query queue info, queue[%u], resId[%u], ret[%d]",
970 : queueId, resId, static_cast<int32_t>(drvRet));
971 0 : return BQS_STATUS_DRIVER_ERROR;
972 : }
973 :
974 27 : const uint32_t authValue = isSrc ? static_cast<uint32_t>(output.get()->queQueryQueueAttrInfo.attr.read) :
975 13 : static_cast<uint32_t>(output.get()->queQueryQueueAttrInfo.attr.write);
976 27 : if (authValue == 0U) {
977 0 : BQS_LOG_ERROR("Queue[%u] res[%u] did not own needed authority, isSrc[%d].",
978 : queueId, resId, static_cast<int32_t>(isSrc));
979 0 : return BQS_STATUS_QUEUE_AHTU_ERROR;
980 : }
981 27 : BQS_LOG_INFO("Queue[%u] res[%u] check authority success, isSrc[%d]",
982 : queueId, resId, static_cast<int32_t>(isSrc));
983 27 : return BQS_STATUS_OK;
984 27 : }
985 :
986 41 : BqsStatus ConfigInfoOperator::CheckAndRecordUpdateCfgInfo(const uintptr_t mbufData, const uint64_t dataLen)
987 : {
988 : // check min dataLen
989 41 : if (dataLen < sizeof(ConfigInfo)) {
990 1 : BQS_LOG_ERROR("dataLen[%lu] is invalid.", dataLen);
991 1 : return BQS_STATUS_PARAM_INVALID;
992 : }
993 : // check cfgInfo
994 40 : ConfigInfo * const cfgInfo = PtrToPtr<void, ConfigInfo>(ValueToPtr(mbufData));
995 40 : if (cfgInfo == nullptr) {
996 0 : BQS_LOG_ERROR("cfgInfo is nullptr.");
997 0 : return BQS_STATUS_PARAM_INVALID;
998 : }
999 :
1000 40 : updateCfgInfo_.reset(new (std::nothrow) UpdateCfgInfo());
1001 40 : if (updateCfgInfo_ == nullptr) {
1002 0 : BQS_LOG_ERROR("Malloc memory for updateCfgInfo_ failed.");
1003 0 : return BQS_STATUS_INNER_ERROR;
1004 : }
1005 : // record mbuf data
1006 40 : updateCfgInfo_->mbufData = mbufData;
1007 40 : updateCfgInfo_->dataLen = dataLen;
1008 : // record cfgInfo
1009 40 : updateCfgInfo_->cfgInfo = cfgInfo;
1010 :
1011 40 : BQS_LOG_INFO("cmd is %d", static_cast<int32_t>(cfgInfo->cmd));
1012 40 : auto ret = BQS_STATUS_OK;
1013 40 : switch (cfgInfo->cmd) {
1014 13 : case ConfigCmd::DGW_CFG_CMD_BIND_ROUTE:
1015 : case ConfigCmd::DGW_CFG_CMD_UNBIND_ROUTE: {
1016 13 : ret = CheckAndRecordRouteInfo();
1017 13 : break;
1018 : }
1019 10 : case ConfigCmd::DGW_CFG_CMD_ADD_GROUP: {
1020 10 : ret = CheckAndRecordAddGrpInfo();
1021 10 : break;
1022 : }
1023 11 : case ConfigCmd::DGW_CFG_CMD_UPDATE_PROFILING:
1024 : case ConfigCmd::DGW_CFG_CMD_DEL_GROUP:
1025 : case ConfigCmd::DGW_CFG_CMD_SET_HCCL_PROTOCOL: {
1026 11 : ret = CheckAndRecordCfgInfo();
1027 11 : break;
1028 : }
1029 4 : case ConfigCmd::DGW_CFG_CMD_INIT_DYNAMIC_SCHEDULE: {
1030 4 : ret = CheckAndRecordCommonCfg(sizeof(ConfigInfo) + sizeof(DynamicSchedConfigV2));
1031 4 : break;
1032 : }
1033 2 : case ConfigCmd::DGW_CFG_CMD_STOP_SCHEDULE:
1034 : case ConfigCmd::DGW_CFG_CMD_CLEAR_AND_RESTART_SCHEDULE: {
1035 2 : ret = CheckAndRecordRedeployCfg();
1036 2 : break;
1037 : }
1038 0 : default: {
1039 0 : ret = BQS_STATUS_PARAM_INVALID;
1040 0 : BQS_LOG_WARN("cmd[%d] is invalid.", static_cast<int32_t>(cfgInfo->cmd));
1041 0 : break;
1042 : }
1043 : }
1044 :
1045 : // check update config info failed, clear updateCfgInfo_
1046 40 : if (ret != BQS_STATUS_OK) {
1047 3 : updateCfgInfo_ = nullptr;
1048 : }
1049 40 : return ret;
1050 : }
1051 :
1052 13 : BqsStatus ConfigInfoOperator::CheckAndRecordRouteInfo() const
1053 : {
1054 13 : const ConfigInfo * const cfgInfo = updateCfgInfo_->cfgInfo;
1055 13 : const size_t routeNum = static_cast<size_t>(cfgInfo->cfg.routesCfg.routeNum);
1056 : // check route num
1057 13 : if ((routeNum == 0UL) || (routeNum > MAX_ROUTES_NUM)) {
1058 0 : BQS_LOG_ERROR("Route num[%zu] is invalid, max allowed value is [%zu].", routeNum, MAX_ROUTES_NUM);
1059 0 : return BQS_STATUS_PARAM_INVALID;
1060 : }
1061 : // calculate and check totalLen
1062 13 : const size_t totalLen = sizeof(ConfigInfo) + (routeNum * sizeof(Route)) + (routeNum * sizeof(CfgRetInfo));
1063 13 : const uint64_t dataLen = updateCfgInfo_->dataLen;
1064 13 : if (totalLen != dataLen) {
1065 0 : BQS_LOG_ERROR("dataLen[%lu] is not equal with totalLen[%zu].", dataLen, totalLen);
1066 0 : return BQS_STATUS_PARAM_INVALID;
1067 : }
1068 :
1069 13 : auto &routeVec = updateCfgInfo_->routes;
1070 13 : auto &resultVec = updateCfgInfo_->results;
1071 13 : auto &entityPairVec = updateCfgInfo_->entitiesInRoutes;
1072 13 : const uintptr_t routesAddr = updateCfgInfo_->mbufData + sizeof(ConfigInfo);
1073 13 : Route * const routes = PtrToPtr<void, Route>(ValueToPtr(routesAddr));
1074 13 : CfgRetInfo * const results = PtrToPtr<void, CfgRetInfo>(ValueToPtr(routesAddr + (routeNum * sizeof(Route))));
1075 40 : for (size_t idx = 0UL; idx < routeNum; idx++) {
1076 27 : Route * const route = PtrAdd<Route>(routes, routeNum, idx);
1077 27 : CfgRetInfo * const result = PtrAdd<CfgRetInfo>(results, routeNum, idx);
1078 : // initialize retCode
1079 27 : result->retCode = static_cast<int32_t>(BQS_STATUS_OK);
1080 27 : routeVec.emplace_back(route);
1081 27 : resultVec.emplace_back(result);
1082 : // create src and dst entity info ptr
1083 27 : const EntityInfoPtr src = CreateEntityInfo(route->src, false);
1084 27 : const EntityInfoPtr dst = CreateEntityInfo(route->dst, false);
1085 27 : if ((src == nullptr) || (dst == nullptr)) {
1086 0 : BQS_LOG_ERROR("Create src or dst entityInfoPtr failed.");
1087 0 : return BQS_STATUS_INNER_ERROR;
1088 : }
1089 27 : entityPairVec.emplace_back(std::make_pair(src, dst));
1090 27 : }
1091 13 : return BQS_STATUS_OK;
1092 : }
1093 :
1094 12 : BqsStatus ConfigInfoOperator::CheckAndRecordAddGrpInfo() const
1095 : {
1096 12 : const ConfigInfo * const cfgInfo = updateCfgInfo_->cfgInfo;
1097 12 : const size_t endpointNum = static_cast<size_t>(cfgInfo->cfg.groupCfg.endpointNum);
1098 : // check endpoint num
1099 12 : if ((endpointNum == 0UL) || (endpointNum > MAX_ENDPOINTS_NUM_IN_SINGLE_GROUP)) {
1100 1 : BQS_LOG_ERROR("Group num[%zu] is invalid, max allowed value is [%u].",
1101 : endpointNum, MAX_ENDPOINTS_NUM_IN_SINGLE_GROUP);
1102 1 : return BQS_STATUS_PARAM_INVALID;
1103 : }
1104 :
1105 : // calculate and check totalLen
1106 11 : const size_t totalLen = sizeof(ConfigInfo) + (endpointNum * sizeof(Endpoint)) + sizeof(CfgRetInfo);
1107 11 : const uint64_t dataLen = updateCfgInfo_->dataLen;
1108 11 : if (totalLen != dataLen) {
1109 1 : BQS_LOG_ERROR("dataLen[%lu] is not equal with totalLen[%zu].", dataLen, totalLen);
1110 1 : return BQS_STATUS_PARAM_INVALID;
1111 : }
1112 :
1113 10 : auto &endpointVec = updateCfgInfo_->endpointsInGroup;
1114 10 : auto &resultVec = updateCfgInfo_->results;
1115 10 : auto &entityVec = updateCfgInfo_->entitiesInGroup;
1116 10 : const uintptr_t endpointsAddr = updateCfgInfo_->mbufData + sizeof(ConfigInfo);
1117 10 : Endpoint * const endpoints = PtrToPtr<void, Endpoint>(ValueToPtr(endpointsAddr));
1118 10 : const uintptr_t results = endpointsAddr + (endpointNum * sizeof(Endpoint));
1119 : // only one result
1120 10 : CfgRetInfo * const result = PtrToPtr<void, CfgRetInfo>(ValueToPtr(results));
1121 10 : resultVec.emplace_back(result);
1122 :
1123 10 : std::set<std::tuple<const uint32_t, const bool, const uint32_t, const dgw::EntityType>> entitySet;
1124 27 : for (size_t idx = 0UL; idx < endpointNum; idx++) {
1125 20 : Endpoint * const endpoint = PtrAdd<Endpoint>(endpoints, endpointNum, idx);
1126 20 : endpointVec.emplace_back(endpoint);
1127 : // create entity info ptr
1128 20 : EntityInfoPtr entity = CreateEntityInfo(*endpoint, false);
1129 20 : if (entity == nullptr) {
1130 3 : BQS_LOG_ERROR("Create entityInfoPtr failed.");
1131 3 : return BQS_STATUS_PARAM_INVALID;
1132 : }
1133 17 : if (entity->GetType() == dgw::EntityType::ENTITY_GROUP) {
1134 0 : BQS_LOG_ERROR("Not allowd group[%s] exist in group.", entity->ToString().c_str());
1135 0 : return BQS_STATUS_PARAM_INVALID;
1136 : }
1137 17 : entityVec.emplace_back(entity);
1138 17 : uint32_t deviceId = ((endpoint->resId & RESOURCE_ID_ENABLE_BIT_MASK) != 0U) ?
1139 10 : (endpoint->resId & ROUCE_ID_DEVICE_ID_DATA_MASK) : deviceId_;
1140 17 : bool isHostQueue = (((endpoint->resId >> RESOURCE_ID_HOST_DEVICE_BIT_NUM) & 1) != 0U) ? true : false;
1141 17 : (void)entitySet.emplace(std::make_tuple(deviceId, isHostQueue, entity->GetId(), entity->GetType()));
1142 20 : }
1143 :
1144 : // check whether group has the same entity
1145 7 : if (entitySet.size() != endpointNum) {
1146 0 : BQS_LOG_ERROR("entitySet size[%lu] is not equal with endpointNum[%lu].", entitySet.size(), endpointNum);
1147 0 : return BQS_STATUS_PARAM_INVALID;
1148 : }
1149 7 : return BQS_STATUS_OK;
1150 10 : }
1151 :
1152 11 : BqsStatus ConfigInfoOperator::CheckAndRecordCfgInfo() const
1153 : {
1154 11 : return CheckAndRecordCommonCfg(sizeof(ConfigInfo));
1155 : }
1156 :
1157 18 : BqsStatus ConfigInfoOperator::CheckAndRecordCommonCfg(const size_t resultOffset) const
1158 : {
1159 : // calculate and check totalLen
1160 18 : const size_t totalLen = resultOffset + sizeof(CfgRetInfo);
1161 18 : const uint64_t dataLen = updateCfgInfo_->dataLen;
1162 18 : if (totalLen != dataLen) {
1163 1 : BQS_LOG_ERROR("dataLen[%lu] is not equal with totalLen[%zu].", dataLen, totalLen);
1164 1 : return BQS_STATUS_PARAM_INVALID;
1165 : }
1166 :
1167 17 : auto &resultVec = updateCfgInfo_->results;
1168 : // only one result
1169 17 : const uintptr_t results = updateCfgInfo_->mbufData + resultOffset;
1170 17 : CfgRetInfo * const result = PtrToPtr<void, CfgRetInfo>(ValueToPtr(results));
1171 17 : resultVec.emplace_back(result);
1172 17 : BQS_LOG_INFO("CheckAndRecordCommonCfg for %zu", resultOffset);
1173 17 : return BQS_STATUS_OK;
1174 : }
1175 :
1176 2 : BqsStatus ConfigInfoOperator::CheckAndRecordRedeployCfg() const
1177 : {
1178 2 : const ConfigInfo * const cfgInfo = updateCfgInfo_->cfgInfo;
1179 2 : const size_t rootModelIdsLen = cfgInfo->cfg.reDeployCfg.rootModelNum * sizeof(uint32_t);
1180 2 : return CheckAndRecordCommonCfg(rootModelIdsLen + sizeof(ConfigInfo));
1181 : }
1182 :
1183 13 : BqsStatus ConfigInfoOperator::ProcessUpdateRoutes(const uint32_t index) const
1184 : {
1185 : // no need to check cfgInfo and updateCfgInfo_ nullptr
1186 13 : ConfigInfo * const cfgInfo = updateCfgInfo_->cfgInfo;
1187 13 : auto &resultVec = updateCfgInfo_->results;
1188 13 : auto &entityPairVec = updateCfgInfo_->entitiesInRoutes;
1189 :
1190 13 : auto returnCode = BQS_STATUS_OK;
1191 13 : size_t idx = 0UL;
1192 40 : for (auto &entityPair : entityPairVec) {
1193 : // check preprocess result
1194 27 : CfgRetInfo * const retInfo = resultVec[idx];
1195 27 : idx++;
1196 27 : const auto preRet = static_cast<BqsStatus>(retInfo->retCode);
1197 27 : if (preRet != BQS_STATUS_OK) {
1198 0 : returnCode = preRet;
1199 0 : continue;
1200 : }
1201 : // create entity info
1202 27 : const auto ret = (cfgInfo->cmd == ConfigCmd::DGW_CFG_CMD_BIND_ROUTE) ?
1203 14 : BindRelation::GetInstance().Bind(*(entityPair.first), *(entityPair.second), index) :
1204 13 : BindRelation::GetInstance().UnBind(*(entityPair.first), *(entityPair.second), index);
1205 27 : if (ret == BQS_STATUS_RETRY) {
1206 0 : continue;
1207 : }
1208 :
1209 27 : retInfo->retCode = static_cast<int32_t>(ret);
1210 27 : returnCode = (returnCode == BQS_STATUS_OK) ? ret : returnCode;
1211 27 : BQS_LOG_RUN_INFO("Bind/unbind relation operate, cmd[%d], stage[server:process],"
1212 : "relation[src:%s, dst:%s, result:%d]", static_cast<int32_t>(cfgInfo->cmd),
1213 : entityPair.first->ToString().c_str(), entityPair.second->ToString().c_str(), static_cast<int32_t>(ret));
1214 : }
1215 13 : BindRelation::GetInstance().Order(index);
1216 13 : return returnCode;
1217 : }
1218 :
1219 7 : BqsStatus ConfigInfoOperator::ProcessAddGroup() const
1220 : {
1221 : // no need to check cfgInfo and updateCfgInfo_ nullptr
1222 : // get endpoint number
1223 7 : auto cfgInfo = updateCfgInfo_->cfgInfo;
1224 7 : auto &resultVec = updateCfgInfo_->results;
1225 7 : auto &entityVec = updateCfgInfo_->entitiesInGroup;
1226 :
1227 : // create group
1228 7 : uint32_t groupId = 0U;
1229 7 : const auto retCode = BindRelation::GetInstance().CreateGroup(entityVec, groupId);
1230 7 : if (retCode == BQS_STATUS_OK) {
1231 7 : cfgInfo->cfg.groupCfg.groupId = static_cast<int32_t>(groupId);
1232 : }
1233 : // set result
1234 7 : resultVec[0UL]->retCode = static_cast<int32_t>(retCode);
1235 :
1236 7 : BQS_LOG_RUN_INFO("Add group operate, cmd[%d], stage[server:process], endpointNum[%zu], groupId[%u], result:[%d]",
1237 : static_cast<int32_t>(cfgInfo->cmd), entityVec.size(), groupId, static_cast<int32_t>(retCode));
1238 7 : return retCode;
1239 : }
1240 :
1241 8 : BqsStatus ConfigInfoOperator::ProcessDelGroup() const
1242 : {
1243 : // get endpoint number
1244 8 : const auto cfgInfo = updateCfgInfo_->cfgInfo;
1245 8 : auto &resultVec = updateCfgInfo_->results;
1246 8 : const uint32_t groupId = static_cast<uint32_t>(cfgInfo->cfg.groupCfg.groupId);
1247 :
1248 : // delete group
1249 8 : const auto retCode = BindRelation::GetInstance().DeleteGroup(groupId);
1250 : // set result
1251 8 : resultVec[0UL]->retCode = static_cast<int32_t>(retCode);
1252 :
1253 8 : BQS_LOG_RUN_INFO("Delete group operate, cmd[%d], stage[server:process], groupId[%u], result:[%d]",
1254 : static_cast<int32_t>(cfgInfo->cmd), groupId, static_cast<int32_t>(retCode));
1255 8 : return retCode;
1256 : }
1257 :
1258 1 : BqsStatus ConfigInfoOperator::ProcessUpdateProfiling() const
1259 : {
1260 : // get prof mode
1261 1 : const auto cfgInfo = updateCfgInfo_->cfgInfo;
1262 1 : auto &resultVec = updateCfgInfo_->results;
1263 1 : const ProfilingMode mode = cfgInfo->cfg.profCfg.profMode;
1264 : // set prof mode
1265 1 : auto retCode = ProfileManager::GetInstance(0U).UpdateProfilingMode(mode);
1266 1 : if ((retCode == BQS_STATUS_OK) && GlobalCfg::GetInstance().GetNumaFlag()) {
1267 0 : retCode = bqs::ProfileManager::GetInstance(1U).UpdateProfilingMode(mode);
1268 : }
1269 : // set result
1270 1 : resultVec[0UL]->retCode = static_cast<int32_t>(retCode);
1271 1 : BQS_LOG_RUN_INFO("Update profiling operate, cmd[%d], stage[server:process], profiling mode[%u], result:[%d]",
1272 : static_cast<int32_t>(cfgInfo->cmd), static_cast<uint32_t>(mode), static_cast<int32_t>(retCode));
1273 1 : return retCode;
1274 : }
1275 :
1276 2 : BqsStatus ConfigInfoOperator::ProcessUpdateHcclProtocol() const
1277 : {
1278 2 : const auto cfgInfo = updateCfgInfo_->cfgInfo;
1279 2 : auto &resultVec = updateCfgInfo_->results;
1280 2 : const HcclProtocolType protocol = cfgInfo->cfg.hcclProtocolCfg.protocol;
1281 :
1282 : // set protocol
1283 2 : auto retCode = BQS_STATUS_OK;
1284 2 : std::string strProtocol = "";
1285 2 : if (protocol == HcclProtocolType::RDMA) {
1286 1 : strProtocol = "RDMA";
1287 1 : } else if (protocol == HcclProtocolType::TCP) {
1288 1 : strProtocol = "TCP";
1289 : } else {
1290 0 : BQS_LOG_ERROR("Invalid protocol type[%d]", static_cast<int32_t>(protocol));
1291 0 : retCode = BQS_STATUS_PARAM_INVALID;
1292 : }
1293 :
1294 2 : if (!strProtocol.empty()) {
1295 2 : const auto ret = setenv("HCCL_NPU_NET_PROTOCOL", strProtocol.c_str(), 1);
1296 2 : if (ret != 0) {
1297 0 : BQS_LOG_ERROR("setenv HCCL_NPU_NET_PROTOCOL failed, ret[%d]", ret);
1298 0 : retCode = BQS_STATUS_INNER_ERROR;
1299 : }
1300 : }
1301 : // set result
1302 2 : resultVec[0UL]->retCode = static_cast<int32_t>(retCode);
1303 2 : BQS_LOG_RUN_INFO("Update hccl_protocol operate, cmd[%d], stage[server:process], protocol[%d], result:[%d]",
1304 : static_cast<int32_t>(cfgInfo->cmd), static_cast<int32_t>(protocol), static_cast<int32_t>(retCode));
1305 2 : return retCode;
1306 2 : }
1307 :
1308 4 : BqsStatus ConfigInfoOperator::ProcessInitDynamicSched() const
1309 : {
1310 4 : BQS_LOG_INFO("ProcessInitDynamicSched");
1311 4 : const auto cfgInfo = updateCfgInfo_->cfgInfo;
1312 4 : auto &resultVec = updateCfgInfo_->results;
1313 :
1314 4 : const uintptr_t dynamicSchedCfgAddr = updateCfgInfo_->mbufData + sizeof(ConfigInfo);
1315 : const DynamicSchedConfigV2 * const dynamicCfg =
1316 4 : PtrToPtr<void, DynamicSchedConfigV2>(ValueToPtr(dynamicSchedCfgAddr));
1317 :
1318 4 : BqsStatus retCode = BQS_STATUS_OK;
1319 4 : uint32_t localRequestQDeviceId = ParseDeviceId(dynamicCfg->requestQ.deviceId);
1320 4 : uint32_t localResponseDeviceId = ParseDeviceId(dynamicCfg->responseQ.deviceId);
1321 :
1322 4 : auto drvRet = halQueueAttach(localRequestQDeviceId, dynamicCfg->requestQ.queueId, 0);
1323 4 : if (drvRet != DRV_ERROR_NONE) {
1324 2 : BQS_LOG_ERROR("Fail to attach queue[%u], device[%u], ret[%d]", dynamicCfg->requestQ.queueId,
1325 : localRequestQDeviceId, static_cast<int32_t>(drvRet));
1326 2 : resultVec[0UL]->retCode = static_cast<int32_t>(BQS_STATUS_DRIVER_ERROR);
1327 2 : return BQS_STATUS_DRIVER_ERROR;
1328 : }
1329 2 : drvRet = halQueueAttach(localResponseDeviceId, dynamicCfg->responseQ.queueId, 0);
1330 2 : if (drvRet != DRV_ERROR_NONE) {
1331 1 : BQS_LOG_ERROR("Fail to attach queue[%u], device[%u], ret[%d]", dynamicCfg->responseQ.queueId,
1332 : localResponseDeviceId, static_cast<int32_t>(drvRet));
1333 1 : resultVec[0UL]->retCode = static_cast<int32_t>(BQS_STATUS_DRIVER_ERROR);
1334 1 : return BQS_STATUS_DRIVER_ERROR;
1335 : }
1336 :
1337 1 : dgw::DynamicSchedMgr::RootModelInfo schedCfgInfo = {};
1338 1 : schedCfgInfo.rootModelId = dynamicCfg->rootModelId;
1339 1 : schedCfgInfo.requestQue = dynamicCfg->requestQ;
1340 1 : schedCfgInfo.requestQue.deviceId = localRequestQDeviceId;
1341 1 : schedCfgInfo.responseQue = dynamicCfg->responseQ;
1342 1 : schedCfgInfo.responseQue.deviceId = localResponseDeviceId;
1343 :
1344 1 : uint32_t resIndex = 0U;
1345 1 : if (GlobalCfg::GetInstance().GetNumaFlag()) {
1346 1 : resIndex = GlobalCfg::GetInstance().GetResIndexByDeviceId(localRequestQDeviceId);
1347 : }
1348 1 : const auto addCfgRet = dgw::DynamicSchedMgr::GetInstance(resIndex).AddRootModelInfo(schedCfgInfo);
1349 1 : if (addCfgRet != dgw::FsmStatus::FSM_SUCCESS) {
1350 0 : BQS_LOG_ERROR("Fail to add dynamic sched config, rootModelId[%u], ret[%d]", schedCfgInfo.rootModelId,
1351 : static_cast<int32_t>(addCfgRet));
1352 0 : resultVec[0UL]->retCode = static_cast<int32_t>(BQS_STATUS_DYNAMIC_SCHEDULE_ERROR);
1353 0 : return BQS_STATUS_DYNAMIC_SCHEDULE_ERROR;
1354 : }
1355 :
1356 : QueueSetInputPara inPutParam;
1357 : QueueSetInput inPut;
1358 1 : inPut.queSetWorkMode.qid = schedCfgInfo.responseQue.queueId;
1359 1 : inPut.queSetWorkMode.workMode = QUEUE_MODE_PUSH;
1360 1 : inPutParam.inBuff = static_cast<void *>(&inPut);
1361 1 : inPutParam.inLen = static_cast<uint32_t>(sizeof(QueueSetInput));
1362 1 : drvRet = halQueueSet(0U, QUEUE_SET_WORK_MODE, &inPutParam);
1363 1 : BQS_LOG_RUN_INFO("Set queue[%u] work mode to push for dynamic schedule.", schedCfgInfo.responseQue.queueId);
1364 :
1365 1 : if ((SubscribeQueueEvent(!schedCfgInfo.requestQue.isClientQ, schedCfgInfo.requestQue.queueId,
1366 2 : schedCfgInfo.requestQue.deviceId, resIndex, false) != BQS_STATUS_OK) ||
1367 1 : (SubscribeQueueEvent(!schedCfgInfo.responseQue.isClientQ, schedCfgInfo.responseQue.queueId,
1368 : schedCfgInfo.responseQue.deviceId, resIndex, true) != BQS_STATUS_OK)) {
1369 0 : BQS_LOG_ERROR("Fail to subscribe enque event of [qid:%u-deviceId:%u-isclientQ:%d] or "
1370 : "subscribe f2nf of [qid:%u-deviceId:%u-isclientQ:%d]",
1371 : schedCfgInfo.responseQue.queueId, schedCfgInfo.responseQue.deviceId, schedCfgInfo.responseQue.isClientQ,
1372 : schedCfgInfo.requestQue.queueId, schedCfgInfo.requestQue.deviceId, schedCfgInfo.responseQue.isClientQ);
1373 0 : resultVec[0UL]->retCode = static_cast<int32_t>(BQS_STATUS_INNER_ERROR);
1374 0 : return BQS_STATUS_INNER_ERROR;
1375 : }
1376 1 : dgw::ScheduleConfig::GetInstance().
1377 1 : RecordConfig(dynamicCfg->rootModelId, schedCfgInfo.requestQue, schedCfgInfo.responseQue);
1378 :
1379 : // set result
1380 1 : resultVec[0UL]->retCode = static_cast<int32_t>(retCode);
1381 1 : BQS_LOG_RUN_INFO("Init dynamicSched[%u] operate, cmd[%d], stage[server:process], rootModelId[%u], result:[%d]",
1382 : localRequestQDeviceId, static_cast<int32_t>(cfgInfo->cmd), dynamicCfg->rootModelId,
1383 : static_cast<int32_t>(retCode));
1384 1 : return retCode;
1385 1 : }
1386 :
1387 8 : uint32_t ConfigInfoOperator::ParseDeviceId(const uint32_t rawDeviceId) const
1388 : {
1389 8 : uint32_t localDeviceId = rawDeviceId;
1390 8 : if ((bqs::GetRunContext() != bqs::RunContext::HOST) && (&drvGetLocalDevIDByHostDevID != nullptr)) {
1391 0 : auto retCode = drvGetLocalDevIDByHostDevID(rawDeviceId, &localDeviceId);
1392 0 : if (retCode != static_cast<int32_t>(DRV_ERROR_NONE)) {
1393 0 : BQS_LOG_INFO("host devid(%u) transform to local devid not success.", rawDeviceId);
1394 0 : localDeviceId = rawDeviceId;
1395 : }
1396 : }
1397 8 : return localDeviceId;
1398 : }
1399 :
1400 2 : BqsStatus ConfigInfoOperator::SubscribeQueueEvent(const bool isLocalQ, const uint32_t queueId, const uint32_t deviceId,
1401 : const uint32_t resIndex, const bool isEnqueue) const
1402 : {
1403 2 : const auto subscribeManager = Subscribers::GetInstance().GetSubscribeManager(resIndex, deviceId);
1404 2 : if (subscribeManager == nullptr) {
1405 0 : BQS_LOG_ERROR("Failed to find subscribeManager for isLocalQ:%d, device: %u, resIndex: %u",
1406 : static_cast<int32_t>(isLocalQ), deviceId, resIndex);
1407 0 : return BQS_STATUS_INNER_ERROR;
1408 : }
1409 2 : return isEnqueue ? subscribeManager->Subscribe(queueId) :
1410 2 : subscribeManager->SubscribeFullToNotFull(queueId);
1411 : }
1412 :
1413 8 : BqsStatus ConfigInfoOperator::CheckCommChannelAttr(const CommChannelAttr &attr, const bool isQry) const
1414 : {
1415 8 : if (attr.localTagId != attr.peerTagId) {
1416 1 : BQS_LOG_ERROR("Local tag id[%u] is not equal with peer tag id[%u]. Please check!",
1417 : attr.localTagId, attr.peerTagId);
1418 1 : return BQS_STATUS_PARAM_INVALID;
1419 : }
1420 7 : if (attr.localRankId == attr.peerRankId) {
1421 1 : BQS_LOG_ERROR("local rank id[%u] is equal with peer rank id[%u]. Please check!",
1422 : attr.localRankId, attr.peerRankId);
1423 1 : return BQS_STATUS_PARAM_INVALID;
1424 : }
1425 6 : if (isQry) {
1426 0 : return BQS_STATUS_OK;
1427 : }
1428 : // when qry route, no need check tag depth
1429 6 : if ((attr.localTagDepth == 0U) || (attr.localTagDepth > MAX_TAG_DEPTH)) {
1430 1 : BQS_LOG_ERROR("Local tag depth[%u] is invalid, max tag depth is [%u].", attr.localTagDepth, MAX_TAG_DEPTH);
1431 1 : return BQS_STATUS_PARAM_INVALID;
1432 : }
1433 5 : if ((attr.peerTagDepth == 0U) || (attr.peerTagDepth > MAX_TAG_DEPTH)) {
1434 0 : BQS_LOG_ERROR("Peer tag depth[%u] is invalid, max tag depth is [%u].", attr.peerTagDepth, MAX_TAG_DEPTH);
1435 0 : return BQS_STATUS_PARAM_INVALID;
1436 : }
1437 5 : return BQS_STATUS_OK;
1438 : }
1439 :
1440 1 : BqsStatus ConfigInfoOperator::ProcessStopSchedule(const uint32_t index) const
1441 : {
1442 1 : BQS_LOG_RUN_INFO("ProcessStopSchedule");
1443 1 : const auto cfgInfo = updateCfgInfo_->cfgInfo;
1444 1 : auto &resultVec = updateCfgInfo_->results;
1445 :
1446 1 : const uintptr_t rootModelIdsAddr = updateCfgInfo_->mbufData + sizeof(ConfigInfo);
1447 1 : const uint32_t * const rootModelIds = PtrToPtr<void, uint32_t>(ValueToPtr(rootModelIdsAddr));
1448 1 : const uint32_t rootModelNum = cfgInfo->cfg.reDeployCfg.rootModelNum;
1449 :
1450 1 : if ((rootModelNum != 0U) && (rootModelIds == nullptr)) {
1451 0 : BQS_LOG_ERROR("Invalid rootModelIds");
1452 0 : resultVec[0UL]->retCode = static_cast<int32_t>(BQS_STATUS_PARAM_INVALID);
1453 0 : return BQS_STATUS_PARAM_INVALID;
1454 : }
1455 :
1456 1 : std::unordered_set<uint32_t> rootModelSet;
1457 2 : for (uint32_t i = 0U; i < rootModelNum; i++) {
1458 1 : dgw::ScheduleConfig::GetInstance().StopSched(rootModelIds[i]);
1459 1 : rootModelSet.insert(rootModelIds[i]);
1460 : }
1461 1 : (void)dgw::DynamicSchedMgr::GetInstance(index).ClearCacheRouteResult();
1462 :
1463 1 : const auto retCode = BindRelation::GetInstance().MakeSureOutputCompletion(index, rootModelSet);
1464 :
1465 1 : resultVec[0UL]->retCode = static_cast<int32_t>(retCode);
1466 1 : BQS_LOG_RUN_INFO("Finish ProcessStopSchedule, retCode is %d", static_cast<int32_t>(retCode));
1467 1 : return retCode;
1468 1 : }
1469 :
1470 1 : BqsStatus ConfigInfoOperator::ProcessRestartSchedule(const uint32_t index) const
1471 : {
1472 1 : BQS_LOG_RUN_INFO("ProcessRestartSchedule");
1473 1 : const auto cfgInfo = updateCfgInfo_->cfgInfo;
1474 1 : auto &resultVec = updateCfgInfo_->results;
1475 :
1476 1 : const uintptr_t rootModelIdsAddr = updateCfgInfo_->mbufData + sizeof(ConfigInfo);
1477 1 : const uint32_t * const rootModelIds = PtrToPtr<void, uint32_t>(ValueToPtr(rootModelIdsAddr));
1478 1 : const uint32_t rootModelNum = cfgInfo->cfg.reDeployCfg.rootModelNum;
1479 :
1480 1 : if ((rootModelNum != 0U) && (rootModelIds == nullptr)) {
1481 0 : BQS_LOG_ERROR("Invalid rootModelIds");
1482 0 : resultVec[0UL]->retCode = static_cast<int32_t>(BQS_STATUS_PARAM_INVALID);
1483 0 : return BQS_STATUS_PARAM_INVALID;
1484 : }
1485 :
1486 1 : std::unordered_set<uint32_t> rootModelSet;
1487 2 : for (uint32_t i = 0U; i < rootModelNum; i++) {
1488 1 : rootModelSet.insert(rootModelIds[i]);
1489 : }
1490 1 : const auto ret = BindRelation::GetInstance().ClearInputQueue(index, rootModelSet);
1491 1 : if (ret == BQS_STATUS_OK) {
1492 1 : std::vector<dgw::DynamicSchedMgr::ResponseInfo> responses;
1493 2 : for (uint32_t i = 0U; i < rootModelNum; i++) {
1494 : do {
1495 1 : responses.clear();
1496 1 : (void)dgw::DynamicSchedMgr::GetInstance(index).GetResponse(rootModelIds[i], responses);
1497 1 : } while (!responses.empty());
1498 1 : dgw::ScheduleConfig::GetInstance().RestartSched(rootModelIds[i]);
1499 : }
1500 1 : }
1501 1 : resultVec[0UL]->retCode = static_cast<int32_t>(ret);
1502 1 : BQS_LOG_RUN_INFO("Finish ProcessRestartSchedule");
1503 1 : return ret;
1504 1 : }
1505 : }
|
