Nginx 源代码笔记 - 延迟操作

本文目录

deferred accept
posted event
timer

这篇讨论一下 Nginx 中关于 “延迟” 相关的话题。Nginx “延迟” 相关的逻辑，会将某些处理拖延到合适的时机，会将某些分散的逻辑集中到一起进行批量处理，也会在计划中的某个将来的时间点完成某个操作。

这些 “延迟” 逻辑，包括 deferred accept、posted event 和 timer。

deferred accept

根据定义，deferred accept 将 accept() 的处理时机推迟到三次握手成功完成后的连接上有可读数据后，然后才向 listening socket 的监听者投递读事件。这样一来，一旦 accept() 成功，Nginx 不用等待新接入连接上有读事件发生，便可以马上从其中读取请求数据。这样，就提高了请求的处理效率。

man 7 tcp

TCP_DEFER_ACCEPT (since Linux 2.4) Allow a listener to be awakened only when data arrives on the socket. Takes an integer value (seconds), this can bound the maximum number of attempts TCP will make to complete the connection.

其实现原理大致是：要求三次握手过程的最后一握为实际数据，这时此连接才真正进行 ESTABLISHED 状态。参数值被协议栈用作最后一握的超时时间。具体实现逻辑可 google 之。

Linux 2.4 以后，才开始支持此特性。Nginx 根据 listen 配置指令的 deferred 选项和 TCP_DEFER_ACCEPT 是否定义，来决定是否打开 listening socket 的这个选项。

关于 deferred_accept 的存储和设置由下面代码完成，add_deferred 和 delete_deferred 指令用于表示在配置 ngx_listening_t 过程中，开启或关闭对应 listening socket 的 deferred accept 属性。初始化和配置代码如下：

    ----------core/ngx_connection.c:17--------
    struct ngx_listening_s {
        ngx_socket_t        fd;
        ...
    #define (NGX_HAVE_DEFERRED_ACCEPT)
        unsigned            deferred_accept:1;
        unsigned            delete_deferred:1;
        unsigned            add_deferred:1;
        ...
    #endif
    };

    ----------http/ngx_http.c:1679------------
    static ngx_listening_t *
    ngx_http_add_listening(ngx_conf_t *cf, ngx_http_conf_addr_t *addr)
    {
        ...
    #if (NGX_HAVE_DEFERRED_ACCEPT && defined TCP_DEFER_ACCEPT)
        ls->deferred_accept = addr->opt.deferred_accept;
    #endif
        ...
    }

    ---------core/ngx_cycle.c:40--------------
    ngx_cycle_t *
    ngx_init_cycle(ngx_cycle_t *old_cycle)
    {
        ...
            ls = cycle->listening.elts;
            for (i = 0; i < cycle->listening.nelts; i++) {
                ls[i].open = 1;
                ...
    #if (NGX_HAVE_DEFERRED_ACCEPT && defined TCP_DEFER_ACCEPT)
                if (ls[i].deferred_accept) {
                    ls[i].add_deferred = 1;
                }
    #endif
            }
         ...
    }


    ----------core/ngx_connection.c:439-------
    void
    ngx_configure_listening_sockets(ngx_cycle_t *cycle)
    {
        ...
        ls = cycle->listening.elts;
        for (i = 0; i < cycle->listening.nelts; i++) {
            ...
    #ifdef TCP_DEFER_ACCEPT
            if (ls[i].add_deferred || ls[i].delete_deferred) {
                if (ls[i].add_deferred) {
                    timeout = (int) (ls[i].post_accept_timeout / 1000);
                } else {
                    timeout = 0;
                }

                setsocketopt(ls[i].fd, IPPROTO_TCP, TCP_DEFER_ACCEPT,
                            &timeout, sizeof(int))
            }

            if ls[i].add_deferred) {
                ls[i].deferred_accept = 1;
            }
    #endif
        ...
        }
    }

对以上代码的补充说明：

ls->post_accept_timeout 的值是配置指令项 client_header_timeout 的值。
delete_deferred 生效时，timeout 被置为 0，随后的 setsocketopt 就禁用了此 socket 的 TCP_DEFER_ACCEPT 特性。

