關于Linux內核中的異步IO的使用
時間：2018-09-21 來源：未知
我們都知道異步IO的作用，就是可以提高我們程序的并發能力，尤其在網絡模型中。在linux中有aio的一系列異步IO的函數接口，但是這類函數都是glibc庫中的函數，是基于多線程實現，不是真正的異步IO，在內核中有真正的異步IO函數接口。下邊我們來學習一下linux內核中的異步IO。
首先我們來看一下內核中異步IO的主要函數接口：
int io_setup(unsigned nr_events, aio_context_t *ctxp);
功能：用來初始化異步IO的上下文。
其中參數ctxp用來描述異步IO上下文， 參數nr_events表示小可處理的異步IO事件的個數。
int io_submit(io_context_t ctx, long nr, struct iocb *iocbs[]);
功能：提交初始化好的異步讀寫事件。
其中ctx是上文的描述句柄， nr表示提交的異步事件個數。Iocbs是異步事件的結構體。
int io_getevents(io_context_t ctx, long nr, struct io_event *events[], struct timespec *timeout);
功能：獲得已經完成的異步IO事件。
其中參數ctx是上下文的句柄，nr 表示期望獲得異步IO事件個數，events用來存放已經完成的異步事件的數據，timeout為超時事件。
int io_destroy(aio_context_t ctx);
功能：用于銷毀異步IO事件句柄。
但是內核的異步IO通常和epoll等IO多路復用配合使用來完成一些異步事件，那么就需要使用epoll來監聽一個可以通知異步IO完成的描述符，那么就需要使用eventfd函數來獲得一個這樣的描述符。
下邊附上一個epoll和內核異步IO配合使用的示例代碼：
#define TEST_FILE "aio_test_file"
#define TEST_FILE_SIZE (127 * 1024)
#define NUM_EVENTS 128
#define ALIGN_SIZE 512
#define RD_WR_SIZE 1024
struct custom_iocb
{
struct iocb iocb;
int nth_request;
};
//異步IO的回調函數
void aio_callback(io_context_t ctx, struct iocb *iocb, long res, long res2)
{
struct custom_iocb *iocbp = (struct custom_iocb *)iocb;
printf("nth_request: %d, request_type: %s, offset: %lld, length: %lu, res: %ld, res2: %ld\n", iocbp->nth_request, (iocb->aio_lio_opcode == IO_CMD_PREAD) ? "READ" : "WRITE",iocb->u.c.offset, iocb->u.c.nbytes, res, res2);
}
int main(int argc, char *argv[])
{
int efd, fd, epfd;
io_context_t ctx;
struct timespec tms;
struct io_event events[NUM_EVENTS];
struct custom_iocb iocbs[NUM_EVENTS];
struct iocb *iocbps[NUM_EVENTS];
struct custom_iocb *iocbp;
int i, j, r;
void *buf;
struct epoll_event epevent;
//創建用于獲取異步事件的通知描述符
efd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
if (efd == -1) {
perror("eventfd");
return 2;
}
fd = open(TEST_FILE, O_RDWR | O_CREAT | O_DIRECT , 0644);
if (fd == -1) {
perror("open");
return 3;
}
ftruncate(fd, TEST_FILE_SIZE);
ctx = 0;
//創建異步IO的句柄
if (io_setup(8192, &ctx)) {
perror("io_setup");
return 4;
}
//申請空間
if (posix_memalign(&buf, ALIGN_SIZE, RD_WR_SIZE)) {
perror("posix_memalign");
return 5;
}
printf("buf: %p\n", buf);
for (i = 0, iocbp = iocbs; i < NUM_EVENTS; ++i, ++iocbp) {
iocbps[i] = &iocbp->iocb;
//設置異步IO讀事件
io_prep_pread(&iocbp->iocb, fd, buf, RD_WR_SIZE, i * RD_WR_SIZE);
//關聯通知描述符
io_set_eventfd(&iocbp->iocb, efd);
//設置回調函數
io_set_callback(&iocbp->iocb, aio_callback);
iocbp->nth_request = i + 1;
}
//提交異步IO事件
if (io_submit(ctx, NUM_EVENTS, iocbps) != NUM_EVENTS) {
perror("io_submit");
return 6;
}
epfd = epoll_create(1);
if (epfd == -1) {
perror("epoll_create");
return 7;
}
epevent.events = EPOLLIN | EPOLLET;
epevent.data.ptr = NULL;
if (epoll_ctl(epfd, EPOLL_CTL_ADD, efd, &epevent)) {
perror("epoll_ctl");
return 8;
}
i = 0;
while (i < NUM_EVENTS) {
uint64_t finished_aio;
//監聽通知描述符
if (epoll_wait(epfd, &epevent, 1, -1) != 1) {
perror("epoll_wait");
return 9;
}
//讀取完成的異步IO事件個數
if (read(efd, &finished_aio, sizeof(finished_aio)) != sizeof(finished_aio)) {
perror("read");
return 10;
}
printf("finished io number: %"PRIu64"\n", finished_aio);
while (finished_aio > 0) {
tms.tv_sec = 0;
tms.tv_nsec = 0;
//獲取完成的異步IO事件
r = io_getevents(ctx, 1, NUM_EVENTS, events, &tms);
if (r > 0) {
for (j = 0; j < r; ++j) {
//調用回調函數
//events[j].data的數據和設置的iocb結構體中的data數據是一致。
((io_callback_t)(events[j].data))(ctx, events[j].obj, events[j].res, events[j].res2);
}
i += r;
finished_aio -= r;
}
}
}
close(epfd);
free(buf);
io_destroy(ctx);
close(fd);
close(efd);
remove(TEST_FILE);
return 0;
}
