Libevent之evconnlistener详解

爱被打了一巴掌 2022-12-31 06:22 140阅读 0赞

# evconnlistener的简介 #

基于event和event\_base已经可以写一个CS模型了。但是对于服务器端来说，仍然需要用户自行调用socket、bind、listen、accept等步骤。这个过程有点繁琐，并且一些细节可能考虑不全，为此Libevent推出了一些对应的封装函数，简化了整个监听的流程，用户仅仅需要在对应回调函数里面处理已完成连接的套接字即可。  
1、省去了用户手动注册事件的过程。  
2、省去了用户去验证系统函数返回是否成功的问题。  
3、帮助用户完成了处理非阻塞套接字accpet的麻烦。  
4、整个过程一气呵成，用户仅仅关系业务逻辑即可，其他细节，libevent都帮你搞定。

# evconnlistener\_new\_bind过程详解 #

用户仅仅需要通过**`evconnlistener_new_bind`传递回调函数**，在aceept成功后，在回调函数里面处理已连接的套接字即可。省去了用户需要处理的一些列麻烦问题。  
evconnlistener其实是对even\_base和event的封装而已。

//一系列的工作函数，因为listener可以用于不同的协议。
    struct evconnlistener_ops {
        int (*enable)(struct evconnlistener *);
        int (*disable)(struct evconnlistener *);
        void (*destroy)(struct evconnlistener *);
        void (*shutdown)(struct evconnlistener *);
        evutil_socket_t (*getfd)(struct evconnlistener *);
        struct event_base *(*getbase)(struct evconnlistener *);
    };
    
    //一层一层封装，加上隔离
    struct evconnlistener {
        const struct evconnlistener_ops *ops;   //操作函数
        void *lock;                             //锁变量，用于线程安全
        evconnlistener_cb cb;                   //用户的回调函数
        evconnlistener_errorcb errorcb;         //发生错误时的回调函数
        void *user_data;                        //回调函数的参数，当回调函数执行时候，通过形参传入回调函数内部
        unsigned flags;                         //属性标志 ，例如socket套接字属性，可以是阻塞，非阻塞，reuse等。
        short refcnt;                           //引用计数
        unsigned enabled : 1;                   //位域为1.即只需一个比特位来存储这个成员 
    };
    struct evconnlistener_event {
        struct evconnlistener base;
        struct event listener;     //内部event,插入到event_base，完成监听
    };

`evconnlistener_ops`包含一些工作函数，因为支持的协议不仅仅是TCP，所以搞一个结构体包装下，后续代码更容易编写。

`evconnlistener`里面包含了一些属性，其中包括当`listener`事件发生的时候，会调用的用户回调函数以及传递给回调函数的参数。这样在回调函数里面就可以应用这些参数了。

`evconnlistener_event`里面封装了一个`listener`事件，首先将`listener`加入epoll监听，当`listener`可读，则会调用Libevent内部的回调函数`listener_read_cb`，在这个回调函数内部，调用用户注册的回调函数`cb`，各个参数的传递通过指针进行，这也是回调函数编写的本质所在。下面具体分析源代码。