随后，listening socket 对应的读事件结构体成员 deferred_accept 也被置 1。这样，一旦从此 listening socket 接收了新的连接，新连接可以被认为已经有数据可用 (新连接的读事件结构体 ready 字段被置为 1)。

    -----------------event/ngx_event.c:579---------------
    static ngx_int_t
    ngx_event_process_init(ngx_cycle_t *cycle)
    {
        ...
        ls = cycle->listening.elts;
        for (i = 0; i < cycle->listening.nelts; i++) {
            ...
    #if (NGX_HAVE_DEFERRED_ACCEPT)
            rev->deferred_accept = ls[i].deferred_accept;
    #endif
            ...
        }
        ...
    }

    ----------------event/ngx_event_accept.c:17------------
    void
    ngx_event_accept(ngx_event_t *ev)
    {
        ...
        do {
            s = accept(lc->fd, (struct sockaddr *) sa, &socklen);
            ...
            if (ev->deferred_accept) {
                rev->ready = 1;
                ...
            }
            ...
        } while (ev->available);
    }

新连接初始化完成后，如果其已有数据可用 (rev->ready == 1)，直接试着从已有数据中读取 HTTP request 的相关信息。否则，Nginx 需要将此连接加入 epoll set 中，重新监听读事件，随后才能继续分析从此连接上来的 HTTP request 信息。

    ---------------http/ngx_http_request.c:177-------------
    void
    ngx_http_init_connection(ngx_connection_t *c)
    {
        ...
        if (rev->ready) {
            /* the deferred accept(), rtsig, aio, iocp */
            ...
            ngx_http_init_request(rev);
            return;
        }
        ...
        ngx_handle_read_event(rev, 0);
        ...
    }

deferred accept 简化了 HTTP request 处理逻辑，将部分数据的读取交由 TCP/IP 栈完成，这样可以提升 Nginx 的执行效率，减少请求的响应时间。

posted event

在 ngx_use_accept_mutex 选项打开情况下，worker 进程接收新的连接之前都需要竞争 accept_mutex 锁 (详细说明)。由于这个锁是进程级的，为了尽量降低 worker 进程间的互斥影响，Nginx 将尽可能多的操作挪到进程级互斥区之外执行。

避免 listening sockets 上产生的读事件造成的 epoll 惊群是加入 accept_mutex 的首要目的，而普通连接产生的读写事件和 listening sockets 上产生的读事件都需要通过 epoll 获取。为了尽量减少互斥区的操作，Nginx 将普通连接上的读写事件处理操作挪到了互斥区外，这就是读写事件延迟链表 ngx_posted_events 的作用。同时，为了事件监听部分代码的一致性，必须在互斥区内完成的新连接接入操作 (accept()) 对应的读事件，也由相似的链表 ngx_posted_accept_events 维护。

主要代码如下：

    -----------event/ngx_event.c:200------------
    void
    ngx_process_events_and_timers(ngx_cycle_t *cycle)
    {
        ...
        if (ngx_posted_accept_events) {
            ngx_event_process_posted(cycle, &ngx_posted_accept_events);
        }

        if (ngx_accept_mutex_held) {
            ngx_shmtx_unlock(&ngx_accept_mutex);
        }
        ...
        if (ngx_posted_events) {
            if (ngx_threaded) {
                ngx_wakeup_worker_thread(cycle);

            } else {
                ngx_event_process_posted(cycle, &ngx_posted_events);
            }
        }
    }

对上述代码的补充：

ngx_posted_accept_events 维护的读事件，在离开互斥区，即释放 accept_mutex 前执行；ngx_posted_events 维护的普通读写事件，可以放到离开互斥区分执行。

而 ngx_event_process_posted 函数的主要逻辑，就是逐个调用事件链表中的回调函数：

    ----------event/ngx_event_posted.c:20----------------
    void
    ngx_event_process_posted(ngx_cycle_t *cycle,
        ngx_thread_volatile ngx_event_t **posted)
    {
        ngx_event_t *ev;

        for ( ;; ) {

            ev = (ngx_event_t *) *posted;
            ...
            ngx_delete_posted_event(ev);

            ev->handler(ev);
        }
    }

对上述代码的补充说明：

注意一下 posted event 链表操作中对 ngx_event_t **prev 指针的使用。

在互斥区中执行的连接初始化等操作，也会尽量延迟到互斥区以外。比如，deferred accept 部分分析的情况，连接被 accept() 后，其上已有请求数据等待读取，在初始化完成后：

    ----------http/ngx_http_request.c:177------------
    void
    ngx_http_init_connection(ngx_connection_t *c)
    {
        ...
        if (rev->read) {
            /* the deferred accept(), rtsig, aio, iocp */

            if (ngx_use_accept_mutex) {
                ngx_post_event(rev, &ngx_posted_events);
                return;
            }

            ngx_http_init_request(rev);
            return;
        }
        ...
    }

