1 // SPDX-License-Identifier: GPL-2.0
3 * Shared application/kernel submission and completion ring pairs, for
4 * supporting fast/efficient IO.
6 * A note on the read/write ordering memory barriers that are matched between
7 * the application and kernel side. When the application reads the CQ ring
8 * tail, it must use an appropriate smp_rmb() to order with the smp_wmb()
9 * the kernel uses after writing the tail. Failure to do so could cause a
10 * delay in when the application notices that completion events available.
11 * This isn't a fatal condition. Likewise, the application must use an
12 * appropriate smp_wmb() both before writing the SQ tail, and after writing
13 * the SQ tail. The first one orders the sqe writes with the tail write, and
14 * the latter is paired with the smp_rmb() the kernel will issue before
15 * reading the SQ tail on submission.
17 * Also see the examples in the liburing library:
19 * git://git.kernel.dk/liburing
21 * io_uring also uses READ/WRITE_ONCE() for _any_ store or load that happens
22 * from data shared between the kernel and application. This is done both
23 * for ordering purposes, but also to ensure that once a value is loaded from
24 * data that the application could potentially modify, it remains stable.
26 * Copyright (C) 2018-2019 Jens Axboe
27 * Copyright (c) 2018-2019 Christoph Hellwig
29 #include <linux/kernel.h>
30 #include <linux/init.h>
31 #include <linux/errno.h>
32 #include <linux/syscalls.h>
33 #include <linux/compat.h>
34 #include <linux/refcount.h>
35 #include <linux/uio.h>
37 #include <linux/sched/signal.h>
39 #include <linux/file.h>
40 #include <linux/fdtable.h>
42 #include <linux/mman.h>
43 #include <linux/mmu_context.h>
44 #include <linux/percpu.h>
45 #include <linux/slab.h>
46 #include <linux/workqueue.h>
47 #include <linux/kthread.h>
48 #include <linux/blkdev.h>
49 #include <linux/bvec.h>
50 #include <linux/net.h>
52 #include <net/af_unix.h>
54 #include <linux/anon_inodes.h>
55 #include <linux/sched/mm.h>
56 #include <linux/uaccess.h>
57 #include <linux/nospec.h>
58 #include <linux/sizes.h>
59 #include <linux/hugetlb.h>
61 #include <uapi/linux/io_uring.h>
65 #define IORING_MAX_ENTRIES 4096
66 #define IORING_MAX_FIXED_FILES 1024
69 u32 head ____cacheline_aligned_in_smp;
70 u32 tail ____cacheline_aligned_in_smp;
87 struct io_uring_cqe cqes[];
90 struct io_mapped_ubuf {
94 unsigned int nr_bvecs;
100 struct list_head list;
109 struct percpu_ref refs;
110 } ____cacheline_aligned_in_smp;
118 struct io_sq_ring *sq_ring;
119 unsigned cached_sq_head;
122 unsigned sq_thread_idle;
123 struct io_uring_sqe *sq_sqes;
124 } ____cacheline_aligned_in_smp;
127 struct workqueue_struct *sqo_wq;
128 struct task_struct *sqo_thread; /* if using sq thread polling */
129 struct mm_struct *sqo_mm;
130 wait_queue_head_t sqo_wait;
135 struct io_cq_ring *cq_ring;
136 unsigned cached_cq_tail;
139 struct wait_queue_head cq_wait;
140 struct fasync_struct *cq_fasync;
141 } ____cacheline_aligned_in_smp;
144 * If used, fixed file set. Writers must ensure that ->refs is dead,
145 * readers must ensure that ->refs is alive as long as the file* is
146 * used. Only updated through io_uring_register(2).
148 struct file **user_files;
149 unsigned nr_user_files;
151 /* if used, fixed mapped user buffers */
152 unsigned nr_user_bufs;
153 struct io_mapped_ubuf *user_bufs;
155 struct user_struct *user;
157 struct completion ctx_done;
160 struct mutex uring_lock;
161 wait_queue_head_t wait;
162 } ____cacheline_aligned_in_smp;
165 spinlock_t completion_lock;
166 bool poll_multi_file;
168 * ->poll_list is protected by the ctx->uring_lock for
169 * io_uring instances that don't use IORING_SETUP_SQPOLL.
170 * For SQPOLL, only the single threaded io_sq_thread() will
171 * manipulate the list, hence no extra locking is needed there.
173 struct list_head poll_list;
174 struct list_head cancel_list;
175 } ____cacheline_aligned_in_smp;
177 struct async_list pending_async[2];
179 #if defined(CONFIG_UNIX)
180 struct socket *ring_sock;
185 const struct io_uring_sqe *sqe;
186 unsigned short index;
189 bool needs_fixed_file;
192 struct io_poll_iocb {
194 struct wait_queue_head *head;
198 struct wait_queue_entry wait;
204 struct io_poll_iocb poll;
207 struct sqe_submit submit;
209 struct io_ring_ctx *ctx;
210 struct list_head list;
213 #define REQ_F_FORCE_NONBLOCK 1 /* inline submission attempt */
214 #define REQ_F_IOPOLL_COMPLETED 2 /* polled IO has completed */
215 #define REQ_F_FIXED_FILE 4 /* ctx owns file */
216 #define REQ_F_SEQ_PREV 8 /* sequential with previous */
220 struct work_struct work;
223 #define IO_PLUG_THRESHOLD 2
224 #define IO_IOPOLL_BATCH 8
226 struct io_submit_state {
227 struct blk_plug plug;
230 * io_kiocb alloc cache
232 void *reqs[IO_IOPOLL_BATCH];
233 unsigned int free_reqs;
234 unsigned int cur_req;
237 * File reference cache
241 unsigned int has_refs;
242 unsigned int used_refs;
243 unsigned int ios_left;
246 static struct kmem_cache *req_cachep;
248 static const struct file_operations io_uring_fops;
250 struct sock *io_uring_get_socket(struct file *file)
252 #if defined(CONFIG_UNIX)
253 if (file->f_op == &io_uring_fops) {
254 struct io_ring_ctx *ctx = file->private_data;
256 return ctx->ring_sock->sk;
261 EXPORT_SYMBOL(io_uring_get_socket);
263 static void io_ring_ctx_ref_free(struct percpu_ref *ref)
265 struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);
267 complete(&ctx->ctx_done);
270 static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
272 struct io_ring_ctx *ctx;
275 ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
279 if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free, 0, GFP_KERNEL)) {
284 ctx->flags = p->flags;
285 init_waitqueue_head(&ctx->cq_wait);
286 init_completion(&ctx->ctx_done);
287 mutex_init(&ctx->uring_lock);
288 init_waitqueue_head(&ctx->wait);
289 for (i = 0; i < ARRAY_SIZE(ctx->pending_async); i++) {
290 spin_lock_init(&ctx->pending_async[i].lock);
291 INIT_LIST_HEAD(&ctx->pending_async[i].list);
292 atomic_set(&ctx->pending_async[i].cnt, 0);
294 spin_lock_init(&ctx->completion_lock);
295 INIT_LIST_HEAD(&ctx->poll_list);
296 INIT_LIST_HEAD(&ctx->cancel_list);
300 static void io_commit_cqring(struct io_ring_ctx *ctx)
302 struct io_cq_ring *ring = ctx->cq_ring;
304 if (ctx->cached_cq_tail != READ_ONCE(ring->r.tail)) {
305 /* order cqe stores with ring update */
306 smp_store_release(&ring->r.tail, ctx->cached_cq_tail);
309 * Write sider barrier of tail update, app has read side. See
310 * comment at the top of this file.
314 if (wq_has_sleeper(&ctx->cq_wait)) {
315 wake_up_interruptible(&ctx->cq_wait);
316 kill_fasync(&ctx->cq_fasync, SIGIO, POLL_IN);
321 static struct io_uring_cqe *io_get_cqring(struct io_ring_ctx *ctx)
323 struct io_cq_ring *ring = ctx->cq_ring;
326 tail = ctx->cached_cq_tail;
327 /* See comment at the top of the file */
329 if (tail + 1 == READ_ONCE(ring->r.head))
332 ctx->cached_cq_tail++;
333 return &ring->cqes[tail & ctx->cq_mask];
336 static void io_cqring_fill_event(struct io_ring_ctx *ctx, u64 ki_user_data,
337 long res, unsigned ev_flags)
339 struct io_uring_cqe *cqe;
342 * If we can't get a cq entry, userspace overflowed the
343 * submission (by quite a lot). Increment the overflow count in
346 cqe = io_get_cqring(ctx);
348 WRITE_ONCE(cqe->user_data, ki_user_data);
349 WRITE_ONCE(cqe->res, res);
350 WRITE_ONCE(cqe->flags, ev_flags);
352 unsigned overflow = READ_ONCE(ctx->cq_ring->overflow);
354 WRITE_ONCE(ctx->cq_ring->overflow, overflow + 1);
358 static void io_cqring_add_event(struct io_ring_ctx *ctx, u64 ki_user_data,
359 long res, unsigned ev_flags)
363 spin_lock_irqsave(&ctx->completion_lock, flags);
364 io_cqring_fill_event(ctx, ki_user_data, res, ev_flags);
365 io_commit_cqring(ctx);
366 spin_unlock_irqrestore(&ctx->completion_lock, flags);
368 if (waitqueue_active(&ctx->wait))
370 if (waitqueue_active(&ctx->sqo_wait))
371 wake_up(&ctx->sqo_wait);
374 static void io_ring_drop_ctx_refs(struct io_ring_ctx *ctx, unsigned refs)
376 percpu_ref_put_many(&ctx->refs, refs);
378 if (waitqueue_active(&ctx->wait))
382 static struct io_kiocb *io_get_req(struct io_ring_ctx *ctx,
383 struct io_submit_state *state)
385 struct io_kiocb *req;
387 if (!percpu_ref_tryget(&ctx->refs))
391 req = kmem_cache_alloc(req_cachep, __GFP_NOWARN);
394 } else if (!state->free_reqs) {
398 sz = min_t(size_t, state->ios_left, ARRAY_SIZE(state->reqs));
399 ret = kmem_cache_alloc_bulk(req_cachep, __GFP_NOWARN, sz,
401 if (unlikely(ret <= 0))
403 state->free_reqs = ret - 1;
405 req = state->reqs[0];
407 req = state->reqs[state->cur_req];
414 /* one is dropped after submission, the other at completion */
415 refcount_set(&req->refs, 2);
418 io_ring_drop_ctx_refs(ctx, 1);
422 static void io_free_req_many(struct io_ring_ctx *ctx, void **reqs, int *nr)
425 kmem_cache_free_bulk(req_cachep, *nr, reqs);
426 io_ring_drop_ctx_refs(ctx, *nr);
431 static void io_free_req(struct io_kiocb *req)
433 io_ring_drop_ctx_refs(req->ctx, 1);
434 kmem_cache_free(req_cachep, req);
437 static void io_put_req(struct io_kiocb *req)
439 if (refcount_dec_and_test(&req->refs))
444 * Find and free completed poll iocbs
446 static void io_iopoll_complete(struct io_ring_ctx *ctx, unsigned int *nr_events,
447 struct list_head *done)
449 void *reqs[IO_IOPOLL_BATCH];
450 int file_count, to_free;
451 struct file *file = NULL;
452 struct io_kiocb *req;
454 file_count = to_free = 0;
455 while (!list_empty(done)) {
456 req = list_first_entry(done, struct io_kiocb, list);
457 list_del(&req->list);
459 io_cqring_fill_event(ctx, req->user_data, req->error, 0);
461 if (refcount_dec_and_test(&req->refs))
462 reqs[to_free++] = req;
466 * Batched puts of the same file, to avoid dirtying the
467 * file usage count multiple times, if avoidable.
469 if (!(req->flags & REQ_F_FIXED_FILE)) {
471 file = req->rw.ki_filp;
473 } else if (file == req->rw.ki_filp) {
476 fput_many(file, file_count);
477 file = req->rw.ki_filp;
482 if (to_free == ARRAY_SIZE(reqs))
483 io_free_req_many(ctx, reqs, &to_free);
485 io_commit_cqring(ctx);
488 fput_many(file, file_count);
489 io_free_req_many(ctx, reqs, &to_free);
492 static int io_do_iopoll(struct io_ring_ctx *ctx, unsigned int *nr_events,
495 struct io_kiocb *req, *tmp;
501 * Only spin for completions if we don't have multiple devices hanging
502 * off our complete list, and we're under the requested amount.
504 spin = !ctx->poll_multi_file && *nr_events < min;
507 list_for_each_entry_safe(req, tmp, &ctx->poll_list, list) {
508 struct kiocb *kiocb = &req->rw;
511 * Move completed entries to our local list. If we find a
512 * request that requires polling, break out and complete
513 * the done list first, if we have entries there.
515 if (req->flags & REQ_F_IOPOLL_COMPLETED) {
516 list_move_tail(&req->list, &done);
519 if (!list_empty(&done))
522 ret = kiocb->ki_filp->f_op->iopoll(kiocb, spin);
531 if (!list_empty(&done))
532 io_iopoll_complete(ctx, nr_events, &done);
538 * Poll for a mininum of 'min' events. Note that if min == 0 we consider that a
539 * non-spinning poll check - we'll still enter the driver poll loop, but only
540 * as a non-spinning completion check.
542 static int io_iopoll_getevents(struct io_ring_ctx *ctx, unsigned int *nr_events,
545 while (!list_empty(&ctx->poll_list)) {
548 ret = io_do_iopoll(ctx, nr_events, min);
551 if (!min || *nr_events >= min)
559 * We can't just wait for polled events to come to us, we have to actively
560 * find and complete them.
562 static void io_iopoll_reap_events(struct io_ring_ctx *ctx)
564 if (!(ctx->flags & IORING_SETUP_IOPOLL))
567 mutex_lock(&ctx->uring_lock);
568 while (!list_empty(&ctx->poll_list)) {
569 unsigned int nr_events = 0;
571 io_iopoll_getevents(ctx, &nr_events, 1);
573 mutex_unlock(&ctx->uring_lock);
576 static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned *nr_events,
584 if (*nr_events < min)
585 tmin = min - *nr_events;
587 ret = io_iopoll_getevents(ctx, nr_events, tmin);
591 } while (min && !*nr_events && !need_resched());
596 static void kiocb_end_write(struct kiocb *kiocb)
598 if (kiocb->ki_flags & IOCB_WRITE) {
599 struct inode *inode = file_inode(kiocb->ki_filp);
602 * Tell lockdep we inherited freeze protection from submission
605 if (S_ISREG(inode->i_mode))
606 __sb_writers_acquired(inode->i_sb, SB_FREEZE_WRITE);
607 file_end_write(kiocb->ki_filp);
611 static void io_fput(struct io_kiocb *req)
613 if (!(req->flags & REQ_F_FIXED_FILE))
614 fput(req->rw.ki_filp);
617 static void io_complete_rw(struct kiocb *kiocb, long res, long res2)
619 struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw);
621 kiocb_end_write(kiocb);
624 io_cqring_add_event(req->ctx, req->user_data, res, 0);
628 static void io_complete_rw_iopoll(struct kiocb *kiocb, long res, long res2)
630 struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw);
632 kiocb_end_write(kiocb);
636 req->flags |= REQ_F_IOPOLL_COMPLETED;
640 * After the iocb has been issued, it's safe to be found on the poll list.
641 * Adding the kiocb to the list AFTER submission ensures that we don't
642 * find it from a io_iopoll_getevents() thread before the issuer is done
643 * accessing the kiocb cookie.
645 static void io_iopoll_req_issued(struct io_kiocb *req)
647 struct io_ring_ctx *ctx = req->ctx;
650 * Track whether we have multiple files in our lists. This will impact
651 * how we do polling eventually, not spinning if we're on potentially
654 if (list_empty(&ctx->poll_list)) {
655 ctx->poll_multi_file = false;
656 } else if (!ctx->poll_multi_file) {
657 struct io_kiocb *list_req;
659 list_req = list_first_entry(&ctx->poll_list, struct io_kiocb,
661 if (list_req->rw.ki_filp != req->rw.ki_filp)
662 ctx->poll_multi_file = true;
666 * For fast devices, IO may have already completed. If it has, add
667 * it to the front so we find it first.
669 if (req->flags & REQ_F_IOPOLL_COMPLETED)
670 list_add(&req->list, &ctx->poll_list);
672 list_add_tail(&req->list, &ctx->poll_list);
675 static void io_file_put(struct io_submit_state *state, struct file *file)
679 } else if (state->file) {
680 int diff = state->has_refs - state->used_refs;
683 fput_many(state->file, diff);
689 * Get as many references to a file as we have IOs left in this submission,
690 * assuming most submissions are for one file, or at least that each file
691 * has more than one submission.
693 static struct file *io_file_get(struct io_submit_state *state, int fd)
699 if (state->fd == fd) {
704 io_file_put(state, NULL);
706 state->file = fget_many(fd, state->ios_left);
711 state->has_refs = state->ios_left;
712 state->used_refs = 1;
718 * If we tracked the file through the SCM inflight mechanism, we could support
719 * any file. For now, just ensure that anything potentially problematic is done
722 static bool io_file_supports_async(struct file *file)
724 umode_t mode = file_inode(file)->i_mode;
726 if (S_ISBLK(mode) || S_ISCHR(mode))
728 if (S_ISREG(mode) && file->f_op != &io_uring_fops)
734 static int io_prep_rw(struct io_kiocb *req, const struct sqe_submit *s,
735 bool force_nonblock, struct io_submit_state *state)
737 const struct io_uring_sqe *sqe = s->sqe;
738 struct io_ring_ctx *ctx = req->ctx;
739 struct kiocb *kiocb = &req->rw;
740 unsigned ioprio, flags;
743 /* For -EAGAIN retry, everything is already prepped */
747 flags = READ_ONCE(sqe->flags);
748 fd = READ_ONCE(sqe->fd);
750 if (flags & IOSQE_FIXED_FILE) {
751 if (unlikely(!ctx->user_files ||
752 (unsigned) fd >= ctx->nr_user_files))
754 kiocb->ki_filp = ctx->user_files[fd];
755 req->flags |= REQ_F_FIXED_FILE;
757 if (s->needs_fixed_file)
759 kiocb->ki_filp = io_file_get(state, fd);
760 if (unlikely(!kiocb->ki_filp))
762 if (force_nonblock && !io_file_supports_async(kiocb->ki_filp))
763 force_nonblock = false;
765 kiocb->ki_pos = READ_ONCE(sqe->off);
766 kiocb->ki_flags = iocb_flags(kiocb->ki_filp);
767 kiocb->ki_hint = ki_hint_validate(file_write_hint(kiocb->ki_filp));
769 ioprio = READ_ONCE(sqe->ioprio);
771 ret = ioprio_check_cap(ioprio);
775 kiocb->ki_ioprio = ioprio;
777 kiocb->ki_ioprio = get_current_ioprio();
779 ret = kiocb_set_rw_flags(kiocb, READ_ONCE(sqe->rw_flags));
782 if (force_nonblock) {
783 kiocb->ki_flags |= IOCB_NOWAIT;
784 req->flags |= REQ_F_FORCE_NONBLOCK;
786 if (ctx->flags & IORING_SETUP_IOPOLL) {
788 if (!(kiocb->ki_flags & IOCB_DIRECT) ||
789 !kiocb->ki_filp->f_op->iopoll)
793 kiocb->ki_flags |= IOCB_HIPRI;
794 kiocb->ki_complete = io_complete_rw_iopoll;
796 if (kiocb->ki_flags & IOCB_HIPRI) {
800 kiocb->ki_complete = io_complete_rw;
804 if (!(flags & IOSQE_FIXED_FILE)) {
806 * in case of error, we didn't use this file reference. drop it.
810 io_file_put(state, kiocb->ki_filp);
815 static inline void io_rw_done(struct kiocb *kiocb, ssize_t ret)
821 case -ERESTARTNOINTR:
822 case -ERESTARTNOHAND:
823 case -ERESTART_RESTARTBLOCK:
825 * We can't just restart the syscall, since previously
826 * submitted sqes may already be in progress. Just fail this
832 kiocb->ki_complete(kiocb, ret, 0);
836 static int io_import_fixed(struct io_ring_ctx *ctx, int rw,
837 const struct io_uring_sqe *sqe,
838 struct iov_iter *iter)
840 size_t len = READ_ONCE(sqe->len);
841 struct io_mapped_ubuf *imu;
842 unsigned index, buf_index;
846 /* attempt to use fixed buffers without having provided iovecs */
847 if (unlikely(!ctx->user_bufs))
850 buf_index = READ_ONCE(sqe->buf_index);
851 if (unlikely(buf_index >= ctx->nr_user_bufs))
854 index = array_index_nospec(buf_index, ctx->nr_user_bufs);
855 imu = &ctx->user_bufs[index];
856 buf_addr = READ_ONCE(sqe->addr);
859 if (buf_addr + len < buf_addr)
861 /* not inside the mapped region */
862 if (buf_addr < imu->ubuf || buf_addr + len > imu->ubuf + imu->len)
866 * May not be a start of buffer, set size appropriately
867 * and advance us to the beginning.
869 offset = buf_addr - imu->ubuf;
870 iov_iter_bvec(iter, rw, imu->bvec, imu->nr_bvecs, offset + len);
872 iov_iter_advance(iter, offset);
876 static int io_import_iovec(struct io_ring_ctx *ctx, int rw,
877 const struct sqe_submit *s, struct iovec **iovec,
878 struct iov_iter *iter)
880 const struct io_uring_sqe *sqe = s->sqe;
881 void __user *buf = u64_to_user_ptr(READ_ONCE(sqe->addr));
882 size_t sqe_len = READ_ONCE(sqe->len);
886 * We're reading ->opcode for the second time, but the first read
887 * doesn't care whether it's _FIXED or not, so it doesn't matter
888 * whether ->opcode changes concurrently. The first read does care
889 * about whether it is a READ or a WRITE, so we don't trust this read
890 * for that purpose and instead let the caller pass in the read/write
893 opcode = READ_ONCE(sqe->opcode);
894 if (opcode == IORING_OP_READ_FIXED ||
895 opcode == IORING_OP_WRITE_FIXED) {
896 int ret = io_import_fixed(ctx, rw, sqe, iter);
906 return compat_import_iovec(rw, buf, sqe_len, UIO_FASTIOV,
910 return import_iovec(rw, buf, sqe_len, UIO_FASTIOV, iovec, iter);
914 * Make a note of the last file/offset/direction we punted to async
915 * context. We'll use this information to see if we can piggy back a
916 * sequential request onto the previous one, if it's still hasn't been
917 * completed by the async worker.
919 static void io_async_list_note(int rw, struct io_kiocb *req, size_t len)
921 struct async_list *async_list = &req->ctx->pending_async[rw];
922 struct kiocb *kiocb = &req->rw;
923 struct file *filp = kiocb->ki_filp;
924 off_t io_end = kiocb->ki_pos + len;
926 if (filp == async_list->file && kiocb->ki_pos == async_list->io_end) {
927 unsigned long max_pages;
929 /* Use 8x RA size as a decent limiter for both reads/writes */
930 max_pages = filp->f_ra.ra_pages;
932 max_pages = VM_MAX_READAHEAD >> (PAGE_SHIFT - 10);
935 /* If max pages are exceeded, reset the state */
937 if (async_list->io_pages + len <= max_pages) {
938 req->flags |= REQ_F_SEQ_PREV;
939 async_list->io_pages += len;
942 async_list->io_pages = 0;
946 /* New file? Reset state. */
947 if (async_list->file != filp) {
948 async_list->io_pages = 0;
949 async_list->file = filp;
951 async_list->io_end = io_end;
954 static int io_read(struct io_kiocb *req, const struct sqe_submit *s,
955 bool force_nonblock, struct io_submit_state *state)
957 struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
958 struct kiocb *kiocb = &req->rw;
959 struct iov_iter iter;
964 ret = io_prep_rw(req, s, force_nonblock, state);
967 file = kiocb->ki_filp;
970 if (unlikely(!(file->f_mode & FMODE_READ)))
973 if (unlikely(!file->f_op->read_iter))
976 ret = io_import_iovec(req->ctx, READ, s, &iovec, &iter);
980 iov_count = iov_iter_count(&iter);
981 ret = rw_verify_area(READ, file, &kiocb->ki_pos, iov_count);
985 /* Catch -EAGAIN return for forced non-blocking submission */
986 ret2 = call_read_iter(file, kiocb, &iter);
987 if (!force_nonblock || ret2 != -EAGAIN) {
988 io_rw_done(kiocb, ret2);
991 * If ->needs_lock is true, we're already in async
995 io_async_list_note(READ, req, iov_count);
1001 /* Hold on to the file for -EAGAIN */
1002 if (unlikely(ret && ret != -EAGAIN))
1007 static int io_write(struct io_kiocb *req, const struct sqe_submit *s,
1008 bool force_nonblock, struct io_submit_state *state)
1010 struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
1011 struct kiocb *kiocb = &req->rw;
1012 struct iov_iter iter;
1017 ret = io_prep_rw(req, s, force_nonblock, state);
1022 file = kiocb->ki_filp;
1023 if (unlikely(!(file->f_mode & FMODE_WRITE)))
1026 if (unlikely(!file->f_op->write_iter))
1029 ret = io_import_iovec(req->ctx, WRITE, s, &iovec, &iter);
1033 iov_count = iov_iter_count(&iter);
1036 if (force_nonblock && !(kiocb->ki_flags & IOCB_DIRECT)) {
1037 /* If ->needs_lock is true, we're already in async context. */
1039 io_async_list_note(WRITE, req, iov_count);
1043 ret = rw_verify_area(WRITE, file, &kiocb->ki_pos, iov_count);
1046 * Open-code file_start_write here to grab freeze protection,
1047 * which will be released by another thread in
1048 * io_complete_rw(). Fool lockdep by telling it the lock got
1049 * released so that it doesn't complain about the held lock when
1050 * we return to userspace.
1052 if (S_ISREG(file_inode(file)->i_mode)) {
1053 __sb_start_write(file_inode(file)->i_sb,
1054 SB_FREEZE_WRITE, true);
1055 __sb_writers_release(file_inode(file)->i_sb,
1058 kiocb->ki_flags |= IOCB_WRITE;
1059 io_rw_done(kiocb, call_write_iter(file, kiocb, &iter));
1064 /* Hold on to the file for -EAGAIN */
1065 if (unlikely(ret && ret != -EAGAIN))
1071 * IORING_OP_NOP just posts a completion event, nothing else.
1073 static int io_nop(struct io_kiocb *req, u64 user_data)
1075 struct io_ring_ctx *ctx = req->ctx;
1078 if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
1082 * Twilight zone - it's possible that someone issued an opcode that
1083 * has a file attached, then got -EAGAIN on submission, and changed
1084 * the sqe before we retried it from async context. Avoid dropping
1085 * a file reference for this malicious case, and flag the error.
1087 if (req->rw.ki_filp) {
1091 io_cqring_add_event(ctx, user_data, err, 0);
1096 static int io_prep_fsync(struct io_kiocb *req, const struct io_uring_sqe *sqe)
1098 struct io_ring_ctx *ctx = req->ctx;
1102 /* Prep already done */
1103 if (req->rw.ki_filp)
1106 if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
1108 if (unlikely(sqe->addr || sqe->ioprio || sqe->buf_index))
1111 fd = READ_ONCE(sqe->fd);
1112 flags = READ_ONCE(sqe->flags);
1114 if (flags & IOSQE_FIXED_FILE) {
1115 if (unlikely(!ctx->user_files || fd >= ctx->nr_user_files))
1117 req->rw.ki_filp = ctx->user_files[fd];
1118 req->flags |= REQ_F_FIXED_FILE;
1120 req->rw.ki_filp = fget(fd);
1121 if (unlikely(!req->rw.ki_filp))
1128 static int io_fsync(struct io_kiocb *req, const struct io_uring_sqe *sqe,
1129 bool force_nonblock)
1131 loff_t sqe_off = READ_ONCE(sqe->off);
1132 loff_t sqe_len = READ_ONCE(sqe->len);
1133 loff_t end = sqe_off + sqe_len;
1134 unsigned fsync_flags;
1137 fsync_flags = READ_ONCE(sqe->fsync_flags);
1138 if (unlikely(fsync_flags & ~IORING_FSYNC_DATASYNC))
1141 ret = io_prep_fsync(req, sqe);
1145 /* fsync always requires a blocking context */
1149 ret = vfs_fsync_range(req->rw.ki_filp, sqe_off,
1150 end > 0 ? end : LLONG_MAX,
1151 fsync_flags & IORING_FSYNC_DATASYNC);
1154 io_cqring_add_event(req->ctx, sqe->user_data, ret, 0);
1159 static void io_poll_remove_one(struct io_kiocb *req)
1161 struct io_poll_iocb *poll = &req->poll;
1163 spin_lock(&poll->head->lock);
1164 WRITE_ONCE(poll->canceled, true);
1165 if (!list_empty(&poll->wait.entry)) {
1166 list_del_init(&poll->wait.entry);
1167 queue_work(req->ctx->sqo_wq, &req->work);
1169 spin_unlock(&poll->head->lock);
1171 list_del_init(&req->list);
1174 static void io_poll_remove_all(struct io_ring_ctx *ctx)
1176 struct io_kiocb *req;
1178 spin_lock_irq(&ctx->completion_lock);
1179 while (!list_empty(&ctx->cancel_list)) {
1180 req = list_first_entry(&ctx->cancel_list, struct io_kiocb,list);
1181 io_poll_remove_one(req);
1183 spin_unlock_irq(&ctx->completion_lock);
1187 * Find a running poll command that matches one specified in sqe->addr,
1188 * and remove it if found.
1190 static int io_poll_remove(struct io_kiocb *req, const struct io_uring_sqe *sqe)
1192 struct io_ring_ctx *ctx = req->ctx;
1193 struct io_kiocb *poll_req, *next;
1196 if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
1198 if (sqe->ioprio || sqe->off || sqe->len || sqe->buf_index ||
1202 spin_lock_irq(&ctx->completion_lock);
1203 list_for_each_entry_safe(poll_req, next, &ctx->cancel_list, list) {
1204 if (READ_ONCE(sqe->addr) == poll_req->user_data) {
1205 io_poll_remove_one(poll_req);
1210 spin_unlock_irq(&ctx->completion_lock);
1212 io_cqring_add_event(req->ctx, sqe->user_data, ret, 0);
1217 static void io_poll_complete(struct io_kiocb *req, __poll_t mask)
1219 io_cqring_add_event(req->ctx, req->user_data, mangle_poll(mask), 0);
1224 static void io_poll_complete_work(struct work_struct *work)
1226 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1227 struct io_poll_iocb *poll = &req->poll;
1228 struct poll_table_struct pt = { ._key = poll->events };
1229 struct io_ring_ctx *ctx = req->ctx;
1232 if (!READ_ONCE(poll->canceled))
1233 mask = vfs_poll(poll->file, &pt) & poll->events;
1236 * Note that ->ki_cancel callers also delete iocb from active_reqs after
1237 * calling ->ki_cancel. We need the ctx_lock roundtrip here to
1238 * synchronize with them. In the cancellation case the list_del_init
1239 * itself is not actually needed, but harmless so we keep it in to
1240 * avoid further branches in the fast path.
1242 spin_lock_irq(&ctx->completion_lock);
1243 if (!mask && !READ_ONCE(poll->canceled)) {
1244 add_wait_queue(poll->head, &poll->wait);
1245 spin_unlock_irq(&ctx->completion_lock);
1248 list_del_init(&req->list);
1249 spin_unlock_irq(&ctx->completion_lock);
1251 io_poll_complete(req, mask);
1254 static int io_poll_wake(struct wait_queue_entry *wait, unsigned mode, int sync,
1257 struct io_poll_iocb *poll = container_of(wait, struct io_poll_iocb,
1259 struct io_kiocb *req = container_of(poll, struct io_kiocb, poll);
1260 struct io_ring_ctx *ctx = req->ctx;
1261 __poll_t mask = key_to_poll(key);
1265 /* for instances that support it check for an event match first: */
1267 unsigned long flags;
1269 if (!(mask & poll->events))
1272 /* try to complete the iocb inline if we can: */
1273 if (spin_trylock_irqsave(&ctx->completion_lock, flags)) {
1274 list_del(&req->list);
1275 spin_unlock_irqrestore(&ctx->completion_lock, flags);
1277 list_del_init(&poll->wait.entry);
1278 io_poll_complete(req, mask);
1283 list_del_init(&poll->wait.entry);
1284 queue_work(ctx->sqo_wq, &req->work);
1288 struct io_poll_table {
1289 struct poll_table_struct pt;
1290 struct io_kiocb *req;
1294 static void io_poll_queue_proc(struct file *file, struct wait_queue_head *head,
1295 struct poll_table_struct *p)
1297 struct io_poll_table *pt = container_of(p, struct io_poll_table, pt);
1299 if (unlikely(pt->req->poll.head)) {
1300 pt->error = -EINVAL;
1305 pt->req->poll.head = head;
1306 add_wait_queue(head, &pt->req->poll.wait);
1309 static int io_poll_add(struct io_kiocb *req, const struct io_uring_sqe *sqe)
1311 struct io_poll_iocb *poll = &req->poll;
1312 struct io_ring_ctx *ctx = req->ctx;
1313 struct io_poll_table ipt;
1319 if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
1321 if (sqe->addr || sqe->ioprio || sqe->off || sqe->len || sqe->buf_index)
1324 INIT_WORK(&req->work, io_poll_complete_work);
1325 events = READ_ONCE(sqe->poll_events);
1326 poll->events = demangle_poll(events) | EPOLLERR | EPOLLHUP;
1328 flags = READ_ONCE(sqe->flags);
1329 fd = READ_ONCE(sqe->fd);
1331 if (flags & IOSQE_FIXED_FILE) {
1332 if (unlikely(!ctx->user_files || fd >= ctx->nr_user_files))
1334 poll->file = ctx->user_files[fd];
1335 req->flags |= REQ_F_FIXED_FILE;
1337 poll->file = fget(fd);
1339 if (unlikely(!poll->file))
1343 poll->woken = false;
1344 poll->canceled = false;
1346 ipt.pt._qproc = io_poll_queue_proc;
1347 ipt.pt._key = poll->events;
1349 ipt.error = -EINVAL; /* same as no support for IOCB_CMD_POLL */
1351 /* initialized the list so that we can do list_empty checks */
1352 INIT_LIST_HEAD(&poll->wait.entry);
1353 init_waitqueue_func_entry(&poll->wait, io_poll_wake);
1355 mask = vfs_poll(poll->file, &ipt.pt) & poll->events;
1356 if (unlikely(!poll->head)) {
1357 /* we did not manage to set up a waitqueue, done */
1361 spin_lock_irq(&ctx->completion_lock);
1362 spin_lock(&poll->head->lock);
1364 /* wake_up context handles the rest */
1367 } else if (mask || ipt.error) {
1368 /* if we get an error or a mask we are done */
1369 WARN_ON_ONCE(list_empty(&poll->wait.entry));
1370 list_del_init(&poll->wait.entry);
1372 /* actually waiting for an event */
1373 list_add_tail(&req->list, &ctx->cancel_list);
1375 spin_unlock(&poll->head->lock);
1376 spin_unlock_irq(&ctx->completion_lock);
1379 if (unlikely(ipt.error)) {
1380 if (!(flags & IOSQE_FIXED_FILE))
1383 * Drop one of our refs to this req, __io_submit_sqe() will
1384 * drop the other one since we're returning an error.
1391 io_poll_complete(req, mask);
1395 static int __io_submit_sqe(struct io_ring_ctx *ctx, struct io_kiocb *req,
1396 const struct sqe_submit *s, bool force_nonblock,
1397 struct io_submit_state *state)
1401 if (unlikely(s->index >= ctx->sq_entries))
1403 req->user_data = READ_ONCE(s->sqe->user_data);
1405 opcode = READ_ONCE(s->sqe->opcode);
1408 ret = io_nop(req, req->user_data);
1410 case IORING_OP_READV:
1411 if (unlikely(s->sqe->buf_index))
1413 ret = io_read(req, s, force_nonblock, state);
1415 case IORING_OP_WRITEV:
1416 if (unlikely(s->sqe->buf_index))
1418 ret = io_write(req, s, force_nonblock, state);
1420 case IORING_OP_READ_FIXED:
1421 ret = io_read(req, s, force_nonblock, state);
1423 case IORING_OP_WRITE_FIXED:
1424 ret = io_write(req, s, force_nonblock, state);
1426 case IORING_OP_FSYNC:
1427 ret = io_fsync(req, s->sqe, force_nonblock);
1429 case IORING_OP_POLL_ADD:
1430 ret = io_poll_add(req, s->sqe);
1432 case IORING_OP_POLL_REMOVE:
1433 ret = io_poll_remove(req, s->sqe);
1443 if (ctx->flags & IORING_SETUP_IOPOLL) {
1444 if (req->error == -EAGAIN)
1447 /* workqueue context doesn't hold uring_lock, grab it now */
1449 mutex_lock(&ctx->uring_lock);
1450 io_iopoll_req_issued(req);
1452 mutex_unlock(&ctx->uring_lock);
1458 static struct async_list *io_async_list_from_sqe(struct io_ring_ctx *ctx,
1459 const struct io_uring_sqe *sqe)
1461 switch (sqe->opcode) {
1462 case IORING_OP_READV:
1463 case IORING_OP_READ_FIXED:
1464 return &ctx->pending_async[READ];
1465 case IORING_OP_WRITEV:
1466 case IORING_OP_WRITE_FIXED:
1467 return &ctx->pending_async[WRITE];
1473 static inline bool io_sqe_needs_user(const struct io_uring_sqe *sqe)
1475 u8 opcode = READ_ONCE(sqe->opcode);
1477 return !(opcode == IORING_OP_READ_FIXED ||
1478 opcode == IORING_OP_WRITE_FIXED);
1481 static void io_sq_wq_submit_work(struct work_struct *work)
1483 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1484 struct io_ring_ctx *ctx = req->ctx;
1485 struct mm_struct *cur_mm = NULL;
1486 struct async_list *async_list;
1487 LIST_HEAD(req_list);
1488 mm_segment_t old_fs;
1491 async_list = io_async_list_from_sqe(ctx, req->submit.sqe);
1494 struct sqe_submit *s = &req->submit;
1495 const struct io_uring_sqe *sqe = s->sqe;
1497 /* Ensure we clear previously set forced non-block flag */
1498 req->flags &= ~REQ_F_FORCE_NONBLOCK;
1499 req->rw.ki_flags &= ~IOCB_NOWAIT;
1502 if (io_sqe_needs_user(sqe) && !cur_mm) {
1503 if (!mmget_not_zero(ctx->sqo_mm)) {
1506 cur_mm = ctx->sqo_mm;
1514 s->has_user = cur_mm != NULL;
1515 s->needs_lock = true;
1517 ret = __io_submit_sqe(ctx, req, s, false, NULL);
1519 * We can get EAGAIN for polled IO even though
1520 * we're forcing a sync submission from here,
1521 * since we can't wait for request slots on the
1529 /* drop submission reference */
1533 io_cqring_add_event(ctx, sqe->user_data, ret, 0);
1537 /* async context always use a copy of the sqe */
1542 if (!list_empty(&req_list)) {
1543 req = list_first_entry(&req_list, struct io_kiocb,
1545 list_del(&req->list);
1548 if (list_empty(&async_list->list))
1552 spin_lock(&async_list->lock);
1553 if (list_empty(&async_list->list)) {
1554 spin_unlock(&async_list->lock);
1557 list_splice_init(&async_list->list, &req_list);
1558 spin_unlock(&async_list->lock);
1560 req = list_first_entry(&req_list, struct io_kiocb, list);
1561 list_del(&req->list);
1565 * Rare case of racing with a submitter. If we find the count has
1566 * dropped to zero AND we have pending work items, then restart
1567 * the processing. This is a tiny race window.
1570 ret = atomic_dec_return(&async_list->cnt);
1571 while (!ret && !list_empty(&async_list->list)) {
1572 spin_lock(&async_list->lock);
1573 atomic_inc(&async_list->cnt);
1574 list_splice_init(&async_list->list, &req_list);
1575 spin_unlock(&async_list->lock);
1577 if (!list_empty(&req_list)) {
1578 req = list_first_entry(&req_list,
1579 struct io_kiocb, list);
1580 list_del(&req->list);
1583 ret = atomic_dec_return(&async_list->cnt);
1595 * See if we can piggy back onto previously submitted work, that is still
1596 * running. We currently only allow this if the new request is sequential
1597 * to the previous one we punted.
1599 static bool io_add_to_prev_work(struct async_list *list, struct io_kiocb *req)
1605 if (!(req->flags & REQ_F_SEQ_PREV))
1607 if (!atomic_read(&list->cnt))
1611 spin_lock(&list->lock);
1612 list_add_tail(&req->list, &list->list);
1613 if (!atomic_read(&list->cnt)) {
1614 list_del_init(&req->list);
1617 spin_unlock(&list->lock);
1621 static int io_submit_sqe(struct io_ring_ctx *ctx, struct sqe_submit *s,
1622 struct io_submit_state *state)
1624 struct io_kiocb *req;
1627 /* enforce forwards compatibility on users */
1628 if (unlikely(s->sqe->flags & ~IOSQE_FIXED_FILE))
1631 req = io_get_req(ctx, state);
1635 req->rw.ki_filp = NULL;
1637 ret = __io_submit_sqe(ctx, req, s, true, state);
1638 if (ret == -EAGAIN) {
1639 struct io_uring_sqe *sqe_copy;
1641 sqe_copy = kmalloc(sizeof(*sqe_copy), GFP_KERNEL);
1643 struct async_list *list;
1645 memcpy(sqe_copy, s->sqe, sizeof(*sqe_copy));
1648 memcpy(&req->submit, s, sizeof(*s));
1649 list = io_async_list_from_sqe(ctx, s->sqe);
1650 if (!io_add_to_prev_work(list, req)) {
1652 atomic_inc(&list->cnt);
1653 INIT_WORK(&req->work, io_sq_wq_submit_work);
1654 queue_work(ctx->sqo_wq, &req->work);
1658 * Queued up for async execution, worker will release
1659 * submit reference when the iocb is actually
1666 /* drop submission reference */
1669 /* and drop final reference, if we failed */
1677 * Batched submission is done, ensure local IO is flushed out.
1679 static void io_submit_state_end(struct io_submit_state *state)
1681 blk_finish_plug(&state->plug);
1682 io_file_put(state, NULL);
1683 if (state->free_reqs)
1684 kmem_cache_free_bulk(req_cachep, state->free_reqs,
1685 &state->reqs[state->cur_req]);
1689 * Start submission side cache.
1691 static void io_submit_state_start(struct io_submit_state *state,
1692 struct io_ring_ctx *ctx, unsigned max_ios)
1694 blk_start_plug(&state->plug);
1695 state->free_reqs = 0;
1697 state->ios_left = max_ios;
1700 static void io_commit_sqring(struct io_ring_ctx *ctx)
1702 struct io_sq_ring *ring = ctx->sq_ring;
1704 if (ctx->cached_sq_head != READ_ONCE(ring->r.head)) {
1706 * Ensure any loads from the SQEs are done at this point,
1707 * since once we write the new head, the application could
1708 * write new data to them.
1710 smp_store_release(&ring->r.head, ctx->cached_sq_head);
1713 * write side barrier of head update, app has read side. See
1714 * comment at the top of this file
1721 * Undo last io_get_sqring()
1723 static void io_drop_sqring(struct io_ring_ctx *ctx)
1725 ctx->cached_sq_head--;
1729 * Fetch an sqe, if one is available. Note that s->sqe will point to memory
1730 * that is mapped by userspace. This means that care needs to be taken to
1731 * ensure that reads are stable, as we cannot rely on userspace always
1732 * being a good citizen. If members of the sqe are validated and then later
1733 * used, it's important that those reads are done through READ_ONCE() to
1734 * prevent a re-load down the line.
1736 static bool io_get_sqring(struct io_ring_ctx *ctx, struct sqe_submit *s)
1738 struct io_sq_ring *ring = ctx->sq_ring;
1742 * The cached sq head (or cq tail) serves two purposes:
1744 * 1) allows us to batch the cost of updating the user visible
1746 * 2) allows the kernel side to track the head on its own, even
1747 * though the application is the one updating it.
1749 head = ctx->cached_sq_head;
1750 /* See comment at the top of this file */
1752 if (head == READ_ONCE(ring->r.tail))
1755 head = READ_ONCE(ring->array[head & ctx->sq_mask]);
1756 if (head < ctx->sq_entries) {
1758 s->sqe = &ctx->sq_sqes[head];
1759 ctx->cached_sq_head++;
1763 /* drop invalid entries */
1764 ctx->cached_sq_head++;
1766 /* See comment at the top of this file */
1771 static int io_submit_sqes(struct io_ring_ctx *ctx, struct sqe_submit *sqes,
1772 unsigned int nr, bool has_user, bool mm_fault)
1774 struct io_submit_state state, *statep = NULL;
1775 int ret, i, submitted = 0;
1777 if (nr > IO_PLUG_THRESHOLD) {
1778 io_submit_state_start(&state, ctx, nr);
1782 for (i = 0; i < nr; i++) {
1783 if (unlikely(mm_fault)) {
1786 sqes[i].has_user = has_user;
1787 sqes[i].needs_lock = true;
1788 sqes[i].needs_fixed_file = true;
1789 ret = io_submit_sqe(ctx, &sqes[i], statep);
1796 io_cqring_add_event(ctx, sqes[i].sqe->user_data, ret, 0);
1800 io_submit_state_end(&state);
1805 static int io_sq_thread(void *data)
1807 struct sqe_submit sqes[IO_IOPOLL_BATCH];
1808 struct io_ring_ctx *ctx = data;
1809 struct mm_struct *cur_mm = NULL;
1810 mm_segment_t old_fs;
1813 unsigned long timeout;
1818 timeout = inflight = 0;
1819 while (!kthread_should_stop() && !ctx->sqo_stop) {
1820 bool all_fixed, mm_fault = false;
1824 unsigned nr_events = 0;
1826 if (ctx->flags & IORING_SETUP_IOPOLL) {
1828 * We disallow the app entering submit/complete
1829 * with polling, but we still need to lock the
1830 * ring to prevent racing with polled issue
1831 * that got punted to a workqueue.
1833 mutex_lock(&ctx->uring_lock);
1834 io_iopoll_check(ctx, &nr_events, 0);
1835 mutex_unlock(&ctx->uring_lock);
1838 * Normal IO, just pretend everything completed.
1839 * We don't have to poll completions for that.
1841 nr_events = inflight;
1844 inflight -= nr_events;
1846 timeout = jiffies + ctx->sq_thread_idle;
1849 if (!io_get_sqring(ctx, &sqes[0])) {
1851 * We're polling. If we're within the defined idle
1852 * period, then let us spin without work before going
1855 if (inflight || !time_after(jiffies, timeout)) {
1861 * Drop cur_mm before scheduling, we can't hold it for
1862 * long periods (or over schedule()). Do this before
1863 * adding ourselves to the waitqueue, as the unuse/drop
1872 prepare_to_wait(&ctx->sqo_wait, &wait,
1873 TASK_INTERRUPTIBLE);
1875 /* Tell userspace we may need a wakeup call */
1876 ctx->sq_ring->flags |= IORING_SQ_NEED_WAKEUP;
1879 if (!io_get_sqring(ctx, &sqes[0])) {
1880 if (kthread_should_stop()) {
1881 finish_wait(&ctx->sqo_wait, &wait);
1884 if (signal_pending(current))
1885 flush_signals(current);
1887 finish_wait(&ctx->sqo_wait, &wait);
1889 ctx->sq_ring->flags &= ~IORING_SQ_NEED_WAKEUP;
1893 finish_wait(&ctx->sqo_wait, &wait);
1895 ctx->sq_ring->flags &= ~IORING_SQ_NEED_WAKEUP;
1902 if (all_fixed && io_sqe_needs_user(sqes[i].sqe))
1906 if (i == ARRAY_SIZE(sqes))
1908 } while (io_get_sqring(ctx, &sqes[i]));
1910 /* Unless all new commands are FIXED regions, grab mm */
1911 if (!all_fixed && !cur_mm) {
1912 mm_fault = !mmget_not_zero(ctx->sqo_mm);
1914 use_mm(ctx->sqo_mm);
1915 cur_mm = ctx->sqo_mm;
1919 inflight += io_submit_sqes(ctx, sqes, i, cur_mm != NULL,
1922 /* Commit SQ ring head once we've consumed all SQEs */
1923 io_commit_sqring(ctx);
1934 static int io_ring_submit(struct io_ring_ctx *ctx, unsigned int to_submit)
1936 struct io_submit_state state, *statep = NULL;
1937 int i, ret = 0, submit = 0;
1939 if (to_submit > IO_PLUG_THRESHOLD) {
1940 io_submit_state_start(&state, ctx, to_submit);
1944 for (i = 0; i < to_submit; i++) {
1945 struct sqe_submit s;
1947 if (!io_get_sqring(ctx, &s))
1951 s.needs_lock = false;
1952 s.needs_fixed_file = false;
1954 ret = io_submit_sqe(ctx, &s, statep);
1956 io_drop_sqring(ctx);
1962 io_commit_sqring(ctx);
1965 io_submit_state_end(statep);
1967 return submit ? submit : ret;
1970 static unsigned io_cqring_events(struct io_cq_ring *ring)
1972 return READ_ONCE(ring->r.tail) - READ_ONCE(ring->r.head);
1976 * Wait until events become available, if we don't already have some. The
1977 * application must reap them itself, as they reside on the shared cq ring.
1979 static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
1980 const sigset_t __user *sig, size_t sigsz)
1982 struct io_cq_ring *ring = ctx->cq_ring;
1983 sigset_t ksigmask, sigsaved;
1987 /* See comment at the top of this file */
1989 if (io_cqring_events(ring) >= min_events)
1993 ret = set_user_sigmask(sig, &ksigmask, &sigsaved, sigsz);
1999 prepare_to_wait(&ctx->wait, &wait, TASK_INTERRUPTIBLE);
2002 /* See comment at the top of this file */
2004 if (io_cqring_events(ring) >= min_events)
2010 if (signal_pending(current))
2014 finish_wait(&ctx->wait, &wait);
2017 restore_user_sigmask(sig, &sigsaved);
2019 return READ_ONCE(ring->r.head) == READ_ONCE(ring->r.tail) ? ret : 0;
2022 static void __io_sqe_files_unregister(struct io_ring_ctx *ctx)
2024 #if defined(CONFIG_UNIX)
2025 if (ctx->ring_sock) {
2026 struct sock *sock = ctx->ring_sock->sk;
2027 struct sk_buff *skb;
2029 while ((skb = skb_dequeue(&sock->sk_receive_queue)) != NULL)
2035 for (i = 0; i < ctx->nr_user_files; i++)
2036 fput(ctx->user_files[i]);
2040 static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
2042 if (!ctx->user_files)
2045 __io_sqe_files_unregister(ctx);
2046 kfree(ctx->user_files);
2047 ctx->user_files = NULL;
2048 ctx->nr_user_files = 0;
2052 static void io_sq_thread_stop(struct io_ring_ctx *ctx)
2054 if (ctx->sqo_thread) {
2057 kthread_stop(ctx->sqo_thread);
2058 ctx->sqo_thread = NULL;
2062 static void io_finish_async(struct io_ring_ctx *ctx)
2064 io_sq_thread_stop(ctx);
2067 destroy_workqueue(ctx->sqo_wq);
2072 #if defined(CONFIG_UNIX)
2073 static void io_destruct_skb(struct sk_buff *skb)
2075 struct io_ring_ctx *ctx = skb->sk->sk_user_data;
2077 io_finish_async(ctx);
2078 unix_destruct_scm(skb);
2082 * Ensure the UNIX gc is aware of our file set, so we are certain that
2083 * the io_uring can be safely unregistered on process exit, even if we have
2084 * loops in the file referencing.
2086 static int __io_sqe_files_scm(struct io_ring_ctx *ctx, int nr, int offset)
2088 struct sock *sk = ctx->ring_sock->sk;
2089 struct scm_fp_list *fpl;
2090 struct sk_buff *skb;
2093 if (!capable(CAP_SYS_RESOURCE) && !capable(CAP_SYS_ADMIN)) {
2094 unsigned long inflight = ctx->user->unix_inflight + nr;
2096 if (inflight > task_rlimit(current, RLIMIT_NOFILE))
2100 fpl = kzalloc(sizeof(*fpl), GFP_KERNEL);
2104 skb = alloc_skb(0, GFP_KERNEL);
2111 skb->destructor = io_destruct_skb;
2113 fpl->user = get_uid(ctx->user);
2114 for (i = 0; i < nr; i++) {
2115 fpl->fp[i] = get_file(ctx->user_files[i + offset]);
2116 unix_inflight(fpl->user, fpl->fp[i]);
2119 fpl->max = fpl->count = nr;
2120 UNIXCB(skb).fp = fpl;
2121 refcount_add(skb->truesize, &sk->sk_wmem_alloc);
2122 skb_queue_head(&sk->sk_receive_queue, skb);
2124 for (i = 0; i < nr; i++)
2131 * If UNIX sockets are enabled, fd passing can cause a reference cycle which
2132 * causes regular reference counting to break down. We rely on the UNIX
2133 * garbage collection to take care of this problem for us.
2135 static int io_sqe_files_scm(struct io_ring_ctx *ctx)
2137 unsigned left, total;
2141 left = ctx->nr_user_files;
2143 unsigned this_files = min_t(unsigned, left, SCM_MAX_FD);
2146 ret = __io_sqe_files_scm(ctx, this_files, total);
2150 total += this_files;
2156 while (total < ctx->nr_user_files) {
2157 fput(ctx->user_files[total]);
2164 static int io_sqe_files_scm(struct io_ring_ctx *ctx)
2170 static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
2173 __s32 __user *fds = (__s32 __user *) arg;
2177 if (ctx->user_files)
2181 if (nr_args > IORING_MAX_FIXED_FILES)
2184 ctx->user_files = kcalloc(nr_args, sizeof(struct file *), GFP_KERNEL);
2185 if (!ctx->user_files)
2188 for (i = 0; i < nr_args; i++) {
2190 if (copy_from_user(&fd, &fds[i], sizeof(fd)))
2193 ctx->user_files[i] = fget(fd);
2196 if (!ctx->user_files[i])
2199 * Don't allow io_uring instances to be registered. If UNIX
2200 * isn't enabled, then this causes a reference cycle and this
2201 * instance can never get freed. If UNIX is enabled we'll
2202 * handle it just fine, but there's still no point in allowing
2203 * a ring fd as it doesn't support regular read/write anyway.
2205 if (ctx->user_files[i]->f_op == &io_uring_fops) {
2206 fput(ctx->user_files[i]);
2209 ctx->nr_user_files++;
2214 for (i = 0; i < ctx->nr_user_files; i++)
2215 fput(ctx->user_files[i]);
2217 kfree(ctx->user_files);
2218 ctx->nr_user_files = 0;
2222 ret = io_sqe_files_scm(ctx);
2224 io_sqe_files_unregister(ctx);
2229 static int io_sq_offload_start(struct io_ring_ctx *ctx,
2230 struct io_uring_params *p)
2234 init_waitqueue_head(&ctx->sqo_wait);
2235 mmgrab(current->mm);
2236 ctx->sqo_mm = current->mm;
2238 ctx->sq_thread_idle = msecs_to_jiffies(p->sq_thread_idle);
2239 if (!ctx->sq_thread_idle)
2240 ctx->sq_thread_idle = HZ;
2243 if (!cpu_possible(p->sq_thread_cpu))
2246 if (ctx->flags & IORING_SETUP_SQPOLL) {
2247 if (p->flags & IORING_SETUP_SQ_AFF) {
2250 cpu = array_index_nospec(p->sq_thread_cpu, NR_CPUS);
2251 ctx->sqo_thread = kthread_create_on_cpu(io_sq_thread,
2255 ctx->sqo_thread = kthread_create(io_sq_thread, ctx,
2258 if (IS_ERR(ctx->sqo_thread)) {
2259 ret = PTR_ERR(ctx->sqo_thread);
2260 ctx->sqo_thread = NULL;
2263 wake_up_process(ctx->sqo_thread);
2264 } else if (p->flags & IORING_SETUP_SQ_AFF) {
2265 /* Can't have SQ_AFF without SQPOLL */
2270 /* Do QD, or 2 * CPUS, whatever is smallest */
2271 ctx->sqo_wq = alloc_workqueue("io_ring-wq", WQ_UNBOUND | WQ_FREEZABLE,
2272 min(ctx->sq_entries - 1, 2 * num_online_cpus()));
2280 io_sq_thread_stop(ctx);
2281 mmdrop(ctx->sqo_mm);
2286 static void io_unaccount_mem(struct user_struct *user, unsigned long nr_pages)
2288 atomic_long_sub(nr_pages, &user->locked_vm);
2291 static int io_account_mem(struct user_struct *user, unsigned long nr_pages)
2293 unsigned long page_limit, cur_pages, new_pages;
2295 /* Don't allow more pages than we can safely lock */
2296 page_limit = rlimit(RLIMIT_MEMLOCK) >> PAGE_SHIFT;
2299 cur_pages = atomic_long_read(&user->locked_vm);
2300 new_pages = cur_pages + nr_pages;
2301 if (new_pages > page_limit)
2303 } while (atomic_long_cmpxchg(&user->locked_vm, cur_pages,
2304 new_pages) != cur_pages);
2309 static void io_mem_free(void *ptr)
2311 struct page *page = virt_to_head_page(ptr);
2313 if (put_page_testzero(page))
2314 free_compound_page(page);
2317 static void *io_mem_alloc(size_t size)
2319 gfp_t gfp_flags = GFP_KERNEL | __GFP_ZERO | __GFP_NOWARN | __GFP_COMP |
2322 return (void *) __get_free_pages(gfp_flags, get_order(size));
2325 static unsigned long ring_pages(unsigned sq_entries, unsigned cq_entries)
2327 struct io_sq_ring *sq_ring;
2328 struct io_cq_ring *cq_ring;
2331 bytes = struct_size(sq_ring, array, sq_entries);
2332 bytes += array_size(sizeof(struct io_uring_sqe), sq_entries);
2333 bytes += struct_size(cq_ring, cqes, cq_entries);
2335 return (bytes + PAGE_SIZE - 1) / PAGE_SIZE;
2338 static int io_sqe_buffer_unregister(struct io_ring_ctx *ctx)
2342 if (!ctx->user_bufs)
2345 for (i = 0; i < ctx->nr_user_bufs; i++) {
2346 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
2348 for (j = 0; j < imu->nr_bvecs; j++)
2349 put_page(imu->bvec[j].bv_page);
2351 if (ctx->account_mem)
2352 io_unaccount_mem(ctx->user, imu->nr_bvecs);
2357 kfree(ctx->user_bufs);
2358 ctx->user_bufs = NULL;
2359 ctx->nr_user_bufs = 0;
2363 static int io_copy_iov(struct io_ring_ctx *ctx, struct iovec *dst,
2364 void __user *arg, unsigned index)
2366 struct iovec __user *src;
2368 #ifdef CONFIG_COMPAT
2370 struct compat_iovec __user *ciovs;
2371 struct compat_iovec ciov;
2373 ciovs = (struct compat_iovec __user *) arg;
2374 if (copy_from_user(&ciov, &ciovs[index], sizeof(ciov)))
2377 dst->iov_base = (void __user *) (unsigned long) ciov.iov_base;
2378 dst->iov_len = ciov.iov_len;
2382 src = (struct iovec __user *) arg;
2383 if (copy_from_user(dst, &src[index], sizeof(*dst)))
2388 static int io_sqe_buffer_register(struct io_ring_ctx *ctx, void __user *arg,
2391 struct vm_area_struct **vmas = NULL;
2392 struct page **pages = NULL;
2393 int i, j, got_pages = 0;
2398 if (!nr_args || nr_args > UIO_MAXIOV)
2401 ctx->user_bufs = kcalloc(nr_args, sizeof(struct io_mapped_ubuf),
2403 if (!ctx->user_bufs)
2406 for (i = 0; i < nr_args; i++) {
2407 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
2408 unsigned long off, start, end, ubuf;
2413 ret = io_copy_iov(ctx, &iov, arg, i);
2418 * Don't impose further limits on the size and buffer
2419 * constraints here, we'll -EINVAL later when IO is
2420 * submitted if they are wrong.
2423 if (!iov.iov_base || !iov.iov_len)
2426 /* arbitrary limit, but we need something */
2427 if (iov.iov_len > SZ_1G)
2430 ubuf = (unsigned long) iov.iov_base;
2431 end = (ubuf + iov.iov_len + PAGE_SIZE - 1) >> PAGE_SHIFT;
2432 start = ubuf >> PAGE_SHIFT;
2433 nr_pages = end - start;
2435 if (ctx->account_mem) {
2436 ret = io_account_mem(ctx->user, nr_pages);
2442 if (!pages || nr_pages > got_pages) {
2445 pages = kmalloc_array(nr_pages, sizeof(struct page *),
2447 vmas = kmalloc_array(nr_pages,
2448 sizeof(struct vm_area_struct *),
2450 if (!pages || !vmas) {
2452 if (ctx->account_mem)
2453 io_unaccount_mem(ctx->user, nr_pages);
2456 got_pages = nr_pages;
2459 imu->bvec = kmalloc_array(nr_pages, sizeof(struct bio_vec),
2463 if (ctx->account_mem)
2464 io_unaccount_mem(ctx->user, nr_pages);
2469 down_read(¤t->mm->mmap_sem);
2470 pret = get_user_pages_longterm(ubuf, nr_pages, FOLL_WRITE,
2472 if (pret == nr_pages) {
2473 /* don't support file backed memory */
2474 for (j = 0; j < nr_pages; j++) {
2475 struct vm_area_struct *vma = vmas[j];
2478 !is_file_hugepages(vma->vm_file)) {
2484 ret = pret < 0 ? pret : -EFAULT;
2486 up_read(¤t->mm->mmap_sem);
2489 * if we did partial map, or found file backed vmas,
2490 * release any pages we did get
2493 for (j = 0; j < pret; j++)
2496 if (ctx->account_mem)
2497 io_unaccount_mem(ctx->user, nr_pages);
2501 off = ubuf & ~PAGE_MASK;
2503 for (j = 0; j < nr_pages; j++) {
2506 vec_len = min_t(size_t, size, PAGE_SIZE - off);
2507 imu->bvec[j].bv_page = pages[j];
2508 imu->bvec[j].bv_len = vec_len;
2509 imu->bvec[j].bv_offset = off;
2513 /* store original address for later verification */
2515 imu->len = iov.iov_len;
2516 imu->nr_bvecs = nr_pages;
2518 ctx->nr_user_bufs++;
2526 io_sqe_buffer_unregister(ctx);
2530 static void io_ring_ctx_free(struct io_ring_ctx *ctx)
2532 io_finish_async(ctx);
2534 mmdrop(ctx->sqo_mm);
2536 io_iopoll_reap_events(ctx);
2537 io_sqe_buffer_unregister(ctx);
2538 io_sqe_files_unregister(ctx);
2540 #if defined(CONFIG_UNIX)
2542 sock_release(ctx->ring_sock);
2545 io_mem_free(ctx->sq_ring);
2546 io_mem_free(ctx->sq_sqes);
2547 io_mem_free(ctx->cq_ring);
2549 percpu_ref_exit(&ctx->refs);
2550 if (ctx->account_mem)
2551 io_unaccount_mem(ctx->user,
2552 ring_pages(ctx->sq_entries, ctx->cq_entries));
2553 free_uid(ctx->user);
2557 static __poll_t io_uring_poll(struct file *file, poll_table *wait)
2559 struct io_ring_ctx *ctx = file->private_data;
2562 poll_wait(file, &ctx->cq_wait, wait);
2563 /* See comment at the top of this file */
2565 if (READ_ONCE(ctx->sq_ring->r.tail) + 1 != ctx->cached_sq_head)
2566 mask |= EPOLLOUT | EPOLLWRNORM;
2567 if (READ_ONCE(ctx->cq_ring->r.head) != ctx->cached_cq_tail)
2568 mask |= EPOLLIN | EPOLLRDNORM;
2573 static int io_uring_fasync(int fd, struct file *file, int on)
2575 struct io_ring_ctx *ctx = file->private_data;
2577 return fasync_helper(fd, file, on, &ctx->cq_fasync);
2580 static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
2582 mutex_lock(&ctx->uring_lock);
2583 percpu_ref_kill(&ctx->refs);
2584 mutex_unlock(&ctx->uring_lock);
2586 io_poll_remove_all(ctx);
2587 io_iopoll_reap_events(ctx);
2588 wait_for_completion(&ctx->ctx_done);
2589 io_ring_ctx_free(ctx);
2592 static int io_uring_release(struct inode *inode, struct file *file)
2594 struct io_ring_ctx *ctx = file->private_data;
2596 file->private_data = NULL;
2597 io_ring_ctx_wait_and_kill(ctx);
2601 static int io_uring_mmap(struct file *file, struct vm_area_struct *vma)
2603 loff_t offset = (loff_t) vma->vm_pgoff << PAGE_SHIFT;
2604 unsigned long sz = vma->vm_end - vma->vm_start;
2605 struct io_ring_ctx *ctx = file->private_data;
2611 case IORING_OFF_SQ_RING:
2614 case IORING_OFF_SQES:
2617 case IORING_OFF_CQ_RING:
2624 page = virt_to_head_page(ptr);
2625 if (sz > (PAGE_SIZE << compound_order(page)))
2628 pfn = virt_to_phys(ptr) >> PAGE_SHIFT;
2629 return remap_pfn_range(vma, vma->vm_start, pfn, sz, vma->vm_page_prot);
2632 SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
2633 u32, min_complete, u32, flags, const sigset_t __user *, sig,
2636 struct io_ring_ctx *ctx;
2641 if (flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP))
2649 if (f.file->f_op != &io_uring_fops)
2653 ctx = f.file->private_data;
2654 if (!percpu_ref_tryget(&ctx->refs))
2658 * For SQ polling, the thread will do all submissions and completions.
2659 * Just return the requested submit count, and wake the thread if
2662 if (ctx->flags & IORING_SETUP_SQPOLL) {
2663 if (flags & IORING_ENTER_SQ_WAKEUP)
2664 wake_up(&ctx->sqo_wait);
2665 submitted = to_submit;
2671 to_submit = min(to_submit, ctx->sq_entries);
2673 mutex_lock(&ctx->uring_lock);
2674 submitted = io_ring_submit(ctx, to_submit);
2675 mutex_unlock(&ctx->uring_lock);
2680 if (flags & IORING_ENTER_GETEVENTS) {
2681 unsigned nr_events = 0;
2683 min_complete = min(min_complete, ctx->cq_entries);
2686 * The application could have included the 'to_submit' count
2687 * in how many events it wanted to wait for. If we failed to
2688 * submit the desired count, we may need to adjust the number
2689 * of events to poll/wait for.
2691 if (submitted < to_submit)
2692 min_complete = min_t(unsigned, submitted, min_complete);
2694 if (ctx->flags & IORING_SETUP_IOPOLL) {
2695 mutex_lock(&ctx->uring_lock);
2696 ret = io_iopoll_check(ctx, &nr_events, min_complete);
2697 mutex_unlock(&ctx->uring_lock);
2699 ret = io_cqring_wait(ctx, min_complete, sig, sigsz);
2704 io_ring_drop_ctx_refs(ctx, 1);
2707 return submitted ? submitted : ret;
2710 static const struct file_operations io_uring_fops = {
2711 .release = io_uring_release,
2712 .mmap = io_uring_mmap,
2713 .poll = io_uring_poll,
2714 .fasync = io_uring_fasync,
2717 static int io_allocate_scq_urings(struct io_ring_ctx *ctx,
2718 struct io_uring_params *p)
2720 struct io_sq_ring *sq_ring;
2721 struct io_cq_ring *cq_ring;
2724 sq_ring = io_mem_alloc(struct_size(sq_ring, array, p->sq_entries));
2728 ctx->sq_ring = sq_ring;
2729 sq_ring->ring_mask = p->sq_entries - 1;
2730 sq_ring->ring_entries = p->sq_entries;
2731 ctx->sq_mask = sq_ring->ring_mask;
2732 ctx->sq_entries = sq_ring->ring_entries;
2734 size = array_size(sizeof(struct io_uring_sqe), p->sq_entries);
2735 if (size == SIZE_MAX)
2738 ctx->sq_sqes = io_mem_alloc(size);
2739 if (!ctx->sq_sqes) {
2740 io_mem_free(ctx->sq_ring);
2744 cq_ring = io_mem_alloc(struct_size(cq_ring, cqes, p->cq_entries));
2746 io_mem_free(ctx->sq_ring);
2747 io_mem_free(ctx->sq_sqes);
2751 ctx->cq_ring = cq_ring;
2752 cq_ring->ring_mask = p->cq_entries - 1;
2753 cq_ring->ring_entries = p->cq_entries;
2754 ctx->cq_mask = cq_ring->ring_mask;
2755 ctx->cq_entries = cq_ring->ring_entries;
2760 * Allocate an anonymous fd, this is what constitutes the application
2761 * visible backing of an io_uring instance. The application mmaps this
2762 * fd to gain access to the SQ/CQ ring details. If UNIX sockets are enabled,
2763 * we have to tie this fd to a socket for file garbage collection purposes.
2765 static int io_uring_get_fd(struct io_ring_ctx *ctx)
2770 #if defined(CONFIG_UNIX)
2771 ret = sock_create_kern(&init_net, PF_UNIX, SOCK_RAW, IPPROTO_IP,
2777 ret = get_unused_fd_flags(O_RDWR | O_CLOEXEC);
2781 file = anon_inode_getfile("[io_uring]", &io_uring_fops, ctx,
2782 O_RDWR | O_CLOEXEC);
2785 ret = PTR_ERR(file);
2789 #if defined(CONFIG_UNIX)
2790 ctx->ring_sock->file = file;
2791 ctx->ring_sock->sk->sk_user_data = ctx;
2793 fd_install(ret, file);
2796 #if defined(CONFIG_UNIX)
2797 sock_release(ctx->ring_sock);
2798 ctx->ring_sock = NULL;
2803 static int io_uring_create(unsigned entries, struct io_uring_params *p)
2805 struct user_struct *user = NULL;
2806 struct io_ring_ctx *ctx;
2810 if (!entries || entries > IORING_MAX_ENTRIES)
2814 * Use twice as many entries for the CQ ring. It's possible for the
2815 * application to drive a higher depth than the size of the SQ ring,
2816 * since the sqes are only used at submission time. This allows for
2817 * some flexibility in overcommitting a bit.
2819 p->sq_entries = roundup_pow_of_two(entries);
2820 p->cq_entries = 2 * p->sq_entries;
2822 user = get_uid(current_user());
2823 account_mem = !capable(CAP_IPC_LOCK);
2826 ret = io_account_mem(user,
2827 ring_pages(p->sq_entries, p->cq_entries));
2834 ctx = io_ring_ctx_alloc(p);
2837 io_unaccount_mem(user, ring_pages(p->sq_entries,
2842 ctx->compat = in_compat_syscall();
2843 ctx->account_mem = account_mem;
2846 ret = io_allocate_scq_urings(ctx, p);
2850 ret = io_sq_offload_start(ctx, p);
2854 ret = io_uring_get_fd(ctx);
2858 memset(&p->sq_off, 0, sizeof(p->sq_off));
2859 p->sq_off.head = offsetof(struct io_sq_ring, r.head);
2860 p->sq_off.tail = offsetof(struct io_sq_ring, r.tail);
2861 p->sq_off.ring_mask = offsetof(struct io_sq_ring, ring_mask);
2862 p->sq_off.ring_entries = offsetof(struct io_sq_ring, ring_entries);
2863 p->sq_off.flags = offsetof(struct io_sq_ring, flags);
2864 p->sq_off.dropped = offsetof(struct io_sq_ring, dropped);
2865 p->sq_off.array = offsetof(struct io_sq_ring, array);
2867 memset(&p->cq_off, 0, sizeof(p->cq_off));
2868 p->cq_off.head = offsetof(struct io_cq_ring, r.head);
2869 p->cq_off.tail = offsetof(struct io_cq_ring, r.tail);
2870 p->cq_off.ring_mask = offsetof(struct io_cq_ring, ring_mask);
2871 p->cq_off.ring_entries = offsetof(struct io_cq_ring, ring_entries);
2872 p->cq_off.overflow = offsetof(struct io_cq_ring, overflow);
2873 p->cq_off.cqes = offsetof(struct io_cq_ring, cqes);
2876 io_ring_ctx_wait_and_kill(ctx);
2881 * Sets up an aio uring context, and returns the fd. Applications asks for a
2882 * ring size, we return the actual sq/cq ring sizes (among other things) in the
2883 * params structure passed in.
2885 static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
2887 struct io_uring_params p;
2891 if (copy_from_user(&p, params, sizeof(p)))
2893 for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
2898 if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
2899 IORING_SETUP_SQ_AFF))
2902 ret = io_uring_create(entries, &p);
2906 if (copy_to_user(params, &p, sizeof(p)))
2912 SYSCALL_DEFINE2(io_uring_setup, u32, entries,
2913 struct io_uring_params __user *, params)
2915 return io_uring_setup(entries, params);
2918 static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
2919 void __user *arg, unsigned nr_args)
2923 percpu_ref_kill(&ctx->refs);
2924 wait_for_completion(&ctx->ctx_done);
2927 case IORING_REGISTER_BUFFERS:
2928 ret = io_sqe_buffer_register(ctx, arg, nr_args);
2930 case IORING_UNREGISTER_BUFFERS:
2934 ret = io_sqe_buffer_unregister(ctx);
2936 case IORING_REGISTER_FILES:
2937 ret = io_sqe_files_register(ctx, arg, nr_args);
2939 case IORING_UNREGISTER_FILES:
2943 ret = io_sqe_files_unregister(ctx);
2950 /* bring the ctx back to life */
2951 reinit_completion(&ctx->ctx_done);
2952 percpu_ref_reinit(&ctx->refs);
2956 SYSCALL_DEFINE4(io_uring_register, unsigned int, fd, unsigned int, opcode,
2957 void __user *, arg, unsigned int, nr_args)
2959 struct io_ring_ctx *ctx;
2968 if (f.file->f_op != &io_uring_fops)
2971 ctx = f.file->private_data;
2973 mutex_lock(&ctx->uring_lock);
2974 ret = __io_uring_register(ctx, opcode, arg, nr_args);
2975 mutex_unlock(&ctx->uring_lock);
2981 static int __init io_uring_init(void)
2983 req_cachep = KMEM_CACHE(io_kiocb, SLAB_HWCACHE_ALIGN | SLAB_PANIC);
2986 __initcall(io_uring_init);