From 578e15c276d72bfbdd707c6c948824daa43d3780 Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期四, 18 一月 2024 14:18:35 +0800
Subject: [PATCH] less memory
---
src/net/net_mod_socket.cpp | 217 +++++++++++++++++++++++++++++++++---------------------
1 files changed, 132 insertions(+), 85 deletions(-)
diff --git a/src/net/net_mod_socket.cpp b/src/net/net_mod_socket.cpp
index a863f6d..d10fbdc 100644
--- a/src/net/net_mod_socket.cpp
+++ b/src/net/net_mod_socket.cpp
@@ -15,37 +15,14 @@
NetModSocket::NetModSocket()
{
- int s;
if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)
logger->error(errno, "NetModSocket::NetModSocket signal");
-
- // gpool = new NetConnPool();
-
- // pthread_mutexattr_t mtxAttr;
- // s = pthread_mutexattr_init(&mtxAttr);
- // if (s != 0)
- // err_exit(s, "pthread_mutexattr_init");
- // s = pthread_mutexattr_settype(&mtxAttr, PTHREAD_MUTEX_ERRORCHECK);
- // if (s != 0)
- // err_exit(s, "pthread_mutexattr_settype");
- // s = pthread_mutex_init(&sendMutex, &mtxAttr);
- // if (s != 0)
- // err_exit(s, "pthread_mutex_init");
-
- // s = pthread_mutexattr_destroy(&mtxAttr);
- // if (s != 0)
- // err_exit(s, "pthread_mutexattr_destroy");
}
NetModSocket::~NetModSocket() {
- // int s;
- // delete gpool;
- // s = pthread_mutex_destroy(&sendMutex);
- // if(s != 0) {
- // err_exit(s, "shm_socket_close");
- // }
+
}
@@ -69,19 +46,40 @@
return shmModSocket.force_bind(key);
}
-int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
- net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
- return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, -1);
+int NetModSocket::reg(void *pData, int len, void **buf, int *size, const int timeout_ms, int flag) {
+
+ return shmModSocket.reg(pData, len, buf, size, timeout_ms, flag);
}
-int NetModSocket::sendandrecv_timeout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
- net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int msec) {
- return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, msec);
-}
-int NetModSocket::sendandrecv_nowait(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
- net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
- return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, 0);
+void NetModSocket::int_set(int data) {
+ int_val = data;
}
+
+void NetModSocket::svr_set(int data) {
+ svr_val = data;
+}
+
+int NetModSocket::int_get(void) {
+ return int_val;
+}
+
+int NetModSocket::svr_get(void) {
+ return svr_val;
+}
+
+// int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
+// net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
+// return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, -1);
+// }
+// int NetModSocket::sendandrecv_timeout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
+// net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int msec) {
+// return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, msec);
+// }
+// int NetModSocket::sendandrecv_nowait(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
+// net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
+// return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, 0);
+
+// }
/* Free thread-specific data buffer */
@@ -90,7 +88,7 @@
NetConnPool *mpool = (NetConnPool *)_pool;
delete mpool;
- logger->debug("destory connPool");
+
}
/* One-time key creation function */
@@ -123,7 +121,6 @@
if (mpool == NULL)
{
/* If first call from this thread, allocate buffer for thread, and save its location */
- logger->debug("Create connPool");
mpool = new NetConnPool();
if (mpool == NULL) {
LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ malloc");
@@ -147,21 +144,22 @@
-int NetModSocket::_sendandrecv_(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
- net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int msec ) {
+int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size,
+ net_mod_recv_msg_t ** recv_arr, int *recv_arr_size,
+ net_mod_err_t ** _err_arr, int *_err_arr_size, int msec ) {
int i, recv_size, connfd;
net_node_t *node;
void *recv_buf = NULL;
struct timespec timeout;
int ret;
- int n_req = 0, n_recv_suc = 0, n_recv_err = 0, n_resp =0;
+ int n_req = 0, n_recv_suc = 0, n_err = 0, n_resp =0;
net_mod_request_head_t request_head = {};
net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t));
- net_mod_recv_err_t err_arr[arrlen];
+ net_mod_err_t *err_arr = (net_mod_err_t *)calloc(arrlen, sizeof(net_mod_err_t));
NetConnPool *mpool = _get_pool();
@@ -189,11 +187,11 @@
ret_arr[n_recv_suc].content_length = recv_size;
n_recv_suc++;
} else {
- err_arr[n_recv_err].port = 0;
- err_arr[n_recv_err].key = node->key;
- err_arr[n_recv_err].code = ret;
- n_recv_err++;
- logger->error("NetModSocket:: %d _sendandrecv_ to key %d failed, %s", get_key(), node->key, bus_strerror(ret));
+ err_arr[n_err].port = 0;
+ err_arr[n_err].key = node->key;
+ err_arr[n_err].code = ret;
+ n_err++;
+ // logger->error("NetModSocket:: _sendandrecv_ to key %d failed. %s", node->key, bus_strerror(ret));
}
@@ -201,11 +199,11 @@
}
if( (connfd = mpool->getConn(node->host, node->port)) < 0 ) {
- memcpy(err_arr[n_recv_err].host, node->host, sizeof(err_arr[n_recv_err].host));
- err_arr[n_recv_err].port = node->port;
- err_arr[n_recv_err].key = node->key;
- err_arr[n_recv_err].code = EBUS_NET;
- n_recv_err++;
+ memcpy(err_arr[n_err].host, node->host, sizeof(err_arr[n_err].host));
+ err_arr[n_err].port = node->port;
+ err_arr[n_err].key = node->key;
+ err_arr[n_err].code = EBUS_NET;
+ n_err++;
continue;
}
@@ -220,11 +218,11 @@
// printf("write_request %s:%d\n", request_head.host, request_head.port);
if(write_request(connfd, request_head, send_buf, send_size, NULL, 0) != 0) {
LoggerFactory::getLogger()->error("write_request failture %s:%d\n", node->host, node->port);
- memcpy(err_arr[n_recv_err].host, node->host, sizeof(err_arr[n_recv_err].host));
- err_arr[n_recv_err].port = node->port;
- err_arr[n_recv_err].key = node->key;
- err_arr[n_recv_err].code = EBUS_NET;
- n_recv_err++;
+ memcpy(err_arr[n_err].host, node->host, sizeof(err_arr[n_err].host));
+ err_arr[n_err].port = node->port;
+ err_arr[n_err].key = node->key;
+ err_arr[n_err].code = EBUS_NET;
+ n_err++;
mpool->closeConn( connfd);
} else {
n_req++;
@@ -250,7 +248,7 @@
{
mpool->nready--;
// printf("POLLIN %d\n", connfd);
- if( (ret = read_response(connfd, ret_arr+n_recv_suc, err_arr + n_recv_err)) == 0) {
+ if( (ret = read_response(connfd, ret_arr+n_recv_suc, err_arr + n_err)) == 0) {
n_recv_suc++;
// 鎴愬姛鏀跺埌杩斿洖娑堟伅锛屾竻绌鸿鍏ヤ綅
mpool->conns[i].fd = -1;
@@ -261,14 +259,14 @@
logger->error("NetModSocket::_sendandrecv_ read_response key = %d , %s", get_key(), bus_strerror(ret));
mpool->closeConn( connfd);
- n_recv_err++;
+ n_err++;
// mpool->conns[i].fd = -1;
} else {
// 浠ｇ悊鏈嶅姟娌℃湁杞彂鎴愬姛
logger->error("NetModSocket::_sendandrecv_ read_response key = %d , %s", get_key(), bus_strerror(ret));
mpool->conns[i].fd = -1;
- n_recv_err++;
+ n_err++;
}
n_resp++;
@@ -300,19 +298,71 @@
mpool->maxi = -1;
+
if(recv_arr != NULL) {
- *recv_arr = ret_arr;
+
+ if(n_recv_suc > 0) {
+ *recv_arr = ret_arr;
+
+ } else {
+ free_recv_msg_arr(ret_arr, n_recv_suc);
+ }
+
} else {
free_recv_msg_arr(ret_arr, n_recv_suc);
}
-
+
if(recv_arr_size != NULL) {
- *recv_arr_size = n_recv_suc;
+ *recv_arr_size = n_recv_suc;
+ }
+
+
+ if(_err_arr != NULL) {
+
+ if(n_err > 0) {
+ *_err_arr = err_arr;
+
+
+ } else {
+ *_err_arr = NULL;
+ *_err_arr_size = 0;
+ free(err_arr);
+ }
+
+ } else {
+ free(err_arr);
+ }
+
+ if(_err_arr_size != NULL) {
+ *_err_arr_size = n_err;
}
return n_recv_suc;
}
+void NetModSocket::buf_data_set(std::string str, int val) {
+ recvbuf.insert({str, val});
+}
+
+int NetModSocket::buf_data_get(std::string str) {
+
+ int i;
+ int val = 0;
+ std::map<std::string, int>::iterator recvIter;
+
+ recvIter = recvbuf.find(str);
+ if(recvIter != recvbuf.end()) {
+
+ val = recvIter->second;
+
+ }
+
+ return val;
+}
+
+void NetModSocket::buf_data_del(std::string str) {
+ recvbuf.erase(str);
+}
void NetModSocket::free_recv_msg_arr(net_mod_recv_msg_t * arr, size_t size) {
@@ -336,9 +386,6 @@
return _pub_(node_arr, arrlen, topic, topic_size, content, content_size, msec);
}
-
-// int pub(char *topic, int topic_size, void *content, int content_size, int port);
-
int NetModSocket::_pub_(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content,
int content_size, int msec) {
int i, connfd;
@@ -353,10 +400,10 @@
int ret;
NetConnPool *mpool = _get_pool();
- net_mod_recv_err_t err_msg;
+ net_mod_err_t err_msg;
// 鏈湴鍙戦��
- if(node_arr == NULL || arrlen == 0) {
+ if ((node_arr == NULL) || (arrlen == 0)) {
if(msec == 0) {
ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, NULL, BUS_NOWAIT_FLAG);
} else if(msec > 0) {
@@ -480,20 +527,20 @@
* @key 鍙戦�佺粰璋�
* @return 0 鎴愬姛锛� 鍏朵粬鍊� 澶辫触鐨勯敊璇爜
*/
-int NetModSocket::sendto(const void *buf, const int size, const int key){
- return shmModSocket.sendto(buf, size, key);
+int NetModSocket::sendto(const void *buf, const int size, const int key, int reset, int data_set){
+ return shmModSocket.sendto(buf, size, key, 0, 0, reset, data_set);
}
// 鍙戦�佷俊鎭秴鏃惰繑鍥炪�� @sec 绉� 锛� @nsec 绾崇
-int NetModSocket::sendto_timeout(const void *buf, const int size, const int key, int sec, int nsec){
+int NetModSocket::sendto_timeout(const void *buf, const int size, const int key, int sec, int nsec, int reset, int data_set){
struct timespec timeout = {sec, nsec};
- return shmModSocket.sendto(buf, size, key, &timeout, BUS_TIMEOUT_FLAG);
+ return shmModSocket.sendto(buf, size, key, &timeout, BUS_TIMEOUT_FLAG, reset, data_set);
}
// 鍙戦�佷俊鎭珛鍒昏繑鍥炪��
-int NetModSocket::sendto_nowait(const void *buf, const int size, const int key){
- return shmModSocket.sendto(buf, size, key, NULL, BUS_NOWAIT_FLAG);
+int NetModSocket::sendto_nowait(const void *buf, const int size, const int key, int reset, int data_set){
+ return shmModSocket.sendto(buf, size, key, NULL, BUS_NOWAIT_FLAG, reset, data_set);
}
@@ -502,23 +549,22 @@
* @key 浠庤皝鍝噷鏀跺埌鐨勪俊鎭�
* @return 0 鎴愬姛锛� 鍏朵粬鍊� 澶辫触鐨勯敊璇爜
*/
-int NetModSocket::recvfrom(void **buf, int *size, int *key) {
+int NetModSocket::recvfrom(void **buf, int *size, int *key, int reset, int data_set) {
- return shmModSocket.recvfrom(buf, size, key);
+ return shmModSocket.recvfrom(buf, size, key, 0, 0, reset, data_set);
}
// 鎺ュ彈淇℃伅瓒呮椂杩斿洖銆� @sec 绉� 锛� @nsec 绾崇
-int NetModSocket::recvfrom_timeout(void **buf, int *size, int *key, int sec, int nsec){
+int NetModSocket::recvfrom_timeout(void **buf, int *size, int *key, int sec, int nsec, int reset, int data_set){
struct timespec timeout = {sec, nsec};
- return shmModSocket.recvfrom(buf, size, key, &timeout, BUS_TIMEOUT_FLAG);
+ return shmModSocket.recvfrom(buf, size, key, &timeout, BUS_TIMEOUT_FLAG, reset, data_set);
}
-int NetModSocket::recvfrom_nowait(void **buf, int *size, int *key){
- return shmModSocket.recvfrom(buf, size, key, NULL, BUS_NOWAIT_FLAG);
+int NetModSocket::recvfrom_nowait(void **buf, int *size, int *key, int reset, int data_set){
+ return shmModSocket.recvfrom(buf, size, key, NULL, BUS_NOWAIT_FLAG, reset, data_set);
}
-
int NetModSocket::recvandsend(recvandsend_callback_fn callback,
const struct timespec *timeout , int flag, void * user_data ) {
@@ -659,7 +705,7 @@
* @return 0 鎴愬姛, EBUS_NET 缃戠粶閿欒锛� 鍏朵粬鍊� 浠ｇ悊鏈嶅姟娌℃湁杞彂鎴愬姛銆�
*
*/
-int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg, net_mod_recv_err_t *err_arr) {
+int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg, net_mod_err_t *err_arr) {
int recv_size;
void *recv_buf;
char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH];
@@ -690,6 +736,7 @@
LoggerFactory::getLogger()->error(errno, "NetModSocket::send malloc recv_buf");
exit(1);
}
+
if ( (recv_size = rio_readn(connfd, recv_buf, response_head.content_length) ) != response_head.content_length) {
memcpy(err_arr->host, response_head.host, sizeof(err_arr->host));
@@ -756,23 +803,23 @@
head.mod = ntohl(GET(tmp_ptr));
- tmp_ptr += 4;
+ tmp_ptr += sizeof(uint32_t);
memcpy(head.host, tmp_ptr, sizeof(head.host));
tmp_ptr += sizeof(head.host);
head.port = ntohl(GET(tmp_ptr));
- tmp_ptr += 4;
+ tmp_ptr += sizeof(uint32_t);
head.key = ntohl(GET(tmp_ptr));
- tmp_ptr += 4;
+ tmp_ptr += sizeof(uint32_t);
head.content_length = ntohl(GET(tmp_ptr));
- tmp_ptr += 4;
+ tmp_ptr += sizeof(uint32_t);
head.topic_length = ntohl(GET(tmp_ptr));
- tmp_ptr += 4;
+ tmp_ptr += sizeof(uint32_t);
head.timeout = ntohl(GET_INT32(tmp_ptr));
return head;
--
Gitblit v1.8.0