From d26e81c4213dfb04391c7b6692f243aede2e6895 Mon Sep 17 00:00:00 2001 From: wangzhengquan <wangzhengquan85@126.com> Date: 星期一, 12 十月 2020 15:25:31 +0800 Subject: [PATCH] version --- src/socket/net_mod_socket.c | 99 ++++++++++++++-- /dev/null | 43 ------- src/libshm_queue.a | 0 src/socket/net_mod_server_socket.h | 5 test_net_socket/net_mod_req_rep.c | 50 ++++++++ src/socket/net_mod_socket.h | 45 ++++++ test_net_socket/Makefile | 2 src/socket/net_mod_server_socket.c | 72 ++++++++--- 8 files changed, 230 insertions(+), 86 deletions(-) diff --git a/src/libshm_queue.a b/src/libshm_queue.a index 3ae19a9..fcda7db 100644 --- a/src/libshm_queue.a +++ b/src/libshm_queue.a Binary files differ diff --git a/src/socket/net_mod_server_socket.c b/src/socket/net_mod_server_socket.c index 97e4fea..4e574d8 100644 --- a/src/socket/net_mod_server_socket.c +++ b/src/socket/net_mod_server_socket.c @@ -2,10 +2,13 @@ #include "socket_io.h" #include "net_mod_socket_io.h" +#include "net_mod_socket.h" -NetModServerSocket::NetModServerSocket(int port, ShmModSocket * modsocket): shm_mod_socket(modsocket) +NetModServerSocket::NetModServerSocket(int port) { char portstr[32]; + + //shmModSocket = new ShmModSocket; sprintf(portstr, "%d", port); listenfd = Open_listenfd(portstr); init_pool(listenfd); @@ -81,44 +84,69 @@ /* $end add_client */ +int NetModServerSocket::process_client(rio_t *rio, int connfd) { + int n; + net_mod_request_head_t request_head; + net_mod_response_head_t response_head; + void *buf, *recv_buf; + int recv_size; + + size_t max_buf = 8096; + + buf = malloc(max_buf); + if(buf == NULL) { + err_exit(errno, "process_client malloc"); + } + if ((n = rio_readnb(rio, &request_head, sizeof(net_mod_request_head_t))) != sizeof(net_mod_request_head_t)) + { + free(buf); + return -1; + } + + if(request_head.content_length > max_buf) { + buf = realloc(buf, request_head.content_length); + max_buf = request_head.content_length; + if(buf == NULL) { + err_exit(errno, "process_client realloc"); + } + } + + if ((n = rio_readnb(rio, buf, request_head.content_length)) != request_head.content_length ) { + free(buf); + return -1; + } + + shmModSocket.sendandrecv(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size); + response_head.content_length = recv_size; + Rio_writen(connfd, &response_head, sizeof(response_head)); + Rio_writen(connfd, recv_buf, recv_size); + free(buf); + return 0; + +} /* $begin check_clients */ void NetModServerSocket::check_clients() { - int i, connfd, n; - char buf[MAXLINE]; - rio_t rio; + int i, connfd; + rio_t *rio; + for (i = 0; (i <= pool.maxi) && (pool.nready > 0); i++) { connfd = pool.clientfd[i]; - rio = pool.clientrio[i]; + rio = &pool.clientrio[i]; /* If the descriptor is ready, echo a text line from it */ if ((connfd > 0) && (FD_ISSET(connfd, &pool.ready_set))) { pool.nready--; - if ((n = rio_readpkgb(&rio, buf, MAXLINE)) > 0) - { - - Rio_writen(connfd, buf, n); - Rio_writen(connfd, PKG_SEP, strlen(PKG_SEP)); - // shm_mod_socket->sendto(buf, n, msg->key); - // net_mod_msg_t *msg = (net_mod_msg_t*)buf; - // if(msg.mod == PUB_SUB) { - // shm_mod_socket->pub(msg->topic, msg->topic_size, msg->content, msg->content_size, msg->key); - // } else { - // shm_mod_socket->sendto(msg->buf, msg->size, msg->key); - // } - } - - /* EOF detected, remove descriptor from pool */ - else - { + if(process_client(rio, connfd) != 0) { Close(connfd); //line:conc:echoservers:closeconnfd FD_CLR(connfd, &pool.read_set); //line:conc:echoservers:beginremove pool.clientfd[i] = -1; //line:conc:echoservers:endremove } + } } } diff --git a/src/socket/net_mod_server_socket.h b/src/socket/net_mod_server_socket.h index d908180..c79837b 100644 --- a/src/socket/net_mod_server_socket.h +++ b/src/socket/net_mod_server_socket.h @@ -28,16 +28,17 @@ private: int listenfd; - ShmModSocket * shm_mod_socket; + ShmModSocket shmModSocket; pool pool; void init_pool(int listenfd); void add_client(int connfd); void check_clients(); + int process_client(rio_t *rio, int connfd); public: - NetModServerSocket(int port, ShmModSocket *_shm_mod_socket); + NetModServerSocket(int port); void start(); ~NetModServerSocket(); diff --git a/src/socket/net_mod_socket.c b/src/socket/net_mod_socket.c index e1f71f0..bdfa85d 100644 --- a/src/socket/net_mod_socket.c +++ b/src/socket/net_mod_socket.c @@ -2,24 +2,95 @@ #include "socket_io.h" #include "net_mod_socket_io.h" -NetModSocket::NetModSocket(const char *host, int port) + +std::map<std::string, rio_t *> NetModSocket::connectionMap; + +NetModSocket::NetModSocket() { - char portstr[32]; - sprintf(portstr, "%d", port); - clientfd = Open_clientfd(host, portstr); - Rio_readinitb(&rio, clientfd); + } -ssize_t NetModSocket::send(void *buf, size_t size) { - int n = rio_writen(clientfd, buf, size); - rio_writen(clientfd, PKG_SEP, strlen(PKG_SEP)); +int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, + net_mod_recv_msg_t ** resp_arr, int *resp_arr_size) { - char resp[MAXLINE]; - int ss; - ss = rio_readpkgb(&rio, resp, MAXLINE); - puts(resp); - return n; + int i, n, clientfd; + char portstr[32]; + net_node_t *node; + char mapKey[256]; + void *recv_buf; + int recv_size; + 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; + 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++) { + + node = &node_arr[i]; + if(node->host == NULL) { + // 鏈湴鍙戦�� + 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}); + + } + + request_head.mod = REQ_REP; + request_head.key = node->key; + request_head.content_length = send_size; + if( (n = rio_writen(rio->rio_fd, &request_head, sizeof(request_head))) != sizeof(request_head)) { + err_exit(errno, "NetModSocket::send head rio_writen"); + + } + + if( (n = rio_writen(rio->rio_fd, send_buf, send_size)) != send_size ) { + err_exit(errno, "NetModSocket::send conent rio_writen"); + } + + if ((n = rio_readnb(rio, &response_head, sizeof(response_head))) != sizeof(response_head)) { + err_exit(errno, "NetModSocket::send rio_readnb"); + } + + recv_buf = malloc(response_head.content_length); + if(recv_buf == NULL) { + err_exit(errno, "NetModSocket::send malloc"); + } + if ( (recv_size = rio_readnb(rio, recv_buf, response_head.content_length) ) != response_head.content_length) { + err_exit(errno, "NetModSocket::send rio_readnb"); + } + +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; + } + *resp_arr = ret_arr; + *resp_arr_size = i; + + return i; + +} + + +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); + } + } // ssize_t recv(void *buf, size_t len) { @@ -29,6 +100,6 @@ // } NetModSocket::~NetModSocket() { - Close(clientfd); + } diff --git a/src/socket/net_mod_socket.h b/src/socket/net_mod_socket.h index 7ce98dc..4ace2bb 100644 --- a/src/socket/net_mod_socket.h +++ b/src/socket/net_mod_socket.h @@ -3,15 +3,52 @@ #include "usg_common.h" #include "shm_mod_socket.h" #include "socket_io.h" + +struct net_node_t +{ + const char *host; + int port; + int key; +}; + + +struct net_mod_request_head_t { + socket_mod_t mod; + int key; + uint32_t content_length; +}; + +struct net_mod_response_head_t { + // socket_mod_t mod; + // int key; + uint32_t content_length; +}; + + +struct net_mod_recv_msg_t +{ + char host[128]; + int port; + int key; + void *content; + uint32_t content_length; + +}; + class NetModSocket { private: - int clientfd; - rio_t rio; + static std::map<std::string, rio_t *> connectionMap; + ShmModSocket shmModSocket; public: - NetModSocket(const char *host, int port); - ssize_t send(void *buf, size_t size); + + NetModSocket(); + int sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, + net_mod_recv_msg_t ** resp_arr, int *resp_arr_size); + ~NetModSocket(); + + static void free_recv_msg_arr(net_mod_recv_msg_t * arr, size_t size); }; diff --git a/test_net_socket/Makefile b/test_net_socket/Makefile index 6060f15..f73ebc3 100644 --- a/test_net_socket/Makefile +++ b/test_net_socket/Makefile @@ -13,7 +13,7 @@ INCLUDES += -I${DEST}/include/shmqueue -I$(ROOT)/include/usgcommon -PROGS = ${DEST}/test_net_mod_socket +PROGS = ${DEST}/net_mod_req_rep DEPENDENCES = $(patsubst %, %.d, $(PROGS)) diff --git a/test_net_socket/net_mod_req_rep.c b/test_net_socket/net_mod_req_rep.c new file mode 100644 index 0000000..726eb06 --- /dev/null +++ b/test_net_socket/net_mod_req_rep.c @@ -0,0 +1,50 @@ +#include "net_mod_server_socket.h" +#include "net_mod_socket.h" +#include "shm_mm.h" +#include "dgram_mod_socket.h" +#include "usg_common.h" + +void server() { + NetModServerSocket *serverSocket = new NetModServerSocket(5000); + serverSocket->start(); +} + +void client(){ + NetModSocket client; + char send_buf[MAXLINE]; + net_mod_recv_msg_t *recv_arr; + int recv_arr_size, i; + net_node_t node_arr[1] = { + {"localhost", 5000, 8} + }; + while (fgets(send_buf, MAXLINE, stdin) != NULL) { + client.sendandrecv( node_arr, 1, send_buf, strlen(send_buf), &recv_arr, &recv_arr_size); + for(i=0; i<recv_arr_size; i++) { + printf("host:%s, port: %d, key:%d, content: %s\n", + recv_arr[i].host, + recv_arr[i].port, + recv_arr[i].key, + recv_arr[i].content + ); + } + } +} + +int main(int argc, char *argv[]) { + shm_init(512); + + if (argc < 2) { + fprintf(stderr, "Usage: %s %s|%s\n", argv[0], "server", "client"); + return 1; + } + + if (strcmp("server", argv[1]) == 0 ) { + server(); + } + + if (strcmp("client", argv[1]) == 0) + client(); +} + + + diff --git a/test_net_socket/test_net_mod_socket.c b/test_net_socket/test_net_mod_socket.c deleted file mode 100644 index 2d9c94e..0000000 --- a/test_net_socket/test_net_mod_socket.c +++ /dev/null @@ -1,43 +0,0 @@ -#include "net_mod_server_socket.h" -#include "net_mod_socket.h" -#include "shm_mm.h" -#include "dgram_mod_socket.h" - -void server() { - // void *socket = dgram_mod_open_socket(); - // ShmModSocket tt; - ShmModSocket * m_socket = new ShmModSocket; - // NetModServerSocket(5000, NULL); - - NetModServerSocket *serverSocket = new NetModServerSocket(5000, m_socket); - serverSocket->start(); -} - -void client(){ - NetModSocket client("localhost", 5000); - char buf[MAXLINE]; - - while (fgets(buf, MAXLINE, stdin) != NULL) { - client.send(buf, strlen(buf)); - - } -} - -int main(int argc, char *argv[]) { - shm_init(512); - - if (argc < 2) { - fprintf(stderr, "Usage: %s %s|%s\n", argv[0], "server", "client"); - return 1; - } - - if (strcmp("server", argv[1]) == 0 ) { - server(); - } - - if (strcmp("client", argv[1]) == 0) - client(); -} - - - -- Gitblit v1.8.0