/*
       @param base The event base to associate the listener with.
       @param cb A callback to be invoked when a new connection arrives. If the
          callback is NULL, the listener will be treated as disabled until the
          callback is set.
       @param ptr A user-supplied pointer to give to the callback.
       @param flags Any number of LEV_OPT_* flags
       @param backlog Passed to the listen() call to determine the length of the
          acceptable connection backlog.  Set to -1 for a reasonable default.
       @param addr The address to listen for connections on.
       @param socklen The length of the address.
    
    上述ptr通常设定为base，因为在cb里面通常要利用bufferevent处理连接socket的读和写。
    */
    struct evconnlistener *
    evconnlistener_new_bind(struct event_base *base, evconnlistener_cb cb,
        void *ptr, unsigned flags, int backlog, const struct sockaddr *sa,
        int socklen)
    {
        struct evconnlistener *listener;
        evutil_socket_t fd;
        int on = 1;
        int family = sa ? sa->sa_family : AF_UNSPEC;
    
        if (backlog == 0)
            return NULL;
    
    
        //创建监听socket
        fd = socket(family, SOCK_STREAM, 0);//创建套接字
        if (fd == -1)
            return NULL;
    
        //将socket套接字设定非阻塞
        if (evutil_make_socket_nonblocking(fd) < 0) {
            evutil_closesocket(fd);
            return NULL;
        }
    
        //设定一些socket套接字选项
        if (flags & LEV_OPT_CLOSE_ON_EXEC) {//设定socket文件描述符属性-FD_CLOEXEC
            if (evutil_make_socket_closeonexec(fd) < 0) {
                evutil_closesocket(fd);
                return NULL;
            }
        }
        if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (void*)&on, sizeof(on))<0) {//设定是否打开SO_KEEPALIVE，默认打开
            evutil_closesocket(fd);
            return NULL;
        }
        if (flags & LEV_OPT_REUSEABLE) {
            if (evutil_make_listen_socket_reuseable(fd) < 0) {//设定socket选项 REUSEABLE
                evutil_closesocket(fd);
                return NULL;
            }
        }
    
        if (sa) {
            if (bind(fd, sa, socklen)<0) {//绑定
                evutil_closesocket(fd);
                return NULL;
            }
        }
    
        //监听并将返回连接描述符加入epoll
        listener = evconnlistener_new(base, cb, ptr, flags, backlog, fd);//socket+blind完后，就创建事件并监听
        if (!listener) {
            evutil_closesocket(fd);
            return NULL;
        }
    
        return listener;
    }
    
    
    
    //封装服务器端套路 listen 并将ListenFD加入epoll监听
    struct evconnlistener *
    evconnlistener_new(struct event_base *base,
        evconnlistener_cb cb, void *ptr, unsigned flags, int backlog,
        evutil_socket_t fd)
    {
        struct evconnlistener_event *lev;
    
    #ifdef WIN32
        if (base && event_base_get_iocp(base)) {
            const struct win32_extension_fns *ext =
                event_get_win32_extension_fns();
            if (ext->AcceptEx && ext->GetAcceptExSockaddrs)
                return evconnlistener_new_async(base, cb, ptr, flags,
                    backlog, fd);
        }
    #endif
    
        //开始listen，并设定backlog-排队最大连接数
        if (backlog > 0) {
            if (listen(fd, backlog) < 0)
                return NULL;
        } else if (backlog < 0) {
            if (listen(fd, 128) < 0)
                return NULL;
        }
    
        lev = mm_calloc(1, sizeof(struct evconnlistener_event));
        if (!lev)
            return NULL;
    
        //将用户传入参数全部保存至evconnlistener_event，供后续回调函数使用。
        lev->base.ops = &evconnlistener_event_ops;
        lev->base.cb = cb;
        lev->base.user_data = ptr;//
        lev->base.flags = flags;//
        lev->base.refcnt = 1;
    
        if (flags & LEV_OPT_THREADSAFE) {//线程安全
            EVTHREAD_ALLOC_LOCK(lev->base.lock, EVTHREAD_LOCKTYPE_RECURSIVE);
        }
    
        event_assign(&lev->listener, base, fd, EV_READ|EV_PERSIST,
            listener_read_cb, lev);//设定事件属性，并加入epoll，EV_PERSIST   
        //默认回调函数listener_read_cb，注意执行传入了lev参数    
    
        evconnlistener_enable(&lev->base);//调用epoll_add将listener加入epoll中监听。
    
        return &lev->base;
    }

从上述函数过程可以看出，用户仅仅需要定义`struct sockaddr_in`套接字地址结构，并完成对应的初始化即可。`evconnlistener_new_bind`帮助你创建监听套接字、绑定地址和端口、监听操作。然后当监听套接字可读，则调用Libevent内部回调函数`listener_read_cb`，进而调用用户注册的回调函数`cb`，因此这里用户仅仅需要关心，事件发送如何处理业务逻辑即可，不必关系操作系统底层需要如何处理。这里有两点需要注意：  
1、创建的监听套接字默认是非阻塞的。所以对于accept的处理很特殊。  
2、一些套接字属性通常需要设定，例如`LEV_OPT_REUSEABLE、LEV_OPT_CLOSE_ON_EXEC`等。  
3、传给`listener_read_cb`的参数分别是：事件fd即监听套接字、触发的事件属性、`struct evconnlistener_event`结构体指针。

