io_uring源码分析
io_uring源码分析
io_uring源码分析
基础
io_uring 就是Linux的系统调用接口，于 2019 年在上游 Linux 内核版本 5.1 中首次引入，它使应用程序可以异步执行的系统调用。虽然相比于 Linux 原有的 Native AIO 而言似乎并没有太多的性能提升，但其仍是一项革命性的新技术。
提交任务的环叫SQ，里面的每个任务叫SQE，获取结果的环叫CQ，里面的每个结果叫CQE。
liburing库安装
自动安装
sudo apt update
sudo apt install liburing-dev手动安装
sudo apt update
sudo apt upgrade
git clone https://github.com/axboe/liburing.git
cd liburing
./configure
make
sudo make install
sudo ldconfig #更新库缓存
ldconfig -p | grep liburing #检查是否安装好liburing库函数使用
1. 核心结构体
struct io_uring
- 表示一个
io_uring实例，包含提交队列（SQ）和完成队列（CQ）的内存映射信息。 - 通过
io_uring_queue_init()初始化。
2. 初始化与销毁
io_uring_queue_init(entries, ring, flags)
- 作用：初始化
io_uring实例，分配队列内存并映射到用户态。 - 参数：
entries：队列大小（必须是 2 的幂，如 1024）。ring：指向struct io_uring的指针。flags：配置标志（如IORING_SETUP_SQPOLL启用内核轮询模式）。
- 返回值：成功返回
0，失败返回负的错误码。
io_uring_queue_exit(ring)
- 作用：释放
io_uring实例资源，解除内存映射，关闭文件描述符。 - 参数：
ring指向已初始化的io_uring实例。
示例：
#include <liburing.h>
struct io_uring ring;
int ret = io_uring_queue_init(1024, &ring, 0);
if (ret < 0) {
perror("io_uring_queue_init");
exit(1);
}
// ... 使用 io_uring ...
io_uring_queue_exit(&ring);3. 提交队列操作
io_uring_get_sqe(ring)
- 作用：从提交队列（SQ）中获取一个空闲的 提交队列条目（SQE）。
- 返回值：指向
struct io_uring_sqe的指针，失败返回NULL。
io_uring_submit(ring)
- 作用：将提交队列中的请求提交到内核，触发异步处理。
- 返回值：成功提交的请求数量，失败返回负的错误码。
示例（提交一个读请求）：
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
if (!sqe) {
fprintf(stderr, "Failed to get SQE\n");
exit(1);
}
int fd = open("test.txt", O_RDONLY);
char buffer[4096];
struct iovec iov = { .iov_base = buffer, .iov_len = sizeof(buffer) };
// 准备读请求
io_uring_prep_readv(sqe, fd, &iov, 1, 0); // 从偏移量 0 读取
sqe->user_data = (uintptr_t)fd; // 可选：关联用户数据
// 提交请求
int submitted = io_uring_submit(&ring);
if (submitted < 0) {
perror("io_uring_submit");
}4. 完成队列操作
io_uring_wait_cqe(ring, cqe_ptr)
- 作用：阻塞等待至少一个完成事件（CQE）。
- 参数：
cqe_ptr是struct io_uring_cqe **类型，用于接收 CQE 指针。 - 返回值：成功返回
0，失败返回负的错误码。
io_uring_peek_cqe(ring, cqe_ptr)
- 作用：非阻塞检查是否有完成的 CQE。
- 返回值：有事件返回
0，无事件返回-EAGAIN。
io_uring_cqe_seen(ring, cqe)
- 作用：标记一个 CQE 已被处理，推进完成队列指针。
示例（处理完成事件）：
struct io_uring_cqe *cqe;
int ret = io_uring_wait_cqe(&ring, &cqe);
if (ret < 0) {
perror("io_uring_wait_cqe");
exit(1);
}
if (cqe->res < 0) {
fprintf(stderr, "Read error: %s\n", strerror(cqe->res));
} else {
printf("Read %d bytes\n", cqe->res);
}
io_uring_cqe_seen(&ring, cqe); // 标记处理完成5. 准备 I/O 请求的辅助函数
liburing 提供了一系列函数用于快速配置 SQE 的常见操作。
读/写操作
io_uring_prep_readv(sqe, fd, iov, iovcnt, offset)
准备一个分散读请求（readv）。io_uring_prep_writev(sqe, fd, iov, iovcnt, offset)
准备一个集中写请求（writev）。
文件操作
io_uring_prep_openat(sqe, dfd, path, flags, mode)
异步打开文件。io_uring_prep_close(sqe, fd)
异步关闭文件描述符。
网络操作
io_uring_prep_accept(sqe, fd, addr, addrlen, flags)
异步接受连接。io_uring_prep_send(sqe, sockfd, buf, len, flags)
异步发送数据。io_uring_prep_recv(sqe, sockfd, buf, len, flags)
异步接收数据。
超时与事件
io_uring_prep_timeout(sqe, ts, count, flags)
设置超时事件。io_uring_prep_poll_add(sqe, fd, poll_mask)
添加文件描述符的轮询事件。
示例（异步接受连接）：
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
if (!sqe) {
fprintf(stderr, "Failed to get SQE\n");
exit(1);
}
int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
// ... bind 和 listen 操作 ...
struct sockaddr_in client_addr;
socklen_t addrlen = sizeof(client_addr);
io_uring_prep_accept(sqe, listen_fd, (struct sockaddr*)&client_addr, &addrlen, 0);
sqe->user_data = 123; // 标识此请求
io_uring_submit(&ring);6. 高级功能
注册资源（减少开销）
- 文件描述符注册：
int files[] = {fd1, fd2};
io_uring_register(ring.ring_fd, IORING_REGISTER_FILES, files, 2);之后可通过 IOSQE_FIXED_FILE 标志使用注册的 fd，避免内核检查开销。
- 缓冲区注册：
struct iovec iov = {buffer, sizeof(buffer)};
io_uring_register(ring.ring_fd, IORING_REGISTER_BUFFERS, &iov, 1);内核预注册缓冲区，提升零拷贝性能。
7. 完整示例：异步读取文件
#include <liburing.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define BUF_SIZE 4096
int main() {
struct io_uring ring;
int ret = io_uring_queue_init(8, &ring, 0);
if (ret < 0) {
perror("io_uring_queue_init");
return 1;
}
int fd = open("test.txt", O_RDONLY);
if (fd < 0) {
perror("open");
return 1;
}
char *buffer = malloc(BUF_SIZE);
struct iovec iov = { .iov_base = buffer, .iov_len = BUF_SIZE };
// 准备读请求
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_readv(sqe, fd, &iov, 1, 0);
sqe->user_data = (uintptr_t)fd;
// 提交请求
ret = io_uring_submit(&ring);
if (ret < 0) {
perror("io_uring_submit");
return 1;
}
// 等待完成
struct io_uring_cqe *cqe;
ret = io_uring_wait_cqe(&ring, &cqe);
if (ret < 0) {
perror("io_uring_wait_cqe");
return 1;
}
if (cqe->res < 0) {
fprintf(stderr, "Read failed: %s\n", strerror(-cqe->res));
} else {
printf("Read %d bytes:\n%.*s\n", cqe->res, cqe->res, buffer);
}
io_uring_cqe_seen(&ring, cqe);
io_uring_queue_exit(&ring);
close(fd);
free(buffer);
return 0;
}io_uring的系统调用:
int io_uring_setup(u32 entries, struct io_uring_params *p){
return syscall(__NR_io_uring_setup, entries, p);
}
int io_uring_enter(int fd, uint32_t to_submit, uint32_t min_complete, uint32_t flags, sigset_t *sig){
return syscall(__NR_io_uring_enter, fd, to_submit, min_complete, flags, sig, _NSIG / 8);
}
int io_uring_register(int fd, unsigned int opcode, const void *arg, unsigned int nr_args){
return syscall(__NR_io_uring_register, fd, opcode, arg, nr_args);
}io_uring三个系统调用的作用：
io_uring_setup():返回文件描述符给用户操作，并注册io_uring的上下文(ctx)。ctx在这里主要是创建CQ,SQ队列。
io_uring_enter():提交新的IO请求，可以选择等待IO完成请求。
io_uring_register():主要用来注册\注销\更新 缓冲区，文件，个性化等。
先看一下前面提到的ctx和一些io_uring用的结构体。
io_ring_ctx结构体太大这里先不展示，这个结构体之后会被经常提到，就是我们常说的上下文。他包含了struct io_rings *rings;。
struct io_rings {
/*
* 提交队列(sq)和完成队列(cq)的头尾指针
* 需要用掩码来获取有效的索引
*
* 内核控制sq的head和cq的tail
* 应用程序控制sq的tail和cq的head
*/
struct io_uring sq, cq;
/*
* 用于head和tail偏移量的位掩码
* 常量,等于ring_entries - 1
*/
u32 sq_ring_mask, cq_ring_mask;
/*
* 环形缓冲区大小(常量,2的幂)
*/
u32 sq_ring_entries, cq_ring_entries;
/*
* 由于数组中存储的索引无效而被内核丢弃的无效条目数
*
* 由内核写入,应用程序不应修改
* (即通过与缓存值比较来获取"新事件"的数量)
*
* 当应用程序读取新的SQ head值后,此计数器包括
* 到达新SQ head时丢弃的所有提交(可能更多)
*/
u32 sq_dropped;
/*
* 运行时SQ标志
*
* 由内核写入,应用程序不应修改
*
* 应用程序在更新sq tail后,需要完整的内存屏障
* 才能检查IORING_SQ_NEED_WAKEUP标志
*/
u32 sq_flags;
/*
* 运行时CQ标志
*
* 由应用程序写入,内核不应修改
*/
u32 cq_flags;
/*
* 由于队列已满而丢失的完成事件数量
* 应用程序应通过确保待处理请求数不超过完成队列空间来避免这种情况
*
* 由内核写入,应用程序不应修改
* (即通过与缓存值比较来获取"新事件"的数量)
*
* 由于完成事件是无序到达的,此计数器与其他数据无关
*/
u32 cq_overflow;
/*
* 完成事件的环形缓冲区
*
* 内核每次产生完成事件时都会重新写入,
* 因此应用程序可以修改待处理的条目
*/
struct io_uring_cqe cqes[] ____cacheline_aligned_in_smp;
};CQ、SQ为struct io_uring，就是一个head、一个tail。
struct io_uring {
u32 head ____cacheline_aligned_in_smp;
u32 tail ____cacheline_aligned_in_smp;
};io_uring_setup系统调用
几个关键的函数如下图
该函数完成以下工作，可见核心在于调用了io_uring_create函数，继续跟进io_uring_create()。
1、将用户空间的参数拷贝到内核空间
2、检查保留字段是否为0
3、检查flags的合法性
4、调用io_uring_create()创建io_uring上下文，并返回文件描述符。
/*
* @entries: 请求的提交队列(SQ)和完成队列(CQ)的大小
* @params: 用户空间传入的io_uring参数结构体指针
*/
static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
{
struct io_uring_params p;
int i;
/* 将用户空间的io_uring_params结构体拷贝到内核空间 */
if (copy_from_user(&p, params, sizeof(p)))
return -EFAULT;
/* 检查保留字段数组是否全部为0,防止未来扩展时出现兼容性问题 */
for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
if (p.resv[i])
return -EINVAL;
}
/*
* 验证flags是否合法,只允许以下标志位:
* IORING_SETUP_IOPOLL: 启用io polling模式
* IORING_SETUP_SQPOLL: 启用SQ polling线程
* IORING_SETUP_SQ_AFF: 设置SQ线程CPU亲和性
* IORING_SETUP_CQSIZE: 允许单独设置CQ大小
* IORING_SETUP_CLAMP: 限制ring buffer大小
* IORING_SETUP_ATTACH_WQ: 附加到已存在的io_wq
* IORING_SETUP_R_DISABLED: 禁用ring
*/
if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
IORING_SETUP_SQ_AFF | IORING_SETUP_CQSIZE |
IORING_SETUP_CLAMP | IORING_SETUP_ATTACH_WQ |
IORING_SETUP_R_DISABLED))
return -EINVAL;
/* 创建和初始化io_uring上下文,返回文件描述符 */
return io_uring_create(entries, &p, params);
}io_uring_create函数
该函数完成以下工作，分配上下文调用io_ring_ctx_alloc(),在SQ\CQ环申请空间时使用io_allocate_scq_urings()。
1、检查和调整SQ/CQ环的大小。
2、默认CQsize为SQ的两倍，但是如果设置IORING_SETUP_CQSIZE，可自定义CQsize(需要大于SQsize)。
3、分配并初始化**上下文(ctx)**空间(结构体io_ring_ctx)
4、分配和初始化SQ/CQ环。
5、获取文件描述符并安装。
static int io_uring_create(unsigned entries, struct io_uring_params *p,
struct io_uring_params __user *params)
{
struct user_struct *user = NULL;
struct io_ring_ctx *ctx;
struct file *file;
bool limit_mem;
int ret;
/* 验证entries参数 */
//大小不能为0
if (!entries)
return -EINVAL;
//大小最大为32768(0x8000)
if (entries > IORING_MAX_ENTRIES) {
if (!(p->flags & IORING_SETUP_CLAMP))
return -EINVAL;
entries = IORING_MAX_ENTRIES;
}
//向上舍入到最近的 2 的幂次方
p->sq_entries = roundup_pow_of_two(entries);
/* 处理CQ环大小 */
//设置CQsize
if (p->flags & IORING_SETUP_CQSIZE) {
/*
* If IORING_SETUP_CQSIZE is set, we do the same roundup
* to a power-of-two, if it isn't already. We do NOT impose
* any cq vs sq ring sizing.
*/
//大小首先检查为0和最大值
if (!p->cq_entries)
return -EINVAL;
if (p->cq_entries > IORING_MAX_CQ_ENTRIES) {
if (!(p->flags & IORING_SETUP_CLAMP))
return -EINVAL;
p->cq_entries = IORING_MAX_CQ_ENTRIES;
}
//如果使用CQSIZE直接设置2的幂次方
p->cq_entries = roundup_pow_of_two(p->cq_entries);
//完成环需要大于提交环（CQsize > SQsize）
if (p->cq_entries < p->sq_entries)
return -EINVAL;
} else {
/* 默认CQ环大小为SQ环的2倍 */
p->cq_entries = 2 * p->sq_entries;
}
/* 获取当前用户信息 */
user = get_uid(current_user());
/* 检查是否有CAP_IPC_LOCK权限 ，允许对共享内存段进行锁定 */
limit_mem = !capable(CAP_IPC_LOCK);
//如果没有锁定权限，增加当前可以锁定的内存
if (limit_mem) {
ret = __io_account_mem(user,
ring_pages(p->sq_entries, p->cq_entries));
if (ret) {
free_uid(user);
return ret;
}
}
/* 分配和初始化io_ring上下文 */
//分配ctx空间
ctx = io_ring_ctx_alloc(p);
if (!ctx) {
if (limit_mem)
__io_unaccount_mem(user, ring_pages(p->sq_entries,
p->cq_entries));
free_uid(user);
return -ENOMEM;
}
/* 设置上下文的各种属性 */
//...
/* 保存当前进程的mm_struct用于内存记账 */
mmgrab(current->mm);
ctx->mm_account = current->mm;
/* 对ring内存进行记账 */
io_account_mem(ctx, ring_pages(p->sq_entries, p->cq_entries),
ACCT_LOCKED);
ctx->limit_mem = limit_mem;
/* 分配和初始化SQ/CQ环 */
// 分配 io_rings
// 直接分配页 __get_free_pages()
ret = io_allocate_scq_urings(ctx, p);
if (ret)
goto err;
//初始化工作队列
ret = io_sq_offload_create(ctx, p);
if (ret)
goto err;
//尝试唤醒线程启动 iopoll
if (!(p->flags & IORING_SETUP_R_DISABLED))
io_sq_offload_start(ctx);
/* 设置SQ、CQ环的偏移量 */
//...
/* 设置io_uring支持的特性 */
p->features = IORING_FEAT_SINGLE_MMAP | IORING_FEAT_NODROP |
IORING_FEAT_SUBMIT_STABLE | IORING_FEAT_RW_CUR_POS |
IORING_FEAT_CUR_PERSONALITY | IORING_FEAT_FAST_POLL |
IORING_FEAT_POLL_32BITS | IORING_FEAT_SQPOLL_NONFIXED |
IORING_FEAT_EXT_ARG;
//拷贝上下文回用户态
if (copy_to_user(params, p, sizeof(*p))) {
ret = -EFAULT;
goto err;
}
//获取文件描述符inode
file = io_uring_get_file(ctx);
if (IS_ERR(file)) {
ret = PTR_ERR(file);
goto err;
}
/*
* Install ring fd as the very last thing, so we don't risk someone
* having closed it before we finish setup
*/
//绑定fd
ret = io_uring_install_fd(ctx, file);
if (ret < 0) {
io_disable_sqo_submit(ctx);
/* fput will clean it up */
fput(file);
return ret;
}
//日志记录
trace_io_uring_create(ret, ctx, p->sq_entries, p->cq_entries, p->flags);
return ret;
err:
io_disable_sqo_submit(ctx);
io_ring_ctx_wait_and_kill(ctx);
return ret;
}主要函数
1、io_ring_ctx_alloc()分配上下文空间，初始化列表头、旋转锁等
2、io_allocate_scq_urings()分配与用户空间共享的 rings 本体(结构体 struct io_rings )，初始化CQ\SQ环，分配SQE数组的空间。
3、sqd->thread关于内核线程，io_sq_offload_create()用来初始化工作队列，io_sq_offload_start()唤醒工作线程。
4、io_uring_get_file()获取文件描述符，将struct io_ring_ctx *ctx 暴露给用户态使用。
io_ring_ctx_alloc分配上下文空间
分配走的都是常规的分配路径GFP_KERNEL。
static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
{
struct io_ring_ctx *ctx;
int hash_bits;
//分配上下文空间
ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
if (!ctx)
return NULL;
/* 分配fallback请求缓存,用于异常情况 */
ctx->fallback_req = kmem_cache_alloc(req_cachep, GFP_KERNEL);
if (!ctx->fallback_req)
goto err;
//初始化
return ctx;
err:
//...
return NULL;
}io_allocate_scq_urings分配SQ\CQ环
先获取rings大小然后调用io_mem_alloc(),之后会调用__get_free_pages()，直接分配页。因为 io_rings 结构体要与用户进程通过 mmap() 直接进行共享。
static int io_allocate_scq_urings(struct io_ring_ctx *ctx,
struct io_uring_params *p)
{
struct io_rings *rings;
size_t size, sq_array_offset;
/* make sure these are sane, as we already accounted them */
ctx->sq_entries = p->sq_entries;
ctx->cq_entries = p->cq_entries;
//获取rings大小
size = rings_size(p->sq_entries, p->cq_entries, &sq_array_offset);
if (size == SIZE_MAX)
return -EOVERFLOW;
/* 与用户空间共享的 rings 本体 */
//调用 __get_free_pages
rings = io_mem_alloc(size);
//...
//上下文设置，SQsize设置
//...
ctx->sq_sqes = io_mem_alloc(size);
//...
return 0;
}
static void *io_mem_alloc(size_t size)
{
gfp_t gfp_flags = GFP_KERNEL | __GFP_ZERO | __GFP_NOWARN | __GFP_COMP |
__GFP_NORETRY;
return (void *) __get_free_pages(gfp_flags, get_order(size));
}io_sq_offload_create初始化工作队列
「1」 主要工作进行的就是SQ轮询线程的初始化，在函数最开始就判断ctx->flags是否开启SQ轮询模式，如果没有就跳过一大段线程的初始化工作，否则初始化kernel thread的初始化。
「2」 初始化SQ轮询线程会先检查权限是否足够(需要SYS_ADMIN或SYS_NICE),然后将上下文的SQ数据链添加到sqd的新列表中。然后进行线程超时设置(默认为1秒)，设置线程是否需要指定CPU，最后为线程的task_struct分配内存以及初始化。
「3」 最后调用io_init_wq_offload()进行工作队列的初始化，主要就是ctx->io_wq队列的一些初始化。
/* fs/io_uring.c
* 创建和初始化io_uring的SQ轮询线程
* @ctx: io_uring上下文
* @p: 内核态io_uring参数
*/
static int io_sq_offload_create(struct io_ring_ctx *ctx,
struct io_uring_params *p)
{
int ret;
/* 检查是否开启了SQ轮询模式 */
if (ctx->flags & IORING_SETUP_SQPOLL) {
struct io_sq_data *sqd;
/* 检查权限 - 需要SYS_ADMIN或SYS_NICE权限 */
ret = -EPERM;
if (!capable(CAP_SYS_ADMIN) && !capable(CAP_SYS_NICE))
goto err;
/* 获取io_sq_data */
sqd = io_get_sq_data(p);
if (IS_ERR(sqd)) {
ret = PTR_ERR(sqd);
goto err;
}
ctx->sq_data = sqd;
//暂停 SQ线程
io_sq_thread_park(sqd);
mutex_lock(&sqd->ctx_lock);
//添加新的上下文到队列
list_add(&ctx->sqd_list, &sqd->ctx_new_list);
mutex_unlock(&sqd->ctx_lock);
//恢复 SQ线程
io_sq_thread_unpark(sqd);
/* 设置SQ线程空闲超时时间,默认为1秒(HZ) */
ctx->sq_thread_idle = msecs_to_jiffies(p->sq_thread_idle);
if (!ctx->sq_thread_idle)
ctx->sq_thread_idle = HZ;
//如果线程存在直接跳到done
if (sqd->thread)
goto done;
/* 是否需要指定cpu */
if (p->flags & IORING_SETUP_SQ_AFF) {
int cpu = p->sq_thread_cpu;
ret = -EINVAL;
if (cpu >= nr_cpu_ids)
goto err;
if (!cpu_online(cpu))
goto err;
//开启内核线程在指定cpu上
sqd->thread = kthread_create_on_cpu(io_sq_thread, sqd,
cpu, "io_uring-sq");
} else {
sqd->thread = kthread_create(io_sq_thread, sqd,
"io_uring-sq");
}
if (IS_ERR(sqd->thread)) {
ret = PTR_ERR(sqd->thread);
sqd->thread = NULL;
goto err;
}
// 为SQ线程[sqd->thread(struct task_struct*)]分配io_uring 任务上下文
ret = io_uring_alloc_task_context(sqd->thread);
if (ret)
goto err;
} else if (p->flags & IORING_SETUP_SQ_AFF) {
// 不能在没有SQPOLL的情况下设置CPU
ret = -EINVAL;
goto err;
}
done:
//初始化工作队列
ret = io_init_wq_offload(ctx, p);
if (ret)
goto err;
return 0;
err:
//清理异步资源
io_finish_async(ctx);
return ret;
}io_sq_offload_start线程启动
「1」之前通过io_sq_offload_create()以及完成线程的初始化工作，在这里唤醒线程即可。在最开始的源码中并没有分为create、start两步直接写在一起的，不过大体都一致。
/* fs/io_uring.c */
static void io_sq_offload_start(struct io_ring_ctx *ctx)
{
struct io_sq_data *sqd = ctx->sq_data;
if ((ctx->flags & IORING_SETUP_SQPOLL) && sqd->thread)
wake_up_process(sqd->thread);
}io_uring_get_file获取暴露文件结构体
「1」io_uring_get_file就是获取了文件结构体，之后会会调用io_uring_install_fd()安装文件描述符之后返回给用户，这样就把struct io_ring_ctx *ctx 暴露给用户态使用。
「2」值得注意的是调用了sock_craeate_kern()来创建内核的UNIX套接字，这就表明io_uring其实是sock类型的，这点在linux中飞行计数很重要，之后复现CVE的时候也会重点分析。然后通过anon_inode_getfile()创建一个匿名 inode并返回ctx的文件结构体。
/* fs/io_uring.c */
static struct file *io_uring_get_file(struct io_ring_ctx *ctx)
{
struct file *file;
#if defined(CONFIG_UNIX)
int ret;
ret = sock_create_kern(&init_net, PF_UNIX, SOCK_RAW, IPPROTO_IP,
&ctx->ring_sock);
if (ret)
return ERR_PTR(ret);
#endif
file = anon_inode_getfile("[io_uring]", &io_uring_fops, ctx,
O_RDWR | O_CLOEXEC);
#if defined(CONFIG_UNIX)
if (IS_ERR(file)) {
sock_release(ctx->ring_sock);
ctx->ring_sock = NULL;
} else {
ctx->ring_sock->file = file;
}
#endif
return file;
}io_uring_enter系统调用
该函数完成以下工作，主要是管理用户 空间对异步 I/O 操作的提交和完成事件的处理。
1、flags检查、检查文件描述符。
2、根据ctx判断是否需要SQ轮询，否则采用to_submit数量提交。
3、设置IORING_ENTER_GETEVENTS则处理完成事件，
- 如果启用了
IOPOLL但未启用SQPOLL模式，则调用io_iopoll_check进行轮询检查。 - 否则，调用
io_cqring_wait等待完成事件，尝试加入等待队列并唤醒执行。
4、释放资源后返回SQE提交数量。
/*
* @fd: io_uring实例的文件描述符
* @to_submit: 要提交的SQE数量
* @min_complete: 最少需要完成的事件数量
* @flags: 控制标志
* @argp: 用户空间参数指针
* @argsz: 参数大小
*/
SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
u32, min_complete, u32, flags, const void __user *, argp,
size_t, argsz)
{
struct io_ring_ctx *ctx;
long ret = -EBADF;
int submitted = 0; // 已提交的SQE数量
struct fd f;
io_run_task_work(); // 执行待处理的任务工作
if (flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP |
IORING_ENTER_SQ_WAIT | IORING_ENTER_EXT_ARG))
return -EINVAL;
f = fdget(fd);
if (!f.file)
return -EBADF;
// 检查文件是否为io_uring操作
ret = -EOPNOTSUPP;
if (f.file->f_op != &io_uring_fops)
goto out_fput;
ret = -ENXIO;
//获取上下文
ctx = f.file->private_data;
//如果引用计数为0，则退出
if (!percpu_ref_tryget(&ctx->refs))
goto out_fput;
// 检查io_uring是否被禁用
ret = -EBADFD;
if (ctx->flags & IORING_SETUP_R_DISABLED)
goto out;
ret = 0;
//SQ轮询
if (ctx->flags & IORING_SETUP_SQPOLL) {
// 刷新CQ溢出的事件
io_cqring_overflow_flush(ctx, false, NULL, NULL);
// 检查SQ线程是否已死亡
ret = -EOWNERDEAD;
if (unlikely(ctx->sqo_dead))
goto out;
//唤醒等待队列
if (flags & IORING_ENTER_SQ_WAKEUP)
wake_up(&ctx->sq_data->wait);
//启动SQ_WAIT模式，尝试把当前进程添加到wait队列
if (flags & IORING_ENTER_SQ_WAIT) {
ret = io_sqpoll_wait_sq(ctx);
if (ret)
goto out;
}
submitted = to_submit;
} else if (to_submit) {
/* 手动提交任务 */
// 将当前文件添加到任务文件列表
ret = io_uring_add_task_file(ctx, f.file);
if (unlikely(ret))
goto out;
mutex_lock(&ctx->uring_lock);
//提交SQE
submitted = io_submit_sqes(ctx, to_submit);
mutex_unlock(&ctx->uring_lock);
// 检查提交数量是否符合预期
if (submitted != to_submit)
goto out;
}
//完成个数
if (flags & IORING_ENTER_GETEVENTS) {
const sigset_t __user *sig; // 信号掩码
struct __kernel_timespec __user *ts;// 超时时间
// 获取扩展参数
ret = io_get_ext_arg(flags, argp, &argsz, &ts, &sig);
if (unlikely(ret))
goto out;
// 最小完成数不超过CQ容量
min_complete = min(min_complete, ctx->cq_entries);
// 根据配置选择轮询方式
if (ctx->flags & IORING_SETUP_IOPOLL &&
!(ctx->flags & IORING_SETUP_SQPOLL)) {
//调用核心IOpoll函数，进行轮询检查是否完成
ret = io_iopoll_check(ctx, min_complete);
} else {
//先尝试执行任务，失败后进行加入等待队列，并尝试唤醒执行
ret = io_cqring_wait(ctx, min_complete, sig, argsz, ts);
}
}
out:
// 减少引用计数
percpu_ref_put(&ctx->refs);
out_fput:
fdput(f);
return submitted ? submitted : ret;
}主要函数
io_sqpoll_wait_sq(): 尝试把当前进程添加到ctx->sqo_sq_wait队列中，然后挂起当前进程，等待提交队列(SQ)有空闲空间。
io_submit_sqes():提交SQES到io_uring。
io_iopoll_check()：会调用核心IOpoll函数，进行轮询检查，判断是否有事件完成。
io_cqring_wait()：会一直等待完成队列(CQ)中有足够的事件，足够条件取决于提交的最小数量和CQ队列大小。
io_sqpoll_wait_sq等待SQ有空间
「1」 io_sqpoll_wait_sq() 作用就是在高并发的时候，SQ队列处于满状态，就不准在加入SQ了，需要挂起当前进程，阻塞等待SQ有空闲空间为止。
「2」通过io_sqring_full()函数判断SQring是否为满状态，如果不满则直接退出循环，否则会把当前进程添加到ctx->sqo_sq_wait的等待队列中，然后进行进程调度，直到SQring有空闲、收到信号、SQ线程死亡才会返回。
/* fs/io_uring.c */
static int io_sqpoll_wait_sq(struct io_ring_ctx *ctx)
{
int ret = 0;
DEFINE_WAIT(wait);
do {
//检查SQ是否为满，如果不满则退出循环
if (!io_sqring_full(ctx))
break;
//当前进程添加到等待队列
prepare_to_wait(&ctx->sqo_sq_wait, &wait, TASK_INTERRUPTIBLE);
//检查SQ线程是否死亡
if (unlikely(ctx->sqo_dead)) {
ret = -EOWNERDEAD;
goto out;
}
//再次检查SQ是否为满
if (!io_sqring_full(ctx))
break;
//进程调度
schedule();
/* 循环直到收到信号 */
} while (!signal_pending(current));
// 删除当前进程从等待队列
finish_wait(&ctx->sqo_sq_wait, &wait);
out:
return ret;
}io_submit_sqes提交SQES
「1」 该函数会先检查CQ溢出风险，然后需要确定提交多少个事件，多少取决于用户提交数量、SQ事件数量、SQ队列长度。三者取最小，之后增加计数引用，这里会增加当前任务的inflight计数和usage计数，这里的 inflight 就是之前说的飞行计数，这个特性之后单独分析，暂时不展开叙述。
「2」 循环处理每个SQE，先通过上下文获取SQE,然后对io请求(req)进行初始化，调用io_submit_sqe()进行单个SQE请求的提交，之后单个SQE转换为内核可处理的I/O请求，并触发内核的异步处理流程完成事件。
/* fs/io_uring.c */
static int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr)
{
struct io_submit_state state;
struct io_submit_link link;
int i, submitted = 0;
/* 如果有未处理的CQ溢出,返回BUSY */
if (test_bit(0, &ctx->sq_check_overflow)) {
if (!__io_cqring_overflow_flush(ctx, false, NULL, NULL))
return -EBUSY;
}
// 确保SQ条目数量合法,取三者最小值
nr = min3(nr, ctx->sq_entries, io_sqring_entries(ctx));
// 尝试增加引用计数
if (!percpu_ref_tryget_many(&ctx->refs, nr))
return -EAGAIN;
// 增加当前任务的inflight计数和usage计数
percpu_counter_add(¤t->io_uring->inflight, nr);
refcount_add(nr, ¤t->usage);
//初始化IO提交状态结构体
io_submit_state_start(&state, ctx, nr);
link.head = NULL;
//循环循环处理每个SQE
for (i = 0; i < nr; i++) {
const struct io_uring_sqe *sqe;
struct io_kiocb *req;
int err;
//获取一个SQE
sqe = io_get_sqe(ctx);
if (unlikely(!sqe)) {
io_consume_sqe(ctx);
break;
}
//为请求分配内存
req = io_alloc_req(ctx, &state);
if (unlikely(!req)) {
if (!submitted)
submitted = -EAGAIN;
break;
}
//增加cached_sq_head
io_consume_sqe(ctx);//ctx->cached_sq_head++;
submitted++;
//初始化请求结构
err = io_init_req(ctx, req, sqe, &state);
if (unlikely(err)) {
fail_req:
io_put_req(req);
io_req_complete(req, err);
break;
}
trace_io_uring_submit_sqe(ctx, req->opcode, req->user_data,
true, io_async_submit(ctx));
//提交SQE请求
err = io_submit_sqe(req, sqe, &link, &state.comp);
if (err)
goto fail_req;
}
//处理提交失败的情况
if (unlikely(submitted != nr)) {
int ref_used = (submitted == -EAGAIN) ? 0 : submitted;
struct io_uring_task *tctx = current->io_uring;
int unused = nr - ref_used;
//减少未使用的引用计数
percpu_ref_put_many(&ctx->refs, unused);
percpu_counter_sub(&tctx->inflight, unused);
put_task_struct_many(current, unused);
}
//处理链接的请求
if (link.head)
io_queue_link_head(link.head, &state.comp);
//结束提交状态
io_submit_state_end(&state);
/* 提交完成后更新SQ ring的head */
io_commit_sqring(ctx);
return submitted;
}io_iopoll_check轮询事件完成
「1」 如果设置了IORING_SETUP_IOPOLL并且没开启SQ轮询的时候才会调用io_iopoll_check，该函数主要用来判断是否有事件完成。
「2」 通过循环一直调用核心IO轮询函数io_iopoll_getevents()，这个函数作用就是查看完成队列是否有事件完成，会对传入参数nr_events进行累加，在循环中每8次就启动一次任务，尝试完成I/O请求。只有当有事件完成或者需要进程调度的时候才会退出。
/* fs/io_uring.c */
static int io_iopoll_check(struct io_ring_ctx *ctx, long min)
{
unsigned int nr_events = 0;//已经处理事件的数量
int iters = 0, ret = 0;
mutex_lock(&ctx->uring_lock);
do {
//防止CQ溢出
if (test_bit(0, &ctx->cq_check_overflow))
__io_cqring_overflow_flush(ctx, false, NULL, NULL);
//如果完成队列有事件，直接退出循环
if (io_cqring_events(ctx))
break;
//每8次启动一次任务
if (!(++iters & 7)) {
mutex_unlock(&ctx->uring_lock);
io_run_task_work();
mutex_lock(&ctx->uring_lock);
}
//核心IO轮询函数调用
ret = io_iopoll_getevents(ctx, &nr_events, min);
if (ret <= 0)
break;
ret = 0;
} while (min && !nr_events && !need_resched());
/*
循环退出条件:
1、有事件完成
2、min==0
3、需要进程调度的时候 need_resched()
*/
mutex_unlock(&ctx->uring_lock);
return ret;
}io_cqring_wait等待CQ满足任务数
「1」 首先会使用常见执行方式调用 io_run_task_work()，这一步主要是处理已有事件 ,如果满足最小完成数量会正常返回，但是当 io_run_task_work() 执行失败的时候会进行 **wait **操作。
「2」 然后设置信号掩码、超时时间、超时计数，再次进入一个无限循环，在循环中会挂起当前进程加入等待队列，调用io_run_task_work_sig() 执行待处理任务并检查信号，函数在处理任务之前会先处理信号，防止阻塞。直到没有待执行任务或者其他错误因素才会退出循环。
「3」 最后返回值根据CQ队列，如果CQ队列不为空，即就事件完成就返回0，否则返回错误🐎。
/* fs/io_uring.c */
static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
const sigset_t __user *sig, size_t sigsz,
struct __kernel_timespec __user *uts)
{
// 初始化等待队列结构体
struct io_wait_queue iowq = {
.wq = {
.private = current, // 当前进程
.func = io_wake_function, // 唤醒函数
.entry = LIST_HEAD_INIT(iowq.wq.entry),
},
.ctx = ctx,
.to_wait = min_events, // 等待事件数量
};
struct io_rings *rings = ctx->rings;
struct timespec64 ts;
signed long timeout = 0;
int ret = 0;
// 首先尝试普通执行方式io_run_task_work
do {
// 刷新CQ溢出的事件
io_cqring_overflow_flush(ctx, false, NULL, NULL);
// 完成事件满足最小数量，则返回0
if (io_cqring_events(ctx) >= min_events)
return 0;
// 如果执行失败，则退出循环
if (!io_run_task_work())
break;
} while (1);
// 设置信号掩码、超时时间、超时计数
//...
// 主等待循环
do {
// 刷新CQ溢出事件
io_cqring_overflow_flush(ctx, false, NULL, NULL);
// 将当前进程加入等待队列
prepare_to_wait_exclusive(&ctx->wait, &iowq.wq,
TASK_INTERRUPTIBLE);
// 执行待处理任务并检查信号
ret = io_run_task_work_sig();
if (ret > 0) {
finish_wait(&ctx->wait, &iowq.wq);
continue;
}
else if (ret < 0)
break;
// 检查是否应该唤醒
if (io_should_wake(&iowq))
break;
// 检查是否需要处理CQ溢出
if (test_bit(0, &ctx->cq_check_overflow)) {
finish_wait(&ctx->wait, &iowq.wq);
continue;
}
// 根据是否有超时进行调度
if (uts) {
timeout = schedule_timeout(timeout);
if (timeout == 0) {
ret = -ETIME;
break;
}
} else {
schedule();
}
} while (1);
// 从等待队列中移除
finish_wait(&ctx->wait, &iowq.wq);
restore_saved_sigmask_unless(ret == -EINTR);
// 如果CQ不为空则返回0,否则返回错误码
return READ_ONCE(rings->cq.head) == READ_ONCE(rings->cq.tail) ? ret : 0;
}io_uring_register系统调用
「1」 主要是__io_uring_register()的封装，在最开始的时候会判断文件描述符是否符合io_uring。
/* fs/io_uring.c */
SYSCALL_DEFINE4(io_uring_register, unsigned int, fd, unsigned int, opcode,
void __user *, arg, unsigned int, nr_args)
{
struct io_ring_ctx *ctx;
long ret = -EBADF;
struct fd f;
/* 首先判断fd是否符合io_uring */
f = fdget(fd);
if (!f.file)
return -EBADF;
ret = -EOPNOTSUPP;
if (f.file->f_op != &io_uring_fops)
goto out_fput;
//获取上下文
ctx = f.file->private_data;
mutex_lock(&ctx->uring_lock);
ret = __io_uring_register(ctx, opcode, arg, nr_args);
mutex_unlock(&ctx->uring_lock);
trace_io_uring_register(ctx, opcode, ctx->nr_user_files, ctx->nr_user_bufs,
ctx->cq_ev_fd != NULL, ret);
out_fput:
fdput(f);
return ret;
}__io_uring_register
「1」 首先判断当前上下文是否还存在，通过引用计数来判断，还通过opcode检查操作是否需要暂停，这里不具体展开暂停部分。再次检查opcode的合法性。
「2」 然后是一个大的 switch 选择，opcode选择码如下，根据opcode来选择使用的功能和ioctl一样，在这篇函数就不全部展开分析，选择几个常用的分析——IORING_REGISTER_BUFFERS、IORING_UNREGISTER_BUFFERS、IORING_REGISTER_FILES。
- IORING_REGISTER_BUFFERS: 注册缓冲区
- IORING_UNREGISTER_BUFFERS: 注销缓冲区
- IORING_REGISTER_FILES: 注册文件
- IORING_UNREGISTER_FILES: 注销文件
- IORING_REGISTER_FILES_UPDATE: 更新已注册文件
- IORING_REGISTER_EVENTFD(_ASYNC): 注册事件fd
- IORING_UNREGISTER_EVENTFD: 注销事件fd
- IORING_REGISTER_PROBE: 注册探针
- IORING_REGISTER_PERSONALITY: 注册个性化
- IORING_UNREGISTER_PERSONALITY: 注销个性化
- IORING_REGISTER_ENABLE_RINGS: 启用rings
- IORING_REGISTER_RESTRICTIONS: 注册限制
/* fs/io_uring.c */
static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
void __user *arg, unsigned nr_args)
__releases(ctx->uring_lock)
__acquires(ctx->uring_lock)
{
int ret;
// 检查引用计数是否已死亡
if (percpu_ref_is_dying(&ctx->refs))
return -ENXIO;
// 检查操作是否需要暂停
if (io_register_op_must_quiesce(opcode)) {
//...
}
// 检查上下文限制
if (ctx->restricted) {
//未知的操作🐎
if (opcode >= IORING_REGISTER_LAST) {
ret = -EINVAL;
goto out;
}
//opcode是否被允许
if (!test_bit(opcode, ctx->restrictions.register_op)) {
ret = -EACCES;
goto out;
}
}
switch (opcode) {
case IORING_REGISTER_BUFFERS:
/* 注册用户缓冲区 */
ret = io_sqe_buffer_register(ctx, arg, nr_args);
break;
case IORING_UNREGISTER_BUFFERS:
/* 注销用户缓冲区 */
ret = -EINVAL;
if (arg || nr_args)
break;
ret = io_sqe_buffer_unregister(ctx);
break;
case IORING_REGISTER_FILES:
/* 注册文件描述符 */
ret = io_sqe_files_register(ctx, arg, nr_args);
break;
//其他case...
default:
ret = -EINVAL;
break;
}
out:
//...
return ret;
}io_sqe_buffer_register注册用户缓冲区
「1」 一如既往会先检查各种参数的合法性，首先判断用户缓冲区是否存在，然后判断大小合法之后为ctx->user_bufs分配缓冲区数组(GFP_KERNEL),然后遍历整个数组为其再次开辟空间，可见整个缓冲区采用二维数组的形式。
「2」 循环遍历每个缓冲区，先从用户空间复制IO向量信息到内核态并检查基地址、长度(最大为1GB)，先分配 vmas、pages 空间后面会用来固定用户页面到内存，然后使用kvmalloc_array()为每个缓冲区分配空间，长度和user_bufs一致都为nr_args，标志为GFP_KERNEL。再调用pin_user_pages()把用户页面固定到内存中。
「3」 固定完之后会对每个vma进行检查，不支持非 huge page 的文件内存，最后为 bio_vec 以及当前缓冲区进行初始化。
/* fs/io_uring.c */
static int io_sqe_buffer_register(struct io_ring_ctx *ctx, void __user *arg,
unsigned nr_args)
{
struct vm_area_struct **vmas = NULL; /* 虚拟内存区域数组 */
struct page **pages = NULL; /* 页面数组 */
struct page *last_hpage = NULL; /* 上一个大页面 */
int i, j, got_pages = 0;
int ret = -EINVAL;
// 检查是否已经注册过缓冲区,参数合法性
if (ctx->user_bufs)
return -EBUSY;
if (!nr_args || nr_args > UIO_MAXIOV)
return -EINVAL;
// 分配用户缓冲区数组并初始化为0
ctx->user_bufs = kcalloc(nr_args, sizeof(struct io_mapped_ubuf),
GFP_KERNEL);
if (!ctx->user_bufs)
return -ENOMEM;
//遍历每个缓冲区
for (i = 0; i < nr_args; i++) {
struct io_mapped_ubuf *imu = &ctx->user_bufs[i]; /* 当前缓冲区结构体 */
unsigned long off, start, end, ubuf; /* 偏移量和地址计算变量 */
int pret, nr_pages; /* pin_user_pages的返回值和页面数 */
struct iovec iov; /* IO向量结构体 */
size_t size; /* 缓冲区大小 */
// 从用户空间复制IO向量信息到iov
ret = io_copy_iov(ctx, &iov, arg, i);
if (ret)
goto err;
//检查iov的基地址和长度
ret = -EFAULT;
if (!iov.iov_base || !iov.iov_len)
goto err;
// 限制单个缓冲区最大长度为1GB（0x40000000）
if (iov.iov_len > SZ_1G)
goto err;
ubuf = (unsigned long) iov.iov_base;
end = (ubuf + iov.iov_len + PAGE_SIZE - 1) >> PAGE_SHIFT;
start = ubuf >> PAGE_SHIFT;
nr_pages = end - start;
/* 如果需要更多页面空间，重新分配数组 */
ret = 0;
if (!pages || nr_pages > got_pages) {
kvfree(vmas);
kvfree(pages);
pages = kvmalloc_array(nr_pages, sizeof(struct page *),
GFP_KERNEL);
vmas = kvmalloc_array(nr_pages,
sizeof(struct vm_area_struct *),
GFP_KERNEL);
if (!pages || !vmas) {
ret = -ENOMEM;
goto err;
}
got_pages = nr_pages;
}
// 为bio_vec向量数组 分配内存
imu->bvec = kvmalloc_array(nr_pages, sizeof(struct bio_vec),
GFP_KERNEL);
ret = -ENOMEM;
if (!imu->bvec)
goto err;
ret = 0;
// 上内存映射读锁
mmap_read_lock(current->mm);
/* 固定用户页面到内存中 */
pret = pin_user_pages(ubuf, nr_pages,
FOLL_WRITE | FOLL_LONGTERM,
pages, vmas);
if (pret == nr_pages) {
/*
正常都会执行到这里
检查是否有文件支持的内存
(不支持非huge page的文件内存，即只支持大页)
*/
for (j = 0; j < nr_pages; j++) {
struct vm_area_struct *vma = vmas[j];
if (vma->vm_file &&
!is_file_hugepages(vma->vm_file)) {
ret = -EOPNOTSUPP;
break;
}
}
} else {
ret = pret < 0 ? pret : -EFAULT;
}
mmap_read_unlock(current->mm);
if (ret) {
/* 有错误，释放已获取的页面 */
if (pret > 0)
//有文件不符合条件(不是大页)
unpin_user_pages(pages, pret);
kvfree(imu->bvec);
goto err;
}
/* 记录页面使用情况 */
ret = io_buffer_account_pin(ctx, pages, pret, imu, &last_hpage);
if (ret) {
unpin_user_pages(pages, pret);
kvfree(imu->bvec);
goto err;
}
/* 设置bio_vec */
off = ubuf & ~PAGE_MASK; /* 计算页内偏移 */
size = iov.iov_len;
for (j = 0; j < nr_pages; j++) {
size_t vec_len;
vec_len = min_t(size_t, size, PAGE_SIZE - off);
imu->bvec[j].bv_page = pages[j];
imu->bvec[j].bv_len = vec_len;
imu->bvec[j].bv_offset = off;
off = 0;
size -= vec_len;
}
imu->ubuf = ubuf;
imu->len = iov.iov_len;
imu->nr_bvecs = nr_pages;
ctx->nr_user_bufs++;
}
kvfree(pages);
kvfree(vmas);
return 0;
err:
kvfree(pages);
kvfree(vmas);
io_sqe_buffer_unregister(ctx);
return ret;
}io_sqe_buffer_unregister注销用户缓冲区
「1」 注销缓冲区相对没那么多步骤，如果用户缓冲区存在时，先解除用户映射的 bio_vec 向量，然后释放 imu->bvec ，最后释放ctx->user_bufs并置零。
static int io_sqe_buffer_unregister(struct io_ring_ctx *ctx)
{
int i, j;
// 检查缓冲区是否存在
if (!ctx->user_bufs)
return -ENXIO;
// 根据ctx的用户缓冲区数量遍历
for (i = 0; i < ctx->nr_user_bufs; i++) {
struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
//bio_vec向量解除用户映射
for (j = 0; j < imu->nr_bvecs; j++)
unpin_user_page(imu->bvec[j].bv_page);
if (imu->acct_pages)
io_unaccount_mem(ctx, imu->acct_pages, ACCT_PINNED);
kvfree(imu->bvec);//释放bio_vec
imu->nr_bvecs = 0;
}
//释放用户缓冲区数组
kfree(ctx->user_bufs);
ctx->user_bufs = NULL;
ctx->nr_user_bufs = 0;
return 0;
}io_sqe_files_register注册文件
「1」 遍历用户传入需要注册的所有文件，并使用 fget() 获取file结构体，这个操作会造成文件引用计数+1,需要注意的是注册不支持io_uring本身
「2」 调用 io_sqe_files_scm() 来设置 SCM_RIGHTS ，这个函数在飞行计数中很关键，在之后会详细分析这个函数，这里暂时不咱开。然后分配并设置引用节点后返回，就完成了注册文件的过程。
static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
unsigned nr_args)
{
/* 用户空间传入的文件描述符数组 */
__s32 __user *fds = (__s32 __user *) arg;
//...
/* 遍历注册所有文件 */
for (i = 0; i < nr_args; i++, ctx->nr_user_files++) {
struct fixed_file_table *table;
unsigned index;
// 拷贝用户空间的fd
if (copy_from_user(&fd, &fds[i], sizeof(fd))) {
ret = -EFAULT;
goto out_fput;
}
if (fd == -1)
continue;
// 获取file结构体，fget()引用计数+1
file = fget(fd);
ret = -EBADF;
if (!file)
goto out_fput;
/*
* 不允许注册io_uring实例本身:
* 1. 如果未启用UNIX域套接字,会导致引用循环,无法释放
* 2. 即使启用了UNIX域套接字,io_uring实例也不支持常规读写,所以没有意义
*/
if (file->f_op == &io_uring_fops) {
fput(file);
goto out_fput;
}
//保存file到表格对应位置
table = &file_data->table[i >> IORING_FILE_TABLE_SHIFT];
index = i & IORING_FILE_TABLE_MASK;
table->files[index] = file;
}
// 设置SCM_RIGHTS
ret = io_sqe_files_scm(ctx);
if (ret) {
io_sqe_files_unregister(ctx);
return ret;
}
/* 分配引用节点 */
ref_node = alloc_fixed_file_ref_node(ctx);
if (!ref_node) {
io_sqe_files_unregister(ctx);
return -ENOMEM;
}
/* 设置引用节点 */
io_sqe_files_set_node(file_data, ref_node);
return ret;
out_fput:
//...
}END