LibEvent代码阅读--服务器端Listen、Accept处理

在socket的服务器标准编程过程主要包含如下的基本过程：socket()-->bind()-->lister()-->accept()-->新的socket处理(收发报文等)。在编写最基本服务器代码的过程就是如上的几个步骤，在支持了IO复用之后，只是将accept的处理过程并入到了事件处理中。

今天就libevent的服务器的创建基本过程进行分析，如果需要创建对应的服务器，因此必须有一个进行监听的处理，在libevent中的listener.c文件实现了服务器进行监听的基本过程。

数据结构

在Listener中主要实现了如下的三个数据结构，这三个数据结构中的struct evconnlistener用于对外提供，另外两个数据结构在内部使用。其中的struct evconnlistener_event实现了将监听与libevet中事件进行绑定，也就是改监听结构体对应一个具体的事件。

/* 监听的操作结构体 */
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;

    /* 新连接的callback */
    evconnlistener_cb cb;
    evconnlistener_errorcb errorcb;
    void *user_data;
    unsigned flags;
    short refcnt;
    unsigned enabled : 1;
};

/* 监听事件，包含了其中的监听对应的事件和监听结构体 */
struct evconnlistener_event {
      /* 监听结构体 */
    struct evconnlistener base;
    /* 具体的事件 */
    struct event listener;
};

最后一个结构体在使用中都是通过base这个成员确定的，也就是根据变量的偏移地址确定，这也是根据成员变量计算结构体起始地址进程会使用的。
#define EVUTIL_UPCAST(ptr, type, field) \
((type *)(((char*)(ptr)) - evutil_offsetof(type, field)))
struct evconnlistener_event *lev_e =
   EVUTIL_UPCAST(lev, struct evconnlistener_event, base);
在内部代码使用中主要通过evconnlistener_event进行代码的编写。

对外提供接口

Listener.h中对外提供的接口主要包括：

//该接口用于实现监听操作的实现
struct evconnlistener *evconnlistener_new(struct event_base *base,
    evconnlistener_cb cb, void *ptr, unsigned flags, int backlog,
    evutil_socket_t fd);
//该接口用于实现绑定和监听
struct evconnlistener *evconnlistener_new_bind(struct event_base *base,
    evconnlistener_cb cb, void *ptr, unsigned flags, int backlog,
    const struct sockaddr *sa, int socklen);

其中的cb主要用于使用者对新accept的socket进行处理，这部分见后面的代码分析。

代码分析

首先是监听操作函数的初始化:

/* 监听器的操作函数结合初始化 */
static const struct evconnlistener_ops evconnlistener_event_ops = {
    event_listener_enable,
    event_listener_disable,
    event_listener_destroy,
    NULL, /* shutdown */
    /* 获得对应的fd，监听的fd */
    event_listener_getfd,
    /* 获得对应的event_base */
    event_listener_getbase
};

从接口进行分析:
evconnlistener_new():

/* 对外提供的接口，用于创建一个用于监听连接的数据结构,其中的参数cb是有客户端连接过来之后的处理 */
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;

    /* 实际的处理函数，创建对应的监听，默认的log为128，提高并发的一个关键参数 */
    if (backlog > 0) {
        if (listen(fd, backlog) < 0)
            return NULL;
    } else if (backlog < 0) {
        if (listen(fd, 128) < 0)
            return NULL;
    }

    /* 创建一个监听事件(其中包含了具体的事件),也就是为对应的fd创建关注的事件 */
    lev = mm_calloc(1, sizeof(struct evconnlistener_event));
    if (!lev)
        return NULL;
    /* 连接的操作函数 */
    lev->base.ops = &evconnlistener_event_ops;
    lev->base.cb = cb;
    lev->base.user_data = ptr;
    lev->base.flags = flags;
    lev->base.refcnt = 1;

    /* 分配一个对应的读事件和永久事件标号给对应的监听fd,对应的响应函数为listener_read_cb */
    event_assign(&lev->listener, base, fd, EV_READ|EV_PERSIST,
        listener_read_cb, lev);

    /* 使能对应的监听，实际上实现了上述分配事件的添加 */
    evconnlistener_enable(&lev->base);

    return &lev->base;
}

其中最后的几行代码主要采用了event_assign()分配了一个新的事件，主要接收读事件，对应的回调函数为listener_read_cb。evconnlistener_enable()中实现了事件的添加操作，见后面的分析。
evconnlistener_new_bind():