struct sockaddr_in sin;
        sin.sin_family = AF_INET;
        sin.sin_addr.s_addr = htonl(INADDR_ANY);
        sin.sin_port = htons(9999);

下面看看`listener_read_cb`

static void
    listener_read_cb(evutil_socket_t fd, short what, void *p)
    {
        struct evconnlistener *lev = p;
        //注意本来传入参数是evconnlistener_event，但是这里会强制转换成evconnlistener。
        //也就是这里的lev仅仅能访问evconnlistener_event的evconnlistener部分。
        int err;
        evconnlistener_cb cb;
        evconnlistener_errorcb errorcb;
        void *user_data;
        LOCK(lev);
        while (1) {//可能有多个客户端同时请求连接,将完成连接缓冲队列中全部accept
            struct sockaddr_storage ss;
    #ifdef WIN32
            int socklen = sizeof(ss);
    #else
            socklen_t socklen = sizeof(ss);
    #endif
            //对于非阻塞socket，Listen之后，用epoll监听Listenfd是否可读即可搞定。
            evutil_socket_t new_fd = accept(fd, (struct sockaddr*)&ss, &socklen);//因为Listenfd非阻塞,此时accept也不会阻塞
            if (new_fd < 0)//尚无连接，则出错，立即返回小于0数，并且将errno=EWOULDBLOCK/EAGAIN。
                break;
            if (socklen == 0) {
                /* This can happen with some older linux kernels in
                 * response to nmap. */
                evutil_closesocket(new_fd);
                continue;
            }
    
            //默认将new_fd设为非阻塞。
            if (!(lev->flags & LEV_OPT_LEAVE_SOCKETS_BLOCKING))
                evutil_make_socket_nonblocking(new_fd);
    
            //用户没有设置回调函数，则直接关闭new_fd并返回。
            if (lev->cb == NULL) {
                evutil_closesocket(new_fd);
                UNLOCK(lev);
                return;
            }
    
            //由于refcnt被初始化为1.这里有++了，所以一般情况下并不会进入下面的  
            //if判断里面。但如果程在下面UNLOCK之后，第二个线调用evconnlistener_free  
            //释放这个evconnlistener时，就有可能使得refcnt为1了。即进入那个判断体里  
            //执行listener_decref_and_unlock。在下面会讨论这个问题。          
            ++lev->refcnt;
            cb = lev->cb;
            user_data = lev->user_data;
            UNLOCK(lev);
            cb(lev, new_fd, (struct sockaddr*)&ss, (int)socklen,
                user_data);//执行用户回调函数，让用户处理已经完成连接的new_fd
            LOCK(lev);
            if (lev->refcnt == 1) {
                int freed = listener_decref_and_unlock(lev);
                EVUTIL_ASSERT(freed);
                return;
            }
            --lev->refcnt;
        }
        err = evutil_socket_geterror(fd);//判断listenfd的错误，忽略一些错误
    //当错误是下面三种，直接返回，表示可以重新accept
    //EINTR:
    //EAGAIN:
    //ECONNABORTED:
        if (EVUTIL_ERR_ACCEPT_RETRIABLE(err)) {//还可以accept
            UNLOCK(lev);
            return;
        }
    
        //当真有错误发生时才会运行到这里  
        if (lev->errorcb != NULL) {
            ++lev->refcnt;
            errorcb = lev->errorcb;
            user_data = lev->user_data;
            UNLOCK(lev);
            errorcb(lev, user_data);//执行错误函数
            LOCK(lev);
            listener_decref_and_unlock(lev);
        } else {
            event_sock_warn(fd, "Error from accept() call");
        }
    }

