2021SC@SDUSC Linux内核—原生异步I/O（2）

提交异步 IO 操作

提交异步 IO 操作是通过 io_submit 函数完成的，io_submit 需要提供一个类型为 iocb 结构的数组，表示要进行的异步 IO 操作相关的信息，我们先来看看 iocb 结构的定义：

struct iocb {__u64   aio_data;       // 用户自定义数据, 可用于标识IO操作或者设置回调函数__u16   aio_lio_opcode; // IO操作类型, 如读(IOCB_CMD_PREAD)或者写(IOCB_CMD_PWRITE)操作__s16   aio_reqprio;__u32   aio_fildes;     // 进行IO操作的文件句柄__u64   aio_buf;        // 进行IO操作的缓冲区(如写操作的话就是写到文件的数据)__u64   aio_nbytes;     // 缓冲区的大小__s64   aio_offset;     // IO操作的文件偏移量
};

io_submit 函数最终会调用内核函数 sys_io_submit 来实现提供异步 IO 操作，我们来分析 sys_io_submit 函数的实现：

asmlinkage long
sys_io_submit(aio_context_t ctx_id, long nr, struct iocb __user **iocbpp)
{struct kioctx *ctx;long ret = 0;int i;ctx = lookup_ioctx(ctx_id); // 通过异步IO上下文标识符获取异步IO上下文对象for (i = 0; i < nr; i++) {struct iocb __user *user_iocb;struct iocb tmp;
if (unlikely(__get_user(user_iocb, iocbpp+i))) {ret = -EFAULT;break;}
// 从用户空间复制异步IO操作到内核空间if (unlikely(copy_from_user(&tmp, user_iocb, sizeof(tmp)))) {ret = -EFAULT;break;}
// 调用 io_submit_one 函数提交异步IO操作ret = io_submit_one(ctx, user_iocb, &tmp);if (ret)break;}
put_ioctx(ctx);return i ? i : ret;
}

sys_io_submit 函数的实现比较简单，主要从用户空间复制异步 IO 操作信息到内核空间，然后调用 io_submit_one 函数提交异步 IO 操作。我们重点分析 io_submit_one 函数的实现：

int io_submit_one(struct kioctx *ctx, struct iocb __user *user_iocb,struct iocb *iocb)
{struct kiocb *req;struct file *file;ssize_t ret;char *buf;file = fget(iocb->aio_fildes);      // 通过文件句柄获取文件对象req = aio_get_req(ctx);             // 获取一个异步IO操作对象req->ki_filp = file;                // 要进行异步IO的文件对象req->ki_user_obj = user_iocb;       // 指向用户空间的iocb对象req->ki_user_data = iocb->aio_data; // 设置用户自定义数据req->ki_pos = iocb->aio_offset;     // 设置异步IO操作的文件偏移量
buf = (char *)(unsigned long)iocb->aio_buf; // 要进行异步IO操作的数据缓冲区
// 根据不同的异步IO操作类型来进行不同的处理switch (iocb->aio_lio_opcode) {case IOCB_CMD_PREAD: // 异步读操作ret = -EINVAL;// 发起异步IO操作, 会根据不同的文件系统调用不同的函数:// 如ext3文件系统会调用 generic_file_aio_read 函数if (file->f_op->aio_read)ret = file->f_op->aio_read(req, buf, iocb->aio_nbytes, req->ki_pos);break;}// 异步IO操作或许会在调用 aio_read 时已经完成, 或者会被添加到IO请求队列中。// 所以, 如果异步IO操作被提交到IO请求队列中, 直接返回if (likely(-EIOCBQUEUED == ret)) return 0;
aio_complete(req, ret, 0); // 如果IO操作已经完成, 调用 aio_complete 函数完成收尾工作return 0;
}

上面代码已经对 io_submit_one 函数进行了详细的注释，这里总结一下 io_submit_one 函数主要完成的工作：

通过调用 fget 函数获取文件句柄对应的文件对象。
调用 aio_get_req 函数获取一个类型为 kiocb 结构的异步 IO 操作对象，这个结构前面已经分析过。另外，aio_get_req 函数还会把异步 IO 操作对象添加到异步 IO 上下文的 active_reqs 队列中。
根据不同的异步 IO 操作类型来进行不同的处理，如 异步读操作 会调用文件对象的 aio_read 方法来进行处理。不同的文件系统，其 aio_read 方法的实现不一样，如 Ext3 文件系统的 aio_read 方法会指向 generic_file_aio_read 函数。
如果异步 IO 操作被添加到内核的 IO 请求队列中，那么就直接返回。否则就代表 IO 操作已经完成，那么就调用 aio_complete 函数完成收尾工作。

io_submit_one 函数的操作过程如图5 所示：

所以，io_submit_one 函数的主要任务就是向内核提交 IO 请求。

异步 IO 操作完成

当异步 IO 操作完成后，内核会调用 aio_complete 函数来把处理结果放进异步 IO 上下文的环形缓冲区 ring_info 中，我们来分析一下 aio_complete 函数的实现：

