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