9.9.2　ngx_epoll_module模块中实现的针对文件的异步I/O

在Nginx中，文件异步I/O事件完成后的通知是集成到epoll中的，它是通过9.9.1节中介绍的IOCB_FLAG_RESFD标志位完成的。下面看看文件异步I/O事件在ngx_epoll_module模块中是如何实现的，其中在文件异步I/O机制中定义的全局变量如下。

//用于通知异步I/O事件的描述符，它与iocb结构体中的aio_resfd成员是一致的

int ngx_eventfd=-1；

//异步I/O的上下文，全局唯一，必须经过io_setup初始化才能使用

aio_context_t ngx_aio_ctx=0；

/异步I/O事件完成后进行通知的描述符，也就是ngx_eventfd所对应的ngx_event_t事件/

static ngx_event_t ngx_eventfd_event；

/异步I/O事件完成后进行通知的描述符ngx_eventfd所对应的ngx_connection_t连接/

static ngx_connection_t ngx_eventfd_conn；

在9.6.3节的ngx_epoll_init代码中，在epoll_create执行完成后如果开启了文件异步I/O功能，则会调用ngx_epoll_aio_init方法。现在详细描述一下ngx_epoll_aio_init方法中做了些什么，如下所示。

define SYS_eventfd 323

static void ngx_epoll_aio_init（ngx_cycle_tcycle,ngx_epoll_conf_tepcf）

{

int n；

struct epoll_event ee；

//使用Linux中第323个系统调用获取一个描述符句柄

ngx_eventfd=syscall（SYS_eventfd，0）；

……

//设置ngx_eventfd为无阻塞

if（ioctl（ngx_eventfd,FIONBIO，＆n）==-1）{

……

}

//初始化文件异步I/O的上下文

if（io_setup（epcf-＞aio_requests，＆ngx_aio_ctx）==-1）{

……

}

/设置用于异步I/O完成通知的ngx_eventfd_event事件，它与ngx_eventfd_conn连接是对应的/

ngx_eventfd_event.data=＆ngx_eventfd_conn；

//在异步I/O事件完成后，使用ngx_epoll_eventfd_handler方法处理

ngx_eventfd_event.handler=ngx_epoll_eventfd_handler；

ngx_eventfd_event.log=cycle-＞log；

ngx_eventfd_event.active=1；

//初始化ngx_eventfd_conn连接

ngx_eventfd_conn.fd=ngx_eventfd；

//ngx_eventfd_conn连接的读事件就是上面的ngx_eventfd_event

ngx_eventfd_conn.read=＆ngx_eventfd_event；

ngx_eventfd_conn.log=cycle-＞log；

ee.events=EPOLLIN|EPOLLET；

ee.data.ptr=＆ngx_eventfd_conn；

//向epoll中添加到异步I/O的通知描述符ngx_eventfd

if（epoll_ctl（ep,EPOLL_CTL_ADD,ngx_eventfd，＆ee）！=-1）{

return；

}

……

}

这样，ngx_epoll_aio_init方法会把异步I/O与epoll结合起来，当某一个异步I/O事件完成后，ngx_eventfd句柄就处于可用状态，这样epoll_wait在返回ngx_eventfd_event事件后就会调用它的回调方法ngx_epoll_eventfd_handler处理已经完成的异步I/O事件，下面看一下ngx_epoll_eventfd_handler方法主要在做些什么，代码如下所示。

static void ngx_epoll_eventfd_handler（ngx_event_t*ev）

{

int n,events；

uint64_t ready；

ngx_event_t*e；

//一次性最多处理64个事件

struct io_event event[64]；

struct timespec ts；

/获取已经完成的事件数目，并设置到ready中，注意，这个ready是可以大于64的/

n=read（ngx_eventfd，＆ready，8）；

……

//ready表示还未处理的事件。当ready大于0时继续处理

while（ready）{

//调用io_getevents获取已经完成的异步I/O事件

events=io_getevents（ngx_aio_ctx，1，64，event，＆ts）；

if（events＞0）{

//将ready减去已经取出的事件

ready-=events；

//处理event数组里的事件

for（i=0；i＜events；i++）{

//data成员指向这个异步I/O事件对应着的实际事件

e=（ngx_event_t*）（uintptr_t）event[i].data；

……

//将该事件放到ngx_posted_events队列中延后执行

ngx_post_event（e，＆ngx_posted_events）；

}

continue；

}

if（events==0）{

return；

}

return；

}

}