int aio_complete(struct kiocb *iocb, long res, long res2)
{struct kioctx *ctx = iocb->ki_ctx;struct aio_ring_info *info;struct aio_ring *ring;struct io_event *event;unsigned long flags;unsigned long tail;int ret;info = &ctx->ring_info; // 环形缓冲区对象
spin_lock_irqsave(&ctx->ctx_lock, flags);         // 对异步IO上下文进行上锁ring = kmap_atomic(info->ring_pages[0], KM_IRQ1); // 对内存页进行虚拟内存地址映射
tail = info->tail;                           // 环形缓冲区下一个空闲的位置event = aio_ring_event(info, tail, KM_IRQ0); // 从环形缓冲区获取空闲的位置保存结果tail = (tail + 1) % info->nr;                // 更新下一个空闲的位置
// 保存异步IO结果到环形缓冲区中event->obj = (u64)(unsigned long)iocb->ki_user_obj;event->data = iocb->ki_user_data;event->res = res;event->res2 = res2;info->tail = tail;ring->tail = tail; // 更新环形缓冲区下一个空闲的位置
put_aio_ring_event(event, KM_IRQ0); // 解除虚拟内存地址映射kunmap_atomic(ring, KM_IRQ1);       // 解除虚拟内存地址映射
// 释放异步IO对象ret = __aio_put_req(ctx, iocb);spin_unlock_irqrestore(&ctx->ctx_lock, flags);return ret;
}

aio_complete 函数的 iocb 参数是我们通过调用 io_submit_once 函数提交的异步 IO 对象，而参数 res 和 res2 是用内核进行 IO 操作完成后返回的结果。

aio_complete 函数的主要工作如下：

根据环形缓冲区的 tail 指针获取一个空闲的 io_event 对象来保存 IO 操作的结果。
对环形缓冲区的 tail 指针进行加一操作，指向下一个空闲的位置。

当把异步 IO 操作的结果保存到环形缓冲区后，用户层就可以通过调用 io_getevents 函数来读取 IO 操作的结果，io_getevents 函数最终会调用 sys_io_getevents 函数。

我们来分析 sys_io_getevents 函数的实现：

asmlinkage long sys_io_getevents(aio_context_t ctx_id,long min_nr,long nr,struct io_event *events,struct timespec *timeout)
{struct kioctx *ioctx = lookup_ioctx(ctx_id);long ret = -EINVAL;if (likely(NULL != ioctx)) {// 调用 read_events 函数读取IO操作的结果ret = read_events(ioctx, min_nr, nr, events, timeout);put_ioctx(ioctx);}return ret;
}

从上面的代码可以看出，sys_io_getevents 函数主要调用 read_events 函数来读取异步 IO 操作的结果，我们接着分析 read_events 函数：

static int read_events(struct kioctx *ctx,long min_nr, long nr,struct io_event *event,struct timespec *timeout)
{long start_jiffies = jiffies;struct task_struct *tsk = current;DECLARE_WAITQUEUE(wait, tsk);int ret;int i = 0;struct io_event ent;struct timeout to;
memset(&ent, 0, sizeof(ent));ret = 0;
while (likely(i < nr)) {ret = aio_read_evt(ctx, &ent); // 从环形缓冲区中读取一个IO处理结果if (unlikely(ret <= 0))        // 如果环形缓冲区没有IO处理结果, 退出循环break;
ret = -EFAULT;// 把IO处理结果复制到用户空间if (unlikely(copy_to_user(event, &ent, sizeof(ent)))) {break;}
ret = 0;event++;i++;}
if (min_nr <= i)return i;if (ret)return ret;
}

read_events 函数主要还是调用 aio_read_evt 函数来从环形缓冲区中读取异步 IO 操作的结果，如果读取成功，就把结果复制到用户空间中。

aio_read_evt 函数是从环形缓冲区中读取异步 IO 操作的结果，其实现如下：

static int aio_read_evt(struct kioctx *ioctx, struct io_event *ent)
{struct aio_ring_info *info = &ioctx->ring_info;struct aio_ring *ring;unsigned long head;int ret = 0;
ring = kmap_atomic(info->ring_pages[0], KM_USER0);
// 如果环形缓冲区的head指针与tail指针相等, 代表环形缓冲区为空, 所以直接返回if (ring->head == ring->tail) goto out;
spin_lock(&info->ring_lock);
head = ring->head % info->nr;if (head != ring->tail) {// 根据环形缓冲区的head指针从环形缓冲区中读取结果struct io_event *evp = aio_ring_event(info, head, KM_USER1);
*ent = *evp;                  // 将结果保存到ent参数中head = (head + 1) % info->nr; // 移动环形缓冲区的head指针到下一个位置ring->head = head;            // 保存环形缓冲区的head指针ret = 1;put_aio_ring_event(evp, KM_USER1);}
spin_unlock(&info->ring_lock);

out:kunmap_atomic(ring, KM_USER0);return ret;
}

aio_read_evt 函数的主要工作就是判断环形缓冲区是否为空，如果不为空就从环形缓冲区中读取异步 IO 操作的结果，并且保存到参数 ent 中，并且移动环形缓冲区的 head 指针到下一个位置。