注意两点：  
1、在`while`循环里面处理`accept`，因为可能此时在已连接缓冲队列中有多条，所以需要不停accept。直到accept返回负数，并且errno是EAGAIN为止，否则调用先前注册的回调函数，表明accept错误。  
2、在这个回调函数里面执行用户注册的回调函数传入参数分别是：  
`cb(lev, new_fd, (struct sockaddr*)&ss, (int)socklen,user_data);`  
上述的`new_fd`是已经成功建立连接的`fd`，ss是客户端套接字地址结构，`user_data`是用户`evconnlistener_new_bind`时候传入的`ptr`数据。注意地址如何传递的。

# evconnlistener\_ops操作 #

`evconnlistener_ops`中存储了操作监听事件的一些函数封装。  
注意下面这些函数是静态的，仅仅在当前文件中有效，供Libevent内部调用。

//已知结构体成员地址，求结构体首地址，原理是结构体在内存上是顺序排放的。
    #define EVUTIL_UPCAST(ptr, type, field)             \
        ((type *)(((char*)(ptr)) - evutil_offsetof(type, field)))
    
    //将evconnlistener_event里面的事件listener添加epoll
    
    static int
    event_listener_enable(struct evconnlistener *lev)
    {
    
    //通过evconnlistener地址获取 evconnlistener_event 的地址
        struct evconnlistener_event *lev_e =
            EVUTIL_UPCAST(lev, struct evconnlistener_event, base);
        return event_add(&lev_e->listener, NULL);//将lister事件添加epoll。
    }
    
    //从epoll中删除事件
    static int
    event_listener_disable(struct evconnlistener *lev)
    {
        struct evconnlistener_event *lev_e =
            EVUTIL_UPCAST(lev, struct evconnlistener_event, base);
        return event_del(&lev_e->listener);//从IO事件列表中移除，并调用epoll_ctl。
    }
    
    
    //将事件从epoll中移除，并且释放删除监听套接字。
    //注意LEV_OPT_CLOSE_ON_FREE选项关闭的是服务器端的监听socket，而非那些连接客户端的socket
    static void
    event_listener_destroy(struct evconnlistener *lev)
    {
        struct evconnlistener_event *lev_e =
            EVUTIL_UPCAST(lev, struct evconnlistener_event, base);
    
        event_del(&lev_e->listener);//把event从event_base中删除
        if (lev->flags & LEV_OPT_CLOSE_ON_FREE)
        //注意LEV_OPT_CLOSE_ON_FREE选项关闭的是服务器端的监听socket，而非那些连接客户端的socket
    
            evutil_closesocket(event_get_fd(&lev_e->listener));//如果用户设置了这个选项，那么要关闭socket
        event_debug_unassign(&lev_e->listener);
    }

下面这些函数是全局的供用户调用，中间仅仅调用了上述的静态函数。

//简单调用上述的event_listener_enable
    int
    evconnlistener_enable(struct evconnlistener *lev)
    {
        int r;
        LOCK(lev);
        lev->enabled = 1;
        if (lev->cb)
            r = lev->ops->enable(lev);
        else
            r = 0;
        UNLOCK(lev);
        return r;
    }
    
    int
    evconnlistener_disable(struct evconnlistener *lev)
    {
        int r;
        LOCK(lev);
        lev->enabled = 0;
        r = lev->ops->disable(lev);
        UNLOCK(lev);
        return r;
    }
    
    //释放evconnlistener
    void
    evconnlistener_free(struct evconnlistener *lev)
    {
        LOCK(lev);
        lev->cb = NULL;
        lev->errorcb = NULL;
        if (lev->ops->shutdown)//如果有shutdown，则shutdown
            lev->ops->shutdown(lev);
        //引用次数减一，并解锁
        listener_decref_and_unlock(lev);
    }

static int
    listener_decref_and_unlock(struct evconnlistener *listener)
    {
        int refcnt = --listener->refcnt;//引用计数等于0，则删除
        if (refcnt == 0) {
            //调用event_listener_destroy
            listener->ops->destroy(listener);
            UNLOCK(listener);
            EVTHREAD_FREE_LOCK(listener->lock, EVTHREAD_LOCKTYPE_RECURSIVE);
            mm_free(listener);//释放内存
            return 1;
        } else {
            UNLOCK(listener);
            return 0;
        }
    }

