From 9b29a67af93e4ab9948cc60f743d73c4bb39e738 Mon Sep 17 00:00:00 2001 From: wangzhengquan <wangzhengquan85@126.com> Date: 星期二, 20 十月 2020 16:46:55 +0800 Subject: [PATCH] update --- src/socket/net_mod_socket.c | 605 +++++++++++++++++++++++++++++++++++++++++++++++++----- 1 files changed, 545 insertions(+), 60 deletions(-) diff --git a/src/socket/net_mod_socket.c b/src/socket/net_mod_socket.c index bb50c59..660c1f3 100644 --- a/src/socket/net_mod_socket.c +++ b/src/socket/net_mod_socket.c @@ -1,34 +1,345 @@ #include "net_mod_socket.h" #include "socket_io.h" #include "net_mod_socket_io.h" +#include <sys/types.h> /* See NOTES */ +#include <sys/socket.h> -std::map<std::string, rio_t *> NetModSocket::connectionMap; + 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); + + } } + + +void NetModSocket::init_req_rep_req_resp_pool() +{ + /* Initially, there are no connected descriptors */ + int i; + req_resp_pool.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; + } + +} + +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); + +} + 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 i, n, clientfd; - char portstr[32]; + int i, n, recv_size, connfd; net_node_t *node; - char mapKey[256]; + void *recv_buf; + int timeout = 5 * 1000; + + 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(); + + for (i = 0; i< arrlen; i++) { + + node = &node_arr[i]; + if(node->host == NULL) { + // 鏈湴鍙戦�� + shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size); + strcpy( ret_arr[n_recv_suc].host,"localshm"); + ret_arr[n_recv_suc].port = 0; + ret_arr[n_recv_suc].key = node->key; + ret_arr[n_recv_suc].content = recv_buf; + ret_arr[n_recv_suc].content_length = recv_size; + n_recv_suc++; + continue; + } + + if( (connfd = connect(node)) < 0 ) { + continue; + } + +// printf("write_request %s:%d\n", node->host, node->port); + if(write_request(connfd, node->key, 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; + } else { + n_req++; + } + + } + +// printf(" req_resp_pool.maxi = %d\n", req_resp_pool.maxi); +// printf(" n_req = %d\n", n_req); + +// int tmp = 0; + while(n_resp < n_req) + { +// 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) { + // 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 ) { + /* If the descriptor is ready, echo a text line from it */ + if (req_resp_pool.conns[i].revents & POLLIN ) + { + req_resp_pool.nready--; +// printf("POLLIN %d\n", connfd); + if( (n = read_response(connfd, ret_arr+n_recv_suc)) == 0) { + + // 鎴愬姛鏀跺埌杩斿洖娑堟伅锛屾竻绌鸿鍏ヤ綅 + req_resp_pool.conns[i].fd = -1; + n_recv_suc++; + + } else if(n == -1) { + req_resp_pool.conns[i].fd = -1; + close_connect(connfd); + } else { + req_resp_pool.conns[i].fd = -1; + + } + n_resp++; +// printf("read response %d\n", n); + + } + + if (req_resp_pool.conns[i].revents & POLLOUT ) { + // printf("poll POLLOUT %d\n", connfd); + } + + if (req_resp_pool.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; + } + } + } + } + + for (i = 0; i <= req_resp_pool.maxi; i++) { + if ( (connfd = req_resp_pool.conns[i].fd) > 0 ) { + // 鍏抽棴骞舵竻闄ゅ啓鍏ユ垨璇诲彇澶辫触鐨勮繛鎺� + close_connect(connfd); + req_resp_pool.conns[i].fd = -1; + } + } + + req_resp_pool.maxi = -1; + + *recv_arr = ret_arr; + if(recv_arr_size != NULL) { + *recv_arr_size = n_recv_suc; + } + return n_recv_suc; + +} + +int NetModSocket::write_request(int clientfd, int key, void *send_buf, int send_size) { + net_mod_request_head_t request_head = {}; + int buf_size; + char *buf; + int max_buf_size; + if((buf = (char *)malloc(MAXBUF)) == NULL) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::NetModSocket malloc"); + exit(1); + } else { + max_buf_size = MAXBUF; + } + + buf_size = send_size + NET_MODE_REQUEST_HEAD_LENGTH; + if(max_buf_size < buf_size) { + + if((buf = (char *)realloc(buf, buf_size)) == NULL) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::write_request realloc buf"); + exit(1); + } else { + max_buf_size = buf_size; + } + } + + request_head.mod = REQ_REP; + request_head.key = key; + request_head.content_length = send_size; + request_head.topic_length = 0; + + // optval = 1; + // setsockopt(clientfd, IPPROTO_TCP, TCP_CORK, &optval, sizeof(optval)); + memcpy(buf, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH); + memcpy(buf + NET_MODE_REQUEST_HEAD_LENGTH, send_buf, send_size); + + + if(rio_writen(clientfd, buf, buf_size) != buf_size ) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::write_request rio_writen"); + free(buf); + return -1; + } + free(buf); + return 0; +} + +/** + * @return 0 鎴愬姛, 1 瀵规柟娌℃湁瀵瑰簲鐨刱ey, -1 缃戠粶閿欒 + * + */ +int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg) { + int recv_size; + void *recv_buf; + char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH]; + + net_mod_response_head_t response_head; + if ( rio_readn(connfd, response_head_bs, NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::read_response rio_readnb response_head"); + + return -1; + } + + response_head = NetModSocket::decode_response_head(response_head_bs); + if(response_head.code != 0) { + // 瀵规柟娌℃湁瀵瑰簲鐨刱ey + return 1; + } + + recv_buf = malloc(response_head.content_length); + if(recv_buf == NULL) { + 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) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::read_response rio_readnb recv_buf"); + + return -1; + } + + strcpy( recv_msg->host, response_head.host); + recv_msg->port = response_head.port; + recv_msg->key = response_head.key; + recv_msg->content = recv_buf; + recv_msg->content_length = recv_size; + return 0; +} + +int NetModSocket::sendandrecv_safe(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 i, clientfd; + net_node_t *node; void *recv_buf; int recv_size; char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH]; - net_mod_request_head_t request_head; + net_mod_request_head_t request_head = {}; net_mod_response_head_t response_head; - std::map<std::string, rio_t*>::iterator mapIter; - rio_t *rio; + + + char portstr[32]; + char *buf = NULL; + int buf_size, max_buf_size; + + if(buf == NULL) { + buf = (char *)malloc(MAXBUF); + max_buf_size = MAXBUF; + LoggerFactory::getLogger()->error(errno, "NetModSocket::sendandrecv malloc"); + } + + int nsuc = 0; net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t)); for (i = 0; i< arrlen; i++) { @@ -39,100 +350,274 @@ shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size); goto LABEL_ARR_PUSH; } - sprintf(mapKey, "%s:%d", node->host, node->port); - if( ( mapIter = connectionMap.find(mapKey)) != connectionMap.end()) { - rio = mapIter->second; - } else { - rio = (rio_t *)malloc(sizeof(rio_t)); - sprintf(portstr, "%d", node->port); - clientfd = Open_clientfd(node-> host, portstr); - Rio_readinitb(rio, clientfd); - connectionMap.insert({mapKey, rio}); + sprintf(portstr, "%d", node->port); + clientfd = open_clientfd(node->host, portstr); + if(clientfd < 0) { + continue; } + + buf_size = send_size + NET_MODE_REQUEST_HEAD_LENGTH; + if(max_buf_size < buf_size) { + if((buf = (char *)realloc(buf, buf_size)) == NULL) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::sendandrecv_safe realloc buf"); + } else { + max_buf_size = buf_size; + } + + } + request_head.mod = REQ_REP; request_head.key = node->key; request_head.content_length = send_size; - if(rio_writen(rio->rio_fd, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH) != NET_MODE_REQUEST_HEAD_LENGTH) { - err_exit(errno, "NetModSocket::send head rio_writen"); + request_head.topic_length = 0; + // optval = 1; + // setsockopt(clientfd, IPPROTO_TCP, TCP_CORK, &optval, sizeof(optval)); + memcpy(buf, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH); + memcpy(buf + NET_MODE_REQUEST_HEAD_LENGTH, send_buf, send_size); + + + if(rio_writen(clientfd, buf, buf_size) != buf_size ) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::sendandrecv_safe rio_writen buf"); + + close(clientfd); + continue; } + // optval = 0; + // setsockopt(clientfd, IPPROTO_TCP, TCP_CORK, &optval, sizeof(optval)); - if(rio_writen(rio->rio_fd, send_buf, send_size) != send_size ) { - err_exit(errno, "NetModSocket::send conent rio_writen"); - } - - - if ( rio_readnb(rio, response_head_bs, NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH) { - err_exit(errno, "NetModSocket::send rio_readnb"); + if ( rio_readn(clientfd, response_head_bs, NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::sendandrecv_safe rio_readnb response_head_bs"); + + close(clientfd); + continue; } response_head = NetModSocket::decode_response_head(response_head_bs); + if(response_head.code != 0) { + continue; + } recv_buf = malloc(response_head.content_length); if(recv_buf == NULL) { - err_exit(errno, "NetModSocket::send malloc"); + LoggerFactory::getLogger()->error(errno, "NetModSocket::send malloc"); + exit(1); } - if ( (recv_size = rio_readnb(rio, recv_buf, response_head.content_length) ) != response_head.content_length) { - err_exit(errno, "NetModSocket::send rio_readnb"); + if ( (recv_size = rio_readn(clientfd, recv_buf, response_head.content_length) ) != response_head.content_length) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::sendandrecv_safe rio_readnb recv_buf"); + + close(clientfd); + continue; } LABEL_ARR_PUSH: - strcpy( ret_arr[i].host, node->host); - ret_arr[i].port = node->port; - ret_arr[i].key = node->key; - ret_arr[i].content = recv_buf; - ret_arr[i].content_length = recv_size; - } - *recv_arr = ret_arr; - *recv_arr_size = i; + if(node->host != NULL) { + strcpy(ret_arr[nsuc].host, node->host); + } else { + strcpy(ret_arr[nsuc].host, "local"); + } + + ret_arr[nsuc].port = node->port; + ret_arr[nsuc].key = node->key; + ret_arr[nsuc].content = recv_buf; + ret_arr[nsuc].content_length = recv_size; - return i; + nsuc++; + } + + *recv_arr = ret_arr; + if(recv_arr_size != NULL) { + *recv_arr_size = nsuc; + } + + free(buf); + return nsuc; } +// int pub(char *topic, int topic_size, void *content, int content_size, int port); + +int NetModSocket::pub(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content, int content_size) { + int i, clientfd; + net_node_t *node; + + char *buf; + int max_buf_size, buf_size; + + net_mod_request_head_t request_head; + char portstr[32]; + int nsuc = 0; + + if((buf = (char *)malloc(MAXBUF)) == NULL) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::sendandrecv malloc"); + exit(1); + } else { + max_buf_size = MAXBUF; + } + + for (i = 0; i< arrlen; i++) { + + node = &node_arr[i]; + if(node->host == NULL) { + // 鏈湴鍙戦�� + if(shmModSocket.pub(topic, topic_size, content, content_size, node->key) == 0 ) { + nsuc++; + } + + } else { + sprintf(portstr, "%d", node->port); + clientfd = open_clientfd(node->host, portstr); + if(clientfd < 0) { + continue; + } + + request_head.mod = BUS; + request_head.key = node->key; + request_head.content_length = content_size; + request_head.topic_length = strlen(topic) + 1; + + buf_size = NET_MODE_REQUEST_HEAD_LENGTH + content_size + request_head.topic_length; + + if(max_buf_size < buf_size) { + if( ( buf = (char *)realloc(buf, buf_size)) == NULL) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::pub realloc buf "); + } else { + max_buf_size = buf_size; + } + + } + + memcpy(buf, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH); + memcpy(buf + NET_MODE_REQUEST_HEAD_LENGTH, content, content_size); + memcpy(buf + NET_MODE_REQUEST_HEAD_LENGTH + content_size, topic, request_head.topic_length); + + if(rio_writen(clientfd, buf, buf_size) != buf_size ) { + LoggerFactory::getLogger()->error(errno, "NetModSocket::pub rio_writen conent "); + } else { + nsuc++; + } + close(clientfd); + } + + + } + + free(buf); + return nsuc; +} void NetModSocket::free_recv_msg_arr(net_mod_recv_msg_t * arr, size_t size) { for(int i =0; i< size; i++) { free(arr[i].content); } - + free(arr); } -// ssize_t recv(void *buf, size_t len) { - -// return rio_readlineb(&rio, buf, MAXLINE); - -// } +/** + uint32_t mod; + char host[NI_MAXHOST]; + uint32_t port; + uint32_t key; + uint32_t content_length; + uint32_t topic_length; +*/ void * NetModSocket::encode_request_head(net_mod_request_head_t & request) { - char * head = (char *)malloc(NET_MODE_REQUEST_HEAD_LENGTH); - PUT(head, htonl(request.mod)); - PUT(head + 4, htonl(request.key)); - PUT(head + 8, htonl(request.content_length)); - return head; + void * headbs = malloc(NET_MODE_REQUEST_HEAD_LENGTH); + char *tmp_ptr = (char *)headbs; + PUT(tmp_ptr, htonl(request.mod)); + + tmp_ptr += 4; + memcpy(tmp_ptr, request.host, NI_MAXHOST); + + tmp_ptr += NI_MAXHOST; + PUT(tmp_ptr, htonl(request.port)); + + tmp_ptr += 4; + PUT(tmp_ptr, htonl(request.key)); + + tmp_ptr += 4; + PUT(tmp_ptr, htonl(request.content_length)); + + tmp_ptr += 4; + PUT(tmp_ptr, htonl(request.topic_length)); + + + return headbs; } -net_mod_request_head_t NetModSocket::decode_request_head(void *_headbs) { - char *headbs = (char *)_headbs; +net_mod_request_head_t NetModSocket::decode_request_head(void *headbs) { + char *tmp_ptr = (char *)headbs; net_mod_request_head_t head; - head.mod = ntohl(GET(headbs)); - head.key = ntohl(GET(headbs + 4)); - head.content_length = ntohl(GET(headbs + 8)); + + head.mod = ntohl(GET(tmp_ptr)); + + tmp_ptr += NI_MAXHOST; + memcpy(head.host, tmp_ptr, NI_MAXHOST); + + tmp_ptr += 4; + head.port = ntohl(GET(tmp_ptr)); + + tmp_ptr += 4; + head.key = ntohl(GET(tmp_ptr)); + + tmp_ptr += 4; + head.content_length = ntohl(GET(tmp_ptr)); + + tmp_ptr += 4; + head.topic_length = ntohl(GET(tmp_ptr)); + return head; } +/** + char host[NI_MAXHOST]; + uint32_t port; + uint32_t key; + uint32_t content_length; + uint32_t code; + */ void * NetModSocket::encode_response_head(net_mod_response_head_t & response) { - char * head = (char *)malloc(NET_MODE_RESPONSE_HEAD_LENGTH); - PUT(head, htonl(response.content_length)); - return head; + void * headbs = malloc(NET_MODE_RESPONSE_HEAD_LENGTH); + char *tmp_ptr = (char *)headbs; + + memcpy(tmp_ptr, response.host, NI_MAXHOST); + + tmp_ptr += NI_MAXHOST; + PUT(tmp_ptr, htonl(response.port)); + + tmp_ptr += 4; + PUT(tmp_ptr , htonl(response.key)); + + tmp_ptr += 4; + PUT(tmp_ptr, htonl(response.content_length)); + + tmp_ptr += 4; + PUT(tmp_ptr, htonl(response.code)); + + return headbs; } -net_mod_response_head_t NetModSocket::decode_response_head(void *_headbs) { - char *headbs = (char *)_headbs; +net_mod_response_head_t NetModSocket::decode_response_head(void *headbs) { + char *tmp_ptr = (char *)headbs; net_mod_response_head_t head; - head.content_length = ntohl(GET(headbs)); + + memcpy(head.host, tmp_ptr, NI_MAXHOST); + + tmp_ptr += NI_MAXHOST; + head.port = ntohl(GET(tmp_ptr)); + + tmp_ptr += 4; + head.key = ntohl(GET(tmp_ptr)); + + tmp_ptr += 4; + head.content_length = ntohl(GET(tmp_ptr)); + + tmp_ptr += 4; + head.code = ntohl(GET(tmp_ptr)); + return head; } -- Gitblit v1.8.0