Kernel源码笔记目录

io-wq

源码基于5.10

io-wq是io_uring用来做异步任务。

创建 io-wq

struct io_wq *io_wq_create(unsigned bounded, struct io_wq_data *data)
{int ret = -ENOMEM, node;struct io_wq *wq;if (WARN_ON_ONCE(!data->free_work || !data->do_work))return ERR_PTR(-EINVAL);if (WARN_ON_ONCE(!bounded))return ERR_PTR(-EINVAL);wq = kzalloc(sizeof(*wq), GFP_KERNEL);if (!wq)return ERR_PTR(-ENOMEM);// wqe是每个node上申请一个wq->wqes = kcalloc(nr_node_ids, sizeof(struct io_wqe *), GFP_KERNEL);if (!wq->wqes)goto err_wq;// todo: 后面看。好像会设置运行任务task的亲和性，应该是附上迁移ret = cpuhp_state_add_instance_nocalls(io_wq_online, &wq->cpuhp_node);if (ret)goto err_wqes;// 释放work的函数wq->free_work = data->free_work;// 提交work的函数wq->do_work = data->do_work;wq->user = data->user;ret = -ENOMEM;// 在对应的node上分配wqefor_each_node(node) {struct io_wqe *wqe;int alloc_node = node;// node不在线if (!node_online(alloc_node))// todo: 后面看，这种情况一般很少alloc_node = NUMA_NO_NODE;// 在对应的node上分配wqe内存wqe = kzalloc_node(sizeof(struct io_wqe), GFP_KERNEL, alloc_node);if (!wqe)goto err;// 设置到对应的wqes里wq->wqes[node] = wqe;wqe->node = alloc_node;// 下面是统计相关// 最大worker数量wqe->acct[IO_WQ_ACCT_BOUND].max_workers = bounded;// 当前运行进程数量atomic_set(&wqe->acct[IO_WQ_ACCT_BOUND].nr_running, 0);if (wq->user) {// todo: IO_WQ_ACCT_UNBOUND?wqe->acct[IO_WQ_ACCT_UNBOUND].max_workers =task_rlimit(current, RLIMIT_NPROC);}atomic_set(&wqe->acct[IO_WQ_ACCT_UNBOUND].nr_running, 0);// wqe对wq的引用wqe->wq = wq;raw_spin_lock_init(&wqe->lock);// 这个是工作列表，异步任务就挂在这个列表上INIT_WQ_LIST(&wqe->work_list);// todo:INIT_HLIST_NULLS_HEAD(&wqe->free_list, 0);// todo:INIT_LIST_HEAD(&wqe->all_list);}// 初始化完成量init_completion(&wq->done);// 创建wq管理器线程wq->manager = kthread_create(io_wq_manager, wq, "io_wq_manager");if (!IS_ERR(wq->manager)) { // 创建成功// 唤醒io_wq_manager线程wake_up_process(wq->manager);// 等待线程准备完成wait_for_completion(&wq->done);// 有错误if (test_bit(IO_WQ_BIT_ERROR, &wq->state)) {ret = -ENOMEM;goto err;}refcount_set(&wq->use_refs, 1);// 重新初始化完成量reinit_completion(&wq->done);return wq;}// 走到这儿，说明创建io_wq_manager失败ret = PTR_ERR(wq->manager);// todo: 创建失败后，为啥要调用完成 ？complete(&wq->done);
err:cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node);for_each_node(node)kfree(wq->wqes[node]);
err_wqes:kfree(wq->wqes);
err_wq:kfree(wq);return ERR_PTR(ret);
}

io_wq_create的主流程：

创建io_wq实例
创建wq->wqes，这是数组，大小是numa节点的数量
给每个node分配wqe节点
创建 io_wq_manager 线程，这个线程负责创建每个wq_worker线程
唤醒 io_wq_manager 线程，并等待wq->done完成

io_wq_manager

