linux 2.6内核epoll用法举例说明(续)--给echo服务器增加读线程池

peakzhang

linux 2.6内核epoll用法举例说明(续)--给echo服务器增加读线程池
上篇文章使用linux内核2.6提供的epoll机制实现了一个反应式echo服务器，使用反应式服务器的最大好处就是可以按cpu的数量来配置线程池内线程的线程数而不是根据客户端的并发量配置线程池。我是第一次使用pthread库来写线程池，使用的是工作队列方式的线程池。我感觉作队列方式的线程池可以当成一种设计模式来用的，在很多平台上都是可以按这种方式来实现线程池，从win32 ,unix到jvm都是适用的

#include <iostream>

#include <sys/socket.h>

#include <sys/epoll.h>

#include <netinet/in.h>

#include <arpa/inet.h>

#include <fcntl.h>

#include <unistd.h>

#include <stdio.h>

#include <pthread.h>

#define MAXLINE 10

#define OPEN_MAX 100

#define LISTENQ 20

#define SERV_PORT 5555

#define INFTIM 1000

//线程池任务队列结构体

struct task{

  int fd;            //需要读写的文件描述符

  struct task *next; //下一个任务

};

//用于读写两个的两个方面传递参数

struct user_data{

  int fd;

  unsigned int n_size;

  char line[MAXLINE];

};

//线程的任务函数

void * readtask(void *args);

void * writetask(void *args);

//声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件

struct epoll_event ev,events[20];

int epfd;

pthread_mutex_t mutex;

pthread_cond_t cond1;

struct task *readhead=NULL,*readtail=NULL,*writehead=NULL;

void setnonblocking(int sock)

{

     int opts;

     opts=fcntl(sock,F_GETFL);

     if(opts<0)

     {

          perror("fcntl(sock,GETFL)");

          exit(1);

     }

    opts = opts|O_NONBLOCK;

     if(fcntl(sock,F_SETFL,opts)<0)

     {

          perror("fcntl(sock,SETFL,opts)");

          exit(1);

     }   

}

int main()

{

     int i, maxi, listenfd, connfd, sockfd,nfds;

     pthread_t tid1,tid2;

     struct task *new_task=NULL;

     struct user_data *rdata=NULL;

     socklen_t clilen;

     pthread_mutex_init(&mutex,NULL);

     pthread_cond_init(&cond1,NULL);

     //初始化用于读线程池的线程

     pthread_create(&tid1,NULL,readtask,NULL);

     pthread_create(&tid2,NULL,readtask,NULL);

     //生成用于处理accept的epoll专用的文件描述符   

     epfd=epoll_create(256);

     struct sockaddr_in clientaddr;

     struct sockaddr_in serveraddr;

     listenfd = socket(AF_INET, SOCK_STREAM, 0);

     //把socket设置为非阻塞方式

     setnonblocking(listenfd);

     //设置与要处理的事件相关的文件描述符

     ev.data.fd=listenfd;

     //设置要处理的事件类型

     ev.events=EPOLLIN|EPOLLET;

     //注册epoll事件

     epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);

     bzero(&serveraddr, sizeof(serveraddr));

     serveraddr.sin_family = AF_INET;

     char *local_addr="200.200.200.222";

     inet_aton(local_addr,&(serveraddr.sin_addr));//htons(SERV_PORT);

     serveraddr.sin_port=htons(SERV_PORT);

     bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr));

     listen(listenfd, LISTENQ);

     maxi = 0;

     for ( ; ; ) {

          //等待epoll事件的发生

          nfds=epoll_wait(epfd,events,20,500);

          //处理所发生的所有事件      

        for(i=0;i<nfds;++i)

        {

               if(events.data.fd==listenfd)

               {

                    connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen);

                    if(connfd<0){

                      perror("connfd<0");

                      exit(1);

                   }

                    setnonblocking(connfd);

                    char *str = inet_ntoa(clientaddr.sin_addr);

                    std::cout<<"connec_ from >>"<<str<<std::endl;

                    //设置用于读操作的文件描述符

                    ev.data.fd=connfd;

                    //设置用于注测的读操作事件

                 ev.events=EPOLLIN|EPOLLET;

                    //注册ev

                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);

               }

            else if(events.events&EPOLLIN)

            {

                    printf("reading!\n");                 

                    if ( (sockfd = events.data.fd) < 0) continue;

                    new_task=new task();

                    new_task->fd=sockfd;

                    new_task->next=NULL;

                    //添加新的读任务

                    pthread_mutex_lock(&mutex);

                    if(readhead==NULL)

                    {

                      readhead=new_task;

                      readtail=new_task;

                    }   

                    else

                    {   

                     readtail->next=new_task;

                      readtail=new_task;

                    }   

                   //唤醒所有等待cond1条件的线程

                    pthread_cond_broadcast(&cond1);

                    pthread_mutex_unlock(&mutex);  

              }

               else if(events.events&EPOLLOUT)

               {   

              rdata=(struct user_data *)events.data.ptr;

                 sockfd = rdata->fd;

                 write(sockfd, rdata->line, rdata->n_size);

                 delete rdata;

                 //设置用于读操作的文件描述符

                 ev.data.fd=sockfd;

                 //设置用于注测的读操作事件

               ev.events=EPOLLIN|EPOLLET;

                 //修改sockfd上要处理的事件为EPOLIN

               epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);

               }

          }

     }

}

void * readtask(void *args)

{

   int fd=-1;

   unsigned int n;

   //用于把读出来的数据传递出去

   struct user_data *data = NULL;

   while(1){

        pthread_mutex_lock(&mutex);

        //等待到任务队列不为空

        while(readhead==NULL)

             pthread_cond_wait(&cond1,&mutex);

        fd=readhead->fd;

        //从任务队列取出一个读任务

        struct task *tmp=readhead;

        readhead = readhead->next;

        delete tmp;

        pthread_mutex_unlock(&mutex);

        data = new user_data();

        data->fd=fd;

        if ( (n = read(fd, data->line, MAXLINE)) < 0) {

           if (errno == ECONNRESET) {

             close(fd);

          } else

             std::cout<<"readline error"<<std::endl;

           if(data!=NULL)delete data;

        } else if (n == 0) {

            close(fd);

           printf("Client close connect!\n");

           if(data!=NULL)delete data;

        } else{

        data->n_size=n;

        //设置需要传递出去的数据

        ev.data.ptr=data;

        //设置用于注测的写操作事件

        ev.events=EPOLLOUT|EPOLLET;

        //修改sockfd上要处理的事件为EPOLLOUT

        epoll_ctl(epfd,EPOLL_CTL_MOD,fd,&ev);

       }

   }

}

作者Blog：[url=http://blog.csdn.net/mote_li/]http://blog.csdn.net/mote_li/[/url]