diff --git a/tx_server_v4.cpp b/tx_server_v4.cpp new file mode 100644 index 0000000000000000000000000000000000000000..9f6a7f1dd17cebadbc71d8ef972c04da8cf78850 --- /dev/null +++ b/tx_server_v4.cpp @@ -0,0 +1,132 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "threadpool_v4.h" +// 线程池版本的 服务器 + +// 分配给socket的msgid +static uint64_t msgid = 0; +static threadpool *pool; + +void et(epoll_event *events, int number, int epollfd, int listenfd) +{ + for (int i = 0; i < number; ++i) + { + int sockfd = ((client_data *)events[i].data.ptr)->sockfd; + if (sockfd == listenfd) + { + struct sockaddr_in client_address; + socklen_t client_addrlength = sizeof(client_address); + int connfd = 0; + while ((connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength)) > 0) + { + //有新的链接来了 + client_data *cd = (client_data *)malloc(sizeof(client_data)); + cd->address = client_address; + cd->sockfd = connfd; + cd->epollfd = epollfd; + cd->msgid = msgid++; //在此任务中,由于每个socket都是短连接,只传输一次消息,可分配一个id用于后续接受到消息的msgid + + addfd(epollfd, connfd, true, cd); + } + if (connfd == -1) + { + if (errno != EAGAIN && errno != ECONNABORTED && errno != EPROTO && errno != EINTR) + perror("accept"); + } + } + else if (events[i].events & EPOLLIN) + { + // pthread_t thread; + // pthread_create(&thread, NULL, worker, events[i].data.ptr); + // pthread_detach(thread); + if (pool->append((client_data *)events[i].data.ptr) == false) + { + client_data *reset_client = (client_data *)events[i].data.ptr; + reset_oneshot(reset_client->epollfd, reset_client->sockfd, reset_client); //////////////这里如果任务池满了就重新注册该客户端,使得下一次epoll_wait重新响应 + } + } + else + { + printf("something else happend\n"); + } + } +} + +int main(int argc, char *argv[]) +{ + if (argc <= 2) + { + printf("usage: %s ip_address port_number\n", basename(argv[0])); + } + const char *ip = argv[1]; + int port = atoi(argv[2]); + + int ret = 0; + struct sockaddr_in address; + bzero(&address, sizeof(address)); + address.sin_family = AF_INET; + inet_pton(AF_INET, ip, &address.sin_addr); + address.sin_port = htons(port); + + int listenfd = socket(PF_INET, SOCK_STREAM, 0); + assert(listenfd >= 0); + + int opt = 1; + setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (const void *)&opt, sizeof(opt)); ///////设置端口复用 + + ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address)); + assert(ret != -1); + + ret = listen(listenfd, SOMAXCONN); + assert(ret != -1); + + epoll_event events[MAX_EVENT_NUMBER]; + int epollfd = epoll_create(SOMAXCONN); + assert(epollfd != -1); + + //储存本机socket的数据 + client_data *cd = (client_data *)malloc(sizeof(client_data)); + cd->epollfd = epollfd; + cd->sockfd = listenfd; + cd->address = address; + cd->msgid = msgid++; + + addfd(epollfd, listenfd, false, cd); /////////////////////////////////////已经修改et触发情况的while读取(第36行) + + //初始化线程池 + pool = new threadpool; + + while (1) + { + int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1); + // printf("return epoll wait\n"); + if (ret < 0) + { + printf("epoll failure!\n"); + break; + } + else + { + et(events, ret, epollfd, listenfd); + } + } + close(listenfd); + //释放本机socket相关数据 + free(cd); + delete pool; + return 0; +} \ No newline at end of file