nginx 源码学习笔记（二十二）—— event 模块(三) ——epoll模块

上一节我们讲到了事件驱动的模块，它把我们引入epoll模块，今天我们主要学习下nginx如何使用epoll完成时间驱动，实现高并发；这里不详细讲解epoll原理，如果有机会再做一次单独的epoll的学习。

本文来自于：http://blog.csdn.net/lengzijian

回忆一下上一节的内容，在我们讲到ngx_process_events_and_timers时，在源码最后提到了ngx_process_events,这里是把我们引入epoll的入口：

1.先来看下ngx_process_events的宏定义：

src/event/ngx_event.h

#define ngx_process_events   ngx_event_actions.process_events

2.继续查找ngx_event_actions，我们找到如下结构体：

src/event/ngx_event.h

typedef struct {
    ngx_int_t  (*add)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);
    ngx_int_t  (*del)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);

    ngx_int_t  (*enable)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);
    ngx_int_t  (*disable)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);

    ngx_int_t  (*add_conn)(ngx_connection_t *c);
    ngx_int_t  (*del_conn)(ngx_connection_t *c, ngx_uint_t flags);

    ngx_int_t  (*process_changes)(ngx_cycle_t *cycle, ngx_uint_t nowait);
    ngx_int_t  (*process_events)(ngx_cycle_t *cycle, ngx_msec_t timer,
                   ngx_uint_t flags);

    ngx_int_t  (*init)(ngx_cycle_t *cycle, ngx_msec_t timer);
    void       (*done)(ngx_cycle_t *cycle);
} ngx_event_actions_t;

a.我们去源代码中搜索下关键字ngx_event_actions：

modules/ngx_epoll_module.c:    ngx_event_actions = ngx_epoll_module_ctx.actions;
modules/ngx_select_module.c:    ngx_event_actions = ngx_select_module_ctx.actions;
modules/ngx_poll_module.c:    ngx_event_actions = ngx_poll_module_ctx.actions;

ngx_event.c:ngx_event_actions_t   ngx_event_actions;

前面三行表示：所有event模块对象中的actions就是ngx_event_actions_t对象，而ngx_event_action在第四行定义为全局变量，用于同一接口，下面又存在一个疑问，event模块到底做了些什么？

b.先找到ngx_event_module_t的结构体：

src/event/ngx_event.h

typedef struct {
    ngx_str_t              *name;           //模块名

    void                 *(*create_conf)(ngx_cycle_t *cycle);  //钩子函数，之前讲过
    char                 *(*init_conf)(ngx_cycle_t *cycle, void *conf);//同上

    ngx_event_actions_t     actions;       //接下来主要看
} ngx_event_module_t;

我们找一个例子来详细讲解下

src/event/modules/ngx_epoll_module.c

ngx_event_module_t  ngx_epoll_module_ctx = {
    &epoll_name,
    ngx_epoll_create_conf,               /* create configuration */
    ngx_epoll_init_conf,                 /* init configuration */

    {
        ngx_epoll_add_event,             /* add an event */
        ngx_epoll_del_event,             /* delete an event */
        ngx_epoll_add_event,             /* enable an event */
        ngx_epoll_del_event,             /* disable an event */
        ngx_epoll_add_connection,        /* add an connection */
        ngx_epoll_del_connection,        /* delete an connection */
        NULL,                            /* process the changes */
        ngx_epoll_process_events,        /* process the events */
        ngx_epoll_init,                  /* init the events */
        ngx_epoll_done,                  /* done the events */
    }
};

这里有注释就不详细讲解了。

ngx_process_events这个函数就是我们要找的，要了好大一圈，ngx_process_events实际上就是调用这个函数，此处本人纠结，为什么作者不加点注释呢。

3.下面正式观察下ngx_epoll_init函数：

src/event/modules/ngx_epoll_module.c

static ngx_int_t
ngx_epoll_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
    ngx_epoll_conf_t  *epcf;
    //获取epoll模块的配置结构
    epcf = ngx_event_get_conf(cycle->conf_ctx, ngx_epoll_module);
    //ep是epoll模块定义的一个全局变量，初始化为-1
    if (ep == -1) {
        //创建一个epoll对象，容量为总连接数的一半
        ep = epoll_create(cycle->connection_n / 2);

        if (ep == -1) {
            ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
                          "epoll_create() failed");
            return NGX_ERROR;
        }
    }
    //nevents也是epoll模块的全局变量，初始化为0
    if (nevents < epcf->events) {
        if (event_list) {
            ngx_free(event_list);
        }
        //event_list存储产生时间的数组
        event_list = ngx_alloc(sizeof(struct epoll_event) * epcf->events,
                               cycle->log);
        if (event_list == NULL) {
            return NGX_ERROR;
        }
    }

    nevents = epcf->events;
    /*初始化全局变量ngx_io，ngx_os_io定义为：
    ngx_os_io_t ngx_os_io = {
        ngx_unix_recv,
        ngx_readv_chain,
        ngx_udp_unix_recv,
        ngx_unix_send,
        ngx_writev_chain,
        0
    };(src/os/unix/ngx_posix_init.c)
    */
    ngx_io = ngx_os_io;
    //这里之前讲过
    ngx_event_actions = ngx_epoll_module_ctx.actions;

