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