HTTP request 处理过程另外一处对 posted event 的使用出现在 ngx_http_set_keepalive 函数中。当连接使用 keepalive 模式时，一个请求处理完毕后并不立即关闭连接，而继续接收处理此连续上更多的请求 (pipeline)。如果，此连接上还有数据可读或者连接 buffer 还有数据可用，Nginx 就将连接的读事件结构体加入到 ngx_posted_events 链表中，在 ngx_event_process_posted 开始下一个请求的处理。

    ------------http/ngx_http_request.c:2354------------
    static void
    ngx_http_set_keepalive(ngx_http_request_t *r)
    {
        ...
        if (b->pos < b->last) {
            ...
            hc->pipeline = 1;
            ...
            rev->handler = ngx_http_init_request;
            ngx_post_event(rev, &ngx_posted_events);
            return;
        }
        ...
        if (rev->ready) {
            ngx_post_event(rev, &ngx_posted_events);
        }
    }

timer

定时器在程用中用于在将来的某个时刻执行某些操作，它在任何一个服务端进程中不可或缺。它在 Nginx 中用于连接的超时管理、服务器时间戳更新、缓存管理和其它需要延迟一定时间后执行的操作。

应用层的定时器的驱动机制和节点存储访问也多种多样：

* 常见的驱动方式有独立线程驱动、系统定时信号驱动和带有超时特性的系统阻塞调
用等等。 Nginx 使用带有超时特性的事件监听系统调用，比如 `epoll_wait`，从而将
网络事件处理和超时处理集中到一个逻辑中统一处理。

* 而定时器节点的存储方式也多种多样，`heap`, `queue`, `binary search tree`,
`hash` 等等。Nginx 使用 `rbtree` 存储定时器节点，对各种操作 (插入，查找) 复
杂度进行了折衷。

这里，对定时器的各种实现不做分析和比较，只来关注一下 Nginx 定时器的实现和各种操作。

Nginx 定时器由 rbtree 树 ngx_event_timer_rbtree 表示，定时器的每个节点都是一个 ngx_event_t 类型的变量。

    struct ngx_event_s {
        ...
        unsigned            timedout:1;
        unsigned            timer_set:1;
        ...
        ngx_rbtree_node_t   timer;
        ...
    };

基本操作

定时器 ngx_event_timer_rbtree 支持 ngx_event_t 类型节点的添加和删除，同时使用者也得到第一个超时节点的超时时间 (第一个超时事件发生时间)。

节点加入，基本操作就是往红黑树中插入新节点的过程。

    --------------event/ngx_event_timer.h:57------------
    static ngx_inline void
    ngx_event_add_timer(ngx_event_t *ev, ngx_msec_t timer)
    {
        ...
        key = ngx_current_msec + timer;

        if (ev->timer_set) {
            ...
            diff = (ngx_msec_int_t) (key - ev->timer.key);

            if (ngx_abs(diff) < NGX_TIMER_LAZY_DELAY) {
                ...
                return;
            }

            ngx_del_timer(ev);
        }

        ev->timer.key = key;
        ...
        ngx_rbtree_insert(&ngx_event_timer_rbtree, &ev->timer);
        ...
        ev->timer_set = 1;
    }

对上述代码的补充说明：

ev->timer_set 字段记录 ngx_event_t 节点是否已经加入到了定时器中。
使用节点超时时间作为其在 rbtree 中的 key。
在加入节点前，先检查此节点是否已经在定时器中。如果它已经在定时器时，再检查一下节点的超时时间和新设定的超时时间是否相差不大。差异可以容忍时，不再重新插入此节点。

节点删除操作也比较简单：从 rbtree 中删除此节点，然后将其 timer_set 字段置 0。这里就不摘录代码了。

定时器第一个超时节点查找，是很重要的操作。这个函数能告诉调用者，它如果需要进入阻塞状态的话，阻塞多长时间才不会错过第一个超时事件的处理。

    ----------------event/ngx_event_timer.c:50------------
    ngx_msec_t
    ngx_event_find_timer(void)
    {
        ...
        root = ngx_event_timer_rbtree.root;
        sentinel = ngx_event_timer_rbtree.sentinal;

        node = ngx_rbtree_min(root, sentinel);
        ...
        timer = (ngx_msec_int_t) node->key - (ngx_msec_int_t) ngx_current_msec;

        return (ngx_msec_t) (timer > 0 ? timer : 0);
    }

超时事件处理

接下来，再回到事件处理函数 ngx_process_events_and_timers 中。这个函数就是 worker 进程阻塞等待网络事件和超时事件，并处理这些事件的主函数。

