libevent简单实现

怼烎@ 2022-06-17 00:54 251阅读 0赞

对epoll进行的封装

## epoll(7) ##

还记得前段时间去百度面试，被问道select、poll、epoll的区别，机械式的把书本上看到的内容复述了一遍。

面试官：谈谈你自己的理解。

一时语塞，说道，epoll写代码难度有点大。面试官点了点头目测很认可。

这就与epoll的结构体有关了，比如它的void\*指针可以只想用户结构体，用户结构提又可以封装一系列内容例如fd、回调等等。

`man 2 epola_wait` 看看它的结构体里面到底有什么。

typedef union epoll_data{
        void * ptr;
        int fd;
        uint32_t u32;
        uint64_t u64;
    }epoll_data_t;
    
    struct epoll_event{
        uint32_t events;
        epoll_data_t data;
    };

## Reactor ##

Reactor是一个设计模式，它包含了几个部分：

> Handle句柄：用来标志描述符  
> Demultiplexer：事件分发器，即select/poll/epoll  
> Event Handler : 事件处理接口  
> Reactor： 用来注册删除事件句柄，运行事件循环，就绪事件时，分发事件到回调函数上处理

![这里写图片描述][view]

## libevent简单实现 ##

#include    <stdlib.h>
    #include    <stdio.h>   
    #include    <stdio.h>
    #include    <sys/socket.h>
    #include    <sys/epoll.h>
    #include    <arpa/inet.h>
    #include    <fcntl.h>
    #include    <unistd.h>
    #include    <errno.h>
    #include    <string.h>
    #include    <time.h>
    #define MAX_EVENTS      1024
    #define BUFLEN  128
    #define SERV_PORT           8080
    
    
    /*
     *  status:1表示在监听事件中,0表示不在  
     *  last_active:记录最后一次响应时间,做超时处理
     */
    struct  myevent_s   {
        int fd; //cfd   listenfd
        int events; //EPOLLIN       EPLLOUT
        void    *arg;   //指向自己结构体指针
        void    (*call_back)(int    fd, int events, void    *arg);
        int status;
        char    buf[BUFLEN];
        int len;
        long    last_active;
    };
    
    int g_efd;  /*  epoll_create返回的句柄   */
    struct  myevent_s g_events[MAX_EVENTS+1];   /*+1最后一个用于listenfd*/
    
    void    eventset(struct myevent_s *ev,  int fd, void (*call_back)(int,int,void  *),void *arg)
    {
        ev->fd  =   fd;
        ev->call_back   =   call_back;
        ev->events  =   0;
        ev->arg =   arg;
        ev->status  =   0;
        //memset(ev->buf,   0,  sizeof(ev->buf));
        //ev->len   =   0;
        ev->last_active =   time(NULL);
        return;
    }
    
    void    eventset(struct myevent_s *ev,int fd, void (*call_back)(int,int,void*), void*arg)
    {
        ev->fd  =   fd;
        ev->call_back   =   call_back;
        ev->events  =   0;
        ev->arg =   arg;
        ev->status  =   0;
        //memset(ev->buf,   0,  sizeof(ev->buf));
        //ev->len   =   0;
        ev->last_active =   time(NULL);
        return;
    }
    void    recvdata(int    fd, int events, void    *arg);
    void    senddata(int    fd, int events, void    *arg);
    void    eventadd(int efd,   int events, struct  myevent_s   *ev)
    {
        struct epoll_event epv ={
       0, {
       0}};
        int op;
        epv.data.ptr = ev;
        epv.events= ev->events=events;
        if  (ev->status == 1)
        {
        op = EPOLL_CTL_MOD;
        }   
        else    
        {
        op = EPOLL_CTL_ADD;
        ev->status = 1;
        }
        if  (epoll_ctl(efd, op, ev->fd, &epv) < 0)
        printf("event add failed    [fd=%d],events[%d]\n",  ev->fd, events);
        else
        printf("event add   OK  [fd=%d],    op=%d,  events[%0X]\n", ev->fd, op,events);
        return;
    }
    void eventdel(int efd,  struct  myevent_s *ev)
    {
        struct epoll_event  epv = {
       0,   {
       0}};
        if  (ev->status !=1)
        return;
        epv.data.ptr = ev;
        ev->status  = 0;
        epoll_ctl(efd, EPOLL_CTL_DEL, ev->fd, &epv);
        return;
    }
    void acceptconn(int lfd,int events,void *arg)
    {
        struct sockaddr_in cin;
        socklen_t len = sizeof(cin);
        int cfd,    i;
        if  ((cfd =accept(lfd, (struct sockaddr *)&cin, &len)) == -1)
        {
        if (errno != EAGAIN && errno != EINTR)
        {
            /*  暂时不做出错处理    */
        }
        printf("%s:accept,  %s\n",  __func__,strerror(errno));
        return;
        }
        do  
        {
        for(i=0;i<MAX_EVENTS;i++)
        {
            if  (g_events[i].status==0)
            break;
        }
        if(i==MAX_EVENTS)   
        {
            printf("%s: max connect limit[%d]\n",__func__,MAX_EVENTS);
            break;
        }
        int flag=0;
        if((flag=fcntl(cfd,F_SETFL,O_NONBLOCK)) <0)
        {
            printf("%s: fcntl   nonblocking failed, %s\n",  __func__,strerror(errno));
            break;
        }
        eventset(&g_events[i],cfd,recvdata,&g_events[i]);
        eventadd(g_efd, EPOLLIN,&g_events[i]);
        }while(0);
        printf("new connect [%s:%d][time:%ld],pos[%d]\n",inet_ntoa (cin.sin_addr),ntohs(cin.sin_port),g_events[i].last_active,i);
        return;
    }
    void    recvdata(int fd,int events, void*arg)
    {
        struct myevent_s *ev=(struct myevent_s*)arg;
        int len;
        len =recv(fd,ev->buf,sizeof(ev->buf),0);
        eventdel(g_efd,ev);
        if  (len>0) 
        {
        ev->len =len;
        ev->buf[len]='\0';
        printf("C[%d]:%s\n",fd, ev->buf);
        /*  转换为发送事件 */
        eventset(ev,fd,senddata,ev);
        eventadd(g_efd,EPOLLOUT,ev);
        }
        else if(len==0)
        {
        close(ev->fd);
        /*ev-g_events地址相减得到偏移元素位置   */
        printf("[fd=%d] pos[%d],closed\n",fd,(int)(ev-g_events));
        }
        else
        {
        close(ev->fd);
        printf("recv[fd=%d] error[%d]:%s\n",fd,errno,strerror(errno));
        }
        return;
    }
    void senddata(int fd,int events,void *arg)
    {
        struct myevent_s *ev=(struct myevent_s*)arg;
        int len;
        len=send(fd,ev->buf,ev->len,0);
        //printf("fd=%d\tev->buf=%s\ttev->len=%d\n",    fd, ev->buf,    ev->len);
        //printf("send  len =   %d\n",  len);
        eventdel(g_efd,ev);
        if  (len>0)
        {
        printf("send[fd=%d],[%d]%s\n",fd,len,ev->buf);
        eventset(ev,fd,recvdata,ev);
        eventadd(g_efd, EPOLLIN,ev);
        }
        else    
        {
        close(ev->fd);
        printf("send[fd=%d] error   %s\n",fd,strerror(errno));
        }
        return;
    }
    void initlistensocket(int efd,short port)
    {
        int lfd=socket(AF_INET,SOCK_STREAM,0);
        fcntl(lfd,F_SETFL,O_NONBLOCK);
        eventset(&g_events[MAX_EVENTS],lfd,acceptconn,&g_events[MAX_EVENTS]);
        eventadd(efd,EPOLLIN,&g_events[MAX_EVENTS]);
        struct sockaddr_in sin;
        memset(&sin,0,sizeof(sin));
        sin.sin_family=AF_INET;
        sin.sin_addr.s_addr=INADDR_ANY;
        sin.sin_port=htons(port);
        bind(lfd,(struct sockaddr*)&sin,sizeof(sin));
        listen(lfd,20);
        return;
    }
    int main(int argc,char*argv[])
    {
        unsigned short port =SERV_PORT;
        if  (argc==2)
        port=atoi(argv[1]);
        g_efd=epoll_create(MAX_EVENTS+1);
        if  (g_efd<=0)
        printf("create efd in   %s  err %s\n",  __func__,strerror(errno));
        initlistensocket(g_efd,port);
        /*  事件循环    */
        struct epoll_event events[MAX_EVENTS+1];
        printf("server  running:port[%d]\n",    port);
        int checkpos=0,i;
        while(1){
        /*  超时验证,每次测试100个链接,不测试listenfd 当客户端60秒内没
            有和服务器通信,则关闭此客户端链接   */
        long now=time(NULL);
        for(i=0;i<100;i++,checkpos++) 
        {
            if(checkpos ==MAX_EVENTS)
            checkpos=0;
            if  (g_events[checkpos].status!=1)
            continue;
            long duration=now-g_events[checkpos].last_active;
            if  (duration>=60)
            {
            close(g_events[checkpos].fd);
            printf("[fd=%d] timeout\n",g_events[checkpos].fd);
            eventdel(g_efd, &g_events[checkpos]);
            }
        }
        /*  等待事件发生  */
        int nfd =epoll_wait(g_efd,events,MAX_EVENTS+1,1000);
        if(nfd<0)   
        {
            printf("epoll_wait  error,  exit\n");
            break;
        }
        for(i=0;i<nfd;i++)
        {
            struct myevent_s*ev =(struct myevent_s*)events[i].data.ptr;
            if  ((events[i].events&EPOLLIN)&&(ev->events&EPOLLIN))
            {
            ev->call_back(ev->fd,events[i].events,ev->arg);
            }
            if  ((events[i].events&EPOLLOUT)&&(ev->events&EPOLLOUT))
            {
            ev->call_back(ev->fd,events[i].events,ev->arg);
            }
        }
        }
        /*退出前释放所有资源*/
        return 0;
    }

## 参考 ##

1.  深入浅出 Libevent 作者：刘丹冰
2.  [https://segmentfault.com/a/1190000002715832][https_segmentfault.com_a_1190000002715832] （Reactor 和 Proactor）
3.  Linux man page

[view]: /images/20220617/bf0bc9f804094e4daf9648102168fdd6.png
[https_segmentfault.com_a_1190000002715832]: https://segmentfault.com/a/1190000002715832