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 "router_server.h"
12 :
13 : #include <map>
14 : #include <csignal>
15 : #include <sstream>
16 : #include <securec.h>
17 : #include <sys/types.h>
18 : #include <unistd.h>
19 : #include "statistic_manager.h"
20 : #include "bqs_log.h"
21 : #include "common/bqs_util.h"
22 : #include "hccl_process.h"
23 : #include "hccl/hccl_so_manager.h"
24 : #include "common/type_def.h"
25 : #include "queue_manager.h"
26 : #include "queue_schedule_hal_interface_ref.h"
27 : #include "qs_interface_process.h"
28 : #include "queue_schedule_sub_module_interface.h"
29 : #include "entity_manager.h"
30 :
31 : namespace bqs {
32 : namespace {
33 : const std::string PIPELINE_QUEUE_NAME = "QsPipeQueue";
34 : constexpr const uint16_t MAJOR_VERSION = 3U;
35 : constexpr const uint16_t MINOR_VERSION = 0U;
36 : constexpr const QueueShareAttr ADMIN_QUEUE_ATTR = {1U, 1U, 1U, 0U};
37 : constexpr const char_t *ROUTER_SERVER_THREAD_NAME_PREFIX = "router_server";
38 :
39 : // mapping of request subeventId and response subeventId
40 : const std::map<int32_t, int32_t> g_reqRspMapping = {
41 : {AICPU_BIND_QUEUE, AICPU_BIND_QUEUE_RES},
42 : {AICPU_BIND_QUEUE_INIT, AICPU_BIND_QUEUE_INIT_RES},
43 : {AICPU_UNBIND_QUEUE, AICPU_UNBIND_QUEUE_RES},
44 : {AICPU_QUERY_QUEUE, AICPU_QUERY_QUEUE_RES},
45 : {AICPU_QUERY_QUEUE_NUM, AICPU_QUERY_QUEUE_NUM_RES}
46 : };
47 : // mapping of request subeventId and qs operate type
48 : const std::map<uint32_t, QsOperType> g_qsOperation = {
49 : {static_cast<uint32_t>(AICPU_BIND_QUEUE_INIT), QsOperType::BIND_INIT},
50 : {static_cast<uint32_t>(AICPU_BIND_QUEUE_INIT), QsOperType::BIND_INIT},
51 : {static_cast<uint32_t>(AICPU_QUERY_QUEUE_NUM), QsOperType::QUERY_NUM},
52 : {static_cast<uint32_t>(AICPU_QUEUE_RELATION_PROCESS), QsOperType::RELATION_PROCESS},
53 : {static_cast<uint32_t>(ACL_BIND_QUEUE_INIT), QsOperType::BIND_INIT},
54 : {static_cast<uint32_t>(ACL_BIND_QUEUE), QsOperType::RELATION_PROCESS},
55 : {static_cast<uint32_t>(ACL_UNBIND_QUEUE), QsOperType::RELATION_PROCESS},
56 : {static_cast<uint32_t>(ACL_QUERY_QUEUE_NUM), QsOperType::QUERY_NUM},
57 : {static_cast<uint32_t>(ACL_QUERY_QUEUE), QsOperType::RELATION_PROCESS},
58 : {static_cast<uint32_t>(DGW_CREATE_HCOM_HANDLE), QsOperType::CREATE_HCOM_HANDLE},
59 : {static_cast<uint32_t>(DGW_DESTORY_HCOM_HANDLE), QsOperType::DESTROY_HCOM_HANDLE},
60 : {static_cast<uint32_t>(BIND_HOSTPID), QsOperType::BIND_HOST_PID},
61 : // new operation type for flow gateway
62 : {static_cast<uint32_t>(UPDATE_CONFIG), QsOperType::UPDATE_CONFIG},
63 : {static_cast<uint32_t>(QUERY_CONFIG_NUM), QsOperType::QUERY_CONFIG_NUM},
64 : {static_cast<uint32_t>(QUERY_CONFIG), QsOperType::QUERY_CONFIG},
65 : {static_cast<uint32_t>(QUERY_LINKSTATUS), QsOperType::QUERY_LINKSTATUS},
66 : {static_cast<uint32_t>(QUERY_LINKSTATUS_V2), QsOperType::QUERY_LINKSTATUS_V2},
67 : };
68 :
69 : // 老版驱动结构体——解析数据时需要使用与驱动相同的结构体
70 : struct old_event_sync_msg {
71 : int pid; /* local pid */
72 : unsigned int dst_engine : 4; /* local engine */
73 : unsigned int gid : 6;
74 : unsigned int event_id : 6;
75 : unsigned int subevent_id : 16; /* Range: 0 ~ 4095 */
76 : char msg[];
77 : };
78 :
79 : }
80 :
81 4 : RouterServer::RouterServer() : processing_(false), done_(false), processingExtra_(false), doneExtra_(true),
82 2 : bindQueueGroupId_(static_cast<uint32_t>(BIND_QUEUE_GROUP_ID)),
83 2 : running_(false), deviceId_(0U), srcPid_(-1), srcVersion_(0U),
84 2 : srcGroupId_(-1), pipelineQueueId_(MAX_QUEUE_ID_NUM), subEventId_(0U),
85 2 : deployMode_(QueueSchedulerRunMode::MULTI_PROCESS),
86 2 : retCode_(static_cast<int32_t>(BQS_STATUS_OK)), attachedFlag_(false),
87 2 : isAicpuEvent_(false), qsRouteListPtr_(nullptr), qsRouterHeadPtr_(nullptr),
88 2 : qsRouterQueryPtr_(nullptr), drvSyncMsg_(nullptr), aicpuRspHead_(0UL),
89 2 : f2nfGroupId_(0U), schedPolicy_(0UL), cfgInfoOperator_(nullptr), callHcclFlag_(false),
90 4 : numaFlag_(false), readyToHandleMsg_(false), manageThreadStatus_(ThreadStatus::NOT_INIT),
91 6 : needAttachGroup_(false), compatMsg_(false)
92 2 : {}
93 :
94 176 : void RouterServer::Destroy()
95 : {
96 176 : if (!running_) {
97 170 : return;
98 : }
99 6 : BQS_LOG_INFO("[RouterServer]QS Server destroy.");
100 6 : running_ = false;
101 6 : queueRouteQueryList_.clear();
102 6 : cv_.notify_all();
103 6 : if (monitorQsEvent_.joinable()) {
104 5 : monitorQsEvent_.join();
105 : }
106 6 : manageThreadStatus_ = ThreadStatus::NOT_INIT;
107 6 : if (pipelineQueueId_ < MAX_QUEUE_ID_NUM) {
108 3 : const auto ret = halQueueDestroy(deviceId_, pipelineQueueId_);
109 3 : BQS_LOG_ERROR_WHEN(ret != DRV_ERROR_NONE,
110 : "[RouterServer]Destroy relation event buff queue error, queue id[%u], ret[%d]",
111 : pipelineQueueId_.load(), static_cast<int32_t>(ret));
112 : }
113 6 : cfgInfoOperator_ = nullptr;
114 6 : BQS_LOG_RUN_INFO("[RouterServer]QS Server finish destroy.");
115 : }
116 :
117 2 : RouterServer::~RouterServer()
118 : {
119 2 : Destroy();
120 2 : }
121 :
122 0 : bool RouterServer::GetCallHcclFlag() const
123 : {
124 0 : return callHcclFlag_;
125 : }
126 :
127 3 : uint32_t RouterServer::GetPipelineQueueId() const
128 : {
129 3 : return pipelineQueueId_;
130 : }
131 :
132 534 : RouterServer &RouterServer::GetInstance()
133 : {
134 534 : static RouterServer instance;
135 534 : return instance;
136 : }
137 :
138 106 : void RouterServer::HandleBqsMsg(event_info &info)
139 : {
140 106 : if (info.comm.event_id != EVENT_QS_MSG) {
141 1 : BQS_LOG_ERROR("[RouterServer]Queue schedule does not support [%d] event",
142 : static_cast<int32_t>(info.comm.event_id));
143 1 : return;
144 : }
145 105 : subEventId_ = info.comm.subevent_id;
146 : // check aicpu event
147 105 : isAicpuEvent_ = (subEventId_ <= static_cast<uint32_t>(AICPU_RELATED_MESSAGE_SPLIT)) ? true : false;
148 105 : if (!isAicpuEvent_) {
149 : // msg is char array, not need check nullptr
150 87 : drvSyncMsg_ = PtrToPtr<char_t, event_sync_msg>(info.priv.msg);
151 : }
152 :
153 105 : if (!readyToHandleMsg_) {
154 1 : BQS_LOG_WARN("[RouterServer] is not ready to HandleBqsMsg");
155 1 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_NOT_INIT));
156 1 : return;
157 : }
158 :
159 : // check subEventId, SendRspEvent need drvSyncMsg if event from acl
160 104 : const auto iter = g_qsOperation.find(subEventId_);
161 104 : if (iter == g_qsOperation.end()) {
162 2 : BQS_LOG_RUN_INFO("[RouterServer]SubEventId is invalid[%d]", subEventId_);
163 2 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_PARAM_INVALID));
164 2 : return;
165 : }
166 :
167 : // aicpu message is not allowed in thread mode.
168 102 : if (isAicpuEvent_ && (deployMode_ == QueueSchedulerRunMode::MULTI_THREAD)) {
169 1 : BQS_LOG_ERROR("[RouterServer]Thread mode[%u] does not sopport event[%u] from aicpu.",
170 : static_cast<int32_t>(deployMode_), subEventId_);
171 1 : return;
172 : }
173 101 : PreProcessEvent(info);
174 101 : BQS_LOG_INFO("[RouterServer]HandleBqsMsg end, isAicpuEvent[%d].", static_cast<int32_t>(isAicpuEvent_));
175 101 : return;
176 : }
177 :
178 106 : void RouterServer::PreProcessEvent(const event_info &info)
179 : {
180 : // get event head for responseing by sync interface
181 106 : BQS_LOG_INFO("[RouterServer]PreProcess start operation[%u]", subEventId_);
182 106 : const auto iter = g_qsOperation.find(subEventId_);
183 106 : if (iter == g_qsOperation.end()) {
184 1 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_PARAM_INVALID));
185 1 : BQS_LOG_ERROR("[RouterServer] RouterServer receive unsupported msg type:%d", subEventId_);
186 1 : return;
187 : }
188 :
189 105 : const QsOperType operType = iter->second;
190 105 : switch (operType) {
191 4 : case QsOperType::BIND_INIT:
192 4 : SendRspEvent(static_cast<int32_t>(ProcessBindInit(info)));
193 4 : break;
194 4 : case QsOperType::QUERY_NUM:
195 4 : StatisticManager::GetInstance().GetBindStat();
196 4 : ParseGetBindNumMsg(info);
197 4 : break;
198 10 : case QsOperType::RELATION_PROCESS: {
199 : // get message from mbuf, eventid also in mbuf
200 10 : Mbuf *mBuf = nullptr;
201 10 : const auto resultCode = ParseRelationInfo(&mBuf);
202 10 : if (resultCode != BQS_STATUS_OK) {
203 1 : BQS_LOG_ERROR("[RouterServer]Get detail message from mbuf failed ret[%d].",
204 : static_cast<int32_t>(resultCode));
205 1 : SendRspEvent(static_cast<int32_t>(resultCode));
206 1 : if ((mBuf != nullptr) && (srcVersion_ != 0U)) {
207 0 : const auto freeRet = halMbufFree(mBuf);
208 0 : BQS_LOG_ERROR_WHEN(freeRet != static_cast<int32_t>(DRV_ERROR_NONE),
209 : "Free mbuf failed, ret is %d", freeRet);
210 : }
211 1 : return;
212 : }
213 :
214 9 : BQS_LOG_INFO("[RouterServer]Start to process relation event[%u].", subEventId_);
215 9 : ProcessQueueRelationEvent(mBuf);
216 9 : break;
217 : }
218 85 : case QsOperType::UPDATE_CONFIG:
219 : case QsOperType::CREATE_HCOM_HANDLE:
220 : case QsOperType::DESTROY_HCOM_HANDLE:
221 : case QsOperType::QUERY_CONFIG:
222 : case QsOperType::QUERY_CONFIG_NUM: {
223 85 : ProcessConfigEvent(operType);
224 85 : break;
225 : }
226 0 : case QsOperType::QUERY_LINKSTATUS: {
227 0 : ProcessQueryLinkStatusEvent();
228 0 : break;
229 : }
230 1 : case QsOperType::QUERY_LINKSTATUS_V2: {
231 1 : ProcessQueryLinkStatusEvent();
232 1 : break;
233 : }
234 1 : default: {
235 1 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_PARAM_INVALID));
236 1 : BQS_LOG_RUN_INFO("[RouterServer]RouterServer receive unsupported msg type:%u", subEventId_);
237 1 : break;
238 : }
239 : }
240 104 : return;
241 : }
242 :
243 87 : void RouterServer::ProcessConfigEvent(const QsOperType operType)
244 : {
245 87 : void *mbuf = nullptr;
246 87 : auto drvRet = halQueueDeQueue(deviceId_, pipelineQueueId_, &mbuf);
247 87 : if ((drvRet != DRV_ERROR_NONE) || (mbuf == nullptr)) {
248 2 : BQS_LOG_ERROR("halQueueDeQueue from queue[%u] in device[%u] failed, error[%d]", pipelineQueueId_.load(),
249 : deviceId_, static_cast<int32_t>(drvRet));
250 1 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_DRIVER_ERROR));
251 2 : return;
252 : }
253 86 : auto resultCode = (cfgInfoOperator_ == nullptr) ? BQS_STATUS_INNER_ERROR :
254 85 : cfgInfoOperator_->ParseConfigEvent(subEventId_, pipelineQueueId_, mbuf, srcVersion_);
255 86 : if (resultCode == BQS_STATUS_WAIT) {
256 37 : resultCode = WaitSyncMsgProc();
257 : }
258 86 : if (operType == QsOperType::CREATE_HCOM_HANDLE) {
259 8 : callHcclFlag_ = true;
260 : }
261 :
262 86 : if (srcVersion_ != 0U) {
263 86 : BQS_LOG_INFO("Enque mbuf back");
264 86 : drvRet = halQueueEnQueue(deviceId_, pipelineQueueId_, mbuf);
265 86 : if (drvRet != DRV_ERROR_NONE) {
266 2 : BQS_LOG_ERROR("halQueueEnQueue into queue[%u] in device[%u] failed, error[%d]", pipelineQueueId_.load(),
267 : deviceId_, static_cast<int32_t>(drvRet));
268 1 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_DRIVER_ERROR));
269 1 : const auto freeRet = halMbufFree(PtrToPtr<void, Mbuf>(mbuf));
270 1 : BQS_LOG_ERROR_WHEN(freeRet != static_cast<int32_t>(DRV_ERROR_NONE),
271 : "Free mbuf failed, ret is %d", freeRet);
272 1 : return;
273 : }
274 : }
275 85 : BQS_LOG_INFO("config for operate[%d] resultCode is %d.", static_cast<int32_t>(operType),
276 : static_cast<int32_t>(resultCode));
277 85 : SendRspEvent(static_cast<int32_t>(resultCode));
278 : }
279 :
280 2 : void RouterServer::ProcessQueryLinkStatusEvent()
281 : {
282 2 : int32_t ret = static_cast<int32_t>(dgw::EntityManager::Instance(0U).CheckLinkStatus());
283 2 : if ((ret == 0) && numaFlag_) {
284 0 : ret = static_cast<int32_t>(dgw::EntityManager::Instance(1U).CheckLinkStatus());
285 : }
286 2 : SendRspEvent(ret);
287 2 : }
288 :
289 9 : void RouterServer::ProcessQueueRelationEvent(Mbuf *mbuf)
290 : {
291 9 : BqsStatus ret = BQS_STATUS_INNER_ERROR;
292 9 : switch (subEventId_) {
293 0 : case AICPU_BIND_QUEUE: {
294 0 : StatisticManager::GetInstance().BindStat();
295 0 : ret = WaitBindMsgProc();
296 0 : break;
297 : }
298 2 : case ACL_BIND_QUEUE: {
299 2 : StatisticManager::GetInstance().BindStat();
300 2 : ret = WaitBindMsgProc();
301 2 : break;
302 : }
303 1 : case AICPU_UNBIND_QUEUE: {
304 1 : StatisticManager::GetInstance().UnbindStat();
305 1 : ret = WaitBindMsgProc();
306 1 : break;
307 : }
308 0 : case ACL_UNBIND_QUEUE: {
309 0 : StatisticManager::GetInstance().UnbindStat();
310 0 : ret = WaitBindMsgProc();
311 0 : break;
312 : }
313 4 : case AICPU_QUERY_QUEUE: {
314 4 : StatisticManager::GetInstance().GetBindStat();
315 4 : ret = ParseGetBindDetailMsg();
316 4 : break;
317 : }
318 0 : case ACL_QUERY_QUEUE: {
319 0 : StatisticManager::GetInstance().GetBindStat();
320 0 : ret = ParseGetBindDetailMsg();
321 0 : break;
322 : }
323 2 : default:
324 2 : BQS_LOG_ERROR("[RouterServer]unsupport subEventId[%u] in bind relation procedure", subEventId_);
325 2 : break;
326 : }
327 :
328 9 : if (srcVersion_ != 0U) {
329 7 : BQS_LOG_INFO("Enque mbuf back.");
330 7 : const auto drvRet = halQueueEnQueue(deviceId_, pipelineQueueId_, mbuf);
331 7 : if (drvRet != DRV_ERROR_NONE) {
332 2 : BQS_LOG_ERROR("halQueueEnQueue into queue[%u] in device[%u] failed, error[%d].", pipelineQueueId_.load(),
333 : deviceId_, static_cast<int32_t>(drvRet));
334 1 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_DRIVER_ERROR));
335 1 : const auto freeRet = halMbufFree(mbuf);
336 1 : BQS_LOG_ERROR_WHEN(freeRet != static_cast<int32_t>(DRV_ERROR_NONE),
337 : "Free mbuf failed, ret is %d", freeRet);
338 1 : return;
339 : }
340 : }
341 8 : SendRspEvent(static_cast<int32_t>(ret));
342 8 : return;
343 : }
344 :
345 3 : BqsStatus RouterServer::WaitBindMsgProc()
346 : {
347 3 : BQS_LOG_INFO("[RouterServer]Bind relation [add/del], stage [wait]");
348 3 : auto ret = ParseBindUnbindMsg();
349 3 : if (ret == BQS_STATUS_OK) {
350 0 : ret = WaitSyncMsgProc();
351 : }
352 3 : BQS_LOG_INFO("[RouterServer]RouterServer WaitBindMsgProc end");
353 3 : return ret;
354 : }
355 :
356 2 : BqsStatus RouterServer::AttachAndInitGroup()
357 : {
358 2 : BQS_LOG_INFO("[RouterServer]Attach and init group begin.");
359 2 : int32_t drvRet = 0;
360 : // 针对aicpusd 与qs合设的情况，如果qs以模块方式启动，则不需要重复加组，因为aicpusd已经加组
361 2 : if (qsInitGroupName_.empty() && (!SubModuleInterface::GetInstance().GetStartFlag())) {
362 1 : const std::unique_ptr<GroupQueryOutput> groupInfoPtr(new (std::nothrow) GroupQueryOutput());
363 1 : if (groupInfoPtr == nullptr) {
364 0 : BQS_LOG_ERROR("[RouterServer] Fail to allocate GroupQueryOutput");
365 0 : return BQS_STATUS_INNER_ERROR;
366 : }
367 1 : GroupQueryOutput &groupInfo = *(groupInfoPtr.get());
368 1 : uint32_t groupInfoLen = 0U;
369 1 : pid_t curPid = drvDeviceGetBareTgid();
370 : // query group info for current qs process
371 1 : drvRet = halGrpQuery(GRP_QUERY_GROUPS_OF_PROCESS, &curPid, static_cast<uint32_t>(sizeof(curPid)),
372 : reinterpret_cast<void *>(&groupInfo), &groupInfoLen);
373 1 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
374 0 : BQS_LOG_ERROR("[RouterServer]halGrpQuery of qs[%d] failed before attached,ret[%d]", curPid, drvRet);
375 0 : return BQS_STATUS_DRIVER_ERROR;
376 : }
377 : // not in any group, cannot do attach process
378 1 : if (groupInfoLen == 0U) {
379 0 : BQS_LOG_ERROR("[RouterServer]QS should be add sharepool group before initial by aicpu or acl.");
380 0 : return BQS_STATUS_INNER_ERROR;
381 : }
382 1 : if ((groupInfoLen % sizeof(groupInfo.grpQueryGroupsOfProcInfo[0])) != 0U) {
383 0 : BQS_LOG_ERROR("[RouterServer]Group info size[%d] is invalid", groupInfoLen);
384 0 : return BQS_STATUS_DRIVER_ERROR;
385 : }
386 1 : const uint32_t groupNum = static_cast<uint32_t>(groupInfoLen / sizeof(groupInfo.grpQueryGroupsOfProcInfo[0]));
387 2 : for (uint32_t i = 0U; i < groupNum; ++i) {
388 : // attach and initial
389 1 : drvRet = halGrpAttach(groupInfo.grpQueryGroupsOfProcInfo[i].groupName, 0);
390 1 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
391 0 : BQS_LOG_ERROR("[RouterServer]Group[%s] attach failed for slave aicpusd[%d] ret[%d]",
392 : groupInfo.grpQueryGroupsOfProcInfo[i].groupName, curPid, drvRet);
393 0 : return BQS_STATUS_DRIVER_ERROR;
394 : }
395 1 : BQS_LOG_INFO("[RouterServer] halGrpAttach execute succ. group[%s] was attached by QS",
396 : groupInfo.grpQueryGroupsOfProcInfo[i].groupName);
397 : }
398 1 : }
399 2 : attachedFlag_ = true;
400 2 : BuffCfg defaultCfg = {};
401 2 : drvRet = halBuffInit(&defaultCfg);
402 2 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) &&
403 : (drvRet != static_cast<int32_t>(DRV_ERROR_REPEATED_INIT))) {
404 1 : BQS_LOG_ERROR("[RouterServer] Buffer initial failed for qs. ret[%d]", drvRet);
405 1 : return BQS_STATUS_DRIVER_ERROR;
406 : }
407 1 : BQS_LOG_INFO("[RouterServer] Buffer init success ret[%d]", drvRet);
408 1 : return BQS_STATUS_OK;
409 : }
410 :
411 2 : BqsStatus RouterServer::CreateAndGrantPipelineQueue()
412 : {
413 2 : BQS_LOG_INFO("[RouterServer]Create and grant pipeline queue begin.");
414 : // do initial process
415 2 : const std::unique_lock<std::mutex> lk(mutex_);
416 2 : QueueAttr queAttr = {};
417 2 : std::string nameStr(PIPELINE_QUEUE_NAME);
418 2 : pid_t curPidTemp = 0;
419 2 : if (bqs::GetRunContext() == bqs::RunContext::HOST) {
420 2 : curPidTemp = getpid();
421 2 : queAttr.deploy_type = LOCAL_QUEUE_DEPLOY;
422 : } else {
423 0 : curPidTemp = drvDeviceGetBareTgid();
424 0 : queAttr.deploy_type = CLIENT_QUEUE_DEPLOY;
425 : }
426 2 : const uint32_t curPid = static_cast<uint32_t>(curPidTemp);
427 2 : nameStr += std::to_string(curPid);
428 2 : const auto memcpyRet = memcpy_s(queAttr.name, static_cast<uint32_t>(QUEUE_MAX_STR_LEN),
429 2 : nameStr.c_str(), nameStr.length() + 1UL);
430 2 : if (memcpyRet != EOK) {
431 0 : BQS_LOG_ERROR("[RouterServer]CreateAndGrantPipelineQueue memcpy_s failed, ret=%d.", memcpyRet);
432 0 : return BQS_STATUS_INNER_ERROR;
433 : }
434 2 : queAttr.depth = 2U;
435 2 : uint32_t queueId = 0U;
436 : // create queue
437 2 : auto drvRet = halQueueCreate(deviceId_, &queAttr, &queueId);
438 2 : if ((drvRet != DRV_ERROR_NONE) || (queueId >= MAX_QUEUE_ID_NUM)) {
439 0 : BQS_LOG_ERROR("[RouterServer]Create queue[%s] error or qID[%u] is invalid, ret[%d]",
440 : PIPELINE_QUEUE_NAME.c_str(), queueId, static_cast<int32_t>(drvRet));
441 0 : return BQS_STATUS_DRIVER_ERROR;
442 : }
443 2 : drvRet = halQueueAttach(deviceId_, queueId, 0);
444 2 : if (drvRet != DRV_ERROR_NONE) {
445 0 : BQS_LOG_ERROR("Fail to attach queue[%ud], result[%d]", queueId, static_cast<int32_t>(drvRet));
446 0 : return BQS_STATUS_DRIVER_ERROR;
447 : }
448 2 : pipelineQueueId_ = queueId;
449 2 : if (deployMode_ == QueueSchedulerRunMode::MULTI_THREAD) {
450 0 : BQS_LOG_INFO("[RouterServer]Thread mode need not grant queue to other process");
451 0 : return BQS_STATUS_OK;
452 : }
453 : // grant pipeline queue to src process
454 :
455 2 : drvRet = halQueueGrant(deviceId_, static_cast<int32_t>(queueId), srcPid_, ADMIN_QUEUE_ATTR);
456 2 : if (drvRet != DRV_ERROR_NONE) {
457 0 : BQS_LOG_ERROR("[RouterServer]Fail to add queue[%d] authority for aicpusd[%d], result[%d].",
458 : queueId, srcPid_, static_cast<int32_t>(drvRet));
459 0 : return BQS_STATUS_DRIVER_ERROR;
460 : }
461 4 : BQS_LOG_RUN_INFO("Success to init pipelineQ[%u].", pipelineQueueId_.load());
462 2 : return BQS_STATUS_OK;
463 2 : }
464 :
465 4 : BqsStatus RouterServer::ProcessBindInit(const event_info &info)
466 : {
467 4 : if (info.priv.msg_len != sizeof(QsBindInit)) {
468 0 : BQS_LOG_ERROR("[RouterServer]Bind initial event message invalid, msgLen[%u]", info.priv.msg_len);
469 0 : return BQS_STATUS_PARAM_INVALID;
470 : }
471 : // bind initial already done
472 4 : if ((pipelineQueueId_ < MAX_QUEUE_ID_NUM) && (srcPid_ != -1)) {
473 2 : BQS_LOG_RUN_INFO("Pipeline queue already existed[%d], return pipelienQueueid", pipelineQueueId_.load());
474 1 : return BQS_STATUS_OK;
475 : }
476 3 : const QsBindInit * const bindInitMsg = reinterpret_cast<const QsBindInit *>(info.priv.msg);
477 3 : aicpuRspHead_ = bindInitMsg->syncEventHead;
478 3 : srcPid_ = bindInitMsg->pid;
479 3 : srcVersion_ = bindInitMsg->majorVersion;
480 3 : srcGroupId_ = static_cast<int32_t>(bindInitMsg->grpId);
481 6 : BQS_LOG_RUN_INFO("[RouterServer]Get hostpid[%d] srcGroup[%d], srcVersion[%u]",
482 : srcPid_, srcGroupId_.load(), srcVersion_);
483 :
484 : // process mode need to attach group at first
485 3 : if (((deployMode_ == QueueSchedulerRunMode::SINGLE_PROCESS) ||
486 3 : (deployMode_ == QueueSchedulerRunMode::MULTI_PROCESS)) && (!attachedFlag_)) {
487 1 : BQS_LOG_INFO("[RouterServer]start up attach and init group.");
488 1 : const auto attachRet = AttachAndInitGroup();
489 1 : if (attachRet != BQS_STATUS_OK) {
490 0 : BQS_LOG_ERROR("[RouterServer]AttachAndInitGroup failed, ret[%d].", static_cast<int32_t>(attachRet));
491 0 : return attachRet;
492 : }
493 : }
494 :
495 3 : if (qsInitGroupName_.empty()) {
496 2 : auto queueInitRet = QueueManager::GetInstance().InitQueue();
497 2 : if (queueInitRet != BQS_STATUS_OK) {
498 1 : BQS_LOG_ERROR("[RouterServer] Queue init failed");
499 1 : return queueInitRet;
500 : }
501 1 : if (numaFlag_) {
502 0 : queueInitRet = QueueManager::GetInstance().InitQueueExtra();
503 0 : if (queueInitRet != BQS_STATUS_OK) {
504 0 : BQS_LOG_ERROR("[RouterServer] Queue init failed");
505 0 : return queueInitRet;
506 : }
507 : }
508 : }
509 :
510 2 : const auto ret = CreateAndGrantPipelineQueue();
511 2 : if (ret != BQS_STATUS_OK) {
512 0 : return ret;
513 : }
514 6 : BQS_LOG_RUN_INFO("First bind initial success, srcPid[%d], srvGroupId[%d], pipelineQueueId[%u]",
515 : srcPid_, srcGroupId_.load(), pipelineQueueId_.load());
516 2 : return BQS_STATUS_OK;
517 : }
518 :
519 112 : void RouterServer::FillRspContent(QsProcMsgRsp &retRsp, const int32_t resultCode)
520 : {
521 : // only aicpu event need fill in aicpuRspHead
522 112 : retRsp.syncEventHead = isAicpuEvent_ ? aicpuRspHead_ : 0UL;
523 112 : retRsp.retCode = resultCode;
524 112 : retRsp.minorVersion = MINOR_VERSION;
525 112 : retRsp.majorVersion = MAJOR_VERSION;
526 112 : if ((subEventId_ == static_cast<uint32_t>(AICPU_BIND_QUEUE_INIT)) ||
527 107 : (subEventId_ == static_cast<uint32_t>(ACL_BIND_QUEUE_INIT))) {
528 : // init message return pipelineID
529 8 : retRsp.retValue = (resultCode == static_cast<int32_t>(BQS_STATUS_OK)) ? pipelineQueueId_.load()
530 : : MAX_QUEUE_ID_NUM;
531 : }
532 112 : if ((subEventId_ == static_cast<uint32_t>(AICPU_QUERY_QUEUE_NUM)) ||
533 108 : (subEventId_ == static_cast<uint32_t>(ACL_QUERY_QUEUE_NUM))) {
534 : // query num message return bind num
535 8 : retRsp.retValue = (resultCode != static_cast<int32_t>(BQS_STATUS_OK)) ? 0U :
536 4 : static_cast<uint32_t>(queueRouteQueryList_.size());
537 4 : queueRouteQueryList_.clear();
538 : }
539 112 : if ((subEventId_ == static_cast<uint32_t>(AICPU_QUERY_QUEUE)) ||
540 108 : (subEventId_ == static_cast<uint32_t>(ACL_QUERY_QUEUE)) ||
541 108 : (subEventId_ == static_cast<uint32_t>(AICPU_BIND_QUEUE)) ||
542 107 : (subEventId_ == static_cast<uint32_t>(ACL_BIND_QUEUE)) ||
543 105 : (subEventId_ == static_cast<uint32_t>(AICPU_UNBIND_QUEUE)) ||
544 104 : (subEventId_ == static_cast<uint32_t>(ACL_UNBIND_QUEUE))) {
545 8 : retRsp.retValue = pipelineQueueId_;
546 : }
547 112 : return;
548 : }
549 :
550 112 : void RouterServer::SendRspEvent(const int32_t result)
551 : {
552 112 : BQS_LOG_INFO("[RouterServer]Start to response message subeventid[%u]", subEventId_);
553 112 : QsProcMsgRsp retRsp = {};
554 112 : FillRspContent(retRsp, result);
555 112 : event_summary qsEvent = {};
556 112 : qsEvent.msg = PtrToPtr<QsProcMsgRsp, char_t>(&retRsp);
557 112 : qsEvent.msg_len = static_cast<uint32_t>(sizeof(retRsp));
558 112 : auto drvRet = DRV_ERROR_NONE;
559 112 : if (!isAicpuEvent_) {
560 87 : BQS_LOG_INFO("[RouterServer] Do ACL response");
561 87 : qsEvent.dst_engine = compatMsg_ ? PtrToPtr<event_sync_msg, old_event_sync_msg>(drvSyncMsg_)->dst_engine
562 0 : : drvSyncMsg_->dst_engine;
563 87 : qsEvent.policy = ONLY;
564 87 : qsEvent.pid = compatMsg_ ? PtrToPtr<event_sync_msg, old_event_sync_msg>(drvSyncMsg_)->pid
565 0 : : drvSyncMsg_->pid;
566 87 : qsEvent.grp_id = compatMsg_ ? PtrToPtr<event_sync_msg, old_event_sync_msg>(drvSyncMsg_)->gid
567 0 : : drvSyncMsg_->gid;
568 87 : const int32_t eventId = compatMsg_ ? PtrToPtr<event_sync_msg, old_event_sync_msg>(drvSyncMsg_)->event_id
569 0 : : drvSyncMsg_->event_id;
570 87 : qsEvent.event_id = static_cast<EVENT_ID>(eventId);
571 87 : qsEvent.subevent_id = compatMsg_ ? PtrToPtr<event_sync_msg, old_event_sync_msg>(drvSyncMsg_)->subevent_id
572 0 : : drvSyncMsg_->subevent_id;
573 87 : drvRet = halEschedSubmitEvent(deviceId_, &qsEvent); // drv interface require use 0
574 87 : drvSyncMsg_ = nullptr;
575 87 : BQS_LOG_INFO("[SendRspEvent] dst_engine[%u], pid[%d], grp_id[%u], eventId[%d], subevent_id[%u], deviceId[%u]",
576 : qsEvent.dst_engine, qsEvent.pid, qsEvent.grp_id, qsEvent.event_id, qsEvent.subevent_id, deviceId_);
577 : } else {
578 25 : BQS_LOG_INFO("[RouterServer] Do AICPU response");
579 25 : qsEvent.pid = srcPid_;
580 25 : qsEvent.grp_id = static_cast<uint32_t>(srcGroupId_);
581 25 : qsEvent.event_id = EVENT_QS_MSG;
582 25 : qsEvent.msg_len = static_cast<uint32_t>(sizeof(QsProcMsgRspDstAicpu));
583 25 : const auto iter = g_reqRspMapping.find(static_cast<int32_t>(subEventId_));
584 25 : if (iter != g_reqRspMapping.end()) {
585 15 : qsEvent.subevent_id = static_cast<uint32_t>(iter->second);
586 : } else {
587 10 : BQS_LOG_ERROR("[RouterServer]ERROR Invalid subeventId[%u]", subEventId_);
588 : }
589 25 : drvRet = halEschedSubmitEvent(deviceId_, &qsEvent); // drv interface require use 0
590 25 : aicpuRspHead_ = 0UL;
591 : }
592 112 : BQS_LOG_ERROR_WHEN(drvRet != DRV_ERROR_NONE, "[RouterServer]ERROR failed to submit event[%u], result[%d].",
593 : subEventId_, static_cast<int32_t>(drvRet));
594 112 : BQS_LOG_INFO("[RouterServer]Finish response message subeventid[%u] ret[%d]", subEventId_, result);
595 112 : qsRouteListPtr_ = nullptr;
596 112 : qsRouterHeadPtr_ = nullptr;
597 112 : subEventId_ = 0U;
598 224 : return;
599 : }
600 :
601 2 : void RouterServer::ProcessBindQueue(const uint32_t index)
602 : {
603 2 : BQS_LOG_INFO("[RouterServer]Bind relation [add], stage [server:process].");
604 2 : auto &relationInstance = BindRelation::GetInstance();
605 2 : retCode_ = static_cast<int32_t>(BQS_STATUS_OK);
606 2 : QueueRoute *queueRouteList = qsRouteListPtr_;
607 8 : for (uint32_t i = 0U; i < qsRouterHeadPtr_->routeNum; ++i) {
608 6 : if (queueRouteList->status != static_cast<int32_t>(BQS_STATUS_OK)) {
609 3 : retCode_ = static_cast<int32_t>(BQS_STATUS_QUEUE_AHTU_ERROR);
610 3 : queueRouteList->status = 0;
611 3 : queueRouteList = queueRouteList + 1U;
612 3 : continue;
613 : }
614 : // only queue
615 : EntityInfo srcEntity = CreateBasicEntityInfo(queueRouteList->srcId,
616 3 : static_cast<dgw::EntityType>(queueRouteList->srcType));
617 : EntityInfo dstEntity = CreateBasicEntityInfo(queueRouteList->dstId,
618 3 : static_cast<dgw::EntityType>(queueRouteList->dstType));
619 3 : const auto result = relationInstance.Bind(srcEntity, dstEntity, index);
620 3 : if (result == BQS_STATUS_RETRY) {
621 0 : queueRouteList = queueRouteList + 1U;
622 0 : continue;
623 3 : } else if (result != BQS_STATUS_OK) {
624 0 : retCode_ = static_cast<int32_t>(result);
625 0 : queueRouteList->status = 0;
626 : } else {
627 3 : queueRouteList->status = 1;
628 : }
629 3 : queueRouteList = queueRouteList + 1U;
630 3 : BQS_LOG_RUN_INFO("Bind relation [add], stage [server:process], relation [src:%s, dst:%s, result:%d]",
631 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), static_cast<int32_t>(result));
632 3 : }
633 2 : relationInstance.Order(index);
634 2 : return;
635 : }
636 :
637 2 : void RouterServer::ProcessUnbindQueue(const uint32_t index)
638 : {
639 2 : BQS_LOG_INFO("[RouterServer]Unbind relation [del], stage [server:process].");
640 2 : QueueRoute *queueRouteList = qsRouteListPtr_;
641 2 : retCode_ = static_cast<int32_t>(BQS_STATUS_OK);
642 2 : auto &relationInstance = BindRelation::GetInstance();
643 8 : for (uint32_t i = 0U; i < qsRouterHeadPtr_->routeNum; ++i) {
644 6 : if (queueRouteList->status != static_cast<int32_t>(BQS_STATUS_OK)) {
645 3 : retCode_ = static_cast<int32_t>(BQS_STATUS_QUEUE_ID_ERROR);
646 3 : queueRouteList->status = 1;
647 3 : queueRouteList = queueRouteList + 1U;
648 3 : continue;
649 : }
650 : EntityInfo srcEntity = CreateBasicEntityInfo(queueRouteList->srcId,
651 3 : static_cast<dgw::EntityType>(queueRouteList->srcType));
652 : EntityInfo dstEntity = CreateBasicEntityInfo(queueRouteList->dstId,
653 3 : static_cast<dgw::EntityType>(queueRouteList->dstType));
654 3 : const auto result = relationInstance.UnBind(srcEntity, dstEntity, index);
655 3 : if (result == BQS_STATUS_RETRY) {
656 0 : queueRouteList = queueRouteList + 1U;
657 0 : continue;
658 3 : } else if (result != BQS_STATUS_OK) {
659 0 : retCode_ = static_cast<int32_t>(result);
660 0 : queueRouteList->status = 1;
661 : } else {
662 3 : queueRouteList->status = 0;
663 : }
664 3 : queueRouteList = queueRouteList + 1U;
665 3 : BQS_LOG_RUN_INFO("Bind relation [del], stage [server:process], relation [src %s," \
666 : "dst %s, result:%d]",
667 : srcEntity.ToString().c_str(), dstEntity.ToString().c_str(), static_cast<int32_t>(result));
668 3 : }
669 2 : relationInstance.Order(index);
670 2 : return;
671 : }
672 :
673 : /**
674 : * Bqs server enqueue bind msg request process
675 : * @return NA
676 : */
677 41 : void RouterServer::BindMsgProc(const uint32_t index)
678 : {
679 41 : BQS_LOG_INFO("[RouterServer]RouterServer BindMsgProc begin.");
680 41 : auto &processing = (index == 0U) ? processing_ : processingExtra_;
681 41 : auto &done = (index == 0U) ? done_ : doneExtra_;
682 :
683 41 : const std::unique_lock<std::mutex> lk(mutex_);
684 41 : processing = true;
685 :
686 : // parse bind and unbind BQSMsg
687 41 : if ((subEventId_ == static_cast<uint32_t>(AICPU_BIND_QUEUE)) ||
688 41 : (subEventId_ == static_cast<uint32_t>(ACL_BIND_QUEUE))) {
689 2 : ProcessBindQueue(index);
690 39 : } else if ((subEventId_ == static_cast<uint32_t>(AICPU_UNBIND_QUEUE)) ||
691 37 : (subEventId_ == static_cast<uint32_t>(ACL_UNBIND_QUEUE))) {
692 2 : ProcessUnbindQueue(index);
693 37 : } else if ((subEventId_ == static_cast<uint32_t>(QueueSubEventType::UPDATE_CONFIG))) {
694 37 : retCode_ = (cfgInfoOperator_ == nullptr) ? static_cast<int32_t>(BQS_STATUS_INNER_ERROR) :
695 37 : static_cast<int32_t>(cfgInfoOperator_->ProcessUpdateConfig(index));
696 37 : BQS_LOG_INFO("[RouterServer] Process update config ret is %d.", retCode_);
697 : } else {
698 0 : BQS_LOG_ERROR("[RouterServer]Invalid subEventId_[%d] in bind relation process.", subEventId_);
699 : }
700 :
701 41 : processing = false;
702 41 : done = true;
703 41 : cv_.notify_one();
704 :
705 41 : BQS_LOG_INFO("[RouterServer]RouterServer BindMsgProc end.");
706 82 : return;
707 41 : }
708 :
709 : /**
710 : * Init bqs server, including init easycomm server and bind relation
711 : * @return BQS_STATUS_OK:success other:failed
712 : */
713 5 : BqsStatus RouterServer::InitRouterServer(const InitQsParams ¶ms)
714 : {
715 5 : BQS_LOG_INFO("[RouterServer]RouterServer Init begin");
716 5 : (void)signal(SIGPIPE, SIG_IGN);
717 :
718 5 : qsInitGroupName_ = params.qsInitGrpName;
719 5 : f2nfGroupId_ = params.f2nfGroupId;
720 5 : schedPolicy_ = params.schedPolicy;
721 : // create config info operator
722 5 : cfgInfoOperator_.reset(new (std::nothrow) ConfigInfoOperator(params.deviceId, qsInitGroupName_));
723 5 : if (cfgInfoOperator_ == nullptr) {
724 0 : BQS_LOG_ERROR("malloc memory for cfgInfoOperator_ failed.");
725 0 : return BQS_STATUS_INNER_ERROR;
726 : }
727 5 : SubscribeBufEvent();
728 :
729 5 : if (!running_) {
730 5 : running_ = true;
731 5 : deviceId_ = params.deviceId;
732 5 : deployMode_ = params.runMode;
733 5 : numaFlag_ = params.numaFlag;
734 5 : needAttachGroup_ = params.needAttachGroup;
735 : // halShrIdGetAttribute为新版驱动中才存在的接口，若没有，说明驱动为老版本，需兼容处理
736 5 : if ((bqs::GetRunContext() == bqs::RunContext::HOST) && (&halShrIdGetAttribute == nullptr)) {
737 5 : compatMsg_ = true;
738 : }
739 5 : BQS_LOG_INFO("compatMsg_ is %d", compatMsg_);
740 :
741 : try {
742 5 : monitorQsEvent_ = std::thread(&RouterServer::ManageQsEvent, this);
743 0 : } catch(std::exception &threadException) {
744 0 : BQS_LOG_ERROR("RouterServer Init thread failure, %s", threadException.what());
745 0 : return BQS_STATUS_INNER_ERROR;
746 0 : }
747 :
748 5 : std::unique_lock<std::mutex> lk(manageThreadMutex_);
749 12 : manageThreadCv_.wait(lk, [this] { return manageThreadStatus_ != ThreadStatus::NOT_INIT; });
750 5 : if (manageThreadStatus_ != ThreadStatus::INIT_SUCCESS) {
751 2 : BQS_LOG_ERROR("RouterServer thread fail to start");
752 2 : return BQS_STATUS_INNER_ERROR;
753 : }
754 :
755 3 : BQS_LOG_INFO("[RouterServer]RouterServer Init success.");
756 5 : } else {
757 0 : BQS_LOG_WARN("RouterServer is already inited");
758 : }
759 3 : return BQS_STATUS_OK;
760 : }
761 :
762 8 : BqsStatus RouterServer::ParseRelationInfo(Mbuf **mbufPtr)
763 : {
764 8 : Mbuf *mBuf = nullptr;
765 8 : const auto drvRet = halQueueDeQueue(deviceId_, pipelineQueueId_, PtrToPtr<Mbuf *, void*>(&mBuf));
766 8 : if ((drvRet != DRV_ERROR_NONE) || (mBuf == nullptr)) {
767 2 : BQS_LOG_ERROR("[RouterServer]halQueueDeQueue from queue[%u] in device[%u] failed, error[%d]",
768 : pipelineQueueId_.load(), deviceId_, static_cast<int32_t>(drvRet));
769 1 : return BQS_STATUS_DRIVER_ERROR;
770 : }
771 7 : *mbufPtr = mBuf;
772 7 : qsRouterHeadPtr_ = nullptr;
773 7 : const auto getBuffRet = halMbufGetBuffAddr(mBuf, reinterpret_cast<void **>(&qsRouterHeadPtr_));
774 7 : if ((getBuffRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (qsRouterHeadPtr_ == nullptr)) {
775 0 : BQS_LOG_ERROR("[RouterServer]halMbufGetBuffAddr from queue[%u] in device[%u] failed, error[%d]",
776 : pipelineQueueId_.load(), deviceId_, getBuffRet);
777 0 : return BQS_STATUS_DRIVER_ERROR;
778 : }
779 7 : if (isAicpuEvent_) {
780 5 : aicpuRspHead_ = qsRouterHeadPtr_->userData;
781 : }
782 : // aicpuRspHead_ is valid only in aicpu event senario, will be 0 in acl event senario
783 7 : subEventId_ = qsRouterHeadPtr_->subEventId;
784 7 : BQS_LOG_INFO("[RouterServer]Parse head[%lu] subEvnetId[%u] from mbuff success.", aicpuRspHead_, subEventId_);
785 :
786 : // query message need to get query info
787 7 : if ((subEventId_ == static_cast<uint32_t>(AICPU_QUERY_QUEUE)) ||
788 3 : (subEventId_ == static_cast<uint32_t>(ACL_QUERY_QUEUE))) {
789 4 : if ((((qsRouterHeadPtr_->routeNum * sizeof(QueueRoute)) + sizeof(QsRouteHead)) + sizeof(QueueRouteQuery)) !=
790 4 : qsRouterHeadPtr_->length) {
791 0 : BQS_LOG_ERROR("[RouterServer]RouteNum[%d] is inconsistence with dataLen[%d] in subEventId[%u]",
792 : qsRouterHeadPtr_->routeNum, qsRouterHeadPtr_->length, subEventId_);
793 0 : return BQS_STATUS_PARAM_INVALID;
794 : }
795 4 : qsRouterQueryPtr_ = reinterpret_cast<QueueRouteQuery *>(
796 4 : reinterpret_cast<uint8_t *>(qsRouterHeadPtr_) + sizeof(QsRouteHead));
797 4 : BQS_LOG_INFO("[RouterServer]Get query info success. queryType[%d]", qsRouterQueryPtr_->queryType);
798 4 : qsRouteListPtr_ = reinterpret_cast<QueueRoute *>(
799 4 : reinterpret_cast<uint8_t *>(qsRouterQueryPtr_) + sizeof(QueueRouteQuery));
800 : } else {
801 3 : if (((qsRouterHeadPtr_->routeNum * sizeof(QueueRoute)) + sizeof(QsRouteHead)) != qsRouterHeadPtr_->length) {
802 0 : BQS_LOG_ERROR("[RouterServer]RouteNum[%d] is inconsistence with dataLen[%d] in subEventId[%u]",
803 : qsRouterHeadPtr_->routeNum, qsRouterHeadPtr_->length, subEventId_);
804 0 : return BQS_STATUS_PARAM_INVALID;
805 : }
806 3 : qsRouterQueryPtr_ = nullptr;
807 3 : qsRouteListPtr_ = reinterpret_cast<QueueRoute *>(
808 3 : reinterpret_cast<uint8_t *>(qsRouterHeadPtr_) + sizeof(QsRouteHead));
809 : }
810 7 : BQS_LOG_INFO("[RouterServer]Get relation mbuff success, bind/unbind queue num[%d]", qsRouterHeadPtr_->routeNum);
811 7 : return BQS_STATUS_OK;
812 : }
813 :
814 3 : BqsStatus RouterServer::ParseBindUnbindMsg() const
815 : {
816 3 : BQS_LOG_INFO("[RouterServer]Bind relation [add/del], stage [server:parse and check]");
817 3 : auto resultCode = BQS_STATUS_QUEUE_AHTU_ERROR;
818 3 : QueueRoute *queueRouteList = qsRouteListPtr_;
819 12 : for (uint32_t i = 0U; i < qsRouterHeadPtr_->routeNum; ++i) {
820 : EntityInfo srcEntity = CreateBasicEntityInfo(queueRouteList->srcId,
821 9 : static_cast<dgw::EntityType>(queueRouteList->srcType));
822 : EntityInfo dstEntity = CreateBasicEntityInfo(queueRouteList->dstId,
823 9 : static_cast<dgw::EntityType>(queueRouteList->dstType));
824 9 : BQS_LOG_INFO("[RouterServer]Src[id:%u type:%d] Dst[id:%u type:%d]",
825 : srcEntity.GetId(), static_cast<int32_t>(srcEntity.GetType()),
826 : dstEntity.GetId(), static_cast<int32_t>(dstEntity.GetType()));
827 9 : if ((srcEntity.GetId() >= MAX_QUEUE_ID_NUM) || (dstEntity.GetId() >= MAX_QUEUE_ID_NUM)) {
828 5 : BQS_LOG_ERROR("[RouterServer]Src[%s] or Dst[%s] is invalid in this "
829 : "bind/unbind relation", srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
830 5 : queueRouteList->status = static_cast<int32_t>(BQS_STATUS_QUEUE_ID_ERROR);
831 5 : queueRouteList = queueRouteList + 1;
832 5 : continue;
833 : }
834 : // preprocess: do attach queue and check src own read auth, dst own write auth
835 4 : if ((subEventId_ == static_cast<uint32_t>(AICPU_BIND_QUEUE)) ||
836 4 : (subEventId_ == static_cast<uint32_t>(ACL_BIND_QUEUE))) {
837 4 : const auto ret = (cfgInfoOperator_ == nullptr) ? BQS_STATUS_INNER_ERROR :
838 0 : cfgInfoOperator_->AttachAndCheckQueue(srcEntity, dstEntity);
839 4 : if (ret != BQS_STATUS_OK) {
840 4 : BQS_LOG_ERROR("[RouterServer]Src[%s] Dst[%s] do attach queue "
841 : "and check auth failed", srcEntity.ToString().c_str(), dstEntity.ToString().c_str());
842 4 : queueRouteList->status = static_cast<int32_t>(BQS_STATUS_QUEUE_AHTU_ERROR);
843 4 : queueRouteList = queueRouteList + 1;
844 4 : continue;
845 : }
846 : }
847 0 : resultCode = BQS_STATUS_OK;
848 0 : queueRouteList->status = static_cast<int32_t>(BQS_STATUS_OK);
849 0 : queueRouteList = queueRouteList + 1;
850 18 : }
851 3 : BQS_LOG_INFO("[RouterServer]Finish parse bind/unbind message,resultCode[%d]", static_cast<int32_t>(resultCode));
852 3 : return resultCode;
853 : }
854 :
855 14 : void RouterServer::FillRoutes(const EntityInfo &src, const EntityInfo &dst, const BindRelationStatus status)
856 : {
857 14 : QueueRoute queueRouteInfo = {};
858 14 : queueRouteInfo.srcId = src.GetId();
859 14 : queueRouteInfo.dstId = dst.GetId();
860 14 : queueRouteInfo.srcType = static_cast<int16_t>(src.GetType());
861 14 : queueRouteInfo.dstType = static_cast<int16_t>(dst.GetType());
862 14 : queueRouteInfo.status = static_cast<int32_t>(status);
863 14 : queueRouteQueryList_.emplace_back(queueRouteInfo);
864 14 : }
865 :
866 28 : void RouterServer::SearchRelation(const MapEnitityInfoToInfoSet &relationMap, const EntityInfo& entityInfo,
867 : const BindRelationStatus status, bool bySrc)
868 : {
869 28 : const auto iter = relationMap.find(entityInfo);
870 28 : if (iter == relationMap.end()) {
871 21 : return;
872 : }
873 12 : const auto dstSet = iter->second;
874 12 : BQS_LOG_INFO("[RouterServer]Bind relation [get], stage [server:process], relation [size:%zu].", dstSet.size());
875 12 : if (bySrc) {
876 12 : for (auto setIter = dstSet.begin(); setIter != dstSet.end(); ++setIter) {
877 7 : FillRoutes(entityInfo, *setIter, status);
878 : }
879 5 : return;
880 : }
881 :
882 14 : for (auto setIter = dstSet.begin(); setIter != dstSet.end(); ++setIter) {
883 7 : FillRoutes(*setIter, entityInfo, status);
884 : }
885 12 : }
886 :
887 : /**
888 : * Assembly response of get bind message according to src entity
889 : * @return Number of query results
890 : */
891 13 : void RouterServer::GetBindRspBySingle(const EntityInfo& entityInfo, const uint32_t &queryType)
892 : {
893 13 : BQS_LOG_INFO("[RouterServer]RouterServer serialize get bind rsponse by entityId[%u], entityType[%d], Type[%d].",
894 : entityInfo.GetId(), static_cast<int32_t>(entityInfo.GetType()), queryType);
895 13 : auto &relationInstance = BindRelation::GetInstance();
896 13 : queueRouteQueryList_.clear();
897 13 : if ((queryType != static_cast<uint32_t>(BQS_QUERY_TYPE_SRC)) &&
898 7 : (queryType != static_cast<uint32_t>(BQS_QUERY_TYPE_DST))) {
899 1 : BQS_LOG_ERROR("[RouterServer]QueryType[%d] is not supported.", queryType);
900 1 : return;
901 : }
902 :
903 12 : if (queryType == static_cast<uint32_t>(BQS_QUERY_TYPE_SRC)) {
904 6 : SearchRelation(relationInstance.GetSrcToDstRelation(), entityInfo, BindRelationStatus::RelationBind, true);
905 6 : if (numaFlag_) {
906 2 : SearchRelation(relationInstance.GetSrcToDstExtraRelation(), entityInfo,
907 : BindRelationStatus::RelationBind, true);
908 : }
909 6 : SearchRelation(relationInstance.GetAbnormalSrcToDstRelation(), entityInfo,
910 : BindRelationStatus::RelationAbnormalForQError, true);
911 :
912 6 : if (queueRouteQueryList_.empty()) {
913 2 : BQS_LOG_WARN("[RouterServer] record does not exist according to src entityId:[%u]", entityInfo.GetId());
914 : }
915 : } else {
916 6 : SearchRelation(relationInstance.GetDstToSrcRelation(), entityInfo, BindRelationStatus::RelationBind, false);
917 6 : if (numaFlag_) {
918 2 : SearchRelation(relationInstance.GetDstToSrcExtraRelation(), entityInfo,
919 : BindRelationStatus::RelationBind, false);
920 : }
921 6 : SearchRelation(relationInstance.GetAbnormalDstToSrcRelation(), entityInfo,
922 : BindRelationStatus::RelationAbnormalForQError, false);
923 : }
924 12 : if (queueRouteQueryList_.empty()) {
925 2 : BQS_LOG_WARN("RouterServer get relation according to dst:%u failed, record does not exist", entityInfo.GetId());
926 : }
927 : }
928 :
929 5 : bool RouterServer::FindRelation(const MapEnitityInfoToInfoSet &relationMap, const EntityInfo& srcInfo,
930 : const EntityInfo& dstInfo) const
931 : {
932 5 : const auto srcIter = relationMap.find(srcInfo);
933 5 : return ((srcIter != relationMap.end()) && (srcIter->second.count(dstInfo) != 0UL));
934 : }
935 :
936 4 : void RouterServer::TransRouteWithEntityInfo(const EntityInfo& srcInfo, const EntityInfo& dstInfo, const int32_t status,
937 : QueueRoute &routeInfo) const
938 : {
939 4 : routeInfo.srcId = srcInfo.GetId();
940 4 : routeInfo.dstId = dstInfo.GetId();
941 4 : routeInfo.srcType = static_cast<int16_t>(srcInfo.GetType());
942 4 : routeInfo.dstType = static_cast<int16_t>(dstInfo.GetType());
943 4 : routeInfo.status = static_cast<int32_t>(status);
944 4 : }
945 :
946 : /**
947 : * Assembly response of get bind message according to dst queueId, one-to-one relation
948 : * @return NA
949 : */
950 5 : void RouterServer::GetBindRspByDouble(const EntityInfo& src, const EntityInfo& dst, const uint32_t &queryType)
951 : {
952 5 : BQS_LOG_INFO("[RouterServer]RouterServer serialize get bind rsponse by srcId[%u], srcType[%d], dstId[%u], "
953 : "dstType[%d], Type[%u]",
954 : src.GetId(), static_cast<int32_t>(src.GetType()), dst.GetId(), static_cast<int32_t>(dst.GetType()), queryType);
955 5 : queueRouteQueryList_.clear();
956 5 : if (queryType == static_cast<uint32_t>(BQS_QUERY_TYPE_SRC_OR_DST)) {
957 2 : GetBindRspBySingle(src, static_cast<uint32_t>(BQS_QUERY_TYPE_SRC));
958 2 : if (queueRouteQueryList_.size() > 0UL) {
959 0 : return;
960 : }
961 2 : GetBindRspBySingle(dst, static_cast<uint32_t>(BQS_QUERY_TYPE_DST));
962 2 : return;
963 : }
964 3 : if (queryType == static_cast<uint32_t>(BQS_QUERY_TYPE_SRC_AND_DST)) {
965 3 : auto status = BindRelationStatus::RelationUnknown;
966 4 : if (FindRelation(BindRelation::GetInstance().GetSrcToDstRelation(), src, dst) ||
967 1 : (numaFlag_ && (FindRelation(BindRelation::GetInstance().GetSrcToDstExtraRelation(), src, dst)))) {
968 2 : status = BindRelationStatus::RelationBind;
969 : } else {
970 1 : if (FindRelation(BindRelation::GetInstance().GetAbnormalSrcToDstRelation(), src, dst)) {
971 1 : status = BindRelationStatus::RelationAbnormalForQError;
972 : }
973 : }
974 :
975 3 : if (status != BindRelationStatus::RelationUnknown) {
976 3 : QueueRoute queueRouteInfo = {};
977 3 : TransRouteWithEntityInfo(src, dst, static_cast<int32_t>(status), queueRouteInfo);
978 3 : queueRouteQueryList_.emplace_back(queueRouteInfo);
979 : }
980 3 : return;
981 : }
982 0 : BQS_LOG_ERROR("[RouterServer]QueryType[%d] is not supported.", queryType);
983 0 : return;
984 : }
985 :
986 1 : void RouterServer::GetAllAbnormalBind()
987 : {
988 1 : BQS_LOG_INFO("[RouterServer]RouterServer serialize get all abnormal bind rsponse");
989 1 : queueRouteQueryList_.clear();
990 1 : auto &relationInstance = BindRelation::GetInstance();
991 1 : auto &abnormalSrcToDstRelation = relationInstance.GetAbnormalSrcToDstRelation();
992 2 : for (auto iter = abnormalSrcToDstRelation.begin(); iter != abnormalSrcToDstRelation.end(); ++iter) {
993 1 : const auto &src = iter->first;
994 1 : const auto &dstSet = iter->second;
995 2 : for (auto &dst : dstSet) {
996 1 : QueueRoute queueRouteInfo = {};
997 1 : TransRouteWithEntityInfo(src, dst, static_cast<int32_t>(BindRelationStatus::RelationAbnormalForQError),
998 : queueRouteInfo);
999 1 : queueRouteQueryList_.emplace_back(queueRouteInfo);
1000 : }
1001 : }
1002 1 : }
1003 :
1004 8 : BqsStatus RouterServer::ProcessGetBindMsg(const uint32_t &queryType, const EntityInfo& src, const EntityInfo& dst)
1005 : {
1006 8 : switch (queryType) {
1007 2 : case BQS_QUERY_TYPE_SRC:
1008 2 : GetBindRspBySingle(src, queryType);
1009 2 : break;
1010 2 : case BQS_QUERY_TYPE_DST:
1011 2 : GetBindRspBySingle(dst, queryType);
1012 2 : break;
1013 4 : case BQS_QUERY_TYPE_SRC_OR_DST:
1014 : case BQS_QUERY_TYPE_SRC_AND_DST:
1015 4 : GetBindRspByDouble(src, dst, queryType);
1016 4 : break;
1017 0 : case BQS_QUERY_TYPE_ABNORMAL_FOR_QUEUE_ERROR:
1018 0 : GetAllAbnormalBind();
1019 0 : break;
1020 0 : default:
1021 0 : BQS_LOG_ERROR("[RouterServer]Unsupported query type"
1022 : "{0:src, 1:dst, 2:src-or-dst, 3:src-and-dst, 100:abnormal-all}:%u", queryType);
1023 0 : break;
1024 : }
1025 :
1026 8 : if ((subEventId_ == static_cast<uint32_t>(AICPU_QUERY_QUEUE_NUM)) ||
1027 4 : (subEventId_ == static_cast<uint32_t>(ACL_QUERY_QUEUE_NUM))) {
1028 4 : return BQS_STATUS_OK;
1029 : }
1030 :
1031 4 : if (queueRouteQueryList_.size() != qsRouterHeadPtr_->routeNum) {
1032 0 : BQS_LOG_ERROR("[RouterServer]Prepare number[%d] is different with real route number[%zu].",
1033 : qsRouterHeadPtr_->routeNum, queueRouteQueryList_.size());
1034 0 : return BQS_STATUS_PARAM_INVALID;
1035 : }
1036 9 : for (size_t i = 0UL; i < qsRouterHeadPtr_->routeNum; i++) {
1037 5 : *qsRouteListPtr_ = queueRouteQueryList_[i];
1038 5 : BQS_LOG_INFO("[RouterServer]Query bind relation srcId[%u] srcType[%d] dstId[%u] dstType[%d]",
1039 : qsRouteListPtr_->srcId, static_cast<int32_t>(qsRouteListPtr_->srcType), qsRouteListPtr_->dstId,
1040 : static_cast<int32_t>(qsRouteListPtr_->dstType));
1041 5 : qsRouteListPtr_ = qsRouteListPtr_ + 1;
1042 : }
1043 4 : queueRouteQueryList_.clear();
1044 4 : return BQS_STATUS_OK;
1045 : }
1046 :
1047 4 : void RouterServer::ParseGetBindNumMsg(const event_info &info)
1048 : {
1049 4 : BQS_LOG_INFO("[RouterServer]Bind relation number [get], stage [server:process], type [request]");
1050 4 : if (info.priv.msg_len != sizeof(QueueRouteQuery)) {
1051 0 : BQS_LOG_ERROR("[RouterServer]Query event[%u] message invalid, msgLen[%u]", subEventId_, info.priv.msg_len);
1052 0 : SendRspEvent(static_cast<int32_t>(BQS_STATUS_PARAM_INVALID));
1053 0 : return;
1054 : }
1055 4 : const QueueRouteQuery * const queueRouteQuery = PtrToPtr<const char_t, const QueueRouteQuery>(info.priv.msg);
1056 : // param syncEventHead is filled by drv when acl (call sync event interface)
1057 4 : aicpuRspHead_ = isAicpuEvent_ ? queueRouteQuery->syncEventHead : 0UL;
1058 4 : const uint32_t keyType = queueRouteQuery->queryType;
1059 4 : const EntityInfo src = CreateBasicEntityInfo(queueRouteQuery->srcId,
1060 4 : static_cast<dgw::EntityType>(queueRouteQuery->srcType));
1061 4 : const EntityInfo dst = CreateBasicEntityInfo(queueRouteQuery->dstId,
1062 4 : static_cast<dgw::EntityType>(queueRouteQuery->dstType));
1063 4 : queueRouteQueryList_.clear();
1064 4 : const auto ret = ProcessGetBindMsg(keyType, src, dst);
1065 4 : SendRspEvent(static_cast<int32_t>(ret));
1066 4 : return;
1067 4 : }
1068 :
1069 5 : BqsStatus RouterServer::ParseGetBindDetailMsg()
1070 : {
1071 5 : BQS_LOG_INFO("[RouterServer]Bind relation [get], stage [server:process], type [request]");
1072 5 : if (qsRouterQueryPtr_ == nullptr) {
1073 1 : BQS_LOG_ERROR("[RouterServer]qsRouterQuery should not be null pointer");
1074 1 : return BQS_STATUS_INNER_ERROR;
1075 : }
1076 4 : const uint32_t keyType = qsRouterQueryPtr_->queryType;
1077 4 : const EntityInfo src = CreateBasicEntityInfo(qsRouterQueryPtr_->srcId,
1078 4 : static_cast<dgw::EntityType>(qsRouterQueryPtr_->srcType));
1079 4 : const EntityInfo dst = CreateBasicEntityInfo(qsRouterQueryPtr_->dstId,
1080 4 : static_cast<dgw::EntityType>(qsRouterQueryPtr_->dstType));
1081 4 : queueRouteQueryList_.clear();
1082 4 : const auto ret = ProcessGetBindMsg(keyType, src, dst);
1083 4 : return ret;
1084 4 : }
1085 :
1086 11 : ThreadStatus RouterServer::PrePareForManageThread()
1087 : {
1088 11 : auto ret = halEschedAttachDevice(deviceId_);
1089 11 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_PROCESS_REPEAT_ADD)) {
1090 1 : BQS_LOG_ERROR("Failed to attach device[%u] for eSched, result[%d].", deviceId_, static_cast<int32_t>(ret));
1091 1 : (void) AttachGroup();
1092 1 : return ThreadStatus::INIT_FAIL;
1093 : }
1094 10 : ret = halEschedCreateGrp(deviceId_, bindQueueGroupId_, GRP_TYPE_BIND_CP_CPU);
1095 10 : if (ret != DRV_ERROR_NONE) {
1096 1 : (void)halEschedDettachDevice(deviceId_);
1097 1 : BQS_LOG_ERROR("Failed to create bindQueueGroup, groupId[%u] result[%d].",
1098 : bindQueueGroupId_, static_cast<int32_t>(ret));
1099 1 : (void) AttachGroup();
1100 1 : return ThreadStatus::INIT_FAIL;
1101 : }
1102 : // subsribe bind/unbind/query event
1103 9 : const uint64_t eventBitmap = (1UL << static_cast<uint32_t>(EVENT_QS_MSG));
1104 9 : BQS_LOG_INFO("[RouterServer]BindQueue group[%u] subscribe event, eventBitmap[%lu]",
1105 : bindQueueGroupId_, eventBitmap);
1106 9 : ret = halEschedSubscribeEvent(deviceId_, bindQueueGroupId_, 0U, eventBitmap);
1107 9 : if (ret != DRV_ERROR_NONE) {
1108 1 : BQS_LOG_ERROR("[RouterServer]halEschedSubscribeEvent failed, groupId[%u] eventBitmap[%lu] result[%d].",
1109 : bindQueueGroupId_, eventBitmap, static_cast<int32_t>(ret));
1110 1 : (void) AttachGroup();
1111 1 : return ThreadStatus::INIT_FAIL;
1112 : }
1113 :
1114 8 : if (!AttachGroup()) {
1115 2 : BQS_LOG_ERROR("[RouterServer] Fail to attach group");
1116 2 : return ThreadStatus::INIT_FAIL;
1117 : }
1118 6 : BQS_LOG_RUN_INFO("[RouterServer] ManageQsEvent of RouterServer is ready.");
1119 6 : return ThreadStatus::INIT_SUCCESS;
1120 : }
1121 :
1122 11 : void RouterServer::ManageQsEvent()
1123 : {
1124 11 : BQS_LOG_INFO("[RouterServer] Manage QS event of router server thread start");
1125 11 : (void)pthread_setname_np(pthread_self(), ROUTER_SERVER_THREAD_NAME_PREFIX);
1126 :
1127 11 : if (bqs::GetRunContext() != bqs::RunContext::HOST) {
1128 1 : const std::vector<uint32_t> &cpuIds = QueueScheduleInterface::GetInstance().GetCtrlCpuIds();
1129 : // bind thread to ctrl cpu
1130 1 : const pthread_t threadId = pthread_self();
1131 1 : (void)BindCpuUtils::SetThreadAffinity(threadId, cpuIds);
1132 : }
1133 :
1134 : {
1135 11 : std::unique_lock<std::mutex> lk(manageThreadMutex_);
1136 11 : manageThreadStatus_ = PrePareForManageThread();
1137 11 : manageThreadCv_.notify_all();
1138 11 : if (manageThreadStatus_ != ThreadStatus::INIT_SUCCESS) {
1139 5 : return;
1140 : }
1141 11 : }
1142 :
1143 6 : struct event_info event = {};
1144 : // default wait timeout 2s
1145 6 : constexpr int32_t waitTimeout = 2000;
1146 10 : while (running_) {
1147 5 : const auto schedRet = halEschedWaitEvent(deviceId_, bindQueueGroupId_, 0U, waitTimeout, &event);
1148 5 : if (schedRet == DRV_ERROR_NONE) {
1149 2 : if (event.comm.event_id == EVENT_QS_MSG) {
1150 1 : HandleBqsMsg(event);
1151 : } else {
1152 1 : BQS_LOG_WARN("Thread[%u] process unsupported eventId[%d] subEventId[%u].",
1153 : 0, static_cast<int32_t>(event.comm.event_id), event.comm.subevent_id);
1154 : }
1155 3 : } else if (schedRet == DRV_ERROR_SCHED_WAIT_TIMEOUT) {
1156 1 : BQS_LOG_DEBUG("ManageQsEvent bind/unbind/query event waiting timeout");
1157 1 : continue;
1158 2 : } else if (schedRet == DRV_ERROR_PARA_ERROR) {
1159 1 : BQS_LOG_ERROR(
1160 : "ManageQsEvent bind/unbind/query event failed, deviceId[%u] groupId[%u] error[%d].",
1161 : deviceId_, bindQueueGroupId_, static_cast<int32_t>(schedRet));
1162 1 : break;
1163 : } else {
1164 : // LOG ERROR
1165 1 : BQS_LOG_ERROR(
1166 : "ManageQsEvent bind/unbind/query event failed, deviceId[%u] groupId[%u] error[%d].",
1167 : deviceId_, bindQueueGroupId_, static_cast<int32_t>(schedRet));
1168 : }
1169 : }
1170 6 : BQS_LOG_INFO("[RouterServer] ManageQsEvent of RouterServer thread exit.");
1171 : }
1172 :
1173 6 : BqsStatus RouterServer::SubscribeBufEvent() const
1174 : {
1175 6 : BQS_LOG_INFO("[RouterServer] SubscribeBufEvent start.");
1176 6 : const bool needSubBufEvent =
1177 6 : static_cast<bool>(schedPolicy_ & static_cast<uint64_t>(SchedPolicy::POLICY_SUB_BUF_EVENT));
1178 6 : if (!needSubBufEvent) {
1179 5 : BQS_LOG_INFO("[RouterServer] needSubBufEvent is [%d]", static_cast<int32_t>(needSubBufEvent));
1180 5 : return BQS_STATUS_OK;
1181 : }
1182 : // load hccl so
1183 1 : dgw::HcclSoManager::GetInstance()->LoadSo();
1184 1 : return BQS_STATUS_OK;
1185 : }
1186 :
1187 3 : BqsStatus RouterServer::WaitSyncMsgProc()
1188 : {
1189 3 : std::unique_lock<std::mutex> lk(mutex_);
1190 :
1191 : // waiting for aicpu thread processing
1192 3 : done_ = false;
1193 : // produce enqueue event
1194 3 : auto ret = QueueManager::GetInstance().EnqueueRelationEvent();
1195 3 : if (ret != BQS_STATUS_OK) {
1196 1 : return ret;
1197 : }
1198 :
1199 : // if numa_flag , produce extra enqueue event
1200 2 : if (numaFlag_) {
1201 2 : doneExtra_ = false;
1202 2 : auto result = QueueManager::GetInstance().EnqueueRelationEventExtra();
1203 2 : if (result != BQS_STATUS_OK) {
1204 1 : return result;
1205 : }
1206 : }
1207 :
1208 1 : BQS_LOG_INFO("[RouterServer] update config[add/del group, bind/unbind route], stage [server:waiting]");
1209 3 : (void)cv_.wait_for(lk, std::chrono::milliseconds(MAX_WAITING_NOTIFY), [this] { return done_ && doneExtra_; });
1210 1 : while ((!done_ || !doneExtra_) && (processing_ || processingExtra_)) {
1211 0 : cv_.wait(lk);
1212 : }
1213 1 : if (!done_ || !doneExtra_) {
1214 1 : QueueManager::GetInstance().LogErrorRelationQueueStatus();
1215 1 : BQS_LOG_ERROR("[RouterServer] update config[add/del group, bind/unbind route], stage [server:wait], timeout, "
1216 : "relation queue[enqueue cnt:%lu, dequeue cnt:%lu].",
1217 : StatisticManager::GetInstance().GetRelationEnqueCnt(),
1218 : StatisticManager::GetInstance().GetRelationDequeCnt());
1219 1 : return BQS_STATUS_TIMEOUT;
1220 : }
1221 : // get config process result
1222 0 : ret = static_cast<BqsStatus>(retCode_);
1223 : // rest retCode_ and updateCfgInfo_
1224 0 : retCode_ = static_cast<int32_t>(BQS_STATUS_OK);
1225 0 : return ret;
1226 3 : }
1227 :
1228 10 : void RouterServer::NotifyInitSuccess()
1229 : {
1230 10 : BQS_LOG_RUN_INFO("schedule finish initing, now router is ready to handle msg");
1231 10 : readyToHandleMsg_ = true;
1232 10 : }
1233 :
1234 11 : bool RouterServer::AttachGroup()
1235 : {
1236 11 : if (!needAttachGroup_) {
1237 9 : return true;
1238 : }
1239 2 : BQS_LOG_INFO("Begin to attach group");
1240 2 : std::stringstream grpNameStream(qsInitGroupName_);
1241 2 : std::string grpNameElement;
1242 2 : std::vector<std::string> groupNameVec;
1243 4 : while (getline(grpNameStream, grpNameElement, ',')) {
1244 2 : groupNameVec.emplace_back(grpNameElement);
1245 : }
1246 :
1247 2 : const int32_t halTimeOut = (RunContext::HOST == GetRunContext()) ? 3000 : -1;
1248 2 : for (const auto &grpName : groupNameVec) {
1249 2 : BQS_LOG_RUN_INFO("Begin to halGrpAttach group[%s].", grpName.c_str());
1250 2 : const auto drvRet = halGrpAttach(grpName.c_str(), halTimeOut);
1251 2 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
1252 2 : BQS_LOG_ERROR("halGrpAttach group[%s] failed. ret[%d]", grpName.c_str(), drvRet);
1253 2 : return false;
1254 : }
1255 0 : BQS_LOG_RUN_INFO("halGrpAttach group[%s] success.", grpName.c_str());
1256 : }
1257 0 : return true;
1258 2 : }
1259 :
1260 46 : EntityInfo RouterServer::CreateBasicEntityInfo(const uint32_t id, const dgw::EntityType eType) const
1261 : {
1262 46 : OptionalArg args = {};
1263 46 : args.eType = eType;
1264 92 : return EntityInfo(id, deviceId_, &args);
1265 : }
1266 :
1267 : } // namespace bqs
|