网络事件何时发生是不确定的，而第一个超时事件何时发生在阻塞前是可以从定时器中获取到的。这样，为了及时处理第一个超时事件，worker 进程的阻塞函数需要在超时事件发生前返回。

    --------------event/ngx_event.c:206--------------
    if (ngx_timer_resolution) {
        timer = NGX_TIMER_INFINITE;
        flags = 0;

    } else {
        timer = ngx_event_find_timer();
        flags = NGX_UPDATE_TIME;

    }
    ...
    delta = ngx_current_msec;

    (void *) ngx_process_events(cycle, timer, flags);

    delta = ngx_current_msec - delta;
    ...
    if (delta) {
        ngx_event_expire_timers();
    }

对上述代码的补充说明：

ngx_timer_resolution 由配置指令 timer_resolution 打开。这个选项可以用于指定计算 ngx_current_msec 的间隔，以提高它的精确度。这个选项打开时，用于计算 ngx_current_msec 的函数 ngx_time_update 由系统定时信号处理函数完成。选项关闭时，ngx_time_update 在阻塞函数 (epoll_wait etc.) 返回时才被设用 (NGX_UPDATE_TIME)。
delta 存储 ngx_process_events 函数的执行耗时 (并不精确)。
由 delta 的值决定是否检查超时事件。

随后，在 ngx_event_expire_timers 函数中检查是否有超时事件发生，并对超时事件进行处理。

    ------------event/ngx_event_timer.c:75--------------
    void
    ngx_event_expire_timers(void)
    {
        ...
        sentinel = ngx_event_timer_rbtree.sentinel;

        for ( ;; ) {
            ...
            root = ngx_event_timer_rbtree.root;
            ...
            node = ngx_rbtree_min(root, sentinel);

            if ((ngx_msec_int_t) node->key - (ngx_msec_int_t) ngx_current_msec <= 0) {
                ev = (ngx_event_t *) ((char *) node - offsetof(ngx_event_t, timer));
                ...
                ngx_rbtree_delete(&ngx_event_timer_rbtree, &ev->timer);
                ...
                ev->timer_set = 0;
                ...
                ev->timedout = 1;

                ev->handler(ev);

                continue;
            }

            break;
        }
    }

定时器用例

这一部分展示一下定时器是如何被使用的。首先定时器节点的增加和删除是通过下面两个宏完成的：

    #define ngx_add_timer        ngx_event_add_timer                                                                                                                |~
    #define ngx_del_timer        ngx_event_del_timer

在请求接收和处理过程中，Nginx 需要对每个涉及到网络事件的操作进行超时跟踪，以便及时清理因为事件超时而占用的资源。

比如，在从连接上读取 HTTP request 包头时，如果此时连接上无数据可用的话，Nginx 就会在监听读事件的同时，对连接设定超时处理：

    ---------------http/ngx_http_request.c:1092------------
    static ssize_t
    ngx_http_read_request_header(ngx_http_request_t *r)
    {
        ...
        if (rev->ready) {
            n = c->recv(c, r->header_in->last,
                        r->header_in->end - r->header_in->last);
        } else {
            n = NGX_AGAIN;
        }

        if (n == NGX_AGAIN) {
            if (!rev->timer_set) {
                cscf = ngx_http_get_module_srv_conf(r, ngx_http_core_module);
                ngx_add_timer(rev, cscf->client_header_timeout);
            }

            if (ngx_handle_read_event(rev, 0) != NGX_OK) {
                ngx_http_close_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
                return NGX_ERROR;
            }

            return NGX_AGAIN;
        }
        ...
    }

这个连接的读事件结构体 ngx_event_t 的回调函数 (handler) 是 ngx_http_process_request_line，也就是说，如果这时 rev 上发生超时事件，Nginx 将 rev->timedout 置 1 后，会调用 ngx_http_process_request_line 处理超时：

    ----------------http/ngx_http_request.c:680--------------
    static void
    ngx_http_process_request_line(ngx_event_t *rev)
    {
        ...
        c = rev->data;
        r = c->data;
        ...
        if (rev->timedout) {
            c->timedout = 1;
            ngx_http_close_request(r, NGX_HTTP_REQUEST_TIME_OUT);
            return;
        }
        ...
    }

于是，当读 HTTP request 包头时，在 cscf->client_header_timeout 的时间内，如果 Nginx 没有收到客户端发来的包头数据时，这个请求及至请求来自的连接都会被回收。

不要轻轻地离开我，请留下点什么...

Comments

Nginx 源代码笔记 - 延迟操作

deferred accept

posted event

timer

基本操作

超时事件处理

定时器用例

Comments

Published

Category

Tags

Contact