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.
9 * After the application reads the CQ ring tail, it must use an
10 * appropriate smp_rmb() to pair with the smp_wmb() the kernel uses
11 * before writing the tail (using smp_load_acquire to read the tail will
12 * do). It also needs a smp_mb() before updating CQ head (ordering the
13 * entry load(s) with the head store), pairing with an implicit barrier
14 * through a control-dependency in io_get_cqring (smp_store_release to
15 * store head will do). Failure to do so could lead to reading invalid
18 * Likewise, the application must use an appropriate smp_wmb() before
19 * writing the SQ tail (ordering SQ entry stores with the tail store),
20 * which pairs with smp_load_acquire in io_get_sqring (smp_store_release
21 * to store the tail will do). And it needs a barrier ordering the SQ
22 * head load before writing new SQ entries (smp_load_acquire to read
25 * When using the SQ poll thread (IORING_SETUP_SQPOLL), the application
26 * needs to check the SQ flags for IORING_SQ_NEED_WAKEUP *after*
27 * updating the SQ tail; a full memory barrier smp_mb() is needed
30 * Also see the examples in the liburing library:
32 * git://git.kernel.dk/liburing
34 * io_uring also uses READ/WRITE_ONCE() for _any_ store or load that happens
35 * from data shared between the kernel and application. This is done both
36 * for ordering purposes, but also to ensure that once a value is loaded from
37 * data that the application could potentially modify, it remains stable.
39 * Copyright (C) 2018-2019 Jens Axboe
40 * Copyright (c) 2018-2019 Christoph Hellwig
42 #include <linux/kernel.h>
43 #include <linux/init.h>
44 #include <linux/errno.h>
45 #include <linux/syscalls.h>
46 #include <linux/compat.h>
47 #include <linux/refcount.h>
48 #include <linux/uio.h>
50 #include <linux/sched/signal.h>
52 #include <linux/file.h>
53 #include <linux/fdtable.h>
55 #include <linux/mman.h>
56 #include <linux/mmu_context.h>
57 #include <linux/percpu.h>
58 #include <linux/slab.h>
59 #include <linux/workqueue.h>
60 #include <linux/kthread.h>
61 #include <linux/blkdev.h>
62 #include <linux/bvec.h>
63 #include <linux/net.h>
65 #include <net/af_unix.h>
67 #include <linux/anon_inodes.h>
68 #include <linux/sched/mm.h>
69 #include <linux/uaccess.h>
70 #include <linux/nospec.h>
71 #include <linux/sizes.h>
72 #include <linux/hugetlb.h>
74 #include <uapi/linux/io_uring.h>
78 #define IORING_MAX_ENTRIES 32768
79 #define IORING_MAX_FIXED_FILES 1024
82 u32 head ____cacheline_aligned_in_smp
;
83 u32 tail ____cacheline_aligned_in_smp
;
87 * This data is shared with the application through the mmap at offsets
88 * IORING_OFF_SQ_RING and IORING_OFF_CQ_RING.
90 * The offsets to the member fields are published through struct
91 * io_sqring_offsets when calling io_uring_setup.
95 * Head and tail offsets into the ring; the offsets need to be
96 * masked to get valid indices.
98 * The kernel controls head of the sq ring and the tail of the cq ring,
99 * and the application controls tail of the sq ring and the head of the
102 struct io_uring sq
, cq
;
104 * Bitmasks to apply to head and tail offsets (constant, equals
107 u32 sq_ring_mask
, cq_ring_mask
;
108 /* Ring sizes (constant, power of 2) */
109 u32 sq_ring_entries
, cq_ring_entries
;
111 * Number of invalid entries dropped by the kernel due to
112 * invalid index stored in array
114 * Written by the kernel, shouldn't be modified by the
115 * application (i.e. get number of "new events" by comparing to
118 * After a new SQ head value was read by the application this
119 * counter includes all submissions that were dropped reaching
120 * the new SQ head (and possibly more).
126 * Written by the kernel, shouldn't be modified by the
129 * The application needs a full memory barrier before checking
130 * for IORING_SQ_NEED_WAKEUP after updating the sq tail.
134 * Number of completion events lost because the queue was full;
135 * this should be avoided by the application by making sure
136 * there are not more requests pending thatn there is space in
137 * the completion queue.
139 * Written by the kernel, shouldn't be modified by the
140 * application (i.e. get number of "new events" by comparing to
143 * As completion events come in out of order this counter is not
144 * ordered with any other data.
148 * Ring buffer of completion events.
150 * The kernel writes completion events fresh every time they are
151 * produced, so the application is allowed to modify pending
154 struct io_uring_cqe cqes
[] ____cacheline_aligned_in_smp
;
157 struct io_mapped_ubuf
{
160 struct bio_vec
*bvec
;
161 unsigned int nr_bvecs
;
167 struct list_head list
;
176 struct percpu_ref refs
;
177 } ____cacheline_aligned_in_smp
;
185 * Ring buffer of indices into array of io_uring_sqe, which is
186 * mmapped by the application using the IORING_OFF_SQES offset.
188 * This indirection could e.g. be used to assign fixed
189 * io_uring_sqe entries to operations and only submit them to
190 * the queue when needed.
192 * The kernel modifies neither the indices array nor the entries
196 unsigned cached_sq_head
;
199 unsigned sq_thread_idle
;
200 struct io_uring_sqe
*sq_sqes
;
202 struct list_head defer_list
;
203 } ____cacheline_aligned_in_smp
;
206 struct workqueue_struct
*sqo_wq
[2];
207 struct task_struct
*sqo_thread
; /* if using sq thread polling */
208 struct mm_struct
*sqo_mm
;
209 wait_queue_head_t sqo_wait
;
210 struct completion sqo_thread_started
;
213 unsigned cached_cq_tail
;
216 struct wait_queue_head cq_wait
;
217 struct fasync_struct
*cq_fasync
;
218 struct eventfd_ctx
*cq_ev_fd
;
219 } ____cacheline_aligned_in_smp
;
221 struct io_rings
*rings
;
224 * If used, fixed file set. Writers must ensure that ->refs is dead,
225 * readers must ensure that ->refs is alive as long as the file* is
226 * used. Only updated through io_uring_register(2).
228 struct file
**user_files
;
229 unsigned nr_user_files
;
231 /* if used, fixed mapped user buffers */
232 unsigned nr_user_bufs
;
233 struct io_mapped_ubuf
*user_bufs
;
235 struct user_struct
*user
;
237 struct completion ctx_done
;
240 struct mutex uring_lock
;
241 wait_queue_head_t wait
;
242 } ____cacheline_aligned_in_smp
;
245 spinlock_t completion_lock
;
246 bool poll_multi_file
;
248 * ->poll_list is protected by the ctx->uring_lock for
249 * io_uring instances that don't use IORING_SETUP_SQPOLL.
250 * For SQPOLL, only the single threaded io_sq_thread() will
251 * manipulate the list, hence no extra locking is needed there.
253 struct list_head poll_list
;
254 struct list_head cancel_list
;
255 } ____cacheline_aligned_in_smp
;
257 struct async_list pending_async
[2];
259 #if defined(CONFIG_UNIX)
260 struct socket
*ring_sock
;
265 const struct io_uring_sqe
*sqe
;
266 unsigned short index
;
270 bool needs_fixed_file
;
274 * First field must be the file pointer in all the
275 * iocb unions! See also 'struct kiocb' in <linux/fs.h>
277 struct io_poll_iocb
{
279 struct wait_queue_head
*head
;
283 struct wait_queue_entry wait
;
287 * NOTE! Each of the iocb union members has the file pointer
288 * as the first entry in their struct definition. So you can
289 * access the file pointer through any of the sub-structs,
290 * or directly as just 'ki_filp' in this struct.
296 struct io_poll_iocb poll
;
299 struct sqe_submit submit
;
301 struct io_ring_ctx
*ctx
;
302 struct list_head list
;
303 struct list_head link_list
;
306 #define REQ_F_NOWAIT 1 /* must not punt to workers */
307 #define REQ_F_IOPOLL_COMPLETED 2 /* polled IO has completed */
308 #define REQ_F_FIXED_FILE 4 /* ctx owns file */
309 #define REQ_F_SEQ_PREV 8 /* sequential with previous */
310 #define REQ_F_IO_DRAIN 16 /* drain existing IO first */
311 #define REQ_F_IO_DRAINED 32 /* drain done */
312 #define REQ_F_LINK 64 /* linked sqes */
313 #define REQ_F_LINK_DONE 128 /* linked sqes done */
314 #define REQ_F_FAIL_LINK 256 /* fail rest of links */
315 #define REQ_F_SHADOW_DRAIN 512 /* link-drain shadow req */
320 struct work_struct work
;
323 #define IO_PLUG_THRESHOLD 2
324 #define IO_IOPOLL_BATCH 8
326 struct io_submit_state
{
327 struct blk_plug plug
;
330 * io_kiocb alloc cache
332 void *reqs
[IO_IOPOLL_BATCH
];
333 unsigned int free_reqs
;
334 unsigned int cur_req
;
337 * File reference cache
341 unsigned int has_refs
;
342 unsigned int used_refs
;
343 unsigned int ios_left
;
346 static void io_sq_wq_submit_work(struct work_struct
*work
);
347 static void __io_free_req(struct io_kiocb
*req
);
349 static struct kmem_cache
*req_cachep
;
351 static const struct file_operations io_uring_fops
;
353 struct sock
*io_uring_get_socket(struct file
*file
)
355 #if defined(CONFIG_UNIX)
356 if (file
->f_op
== &io_uring_fops
) {
357 struct io_ring_ctx
*ctx
= file
->private_data
;
359 return ctx
->ring_sock
->sk
;
364 EXPORT_SYMBOL(io_uring_get_socket
);
366 static void io_ring_ctx_ref_free(struct percpu_ref
*ref
)
368 struct io_ring_ctx
*ctx
= container_of(ref
, struct io_ring_ctx
, refs
);
370 complete(&ctx
->ctx_done
);
373 static struct io_ring_ctx
*io_ring_ctx_alloc(struct io_uring_params
*p
)
375 struct io_ring_ctx
*ctx
;
378 ctx
= kzalloc(sizeof(*ctx
), GFP_KERNEL
);
382 if (percpu_ref_init(&ctx
->refs
, io_ring_ctx_ref_free
,
383 PERCPU_REF_ALLOW_REINIT
, GFP_KERNEL
)) {
388 ctx
->flags
= p
->flags
;
389 init_waitqueue_head(&ctx
->cq_wait
);
390 init_completion(&ctx
->ctx_done
);
391 init_completion(&ctx
->sqo_thread_started
);
392 mutex_init(&ctx
->uring_lock
);
393 init_waitqueue_head(&ctx
->wait
);
394 for (i
= 0; i
< ARRAY_SIZE(ctx
->pending_async
); i
++) {
395 spin_lock_init(&ctx
->pending_async
[i
].lock
);
396 INIT_LIST_HEAD(&ctx
->pending_async
[i
].list
);
397 atomic_set(&ctx
->pending_async
[i
].cnt
, 0);
399 spin_lock_init(&ctx
->completion_lock
);
400 INIT_LIST_HEAD(&ctx
->poll_list
);
401 INIT_LIST_HEAD(&ctx
->cancel_list
);
402 INIT_LIST_HEAD(&ctx
->defer_list
);
406 static inline bool io_sequence_defer(struct io_ring_ctx
*ctx
,
407 struct io_kiocb
*req
)
409 if ((req
->flags
& (REQ_F_IO_DRAIN
|REQ_F_IO_DRAINED
)) != REQ_F_IO_DRAIN
)
412 return req
->sequence
!= ctx
->cached_cq_tail
+ ctx
->rings
->sq_dropped
;
415 static struct io_kiocb
*io_get_deferred_req(struct io_ring_ctx
*ctx
)
417 struct io_kiocb
*req
;
419 if (list_empty(&ctx
->defer_list
))
422 req
= list_first_entry(&ctx
->defer_list
, struct io_kiocb
, list
);
423 if (!io_sequence_defer(ctx
, req
)) {
424 list_del_init(&req
->list
);
431 static void __io_commit_cqring(struct io_ring_ctx
*ctx
)
433 struct io_rings
*rings
= ctx
->rings
;
435 if (ctx
->cached_cq_tail
!= READ_ONCE(rings
->cq
.tail
)) {
436 /* order cqe stores with ring update */
437 smp_store_release(&rings
->cq
.tail
, ctx
->cached_cq_tail
);
439 if (wq_has_sleeper(&ctx
->cq_wait
)) {
440 wake_up_interruptible(&ctx
->cq_wait
);
441 kill_fasync(&ctx
->cq_fasync
, SIGIO
, POLL_IN
);
446 static inline void io_queue_async_work(struct io_ring_ctx
*ctx
,
447 struct io_kiocb
*req
)
451 switch (req
->submit
.sqe
->opcode
) {
452 case IORING_OP_WRITEV
:
453 case IORING_OP_WRITE_FIXED
:
454 rw
= !(req
->rw
.ki_flags
& IOCB_DIRECT
);
461 queue_work(ctx
->sqo_wq
[rw
], &req
->work
);
464 static void io_commit_cqring(struct io_ring_ctx
*ctx
)
466 struct io_kiocb
*req
;
468 __io_commit_cqring(ctx
);
470 while ((req
= io_get_deferred_req(ctx
)) != NULL
) {
471 if (req
->flags
& REQ_F_SHADOW_DRAIN
) {
472 /* Just for drain, free it. */
476 req
->flags
|= REQ_F_IO_DRAINED
;
477 io_queue_async_work(ctx
, req
);
481 static struct io_uring_cqe
*io_get_cqring(struct io_ring_ctx
*ctx
)
483 struct io_rings
*rings
= ctx
->rings
;
486 tail
= ctx
->cached_cq_tail
;
488 * writes to the cq entry need to come after reading head; the
489 * control dependency is enough as we're using WRITE_ONCE to
492 if (tail
- READ_ONCE(rings
->cq
.head
) == rings
->cq_ring_entries
)
495 ctx
->cached_cq_tail
++;
496 return &rings
->cqes
[tail
& ctx
->cq_mask
];
499 static void io_cqring_fill_event(struct io_ring_ctx
*ctx
, u64 ki_user_data
,
502 struct io_uring_cqe
*cqe
;
505 * If we can't get a cq entry, userspace overflowed the
506 * submission (by quite a lot). Increment the overflow count in
509 cqe
= io_get_cqring(ctx
);
511 WRITE_ONCE(cqe
->user_data
, ki_user_data
);
512 WRITE_ONCE(cqe
->res
, res
);
513 WRITE_ONCE(cqe
->flags
, 0);
515 unsigned overflow
= READ_ONCE(ctx
->rings
->cq_overflow
);
517 WRITE_ONCE(ctx
->rings
->cq_overflow
, overflow
+ 1);
521 static void io_cqring_ev_posted(struct io_ring_ctx
*ctx
)
523 if (waitqueue_active(&ctx
->wait
))
525 if (waitqueue_active(&ctx
->sqo_wait
))
526 wake_up(&ctx
->sqo_wait
);
528 eventfd_signal(ctx
->cq_ev_fd
, 1);
531 static void io_cqring_add_event(struct io_ring_ctx
*ctx
, u64 user_data
,
536 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
537 io_cqring_fill_event(ctx
, user_data
, res
);
538 io_commit_cqring(ctx
);
539 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
541 io_cqring_ev_posted(ctx
);
544 static void io_ring_drop_ctx_refs(struct io_ring_ctx
*ctx
, unsigned refs
)
546 percpu_ref_put_many(&ctx
->refs
, refs
);
548 if (waitqueue_active(&ctx
->wait
))
552 static struct io_kiocb
*io_get_req(struct io_ring_ctx
*ctx
,
553 struct io_submit_state
*state
)
555 gfp_t gfp
= GFP_KERNEL
| __GFP_NOWARN
;
556 struct io_kiocb
*req
;
558 if (!percpu_ref_tryget(&ctx
->refs
))
562 req
= kmem_cache_alloc(req_cachep
, gfp
);
565 } else if (!state
->free_reqs
) {
569 sz
= min_t(size_t, state
->ios_left
, ARRAY_SIZE(state
->reqs
));
570 ret
= kmem_cache_alloc_bulk(req_cachep
, gfp
, sz
, state
->reqs
);
573 * Bulk alloc is all-or-nothing. If we fail to get a batch,
574 * retry single alloc to be on the safe side.
576 if (unlikely(ret
<= 0)) {
577 state
->reqs
[0] = kmem_cache_alloc(req_cachep
, gfp
);
582 state
->free_reqs
= ret
- 1;
584 req
= state
->reqs
[0];
586 req
= state
->reqs
[state
->cur_req
];
594 /* one is dropped after submission, the other at completion */
595 refcount_set(&req
->refs
, 2);
599 io_ring_drop_ctx_refs(ctx
, 1);
603 static void io_free_req_many(struct io_ring_ctx
*ctx
, void **reqs
, int *nr
)
606 kmem_cache_free_bulk(req_cachep
, *nr
, reqs
);
607 io_ring_drop_ctx_refs(ctx
, *nr
);
612 static void __io_free_req(struct io_kiocb
*req
)
614 if (req
->file
&& !(req
->flags
& REQ_F_FIXED_FILE
))
616 io_ring_drop_ctx_refs(req
->ctx
, 1);
617 kmem_cache_free(req_cachep
, req
);
620 static void io_req_link_next(struct io_kiocb
*req
)
622 struct io_kiocb
*nxt
;
625 * The list should never be empty when we are called here. But could
626 * potentially happen if the chain is messed up, check to be on the
629 nxt
= list_first_entry_or_null(&req
->link_list
, struct io_kiocb
, list
);
631 list_del(&nxt
->list
);
632 if (!list_empty(&req
->link_list
)) {
633 INIT_LIST_HEAD(&nxt
->link_list
);
634 list_splice(&req
->link_list
, &nxt
->link_list
);
635 nxt
->flags
|= REQ_F_LINK
;
638 nxt
->flags
|= REQ_F_LINK_DONE
;
639 INIT_WORK(&nxt
->work
, io_sq_wq_submit_work
);
640 io_queue_async_work(req
->ctx
, nxt
);
645 * Called if REQ_F_LINK is set, and we fail the head request
647 static void io_fail_links(struct io_kiocb
*req
)
649 struct io_kiocb
*link
;
651 while (!list_empty(&req
->link_list
)) {
652 link
= list_first_entry(&req
->link_list
, struct io_kiocb
, list
);
653 list_del(&link
->list
);
655 io_cqring_add_event(req
->ctx
, link
->user_data
, -ECANCELED
);
660 static void io_free_req(struct io_kiocb
*req
)
663 * If LINK is set, we have dependent requests in this chain. If we
664 * didn't fail this request, queue the first one up, moving any other
665 * dependencies to the next request. In case of failure, fail the rest
668 if (req
->flags
& REQ_F_LINK
) {
669 if (req
->flags
& REQ_F_FAIL_LINK
)
672 io_req_link_next(req
);
678 static void io_put_req(struct io_kiocb
*req
)
680 if (refcount_dec_and_test(&req
->refs
))
684 static unsigned io_cqring_events(struct io_rings
*rings
)
686 /* See comment at the top of this file */
688 return READ_ONCE(rings
->cq
.tail
) - READ_ONCE(rings
->cq
.head
);
692 * Find and free completed poll iocbs
694 static void io_iopoll_complete(struct io_ring_ctx
*ctx
, unsigned int *nr_events
,
695 struct list_head
*done
)
697 void *reqs
[IO_IOPOLL_BATCH
];
698 struct io_kiocb
*req
;
702 while (!list_empty(done
)) {
703 req
= list_first_entry(done
, struct io_kiocb
, list
);
704 list_del(&req
->list
);
706 io_cqring_fill_event(ctx
, req
->user_data
, req
->result
);
709 if (refcount_dec_and_test(&req
->refs
)) {
710 /* If we're not using fixed files, we have to pair the
711 * completion part with the file put. Use regular
712 * completions for those, only batch free for fixed
713 * file and non-linked commands.
715 if ((req
->flags
& (REQ_F_FIXED_FILE
|REQ_F_LINK
)) ==
717 reqs
[to_free
++] = req
;
718 if (to_free
== ARRAY_SIZE(reqs
))
719 io_free_req_many(ctx
, reqs
, &to_free
);
726 io_commit_cqring(ctx
);
727 io_free_req_many(ctx
, reqs
, &to_free
);
730 static int io_do_iopoll(struct io_ring_ctx
*ctx
, unsigned int *nr_events
,
733 struct io_kiocb
*req
, *tmp
;
739 * Only spin for completions if we don't have multiple devices hanging
740 * off our complete list, and we're under the requested amount.
742 spin
= !ctx
->poll_multi_file
&& *nr_events
< min
;
745 list_for_each_entry_safe(req
, tmp
, &ctx
->poll_list
, list
) {
746 struct kiocb
*kiocb
= &req
->rw
;
749 * Move completed entries to our local list. If we find a
750 * request that requires polling, break out and complete
751 * the done list first, if we have entries there.
753 if (req
->flags
& REQ_F_IOPOLL_COMPLETED
) {
754 list_move_tail(&req
->list
, &done
);
757 if (!list_empty(&done
))
760 ret
= kiocb
->ki_filp
->f_op
->iopoll(kiocb
, spin
);
769 if (!list_empty(&done
))
770 io_iopoll_complete(ctx
, nr_events
, &done
);
776 * Poll for a mininum of 'min' events. Note that if min == 0 we consider that a
777 * non-spinning poll check - we'll still enter the driver poll loop, but only
778 * as a non-spinning completion check.
780 static int io_iopoll_getevents(struct io_ring_ctx
*ctx
, unsigned int *nr_events
,
783 while (!list_empty(&ctx
->poll_list
) && !need_resched()) {
786 ret
= io_do_iopoll(ctx
, nr_events
, min
);
789 if (!min
|| *nr_events
>= min
)
797 * We can't just wait for polled events to come to us, we have to actively
798 * find and complete them.
800 static void io_iopoll_reap_events(struct io_ring_ctx
*ctx
)
802 if (!(ctx
->flags
& IORING_SETUP_IOPOLL
))
805 mutex_lock(&ctx
->uring_lock
);
806 while (!list_empty(&ctx
->poll_list
)) {
807 unsigned int nr_events
= 0;
809 io_iopoll_getevents(ctx
, &nr_events
, 1);
812 * Ensure we allow local-to-the-cpu processing to take place,
813 * in this case we need to ensure that we reap all events.
817 mutex_unlock(&ctx
->uring_lock
);
820 static int io_iopoll_check(struct io_ring_ctx
*ctx
, unsigned *nr_events
,
826 * We disallow the app entering submit/complete with polling, but we
827 * still need to lock the ring to prevent racing with polled issue
828 * that got punted to a workqueue.
830 mutex_lock(&ctx
->uring_lock
);
837 * Don't enter poll loop if we already have events pending.
838 * If we do, we can potentially be spinning for commands that
839 * already triggered a CQE (eg in error).
841 if (io_cqring_events(ctx
->rings
))
845 * If a submit got punted to a workqueue, we can have the
846 * application entering polling for a command before it gets
847 * issued. That app will hold the uring_lock for the duration
848 * of the poll right here, so we need to take a breather every
849 * now and then to ensure that the issue has a chance to add
850 * the poll to the issued list. Otherwise we can spin here
851 * forever, while the workqueue is stuck trying to acquire the
854 if (!(++iters
& 7)) {
855 mutex_unlock(&ctx
->uring_lock
);
856 mutex_lock(&ctx
->uring_lock
);
859 if (*nr_events
< min
)
860 tmin
= min
- *nr_events
;
862 ret
= io_iopoll_getevents(ctx
, nr_events
, tmin
);
866 } while (min
&& !*nr_events
&& !need_resched());
868 mutex_unlock(&ctx
->uring_lock
);
872 static void kiocb_end_write(struct kiocb
*kiocb
)
874 if (kiocb
->ki_flags
& IOCB_WRITE
) {
875 struct inode
*inode
= file_inode(kiocb
->ki_filp
);
878 * Tell lockdep we inherited freeze protection from submission
881 if (S_ISREG(inode
->i_mode
))
882 __sb_writers_acquired(inode
->i_sb
, SB_FREEZE_WRITE
);
883 file_end_write(kiocb
->ki_filp
);
887 static void io_complete_rw(struct kiocb
*kiocb
, long res
, long res2
)
889 struct io_kiocb
*req
= container_of(kiocb
, struct io_kiocb
, rw
);
891 kiocb_end_write(kiocb
);
893 if ((req
->flags
& REQ_F_LINK
) && res
!= req
->result
)
894 req
->flags
|= REQ_F_FAIL_LINK
;
895 io_cqring_add_event(req
->ctx
, req
->user_data
, res
);
899 static void io_complete_rw_iopoll(struct kiocb
*kiocb
, long res
, long res2
)
901 struct io_kiocb
*req
= container_of(kiocb
, struct io_kiocb
, rw
);
903 kiocb_end_write(kiocb
);
905 if ((req
->flags
& REQ_F_LINK
) && res
!= req
->result
)
906 req
->flags
|= REQ_F_FAIL_LINK
;
909 req
->flags
|= REQ_F_IOPOLL_COMPLETED
;
913 * After the iocb has been issued, it's safe to be found on the poll list.
914 * Adding the kiocb to the list AFTER submission ensures that we don't
915 * find it from a io_iopoll_getevents() thread before the issuer is done
916 * accessing the kiocb cookie.
918 static void io_iopoll_req_issued(struct io_kiocb
*req
)
920 struct io_ring_ctx
*ctx
= req
->ctx
;
923 * Track whether we have multiple files in our lists. This will impact
924 * how we do polling eventually, not spinning if we're on potentially
927 if (list_empty(&ctx
->poll_list
)) {
928 ctx
->poll_multi_file
= false;
929 } else if (!ctx
->poll_multi_file
) {
930 struct io_kiocb
*list_req
;
932 list_req
= list_first_entry(&ctx
->poll_list
, struct io_kiocb
,
934 if (list_req
->rw
.ki_filp
!= req
->rw
.ki_filp
)
935 ctx
->poll_multi_file
= true;
939 * For fast devices, IO may have already completed. If it has, add
940 * it to the front so we find it first.
942 if (req
->flags
& REQ_F_IOPOLL_COMPLETED
)
943 list_add(&req
->list
, &ctx
->poll_list
);
945 list_add_tail(&req
->list
, &ctx
->poll_list
);
948 static void io_file_put(struct io_submit_state
*state
)
951 int diff
= state
->has_refs
- state
->used_refs
;
954 fput_many(state
->file
, diff
);
960 * Get as many references to a file as we have IOs left in this submission,
961 * assuming most submissions are for one file, or at least that each file
962 * has more than one submission.
964 static struct file
*io_file_get(struct io_submit_state
*state
, int fd
)
970 if (state
->fd
== fd
) {
977 state
->file
= fget_many(fd
, state
->ios_left
);
982 state
->has_refs
= state
->ios_left
;
983 state
->used_refs
= 1;
989 * If we tracked the file through the SCM inflight mechanism, we could support
990 * any file. For now, just ensure that anything potentially problematic is done
993 static bool io_file_supports_async(struct file
*file
)
995 umode_t mode
= file_inode(file
)->i_mode
;
997 if (S_ISBLK(mode
) || S_ISCHR(mode
))
999 if (S_ISREG(mode
) && file
->f_op
!= &io_uring_fops
)
1005 static int io_prep_rw(struct io_kiocb
*req
, const struct sqe_submit
*s
,
1006 bool force_nonblock
)
1008 const struct io_uring_sqe
*sqe
= s
->sqe
;
1009 struct io_ring_ctx
*ctx
= req
->ctx
;
1010 struct kiocb
*kiocb
= &req
->rw
;
1017 if (force_nonblock
&& !io_file_supports_async(req
->file
))
1018 force_nonblock
= false;
1020 kiocb
->ki_pos
= READ_ONCE(sqe
->off
);
1021 kiocb
->ki_flags
= iocb_flags(kiocb
->ki_filp
);
1022 kiocb
->ki_hint
= ki_hint_validate(file_write_hint(kiocb
->ki_filp
));
1024 ioprio
= READ_ONCE(sqe
->ioprio
);
1026 ret
= ioprio_check_cap(ioprio
);
1030 kiocb
->ki_ioprio
= ioprio
;
1032 kiocb
->ki_ioprio
= get_current_ioprio();
1034 ret
= kiocb_set_rw_flags(kiocb
, READ_ONCE(sqe
->rw_flags
));
1038 /* don't allow async punt if RWF_NOWAIT was requested */
1039 if (kiocb
->ki_flags
& IOCB_NOWAIT
)
1040 req
->flags
|= REQ_F_NOWAIT
;
1043 kiocb
->ki_flags
|= IOCB_NOWAIT
;
1045 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
1046 if (!(kiocb
->ki_flags
& IOCB_DIRECT
) ||
1047 !kiocb
->ki_filp
->f_op
->iopoll
)
1050 kiocb
->ki_flags
|= IOCB_HIPRI
;
1051 kiocb
->ki_complete
= io_complete_rw_iopoll
;
1053 if (kiocb
->ki_flags
& IOCB_HIPRI
)
1055 kiocb
->ki_complete
= io_complete_rw
;
1060 static inline void io_rw_done(struct kiocb
*kiocb
, ssize_t ret
)
1066 case -ERESTARTNOINTR
:
1067 case -ERESTARTNOHAND
:
1068 case -ERESTART_RESTARTBLOCK
:
1070 * We can't just restart the syscall, since previously
1071 * submitted sqes may already be in progress. Just fail this
1077 kiocb
->ki_complete(kiocb
, ret
, 0);
1081 static int io_import_fixed(struct io_ring_ctx
*ctx
, int rw
,
1082 const struct io_uring_sqe
*sqe
,
1083 struct iov_iter
*iter
)
1085 size_t len
= READ_ONCE(sqe
->len
);
1086 struct io_mapped_ubuf
*imu
;
1087 unsigned index
, buf_index
;
1091 /* attempt to use fixed buffers without having provided iovecs */
1092 if (unlikely(!ctx
->user_bufs
))
1095 buf_index
= READ_ONCE(sqe
->buf_index
);
1096 if (unlikely(buf_index
>= ctx
->nr_user_bufs
))
1099 index
= array_index_nospec(buf_index
, ctx
->nr_user_bufs
);
1100 imu
= &ctx
->user_bufs
[index
];
1101 buf_addr
= READ_ONCE(sqe
->addr
);
1104 if (buf_addr
+ len
< buf_addr
)
1106 /* not inside the mapped region */
1107 if (buf_addr
< imu
->ubuf
|| buf_addr
+ len
> imu
->ubuf
+ imu
->len
)
1111 * May not be a start of buffer, set size appropriately
1112 * and advance us to the beginning.
1114 offset
= buf_addr
- imu
->ubuf
;
1115 iov_iter_bvec(iter
, rw
, imu
->bvec
, imu
->nr_bvecs
, offset
+ len
);
1119 * Don't use iov_iter_advance() here, as it's really slow for
1120 * using the latter parts of a big fixed buffer - it iterates
1121 * over each segment manually. We can cheat a bit here, because
1124 * 1) it's a BVEC iter, we set it up
1125 * 2) all bvecs are PAGE_SIZE in size, except potentially the
1126 * first and last bvec
1128 * So just find our index, and adjust the iterator afterwards.
1129 * If the offset is within the first bvec (or the whole first
1130 * bvec, just use iov_iter_advance(). This makes it easier
1131 * since we can just skip the first segment, which may not
1132 * be PAGE_SIZE aligned.
1134 const struct bio_vec
*bvec
= imu
->bvec
;
1136 if (offset
<= bvec
->bv_len
) {
1137 iov_iter_advance(iter
, offset
);
1139 unsigned long seg_skip
;
1141 /* skip first vec */
1142 offset
-= bvec
->bv_len
;
1143 seg_skip
= 1 + (offset
>> PAGE_SHIFT
);
1145 iter
->bvec
= bvec
+ seg_skip
;
1146 iter
->nr_segs
-= seg_skip
;
1147 iter
->count
-= bvec
->bv_len
+ offset
;
1148 iter
->iov_offset
= offset
& ~PAGE_MASK
;
1155 static ssize_t
io_import_iovec(struct io_ring_ctx
*ctx
, int rw
,
1156 const struct sqe_submit
*s
, struct iovec
**iovec
,
1157 struct iov_iter
*iter
)
1159 const struct io_uring_sqe
*sqe
= s
->sqe
;
1160 void __user
*buf
= u64_to_user_ptr(READ_ONCE(sqe
->addr
));
1161 size_t sqe_len
= READ_ONCE(sqe
->len
);
1165 * We're reading ->opcode for the second time, but the first read
1166 * doesn't care whether it's _FIXED or not, so it doesn't matter
1167 * whether ->opcode changes concurrently. The first read does care
1168 * about whether it is a READ or a WRITE, so we don't trust this read
1169 * for that purpose and instead let the caller pass in the read/write
1172 opcode
= READ_ONCE(sqe
->opcode
);
1173 if (opcode
== IORING_OP_READ_FIXED
||
1174 opcode
== IORING_OP_WRITE_FIXED
) {
1175 ssize_t ret
= io_import_fixed(ctx
, rw
, sqe
, iter
);
1183 #ifdef CONFIG_COMPAT
1185 return compat_import_iovec(rw
, buf
, sqe_len
, UIO_FASTIOV
,
1189 return import_iovec(rw
, buf
, sqe_len
, UIO_FASTIOV
, iovec
, iter
);
1192 static inline bool io_should_merge(struct async_list
*al
, struct kiocb
*kiocb
)
1194 if (al
->file
== kiocb
->ki_filp
) {
1198 * Allow merging if we're anywhere in the range of the same
1199 * page. Generally this happens for sub-page reads or writes,
1200 * and it's beneficial to allow the first worker to bring the
1201 * page in and the piggy backed work can then work on the
1204 start
= al
->io_start
& PAGE_MASK
;
1205 end
= (al
->io_start
+ al
->io_len
+ PAGE_SIZE
- 1) & PAGE_MASK
;
1206 if (kiocb
->ki_pos
>= start
&& kiocb
->ki_pos
<= end
)
1215 * Make a note of the last file/offset/direction we punted to async
1216 * context. We'll use this information to see if we can piggy back a
1217 * sequential request onto the previous one, if it's still hasn't been
1218 * completed by the async worker.
1220 static void io_async_list_note(int rw
, struct io_kiocb
*req
, size_t len
)
1222 struct async_list
*async_list
= &req
->ctx
->pending_async
[rw
];
1223 struct kiocb
*kiocb
= &req
->rw
;
1224 struct file
*filp
= kiocb
->ki_filp
;
1226 if (io_should_merge(async_list
, kiocb
)) {
1227 unsigned long max_bytes
;
1229 /* Use 8x RA size as a decent limiter for both reads/writes */
1230 max_bytes
= filp
->f_ra
.ra_pages
<< (PAGE_SHIFT
+ 3);
1232 max_bytes
= VM_READAHEAD_PAGES
<< (PAGE_SHIFT
+ 3);
1234 /* If max len are exceeded, reset the state */
1235 if (async_list
->io_len
+ len
<= max_bytes
) {
1236 req
->flags
|= REQ_F_SEQ_PREV
;
1237 async_list
->io_len
+= len
;
1239 async_list
->file
= NULL
;
1243 /* New file? Reset state. */
1244 if (async_list
->file
!= filp
) {
1245 async_list
->io_start
= kiocb
->ki_pos
;
1246 async_list
->io_len
= len
;
1247 async_list
->file
= filp
;
1251 static int io_read(struct io_kiocb
*req
, const struct sqe_submit
*s
,
1252 bool force_nonblock
)
1254 struct iovec inline_vecs
[UIO_FASTIOV
], *iovec
= inline_vecs
;
1255 struct kiocb
*kiocb
= &req
->rw
;
1256 struct iov_iter iter
;
1259 ssize_t read_size
, ret
;
1261 ret
= io_prep_rw(req
, s
, force_nonblock
);
1264 file
= kiocb
->ki_filp
;
1266 if (unlikely(!(file
->f_mode
& FMODE_READ
)))
1268 if (unlikely(!file
->f_op
->read_iter
))
1271 ret
= io_import_iovec(req
->ctx
, READ
, s
, &iovec
, &iter
);
1276 if (req
->flags
& REQ_F_LINK
)
1277 req
->result
= read_size
;
1279 iov_count
= iov_iter_count(&iter
);
1280 ret
= rw_verify_area(READ
, file
, &kiocb
->ki_pos
, iov_count
);
1284 ret2
= call_read_iter(file
, kiocb
, &iter
);
1286 * In case of a short read, punt to async. This can happen
1287 * if we have data partially cached. Alternatively we can
1288 * return the short read, in which case the application will
1289 * need to issue another SQE and wait for it. That SQE will
1290 * need async punt anyway, so it's more efficient to do it
1293 if (force_nonblock
&& ret2
> 0 && ret2
< read_size
)
1295 /* Catch -EAGAIN return for forced non-blocking submission */
1296 if (!force_nonblock
|| ret2
!= -EAGAIN
) {
1297 io_rw_done(kiocb
, ret2
);
1300 * If ->needs_lock is true, we're already in async
1304 io_async_list_note(READ
, req
, iov_count
);
1312 static int io_write(struct io_kiocb
*req
, const struct sqe_submit
*s
,
1313 bool force_nonblock
)
1315 struct iovec inline_vecs
[UIO_FASTIOV
], *iovec
= inline_vecs
;
1316 struct kiocb
*kiocb
= &req
->rw
;
1317 struct iov_iter iter
;
1322 ret
= io_prep_rw(req
, s
, force_nonblock
);
1326 file
= kiocb
->ki_filp
;
1327 if (unlikely(!(file
->f_mode
& FMODE_WRITE
)))
1329 if (unlikely(!file
->f_op
->write_iter
))
1332 ret
= io_import_iovec(req
->ctx
, WRITE
, s
, &iovec
, &iter
);
1336 if (req
->flags
& REQ_F_LINK
)
1339 iov_count
= iov_iter_count(&iter
);
1342 if (force_nonblock
&& !(kiocb
->ki_flags
& IOCB_DIRECT
)) {
1343 /* If ->needs_lock is true, we're already in async context. */
1345 io_async_list_note(WRITE
, req
, iov_count
);
1349 ret
= rw_verify_area(WRITE
, file
, &kiocb
->ki_pos
, iov_count
);
1354 * Open-code file_start_write here to grab freeze protection,
1355 * which will be released by another thread in
1356 * io_complete_rw(). Fool lockdep by telling it the lock got
1357 * released so that it doesn't complain about the held lock when
1358 * we return to userspace.
1360 if (S_ISREG(file_inode(file
)->i_mode
)) {
1361 __sb_start_write(file_inode(file
)->i_sb
,
1362 SB_FREEZE_WRITE
, true);
1363 __sb_writers_release(file_inode(file
)->i_sb
,
1366 kiocb
->ki_flags
|= IOCB_WRITE
;
1368 ret2
= call_write_iter(file
, kiocb
, &iter
);
1369 if (!force_nonblock
|| ret2
!= -EAGAIN
) {
1370 io_rw_done(kiocb
, ret2
);
1373 * If ->needs_lock is true, we're already in async
1377 io_async_list_note(WRITE
, req
, iov_count
);
1387 * IORING_OP_NOP just posts a completion event, nothing else.
1389 static int io_nop(struct io_kiocb
*req
, u64 user_data
)
1391 struct io_ring_ctx
*ctx
= req
->ctx
;
1394 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
1397 io_cqring_add_event(ctx
, user_data
, err
);
1402 static int io_prep_fsync(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
1404 struct io_ring_ctx
*ctx
= req
->ctx
;
1409 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
1411 if (unlikely(sqe
->addr
|| sqe
->ioprio
|| sqe
->buf_index
))
1417 static int io_fsync(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
1418 bool force_nonblock
)
1420 loff_t sqe_off
= READ_ONCE(sqe
->off
);
1421 loff_t sqe_len
= READ_ONCE(sqe
->len
);
1422 loff_t end
= sqe_off
+ sqe_len
;
1423 unsigned fsync_flags
;
1426 fsync_flags
= READ_ONCE(sqe
->fsync_flags
);
1427 if (unlikely(fsync_flags
& ~IORING_FSYNC_DATASYNC
))
1430 ret
= io_prep_fsync(req
, sqe
);
1434 /* fsync always requires a blocking context */
1438 ret
= vfs_fsync_range(req
->rw
.ki_filp
, sqe_off
,
1439 end
> 0 ? end
: LLONG_MAX
,
1440 fsync_flags
& IORING_FSYNC_DATASYNC
);
1442 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
1443 req
->flags
|= REQ_F_FAIL_LINK
;
1444 io_cqring_add_event(req
->ctx
, sqe
->user_data
, ret
);
1449 static int io_prep_sfr(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
1451 struct io_ring_ctx
*ctx
= req
->ctx
;
1457 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
1459 if (unlikely(sqe
->addr
|| sqe
->ioprio
|| sqe
->buf_index
))
1465 static int io_sync_file_range(struct io_kiocb
*req
,
1466 const struct io_uring_sqe
*sqe
,
1467 bool force_nonblock
)
1474 ret
= io_prep_sfr(req
, sqe
);
1478 /* sync_file_range always requires a blocking context */
1482 sqe_off
= READ_ONCE(sqe
->off
);
1483 sqe_len
= READ_ONCE(sqe
->len
);
1484 flags
= READ_ONCE(sqe
->sync_range_flags
);
1486 ret
= sync_file_range(req
->rw
.ki_filp
, sqe_off
, sqe_len
, flags
);
1488 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
1489 req
->flags
|= REQ_F_FAIL_LINK
;
1490 io_cqring_add_event(req
->ctx
, sqe
->user_data
, ret
);
1495 #if defined(CONFIG_NET)
1496 static int io_send_recvmsg(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
1497 bool force_nonblock
,
1498 long (*fn
)(struct socket
*, struct user_msghdr __user
*,
1501 struct socket
*sock
;
1504 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
1507 sock
= sock_from_file(req
->file
, &ret
);
1509 struct user_msghdr __user
*msg
;
1512 flags
= READ_ONCE(sqe
->msg_flags
);
1513 if (flags
& MSG_DONTWAIT
)
1514 req
->flags
|= REQ_F_NOWAIT
;
1515 else if (force_nonblock
)
1516 flags
|= MSG_DONTWAIT
;
1518 msg
= (struct user_msghdr __user
*) (unsigned long)
1519 READ_ONCE(sqe
->addr
);
1521 ret
= fn(sock
, msg
, flags
);
1522 if (force_nonblock
&& ret
== -EAGAIN
)
1526 io_cqring_add_event(req
->ctx
, sqe
->user_data
, ret
);
1532 static int io_sendmsg(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
1533 bool force_nonblock
)
1535 #if defined(CONFIG_NET)
1536 return io_send_recvmsg(req
, sqe
, force_nonblock
, __sys_sendmsg_sock
);
1542 static int io_recvmsg(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
1543 bool force_nonblock
)
1545 #if defined(CONFIG_NET)
1546 return io_send_recvmsg(req
, sqe
, force_nonblock
, __sys_recvmsg_sock
);
1552 static void io_poll_remove_one(struct io_kiocb
*req
)
1554 struct io_poll_iocb
*poll
= &req
->poll
;
1556 spin_lock(&poll
->head
->lock
);
1557 WRITE_ONCE(poll
->canceled
, true);
1558 if (!list_empty(&poll
->wait
.entry
)) {
1559 list_del_init(&poll
->wait
.entry
);
1560 io_queue_async_work(req
->ctx
, req
);
1562 spin_unlock(&poll
->head
->lock
);
1564 list_del_init(&req
->list
);
1567 static void io_poll_remove_all(struct io_ring_ctx
*ctx
)
1569 struct io_kiocb
*req
;
1571 spin_lock_irq(&ctx
->completion_lock
);
1572 while (!list_empty(&ctx
->cancel_list
)) {
1573 req
= list_first_entry(&ctx
->cancel_list
, struct io_kiocb
,list
);
1574 io_poll_remove_one(req
);
1576 spin_unlock_irq(&ctx
->completion_lock
);
1580 * Find a running poll command that matches one specified in sqe->addr,
1581 * and remove it if found.
1583 static int io_poll_remove(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
1585 struct io_ring_ctx
*ctx
= req
->ctx
;
1586 struct io_kiocb
*poll_req
, *next
;
1589 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
1591 if (sqe
->ioprio
|| sqe
->off
|| sqe
->len
|| sqe
->buf_index
||
1595 spin_lock_irq(&ctx
->completion_lock
);
1596 list_for_each_entry_safe(poll_req
, next
, &ctx
->cancel_list
, list
) {
1597 if (READ_ONCE(sqe
->addr
) == poll_req
->user_data
) {
1598 io_poll_remove_one(poll_req
);
1603 spin_unlock_irq(&ctx
->completion_lock
);
1605 io_cqring_add_event(req
->ctx
, sqe
->user_data
, ret
);
1610 static void io_poll_complete(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
,
1613 req
->poll
.done
= true;
1614 io_cqring_fill_event(ctx
, req
->user_data
, mangle_poll(mask
));
1615 io_commit_cqring(ctx
);
1618 static void io_poll_complete_work(struct work_struct
*work
)
1620 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
1621 struct io_poll_iocb
*poll
= &req
->poll
;
1622 struct poll_table_struct pt
= { ._key
= poll
->events
};
1623 struct io_ring_ctx
*ctx
= req
->ctx
;
1626 if (!READ_ONCE(poll
->canceled
))
1627 mask
= vfs_poll(poll
->file
, &pt
) & poll
->events
;
1630 * Note that ->ki_cancel callers also delete iocb from active_reqs after
1631 * calling ->ki_cancel. We need the ctx_lock roundtrip here to
1632 * synchronize with them. In the cancellation case the list_del_init
1633 * itself is not actually needed, but harmless so we keep it in to
1634 * avoid further branches in the fast path.
1636 spin_lock_irq(&ctx
->completion_lock
);
1637 if (!mask
&& !READ_ONCE(poll
->canceled
)) {
1638 add_wait_queue(poll
->head
, &poll
->wait
);
1639 spin_unlock_irq(&ctx
->completion_lock
);
1642 list_del_init(&req
->list
);
1643 io_poll_complete(ctx
, req
, mask
);
1644 spin_unlock_irq(&ctx
->completion_lock
);
1646 io_cqring_ev_posted(ctx
);
1650 static int io_poll_wake(struct wait_queue_entry
*wait
, unsigned mode
, int sync
,
1653 struct io_poll_iocb
*poll
= container_of(wait
, struct io_poll_iocb
,
1655 struct io_kiocb
*req
= container_of(poll
, struct io_kiocb
, poll
);
1656 struct io_ring_ctx
*ctx
= req
->ctx
;
1657 __poll_t mask
= key_to_poll(key
);
1658 unsigned long flags
;
1660 /* for instances that support it check for an event match first: */
1661 if (mask
&& !(mask
& poll
->events
))
1664 list_del_init(&poll
->wait
.entry
);
1666 if (mask
&& spin_trylock_irqsave(&ctx
->completion_lock
, flags
)) {
1667 list_del(&req
->list
);
1668 io_poll_complete(ctx
, req
, mask
);
1669 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
1671 io_cqring_ev_posted(ctx
);
1674 io_queue_async_work(ctx
, req
);
1680 struct io_poll_table
{
1681 struct poll_table_struct pt
;
1682 struct io_kiocb
*req
;
1686 static void io_poll_queue_proc(struct file
*file
, struct wait_queue_head
*head
,
1687 struct poll_table_struct
*p
)
1689 struct io_poll_table
*pt
= container_of(p
, struct io_poll_table
, pt
);
1691 if (unlikely(pt
->req
->poll
.head
)) {
1692 pt
->error
= -EINVAL
;
1697 pt
->req
->poll
.head
= head
;
1698 add_wait_queue(head
, &pt
->req
->poll
.wait
);
1701 static int io_poll_add(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
1703 struct io_poll_iocb
*poll
= &req
->poll
;
1704 struct io_ring_ctx
*ctx
= req
->ctx
;
1705 struct io_poll_table ipt
;
1706 bool cancel
= false;
1710 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
1712 if (sqe
->addr
|| sqe
->ioprio
|| sqe
->off
|| sqe
->len
|| sqe
->buf_index
)
1717 INIT_WORK(&req
->work
, io_poll_complete_work
);
1718 events
= READ_ONCE(sqe
->poll_events
);
1719 poll
->events
= demangle_poll(events
) | EPOLLERR
| EPOLLHUP
;
1723 poll
->canceled
= false;
1725 ipt
.pt
._qproc
= io_poll_queue_proc
;
1726 ipt
.pt
._key
= poll
->events
;
1728 ipt
.error
= -EINVAL
; /* same as no support for IOCB_CMD_POLL */
1730 /* initialized the list so that we can do list_empty checks */
1731 INIT_LIST_HEAD(&poll
->wait
.entry
);
1732 init_waitqueue_func_entry(&poll
->wait
, io_poll_wake
);
1734 INIT_LIST_HEAD(&req
->list
);
1736 mask
= vfs_poll(poll
->file
, &ipt
.pt
) & poll
->events
;
1738 spin_lock_irq(&ctx
->completion_lock
);
1739 if (likely(poll
->head
)) {
1740 spin_lock(&poll
->head
->lock
);
1741 if (unlikely(list_empty(&poll
->wait
.entry
))) {
1747 if (mask
|| ipt
.error
)
1748 list_del_init(&poll
->wait
.entry
);
1750 WRITE_ONCE(poll
->canceled
, true);
1751 else if (!poll
->done
) /* actually waiting for an event */
1752 list_add_tail(&req
->list
, &ctx
->cancel_list
);
1753 spin_unlock(&poll
->head
->lock
);
1755 if (mask
) { /* no async, we'd stolen it */
1757 io_poll_complete(ctx
, req
, mask
);
1759 spin_unlock_irq(&ctx
->completion_lock
);
1762 io_cqring_ev_posted(ctx
);
1768 static int io_req_defer(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
,
1769 const struct io_uring_sqe
*sqe
)
1771 struct io_uring_sqe
*sqe_copy
;
1773 if (!io_sequence_defer(ctx
, req
) && list_empty(&ctx
->defer_list
))
1776 sqe_copy
= kmalloc(sizeof(*sqe_copy
), GFP_KERNEL
);
1780 spin_lock_irq(&ctx
->completion_lock
);
1781 if (!io_sequence_defer(ctx
, req
) && list_empty(&ctx
->defer_list
)) {
1782 spin_unlock_irq(&ctx
->completion_lock
);
1787 memcpy(sqe_copy
, sqe
, sizeof(*sqe_copy
));
1788 req
->submit
.sqe
= sqe_copy
;
1790 INIT_WORK(&req
->work
, io_sq_wq_submit_work
);
1791 list_add_tail(&req
->list
, &ctx
->defer_list
);
1792 spin_unlock_irq(&ctx
->completion_lock
);
1793 return -EIOCBQUEUED
;
1796 static int __io_submit_sqe(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
,
1797 const struct sqe_submit
*s
, bool force_nonblock
)
1801 req
->user_data
= READ_ONCE(s
->sqe
->user_data
);
1803 if (unlikely(s
->index
>= ctx
->sq_entries
))
1806 opcode
= READ_ONCE(s
->sqe
->opcode
);
1809 ret
= io_nop(req
, req
->user_data
);
1811 case IORING_OP_READV
:
1812 if (unlikely(s
->sqe
->buf_index
))
1814 ret
= io_read(req
, s
, force_nonblock
);
1816 case IORING_OP_WRITEV
:
1817 if (unlikely(s
->sqe
->buf_index
))
1819 ret
= io_write(req
, s
, force_nonblock
);
1821 case IORING_OP_READ_FIXED
:
1822 ret
= io_read(req
, s
, force_nonblock
);
1824 case IORING_OP_WRITE_FIXED
:
1825 ret
= io_write(req
, s
, force_nonblock
);
1827 case IORING_OP_FSYNC
:
1828 ret
= io_fsync(req
, s
->sqe
, force_nonblock
);
1830 case IORING_OP_POLL_ADD
:
1831 ret
= io_poll_add(req
, s
->sqe
);
1833 case IORING_OP_POLL_REMOVE
:
1834 ret
= io_poll_remove(req
, s
->sqe
);
1836 case IORING_OP_SYNC_FILE_RANGE
:
1837 ret
= io_sync_file_range(req
, s
->sqe
, force_nonblock
);
1839 case IORING_OP_SENDMSG
:
1840 ret
= io_sendmsg(req
, s
->sqe
, force_nonblock
);
1842 case IORING_OP_RECVMSG
:
1843 ret
= io_recvmsg(req
, s
->sqe
, force_nonblock
);
1853 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
1854 if (req
->result
== -EAGAIN
)
1857 /* workqueue context doesn't hold uring_lock, grab it now */
1859 mutex_lock(&ctx
->uring_lock
);
1860 io_iopoll_req_issued(req
);
1862 mutex_unlock(&ctx
->uring_lock
);
1868 static struct async_list
*io_async_list_from_sqe(struct io_ring_ctx
*ctx
,
1869 const struct io_uring_sqe
*sqe
)
1871 switch (sqe
->opcode
) {
1872 case IORING_OP_READV
:
1873 case IORING_OP_READ_FIXED
:
1874 return &ctx
->pending_async
[READ
];
1875 case IORING_OP_WRITEV
:
1876 case IORING_OP_WRITE_FIXED
:
1877 return &ctx
->pending_async
[WRITE
];
1883 static inline bool io_sqe_needs_user(const struct io_uring_sqe
*sqe
)
1885 u8 opcode
= READ_ONCE(sqe
->opcode
);
1887 return !(opcode
== IORING_OP_READ_FIXED
||
1888 opcode
== IORING_OP_WRITE_FIXED
);
1891 static void io_sq_wq_submit_work(struct work_struct
*work
)
1893 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
1894 struct io_ring_ctx
*ctx
= req
->ctx
;
1895 struct mm_struct
*cur_mm
= NULL
;
1896 struct async_list
*async_list
;
1897 LIST_HEAD(req_list
);
1898 mm_segment_t old_fs
;
1901 async_list
= io_async_list_from_sqe(ctx
, req
->submit
.sqe
);
1904 struct sqe_submit
*s
= &req
->submit
;
1905 const struct io_uring_sqe
*sqe
= s
->sqe
;
1906 unsigned int flags
= req
->flags
;
1908 /* Ensure we clear previously set non-block flag */
1909 req
->rw
.ki_flags
&= ~IOCB_NOWAIT
;
1912 if (io_sqe_needs_user(sqe
) && !cur_mm
) {
1913 if (!mmget_not_zero(ctx
->sqo_mm
)) {
1916 cur_mm
= ctx
->sqo_mm
;
1924 s
->has_user
= cur_mm
!= NULL
;
1925 s
->needs_lock
= true;
1927 ret
= __io_submit_sqe(ctx
, req
, s
, false);
1929 * We can get EAGAIN for polled IO even though
1930 * we're forcing a sync submission from here,
1931 * since we can't wait for request slots on the
1940 /* drop submission reference */
1944 io_cqring_add_event(ctx
, sqe
->user_data
, ret
);
1948 /* async context always use a copy of the sqe */
1951 /* req from defer and link list needn't decrease async cnt */
1952 if (flags
& (REQ_F_IO_DRAINED
| REQ_F_LINK_DONE
))
1957 if (!list_empty(&req_list
)) {
1958 req
= list_first_entry(&req_list
, struct io_kiocb
,
1960 list_del(&req
->list
);
1963 if (list_empty(&async_list
->list
))
1967 spin_lock(&async_list
->lock
);
1968 if (list_empty(&async_list
->list
)) {
1969 spin_unlock(&async_list
->lock
);
1972 list_splice_init(&async_list
->list
, &req_list
);
1973 spin_unlock(&async_list
->lock
);
1975 req
= list_first_entry(&req_list
, struct io_kiocb
, list
);
1976 list_del(&req
->list
);
1980 * Rare case of racing with a submitter. If we find the count has
1981 * dropped to zero AND we have pending work items, then restart
1982 * the processing. This is a tiny race window.
1985 ret
= atomic_dec_return(&async_list
->cnt
);
1986 while (!ret
&& !list_empty(&async_list
->list
)) {
1987 spin_lock(&async_list
->lock
);
1988 atomic_inc(&async_list
->cnt
);
1989 list_splice_init(&async_list
->list
, &req_list
);
1990 spin_unlock(&async_list
->lock
);
1992 if (!list_empty(&req_list
)) {
1993 req
= list_first_entry(&req_list
,
1994 struct io_kiocb
, list
);
1995 list_del(&req
->list
);
1998 ret
= atomic_dec_return(&async_list
->cnt
);
2011 * See if we can piggy back onto previously submitted work, that is still
2012 * running. We currently only allow this if the new request is sequential
2013 * to the previous one we punted.
2015 static bool io_add_to_prev_work(struct async_list
*list
, struct io_kiocb
*req
)
2021 if (!(req
->flags
& REQ_F_SEQ_PREV
))
2023 if (!atomic_read(&list
->cnt
))
2027 spin_lock(&list
->lock
);
2028 list_add_tail(&req
->list
, &list
->list
);
2030 * Ensure we see a simultaneous modification from io_sq_wq_submit_work()
2033 if (!atomic_read(&list
->cnt
)) {
2034 list_del_init(&req
->list
);
2037 spin_unlock(&list
->lock
);
2041 static bool io_op_needs_file(const struct io_uring_sqe
*sqe
)
2043 int op
= READ_ONCE(sqe
->opcode
);
2047 case IORING_OP_POLL_REMOVE
:
2054 static int io_req_set_file(struct io_ring_ctx
*ctx
, const struct sqe_submit
*s
,
2055 struct io_submit_state
*state
, struct io_kiocb
*req
)
2060 flags
= READ_ONCE(s
->sqe
->flags
);
2061 fd
= READ_ONCE(s
->sqe
->fd
);
2063 if (flags
& IOSQE_IO_DRAIN
)
2064 req
->flags
|= REQ_F_IO_DRAIN
;
2066 * All io need record the previous position, if LINK vs DARIN,
2067 * it can be used to mark the position of the first IO in the
2070 req
->sequence
= s
->sequence
;
2072 if (!io_op_needs_file(s
->sqe
))
2075 if (flags
& IOSQE_FIXED_FILE
) {
2076 if (unlikely(!ctx
->user_files
||
2077 (unsigned) fd
>= ctx
->nr_user_files
))
2079 req
->file
= ctx
->user_files
[fd
];
2080 req
->flags
|= REQ_F_FIXED_FILE
;
2082 if (s
->needs_fixed_file
)
2084 req
->file
= io_file_get(state
, fd
);
2085 if (unlikely(!req
->file
))
2092 static int __io_queue_sqe(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
,
2093 struct sqe_submit
*s
, bool force_nonblock
)
2097 ret
= __io_submit_sqe(ctx
, req
, s
, force_nonblock
);
2098 if (ret
== -EAGAIN
&& !(req
->flags
& REQ_F_NOWAIT
)) {
2099 struct io_uring_sqe
*sqe_copy
;
2101 sqe_copy
= kmalloc(sizeof(*sqe_copy
), GFP_KERNEL
);
2103 struct async_list
*list
;
2105 memcpy(sqe_copy
, s
->sqe
, sizeof(*sqe_copy
));
2108 memcpy(&req
->submit
, s
, sizeof(*s
));
2109 list
= io_async_list_from_sqe(ctx
, s
->sqe
);
2110 if (!io_add_to_prev_work(list
, req
)) {
2112 atomic_inc(&list
->cnt
);
2113 INIT_WORK(&req
->work
, io_sq_wq_submit_work
);
2114 io_queue_async_work(ctx
, req
);
2118 * Queued up for async execution, worker will release
2119 * submit reference when the iocb is actually submitted.
2125 /* drop submission reference */
2128 /* and drop final reference, if we failed */
2130 io_cqring_add_event(ctx
, req
->user_data
, ret
);
2131 if (req
->flags
& REQ_F_LINK
)
2132 req
->flags
|= REQ_F_FAIL_LINK
;
2139 static int io_queue_sqe(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
,
2140 struct sqe_submit
*s
, bool force_nonblock
)
2144 ret
= io_req_defer(ctx
, req
, s
->sqe
);
2146 if (ret
!= -EIOCBQUEUED
) {
2148 io_cqring_add_event(ctx
, s
->sqe
->user_data
, ret
);
2153 return __io_queue_sqe(ctx
, req
, s
, force_nonblock
);
2156 static int io_queue_link_head(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
,
2157 struct sqe_submit
*s
, struct io_kiocb
*shadow
,
2158 bool force_nonblock
)
2161 int need_submit
= false;
2164 return io_queue_sqe(ctx
, req
, s
, force_nonblock
);
2167 * Mark the first IO in link list as DRAIN, let all the following
2168 * IOs enter the defer list. all IO needs to be completed before link
2171 req
->flags
|= REQ_F_IO_DRAIN
;
2172 ret
= io_req_defer(ctx
, req
, s
->sqe
);
2174 if (ret
!= -EIOCBQUEUED
) {
2176 io_cqring_add_event(ctx
, s
->sqe
->user_data
, ret
);
2181 * If ret == 0 means that all IOs in front of link io are
2182 * running done. let's queue link head.
2187 /* Insert shadow req to defer_list, blocking next IOs */
2188 spin_lock_irq(&ctx
->completion_lock
);
2189 list_add_tail(&shadow
->list
, &ctx
->defer_list
);
2190 spin_unlock_irq(&ctx
->completion_lock
);
2193 return __io_queue_sqe(ctx
, req
, s
, force_nonblock
);
2198 #define SQE_VALID_FLAGS (IOSQE_FIXED_FILE|IOSQE_IO_DRAIN|IOSQE_IO_LINK)
2200 static void io_submit_sqe(struct io_ring_ctx
*ctx
, struct sqe_submit
*s
,
2201 struct io_submit_state
*state
, struct io_kiocb
**link
,
2202 bool force_nonblock
)
2204 struct io_uring_sqe
*sqe_copy
;
2205 struct io_kiocb
*req
;
2208 /* enforce forwards compatibility on users */
2209 if (unlikely(s
->sqe
->flags
& ~SQE_VALID_FLAGS
)) {
2214 req
= io_get_req(ctx
, state
);
2215 if (unlikely(!req
)) {
2220 ret
= io_req_set_file(ctx
, s
, state
, req
);
2221 if (unlikely(ret
)) {
2225 io_cqring_add_event(ctx
, s
->sqe
->user_data
, ret
);
2230 * If we already have a head request, queue this one for async
2231 * submittal once the head completes. If we don't have a head but
2232 * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be
2233 * submitted sync once the chain is complete. If none of those
2234 * conditions are true (normal request), then just queue it.
2237 struct io_kiocb
*prev
= *link
;
2239 sqe_copy
= kmemdup(s
->sqe
, sizeof(*sqe_copy
), GFP_KERNEL
);
2246 memcpy(&req
->submit
, s
, sizeof(*s
));
2247 list_add_tail(&req
->list
, &prev
->link_list
);
2248 } else if (s
->sqe
->flags
& IOSQE_IO_LINK
) {
2249 req
->flags
|= REQ_F_LINK
;
2251 memcpy(&req
->submit
, s
, sizeof(*s
));
2252 INIT_LIST_HEAD(&req
->link_list
);
2255 io_queue_sqe(ctx
, req
, s
, force_nonblock
);
2260 * Batched submission is done, ensure local IO is flushed out.
2262 static void io_submit_state_end(struct io_submit_state
*state
)
2264 blk_finish_plug(&state
->plug
);
2266 if (state
->free_reqs
)
2267 kmem_cache_free_bulk(req_cachep
, state
->free_reqs
,
2268 &state
->reqs
[state
->cur_req
]);
2272 * Start submission side cache.
2274 static void io_submit_state_start(struct io_submit_state
*state
,
2275 struct io_ring_ctx
*ctx
, unsigned max_ios
)
2277 blk_start_plug(&state
->plug
);
2278 state
->free_reqs
= 0;
2280 state
->ios_left
= max_ios
;
2283 static void io_commit_sqring(struct io_ring_ctx
*ctx
)
2285 struct io_rings
*rings
= ctx
->rings
;
2287 if (ctx
->cached_sq_head
!= READ_ONCE(rings
->sq
.head
)) {
2289 * Ensure any loads from the SQEs are done at this point,
2290 * since once we write the new head, the application could
2291 * write new data to them.
2293 smp_store_release(&rings
->sq
.head
, ctx
->cached_sq_head
);
2298 * Fetch an sqe, if one is available. Note that s->sqe will point to memory
2299 * that is mapped by userspace. This means that care needs to be taken to
2300 * ensure that reads are stable, as we cannot rely on userspace always
2301 * being a good citizen. If members of the sqe are validated and then later
2302 * used, it's important that those reads are done through READ_ONCE() to
2303 * prevent a re-load down the line.
2305 static bool io_get_sqring(struct io_ring_ctx
*ctx
, struct sqe_submit
*s
)
2307 struct io_rings
*rings
= ctx
->rings
;
2308 u32
*sq_array
= ctx
->sq_array
;
2312 * The cached sq head (or cq tail) serves two purposes:
2314 * 1) allows us to batch the cost of updating the user visible
2316 * 2) allows the kernel side to track the head on its own, even
2317 * though the application is the one updating it.
2319 head
= ctx
->cached_sq_head
;
2320 /* make sure SQ entry isn't read before tail */
2321 if (head
== smp_load_acquire(&rings
->sq
.tail
))
2324 head
= READ_ONCE(sq_array
[head
& ctx
->sq_mask
]);
2325 if (head
< ctx
->sq_entries
) {
2327 s
->sqe
= &ctx
->sq_sqes
[head
];
2328 s
->sequence
= ctx
->cached_sq_head
;
2329 ctx
->cached_sq_head
++;
2333 /* drop invalid entries */
2334 ctx
->cached_sq_head
++;
2335 rings
->sq_dropped
++;
2339 static int io_submit_sqes(struct io_ring_ctx
*ctx
, struct sqe_submit
*sqes
,
2340 unsigned int nr
, bool has_user
, bool mm_fault
)
2342 struct io_submit_state state
, *statep
= NULL
;
2343 struct io_kiocb
*link
= NULL
;
2344 struct io_kiocb
*shadow_req
= NULL
;
2345 bool prev_was_link
= false;
2346 int i
, submitted
= 0;
2348 if (nr
> IO_PLUG_THRESHOLD
) {
2349 io_submit_state_start(&state
, ctx
, nr
);
2353 for (i
= 0; i
< nr
; i
++) {
2355 * If previous wasn't linked and we have a linked command,
2356 * that's the end of the chain. Submit the previous link.
2358 if (!prev_was_link
&& link
) {
2359 io_queue_link_head(ctx
, link
, &link
->submit
, shadow_req
,
2363 prev_was_link
= (sqes
[i
].sqe
->flags
& IOSQE_IO_LINK
) != 0;
2365 if (link
&& (sqes
[i
].sqe
->flags
& IOSQE_IO_DRAIN
)) {
2367 shadow_req
= io_get_req(ctx
, NULL
);
2368 shadow_req
->flags
|= (REQ_F_IO_DRAIN
| REQ_F_SHADOW_DRAIN
);
2369 refcount_dec(&shadow_req
->refs
);
2371 shadow_req
->sequence
= sqes
[i
].sequence
;
2374 if (unlikely(mm_fault
)) {
2375 io_cqring_add_event(ctx
, sqes
[i
].sqe
->user_data
,
2378 sqes
[i
].has_user
= has_user
;
2379 sqes
[i
].needs_lock
= true;
2380 sqes
[i
].needs_fixed_file
= true;
2381 io_submit_sqe(ctx
, &sqes
[i
], statep
, &link
, true);
2387 io_queue_link_head(ctx
, link
, &link
->submit
, shadow_req
, true);
2389 io_submit_state_end(&state
);
2394 static int io_sq_thread(void *data
)
2396 struct sqe_submit sqes
[IO_IOPOLL_BATCH
];
2397 struct io_ring_ctx
*ctx
= data
;
2398 struct mm_struct
*cur_mm
= NULL
;
2399 mm_segment_t old_fs
;
2402 unsigned long timeout
;
2404 complete(&ctx
->sqo_thread_started
);
2409 timeout
= inflight
= 0;
2410 while (!kthread_should_park()) {
2411 bool all_fixed
, mm_fault
= false;
2415 unsigned nr_events
= 0;
2417 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
2418 io_iopoll_check(ctx
, &nr_events
, 0);
2421 * Normal IO, just pretend everything completed.
2422 * We don't have to poll completions for that.
2424 nr_events
= inflight
;
2427 inflight
-= nr_events
;
2429 timeout
= jiffies
+ ctx
->sq_thread_idle
;
2432 if (!io_get_sqring(ctx
, &sqes
[0])) {
2434 * We're polling. If we're within the defined idle
2435 * period, then let us spin without work before going
2438 if (inflight
|| !time_after(jiffies
, timeout
)) {
2444 * Drop cur_mm before scheduling, we can't hold it for
2445 * long periods (or over schedule()). Do this before
2446 * adding ourselves to the waitqueue, as the unuse/drop
2455 prepare_to_wait(&ctx
->sqo_wait
, &wait
,
2456 TASK_INTERRUPTIBLE
);
2458 /* Tell userspace we may need a wakeup call */
2459 ctx
->rings
->sq_flags
|= IORING_SQ_NEED_WAKEUP
;
2460 /* make sure to read SQ tail after writing flags */
2463 if (!io_get_sqring(ctx
, &sqes
[0])) {
2464 if (kthread_should_park()) {
2465 finish_wait(&ctx
->sqo_wait
, &wait
);
2468 if (signal_pending(current
))
2469 flush_signals(current
);
2471 finish_wait(&ctx
->sqo_wait
, &wait
);
2473 ctx
->rings
->sq_flags
&= ~IORING_SQ_NEED_WAKEUP
;
2476 finish_wait(&ctx
->sqo_wait
, &wait
);
2478 ctx
->rings
->sq_flags
&= ~IORING_SQ_NEED_WAKEUP
;
2484 if (all_fixed
&& io_sqe_needs_user(sqes
[i
].sqe
))
2488 if (i
== ARRAY_SIZE(sqes
))
2490 } while (io_get_sqring(ctx
, &sqes
[i
]));
2492 /* Unless all new commands are FIXED regions, grab mm */
2493 if (!all_fixed
&& !cur_mm
) {
2494 mm_fault
= !mmget_not_zero(ctx
->sqo_mm
);
2496 use_mm(ctx
->sqo_mm
);
2497 cur_mm
= ctx
->sqo_mm
;
2501 inflight
+= io_submit_sqes(ctx
, sqes
, i
, cur_mm
!= NULL
,
2504 /* Commit SQ ring head once we've consumed all SQEs */
2505 io_commit_sqring(ctx
);
2519 static int io_ring_submit(struct io_ring_ctx
*ctx
, unsigned int to_submit
,
2520 bool block_for_last
)
2522 struct io_submit_state state
, *statep
= NULL
;
2523 struct io_kiocb
*link
= NULL
;
2524 struct io_kiocb
*shadow_req
= NULL
;
2525 bool prev_was_link
= false;
2528 if (to_submit
> IO_PLUG_THRESHOLD
) {
2529 io_submit_state_start(&state
, ctx
, to_submit
);
2533 for (i
= 0; i
< to_submit
; i
++) {
2534 bool force_nonblock
= true;
2535 struct sqe_submit s
;
2537 if (!io_get_sqring(ctx
, &s
))
2541 * If previous wasn't linked and we have a linked command,
2542 * that's the end of the chain. Submit the previous link.
2544 if (!prev_was_link
&& link
) {
2545 io_queue_link_head(ctx
, link
, &link
->submit
, shadow_req
,
2549 prev_was_link
= (s
.sqe
->flags
& IOSQE_IO_LINK
) != 0;
2551 if (link
&& (s
.sqe
->flags
& IOSQE_IO_DRAIN
)) {
2553 shadow_req
= io_get_req(ctx
, NULL
);
2554 shadow_req
->flags
|= (REQ_F_IO_DRAIN
| REQ_F_SHADOW_DRAIN
);
2555 refcount_dec(&shadow_req
->refs
);
2557 shadow_req
->sequence
= s
.sequence
;
2561 s
.needs_lock
= false;
2562 s
.needs_fixed_file
= false;
2566 * The caller will block for events after submit, submit the
2567 * last IO non-blocking. This is either the only IO it's
2568 * submitting, or it already submitted the previous ones. This
2569 * improves performance by avoiding an async punt that we don't
2572 if (block_for_last
&& submit
== to_submit
)
2573 force_nonblock
= false;
2575 io_submit_sqe(ctx
, &s
, statep
, &link
, force_nonblock
);
2577 io_commit_sqring(ctx
);
2580 io_queue_link_head(ctx
, link
, &link
->submit
, shadow_req
,
2583 io_submit_state_end(statep
);
2589 * Wait until events become available, if we don't already have some. The
2590 * application must reap them itself, as they reside on the shared cq ring.
2592 static int io_cqring_wait(struct io_ring_ctx
*ctx
, int min_events
,
2593 const sigset_t __user
*sig
, size_t sigsz
)
2595 struct io_rings
*rings
= ctx
->rings
;
2598 if (io_cqring_events(rings
) >= min_events
)
2602 #ifdef CONFIG_COMPAT
2603 if (in_compat_syscall())
2604 ret
= set_compat_user_sigmask((const compat_sigset_t __user
*)sig
,
2608 ret
= set_user_sigmask(sig
, sigsz
);
2614 ret
= wait_event_interruptible(ctx
->wait
, io_cqring_events(rings
) >= min_events
);
2615 restore_saved_sigmask_unless(ret
== -ERESTARTSYS
);
2616 if (ret
== -ERESTARTSYS
)
2619 return READ_ONCE(rings
->cq
.head
) == READ_ONCE(rings
->cq
.tail
) ? ret
: 0;
2622 static void __io_sqe_files_unregister(struct io_ring_ctx
*ctx
)
2624 #if defined(CONFIG_UNIX)
2625 if (ctx
->ring_sock
) {
2626 struct sock
*sock
= ctx
->ring_sock
->sk
;
2627 struct sk_buff
*skb
;
2629 while ((skb
= skb_dequeue(&sock
->sk_receive_queue
)) != NULL
)
2635 for (i
= 0; i
< ctx
->nr_user_files
; i
++)
2636 fput(ctx
->user_files
[i
]);
2640 static int io_sqe_files_unregister(struct io_ring_ctx
*ctx
)
2642 if (!ctx
->user_files
)
2645 __io_sqe_files_unregister(ctx
);
2646 kfree(ctx
->user_files
);
2647 ctx
->user_files
= NULL
;
2648 ctx
->nr_user_files
= 0;
2652 static void io_sq_thread_stop(struct io_ring_ctx
*ctx
)
2654 if (ctx
->sqo_thread
) {
2655 wait_for_completion(&ctx
->sqo_thread_started
);
2657 * The park is a bit of a work-around, without it we get
2658 * warning spews on shutdown with SQPOLL set and affinity
2659 * set to a single CPU.
2661 kthread_park(ctx
->sqo_thread
);
2662 kthread_stop(ctx
->sqo_thread
);
2663 ctx
->sqo_thread
= NULL
;
2667 static void io_finish_async(struct io_ring_ctx
*ctx
)
2671 io_sq_thread_stop(ctx
);
2673 for (i
= 0; i
< ARRAY_SIZE(ctx
->sqo_wq
); i
++) {
2674 if (ctx
->sqo_wq
[i
]) {
2675 destroy_workqueue(ctx
->sqo_wq
[i
]);
2676 ctx
->sqo_wq
[i
] = NULL
;
2681 #if defined(CONFIG_UNIX)
2682 static void io_destruct_skb(struct sk_buff
*skb
)
2684 struct io_ring_ctx
*ctx
= skb
->sk
->sk_user_data
;
2686 io_finish_async(ctx
);
2687 unix_destruct_scm(skb
);
2691 * Ensure the UNIX gc is aware of our file set, so we are certain that
2692 * the io_uring can be safely unregistered on process exit, even if we have
2693 * loops in the file referencing.
2695 static int __io_sqe_files_scm(struct io_ring_ctx
*ctx
, int nr
, int offset
)
2697 struct sock
*sk
= ctx
->ring_sock
->sk
;
2698 struct scm_fp_list
*fpl
;
2699 struct sk_buff
*skb
;
2702 if (!capable(CAP_SYS_RESOURCE
) && !capable(CAP_SYS_ADMIN
)) {
2703 unsigned long inflight
= ctx
->user
->unix_inflight
+ nr
;
2705 if (inflight
> task_rlimit(current
, RLIMIT_NOFILE
))
2709 fpl
= kzalloc(sizeof(*fpl
), GFP_KERNEL
);
2713 skb
= alloc_skb(0, GFP_KERNEL
);
2720 skb
->destructor
= io_destruct_skb
;
2722 fpl
->user
= get_uid(ctx
->user
);
2723 for (i
= 0; i
< nr
; i
++) {
2724 fpl
->fp
[i
] = get_file(ctx
->user_files
[i
+ offset
]);
2725 unix_inflight(fpl
->user
, fpl
->fp
[i
]);
2728 fpl
->max
= fpl
->count
= nr
;
2729 UNIXCB(skb
).fp
= fpl
;
2730 refcount_add(skb
->truesize
, &sk
->sk_wmem_alloc
);
2731 skb_queue_head(&sk
->sk_receive_queue
, skb
);
2733 for (i
= 0; i
< nr
; i
++)
2740 * If UNIX sockets are enabled, fd passing can cause a reference cycle which
2741 * causes regular reference counting to break down. We rely on the UNIX
2742 * garbage collection to take care of this problem for us.
2744 static int io_sqe_files_scm(struct io_ring_ctx
*ctx
)
2746 unsigned left
, total
;
2750 left
= ctx
->nr_user_files
;
2752 unsigned this_files
= min_t(unsigned, left
, SCM_MAX_FD
);
2754 ret
= __io_sqe_files_scm(ctx
, this_files
, total
);
2758 total
+= this_files
;
2764 while (total
< ctx
->nr_user_files
) {
2765 fput(ctx
->user_files
[total
]);
2772 static int io_sqe_files_scm(struct io_ring_ctx
*ctx
)
2778 static int io_sqe_files_register(struct io_ring_ctx
*ctx
, void __user
*arg
,
2781 __s32 __user
*fds
= (__s32 __user
*) arg
;
2785 if (ctx
->user_files
)
2789 if (nr_args
> IORING_MAX_FIXED_FILES
)
2792 ctx
->user_files
= kcalloc(nr_args
, sizeof(struct file
*), GFP_KERNEL
);
2793 if (!ctx
->user_files
)
2796 for (i
= 0; i
< nr_args
; i
++) {
2798 if (copy_from_user(&fd
, &fds
[i
], sizeof(fd
)))
2801 ctx
->user_files
[i
] = fget(fd
);
2804 if (!ctx
->user_files
[i
])
2807 * Don't allow io_uring instances to be registered. If UNIX
2808 * isn't enabled, then this causes a reference cycle and this
2809 * instance can never get freed. If UNIX is enabled we'll
2810 * handle it just fine, but there's still no point in allowing
2811 * a ring fd as it doesn't support regular read/write anyway.
2813 if (ctx
->user_files
[i
]->f_op
== &io_uring_fops
) {
2814 fput(ctx
->user_files
[i
]);
2817 ctx
->nr_user_files
++;
2822 for (i
= 0; i
< ctx
->nr_user_files
; i
++)
2823 fput(ctx
->user_files
[i
]);
2825 kfree(ctx
->user_files
);
2826 ctx
->user_files
= NULL
;
2827 ctx
->nr_user_files
= 0;
2831 ret
= io_sqe_files_scm(ctx
);
2833 io_sqe_files_unregister(ctx
);
2838 static int io_sq_offload_start(struct io_ring_ctx
*ctx
,
2839 struct io_uring_params
*p
)
2843 init_waitqueue_head(&ctx
->sqo_wait
);
2844 mmgrab(current
->mm
);
2845 ctx
->sqo_mm
= current
->mm
;
2847 if (ctx
->flags
& IORING_SETUP_SQPOLL
) {
2849 if (!capable(CAP_SYS_ADMIN
))
2852 ctx
->sq_thread_idle
= msecs_to_jiffies(p
->sq_thread_idle
);
2853 if (!ctx
->sq_thread_idle
)
2854 ctx
->sq_thread_idle
= HZ
;
2856 if (p
->flags
& IORING_SETUP_SQ_AFF
) {
2857 int cpu
= p
->sq_thread_cpu
;
2860 if (cpu
>= nr_cpu_ids
)
2862 if (!cpu_online(cpu
))
2865 ctx
->sqo_thread
= kthread_create_on_cpu(io_sq_thread
,
2869 ctx
->sqo_thread
= kthread_create(io_sq_thread
, ctx
,
2872 if (IS_ERR(ctx
->sqo_thread
)) {
2873 ret
= PTR_ERR(ctx
->sqo_thread
);
2874 ctx
->sqo_thread
= NULL
;
2877 wake_up_process(ctx
->sqo_thread
);
2878 } else if (p
->flags
& IORING_SETUP_SQ_AFF
) {
2879 /* Can't have SQ_AFF without SQPOLL */
2884 /* Do QD, or 2 * CPUS, whatever is smallest */
2885 ctx
->sqo_wq
[0] = alloc_workqueue("io_ring-wq",
2886 WQ_UNBOUND
| WQ_FREEZABLE
,
2887 min(ctx
->sq_entries
- 1, 2 * num_online_cpus()));
2888 if (!ctx
->sqo_wq
[0]) {
2894 * This is for buffered writes, where we want to limit the parallelism
2895 * due to file locking in file systems. As "normal" buffered writes
2896 * should parellelize on writeout quite nicely, limit us to having 2
2897 * pending. This avoids massive contention on the inode when doing
2898 * buffered async writes.
2900 ctx
->sqo_wq
[1] = alloc_workqueue("io_ring-write-wq",
2901 WQ_UNBOUND
| WQ_FREEZABLE
, 2);
2902 if (!ctx
->sqo_wq
[1]) {
2909 io_finish_async(ctx
);
2910 mmdrop(ctx
->sqo_mm
);
2915 static void io_unaccount_mem(struct user_struct
*user
, unsigned long nr_pages
)
2917 atomic_long_sub(nr_pages
, &user
->locked_vm
);
2920 static int io_account_mem(struct user_struct
*user
, unsigned long nr_pages
)
2922 unsigned long page_limit
, cur_pages
, new_pages
;
2924 /* Don't allow more pages than we can safely lock */
2925 page_limit
= rlimit(RLIMIT_MEMLOCK
) >> PAGE_SHIFT
;
2928 cur_pages
= atomic_long_read(&user
->locked_vm
);
2929 new_pages
= cur_pages
+ nr_pages
;
2930 if (new_pages
> page_limit
)
2932 } while (atomic_long_cmpxchg(&user
->locked_vm
, cur_pages
,
2933 new_pages
) != cur_pages
);
2938 static void io_mem_free(void *ptr
)
2945 page
= virt_to_head_page(ptr
);
2946 if (put_page_testzero(page
))
2947 free_compound_page(page
);
2950 static void *io_mem_alloc(size_t size
)
2952 gfp_t gfp_flags
= GFP_KERNEL
| __GFP_ZERO
| __GFP_NOWARN
| __GFP_COMP
|
2955 return (void *) __get_free_pages(gfp_flags
, get_order(size
));
2958 static unsigned long rings_size(unsigned sq_entries
, unsigned cq_entries
,
2961 struct io_rings
*rings
;
2962 size_t off
, sq_array_size
;
2964 off
= struct_size(rings
, cqes
, cq_entries
);
2965 if (off
== SIZE_MAX
)
2969 off
= ALIGN(off
, SMP_CACHE_BYTES
);
2974 sq_array_size
= array_size(sizeof(u32
), sq_entries
);
2975 if (sq_array_size
== SIZE_MAX
)
2978 if (check_add_overflow(off
, sq_array_size
, &off
))
2987 static unsigned long ring_pages(unsigned sq_entries
, unsigned cq_entries
)
2991 pages
= (size_t)1 << get_order(
2992 rings_size(sq_entries
, cq_entries
, NULL
));
2993 pages
+= (size_t)1 << get_order(
2994 array_size(sizeof(struct io_uring_sqe
), sq_entries
));
2999 static int io_sqe_buffer_unregister(struct io_ring_ctx
*ctx
)
3003 if (!ctx
->user_bufs
)
3006 for (i
= 0; i
< ctx
->nr_user_bufs
; i
++) {
3007 struct io_mapped_ubuf
*imu
= &ctx
->user_bufs
[i
];
3009 for (j
= 0; j
< imu
->nr_bvecs
; j
++)
3010 put_user_page(imu
->bvec
[j
].bv_page
);
3012 if (ctx
->account_mem
)
3013 io_unaccount_mem(ctx
->user
, imu
->nr_bvecs
);
3018 kfree(ctx
->user_bufs
);
3019 ctx
->user_bufs
= NULL
;
3020 ctx
->nr_user_bufs
= 0;
3024 static int io_copy_iov(struct io_ring_ctx
*ctx
, struct iovec
*dst
,
3025 void __user
*arg
, unsigned index
)
3027 struct iovec __user
*src
;
3029 #ifdef CONFIG_COMPAT
3031 struct compat_iovec __user
*ciovs
;
3032 struct compat_iovec ciov
;
3034 ciovs
= (struct compat_iovec __user
*) arg
;
3035 if (copy_from_user(&ciov
, &ciovs
[index
], sizeof(ciov
)))
3038 dst
->iov_base
= (void __user
*) (unsigned long) ciov
.iov_base
;
3039 dst
->iov_len
= ciov
.iov_len
;
3043 src
= (struct iovec __user
*) arg
;
3044 if (copy_from_user(dst
, &src
[index
], sizeof(*dst
)))
3049 static int io_sqe_buffer_register(struct io_ring_ctx
*ctx
, void __user
*arg
,
3052 struct vm_area_struct
**vmas
= NULL
;
3053 struct page
**pages
= NULL
;
3054 int i
, j
, got_pages
= 0;
3059 if (!nr_args
|| nr_args
> UIO_MAXIOV
)
3062 ctx
->user_bufs
= kcalloc(nr_args
, sizeof(struct io_mapped_ubuf
),
3064 if (!ctx
->user_bufs
)
3067 for (i
= 0; i
< nr_args
; i
++) {
3068 struct io_mapped_ubuf
*imu
= &ctx
->user_bufs
[i
];
3069 unsigned long off
, start
, end
, ubuf
;
3074 ret
= io_copy_iov(ctx
, &iov
, arg
, i
);
3079 * Don't impose further limits on the size and buffer
3080 * constraints here, we'll -EINVAL later when IO is
3081 * submitted if they are wrong.
3084 if (!iov
.iov_base
|| !iov
.iov_len
)
3087 /* arbitrary limit, but we need something */
3088 if (iov
.iov_len
> SZ_1G
)
3091 ubuf
= (unsigned long) iov
.iov_base
;
3092 end
= (ubuf
+ iov
.iov_len
+ PAGE_SIZE
- 1) >> PAGE_SHIFT
;
3093 start
= ubuf
>> PAGE_SHIFT
;
3094 nr_pages
= end
- start
;
3096 if (ctx
->account_mem
) {
3097 ret
= io_account_mem(ctx
->user
, nr_pages
);
3103 if (!pages
|| nr_pages
> got_pages
) {
3106 pages
= kvmalloc_array(nr_pages
, sizeof(struct page
*),
3108 vmas
= kvmalloc_array(nr_pages
,
3109 sizeof(struct vm_area_struct
*),
3111 if (!pages
|| !vmas
) {
3113 if (ctx
->account_mem
)
3114 io_unaccount_mem(ctx
->user
, nr_pages
);
3117 got_pages
= nr_pages
;
3120 imu
->bvec
= kvmalloc_array(nr_pages
, sizeof(struct bio_vec
),
3124 if (ctx
->account_mem
)
3125 io_unaccount_mem(ctx
->user
, nr_pages
);
3130 down_read(¤t
->mm
->mmap_sem
);
3131 pret
= get_user_pages(ubuf
, nr_pages
,
3132 FOLL_WRITE
| FOLL_LONGTERM
,
3134 if (pret
== nr_pages
) {
3135 /* don't support file backed memory */
3136 for (j
= 0; j
< nr_pages
; j
++) {
3137 struct vm_area_struct
*vma
= vmas
[j
];
3140 !is_file_hugepages(vma
->vm_file
)) {
3146 ret
= pret
< 0 ? pret
: -EFAULT
;
3148 up_read(¤t
->mm
->mmap_sem
);
3151 * if we did partial map, or found file backed vmas,
3152 * release any pages we did get
3155 put_user_pages(pages
, pret
);
3156 if (ctx
->account_mem
)
3157 io_unaccount_mem(ctx
->user
, nr_pages
);
3162 off
= ubuf
& ~PAGE_MASK
;
3164 for (j
= 0; j
< nr_pages
; j
++) {
3167 vec_len
= min_t(size_t, size
, PAGE_SIZE
- off
);
3168 imu
->bvec
[j
].bv_page
= pages
[j
];
3169 imu
->bvec
[j
].bv_len
= vec_len
;
3170 imu
->bvec
[j
].bv_offset
= off
;
3174 /* store original address for later verification */
3176 imu
->len
= iov
.iov_len
;
3177 imu
->nr_bvecs
= nr_pages
;
3179 ctx
->nr_user_bufs
++;
3187 io_sqe_buffer_unregister(ctx
);
3191 static int io_eventfd_register(struct io_ring_ctx
*ctx
, void __user
*arg
)
3193 __s32 __user
*fds
= arg
;
3199 if (copy_from_user(&fd
, fds
, sizeof(*fds
)))
3202 ctx
->cq_ev_fd
= eventfd_ctx_fdget(fd
);
3203 if (IS_ERR(ctx
->cq_ev_fd
)) {
3204 int ret
= PTR_ERR(ctx
->cq_ev_fd
);
3205 ctx
->cq_ev_fd
= NULL
;
3212 static int io_eventfd_unregister(struct io_ring_ctx
*ctx
)
3214 if (ctx
->cq_ev_fd
) {
3215 eventfd_ctx_put(ctx
->cq_ev_fd
);
3216 ctx
->cq_ev_fd
= NULL
;
3223 static void io_ring_ctx_free(struct io_ring_ctx
*ctx
)
3225 io_finish_async(ctx
);
3227 mmdrop(ctx
->sqo_mm
);
3229 io_iopoll_reap_events(ctx
);
3230 io_sqe_buffer_unregister(ctx
);
3231 io_sqe_files_unregister(ctx
);
3232 io_eventfd_unregister(ctx
);
3234 #if defined(CONFIG_UNIX)
3235 if (ctx
->ring_sock
) {
3236 ctx
->ring_sock
->file
= NULL
; /* so that iput() is called */
3237 sock_release(ctx
->ring_sock
);
3241 io_mem_free(ctx
->rings
);
3242 io_mem_free(ctx
->sq_sqes
);
3244 percpu_ref_exit(&ctx
->refs
);
3245 if (ctx
->account_mem
)
3246 io_unaccount_mem(ctx
->user
,
3247 ring_pages(ctx
->sq_entries
, ctx
->cq_entries
));
3248 free_uid(ctx
->user
);
3252 static __poll_t
io_uring_poll(struct file
*file
, poll_table
*wait
)
3254 struct io_ring_ctx
*ctx
= file
->private_data
;
3257 poll_wait(file
, &ctx
->cq_wait
, wait
);
3259 * synchronizes with barrier from wq_has_sleeper call in
3263 if (READ_ONCE(ctx
->rings
->sq
.tail
) - ctx
->cached_sq_head
!=
3264 ctx
->rings
->sq_ring_entries
)
3265 mask
|= EPOLLOUT
| EPOLLWRNORM
;
3266 if (READ_ONCE(ctx
->rings
->sq
.head
) != ctx
->cached_cq_tail
)
3267 mask
|= EPOLLIN
| EPOLLRDNORM
;
3272 static int io_uring_fasync(int fd
, struct file
*file
, int on
)
3274 struct io_ring_ctx
*ctx
= file
->private_data
;
3276 return fasync_helper(fd
, file
, on
, &ctx
->cq_fasync
);
3279 static void io_ring_ctx_wait_and_kill(struct io_ring_ctx
*ctx
)
3281 mutex_lock(&ctx
->uring_lock
);
3282 percpu_ref_kill(&ctx
->refs
);
3283 mutex_unlock(&ctx
->uring_lock
);
3285 io_poll_remove_all(ctx
);
3286 io_iopoll_reap_events(ctx
);
3287 wait_for_completion(&ctx
->ctx_done
);
3288 io_ring_ctx_free(ctx
);
3291 static int io_uring_release(struct inode
*inode
, struct file
*file
)
3293 struct io_ring_ctx
*ctx
= file
->private_data
;
3295 file
->private_data
= NULL
;
3296 io_ring_ctx_wait_and_kill(ctx
);
3300 static int io_uring_mmap(struct file
*file
, struct vm_area_struct
*vma
)
3302 loff_t offset
= (loff_t
) vma
->vm_pgoff
<< PAGE_SHIFT
;
3303 unsigned long sz
= vma
->vm_end
- vma
->vm_start
;
3304 struct io_ring_ctx
*ctx
= file
->private_data
;
3310 case IORING_OFF_SQ_RING
:
3311 case IORING_OFF_CQ_RING
:
3314 case IORING_OFF_SQES
:
3321 page
= virt_to_head_page(ptr
);
3322 if (sz
> (PAGE_SIZE
<< compound_order(page
)))
3325 pfn
= virt_to_phys(ptr
) >> PAGE_SHIFT
;
3326 return remap_pfn_range(vma
, vma
->vm_start
, pfn
, sz
, vma
->vm_page_prot
);
3329 SYSCALL_DEFINE6(io_uring_enter
, unsigned int, fd
, u32
, to_submit
,
3330 u32
, min_complete
, u32
, flags
, const sigset_t __user
*, sig
,
3333 struct io_ring_ctx
*ctx
;
3338 if (flags
& ~(IORING_ENTER_GETEVENTS
| IORING_ENTER_SQ_WAKEUP
))
3346 if (f
.file
->f_op
!= &io_uring_fops
)
3350 ctx
= f
.file
->private_data
;
3351 if (!percpu_ref_tryget(&ctx
->refs
))
3355 * For SQ polling, the thread will do all submissions and completions.
3356 * Just return the requested submit count, and wake the thread if
3360 if (ctx
->flags
& IORING_SETUP_SQPOLL
) {
3361 if (flags
& IORING_ENTER_SQ_WAKEUP
)
3362 wake_up(&ctx
->sqo_wait
);
3363 submitted
= to_submit
;
3364 } else if (to_submit
) {
3365 bool block_for_last
= false;
3367 to_submit
= min(to_submit
, ctx
->sq_entries
);
3370 * Allow last submission to block in a series, IFF the caller
3371 * asked to wait for events and we don't currently have
3372 * enough. This potentially avoids an async punt.
3374 if (to_submit
== min_complete
&&
3375 io_cqring_events(ctx
->rings
) < min_complete
)
3376 block_for_last
= true;
3378 mutex_lock(&ctx
->uring_lock
);
3379 submitted
= io_ring_submit(ctx
, to_submit
, block_for_last
);
3380 mutex_unlock(&ctx
->uring_lock
);
3382 if (flags
& IORING_ENTER_GETEVENTS
) {
3383 unsigned nr_events
= 0;
3385 min_complete
= min(min_complete
, ctx
->cq_entries
);
3387 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
3388 ret
= io_iopoll_check(ctx
, &nr_events
, min_complete
);
3390 ret
= io_cqring_wait(ctx
, min_complete
, sig
, sigsz
);
3394 io_ring_drop_ctx_refs(ctx
, 1);
3397 return submitted
? submitted
: ret
;
3400 static const struct file_operations io_uring_fops
= {
3401 .release
= io_uring_release
,
3402 .mmap
= io_uring_mmap
,
3403 .poll
= io_uring_poll
,
3404 .fasync
= io_uring_fasync
,
3407 static int io_allocate_scq_urings(struct io_ring_ctx
*ctx
,
3408 struct io_uring_params
*p
)
3410 struct io_rings
*rings
;
3411 size_t size
, sq_array_offset
;
3413 size
= rings_size(p
->sq_entries
, p
->cq_entries
, &sq_array_offset
);
3414 if (size
== SIZE_MAX
)
3417 rings
= io_mem_alloc(size
);
3422 ctx
->sq_array
= (u32
*)((char *)rings
+ sq_array_offset
);
3423 rings
->sq_ring_mask
= p
->sq_entries
- 1;
3424 rings
->cq_ring_mask
= p
->cq_entries
- 1;
3425 rings
->sq_ring_entries
= p
->sq_entries
;
3426 rings
->cq_ring_entries
= p
->cq_entries
;
3427 ctx
->sq_mask
= rings
->sq_ring_mask
;
3428 ctx
->cq_mask
= rings
->cq_ring_mask
;
3429 ctx
->sq_entries
= rings
->sq_ring_entries
;
3430 ctx
->cq_entries
= rings
->cq_ring_entries
;
3432 size
= array_size(sizeof(struct io_uring_sqe
), p
->sq_entries
);
3433 if (size
== SIZE_MAX
)
3436 ctx
->sq_sqes
= io_mem_alloc(size
);
3444 * Allocate an anonymous fd, this is what constitutes the application
3445 * visible backing of an io_uring instance. The application mmaps this
3446 * fd to gain access to the SQ/CQ ring details. If UNIX sockets are enabled,
3447 * we have to tie this fd to a socket for file garbage collection purposes.
3449 static int io_uring_get_fd(struct io_ring_ctx
*ctx
)
3454 #if defined(CONFIG_UNIX)
3455 ret
= sock_create_kern(&init_net
, PF_UNIX
, SOCK_RAW
, IPPROTO_IP
,
3461 ret
= get_unused_fd_flags(O_RDWR
| O_CLOEXEC
);
3465 file
= anon_inode_getfile("[io_uring]", &io_uring_fops
, ctx
,
3466 O_RDWR
| O_CLOEXEC
);
3469 ret
= PTR_ERR(file
);
3473 #if defined(CONFIG_UNIX)
3474 ctx
->ring_sock
->file
= file
;
3475 ctx
->ring_sock
->sk
->sk_user_data
= ctx
;
3477 fd_install(ret
, file
);
3480 #if defined(CONFIG_UNIX)
3481 sock_release(ctx
->ring_sock
);
3482 ctx
->ring_sock
= NULL
;
3487 static int io_uring_create(unsigned entries
, struct io_uring_params
*p
)
3489 struct user_struct
*user
= NULL
;
3490 struct io_ring_ctx
*ctx
;
3494 if (!entries
|| entries
> IORING_MAX_ENTRIES
)
3498 * Use twice as many entries for the CQ ring. It's possible for the
3499 * application to drive a higher depth than the size of the SQ ring,
3500 * since the sqes are only used at submission time. This allows for
3501 * some flexibility in overcommitting a bit.
3503 p
->sq_entries
= roundup_pow_of_two(entries
);
3504 p
->cq_entries
= 2 * p
->sq_entries
;
3506 user
= get_uid(current_user());
3507 account_mem
= !capable(CAP_IPC_LOCK
);
3510 ret
= io_account_mem(user
,
3511 ring_pages(p
->sq_entries
, p
->cq_entries
));
3518 ctx
= io_ring_ctx_alloc(p
);
3521 io_unaccount_mem(user
, ring_pages(p
->sq_entries
,
3526 ctx
->compat
= in_compat_syscall();
3527 ctx
->account_mem
= account_mem
;
3530 ret
= io_allocate_scq_urings(ctx
, p
);
3534 ret
= io_sq_offload_start(ctx
, p
);
3538 ret
= io_uring_get_fd(ctx
);
3542 memset(&p
->sq_off
, 0, sizeof(p
->sq_off
));
3543 p
->sq_off
.head
= offsetof(struct io_rings
, sq
.head
);
3544 p
->sq_off
.tail
= offsetof(struct io_rings
, sq
.tail
);
3545 p
->sq_off
.ring_mask
= offsetof(struct io_rings
, sq_ring_mask
);
3546 p
->sq_off
.ring_entries
= offsetof(struct io_rings
, sq_ring_entries
);
3547 p
->sq_off
.flags
= offsetof(struct io_rings
, sq_flags
);
3548 p
->sq_off
.dropped
= offsetof(struct io_rings
, sq_dropped
);
3549 p
->sq_off
.array
= (char *)ctx
->sq_array
- (char *)ctx
->rings
;
3551 memset(&p
->cq_off
, 0, sizeof(p
->cq_off
));
3552 p
->cq_off
.head
= offsetof(struct io_rings
, cq
.head
);
3553 p
->cq_off
.tail
= offsetof(struct io_rings
, cq
.tail
);
3554 p
->cq_off
.ring_mask
= offsetof(struct io_rings
, cq_ring_mask
);
3555 p
->cq_off
.ring_entries
= offsetof(struct io_rings
, cq_ring_entries
);
3556 p
->cq_off
.overflow
= offsetof(struct io_rings
, cq_overflow
);
3557 p
->cq_off
.cqes
= offsetof(struct io_rings
, cqes
);
3559 p
->features
= IORING_FEAT_SINGLE_MMAP
;
3562 io_ring_ctx_wait_and_kill(ctx
);
3567 * Sets up an aio uring context, and returns the fd. Applications asks for a
3568 * ring size, we return the actual sq/cq ring sizes (among other things) in the
3569 * params structure passed in.
3571 static long io_uring_setup(u32 entries
, struct io_uring_params __user
*params
)
3573 struct io_uring_params p
;
3577 if (copy_from_user(&p
, params
, sizeof(p
)))
3579 for (i
= 0; i
< ARRAY_SIZE(p
.resv
); i
++) {
3584 if (p
.flags
& ~(IORING_SETUP_IOPOLL
| IORING_SETUP_SQPOLL
|
3585 IORING_SETUP_SQ_AFF
))
3588 ret
= io_uring_create(entries
, &p
);
3592 if (copy_to_user(params
, &p
, sizeof(p
)))
3598 SYSCALL_DEFINE2(io_uring_setup
, u32
, entries
,
3599 struct io_uring_params __user
*, params
)
3601 return io_uring_setup(entries
, params
);
3604 static int __io_uring_register(struct io_ring_ctx
*ctx
, unsigned opcode
,
3605 void __user
*arg
, unsigned nr_args
)
3606 __releases(ctx
->uring_lock
)
3607 __acquires(ctx
->uring_lock
)
3612 * We're inside the ring mutex, if the ref is already dying, then
3613 * someone else killed the ctx or is already going through
3614 * io_uring_register().
3616 if (percpu_ref_is_dying(&ctx
->refs
))
3619 percpu_ref_kill(&ctx
->refs
);
3622 * Drop uring mutex before waiting for references to exit. If another
3623 * thread is currently inside io_uring_enter() it might need to grab
3624 * the uring_lock to make progress. If we hold it here across the drain
3625 * wait, then we can deadlock. It's safe to drop the mutex here, since
3626 * no new references will come in after we've killed the percpu ref.
3628 mutex_unlock(&ctx
->uring_lock
);
3629 wait_for_completion(&ctx
->ctx_done
);
3630 mutex_lock(&ctx
->uring_lock
);
3633 case IORING_REGISTER_BUFFERS
:
3634 ret
= io_sqe_buffer_register(ctx
, arg
, nr_args
);
3636 case IORING_UNREGISTER_BUFFERS
:
3640 ret
= io_sqe_buffer_unregister(ctx
);
3642 case IORING_REGISTER_FILES
:
3643 ret
= io_sqe_files_register(ctx
, arg
, nr_args
);
3645 case IORING_UNREGISTER_FILES
:
3649 ret
= io_sqe_files_unregister(ctx
);
3651 case IORING_REGISTER_EVENTFD
:
3655 ret
= io_eventfd_register(ctx
, arg
);
3657 case IORING_UNREGISTER_EVENTFD
:
3661 ret
= io_eventfd_unregister(ctx
);
3668 /* bring the ctx back to life */
3669 reinit_completion(&ctx
->ctx_done
);
3670 percpu_ref_reinit(&ctx
->refs
);
3674 SYSCALL_DEFINE4(io_uring_register
, unsigned int, fd
, unsigned int, opcode
,
3675 void __user
*, arg
, unsigned int, nr_args
)
3677 struct io_ring_ctx
*ctx
;
3686 if (f
.file
->f_op
!= &io_uring_fops
)
3689 ctx
= f
.file
->private_data
;
3691 mutex_lock(&ctx
->uring_lock
);
3692 ret
= __io_uring_register(ctx
, opcode
, arg
, nr_args
);
3693 mutex_unlock(&ctx
->uring_lock
);
3699 static int __init
io_uring_init(void)
3701 req_cachep
= KMEM_CACHE(io_kiocb
, SLAB_HWCACHE_ALIGN
| SLAB_PANIC
);
3704 __initcall(io_uring_init
);