整个网络事件的驱动机制就是这样通过ngx_eventfd通知描述符和ngx_epoll_eventfd_handler回调方法，并与文件异步I/O事件结合起来的。

那么，怎样向异步I/O上下文中提交异步I/O操作呢？看看ngx_linux_aio_read.c文件中的ngx_file_aio_read方法，在打开文件异步I/O后，这个方法将会负责磁盘文件的读取，如下所示。

ssize_t ngx_file_aio_read（ngx_file_tfile,u_charbuf,size_t size,off_t offset,ngx_pool_t*pool）

{

ngx_err_t err；

struct iocb*piocb[1]；

ngx_event_t*ev；

ngx_event_aio_t*aio；

……

aio=file-＞aio；

ev=＆aio-＞event；

……

ngx_memzero（＆aio-＞aiocb,sizeof（struct iocb））；

/设置9.9.1节中介绍过的iocb结构体，这里的aiocb成员就是iocb类型。注意，aio_data已经设置为这个ngx_event_t事件的指针，这样，从io_getevents方法获取的io_event对象中的data也是这个指针/

aio-＞aiocb.aio_data=（uint64_t）（uintptr_t）ev；

aio-＞aiocb.aio_lio_opcode=IOCB_CMD_PREAD；

aio-＞aiocb.aio_fildes=file-＞fd；

aio-＞aiocb.aio_buf=（uint64_t）（uintptr_t）buf；

aio-＞aiocb.aio_nbytes=size；

aio-＞aiocb.aio_offset=offset；

aio-＞aiocb.aio_flags=IOCB_FLAG_RESFD；

aio-＞aiocb.aio_resfd=ngx_eventfd；

/*设置事件的回调方法为ngx_file_aio_event_handler，它的调用关系类似这样：epoll_wait中

调用ngx_epoll_eventfd_handler方法将当前事件放入到ngx_posted_events队列中，在延后执行的队列中

调用ngx_file_aio_event_handler方法*/

ev-＞handler=ngx_file_aio_event_handler；

piocb[0]=＆aio-＞aiocb；

/调用io_submit向ngx_aio_ctx异步I/O上下文中添加1个事件，返回1表示成功/

if（io_submit（ngx_aio_ctx，1，piocb）==1）{

ev-＞active=1；

ev-＞ready=0；

ev-＞complete=0；

return NGX_AGAIN；

}

……

}

下面看一下ngx_event_aio_t结构体的定义。

typedef struct ngx_event_aio_s ngx_event_aio_t；

struct ngx_event_aio_s{

void*data；

//这是真正由业务模块实现的方法，在异步I/O事件完成后被调用

ngx_event_handler_pt handler；

ngx_file_t*file；

ngx_fd_t fd；

if（NGX_HAVE_EVENTFD）

int64_t res；

else

ngx_err_t err；

size_t nbytes；

endif

if（NGX_HAVE_AIO_SENDFILE）

off_t last_offset；

endif

//这里的ngx_aiocb_t就是9.9.1节中介绍的iocb结构体

ngx_aiocb_t aiocb；

ngx_event_t event；

}；

这样，ngx_file_aio_read方法会向异步I/O上下文中添加事件，该epoll_wait在通过ngx_eventfd描述符检测到异步I/O事件后，会再调用ngx_epoll_eventfd_handler方法将io_event事件取出来，放入ngx_posted_events队列中延后执行。ngx_posted_events队列中的事件执行时，则会调用ngx_file_aio_event_handler方法。下面看一下ngx_file_aio_event_handler方法做了些什么，代码如下所示。

static void ngx_file_aio_event_handler（ngx_event_t*ev）

{

ngx_event_aio_t*aio；

aio=ev-＞data；

aio-＞handler（ev）；

}

这里调用了ngx_event_aio_t结构体的handler回调方法，这个回调方法是由真正的业务模块实现的，也就是说，任一个业务模块想使用文件异步I/O，就可以实现handler方法，这样，在文件异步操作完成后，该方法就会被回调。