static int io_wq_manager(void *data)
{struct io_wq *wq = data;int node;refcount_set(&wq->refs, 1);// 在每个node上创建workerfor_each_node(node) {// node不在线if (!node_online(node))continue;// 创建workerif (create_io_worker(wq, wq->wqes[node], IO_WQ_ACCT_BOUND))continue;// 走到这儿说明创建失败，直接退出set_bit(IO_WQ_BIT_ERROR, &wq->state);set_bit(IO_WQ_BIT_EXIT, &wq->state);goto out;}// 走到这儿，说明在每个node上创建成功// 完成，这里对应io_wq_create里的完成量complete(&wq->done);// 下面这个while循环，就是定时遍历wqe的状态，// 根据worker上任务的数量，看是否要为其分配 workerwhile (!kthread_should_stop()) {// 如果进程有work则先运行进程的workif (current->task_works)task_work_run();for_each_node(node) {// 每个node对应的wqestruct io_wqe *wqe = wq->wqes[node];bool fork_worker[2] = { false, false };// node不在线if (!node_online(node))continue;raw_spin_lock_irq(&wqe->lock);// 是否需要创建workderif (io_wqe_need_worker(wqe, IO_WQ_ACCT_BOUND))fork_worker[IO_WQ_ACCT_BOUND] = true;if (io_wqe_need_worker(wqe, IO_WQ_ACCT_UNBOUND))fork_worker[IO_WQ_ACCT_UNBOUND] = true;raw_spin_unlock_irq(&wqe->lock);// 如果需要创建，则创建对应的workerif (fork_worker[IO_WQ_ACCT_BOUND])create_io_worker(wq, wqe, IO_WQ_ACCT_BOUND);if (fork_worker[IO_WQ_ACCT_UNBOUND])create_io_worker(wq, wqe, IO_WQ_ACCT_UNBOUND);}// 设置可中断状态，并延迟一秒set_current_state(TASK_INTERRUPTIBLE);schedule_timeout(HZ);}// 再运行一下进程的workif (current->task_works)task_work_run();out:if (refcount_dec_and_test(&wq->refs)) {// 如果是最后一个走到这儿，还要再调一下完成量，因为有可能是因为出错走到这儿complete(&wq->done);return 0;}if (test_bit(IO_WQ_BIT_ERROR, &wq->state)) {rcu_read_lock();// 如果有错误，就唤醒所有的worker， why?for_each_node(node)io_wq_for_each_worker(wq->wqes[node], io_wq_worker_wake, NULL);rcu_read_unlock();}return 0;
}static inline bool io_wqe_need_worker(struct io_wqe *wqe, int index)__must_hold(wqe->lock)
{struct io_wqe_acct *acct = &wqe->acct[index];// free_list里是空闲的worker// io_wqe_run_queue是判断wqe里有没有work// 如果有空闲的worker，或者没有任务，就不需要再创建workerif (!hlist_nulls_empty(&wqe->free_list) || !io_wqe_run_queue(wqe))return false;// 走到这儿需要创建worker，但是不能超过最大值return acct->nr_workers < acct->max_workers;
}

io_wq_manager的主要流程:

先在每个node上创建一个io_worker,然后调用wq_done的完成接口,因为io_wq_create还在这个完成量上等待
然后进入一个死循环,每隔一秒,判断一下每个node上的wqe的任务情况,根据是否需要创建io_worker
其它一些出错处理和manager线程退出时的一些操作

创建一个worker

static bool create_io_worker(struct io_wq *wq, struct io_wqe *wqe, int index)
{// 统计对象struct io_wqe_acct *acct = &wqe->acct[index];struct io_worker *worker;// 在对应的node上创建worker内存worker = kzalloc_node(sizeof(*worker), GFP_KERNEL, wqe->node);if (!worker)return false;// 设置引用为1refcount_set(&worker->ref, 1);worker->nulls_node.pprev = NULL;worker->wqe = wqe;spin_lock_init(&worker->lock);// 创建worker对应的内核线程worker->task = kthread_create_on_node(io_wqe_worker, worker, wqe->node,"io_wqe_worker-%d/%d", index, wqe->node);if (IS_ERR(worker->task)) {kfree(worker);return false;}// bind_mask表示只能在这个node对应的cpu上运行kthread_bind_mask(worker->task, cpumask_of_node(wqe->node));raw_spin_lock_irq(&wqe->lock);// 把worker挂在free_list上hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);// 把worker加到all_list表尾list_add_tail_rcu(&worker->all_list, &wqe->all_list);// 设置 worker现在空闲worker->flags |= IO_WORKER_F_FREE;if (index == IO_WQ_ACCT_BOUND)worker->flags |= IO_WORKER_F_BOUND;// 如果是node上的第1个线程,则设置固定标志,因为非固定线程在没有任务时会退出if (!acct->nr_workers && (worker->flags & IO_WORKER_F_BOUND))worker->flags |= IO_WORKER_F_FIXED;// 递增worker数量acct->nr_workers++;raw_spin_unlock_irq(&wqe->lock);// 如果是无界限的则增加用户的进程数量？if (index == IO_WQ_ACCT_UNBOUND)atomic_inc(&wq->user->processes);// 增加wq引用refcount_inc(&wq->refs);// 唤醒worker线程wake_up_process(worker->task);return true;
}

