Chinaunix

标题: [转载]linux2.6内核epoll用法举例说明(续)--给echo服务器增加读线程池 [打印本页]

---

作者: ratc    时间: 2007-04-28 22:09
标题: [转载]linux2.6内核epoll用法举例说明(续)--给echo服务器增加读线程池

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

#include
#include
#include
#include
#include
#include
#include
#include
#include

#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
     {
          perror("fcntl(sock,getfl)");
          exit(1);
     }
    opts = opts|o_nonblock;
     if(fcntl(sock,f_setfl,opts)
     {
          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
        {
               if(events.data.fd==listenfd)
               {
                    connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen);
                    if(connfd
                      perror("connfd
                      exit(1);
                   }
                    setnonblocking(connfd);

                    char *str = inet_ntoa(clientaddr.sin_addr);
                    std::cout>"
                    //设置用于读操作的文件描述符
                    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) 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))
           if (errno == econnreset) {
             close(fd);
          } else
             std::cout
           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);
       }
   }
}


本文来自ChinaUnix博客，如果查看原文请点：http://blog.chinaunix.net/u/29350/showart_288427.html

---

欢迎光临 Chinaunix (http://bbs.chinaunix.net/)

Powered by Discuz! X3.2