【Nginx】事件驱动框架处理流程

经过精简的代码如下，注释中的序号对应上图的序号：

void
ngx_event_accept(ngx_event_t *ev)
{
    socklen_t          socklen;
    ngx_err_t          err;
    ngx_log_t         *log;
    ngx_uint_t         level;
    ngx_socket_t       s;
    ngx_event_t       *rev, *wev;
    ngx_listening_t   *ls;
    ngx_connection_t  *c, *lc;
    ngx_event_conf_t  *ecf;
    u_char             sa[NGX_SOCKADDRLEN];
 
    if (ev->timedout) {
        if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) {
            return;
        }
 
        ev->timedout = 0;
    }
 
    ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);
 
    if (ngx_event_flags & NGX_USE_RTSIG_EVENT) {
        ev->available = 1;
 
    } else if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) {
        ev->available = ecf->multi_accept;
    }
 
    lc = ev->data;
    ls = lc->listening;
    ev->ready = 0;
 
    do {
        socklen = NGX_SOCKADDRLEN;
 
        /* 1、accept方法试图建立连接，非阻塞调用 */
        s = accept(lc->fd, (struct sockaddr *) sa, &socklen);
 
        if (s == (ngx_socket_t) -1)
        {
            err = ngx_socket_errno;
 
            if (err == NGX_EAGAIN)
            {
                /* 没有连接，直接返回 */
                return;
            }
 
            level = NGX_LOG_ALERT;
 
            if (err == NGX_ECONNABORTED) {
                level = NGX_LOG_ERR;
 
            } else if (err == NGX_EMFILE || err == NGX_ENFILE) {
                level = NGX_LOG_CRIT;
            }
 
            if (err == NGX_ECONNABORTED) {
                if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) {
                    ev->available--;
                }
 
                if (ev->available) {
                    continue;
                }
            }
 
            if (err == NGX_EMFILE || err == NGX_ENFILE) {
                if (ngx_disable_accept_events((ngx_cycle_t *) ngx_cycle)
                    != NGX_OK)
                {
                    return;
                }
 
                if (ngx_use_accept_mutex) {
                    if (ngx_accept_mutex_held) {
                        ngx_shmtx_unlock(&ngx_accept_mutex);
                        ngx_accept_mutex_held = 0;
                    }
 
                    ngx_accept_disabled = 1;
 
                } else {
                    ngx_add_timer(ev, ecf->accept_mutex_delay);
                }
            }
 
            return;
        }
 
        /* 2、设置负载均衡阈值 */
        ngx_accept_disabled = ngx_cycle->connection_n / 8
                              - ngx_cycle->free_connection_n;
 
        /* 3、从连接池获得一个连接对象 */
        c = ngx_get_connection(s, ev->log);
 
        /* 4、为连接创建内存池 */
        c->pool = ngx_create_pool(ls->pool_size, ev->log);
 
        c->sockaddr = ngx_palloc(c->pool, socklen);
 
        ngx_memcpy(c->sockaddr, sa, socklen);
 
        log = ngx_palloc(c->pool, sizeof(ngx_log_t));
 
        /* set a blocking mode for aio and non-blocking mode for others */
        /* 5、设置套接字属性为阻塞或非阻塞 */
        if (ngx_inherited_nonblocking) {
            if (ngx_event_flags & NGX_USE_AIO_EVENT) {
                if (ngx_blocking(s) == -1) {
                    ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_socket_errno,
                                  ngx_blocking_n " failed");
                    ngx_close_accepted_connection(c);
                    return;
                }
            }
 
        } else {
            if (!(ngx_event_flags & (NGX_USE_AIO_EVENT|NGX_USE_RTSIG_EVENT))) {
                if (ngx_nonblocking(s) == -1) {
                    ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_socket_errno,
                                  ngx_nonblocking_n " failed");
                    ngx_close_accepted_connection(c);
                    return;
                }
            }
        }
 
        *log = ls->log;
 
        c->recv = ngx_recv;
        c->send = ngx_send;
        c->recv_chain = ngx_recv_chain;
        c->send_chain = ngx_send_chain;
 
        c->log = log;
        c->pool->log = log;
 
        c->socklen = socklen;
        c->listening = ls;
        c->local_sockaddr = ls->sockaddr;
        c->local_socklen = ls->socklen;
 
        c->unexpected_eof = 1;
 
        rev = c->read;
        wev = c->write;
 
        wev->ready = 1;
 
        if (ngx_event_flags & (NGX_USE_AIO_EVENT|NGX_USE_RTSIG_EVENT)) {
            /* rtsig, aio, iocp */
            rev->ready = 1;
        }
 
        if (ev->deferred_accept) {
            rev->ready = 1;
 
        }
 
        rev->log = log;
        wev->log = log;
 
        /*
         * TODO: MT: - ngx_atomic_fetch_add()
         *             or protection by critical section or light mutex
         *
         * TODO: MP: - allocated in a shared memory
         *           - ngx_atomic_fetch_add()
         *             or protection by critical section or light mutex
         */
 
        c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
 
        if (ls->addr_ntop) {
            c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len);
            if (c->addr_text.data == NULL) {
                ngx_close_accepted_connection(c);
                return;
            }
 
            c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen,
                                             c->addr_text.data,
                                             ls->addr_text_max_len, 0);
            if (c->addr_text.len == 0) {
                ngx_close_accepted_connection(c);
                return;
            }
        }
 
        /* 6、将新连接对应的读写事件添加到epoll对象中 */
        if (ngx_add_conn && (ngx_event_flags & NGX_USE_EPOLL_EVENT) == 0) {
            if (ngx_add_conn(c) == NGX_ERROR) {
                ngx_close_accepted_connection(c);
                return;
            }
        }
 
        log->data = NULL;
        log->handler = NULL;
 
        /* 7、TCP建立成功调用的方法，这个方法在ngx_listening_t结构体中 */
        ls->handler(c);
 
    } while (ev->available);    /* available标志表示一次尽可能多的建立连接，由配置项multi_accept决定 */
}