创建线程比较简单,创建一个内核线程后,把它分别挂在wq的对应列表上,然后再唤醒这个内核线程,就完事了!

io_wqe_worker

io_wqe_worker是异步任务的主要执行者.

static int io_wqe_worker(void *data)
{struct io_worker *worker = data;struct io_wqe *wqe = worker->wqe;struct io_wq *wq = wqe->wq;// 设置woker运行时的一些标志,及上下文io_worker_start(wqe, worker);while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) { // 是否要退出// 先设置可中断状态,下面可能要睡眠set_current_state(TASK_INTERRUPTIBLE);
loop:raw_spin_lock_irq(&wqe->lock);// 如果有运行的任务if (io_wqe_run_queue(wqe)) {// 把当前进程设置成running__set_current_state(TASK_RUNNING);// 处理任务io_worker_handle_work(worker);// 处理完一个任务后继续循环goto loop;}// 走到这儿表示没有任务要处理// 没有任务时就把当前worker加到空闲列表，返回true表示需要释放锁if (__io_worker_idle(wqe, worker)) {__release(&wqe->lock);goto loop;}raw_spin_unlock_irq(&wqe->lock);// 当前进程有信号要处理if (signal_pending(current))// 处理信号flush_signals(current);// 睡眠， WORKER_IDLE_TIMEOUT是5秒if (schedule_timeout(WORKER_IDLE_TIMEOUT))// 走到这儿说明是在睡眠期间被唤醒,// 睡眠的时候有被唤醒,说明有任务需要处理,继续循环continue;// 走到这儿说明是休眠到期唤醒的// 如果是退出，或者当前worker不是固定，则退出.// 每个node上只有第1个线程是固定,其它都是根据任务的多少动态创建的,// 所以走到这儿既然没有任务,那这个非固定线程也就没用了,直接退出if (test_bit(IO_WQ_BIT_EXIT, &wq->state) ||!(worker->flags & IO_WORKER_F_FIXED))break;}if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) {// 如果是因为设置了退出退出标志,有可能队列中还有工作没做完,// 所以把余下的任务做完raw_spin_lock_irq(&wqe->lock);if (!wq_list_empty(&wqe->work_list))io_worker_handle_work(worker);elseraw_spin_unlock_irq(&wqe->lock);}// woker退出,对应着前面的startio_worker_exit(worker);return 0;
}static void io_worker_start(struct io_wqe *wqe, struct io_worker *worker)
{// 允许的信号？allow_kernel_signal(SIGINT);// 设置worker标志current->flags |= PF_IO_WORKER;// 设置运行标志worker->flags |= (IO_WORKER_F_UP | IO_WORKER_F_RUNNING);// 设置复原时文件系统相关变量为当前进程的worker->restore_files = current->files;worker->restore_nsproxy = current->nsproxy;worker->restore_fs = current->fs;// 增加相应的acct->nr_running数量，这个是运行数量io_wqe_inc_running(wqe, worker);
}

woker的核心流程是处理队列上的任务.没有任务了就去睡眠.如果是临时woker线程的话,没有任务了这个线程就会退出.

处理任务

