From 900c0972923ac0eeac54425f128d38dffefd381b Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期一, 30 十一月 2020 12:06:41 +0800
Subject: [PATCH] sendandrecv改为线程安全的
---
src/socket/net_mod_socket.c | 210 +++++++++++++++++++++++----------------------------
1 files changed, 95 insertions(+), 115 deletions(-)
diff --git a/src/socket/net_mod_socket.c b/src/socket/net_mod_socket.c
index 5ddf221..7b9b577 100644
--- a/src/socket/net_mod_socket.c
+++ b/src/socket/net_mod_socket.c
@@ -9,18 +9,12 @@
NetModSocket::NetModSocket()
{
- init_req_rep_req_resp_pool();
-
if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR) err_msg(errno, "signal");
}
NetModSocket::~NetModSocket() {
- int clientfd;
- for (auto map_iter = req_resp_pool.connectionMap.begin(); map_iter != req_resp_pool.connectionMap.end(); map_iter++) {
- clientfd = map_iter->second;
- Close(clientfd);
- }
+
}
@@ -54,20 +48,83 @@
}
+
+
+int NetModSocket::connect( pool &mpool, net_node_t *node) {
+
+ int connfd;
+ int i;
+ char portstr[32];
+
+ // printf("mis: %s\n", mapKey);
+ sprintf(portstr, "%d", node->port);
+// printf("open before: %s\n", mapKey);
+ connfd = open_clientfd(node->host, portstr);
+// printf("open after: %s\n", mapKey);
+ if(connfd < 0) {
+ LoggerFactory::getLogger()->error(errno, "NetModSocket::connect %s:%d ", node->host, node->port);
+ return -1;
+ }
+
+ for (i = 0; i < OPEN_MAX; i++) { /* Find an available slot */
+ if (mpool.conns[i].fd < 0)
+ {
+ /* Add connected descriptor to the mpool */
+ mpool.conns[i].fd = connfd;
+
+ mpool.conns[i].events = POLLIN;
+ /* Add the descriptor to descriptor set */
+ break;
+ }
+ }
+
+ if (i > mpool.maxi)
+ mpool.maxi = i;
+
+ if (i == OPEN_MAX) {
+ /* Couldn't find an empty slot */
+ LoggerFactory::getLogger()->error(errno, "add_client error: Too many clients");
+ return -1;
+ }
+
+ return connfd;
+}
+
+
+void NetModSocket::close_connect(pool &mpool, int connfd) {
+ int i;
+ if(close(connfd) != 0) {
+ LoggerFactory::getLogger()->error(errno, "NetModSocket::close_connect close");
+ }
+
+
+ for (i = 0; i <= mpool.maxi; i++) {
+ if(mpool.conns[i].fd == connfd) {
+ mpool.conns[i].fd = -1;
+ }
+ }
+
+ LoggerFactory::getLogger()->debug( "closed %d\n", connfd);
+
+}
+
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 timeout = 5 * 1000) {
- int i, n, recv_size, connfd;
+ int i, n, recv_size, connfd;
net_node_t *node;
void *recv_buf;
+
+ pool mpool;
net_mod_request_head_t request_head = {};
int n_req = 0, n_recv_suc = 0, n_resp;
+
net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t));
- //init_req_rep_req_resp_pool();
+ init_conn_pool(mpool);
for (i = 0; i< arrlen; i++) {
@@ -84,7 +141,7 @@
continue;
}
- if( (connfd = connect(node)) < 0 ) {
+ if( (connfd = connect(mpool, node)) < 0 ) {
continue;
}
@@ -99,15 +156,15 @@
// printf("write_request %s:%d\n", request_head.host, request_head.port);
if(write_request(connfd, request_head, send_buf, send_size) != 0) {
LoggerFactory::getLogger()->error("write_request failture %s:%d\n", node->host, node->port);
- close_connect(connfd);
- // req_resp_pool.conns[i].fd = -1;
+ close_connect(mpool, connfd);
+ // mpool.conns[i].fd = -1;
} else {
n_req++;
}
}
-// printf(" req_resp_pool.maxi = %d\n", req_resp_pool.maxi);
+// printf(" mpool.maxi = %d\n", mpool.maxi);
// printf(" n_req = %d\n", n_req);
// int tmp = 0;
@@ -115,29 +172,29 @@
{
// printf(" while %d\n", tmp++);
/* Wait for listening/connected descriptor(s) to become ready */
- if( (req_resp_pool.nready = poll(req_resp_pool.conns, req_resp_pool.maxi + 1, timeout) ) <= 0) {
+ if( (mpool.nready = poll(mpool.conns, mpool.maxi + 1, timeout) ) <= 0) {
// wirite_set 鍜� read_set 鍦ㄦ寚瀹氭椂闂村唴閮芥病鍑嗗濂�
break;
}
-// printf("req_resp_pool.nready =%d\n", req_resp_pool.nready);
- for (i = 0; (i <= req_resp_pool.maxi) && (req_resp_pool.nready > 0); i++) {
- if ( (connfd = req_resp_pool.conns[i].fd) > 0 ) {
+// printf("mpool.nready =%d\n", mpool.nready);
+ for (i = 0; (i <= mpool.maxi) && (mpool.nready > 0); i++) {
+ if ( (connfd = mpool.conns[i].fd) > 0 ) {
/* If the descriptor is ready, echo a text line from it */
- if (req_resp_pool.conns[i].revents & POLLIN )
+ if (mpool.conns[i].revents & POLLIN )
{
- req_resp_pool.nready--;
+ mpool.nready--;
// printf("POLLIN %d\n", connfd);
if( (n = read_response(connfd, ret_arr+n_recv_suc)) == 0) {
// 鎴愬姛鏀跺埌杩斿洖娑堟伅锛屾竻绌鸿鍏ヤ綅
- req_resp_pool.conns[i].fd = -1;
+ mpool.conns[i].fd = -1;
n_recv_suc++;
} else if(n == -1) {
- req_resp_pool.conns[i].fd = -1;
- close_connect(connfd);
+ mpool.conns[i].fd = -1;
+ close_connect(mpool, connfd);
} else {
- req_resp_pool.conns[i].fd = -1;
+ mpool.conns[i].fd = -1;
}
n_resp++;
@@ -145,30 +202,30 @@
}
- if (req_resp_pool.conns[i].revents & POLLOUT ) {
+ if (mpool.conns[i].revents & POLLOUT ) {
// printf("poll POLLOUT %d\n", connfd);
}
- if (req_resp_pool.conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) )
+ if (mpool.conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) )
{
// printf("poll POLLERR %d\n", connfd);
- req_resp_pool.nready--;
- close_connect(connfd);
- req_resp_pool.conns[i].fd = -1;
+ mpool.nready--;
+ close_connect(mpool, connfd);
+ mpool.conns[i].fd = -1;
}
}
}
}
- for (i = 0; i <= req_resp_pool.maxi; i++) {
- if ( (connfd = req_resp_pool.conns[i].fd) > 0 ) {
+ for (i = 0; i <= mpool.maxi; i++) {
+ if ( (connfd = mpool.conns[i].fd) > 0 ) {
// 鍏抽棴骞舵竻闄ゅ啓鍏ユ垨璇诲彇澶辫触鐨勮繛鎺�
- close_connect(connfd);
- req_resp_pool.conns[i].fd = -1;
+ close_connect(mpool, connfd);
+ mpool.conns[i].fd = -1;
}
}
- req_resp_pool.maxi = -1;
+ mpool.maxi = -1;
*recv_arr = ret_arr;
if(recv_arr_size != NULL) {
@@ -598,96 +655,18 @@
-void NetModSocket::init_req_rep_req_resp_pool()
+void NetModSocket::init_conn_pool(pool& mpool)
{
/* Initially, there are no connected descriptors */
int i;
- req_resp_pool.maxi = -1; //line:conc:echoservers:beginempty
+ mpool.maxi = -1; //line:conc:echoservers:beginempty
for (i = 0; i < OPEN_MAX; i++) {
- req_resp_pool.conns[i].fd = -1;
- req_resp_pool.conns[i].events = 0;
+ mpool.conns[i].fd = -1;
+ mpool.conns[i].events = 0;
}
}
-int NetModSocket::connect( net_node_t *node) {
- std::map<std::string, int>::iterator mapIter;
- int connfd;
- int i;
- char mapKey[256];
- char portstr[32];
-
- sprintf(mapKey, "%s:%d", node->host, node->port);
- mapIter = req_resp_pool.connectionMap.find(mapKey);
- if( mapIter != req_resp_pool.connectionMap.end()) {
- connfd = mapIter->second;
-// printf("hit: %s\n", mapKey);
- } else {
-// printf("mis: %s\n", mapKey);
- sprintf(portstr, "%d", node->port);
-// printf("open before: %s\n", mapKey);
- connfd = open_clientfd(node->host, portstr);
-// printf("open after: %s\n", mapKey);
- if(connfd < 0) {
- LoggerFactory::getLogger()->error(errno, "connect %s:%d ", node->host, node->port);
- return -1;
- }
- req_resp_pool.connectionMap.insert({mapKey, connfd});
- }
-
-
- for (i = 0; i < OPEN_MAX; i++) { /* Find an available slot */
- if (req_resp_pool.conns[i].fd < 0)
- {
- /* Add connected descriptor to the req_resp_pool */
- req_resp_pool.conns[i].fd = connfd;
-
- req_resp_pool.conns[i].events = POLLIN;
- /* Add the descriptor to descriptor set */
- break;
- }
- }
-
- if (i > req_resp_pool.maxi)
- req_resp_pool.maxi = i;
-
- if (i == OPEN_MAX) {
- /* Couldn't find an empty slot */
- LoggerFactory::getLogger()->error(errno, "add_client error: Too many clients");
- return -1;
- }
-
-
- return connfd;
-}
-
-void NetModSocket::close_connect(int connfd) {
- int i;
- char mapKey[256];
- std::map<std::string, int>::iterator map_iter;
- if(close(connfd) != 0) {
- LoggerFactory::getLogger()->error(errno, "NetModSocket::close_connect close");
- }
-
-
- for (i = 0; i <= req_resp_pool.maxi; i++) {
- if(req_resp_pool.conns[i].fd == connfd) {
- req_resp_pool.conns[i].fd = -1;
- }
- }
-
- for ( map_iter = req_resp_pool.connectionMap.begin(); map_iter != req_resp_pool.connectionMap.end(); ) {
- if(connfd == map_iter->second) {
-// std::cout << "map_iter->first==" << map_iter->first << std::endl;
- map_iter = req_resp_pool.connectionMap.erase(map_iter);
- } else {
- ++map_iter;
- }
- }
-
- LoggerFactory::getLogger()->debug( "closed %d\n", connfd);
-
-}
/**
uint32_t mod;
char host[NI_MAXHOST];
@@ -795,3 +774,4 @@
return head;
}
+
--
Gitblit v1.8.0