Nginx中的“惊群”问题

Nginx一般会运行多个worker进程，这些进程会同时监听同一端口。当有新连接到来时，内核将这些进程全部唤醒，但只有一个进程能够和客户端连接成功，导致其它进程在唤醒时浪费了大量开销，这被称为“惊群”现象。Nginx解决“惊群”的方法是，让进程获得互斥锁ngx_accept_mutex，让进程互斥地进入某一段临界区。在该临界区中，进程将它所要监听的连接对应的读事件添加到epoll模块中，使得当有“新连接”事件发生时，该worker进程会作出反应。这段加锁并添加事件的过程是在函数ngx_trylock_accept_mutex中完成的。而当其它进程也进入该函数想要添加读事件时，发现互斥锁被另外一个进程持有，所以它只能返回，它所监听的事件也无法添加到epoll模块，从而无法响应“新连接”事件。但这会出现一个问题：持有互斥锁的那个进程在什么时候释放互斥锁呢？如果需要等待它处理完所有的事件才释放锁的话，那么会需要相当长的时间。而在这段时间内，其它worker进程无法建立新连接，这显然是不可取的。Nginx的解决办法是：通过ngx_trylock_accept_mutex获得了互斥锁的进程，在获得就绪读/写事件并从epoll_wait返回后，将这些事件归类放入队列中：

• 新连接事件放入ngx_posted_accept_events队列
• 已有连接事件放入ngx_posted_events队列

if (flags & NGX_POST_EVENTS)
{
        /* 延后处理这批事件 */
        queue = (ngx_event_t **) (rev->accept ? &ngx_posted_accept_events : &ngx_posted_events);
 
        /* 将事件添加到延后执行队列中 */
        ngx_locked_post_event(rev, queue);
}
else
{
        rev->handler(rev);  /* 不需要延后，则立即处理事件 */
}

写事件做类似处理。进程接下来处理ngx_posted_accept_events队列中的事件，处理完后立即释放互斥锁，使该进程占用锁的时间降到了最低。

Nginx中的负载均衡问题

Nginx中每个进程使用了一个处理负载均衡的阈值ngx_accept_disabled，它在上图的第2步中被初始化：

ngx_accept_disabled = ngx_cycle->connection_n / 8 - ngx_cycle->free_connection_n;

它的初值为一个负数，该负数的绝对值等于总连接数的7/8.当阈值小于0时正常响应新连接事件，当阈值大于0时不再响应新连接事件，并将ngx_accept_disabled减1，代码如下：

if (ngx_accept_disabled > 0)
{
        ngx_accept_disabled--;
}
else
{
    if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR)
    {
        return;
    }
    ....
}

这说明，当某个进程当前的连接数达到能够处理的总连接数的7/8时，负载均衡机制被触发，进程停止响应新连接。

参考：

《深入理解Nginx》 P328-P334.