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 "server/bqs_server.h"
12 :
13 : #include <csignal>
14 : #include <algorithm>
15 : #include <securec.h>
16 : #include "easy_comm.h"
17 : #include "driver/ascend_hal.h"
18 :
19 : #include "queue_manager.h"
20 : #include "statistic_manager.h"
21 : #include "router_server.h"
22 : #include "common/bqs_log.h"
23 : #include "aicpu_sched/common/type_def.h"
24 : namespace bqs {
25 : namespace {
26 : // prevents concurrent execution of multiple clients
27 : std::mutex g_bqsMutex;
28 :
29 : constexpr const char_t *BQS_SERVER_THREAD_NAME_PREFIX = "bqs_server";
30 :
31 : /**
32 : * Message process function, need to send a response to avoid blocking
33 : * @return NA
34 : */
35 7 : void RpcHandler(const int32_t fd, EzcomRequest * const req)
36 : {
37 7 : if (req == nullptr) {
38 1 : BQS_LOG_RUN_INFO("Pipe of client has been closed, fd:%d.", fd);
39 1 : (void)EzcomClosePipe(fd);
40 1 : return;
41 : }
42 :
43 6 : const std::unique_lock<std::mutex> lk(g_bqsMutex);
44 6 : BQS_LOG_INFO("BqsServer receive a request, id = %u, msg_size = %u", req->id, req->size);
45 6 : BqsServer::GetInstance().HandleBqsReqMsg(req->id, reinterpret_cast<const char_t *>(req->data), req->size);
46 : // send response
47 6 : BqsServer::GetInstance().SendRspMsg(fd, req->id);
48 6 : BQS_LOG_INFO("BqsServer HandleBqsReqMsg a request success, id = %u, msg_size = %u", req->id, req->size);
49 6 : return;
50 6 : }
51 :
52 0 : void NodeHandlerWrapper(const int32_t fd, const char_t * const clientName, const int32_t nameLen)
53 : {
54 : (void)fd;
55 0 : if ((clientName == nullptr) || (nameLen <= 0)) {
56 0 : BQS_LOG_ERROR("Client name is nullptr");
57 0 : return;
58 : }
59 0 : (void)pthread_setname_np(pthread_self(), BQS_SERVER_THREAD_NAME_PREFIX);
60 : }
61 : } // namespace
62 :
63 1 : BqsServer::BqsServer() : msgId_(0U), processing_(false), done_(false)
64 1 : {}
65 :
66 1 : BqsServer::~BqsServer()
67 1 : {}
68 :
69 134 : BqsServer &BqsServer::GetInstance()
70 : {
71 134 : static BqsServer instance;
72 134 : return instance;
73 : }
74 :
75 5 : void BqsServer::InitBuff() const
76 : {
77 5 : BuffCfg defaultCfg = {};
78 5 : const int32_t drvRet = halBuffInit(&defaultCfg);
79 5 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
80 0 : BQS_LOG_ERROR("[BqsServer]Buffer initial failed ret[%d]", drvRet);
81 0 : return;
82 : }
83 5 : BQS_LOG_INFO("[RouterServer] Buffer init success ret = %d", drvRet);
84 : }
85 :
86 : /**
87 : * Bqs server handle BqsMsg, get/getall deal now, bind/unbind send to work thread to deal
88 : * @return NA
89 : */
90 5 : void BqsServer::HandleBqsReqMsg(const uint32_t msgId, const char_t * const data, const uint32_t dataSize)
91 : {
92 5 : BQS_LOG_INFO("Bind relation, stage [server:receive], type [request], msg [id = %u]", msgId);
93 5 : msgId_ = msgId;
94 5 : bqsRespMsg_.Clear(); // init response msg
95 5 : if (data == nullptr) {
96 0 : BQS_LOG_ERROR("Request of BqsClient is nullptr.");
97 0 : return;
98 : }
99 5 : if (dataSize < BQS_MSG_HEAD_SIZE) {
100 0 : BQS_LOG_ERROR("Request of BqsClient size:%u should be not less than head:%u.", dataSize, BQS_MSG_HEAD_SIZE);
101 0 : return;
102 : }
103 5 : InitBuff();
104 5 : const uint32_t currMsgSize = *(PtrToPtr<const char_t, const uint32_t>(data));
105 5 : if (currMsgSize != dataSize) {
106 1 : BQS_LOG_ERROR("message error, head_msg_content = %u, request_size = %u", currMsgSize, dataSize);
107 1 : return;
108 : }
109 4 : const uint32_t parseLength = currMsgSize - BQS_MSG_HEAD_SIZE;
110 4 : if (bqsReqMsg_.ParseFromArray(data + BQS_MSG_HEAD_SIZE, static_cast<int32_t>(parseLength))) {
111 4 : BQS_LOG_INFO("BqsServer request msg type{%d:BIND, %d:UNBIND, %d:GET_BIND, %d:GET_ALL_BIND}:%d "
112 : "begin to process",
113 : BQSMsg::BIND, BQSMsg::UNBIND, BQSMsg::GET_BIND, BQSMsg::GET_ALL_BIND,
114 : bqsReqMsg_.msg_type());
115 4 : switch (bqsReqMsg_.msg_type()) {
116 1 : case BQSMsg::GET_BIND:
117 1 : StatisticManager::GetInstance().GetBindStat();
118 1 : ParseGetBindMsg(bqsReqMsg_, bqsRespMsg_);
119 1 : break;
120 1 : case BQSMsg::GET_ALL_BIND:
121 1 : StatisticManager::GetInstance().GetAllBindStat();
122 1 : ParseGetPagedBindMsg(bqsReqMsg_, bqsRespMsg_);
123 1 : break;
124 1 : case BQSMsg::BIND:
125 1 : StatisticManager::GetInstance().BindStat();
126 1 : WaitBindMsgProc();
127 1 : break;
128 0 : case BQSMsg::UNBIND:
129 0 : StatisticManager::GetInstance().UnbindStat();
130 0 : WaitBindMsgProc();
131 0 : break;
132 1 : default:
133 1 : BQS_LOG_ERROR("BqsServer receive unsupported msg type:%d", bqsReqMsg_.msg_type());
134 1 : break;
135 : }
136 : }
137 4 : BQS_LOG_INFO("BqsServer HandleBqsMsg end");
138 4 : return;
139 : }
140 :
141 : /**
142 : * Bqs server wait work thread to process msg
143 : * @return NA
144 : */
145 0 : void BqsServer::WaitBindMsgProc()
146 : {
147 0 : BQS_LOG_INFO("Bind relation [add/del], stage [server:enqueue], type [request], msg [id = %u]", msgId_);
148 0 : std::unique_lock<std::mutex> bqsLock(mutex_);
149 0 : const BqsStatus ret = QueueManager::GetInstance().EnqueueRelationEvent();
150 0 : if (ret == BQS_STATUS_OK) {
151 0 : done_ = false;
152 0 : BQS_LOG_INFO("Bind relation [add/del], stage [server:wait], type [request], msg [id = %u]", msgId_);
153 0 : (void)cv_.wait_for(bqsLock, std::chrono::milliseconds(MAX_WAITING_NOTIFY), [this] { return done_; });
154 0 : while ((!done_) && (processing_)) {
155 0 : cv_.wait(bqsLock);
156 : }
157 0 : if (!done_) {
158 0 : QueueManager::GetInstance().LogErrorRelationQueueStatus();
159 0 : BQS_LOG_ERROR("Bind relation [add/del], stage [server:wait], msg [id:%u] timeout, relation queue[enqueue "
160 : "cnt:%lu, dequeue cnt:%lu].",
161 : msgId_,
162 : StatisticManager::GetInstance().GetRelationEnqueCnt(),
163 : StatisticManager::GetInstance().GetRelationDequeCnt());
164 : }
165 : }
166 0 : BQS_LOG_INFO("BqsServer WaitBindMsgProc end, msg [id = %u]", msgId_);
167 0 : return;
168 0 : }
169 :
170 : /**
171 : * Bqs server enqueue bind msg request process
172 : * @return NA
173 : */
174 2 : void BqsServer::BindMsgProc()
175 : {
176 2 : BQS_LOG_INFO("BqsServer BindMsgProc begin.");
177 : {
178 2 : const std::unique_lock<std::mutex> bqsLock(mutex_);
179 2 : processing_ = true;
180 2 : }
181 : // parse bind and unbind BQSMsg
182 2 : if (bqsReqMsg_.msg_type() == BQSMsg::BIND) {
183 0 : ParseBindMsg(bqsReqMsg_, bqsRespMsg_);
184 2 : } else if (bqsReqMsg_.msg_type() == BQSMsg::UNBIND) {
185 0 : ParseUnbindMsg(bqsReqMsg_, bqsRespMsg_);
186 : } else {
187 2 : BQS_LOG_ERROR("Invalid request type[%d]", static_cast<int32_t>(bqsReqMsg_.msg_type()));
188 : }
189 2 : bqsReqMsg_.Clear();
190 :
191 : {
192 2 : const std::unique_lock<std::mutex> bqsLock(mutex_);
193 2 : processing_ = false;
194 2 : done_ = true;
195 2 : cv_.notify_one();
196 2 : }
197 2 : BQS_LOG_INFO("BqsServer BindMsgProc end.");
198 2 : return;
199 : }
200 :
201 : /**
202 : * Init easycomm server, including register handler and start listening
203 : * @return BQS_STATUS_OK:success other:failed
204 : */
205 98 : BqsStatus BqsServer::InitHandler() const
206 : {
207 98 : BQS_LOG_INFO("BqsServer service handler init begin.");
208 : // easycomm start listening
209 98 : struct EzcomServerAttr serverAttr;
210 98 : serverAttr.openCallback = &NodeHandlerWrapper;
211 98 : serverAttr.handler = &RpcHandler;
212 98 : serverAttr.gid = qsGroupId_;
213 98 : const auto err = EzcomCreateServer(&serverAttr);
214 98 : if (err < 0) {
215 0 : BQS_LOG_ERROR("Init server failed, another process may have already owned the server. "
216 : "errno = %d.", err);
217 0 : return BQS_STATUS_EASY_COMM_ERROR;
218 : }
219 98 : return BQS_STATUS_OK;
220 : }
221 :
222 : /**
223 : * Init bqs server, including init easycomm server and bind relation
224 : * @return BQS_STATUS_OK:success other:failed
225 : */
226 98 : BqsStatus BqsServer::InitBqsServer(const std::string &qsInitGrpName, const uint32_t deviceId)
227 : {
228 98 : BQS_LOG_INFO("BqsServer Init begin.");
229 :
230 98 : (void)signal(SIGPIPE, SIG_IGN);
231 :
232 98 : const BqsStatus ret = InitHandler();
233 98 : if (ret != BQS_STATUS_OK) {
234 0 : return ret;
235 : }
236 98 : qsInitGroupName_ = qsInitGrpName;
237 98 : deviceId_ = deviceId;
238 98 : BQS_LOG_INFO("BqsServer Init success.");
239 98 : return BQS_STATUS_OK;
240 : }
241 :
242 : /**
243 : * Bqs server send response msg to client, need to send a response to avoid blocking
244 : * @return NA
245 : */
246 10 : void BqsServer::SendRspMsg(const int32_t fd, const uint32_t msgId) const
247 : {
248 10 : BQS_LOG_INFO("Bind relation, stage [server:send], type [response], msg [fd = %d, id = %u]", fd, msgId);
249 :
250 10 : const uint32_t msgLen = static_cast<uint32_t>(bqsRespMsg_.ByteSizeLong());
251 10 : const uint32_t respLength = msgLen + BQS_MSG_HEAD_SIZE;
252 10 : char_t * const respData = new (std::nothrow) char_t[respLength];
253 10 : if (respData == nullptr) {
254 0 : BQS_LOG_ERROR("Malloc memory error, respData is nullptr");
255 0 : return;
256 : }
257 :
258 : // add msg length to check
259 10 : bool isOverflow = false;
260 10 : BqsCheckAssign32UAdd(msgLen, BQS_MSG_HEAD_SIZE, *(reinterpret_cast<uint32_t *>(respData)), isOverflow);
261 10 : if (isOverflow) {
262 1 : BQS_LOG_ERROR("msgLen[%u] is too big.", msgLen);
263 1 : delete[] respData;
264 1 : return;
265 : }
266 9 : if (!bqsRespMsg_.SerializePartialToArray(respData + BQS_MSG_HEAD_SIZE, static_cast<int32_t>(msgLen))) {
267 1 : BQS_LOG_ERROR("Serialize response msg failed.");
268 1 : delete[] respData;
269 1 : return;
270 : }
271 :
272 8 : EzcomResponse resp = {0U};
273 8 : resp.id = msgId;
274 8 : resp.data = reinterpret_cast<uint8_t *>(respData);
275 8 : resp.size = respLength;
276 8 : BQS_LOG_INFO("EzcomSendResponse begin, fd=%d, msgId=%u", fd, msgId);
277 8 : int32_t ret = EzcomSendResponse(fd, &resp);
278 8 : if (ret == -EAGAIN) {
279 : // just retry one times
280 0 : ret = EzcomSendResponse(fd, &resp);
281 0 : BQS_LOG_INFO("Need to retry ezcom send, fd=%d, msgId=%u", fd, msgId);
282 : }
283 8 : if (ret != 0) {
284 1 : BQS_LOG_ERROR("EzcomSendResponse end, fd=%d, msgId=%u, result=failed, ret=%d", fd, msgId, ret);
285 : } else {
286 7 : BQS_LOG_INFO("EzcomSendResponse end, fd=%d, msgId=%u, result=success", fd, msgId);
287 : }
288 :
289 8 : delete[] respData;
290 8 : StatisticManager::GetInstance().ResponseStat();
291 8 : return;
292 : }
293 :
294 : /**
295 : * Bqs server bind message processing function
296 : * @return NA
297 : */
298 2 : void BqsServer::ParseBindMsg(BQSMsg &requestMsg, BQSMsg &responseMsg) const
299 : {
300 2 : BQS_LOG_INFO("Bind relation [add], stage [server:process], type [request], msg [id:%u].", msgId_);
301 2 : BQSBindQueueMsgs * const bindQueueMsgs = requestMsg.mutable_bind_queue_msgs();
302 :
303 2 : BQSBindQueueRsps * const bqsBindQueueRspBuff = responseMsg.mutable_resp_msgs();
304 2 : auto &relationInstance = BindRelation::GetInstance();
305 :
306 2 : const uint32_t vecSize = static_cast<uint32_t>(bindQueueMsgs->bind_queue_vec_size());
307 15 : for (uint32_t i = 0U; i < vecSize; i++) {
308 13 : const BQSBindQueueMsg bindQueueMsg = bindQueueMsgs->bind_queue_vec(static_cast<int32_t>(i));
309 13 : const uint32_t srcQid = bindQueueMsg.src_queue_id();
310 13 : const uint32_t dstQid = bindQueueMsg.dst_queue_id();
311 :
312 : // add bind relation
313 13 : EntityInfo src(srcQid, deviceId_);
314 13 : EntityInfo dst(dstQid, deviceId_);
315 13 : int32_t result = BQS_STATUS_OK;
316 : // halQueueAttach third para 0 means attach without block
317 13 : auto drvRet = halQueueAttach(deviceId_, srcQid, 0);
318 13 : drvRet = (drvRet == DRV_ERROR_NONE) ? halQueueAttach(deviceId_, dstQid, 0) : drvRet;
319 13 : if (drvRet == DRV_ERROR_NONE) {
320 13 : result = relationInstance.Bind(src, dst);
321 : } else {
322 0 : BQS_LOG_ERROR("Fail to attach src queue[%u] or dst queue[%u], result[%d]", srcQid, dstQid, drvRet);
323 0 : result = BQS_STATUS_DRIVER_ERROR;
324 : }
325 13 : BQSBindQueueRsp * const bqsBindQueueInfo = bqsBindQueueRspBuff->add_bind_result_vec();
326 13 : bqsBindQueueInfo->set_bind_result(result);
327 13 : BQS_LOG_RUN_INFO("Bind relation [add], stage [server:process], relation [srcQid:%u, dstQid:%u, result:%d]",
328 : srcQid, dstQid, result);
329 13 : }
330 2 : relationInstance.Order();
331 2 : return;
332 : }
333 :
334 : /**
335 : * Bqs server unbind message processing function
336 : * @return unbind result, BQS_STATUS_OK:success other:failed
337 : */
338 :
339 40 : int32_t BqsServer::UnbindRelation(BindRelation &relationInstance,
340 : const BQSQueryMsg::QsQueryType &queryType, EntityInfo &srcId, EntityInfo &dstId) const
341 : {
342 40 : int32_t result = BQS_STATUS_INNER_ERROR;
343 40 : switch (queryType) {
344 10 : case BQSQueryMsg::BQS_QUERY_TYPE_SRC:
345 10 : result = relationInstance.UnBindBySrc(srcId);
346 10 : BQS_LOG_RUN_INFO("Bind relation [del], stage [server:process], relation [query type:src, src = %u, "
347 : "result = %d]", srcId.GetId(), result);
348 10 : break;
349 10 : case BQSQueryMsg::BQS_QUERY_TYPE_DST:
350 10 : result = relationInstance.UnBindByDst(dstId);
351 10 : BQS_LOG_RUN_INFO(
352 : "Bind relation [del], stage [server:process], relation [query type:dst, dst:%u, result:%d]",
353 : dstId.GetId(), result);
354 10 : break;
355 10 : case BQSQueryMsg::BQS_QUERY_TYPE_SRC_AND_DST:
356 10 : result = relationInstance.UnBind(srcId, dstId);
357 10 : BQS_LOG_RUN_INFO(
358 : "Bind relation [del], stage [server:process], relation [query type:src-dst, src:%u, dst:%u, result:%d]",
359 : srcId.GetId(), dstId.GetId(), result);
360 10 : break;
361 10 : default:
362 10 : BQS_LOG_ERROR("BqsServer unbind error, unsupported query type{0:src, 1:dst, 2:src-dst}:%d", queryType);
363 10 : break;
364 : }
365 40 : return result;
366 : }
367 :
368 : /**
369 : * Bqs server unbind message processing function
370 : * @return NA
371 : */
372 4 : void BqsServer::ParseUnbindMsg(BQSMsg &requestMsg, BQSMsg &responseMsg) const
373 : {
374 4 : BQS_LOG_INFO("Bind relation [del], stage [server:process], type [request], msg [id = %u].", msgId_);
375 4 : BQSQueryMsgs * const bqsQueryMsgBuff = requestMsg.mutable_query_msgs();
376 :
377 4 : BQSBindQueueRsps * const bqsBindQueueRspBuff = responseMsg.mutable_resp_msgs();
378 :
379 4 : auto &relationInstance = BindRelation::GetInstance();
380 :
381 44 : for (int32_t i = 0; i < bqsQueryMsgBuff->query_msg_vec_size(); i++) {
382 40 : BQSQueryMsg bqsQueryInfo = bqsQueryMsgBuff->query_msg_vec(i);
383 40 : const BQSQueryMsg::QsQueryType keyType = bqsQueryInfo.key_type();
384 40 : BQSBindQueueMsg * const bindQueueinfo = bqsQueryInfo.mutable_bind_queue_item();
385 :
386 40 : const uint32_t srcQid = bindQueueinfo->src_queue_id();
387 40 : const uint32_t dstQid = bindQueueinfo->dst_queue_id();
388 40 : EntityInfo src(srcQid, deviceId_);
389 40 : EntityInfo dst(dstQid, deviceId_);
390 :
391 : // delete bind relation
392 40 : const int32_t result = UnbindRelation(relationInstance, keyType, src, dst);
393 :
394 40 : BQSBindQueueRsp * const relationProcessRsp = bqsBindQueueRspBuff->add_bind_result_vec();
395 40 : relationProcessRsp->set_bind_result(result);
396 40 : }
397 :
398 4 : relationInstance.Order();
399 4 : return;
400 : }
401 :
402 : /**
403 : * Assembly response of get bind message according to src queueId
404 : * @return NA
405 : */
406 4 : void BqsServer::SerializeGetBindRspBySrc(const uint32_t srcId, BQSMsg &responseMsg) const
407 : {
408 4 : BQS_LOG_INFO("BqsServer serialize get bind rsponse by src begin, srcId:%u", srcId);
409 4 : const EntityInfo src(srcId, deviceId_);
410 4 : auto &relationInstance = BindRelation::GetInstance();
411 :
412 : // Find all dst queue id who has subscribed to the src queue id
413 4 : auto &srcToDstRelation = relationInstance.GetSrcToDstRelation();
414 4 : const auto iter = srcToDstRelation.find(src);
415 :
416 4 : const auto &abnormalSrcToDstRelation = relationInstance.GetAbnormalSrcToDstRelation();
417 4 : const auto abnormalIter = abnormalSrcToDstRelation.find(src);
418 4 : if (iter == srcToDstRelation.end() && abnormalIter == abnormalSrcToDstRelation.end()) {
419 2 : BQS_LOG_WARN("BqsServer get relation according to src:%u failed, record does not exist", src.GetId());
420 2 : return;
421 : }
422 :
423 2 : BQSBindQueueMsgs * const bqsBindQueueMsgBuff = responseMsg.mutable_bind_queue_msgs();
424 2 : if (iter != srcToDstRelation.end()) {
425 2 : FillGetBindRspBySrc(srcId, iter->second, false, bqsBindQueueMsgBuff);
426 : }
427 2 : if (abnormalIter != abnormalSrcToDstRelation.end()) {
428 1 : FillGetBindRspBySrc(srcId, abnormalIter->second, true, bqsBindQueueMsgBuff);
429 : }
430 4 : }
431 :
432 : /**
433 : * Fill getBind response by src and dstSet, one-to-one relation
434 : * @return NA
435 : */
436 3 : void BqsServer::FillGetBindRspBySrc(const uint32_t srcId, const std::unordered_set<EntityInfo, EntityInfoHash> &dstSet,
437 : bool isAbnormal, BQSBindQueueMsgs *const bqsBindQueueMsgBuff) const
438 : {
439 3 : BQS_LOG_INFO("Bind relation [get], stage [server:process], relation [size:%zu].", dstSet.size());
440 3 : int32_t i = 0;
441 6 : for (auto setIter = dstSet.begin(); setIter != dstSet.end(); ++setIter) {
442 3 : BQSBindQueueMsg * const bqsBindQueueInfo = bqsBindQueueMsgBuff->add_bind_queue_vec();
443 3 : bqsBindQueueInfo->set_src_queue_id(srcId);
444 3 : const EntityInfo dstQ = *setIter;
445 3 : bqsBindQueueInfo->set_dst_queue_id(dstQ.GetId());
446 3 : ++i;
447 3 : BQS_LOG_INFO(
448 : "Bind relation [get], stage [server:process], relation [abnormal:%d, index:%d, src:%u, dst:%u]",
449 : static_cast<int32_t>(isAbnormal), i, srcId, dstQ.GetId());
450 3 : }
451 3 : }
452 :
453 : /**
454 : * Assembly response of get bind message according to dst queueId, one-to-one relation
455 : * @return NA
456 : */
457 3 : void BqsServer::SerializeGetBindRspByDst(const uint32_t dstId, BQSMsg &responseMsg) const
458 : {
459 3 : BQS_LOG_INFO("BqsServer serialize get bind rsponse by dst begin, dstId:%u", dstId);
460 3 : auto &relationInstance = BindRelation::GetInstance();
461 3 : const EntityInfo dst(dstId, deviceId_);
462 :
463 3 : auto &dstToSrcRelation = relationInstance.GetDstToSrcRelation();
464 3 : const auto iter = dstToSrcRelation.find(dst);
465 :
466 3 : const auto &abnormalDstToSrcRelation = relationInstance.GetAbnormalDstToSrcRelation();
467 3 : const auto abnormalIter = abnormalDstToSrcRelation.find(dst);
468 3 : if ((iter == dstToSrcRelation.end()) && (abnormalIter == abnormalDstToSrcRelation.end())) {
469 1 : BQS_LOG_WARN("BqsServer get relation according to dst:%u failed, record does not exist", dstId);
470 1 : return;
471 : }
472 :
473 2 : BQSBindQueueMsgs * const bqsBindQueueMsgBuff = responseMsg.mutable_bind_queue_msgs();
474 2 : if (iter != dstToSrcRelation.end()) {
475 2 : FillGetBindRspByDst(iter->second, dstId, false, bqsBindQueueMsgBuff);
476 : }
477 2 : if (abnormalIter != abnormalDstToSrcRelation.end()) {
478 1 : FillGetBindRspByDst(abnormalIter->second, dstId, true, bqsBindQueueMsgBuff);
479 : }
480 3 : }
481 :
482 : /**
483 : * Fill getBind response by srcSet and dst, one-to-one relation
484 : * @return NA
485 : */
486 3 : void BqsServer::FillGetBindRspByDst(const std::unordered_set<EntityInfo, EntityInfoHash> &srcSet, const uint32_t dstId,
487 : bool isAbnormal, BQSBindQueueMsgs *const bqsBindQueueMsgBuff) const
488 : {
489 3 : BQS_LOG_INFO("Bind relation [get], stage [server:process], relation [size:%zu].", srcSet.size());
490 3 : int32_t i = 0;
491 6 : for (auto setIter = srcSet.begin(); setIter != srcSet.end(); ++setIter) {
492 3 : BQSBindQueueMsg * const bqsBindQueueInfo = bqsBindQueueMsgBuff->add_bind_queue_vec();
493 3 : bqsBindQueueInfo->set_src_queue_id(setIter->GetId());
494 3 : bqsBindQueueInfo->set_dst_queue_id(dstId);
495 3 : ++i;
496 3 : BQS_LOG_INFO("Bind relation [get], stage [server:process], relation [abnormal:%d, index:%d, src:%u, dst:%u]",
497 : static_cast<int32_t>(isAbnormal), i, setIter->GetId(), dstId);
498 : }
499 3 : }
500 :
501 : /**
502 : * Assembly response of get bind message
503 : * @return NA
504 : */
505 6 : void BqsServer::SerializeGetBindRsp(
506 : const BQSQueryMsg::QsQueryType &queryType, const uint32_t srcId, const uint32_t dstId, BQSMsg &responseMsg) const
507 : {
508 6 : switch (queryType) {
509 3 : case BQSQueryMsg::BQS_QUERY_TYPE_SRC:
510 3 : SerializeGetBindRspBySrc(srcId, responseMsg);
511 3 : break;
512 2 : case BQSQueryMsg::BQS_QUERY_TYPE_DST:
513 2 : SerializeGetBindRspByDst(dstId, responseMsg);
514 2 : break;
515 1 : default:
516 1 : BQS_LOG_ERROR("BqsServer get bind error, unsupported query type{0:src, 1:dst, 2:src-dst}:%d", queryType);
517 1 : break;
518 : }
519 6 : return;
520 : }
521 :
522 : /**
523 : * Bqs server get bind message processing function
524 : * @return NA
525 : */
526 6 : void BqsServer::ParseGetBindMsg(BQSMsg &requestMsg, BQSMsg &responseMsg) const
527 : {
528 6 : BQS_LOG_INFO("Bind relation [get], stage [server:process], type [request], msg [id:%u].", msgId_);
529 6 : BQSQueryMsg * const bqsQueryInfo = requestMsg.mutable_query_msg();
530 :
531 6 : const BQSQueryMsg::QsQueryType keyType = bqsQueryInfo->key_type();
532 6 : BQSBindQueueMsg * const bqsBindQueueInfo = bqsQueryInfo->mutable_bind_queue_item();
533 :
534 6 : const uint32_t src = bqsBindQueueInfo->src_queue_id();
535 6 : const uint32_t dst = bqsBindQueueInfo->dst_queue_id();
536 :
537 6 : SerializeGetBindRsp(keyType, src, dst, responseMsg);
538 12 : return;
539 : }
540 :
541 : /**
542 : * Bqs server get paged bind message processing function
543 : * @return NA
544 : */
545 2 : void BqsServer::ParseGetPagedBindMsg(BQSMsg &requestMsg, BQSMsg &responseMsg) const
546 : {
547 2 : BQS_LOG_INFO("Bind relation [get_all], stage [server:process], type [request], msg [id:%u].", msgId_);
548 2 : BQSBindQueueMsgs * const bqsBindQueueMsgBuff = responseMsg.mutable_bind_queue_msgs();
549 :
550 2 : BQSPagedMsg * const pagedMsg = requestMsg.mutable_paged_msg();
551 :
552 2 : BQSPagedMsg * const pagedRspMsg = responseMsg.mutable_paged_msg();
553 :
554 2 : auto &relationInstance = BindRelation::GetInstance();
555 :
556 2 : static std::vector<std::tuple<uint32_t, uint32_t>> relations;
557 : static uint32_t offsetSave = 0U;
558 : static uint32_t total = 0U;
559 2 : const uint32_t msgOffset = pagedMsg->offset();
560 2 : if ((msgOffset == 0U) || (msgOffset < offsetSave) || relations.empty()) {
561 2 : auto &srcToDstRelation = relationInstance.GetSrcToDstRelation();
562 2 : RelationsCopy(relations, total, srcToDstRelation);
563 2 : AppendRelations(relations, relationInstance.GetAbnormalSrcToDstRelation());
564 2 : offsetSave = msgOffset;
565 2 : total = static_cast<uint32_t>(relations.size());
566 : }
567 2 : pagedRspMsg->set_total(total);
568 2 : const uint32_t offset = (msgOffset > total) ? total : msgOffset;
569 2 : const uint32_t limit = pagedMsg->limit();
570 :
571 : // get bind relation
572 2 : uint32_t i = 0U;
573 2 : auto iter = relations.begin();
574 2 : BQS_LOG_INFO("Bind relation [get_paged], stage [server:process], relation [offset:%u, limit:%u, size:%u]",
575 : pagedMsg->offset(),
576 : limit,
577 : total);
578 : std::advance(iter, offset);
579 12 : while ((iter != relations.end()) && (i < limit)) {
580 10 : const uint32_t srcId = std::get<0>(*iter);
581 10 : const uint32_t dstId = std::get<1>(*iter);
582 10 : BQSBindQueueMsg * const bqsBindQueueInfo = bqsBindQueueMsgBuff->add_bind_queue_vec();
583 :
584 10 : bqsBindQueueInfo->set_src_queue_id(srcId);
585 10 : bqsBindQueueInfo->set_dst_queue_id(dstId);
586 10 : ++iter;
587 10 : ++i;
588 : }
589 4 : return;
590 : }
591 :
592 : /**
593 : * Copy relation map to a vector container
594 : * @return NA
595 : */
596 2 : void BqsServer::RelationsCopy(std::vector<std::tuple<uint32_t, uint32_t>> &relations, const uint32_t oldSize,
597 : const std::unordered_map<EntityInfo, std::unordered_set<EntityInfo, EntityInfoHash>, EntityInfoHash> &srcMap) const
598 : {
599 2 : relations.clear();
600 2 : if (oldSize != 0U) {
601 0 : relations.reserve(static_cast<std::vector<std::tuple<uint32_t, uint32_t>>::size_type>(oldSize));
602 : }
603 3 : for (const auto &iter : srcMap) {
604 1 : (void) std::transform(iter.second.begin(), iter.second.end(), std::back_inserter(relations),
605 10 : [&](const EntityInfo entityInfo) {
606 10 : return std::make_pair(iter.first.GetId(), entityInfo.GetId());
607 : });
608 : }
609 2 : }
610 :
611 : /**
612 : * append relations to a vector container
613 : * @return NA
614 : */
615 3 : void BqsServer::AppendRelations(std::vector<std::tuple<uint32_t, uint32_t>> &relations,
616 : const std::unordered_map<EntityInfo, std::unordered_set<EntityInfo, EntityInfoHash>, EntityInfoHash> &srcMap) const
617 : {
618 5 : for (const auto &iter : srcMap) {
619 2 : (void) std::transform(iter.second.begin(), iter.second.end(), std::back_inserter(relations),
620 3 : [&](const EntityInfo entityInfo) {
621 3 : return std::make_pair(iter.first.GetId(), entityInfo.GetId());
622 : });
623 : }
624 3 : }
625 :
626 : } // namespace bqs
|
