From 5ec3e58ef7720a165a1d8a080cb596780565a9d0 Mon Sep 17 00:00:00 2001 From: LHB Date: Wed, 17 Nov 2021 16:55:21 +0000 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=94=B9=E5=8F=91=E9=80=81=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=B8=BA=E6=94=B6=E5=88=B0=E7=9A=84=E4=BF=A1=E6=81=AF?= =?UTF-8?q?=E5=92=8C=E7=9B=B8=E5=BA=94=E7=9A=84=E5=AD=97=E8=8A=82=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- threadpool_v4.h | 218 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 threadpool_v4.h diff --git a/threadpool_v4.h b/threadpool_v4.h new file mode 100644 index 0000000..85bfd89 --- /dev/null +++ b/threadpool_v4.h @@ -0,0 +1,218 @@ +#ifndef THREADPOOL_H +#define THREADPOOL_H + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +//需要用到c++ 的链表库 +#include +#include + +struct client_data +{ + int epollfd; + int sockfd; + sockaddr_in address; + uint64_t msgid; //socket id 充当msgid + /* data */ +}; + +const int MAX_EVENT_NUMBER = 1024; +const int BUFFER_SIZE = 100; + +int setnonblocking(int fd) +{ + int old_option = fcntl(fd, F_GETFL); + int new_option = old_option | O_NONBLOCK; + fcntl(fd, F_SETFL, new_option); + return old_option; +} + +void addfd(int epollfd, int fd, bool oneshot, client_data *cd) +{ + epoll_event event; + event.data.ptr = cd; + event.events = EPOLLIN | EPOLLET; + if (oneshot) + { + event.events |= EPOLLONESHOT; + } + epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event); + setnonblocking(fd); +} + +void reset_oneshot(int epollfd, int fd, client_data *cd) +{ + epoll_event event; + event.data.ptr = cd; + event.events = EPOLLIN | EPOLLET | EPOLLONESHOT; + epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &event); +} + +void send_http_response(int fd, char *buf) +{ + //char data[] = "HTTP/1.1 200 OK\r\nServer: aiserver<0.1>\r\nAccept-Ranges: bytes\r\nContent-Length: 12\r\nConnection: close\r\nContent-Type: text/html\r\n\r\nhello world!"; + // char *data = buf; + char data[] = "hello client"; + size_t ret = send(fd, data, sizeof(data), 0); + // printf("%lu %lu \n", sizeof(data), ret); +} + +//线程池相关代码如下 +class threadpool +{ +private: + int m_thread_number; + int m_max_requests; + pthread_t *m_threads; + std::list m_workqueue; + bool m_stop; + + sem_t m_sem; //用于通知线程 + pthread_mutex_t m_mutex; //用于给list加锁 + +private: + void process_client_data(void *arg) + { + int sockfd = ((client_data *)arg)->sockfd; + int epollfd = ((client_data *)arg)->epollfd; + sockaddr_in addr = ((client_data *)arg)->address; + uint64_t msgid_sock = ((client_data *)arg)->msgid; + // printf("start new thread to receive data on fd: %d\n", sockfd); + char buf[BUFFER_SIZE]; + memset(buf, '\0', BUFFER_SIZE); + for (;;) + { + int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0); + if (ret == 0) + { + close(sockfd); + //释放掉客户端的clientdata + free(arg); //释放掉了meollc的内存空间 + break; + } + else if (ret < 0) + { + if (errno == EAGAIN) + { + reset_oneshot(epollfd, sockfd, (client_data *)arg); ////////////////发完可以主动断开链接？ + // printf("read later\n"); + break; //已经读取完全部数据 + } + } + else + { + struct timeval tv; + static struct tm t; + + gettimeofday(&tv, NULL); + localtime_r(&tv.tv_sec, &t); + // send_http_response(sockfd, buf); + send(sockfd, buf, ret + 1, 0);/////////////收到什么发什么，并在结尾补\0 + printf("[%04d:%02d:%02d %02d:%02d:%02d.%03ld] \"recv %s msg, msgid:%011lu, msg content: %s\"\n", (1900 + t.tm_year), + (1 + t.tm_mon), t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec, tv.tv_usec / 1000, inet_ntoa(addr.sin_addr), msgid_sock, buf); + // usleep(100000); + } + // reset_oneshot(epollfd, sockfd, (client_data *)arg); + // printf("reset oneshot"); + } + } + + void run() + { + while (!m_stop) + { + sem_wait(&m_sem); + pthread_mutex_lock(&m_mutex); ////////////////////////////////把用于记录当前活跃线程的num也锁住，就可以动态监控线程数了。 + if (m_workqueue.empty()) + { + pthread_mutex_unlock(&m_mutex); + continue; + } + client_data *cd = m_workqueue.front(); + m_workqueue.pop_front(); + pthread_mutex_unlock(&m_mutex); + if (!cd) + { + continue; + } + //执行客户端的任务 + process_client_data(cd); + } + } + + static void *worker(void *arg) + { + threadpool *p = (threadpool *)arg; + p->run(); + return p; + } + +public: + //初始化 + threadpool(int thread_number = 10, int max_requests = 100) : m_thread_number(thread_number), m_max_requests(max_requests), m_threads(NULL), m_stop(false) + { + //先要初始化 信号量 和锁 + if (sem_init(&m_sem, 0, 0) != 0 || pthread_mutex_init(&m_mutex, NULL) != 0) + { + throw std::exception(); + } + m_threads = new pthread_t[m_thread_number]; + if (!m_threads) + { + throw std::exception(); + } + for (int i = 0; i < m_thread_number; ++i) + { + if (pthread_create(&m_threads[i], NULL, worker, this) != 0) + { + delete[] m_threads; + throw std::exception(); + } + if (pthread_detach(m_threads[i])) + { + delete[] m_threads; + throw std::exception(); + } + } + } + //添加任务 + bool append(client_data *cd) + { + pthread_mutex_lock(&m_mutex); + if (m_workqueue.size() > m_max_requests) + { + pthread_mutex_unlock(&m_mutex); + printf("\n\n\n\n\n\n\n\nmemory size full\n\n\n\n\n\n"); + return false; + } + m_workqueue.push_back(cd); + pthread_mutex_unlock(&m_mutex); + sem_post(&m_sem); //通知其他线程 任务增加了 + return true; + } + + //销毁 + ~threadpool() + { + delete[] m_threads; + m_stop = true; + sem_destroy(&m_sem); + pthread_mutex_destroy(&m_mutex); + } +}; + +#endif \ No newline at end of file -- Gitee