#if (NGX_HAVE_CLEAR_EVENT)
    //实现边沿触发
    ngx_event_flags = NGX_USE_CLEAR_EVENT
#else
    //实现水平出发
    ngx_event_flags = NGX_USE_LEVEL_EVENT
#endif
                      |NGX_USE_GREEDY_EVENT  //io是知道收到DAGAIN为止
                      |NGX_USE_EPOLL_EVENT;  //epoll标志

    return NGX_OK;
}

4.下面观察下主要的函数ngx_epoll_process_events：

src/event/modules/ngx_epoll_module.c

static ngx_int_t
ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
    int                events;
    uint32_t           revents;
    ngx_int_t          instance, i;
    ngx_uint_t         level;
    ngx_err_t          err;
    ngx_event_t       *rev, *wev, **queue;
    ngx_connection_t  *c;

    /* NGX_TIMER_INFINITE == INFTIM */

    ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
                   "epoll timer: %M", timer);
    //一开支就等待时间，最长等待时间为timer，这里的timer下一节会详细讲解
    events = epoll_wait(ep, event_list, (int) nevents, timer);

    err = (events == -1) ? ngx_errno : 0;

    if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
        //执行一次时间更新，nginx将时间缓存到一组全局变量中，方便程序高效获取事件
        ngx_time_update();
    }
    //wait出错
    if (err) {
        if (err == NGX_EINTR) {

            if (ngx_event_timer_alarm) {
                ngx_event_timer_alarm = 0;
                return NGX_OK;
            }

            level = NGX_LOG_INFO;

        } else {
            level = NGX_LOG_ALERT;
        }

        ngx_log_error(level, cycle->log, err, "epoll_wait() failed");
        return NGX_ERROR;
    }
    //wait返回事件数0，可能是timeout返回，如果不是timeout返回，那么就是error
    if (events == 0) {
        //这里限定timer不是无线超时
        if (timer != NGX_TIMER_INFINITE) {
            return NGX_OK;
        }

        ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
                      "epoll_wait() returned no events without timeout");
        return NGX_ERROR;
    }
    //ngx_posted_events_mutex上锁
    ngx_mutex_lock(ngx_posted_events_mutex);
    //
    for (i = 0; i < events; i++) {
        c = event_list[i].data.ptr;

        instance = (uintptr_t) c & 1;
        //从发生的epoll事件对象中取得ngx_connection_t对象
        c = (ngx_connection_t *) ((uintptr_t) c & (uintptr_t) ~1);
        //取出读事件
        rev = c->read;
        //....
        //取得发生的时间
        revents = event_list[i].events;
        //记录wait的错误返回状态
        if (revents & (EPOLLERR|EPOLLHUP)) {
            ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
                           "epoll_wait() error on fd:%d ev:%04XD",
                           c->fd, revents);
        }
        
        if ((revents & (EPOLLERR|EPOLLHUP))
             && (revents & (EPOLLIN|EPOLLOUT)) == 0)
        {
            /*
             * if the error events were returned without EPOLLIN or EPOLLOUT,
             * then add these flags to handle the events at least in one
             * active handler
             */

            revents |= EPOLLIN|EPOLLOUT;
        }
        //读取一个事件，并且该连接上注册的读时间是active的
        if ((revents & EPOLLIN) && rev->active) {
            if ((flags & NGX_POST_THREAD_EVENTS) && !rev->accept) {
                rev->posted_ready = 1;

            } else {
                rev->ready = 1;
            }
            //时间放入相应的队列中；之前有说过nginx是先放入队列，释放锁之后再做处理
            if (flags & NGX_POST_EVENTS) {
                queue = (ngx_event_t **) (rev->accept ?
                               &ngx_posted_accept_events : &ngx_posted_events);
                /*
                这里根据accept状态
                如果accept为真：加入到ngx_posted_accept_events事件队列中
                如果accept为假：加入到ngx_posted_events事件队列中
                */
                ngx_locked_post_event(rev, queue);//加入到ngx_posted_accept_events队里面

            } else {
                rev->handler(rev);//调用读事件处理函数，通常就是读取事件了
            }
        }
        //取出写事件
        wev = c->write;
        //如果是写事件并且active状态
        if ((revents & EPOLLOUT) && wev->active) {
        
            if (c->fd == -1 || wev->instance != instance) {

                /*
                 * the stale event from a file descriptor
                 * that was just closed in this iteration
                 */

                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
                               "epoll: stale event %p", c);
                continue;
            }

            if (flags & NGX_POST_THREAD_EVENTS) {
                //写事件设为posted_ready状态
                wev->posted_ready = 1;

            } else {
                写事件设为ready状态
                wev->ready = 1;
            }

            if (flags & NGX_POST_EVENTS) {
                //不立即处理，要时间排队，加入队列
                ngx_locked_post_event(wev, &ngx_posted_events);

            } else {
                //调用写事件处理函数，通常就是写入数据
                wev->handler(wev);
            }
        }
    }
    //ngx_mutex_unlock解锁
    ngx_mutex_unlock(ngx_posted_events_mutex);

    return NGX_OK;
}
//其中用到了ngx_locked_post_event()这个宏，它把事件放到事件队列的头部。