在函数`listener_read_cb`中，一般情况下是不会调用`listener_decref_and_unlock`，但在多线程的时候可能会调用。这种特殊情况是：当主线程accept到一个新客户端时，会解锁，并调用用户设置的回调函数。此时，引用计数等于2。就在这个时候，第二个线程执行`evconnlistener_free`函数。该函数会执行`listener_decref_and_unlock`。明显主线程还在用这个`evconnlistener`，肯定不能删除。此时引用计数也等于2也不会删除。但用户已经调用了`evconnlistener_free`。Libevent必须要响应。当第二个线程执行完后，主线程抢到CPU，此时引用计数就变成1了，也就进入到if判断里面了。在判断体里面执行函数`listener_decref_and_unlock`，并且完成删除工作。

# 例子 #

#include<stdio.h>
    #include<errno.h>
    #include<event.h>
    #include<event2/listener.h>
    
    //accept成功回调
    void listener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *sock, int socklen, void *arg);
    
    void socket_error_cb(bufferevent *bev, short events, void *arg);//bufferevent错误回调
    
    void socket_read_cb(bufferevent* bev, void* arg);//bufferevent可读回调
    
    //先main函数超级简单
    int main(int argc, char** argv)
    {
        struct sockaddr_in sin;
        sin.sin_family = AF_INET;
        sin.sin_addr.s_addr = htonl(INADDR_ANY);
        sin.sin_port = htons(9999);
    
        struct event_base* base = event_base_new();//建立base
    
        //建立listen事件，aceept成功则调用listener_cb
        struct evconnlistener *listener = evconnlistener_new_bind(base ,
                                                                  listener_cb ,
                                                                  base ,
                                                                  LEV_OPT_REUSEABLE|LEV_OPT_CLOSE_ON_FREE | LEV_OPT_THREADSAFE,
                                                                  1024,
                                                                  (struct sockaddr *)&sin,
                                                                  sizeof(struct sockaddr_in));
    
        event_base_dispatch(base);
    
        event_base_free(base);//释放内存
        evconnlistener_free(listener);//释放内存
    
        return 0;
    }
    void
    socket_error_cb(bufferevent *bev, short events, void *arg)
    {
        if (events & BEV_EVENT_EOF)
            printf("connection closed\n");
        else if (events & BEV_EVENT_ERROR)
            printf("some other error\n");
    
        //这将自动close套接字和free读写缓冲区
        bufferevent_free(bev);
    }
    void
    listener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *sock, int socklen, void *arg)
    {
        struct event_base* base = (event_base*)arg;//传递用户参数
    
        //建立bufferevent事件，也就是连接fd
        bufferevent* bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
        //设定读回调函数
        bufferevent_setcb(bev, socket_read_cb , NULL , socket_error_cb, arg);
        //加入epoll
        bufferevent_enable(bev, EV_READ | EV_PERSIST);
    
    }
    void socket_read_cb(bufferevent* bev, void* arg)
    {
        char msg[4096];
        size_t len = bufferevent_read(bev, msg, sizeof(msg));
        msg[len] = '\0';
        printf("recv the client msg: %s", msg);
        char reply_msg[4096] = "I have recvieced the msg: ";
        strcat(reply_msg + strlen(reply_msg), msg);
        bufferevent_write(bev, reply_msg, strlen(reply_msg));
    }
    
    void event_cb(struct bufferevent *bev, short event, void *arg)
    {
    
        if (event & BEV_EVENT_EOF)
            printf("connection closed\n");
        else if (event & BEV_EVENT_ERROR)
            printf("some other error\n");
    
        //这将自动close套接字和free读写缓冲区
        bufferevent_free(bev);
    }

# 参考 #

[https://blog.csdn.net/luotuo44/article/details/38800363][https_blog.csdn.net_luotuo44_article_details_38800363]

[https_blog.csdn.net_luotuo44_article_details_38800363]: https://blog.csdn.net/luotuo44/article/details/38800363