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 :
65 1 : BqsServer::~BqsServer() {}
66 :
67 134 : BqsServer& BqsServer::GetInstance()
68 : {
69 134 : static BqsServer instance;
70 134 : return instance;
71 : }
72 :
73 5 : void BqsServer::InitBuff() const
74 : {
75 5 : BuffCfg defaultCfg = {};
76 5 : const int32_t drvRet = halBuffInit(&defaultCfg);
77 5 : if ((drvRet != DRV_ERROR_NONE) && (drvRet != DRV_ERROR_REPEATED_INIT)) {
78 0 : BQS_LOG_ERROR("[BqsServer]Buffer initial failed ret[%d]", drvRet);
79 0 : return;
80 : }
81 5 : BQS_LOG_INFO("[RouterServer] Buffer init success ret = %d", drvRet);
82 : }
83 :
84 : /**
85 : * Bqs server handle BqsMsg, get/getall deal now, bind/unbind send to work thread to deal
86 : * @return NA
87 : */
88 5 : void BqsServer::HandleBqsReqMsg(const uint32_t msgId, const char_t* const data, const uint32_t dataSize)
89 : {
90 5 : BQS_LOG_INFO("Bind relation, stage [server:receive], type [request], msg [id = %u]", msgId);
91 5 : msgId_ = msgId;
92 5 : bqsRespMsg_.Clear(); // init response msg
93 5 : if (data == nullptr) {
94 0 : BQS_LOG_ERROR("Request of BqsClient is nullptr.");
95 0 : return;
96 : }
97 5 : if (dataSize < BQS_MSG_HEAD_SIZE) {
98 0 : BQS_LOG_ERROR("Request of BqsClient size:%u should be not less than head:%u.", dataSize, BQS_MSG_HEAD_SIZE);
99 0 : return;
100 : }
101 5 : InitBuff();
102 5 : const uint32_t currMsgSize = *(PtrToPtr<const char_t, const uint32_t>(data));
103 5 : if (currMsgSize != dataSize) {
104 1 : BQS_LOG_ERROR("message error, head_msg_content = %u, request_size = %u", currMsgSize, dataSize);
105 1 : return;
106 : }
107 4 : const uint32_t parseLength = currMsgSize - BQS_MSG_HEAD_SIZE;
108 4 : if (bqsReqMsg_.ParseFromArray(data + BQS_MSG_HEAD_SIZE, static_cast<int32_t>(parseLength))) {
109 4 : BQS_LOG_INFO(
110 : "BqsServer request msg type{%d:BIND, %d:UNBIND, %d:GET_BIND, %d:GET_ALL_BIND}:%d "
111 : "begin to process",
112 : BQSMsg::BIND, BQSMsg::UNBIND, BQSMsg::GET_BIND, BQSMsg::GET_ALL_BIND, bqsReqMsg_.msg_type());
113 4 : switch (bqsReqMsg_.msg_type()) {
114 1 : case BQSMsg::GET_BIND:
115 1 : StatisticManager::GetInstance().GetBindStat();
116 1 : ParseGetBindMsg(bqsReqMsg_, bqsRespMsg_);
117 1 : break;
118 1 : case BQSMsg::GET_ALL_BIND:
119 1 : StatisticManager::GetInstance().GetAllBindStat();
120 1 : ParseGetPagedBindMsg(bqsReqMsg_, bqsRespMsg_);
121 1 : break;
122 1 : case BQSMsg::BIND:
123 1 : StatisticManager::GetInstance().BindStat();
124 1 : WaitBindMsgProc();
125 1 : break;
126 0 : case BQSMsg::UNBIND:
127 0 : StatisticManager::GetInstance().UnbindStat();
128 0 : WaitBindMsgProc();
129 0 : break;
130 1 : default:
131 1 : BQS_LOG_ERROR("BqsServer receive unsupported msg type:%d", bqsReqMsg_.msg_type());
132 1 : break;
133 : }
134 : }
135 4 : BQS_LOG_INFO("BqsServer HandleBqsMsg end");
136 4 : return;
137 : }
138 :
139 : /**
140 : * Bqs server wait work thread to process msg
141 : * @return NA
142 : */
143 0 : void BqsServer::WaitBindMsgProc()
144 : {
145 0 : BQS_LOG_INFO("Bind relation [add/del], stage [server:enqueue], type [request], msg [id = %u]", msgId_);
146 0 : std::unique_lock<std::mutex> bqsLock(mutex_);
147 0 : const BqsStatus ret = QueueManager::GetInstance().EnqueueRelationEvent();
148 0 : if (ret == BQS_STATUS_OK) {
149 0 : done_ = false;
150 0 : BQS_LOG_INFO("Bind relation [add/del], stage [server:wait], type [request], msg [id = %u]", msgId_);
151 0 : (void)cv_.wait_for(bqsLock, std::chrono::milliseconds(MAX_WAITING_NOTIFY), [this] { return done_; });
152 0 : while ((!done_) && (processing_)) {
153 0 : cv_.wait(bqsLock);
154 : }
155 0 : if (!done_) {
156 0 : QueueManager::GetInstance().LogErrorRelationQueueStatus();
157 0 : BQS_LOG_ERROR(
158 : "Bind relation [add/del], stage [server:wait], msg [id:%u] timeout, relation queue[enqueue "
159 : "cnt:%lu, dequeue cnt:%lu].",
160 : msgId_, StatisticManager::GetInstance().GetRelationEnqueCnt(),
161 : StatisticManager::GetInstance().GetRelationDequeCnt());
162 : }
163 : }
164 0 : BQS_LOG_INFO("BqsServer WaitBindMsgProc end, msg [id = %u]", msgId_);
165 0 : return;
166 0 : }
167 :
168 : /**
169 : * Bqs server enqueue bind msg request process
170 : * @return NA
171 : */
172 2 : void BqsServer::BindMsgProc()
173 : {
174 2 : BQS_LOG_INFO("BqsServer BindMsgProc begin.");
175 : {
176 2 : const std::unique_lock<std::mutex> bqsLock(mutex_);
177 2 : processing_ = true;
178 2 : }
179 : // parse bind and unbind BQSMsg
180 2 : if (bqsReqMsg_.msg_type() == BQSMsg::BIND) {
181 0 : ParseBindMsg(bqsReqMsg_, bqsRespMsg_);
182 2 : } else if (bqsReqMsg_.msg_type() == BQSMsg::UNBIND) {
183 0 : ParseUnbindMsg(bqsReqMsg_, bqsRespMsg_);
184 : } else {
185 2 : BQS_LOG_ERROR("Invalid request type[%d]", static_cast<int32_t>(bqsReqMsg_.msg_type()));
186 : }
187 2 : bqsReqMsg_.Clear();
188 :
189 : {
190 2 : const std::unique_lock<std::mutex> bqsLock(mutex_);
191 2 : processing_ = false;
192 2 : done_ = true;
193 2 : cv_.notify_one();
194 2 : }
195 2 : BQS_LOG_INFO("BqsServer BindMsgProc end.");
196 2 : return;
197 : }
198 :
199 : /**
200 : * Init easycomm server, including register handler and start listening
201 : * @return BQS_STATUS_OK:success other:failed
202 : */
203 98 : BqsStatus BqsServer::InitHandler() const
204 : {
205 98 : BQS_LOG_INFO("BqsServer service handler init begin.");
206 : // easycomm start listening
207 98 : struct EzcomServerAttr serverAttr;
208 98 : serverAttr.openCallback = &NodeHandlerWrapper;
209 98 : serverAttr.handler = &RpcHandler;
210 98 : serverAttr.gid = qsGroupId_;
211 98 : const auto err = EzcomCreateServer(&serverAttr);
212 98 : if (err < 0) {
213 0 : BQS_LOG_ERROR(
214 : "Init server failed, another process may have already owned the server. "
215 : "errno = %d.",
216 : 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(
328 : "Bind relation [add], stage [server:process], relation [srcQid:%u, dstQid:%u, result:%d]", srcQid, dstQid,
329 : result);
330 13 : }
331 2 : relationInstance.Order();
332 2 : return;
333 : }
334 :
335 : /**
336 : * Bqs server unbind message processing function
337 : * @return unbind result, BQS_STATUS_OK:success other:failed
338 : */
339 :
340 40 : int32_t BqsServer::UnbindRelation(
341 : BindRelation& relationInstance, const BQSQueryMsg::QsQueryType& queryType, EntityInfo& srcId,
342 : EntityInfo& dstId) const
343 : {
344 40 : int32_t result = BQS_STATUS_INNER_ERROR;
345 40 : switch (queryType) {
346 10 : case BQSQueryMsg::BQS_QUERY_TYPE_SRC:
347 10 : result = relationInstance.UnBindBySrc(srcId);
348 10 : BQS_LOG_RUN_INFO(
349 : "Bind relation [del], stage [server:process], relation [query type:src, src = %u, "
350 : "result = %d]",
351 : srcId.GetId(), result);
352 10 : break;
353 10 : case BQSQueryMsg::BQS_QUERY_TYPE_DST:
354 10 : result = relationInstance.UnBindByDst(dstId);
355 10 : BQS_LOG_RUN_INFO(
356 : "Bind relation [del], stage [server:process], relation [query type:dst, dst:%u, result:%d]",
357 : dstId.GetId(), result);
358 10 : break;
359 10 : case BQSQueryMsg::BQS_QUERY_TYPE_SRC_AND_DST:
360 10 : result = relationInstance.UnBind(srcId, dstId);
361 10 : BQS_LOG_RUN_INFO(
362 : "Bind relation [del], stage [server:process], relation [query type:src-dst, src:%u, dst:%u, result:%d]",
363 : srcId.GetId(), dstId.GetId(), result);
364 10 : break;
365 10 : default:
366 10 : BQS_LOG_ERROR("BqsServer unbind error, unsupported query type{0:src, 1:dst, 2:src-dst}:%d", queryType);
367 10 : break;
368 : }
369 40 : return result;
370 : }
371 :
372 : /**
373 : * Bqs server unbind message processing function
374 : * @return NA
375 : */
376 4 : void BqsServer::ParseUnbindMsg(BQSMsg& requestMsg, BQSMsg& responseMsg) const
377 : {
378 4 : BQS_LOG_INFO("Bind relation [del], stage [server:process], type [request], msg [id = %u].", msgId_);
379 4 : BQSQueryMsgs* const bqsQueryMsgBuff = requestMsg.mutable_query_msgs();
380 :
381 4 : BQSBindQueueRsps* const bqsBindQueueRspBuff = responseMsg.mutable_resp_msgs();
382 :
383 4 : auto& relationInstance = BindRelation::GetInstance();
384 :
385 44 : for (int32_t i = 0; i < bqsQueryMsgBuff->query_msg_vec_size(); i++) {
386 40 : BQSQueryMsg bqsQueryInfo = bqsQueryMsgBuff->query_msg_vec(i);
387 40 : const BQSQueryMsg::QsQueryType keyType = bqsQueryInfo.key_type();
388 40 : BQSBindQueueMsg* const bindQueueinfo = bqsQueryInfo.mutable_bind_queue_item();
389 :
390 40 : const uint32_t srcQid = bindQueueinfo->src_queue_id();
391 40 : const uint32_t dstQid = bindQueueinfo->dst_queue_id();
392 40 : EntityInfo src(srcQid, deviceId_);
393 40 : EntityInfo dst(dstQid, deviceId_);
394 :
395 : // delete bind relation
396 40 : const int32_t result = UnbindRelation(relationInstance, keyType, src, dst);
397 :
398 40 : BQSBindQueueRsp* const relationProcessRsp = bqsBindQueueRspBuff->add_bind_result_vec();
399 40 : relationProcessRsp->set_bind_result(result);
400 40 : }
401 :
402 4 : relationInstance.Order();
403 4 : return;
404 : }
405 :
406 : /**
407 : * Assembly response of get bind message according to src queueId
408 : * @return NA
409 : */
410 4 : void BqsServer::SerializeGetBindRspBySrc(const uint32_t srcId, BQSMsg& responseMsg) const
411 : {
412 4 : BQS_LOG_INFO("BqsServer serialize get bind rsponse by src begin, srcId:%u", srcId);
413 4 : const EntityInfo src(srcId, deviceId_);
414 4 : auto& relationInstance = BindRelation::GetInstance();
415 :
416 : // Find all dst queue id who has subscribed to the src queue id
417 4 : auto& srcToDstRelation = relationInstance.GetSrcToDstRelation();
418 4 : const auto iter = srcToDstRelation.find(src);
419 :
420 4 : const auto& abnormalSrcToDstRelation = relationInstance.GetAbnormalSrcToDstRelation();
421 4 : const auto abnormalIter = abnormalSrcToDstRelation.find(src);
422 4 : if (iter == srcToDstRelation.end() && abnormalIter == abnormalSrcToDstRelation.end()) {
423 2 : BQS_LOG_WARN("BqsServer get relation according to src:%u failed, record does not exist", src.GetId());
424 2 : return;
425 : }
426 :
427 2 : BQSBindQueueMsgs* const bqsBindQueueMsgBuff = responseMsg.mutable_bind_queue_msgs();
428 2 : if (iter != srcToDstRelation.end()) {
429 2 : FillGetBindRspBySrc(srcId, iter->second, false, bqsBindQueueMsgBuff);
430 : }
431 2 : if (abnormalIter != abnormalSrcToDstRelation.end()) {
432 1 : FillGetBindRspBySrc(srcId, abnormalIter->second, true, bqsBindQueueMsgBuff);
433 : }
434 4 : }
435 :
436 : /**
437 : * Fill getBind response by src and dstSet, one-to-one relation
438 : * @return NA
439 : */
440 3 : void BqsServer::FillGetBindRspBySrc(
441 : const uint32_t srcId, const std::unordered_set<EntityInfo, EntityInfoHash>& dstSet, bool isAbnormal,
442 : BQSBindQueueMsgs* const bqsBindQueueMsgBuff) const
443 : {
444 3 : BQS_LOG_INFO("Bind relation [get], stage [server:process], relation [size:%zu].", dstSet.size());
445 3 : int32_t i = 0;
446 6 : for (auto setIter = dstSet.begin(); setIter != dstSet.end(); ++setIter) {
447 3 : BQSBindQueueMsg* const bqsBindQueueInfo = bqsBindQueueMsgBuff->add_bind_queue_vec();
448 3 : bqsBindQueueInfo->set_src_queue_id(srcId);
449 3 : const EntityInfo dstQ = *setIter;
450 3 : bqsBindQueueInfo->set_dst_queue_id(dstQ.GetId());
451 3 : ++i;
452 3 : BQS_LOG_INFO(
453 : "Bind relation [get], stage [server:process], relation [abnormal:%d, index:%d, src:%u, dst:%u]",
454 : static_cast<int32_t>(isAbnormal), i, srcId, dstQ.GetId());
455 3 : }
456 3 : }
457 :
458 : /**
459 : * Assembly response of get bind message according to dst queueId, one-to-one relation
460 : * @return NA
461 : */
462 3 : void BqsServer::SerializeGetBindRspByDst(const uint32_t dstId, BQSMsg& responseMsg) const
463 : {
464 3 : BQS_LOG_INFO("BqsServer serialize get bind rsponse by dst begin, dstId:%u", dstId);
465 3 : auto& relationInstance = BindRelation::GetInstance();
466 3 : const EntityInfo dst(dstId, deviceId_);
467 :
468 3 : auto& dstToSrcRelation = relationInstance.GetDstToSrcRelation();
469 3 : const auto iter = dstToSrcRelation.find(dst);
470 :
471 3 : const auto& abnormalDstToSrcRelation = relationInstance.GetAbnormalDstToSrcRelation();
472 3 : const auto abnormalIter = abnormalDstToSrcRelation.find(dst);
473 3 : if ((iter == dstToSrcRelation.end()) && (abnormalIter == abnormalDstToSrcRelation.end())) {
474 1 : BQS_LOG_WARN("BqsServer get relation according to dst:%u failed, record does not exist", dstId);
475 1 : return;
476 : }
477 :
478 2 : BQSBindQueueMsgs* const bqsBindQueueMsgBuff = responseMsg.mutable_bind_queue_msgs();
479 2 : if (iter != dstToSrcRelation.end()) {
480 2 : FillGetBindRspByDst(iter->second, dstId, false, bqsBindQueueMsgBuff);
481 : }
482 2 : if (abnormalIter != abnormalDstToSrcRelation.end()) {
483 1 : FillGetBindRspByDst(abnormalIter->second, dstId, true, bqsBindQueueMsgBuff);
484 : }
485 3 : }
486 :
487 : /**
488 : * Fill getBind response by srcSet and dst, one-to-one relation
489 : * @return NA
490 : */
491 3 : void BqsServer::FillGetBindRspByDst(
492 : const std::unordered_set<EntityInfo, EntityInfoHash>& srcSet, const uint32_t dstId, bool isAbnormal,
493 : BQSBindQueueMsgs* const bqsBindQueueMsgBuff) const
494 : {
495 3 : BQS_LOG_INFO("Bind relation [get], stage [server:process], relation [size:%zu].", srcSet.size());
496 3 : int32_t i = 0;
497 6 : for (auto setIter = srcSet.begin(); setIter != srcSet.end(); ++setIter) {
498 3 : BQSBindQueueMsg* const bqsBindQueueInfo = bqsBindQueueMsgBuff->add_bind_queue_vec();
499 3 : bqsBindQueueInfo->set_src_queue_id(setIter->GetId());
500 3 : bqsBindQueueInfo->set_dst_queue_id(dstId);
501 3 : ++i;
502 3 : BQS_LOG_INFO(
503 : "Bind relation [get], stage [server:process], relation [abnormal:%d, index:%d, src:%u, dst:%u]",
504 : static_cast<int32_t>(isAbnormal), i, setIter->GetId(), dstId);
505 : }
506 3 : }
507 :
508 : /**
509 : * Assembly response of get bind message
510 : * @return NA
511 : */
512 6 : void BqsServer::SerializeGetBindRsp(
513 : const BQSQueryMsg::QsQueryType& queryType, const uint32_t srcId, const uint32_t dstId, BQSMsg& responseMsg) const
514 : {
515 6 : switch (queryType) {
516 3 : case BQSQueryMsg::BQS_QUERY_TYPE_SRC:
517 3 : SerializeGetBindRspBySrc(srcId, responseMsg);
518 3 : break;
519 2 : case BQSQueryMsg::BQS_QUERY_TYPE_DST:
520 2 : SerializeGetBindRspByDst(dstId, responseMsg);
521 2 : break;
522 1 : default:
523 1 : BQS_LOG_ERROR("BqsServer get bind error, unsupported query type{0:src, 1:dst, 2:src-dst}:%d", queryType);
524 1 : break;
525 : }
526 6 : return;
527 : }
528 :
529 : /**
530 : * Bqs server get bind message processing function
531 : * @return NA
532 : */
533 6 : void BqsServer::ParseGetBindMsg(BQSMsg& requestMsg, BQSMsg& responseMsg) const
534 : {
535 6 : BQS_LOG_INFO("Bind relation [get], stage [server:process], type [request], msg [id:%u].", msgId_);
536 6 : BQSQueryMsg* const bqsQueryInfo = requestMsg.mutable_query_msg();
537 :
538 6 : const BQSQueryMsg::QsQueryType keyType = bqsQueryInfo->key_type();
539 6 : BQSBindQueueMsg* const bqsBindQueueInfo = bqsQueryInfo->mutable_bind_queue_item();
540 :
541 6 : const uint32_t src = bqsBindQueueInfo->src_queue_id();
542 6 : const uint32_t dst = bqsBindQueueInfo->dst_queue_id();
543 :
544 6 : SerializeGetBindRsp(keyType, src, dst, responseMsg);
545 12 : return;
546 : }
547 :
548 : /**
549 : * Bqs server get paged bind message processing function
550 : * @return NA
551 : */
552 2 : void BqsServer::ParseGetPagedBindMsg(BQSMsg& requestMsg, BQSMsg& responseMsg) const
553 : {
554 2 : BQS_LOG_INFO("Bind relation [get_all], stage [server:process], type [request], msg [id:%u].", msgId_);
555 2 : BQSBindQueueMsgs* const bqsBindQueueMsgBuff = responseMsg.mutable_bind_queue_msgs();
556 :
557 2 : BQSPagedMsg* const pagedMsg = requestMsg.mutable_paged_msg();
558 :
559 2 : BQSPagedMsg* const pagedRspMsg = responseMsg.mutable_paged_msg();
560 :
561 2 : auto& relationInstance = BindRelation::GetInstance();
562 :
563 2 : static std::vector<std::tuple<uint32_t, uint32_t>> relations;
564 : static uint32_t offsetSave = 0U;
565 : static uint32_t total = 0U;
566 2 : const uint32_t msgOffset = pagedMsg->offset();
567 2 : if ((msgOffset == 0U) || (msgOffset < offsetSave) || relations.empty()) {
568 2 : auto& srcToDstRelation = relationInstance.GetSrcToDstRelation();
569 2 : RelationsCopy(relations, total, srcToDstRelation);
570 2 : AppendRelations(relations, relationInstance.GetAbnormalSrcToDstRelation());
571 2 : offsetSave = msgOffset;
572 2 : total = static_cast<uint32_t>(relations.size());
573 : }
574 2 : pagedRspMsg->set_total(total);
575 2 : const uint32_t offset = (msgOffset > total) ? total : msgOffset;
576 2 : const uint32_t limit = pagedMsg->limit();
577 :
578 : // get bind relation
579 2 : uint32_t i = 0U;
580 2 : auto iter = relations.begin();
581 2 : BQS_LOG_INFO(
582 : "Bind relation [get_paged], stage [server:process], relation [offset:%u, limit:%u, size:%u]",
583 : pagedMsg->offset(), limit, total);
584 : std::advance(iter, offset);
585 12 : while ((iter != relations.end()) && (i < limit)) {
586 10 : const uint32_t srcId = std::get<0>(*iter);
587 10 : const uint32_t dstId = std::get<1>(*iter);
588 10 : BQSBindQueueMsg* const bqsBindQueueInfo = bqsBindQueueMsgBuff->add_bind_queue_vec();
589 :
590 10 : bqsBindQueueInfo->set_src_queue_id(srcId);
591 10 : bqsBindQueueInfo->set_dst_queue_id(dstId);
592 10 : ++iter;
593 10 : ++i;
594 : }
595 4 : return;
596 : }
597 :
598 : /**
599 : * Copy relation map to a vector container
600 : * @return NA
601 : */
602 2 : void BqsServer::RelationsCopy(
603 : std::vector<std::tuple<uint32_t, uint32_t>>& relations, const uint32_t oldSize,
604 : const std::unordered_map<EntityInfo, std::unordered_set<EntityInfo, EntityInfoHash>, EntityInfoHash>& srcMap) const
605 : {
606 2 : relations.clear();
607 2 : if (oldSize != 0U) {
608 0 : relations.reserve(static_cast<std::vector<std::tuple<uint32_t, uint32_t>>::size_type>(oldSize));
609 : }
610 3 : for (const auto& iter : srcMap) {
611 1 : (void)std::transform(
612 : iter.second.begin(), iter.second.end(), std::back_inserter(relations),
613 10 : [&](const EntityInfo entityInfo) { return std::make_pair(iter.first.GetId(), entityInfo.GetId()); });
614 : }
615 2 : }
616 :
617 : /**
618 : * append relations to a vector container
619 : * @return NA
620 : */
621 3 : void BqsServer::AppendRelations(
622 : std::vector<std::tuple<uint32_t, uint32_t>>& relations,
623 : const std::unordered_map<EntityInfo, std::unordered_set<EntityInfo, EntityInfoHash>, EntityInfoHash>& srcMap) const
624 : {
625 5 : for (const auto& iter : srcMap) {
626 2 : (void)std::transform(
627 : iter.second.begin(), iter.second.end(), std::back_inserter(relations),
628 3 : [&](const EntityInfo entityInfo) { return std::make_pair(iter.first.GetId(), entityInfo.GetId()); });
629 : }
630 3 : }
631 :
632 : } // namespace bqs
|
