nginx 异步 IO 初始化 -- ngx_epoll_aio_init

2015-06-24 13:38:20   最后更新: 2015-06-24 14:33:47   访问数量:1132




在下面两篇日志中,我们详细介绍了 linux 异步 IO 的两种实现:

POSIX AIO -- glibc 版本异步 IO 简介

linux AIO -- libaio 实现的异步 IO 简介及实现原理

在文章中,我们清楚 libaio 实现的异步 IO 需要 linux 内核版本支持,然而,他却比 glibc 实现的异步 IO 具有很多优势,如资源使用上、执行效率上以及设备的协调使用等方面均有着明显的优势

无论如何,他毕竟是 linux 内核原生支持的异步 IO 嘛

也因此,nginx 选择了 libaio 实现的异步 IO 实现事件的异步处理,那么他是怎么做的呢?

 

ngx_epoll_aio_init 函数是 nginx epoll 模块初始化的一个重要环节,他决定了 nginx 是否使用异步 IO 来实现事件的读写响应

// static void ngx_epoll_aio_init(ngx_cycle_t *cycle, ngx_epoll_conf_t *epcf) // 初始化异步 IO 机制 {{{ static void ngx_epoll_aio_init(ngx_cycle_t *cycle, ngx_epoll_conf_t *epcf) { int n; struct epoll_event ee; // 调用 eventfd 获取 eventfd 对象,用来在内核态与用户态之间传递通知 #if (NGX_HAVE_SYS_EVENTFD_H) ngx_eventfd = eventfd(0, 0); #else // syscall 直接调用指定 id 的系统调用,即系统并没有开放用户态接口 ngx_eventfd = syscall(SYS_eventfd, 0); #endif if (ngx_eventfd == -1) { ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno, "eventfd() failed"); ngx_file_aio = 0; return; } ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "eventfd: %d", ngx_eventfd); n = 1; // 设置 eventfd 为非阻塞 if (ioctl(ngx_eventfd, FIONBIO, &n) == -1) { ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno, "ioctl(eventfd, FIONBIO) failed"); goto failed; } // 创建异步 IO 上下文 if (io_setup(epcf->aio_requests, &ngx_aio_ctx) == -1) { ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno, "io_setup() failed"); goto failed; } // 异步 IO 回调函数 ngx_epoll_eventfd_handler 的参数,nginx 连接结构 ngx_eventfd_event.data = &ngx_eventfd_conn; // 绑定事件回调函数 ngx_eventfd_event.handler = ngx_epoll_eventfd_handler; ngx_eventfd_event.log = cycle->log; ngx_eventfd_event.active = 1; ngx_eventfd_conn.fd = ngx_eventfd; ngx_eventfd_conn.read = &ngx_eventfd_event; ngx_eventfd_conn.log = cycle->log; ee.events = EPOLLIN|EPOLLET; ee.data.ptr = &ngx_eventfd_conn; // 将 eventfd 添加到 epoll 中 if (epoll_ctl(ep, EPOLL_CTL_ADD, ngx_eventfd, &ee) != -1) { return; } ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno, "epoll_ctl(EPOLL_CTL_ADD, eventfd) failed"); // 执行失败则删除异步 IO 上下文 if (io_destroy(ngx_aio_ctx) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "io_destroy() failed"); } failed: // 关闭 eventfd if (close(ngx_eventfd) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "eventfd close() failed"); } ngx_eventfd = -1; ngx_aio_ctx = 0; ngx_file_aio = 0; } // }}}

 

 

这里用到了一个较新的系统调用 eventfd,他的用法可以参看:

用户态与内核态的事件通知 -- eventfd

它实现了用户态应用对内核态通知的获取

 

// static void ngx_epoll_eventfd_handler(ngx_event_t *ev) // 异步 IO 回调函数 {{{ static void ngx_epoll_eventfd_handler(ngx_event_t *ev) { int n, events; long i; uint64_t ready; ngx_err_t err; ngx_event_t *e; ngx_event_aio_t *aio; struct io_event event[64]; struct timespec ts; ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, 0, "eventfd handler"); n = read(ngx_eventfd, &ready, 8); err = ngx_errno; ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0, "eventfd: %d", n); if (n != 8) { if (n == -1) { if (err == NGX_EAGAIN) { return; } ngx_log_error(NGX_LOG_ALERT, ev->log, err, "read(eventfd) failed"); return; } ngx_log_error(NGX_LOG_ALERT, ev->log, 0, "read(eventfd) returned only %d bytes", n); return; } ts.tv_sec = 0; ts.tv_nsec = 0; while (ready) { // 获取异步 IO 请求的事件 events = io_getevents(ngx_aio_ctx, 1, 64, event, &ts); ngx_log_debug1(NGX_LOG_DEBUG_EVENT, ev->log, 0, "io_getevents: %l", events); if (events > 0) { ready -= events; for (i = 0; i < events; i++) { ngx_log_debug4(NGX_LOG_DEBUG_EVENT, ev->log, 0, "io_event: %uXL %uXL %L %L", event[i].data, event[i].obj, event[i].res, event[i].res2); e = (ngx_event_t *) (uintptr_t) event[i].data; e->complete = 1; e->active = 0; e->ready = 1; aio = e->data; aio->res = event[i].res; // 将事件添加到事件消息队列中 ngx_post_event(e, &ngx_posted_events); } continue; } if (events == 0) { return; } /* events == -1 */ ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno, "io_getevents() failed"); return; } } // }}}

 

在回调函数中,nginx 通过 ngx_eventfd 获取到已经完成异步 IO 的事件数,然后对相应的事件循环处理

首先调用 io_getevents 获取到事件,然后提交到 ngx_posted_events 队列中进行后续处理

 

在 aio 初始化时,nginx 受限调用 eventfd 创建了一个 eventfd 对象

eventfd 系统调用返回一个与 eventfd 对象关联的文件描述符,然后 nginx 把这个描述符设为非阻塞并添加到 epoll 中,这样当该描述符可读时 epoll_wait 函数就会返回,然后调用 read 函数就可以读取当前完成的 I/O 操作个数

我们通过调用 io_getevents 系统调用就可以获取已完成的 I/O 事件

 

但eventfd描述符什么时候可读呢?这需要在提交 I/O 事件时将 eventfd 与 aio 关联起来 (ngx_file_aio_read 函数中),提交 I/O 事件后如果 I/O 事件已完成系统就会将当前完成的事件个数写入到 eventfd 描述符相关的计数器中并标识 eventfd 可读

 

这样,很巧妙的实现了操作系统内核到用户应用的异步通知机制

 






linux      龙潭书斋      apue      posix      nginx      源码      opensource      sourcecode      开源      glibc      aio      libaio      技术贴      eventfd     


京ICP备15018585号