static void io_worker_handle_work(struct io_worker *worker)__releases(wqe->lock)
{struct io_wqe *wqe = worker->wqe;struct io_wq *wq = wqe->wq;do {struct io_wq_work *work;
get_next:// 取出一个任务work = io_get_next_work(wqe);if (work)// 这个函数去除当前空闲标志,也就是设置忙标志,// 主要是从空闲列表里移出__io_worker_busy(wqe, worker, work);else if (!wq_list_empty(&wqe->work_list))// 如果获取work为空,但是work_list不为空,那肯定是在哈希中// todo: 这个条件没看懂wqe->flags |= IO_WQE_FLAG_STALLED;raw_spin_unlock_irq(&wqe->lock);if (!work)break;// 这个主要设置worker->cur_workio_assign_current_work(worker, work);/* handle a whole dependent link */do {struct io_wq_work *old_work, *next_hashed, *linked;unsigned int hash = io_get_work_hash(work);// 下一个任务, work是一个链表next_hashed = wq_next_work(work);// work的运行上下文与当前worker的不一致的时候，要设置成work的上下文io_impersonate_work(worker, work);// 当前work已经取消?if (test_bit(IO_WQ_BIT_CANCEL, &wq->state))work->flags |= IO_WQ_WORK_CANCEL;old_work = work;// 调用相应work的处理函数// todo: 返回的linked是什么？linked = wq->do_work(work);work = next_hashed;// 如果下一个work是空，但是link不为空，且不是哈希表头，设置work为linkif (!work && linked && !io_wq_is_hashed(linked)) {work = linked;linked = NULL;}// 设置下一个workio_assign_current_work(worker, work);// 释放老的work,也就是处理完的workwq->free_work(old_work);// 如果link没有处理，说明当前work相关的下一个work不为空，// 先把link入队if (linked)io_wqe_enqueue(wqe, linked);// 后面再看if (hash != -1U && !next_hashed) {raw_spin_lock_irq(&wqe->lock);wqe->hash_map &= ~BIT_ULL(hash);wqe->flags &= ~IO_WQE_FLAG_STALLED;/* skip unnecessary unlock-lock wqe->lock */if (!work)goto get_next;raw_spin_unlock_irq(&wqe->lock);}} while (work);raw_spin_lock_irq(&wqe->lock);} while (1);
}static struct io_wq_work *io_get_next_work(struct io_wqe *wqe)__must_hold(wqe->lock)
{struct io_wq_work_node *node, *prev;struct io_wq_work *work, *tail;unsigned int hash;// 遍历wqe的work_listwq_list_for_each(node, prev, &wqe->work_list) {work = container_of(node, struct io_wq_work, list);// io_wq_is_hashed判断有没有IO_WQ_WORK_HASHED标志,// 这个标志代表是哈希表头if (!io_wq_is_hashed(work)) {// 不是哈希表头的直接返回wq_list_del(&wqe->work_list, node, prev);return work;}// 说明这个work是哈希表头，// io_get_work_hash 获取它的哈希值hash = io_get_work_hash(work);// BIT(nr) = 1 << nr// todo: hash_map是什么if (!(wqe->hash_map & BIT(hash))) {// 把hash对应的那一位设置后,就把hash值对应的// 整个哈希表都放到work_list上wqe->hash_map |= BIT(hash);// 取出这个哈希表tail = wqe->hash_tail[hash];wqe->hash_tail[hash] = NULL;// 把整个哈希表加到work_list上wq_list_cut(&wqe->work_list, &tail->list, prev);return work;}}return NULL;
}static void __io_worker_busy(struct io_wqe *wqe, struct io_worker *worker,struct io_wq_work *work)__must_hold(wqe->lock)
{bool worker_bound, work_bound;// 去除空闲标志,并从空闲列表里移出if (worker->flags & IO_WORKER_F_FREE) {worker->flags &= ~IO_WORKER_F_FREE;hlist_nulls_del_init_rcu(&worker->nulls_node);}// 计算bound/unbound计数器worker_bound = (worker->flags & IO_WORKER_F_BOUND) != 0;work_bound = (work->flags & IO_WQ_WORK_UNBOUND) == 0;if (worker_bound != work_bound) {io_wqe_dec_running(wqe, worker);if (work_bound) {worker->flags |= IO_WORKER_F_BOUND;wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers--;wqe->acct[IO_WQ_ACCT_BOUND].nr_workers++;atomic_dec(&wqe->wq->user->processes);} else {worker->flags &= ~IO_WORKER_F_BOUND;wqe->acct[IO_WQ_ACCT_UNBOUND].nr_workers++;wqe->acct[IO_WQ_ACCT_BOUND].nr_workers--;atomic_inc(&wqe->wq->user->processes);}io_wqe_inc_running(wqe, worker);}
}static void io_assign_current_work(struct io_worker *worker,struct io_wq_work *work)
{if (work) {// 如果有信号先处理信号if (signal_pending(current))flush_signals(current);// 让出cpucond_resched();}#ifdef CONFIG_AUDIT// 统计相关current->loginuid = KUIDT_INIT(AUDIT_UID_UNSET);current->sessionid = AUDIT_SID_UNSET;
#endifspin_lock_irq(&worker->lock);// 设置当前worker的workworker->cur_work = work;spin_unlock_irq(&worker->lock);
}static void io_impersonate_work(struct io_worker *worker,struct io_wq_work *work)
{// 如果work的files与当前worker不同，需要设置？if ((work->flags & IO_WQ_WORK_FILES) &&current->files != work->identity->files) {task_lock(current);current->files = work->identity->files;current->nsproxy = work->identity->nsproxy;task_unlock(current);if (!work->identity->files) {/* failed grabbing files, ensure work gets cancelled */work->flags |= IO_WQ_WORK_CANCEL;}}// 同上，设置文件系统if ((work->flags & IO_WQ_WORK_FS) && current->fs != work->identity->fs)current->fs = work->identity->fs;// 同上，设置内存if ((work->flags & IO_WQ_WORK_MM) && work->identity->mm != worker->mm)io_wq_switch_mm(worker, work);// credif ((work->flags & IO_WQ_WORK_CREDS) &&worker->cur_creds != work->identity->creds)io_wq_switch_creds(worker, work);// 信号集if (work->flags & IO_WQ_WORK_FSIZE)current->signal->rlim[RLIMIT_FSIZE].rlim_cur = work->identity->fsize;else if (current->signal->rlim[RLIMIT_FSIZE].rlim_cur != RLIM_INFINITY)current->signal->rlim[RLIMIT_FSIZE].rlim_cur = RLIM_INFINITY;// blk控制组io_wq_switch_blkcg(worker, work);
#ifdef CONFIG_AUDIT// 审计的会话相关idcurrent->loginuid = work->identity->loginuid;current->sessionid = work->identity->sessionid;
#endif
}
static bool __io_worker_idle(struct io_wqe *wqe, struct io_worker *worker)__must_hold(wqe->lock)
{// 设置空闲标志,加入空闲列表if (!(worker->flags & IO_WORKER_F_FREE)) {worker->flags |= IO_WORKER_F_FREE;hlist_nulls_add_head_rcu(&worker->nulls_node, &wqe->free_list);}return __io_worker_unuse(wqe, worker);
}static void io_worker_exit(struct io_worker *worker)
{struct io_wqe *wqe = worker->wqe;struct io_wqe_acct *acct = io_wqe_get_acct(wqe, worker);set_current_state(TASK_INTERRUPTIBLE);// 返回0说明还有别人在引用,所以调度出去if (!refcount_dec_and_test(&worker->ref))schedule();// 走到这儿说明没有人引用了// 先把状态改回来__set_current_state(TASK_RUNNING);// 关中断preempt_disable();// 删除worker标记current->flags &= ~PF_IO_WORKER;// 减少running计数if (worker->flags & IO_WORKER_F_RUNNING)atomic_dec(&acct->nr_running);// 减少用户的processes计数if (!(worker->flags & IO_WORKER_F_BOUND))atomic_dec(&wqe->wq->user->processes);worker->flags = 0;preempt_enable();raw_spin_lock_irq(&wqe->lock);// 删除worker空闲列表hlist_nulls_del_rcu(&worker->nulls_node);// 删除all_listlist_del_rcu(&worker->all_list);// 清除所占用的资源if (__io_worker_unuse(wqe, worker)) {__release(&wqe->lock);raw_spin_lock_irq(&wqe->lock);}acct->nr_workers--;raw_spin_unlock_irq(&wqe->lock);// 释放workerkfree_rcu(worker, rcu);// 没有人再使用它了,调用完成量if (refcount_dec_and_test(&wqe->wq->refs))complete(&wqe->wq->done);
}static bool __io_worker_unuse(struct io_wqe *wqe, struct io_worker *worker)
{bool dropped_lock = false;if (worker->saved_creds) {revert_creds(worker->saved_creds);worker->cur_creds = worker->saved_creds = NULL;}if (current->files != worker->restore_files) {__acquire(&wqe->lock);raw_spin_unlock_irq(&wqe->lock);dropped_lock = true;task_lock(current);current->files = worker->restore_files;current->nsproxy = worker->restore_nsproxy;task_unlock(current);}if (current->fs != worker->restore_fs)current->fs = worker->restore_fs;/** If we have an active mm, we need to drop the wq lock before unusing* it. If we do, return true and let the caller retry the idle loop.*/if (worker->mm) {if (!dropped_lock) {__acquire(&wqe->lock);raw_spin_unlock_irq(&wqe->lock);dropped_lock = true;}__set_current_state(TASK_RUNNING);kthread_unuse_mm(worker->mm);mmput(worker->mm);worker->mm = NULL;}#ifdef CONFIG_BLK_CGROUPif (worker->blkcg_css) {kthread_associate_blkcg(NULL);worker->blkcg_css = NULL;}
#endifif (current->signal->rlim[RLIMIT_FSIZE].rlim_cur != RLIM_INFINITY)current->signal->rlim[RLIMIT_FSIZE].rlim_cur = RLIM_INFINITY;return dropped_lock;
}

文件系统：3. io_uring-io_wq原理相关推荐

FastDFS文件系统简介与架构原理
FastDFS分布式文件系统概述概述 FastDFS是一个轻量级的开源分布式文件系统 FastDFS主要解决了大容量的文件存储和高并发访问的问题,文件存取时实现了负载均衡 FastDFS实现了软件方 ...
Hadoop分布式文件系统HDFS的工作原理详述
Hadoop分布式文件系统(HDFS)是一种被设计成适合运行在通用硬件上的分布式文件系统.HDFS是一个高度容错性的系统,适合部署在廉价的机器上.它能提供高吞吐量的数据访问,非常适合大规模数据集上的应 ...
企业级别应用--GFS分布式文件系统（GlusterFS工作原理、弹性 HASH 算法、GlusterFS卷的类型、部署GlusterFS）
文章目录一. GlusterFS 概述 1.1 GlusterFS 简介与传统分布式相比的优点 1.2 GlusterFS 的特点扩展性和高性能高可用性全局统一命名空间弹性卷管理基于标准 ...
分布式文件系统HDFS实践及原理详解part3
HDFS原理说明:3.5开头目录是因为和上篇文章内容同属一章,所以开头使用了3.5 3.5 HDFS核心设计 3.5.1 心跳机制 1. Hadoop 是 Master/Slave 结构,Maste ...
【docker知识】联合文件系统（unionFS）原理
一.说明 Docker CLI 操作起来比较简单--您只需掌握Create.Run.InspPull和Push容器和图像,但是谁想过Docker 背后的内部机制是如何工作的?在这个简单的表象背后隐藏着 ...
分布式文件系统：HDFS 核心原理
点击上方蓝色字体,选择"设为星标" 回复"资源"获取更多资源大数据技术与架构点击右侧关注,大数据开发领域最强公众号! 大数据真好玩点击右侧关注,大数据真好 ...
Windows文件系统以及文件粉碎原理
1.1硬盘揭秘: 物理存储方式: 目前的存储方式有磁存储,电存储,光存储.U盘就是电存储,VCD,DVD光盘用的是光存储,我们计算机用的硬盘就是用的磁存储.各种存储方式在物理存储介质不同外,在逻辑层面 ...
全面详细介绍Linux 虚拟文件系统（VFS）原理
一. 通用文件模型 Linux内核支持装载不同的文件系统类型,不同的文件系统有各自管理文件的方式.Linux中标准的文件系统为Ext文件系统族,当然,开发者不能为他们使用的每种文件系统采用不同的文件存 ...
linux中的根文件系统（rootfs的原理和介绍）
一.什么是文件系统文件系统是操作系统用于明确存储设备(常见的是磁盘,也有基于NAND Flash的固态硬盘)或分区上的文件的方法和数据结构:即在存储设备上组织文件的方法.操作系统中负责管理和存储文件 ...

文件系统：3. io_uring-io_wq原理