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 "ezcom_client.h"
12 :
13 : #include <securec.h>
14 : #include "easy_comm.h"
15 :
16 : #include "bqs_log.h"
17 : #include "bqs_msg.h"
18 : #include "bqs_util.h"
19 :
20 : namespace {
21 : std::mutex g_ezClientMut;
22 : }
23 : namespace bqs {
24 : int32_t EzcomClient::clientFd_(0);
25 : std::atomic<bool> EzcomClient::initFlag_(false);
26 :
27 : /* *
28 : * Create instance of EzcomClient.
29 : * @return EzcomClient*: success
30 : */
31 60 : EzcomClient* EzcomClient::GetInstance(const int32_t fd)
32 : {
33 60 : const std::lock_guard<std::mutex> lk(g_ezClientMut);
34 60 : if (!initFlag_) {
35 1 : clientFd_ = fd;
36 1 : initFlag_ = true;
37 : }
38 : static EzcomClient instance;
39 60 : return &instance;
40 60 : }
41 :
42 : /* *
43 : * Assembly bind BQSMsg message
44 : * @return BQS_STATUS_OK: success, other: error
45 : */
46 8 : BqsStatus EzcomClient::SerializeBindMsg(const std::vector<BQSBindQueueItem>& bindQueueVec, BQSMsg& bqsClientMsg) const
47 : {
48 8 : BQS_LOG_INFO("Bind relation [add], stage [client], type [request], relation [size:%zu]", bindQueueVec.size());
49 8 : if (bindQueueVec.empty()) {
50 3 : BQS_LOG_WARN("The number of bind relation to be added should be greater than zero.");
51 3 : return BQS_STATUS_PARAM_INVALID;
52 : }
53 :
54 5 : bqsClientMsg.set_msg_type(BQSMsg::BIND);
55 5 : BQSBindQueueMsgs* const bqsBindQueueMsgBuff = bqsClientMsg.mutable_bind_queue_msgs();
56 :
57 532 : for (size_t i = 0U; i < bindQueueVec.size(); i++) {
58 527 : BQSBindQueueMsg* const bqsBindQueueEzMsg = bqsBindQueueMsgBuff->add_bind_queue_vec();
59 :
60 527 : bqsBindQueueEzMsg->set_src_queue_id(bindQueueVec[i].srcQueueId_);
61 527 : bqsBindQueueEzMsg->set_dst_queue_id(bindQueueVec[i].dstQueueId_);
62 527 : BQS_LOG_INFO(
63 : "Bind relation [add], stage [client], type [request], relation [src:%u, dst:%u]",
64 : bindQueueVec[i].srcQueueId_, bindQueueVec[i].dstQueueId_);
65 : }
66 5 : return BQS_STATUS_OK;
67 : }
68 :
69 : /* *
70 : * Send BQSMsg message to server
71 : * @return BQS_STATUS_OK: success, other: error
72 : */
73 22 : BqsStatus EzcomClient::SendBqsMsg(const BQSMsg& bqsReqMsg, BQSMsg& bqsRespMsg) const
74 : {
75 22 : const uint32_t bqsMsgLen = static_cast<uint32_t>(bqsReqMsg.ByteSizeLong());
76 22 : uint32_t reqLength = 0U;
77 22 : bool isOverflow = false;
78 22 : BqsCheckAssign32UAdd(bqsMsgLen, BQS_MSG_HEAD_SIZE, reqLength, isOverflow);
79 22 : if (isOverflow) {
80 0 : return BQS_STATUS_INNER_ERROR;
81 : }
82 :
83 22 : std::unique_ptr<char_t[]> reqData(new (std::nothrow) char_t[reqLength], std::default_delete<char_t[]>());
84 22 : if (reqData == nullptr) {
85 0 : BQS_LOG_ERROR("Malloc memory error, reqData is nullptr");
86 0 : return BQS_STATUS_INNER_ERROR;
87 : }
88 : // met Exceptions, no need to checks for security functions
89 22 : const auto ret = memset_s(reqData.get(), static_cast<size_t>(reqLength), 0, static_cast<size_t>(reqLength));
90 22 : if (ret != EOK) {
91 1 : BQS_LOG_ERROR("memset_s fail, ret is %d.", ret);
92 1 : return BQS_STATUS_INNER_ERROR;
93 : }
94 : // Add msg length to check
95 21 : *(PtrToPtr<char_t, uint32_t>(reqData.get())) = bqsMsgLen + BQS_MSG_HEAD_SIZE;
96 :
97 21 : if (!bqsReqMsg.SerializePartialToArray(reqData.get() + BQS_MSG_HEAD_SIZE, static_cast<int32_t>(bqsMsgLen))) {
98 1 : BQS_LOG_ERROR("serialize bqsReqMsg fail.");
99 1 : return BQS_STATUS_INNER_ERROR;
100 : }
101 :
102 20 : EzcomRequest req = {.id = 0U, .data = PtrToPtr<char_t, uint8_t>(reqData.get()), .size = reqLength};
103 20 : struct EzcomResponse resp = {0U};
104 : // Send msg and get response
105 20 : BQS_LOG_INFO("EzcomRPCSync begin, fd:%d, msg size:%u", clientFd_, reqLength);
106 20 : int32_t err = EzcomRPCSync(clientFd_, &req, &resp);
107 20 : if (err == -EAGAIN) {
108 1 : err = EzcomRPCSync(clientFd_, &req, &resp);
109 1 : BQS_LOG_INFO("Need to retry ezcom send, fd=%d", clientFd_);
110 : }
111 :
112 : // scope resp.data
113 0 : const ScopeGuard respDataGuard([&resp]() {
114 20 : if (resp.data != nullptr) {
115 16 : delete[] resp.data;
116 16 : resp.data = nullptr;
117 : }
118 20 : });
119 :
120 20 : if ((err < 0) || (resp.data == nullptr)) {
121 4 : BQS_LOG_ERROR("EasyRPCSync failed: %d", err);
122 4 : return BQS_STATUS_EASY_COMM_ERROR;
123 : }
124 :
125 16 : if (resp.size < BQS_MSG_HEAD_SIZE) {
126 0 : BQS_LOG_ERROR(
127 : "EasyRPCSync response size:%u error, should be not less than head size:%u.", resp.size, BQS_MSG_HEAD_SIZE);
128 0 : return BQS_STATUS_EASY_COMM_ERROR;
129 : }
130 :
131 16 : BQS_LOG_INFO("EzcomRPCSync end, response id:%u, response length:%u", resp.id, resp.size);
132 :
133 : // Check response msg
134 16 : const uint32_t currMsgSize = *(PtrToPtr<uint8_t, uint32_t>(resp.data));
135 16 : if (currMsgSize != resp.size) {
136 4 : BQS_LOG_ERROR("message error, head msg content:%u, response size:%u", currMsgSize, resp.size);
137 4 : return BQS_STATUS_EASY_COMM_ERROR;
138 : }
139 :
140 : // Parse response msg to BQSMsg
141 12 : char_t* const respData = PtrToPtr<uint8_t, char_t>(resp.data);
142 12 : const uint32_t parseLength = currMsgSize - BQS_MSG_HEAD_SIZE;
143 12 : if (!bqsRespMsg.ParseFromArray(respData + BQS_MSG_HEAD_SIZE, static_cast<int32_t>(parseLength))) {
144 1 : BQS_LOG_ERROR("parse bqsRespMsg fail.");
145 1 : return BQS_STATUS_INNER_ERROR;
146 : }
147 11 : return BQS_STATUS_OK;
148 22 : }
149 :
150 : /* *
151 : * Parse bind response BQSMsg message
152 : * @return number of bind relation success
153 : */
154 9 : uint32_t EzcomClient::ParseBindRespMsg(BQSMsg& bqsRespMsg, std::vector<BQSBindQueueResult>& bindResultVec) const
155 : {
156 9 : uint32_t bindNum = 0U;
157 9 : const BQSBindQueueRsps* const bqsBindQueueRspBuff = bqsRespMsg.mutable_resp_msgs();
158 :
159 9 : BQS_LOG_INFO(
160 : "Bind relation [add/del], stage [client], type [response], relation [size:%d]",
161 : bqsBindQueueRspBuff->bind_result_vec_size());
162 1047 : for (int32_t i = 0; i < bqsBindQueueRspBuff->bind_result_vec_size(); i++) {
163 1038 : const BQSBindQueueRsp bqsBindQueueEzRsp = bqsBindQueueRspBuff->bind_result_vec(i);
164 1038 : const BQSBindQueueResult bindResult = {bqsBindQueueEzRsp.bind_result()};
165 1038 : bindResultVec.push_back(bindResult);
166 :
167 1038 : const int32_t result = bqsBindQueueEzRsp.bind_result();
168 1038 : BQS_LOG_INFO(
169 : "Bind relation [add/del], stage [client], type [response], relation [index:%d, result:%d].", i, result);
170 1038 : if (result == BQS_STATUS_OK) {
171 1033 : bindNum++;
172 : }
173 1038 : }
174 9 : return bindNum;
175 : }
176 :
177 : /* *
178 : * Assembly unbind BQSMsg message
179 : * @return BQS_STATUS_OK: success, other: failed
180 : */
181 7 : BqsStatus EzcomClient::SerializeUnbindMsg(const std::vector<BQSQueryPara>& bqsQueryParaVec, BQSMsg& bqsClientMsg) const
182 : {
183 7 : BQS_LOG_INFO("Bind relation [del], stage [client], type [request], relation [size:%zu]", bqsQueryParaVec.size());
184 7 : if (bqsQueryParaVec.size() == 0U) {
185 1 : BQS_LOG_WARN("The number of bind relation to be deleted should be greater than zero.");
186 1 : return BQS_STATUS_PARAM_INVALID;
187 : }
188 :
189 6 : bqsClientMsg.set_msg_type(BQSMsg::UNBIND);
190 6 : BQSQueryMsgs* const bqsQueryMsgBuff = bqsClientMsg.mutable_query_msgs();
191 :
192 519 : for (size_t i = 0U; i < bqsQueryParaVec.size(); i++) {
193 513 : BQSQueryMsg* const bqsQueryEzMsg = bqsQueryMsgBuff->add_query_msg_vec();
194 :
195 513 : bqsQueryEzMsg->set_key_type(static_cast<BQSQueryMsg::QsQueryType>(bqsQueryParaVec[i].keyType_));
196 513 : BQSBindQueueMsg* const bqsBindQueueEzMsg = bqsQueryEzMsg->mutable_bind_queue_item();
197 :
198 513 : bqsBindQueueEzMsg->set_src_queue_id(bqsQueryParaVec[i].bqsBindQueueItem_.srcQueueId_);
199 513 : bqsBindQueueEzMsg->set_dst_queue_id(bqsQueryParaVec[i].bqsBindQueueItem_.dstQueueId_);
200 :
201 513 : BQS_LOG_INFO(
202 : "Bind relation [del], stage [client], type [request], relation [type{0:src, 1:dst, 2:src-dst}:%d, "
203 : "src:%u, dst:%u]",
204 : bqsQueryParaVec[i].keyType_, bqsQueryParaVec[i].bqsBindQueueItem_.srcQueueId_,
205 : bqsQueryParaVec[i].bqsBindQueueItem_.dstQueueId_);
206 : }
207 6 : return BQS_STATUS_OK;
208 : }
209 :
210 : /**
211 : * Assembly get bind BQSMsg message
212 : * @return BQS_STATUS_OK: success, other: failed
213 : */
214 4 : BqsStatus EzcomClient::SerializeGetBindMsg(const BQSQueryPara& queryPara, BQSMsg& bqsClientMsg) const
215 : {
216 4 : BQS_LOG_INFO("Bind relation [get], stage [client], type [request]");
217 4 : bqsClientMsg.set_msg_type(BQSMsg::GET_BIND);
218 4 : BQSQueryMsg* const bqsQueryEzMsg = bqsClientMsg.mutable_query_msg();
219 :
220 4 : bqsQueryEzMsg->set_key_type(static_cast<BQSQueryMsg::QsQueryType>(queryPara.keyType_));
221 4 : BQSBindQueueMsg* const bqsBindQueueEzMsg = bqsQueryEzMsg->mutable_bind_queue_item();
222 :
223 4 : bqsBindQueueEzMsg->set_src_queue_id(queryPara.bqsBindQueueItem_.srcQueueId_);
224 4 : bqsBindQueueEzMsg->set_dst_queue_id(queryPara.bqsBindQueueItem_.dstQueueId_);
225 4 : BQS_LOG_INFO(
226 : "Bind relation [get], stage [client], type [request], relation [type{0:src, 1:dst, 2:src-dst}:%d, "
227 : "src:%u, dst:%u]",
228 : queryPara.keyType_, queryPara.bqsBindQueueItem_.srcQueueId_, queryPara.bqsBindQueueItem_.dstQueueId_);
229 4 : return BQS_STATUS_OK;
230 : }
231 :
232 : /**
233 : * Parse get bind response BQSMsg message
234 : * @return number of bind relation
235 : */
236 2 : uint32_t EzcomClient::ParseGetBindRespMsg(BQSMsg& bqsRespMsg, std::vector<BQSBindQueueItem>& bindQueueVec) const
237 : {
238 2 : uint32_t bindNum = 0U;
239 2 : const BQSBindQueueMsgs* const bqsBindQueueMsgBuff = bqsRespMsg.mutable_bind_queue_msgs();
240 :
241 2 : bindNum = static_cast<uint32_t>(bqsBindQueueMsgBuff->bind_queue_vec_size());
242 2 : BQS_LOG_INFO("Bind relation [get], stage [client], type [response], relation [size:%u]", bindNum);
243 12 : for (uint32_t i = 0U; i < bindNum; i++) {
244 10 : const BQSBindQueueMsg bqsBindQueueEzMsg = bqsBindQueueMsgBuff->bind_queue_vec(static_cast<int32_t>(i));
245 10 : const BQSBindQueueItem bindItem = {bqsBindQueueEzMsg.src_queue_id(), bqsBindQueueEzMsg.dst_queue_id()};
246 10 : bindQueueVec.push_back(bindItem);
247 10 : BQS_LOG_INFO(
248 : "Bind relation [get], stage [client], type [response], relation [index:%u, src:%u, dst:%u]", i,
249 : bindItem.srcQueueId_, bindItem.dstQueueId_);
250 10 : }
251 2 : return bindNum;
252 : }
253 :
254 : /* *
255 : * Assembly get paged bind BQSMsg message
256 : * @return BQS_STATUS_OK: success, other: failed
257 : */
258 5 : BqsStatus EzcomClient::SerializeGetPagedBindMsg(const uint32_t offset, const uint32_t limit, BQSMsg& bqsClientMsg) const
259 : {
260 5 : BQS_LOG_INFO("Bind relation [get_paged], stage [client], type [request]");
261 5 : bqsClientMsg.set_msg_type(BQSMsg::GET_ALL_BIND);
262 5 : BQSPagedMsg* const bqsPagedEzMsg = bqsClientMsg.mutable_paged_msg();
263 5 : bqsPagedEzMsg->set_offset(offset);
264 5 : bqsPagedEzMsg->set_limit(limit);
265 5 : return BQS_STATUS_OK;
266 : }
267 :
268 : /**
269 : * Parse get paged bind response BQSMsg message
270 : * @return number of bind relation
271 : */
272 3 : uint32_t EzcomClient::ParseGetPagedBindRespMsg(
273 : BQSMsg& bqsRespMsg, std::vector<BQSBindQueueItem>& bindQueueVec, uint32_t& total) const
274 : {
275 3 : uint32_t bindNum = 0U;
276 3 : const BQSBindQueueMsgs* const bqsBindQueueMsgBuff = bqsRespMsg.mutable_bind_queue_msgs();
277 :
278 3 : const BQSPagedMsg* const bqsPagedEzMsg = bqsRespMsg.mutable_paged_msg();
279 3 : total = bqsPagedEzMsg->total();
280 :
281 3 : bindNum = static_cast<uint32_t>(bqsBindQueueMsgBuff->bind_queue_vec_size());
282 3 : BQS_LOG_INFO("Bind relation [get_paged], stage [client], type [response], relation [size:%u]", bindNum);
283 15 : for (uint32_t i = 0U; i < bindNum; i++) {
284 12 : const BQSBindQueueMsg bqsBindQueueEzMsg = bqsBindQueueMsgBuff->bind_queue_vec(static_cast<int32_t>(i));
285 12 : const BQSBindQueueItem bindItem = {bqsBindQueueEzMsg.src_queue_id(), bqsBindQueueEzMsg.dst_queue_id()};
286 12 : bindQueueVec.push_back(bindItem);
287 12 : BQS_LOG_INFO(
288 : "Bind relation [get_paged], stage [client], type [response], relation [index:%u, src:%u, dst:%u]", i,
289 : bindItem.srcQueueId_, bindItem.dstQueueId_);
290 12 : }
291 3 : return bindNum;
292 : }
293 : } // namespace bqs
|