/* 从创建到bind到监听的一系列处理过程 */
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;

    fd = socket(family, SOCK_STREAM, 0);
    if (fd == -1)
        return NULL;

    if (evutil_make_socket_nonblocking(fd) < 0) {
        evutil_closesocket(fd);
        return NULL;
    }

    if (flags & LEV_OPT_CLOSE_ON_EXEC) {
        if (evutil_make_socket_closeonexec(fd) < 0) {
            evutil_closesocket(fd);
            return NULL;
        }
    }

    /* 设置保活 */
    if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (void*)&on, sizeof(on))<0) {
        evutil_closesocket(fd);
        return NULL;
    }
    /* 设置地址可反复的利用 */
    if (flags & LEV_OPT_REUSEABLE) {
        if (evutil_make_listen_socket_reuseable(fd) < 0) {
            evutil_closesocket(fd);
            return NULL;
        }
    }

    if (sa) {
        /* 执行对应的bind操作，也就是将对应的fd绑定的对应的地址和端口上 */
        if (bind(fd, sa, socklen)<0) {
            evutil_closesocket(fd);
            return NULL;
        }
    }

    /* 创建对应的监听 */
    listener = evconnlistener_new(base, cb, ptr, flags, backlog, fd);
    if (!listener) {
        evutil_closesocket(fd);
        return NULL;
    }

    return listener;
}

从上述的代码中可知，evconnlistener_new_bind中包含了所有的创建，bind，监听操作。

listener_read_cb()在evconnlistener_new作为监听事件的回调函数进行添加，该事件在有客户端连接到服务以后就会被触发调用，基本的实现逻辑如下所示:

/* 监听socket接收到对应的新客户端连接事件后的处理 */
static void
listener_read_cb(evutil_socket_t fd, short what, void *p)
{
    struct evconnlistener *lev = p; //强制类型转换，evconnlistener和evconnlistener_event地址一致
    int err;
    evconnlistener_cb cb;
    evconnlistener_errorcb errorcb;
    void *user_data;
    LOCK(lev);
    while (1) {
        struct sockaddr_storage ss;
        socklen_t socklen = sizeof(ss);

        /* accept返回值为新的fd */
        evutil_socket_t new_fd = accept(fd, (struct sockaddr*)&ss, &socklen);
        if (new_fd < 0)
            break;//非阻塞模式的好处
        if (socklen == 0) {
            evutil_closesocket(new_fd);
            continue;
        }

        /* 设置属性，通常设备为不可阻塞 */
        if (!(lev->flags & LEV_OPT_LEAVE_SOCKETS_BLOCKING))
            evutil_make_socket_nonblocking(new_fd);

        /* 如无cb，理论上应该不会accept，因此出错 */
        if (lev->cb == NULL) {
            UNLOCK(lev);
            return;
        }
        ++lev->refcnt; //增加接收到socket个数
        /* 对应的回调函数 */
        cb = lev->cb;
        user_data = lev->user_data;
        /* XXX：这边释放锁的操作，主要防止对调函数中进行不可预估的操作，因此释放锁 */
        UNLOCK(lev);
        /* 执行对应的回调函数，也就是说具体的处理由使用者决定，由使用者对new_fd处理 */
        cb(lev, new_fd, (struct sockaddr*)&ss, (int)socklen,
            user_data);
        /* XXX：这边重新添加锁，是编写lib库需要值得学习的地方，能避免反复加锁 */
        LOCK(lev);
        if (lev->refcnt == 1) {
            int freed = listener_decref_and_unlock(lev);
            EVUTIL_ASSERT(freed);
            return;
        }
        --lev->refcnt;
    }
    ...
}

这部分有关锁的处理值得借鉴，我之前在写lib库的过程中提供的回调函数，在使用的过程中通常强制要求用户不能进行锁的操作，而这边采用了回调前后释放和重新加锁的操作，这样防止使用者反复的加锁，同时使得使用者更加的方便处理。

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;
}

由该函数分析可知，在当前监听器的回调函数不存在的情况下，实际上监听操作的使能操作并没有执行，因此回调函数是必须提供的。其中的ops->enable()如下所示:

/* 监听事件的添加操作 */
static int
event_listener_enable(struct evconnlistener *lev)
{
    /* 根据对外提供的evconnlistener获取evconnlistener_event的地址，本来强制转换即可，但是为了可移植性 */
    struct evconnlistener_event *lev_e =
        EVUTIL_UPCAST(lev, struct evconnlistener_event, base);  //计算event的地址
    /* 将之前分配的事件添加到base中，因此完成了事件的添加操作 */
    return event_add(&lev_e->listener, NULL);
}

上述主要实现了evconnlistener_event两个成员之间的转换，并将对应事件添加到IO复用模型中。因此在没有提供回调的情况下，该函数并不会被调用，也就不会触发listener_read_cb()执行，因此在listener_read_cb中如cb为null,说明出现了异常。

以上的代码基本实现了服务创建的基本过程，实现了socket从创建到accept的过程。