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/kthread.h>
60 #include <linux/blkdev.h>
61 #include <linux/bvec.h>
62 #include <linux/net.h>
64 #include <net/af_unix.h>
66 #include <linux/anon_inodes.h>
67 #include <linux/sched/mm.h>
68 #include <linux/uaccess.h>
69 #include <linux/nospec.h>
70 #include <linux/sizes.h>
71 #include <linux/hugetlb.h>
72 #include <linux/highmem.h>
74 #define CREATE_TRACE_POINTS
75 #include <trace/events/io_uring.h>
77 #include <uapi/linux/io_uring.h>
82 #define IORING_MAX_ENTRIES 32768
83 #define IORING_MAX_CQ_ENTRIES (2 * IORING_MAX_ENTRIES)
86 * Shift of 9 is 512 entries, or exactly one page on 64-bit archs
88 #define IORING_FILE_TABLE_SHIFT 9
89 #define IORING_MAX_FILES_TABLE (1U << IORING_FILE_TABLE_SHIFT)
90 #define IORING_FILE_TABLE_MASK (IORING_MAX_FILES_TABLE - 1)
91 #define IORING_MAX_FIXED_FILES (64 * IORING_MAX_FILES_TABLE)
94 u32 head ____cacheline_aligned_in_smp
;
95 u32 tail ____cacheline_aligned_in_smp
;
99 * This data is shared with the application through the mmap at offsets
100 * IORING_OFF_SQ_RING and IORING_OFF_CQ_RING.
102 * The offsets to the member fields are published through struct
103 * io_sqring_offsets when calling io_uring_setup.
107 * Head and tail offsets into the ring; the offsets need to be
108 * masked to get valid indices.
110 * The kernel controls head of the sq ring and the tail of the cq ring,
111 * and the application controls tail of the sq ring and the head of the
114 struct io_uring sq
, cq
;
116 * Bitmasks to apply to head and tail offsets (constant, equals
119 u32 sq_ring_mask
, cq_ring_mask
;
120 /* Ring sizes (constant, power of 2) */
121 u32 sq_ring_entries
, cq_ring_entries
;
123 * Number of invalid entries dropped by the kernel due to
124 * invalid index stored in array
126 * Written by the kernel, shouldn't be modified by the
127 * application (i.e. get number of "new events" by comparing to
130 * After a new SQ head value was read by the application this
131 * counter includes all submissions that were dropped reaching
132 * the new SQ head (and possibly more).
138 * Written by the kernel, shouldn't be modified by the
141 * The application needs a full memory barrier before checking
142 * for IORING_SQ_NEED_WAKEUP after updating the sq tail.
146 * Number of completion events lost because the queue was full;
147 * this should be avoided by the application by making sure
148 * there are not more requests pending than there is space in
149 * the completion queue.
151 * Written by the kernel, shouldn't be modified by the
152 * application (i.e. get number of "new events" by comparing to
155 * As completion events come in out of order this counter is not
156 * ordered with any other data.
160 * Ring buffer of completion events.
162 * The kernel writes completion events fresh every time they are
163 * produced, so the application is allowed to modify pending
166 struct io_uring_cqe cqes
[] ____cacheline_aligned_in_smp
;
169 struct io_mapped_ubuf
{
172 struct bio_vec
*bvec
;
173 unsigned int nr_bvecs
;
176 struct fixed_file_table
{
182 struct percpu_ref refs
;
183 } ____cacheline_aligned_in_smp
;
189 bool cq_overflow_flushed
;
193 * Ring buffer of indices into array of io_uring_sqe, which is
194 * mmapped by the application using the IORING_OFF_SQES offset.
196 * This indirection could e.g. be used to assign fixed
197 * io_uring_sqe entries to operations and only submit them to
198 * the queue when needed.
200 * The kernel modifies neither the indices array nor the entries
204 unsigned cached_sq_head
;
207 unsigned sq_thread_idle
;
208 unsigned cached_sq_dropped
;
209 atomic_t cached_cq_overflow
;
210 struct io_uring_sqe
*sq_sqes
;
212 struct list_head defer_list
;
213 struct list_head timeout_list
;
214 struct list_head cq_overflow_list
;
216 wait_queue_head_t inflight_wait
;
217 } ____cacheline_aligned_in_smp
;
219 struct io_rings
*rings
;
223 struct task_struct
*sqo_thread
; /* if using sq thread polling */
224 struct mm_struct
*sqo_mm
;
225 wait_queue_head_t sqo_wait
;
228 * If used, fixed file set. Writers must ensure that ->refs is dead,
229 * readers must ensure that ->refs is alive as long as the file* is
230 * used. Only updated through io_uring_register(2).
232 struct fixed_file_table
*file_table
;
233 unsigned nr_user_files
;
235 /* if used, fixed mapped user buffers */
236 unsigned nr_user_bufs
;
237 struct io_mapped_ubuf
*user_bufs
;
239 struct user_struct
*user
;
241 const struct cred
*creds
;
243 /* 0 is for ctx quiesce/reinit/free, 1 is for sqo_thread started */
244 struct completion
*completions
;
246 /* if all else fails... */
247 struct io_kiocb
*fallback_req
;
249 #if defined(CONFIG_UNIX)
250 struct socket
*ring_sock
;
254 unsigned cached_cq_tail
;
257 atomic_t cq_timeouts
;
258 struct wait_queue_head cq_wait
;
259 struct fasync_struct
*cq_fasync
;
260 struct eventfd_ctx
*cq_ev_fd
;
261 } ____cacheline_aligned_in_smp
;
264 struct mutex uring_lock
;
265 wait_queue_head_t wait
;
266 } ____cacheline_aligned_in_smp
;
269 spinlock_t completion_lock
;
270 bool poll_multi_file
;
272 * ->poll_list is protected by the ctx->uring_lock for
273 * io_uring instances that don't use IORING_SETUP_SQPOLL.
274 * For SQPOLL, only the single threaded io_sq_thread() will
275 * manipulate the list, hence no extra locking is needed there.
277 struct list_head poll_list
;
278 struct hlist_head
*cancel_hash
;
279 unsigned cancel_hash_bits
;
281 spinlock_t inflight_lock
;
282 struct list_head inflight_list
;
283 } ____cacheline_aligned_in_smp
;
287 * First field must be the file pointer in all the
288 * iocb unions! See also 'struct kiocb' in <linux/fs.h>
290 struct io_poll_iocb
{
292 struct wait_queue_head
*head
;
296 struct wait_queue_entry
*wait
;
299 struct io_timeout_data
{
300 struct io_kiocb
*req
;
301 struct hrtimer timer
;
302 struct timespec64 ts
;
303 enum hrtimer_mode mode
;
307 struct io_async_connect
{
308 struct sockaddr_storage address
;
311 struct io_async_msghdr
{
312 struct iovec fast_iov
[UIO_FASTIOV
];
314 struct sockaddr __user
*uaddr
;
319 struct iovec fast_iov
[UIO_FASTIOV
];
325 struct io_async_ctx
{
326 struct io_uring_sqe sqe
;
328 struct io_async_rw rw
;
329 struct io_async_msghdr msg
;
330 struct io_async_connect connect
;
331 struct io_timeout_data timeout
;
336 * NOTE! Each of the iocb union members has the file pointer
337 * as the first entry in their struct definition. So you can
338 * access the file pointer through any of the sub-structs,
339 * or directly as just 'ki_filp' in this struct.
345 struct io_poll_iocb poll
;
348 const struct io_uring_sqe
*sqe
;
349 struct io_async_ctx
*io
;
350 struct file
*ring_file
;
354 bool needs_fixed_file
;
356 struct io_ring_ctx
*ctx
;
358 struct list_head list
;
359 struct hlist_node hash_node
;
361 struct list_head link_list
;
364 #define REQ_F_NOWAIT 1 /* must not punt to workers */
365 #define REQ_F_IOPOLL_COMPLETED 2 /* polled IO has completed */
366 #define REQ_F_FIXED_FILE 4 /* ctx owns file */
367 #define REQ_F_LINK_NEXT 8 /* already grabbed next link */
368 #define REQ_F_IO_DRAIN 16 /* drain existing IO first */
369 #define REQ_F_IO_DRAINED 32 /* drain done */
370 #define REQ_F_LINK 64 /* linked sqes */
371 #define REQ_F_LINK_TIMEOUT 128 /* has linked timeout */
372 #define REQ_F_FAIL_LINK 256 /* fail rest of links */
373 #define REQ_F_DRAIN_LINK 512 /* link should be fully drained */
374 #define REQ_F_TIMEOUT 1024 /* timeout request */
375 #define REQ_F_ISREG 2048 /* regular file */
376 #define REQ_F_MUST_PUNT 4096 /* must be punted even for NONBLOCK */
377 #define REQ_F_TIMEOUT_NOSEQ 8192 /* no timeout sequence */
378 #define REQ_F_INFLIGHT 16384 /* on inflight list */
379 #define REQ_F_COMP_LOCKED 32768 /* completion under lock */
384 struct list_head inflight_entry
;
386 struct io_wq_work work
;
389 #define IO_PLUG_THRESHOLD 2
390 #define IO_IOPOLL_BATCH 8
392 struct io_submit_state
{
393 struct blk_plug plug
;
396 * io_kiocb alloc cache
398 void *reqs
[IO_IOPOLL_BATCH
];
399 unsigned int free_reqs
;
400 unsigned int cur_req
;
403 * File reference cache
407 unsigned int has_refs
;
408 unsigned int used_refs
;
409 unsigned int ios_left
;
412 static void io_wq_submit_work(struct io_wq_work
**workptr
);
413 static void io_cqring_fill_event(struct io_kiocb
*req
, long res
);
414 static void __io_free_req(struct io_kiocb
*req
);
415 static void io_put_req(struct io_kiocb
*req
);
416 static void io_double_put_req(struct io_kiocb
*req
);
417 static void __io_double_put_req(struct io_kiocb
*req
);
418 static struct io_kiocb
*io_prep_linked_timeout(struct io_kiocb
*req
);
419 static void io_queue_linked_timeout(struct io_kiocb
*req
);
421 static struct kmem_cache
*req_cachep
;
423 static const struct file_operations io_uring_fops
;
425 struct sock
*io_uring_get_socket(struct file
*file
)
427 #if defined(CONFIG_UNIX)
428 if (file
->f_op
== &io_uring_fops
) {
429 struct io_ring_ctx
*ctx
= file
->private_data
;
431 return ctx
->ring_sock
->sk
;
436 EXPORT_SYMBOL(io_uring_get_socket
);
438 static void io_ring_ctx_ref_free(struct percpu_ref
*ref
)
440 struct io_ring_ctx
*ctx
= container_of(ref
, struct io_ring_ctx
, refs
);
442 complete(&ctx
->completions
[0]);
445 static struct io_ring_ctx
*io_ring_ctx_alloc(struct io_uring_params
*p
)
447 struct io_ring_ctx
*ctx
;
450 ctx
= kzalloc(sizeof(*ctx
), GFP_KERNEL
);
454 ctx
->fallback_req
= kmem_cache_alloc(req_cachep
, GFP_KERNEL
);
455 if (!ctx
->fallback_req
)
458 ctx
->completions
= kmalloc(2 * sizeof(struct completion
), GFP_KERNEL
);
459 if (!ctx
->completions
)
463 * Use 5 bits less than the max cq entries, that should give us around
464 * 32 entries per hash list if totally full and uniformly spread.
466 hash_bits
= ilog2(p
->cq_entries
);
470 ctx
->cancel_hash_bits
= hash_bits
;
471 ctx
->cancel_hash
= kmalloc((1U << hash_bits
) * sizeof(struct hlist_head
),
473 if (!ctx
->cancel_hash
)
475 __hash_init(ctx
->cancel_hash
, 1U << hash_bits
);
477 if (percpu_ref_init(&ctx
->refs
, io_ring_ctx_ref_free
,
478 PERCPU_REF_ALLOW_REINIT
, GFP_KERNEL
))
481 ctx
->flags
= p
->flags
;
482 init_waitqueue_head(&ctx
->cq_wait
);
483 INIT_LIST_HEAD(&ctx
->cq_overflow_list
);
484 init_completion(&ctx
->completions
[0]);
485 init_completion(&ctx
->completions
[1]);
486 mutex_init(&ctx
->uring_lock
);
487 init_waitqueue_head(&ctx
->wait
);
488 spin_lock_init(&ctx
->completion_lock
);
489 INIT_LIST_HEAD(&ctx
->poll_list
);
490 INIT_LIST_HEAD(&ctx
->defer_list
);
491 INIT_LIST_HEAD(&ctx
->timeout_list
);
492 init_waitqueue_head(&ctx
->inflight_wait
);
493 spin_lock_init(&ctx
->inflight_lock
);
494 INIT_LIST_HEAD(&ctx
->inflight_list
);
497 if (ctx
->fallback_req
)
498 kmem_cache_free(req_cachep
, ctx
->fallback_req
);
499 kfree(ctx
->completions
);
500 kfree(ctx
->cancel_hash
);
505 static inline bool __req_need_defer(struct io_kiocb
*req
)
507 struct io_ring_ctx
*ctx
= req
->ctx
;
509 return req
->sequence
!= ctx
->cached_cq_tail
+ ctx
->cached_sq_dropped
510 + atomic_read(&ctx
->cached_cq_overflow
);
513 static inline bool req_need_defer(struct io_kiocb
*req
)
515 if ((req
->flags
& (REQ_F_IO_DRAIN
|REQ_F_IO_DRAINED
)) == REQ_F_IO_DRAIN
)
516 return __req_need_defer(req
);
521 static struct io_kiocb
*io_get_deferred_req(struct io_ring_ctx
*ctx
)
523 struct io_kiocb
*req
;
525 req
= list_first_entry_or_null(&ctx
->defer_list
, struct io_kiocb
, list
);
526 if (req
&& !req_need_defer(req
)) {
527 list_del_init(&req
->list
);
534 static struct io_kiocb
*io_get_timeout_req(struct io_ring_ctx
*ctx
)
536 struct io_kiocb
*req
;
538 req
= list_first_entry_or_null(&ctx
->timeout_list
, struct io_kiocb
, list
);
540 if (req
->flags
& REQ_F_TIMEOUT_NOSEQ
)
542 if (!__req_need_defer(req
)) {
543 list_del_init(&req
->list
);
551 static void __io_commit_cqring(struct io_ring_ctx
*ctx
)
553 struct io_rings
*rings
= ctx
->rings
;
555 if (ctx
->cached_cq_tail
!= READ_ONCE(rings
->cq
.tail
)) {
556 /* order cqe stores with ring update */
557 smp_store_release(&rings
->cq
.tail
, ctx
->cached_cq_tail
);
559 if (wq_has_sleeper(&ctx
->cq_wait
)) {
560 wake_up_interruptible(&ctx
->cq_wait
);
561 kill_fasync(&ctx
->cq_fasync
, SIGIO
, POLL_IN
);
566 static inline bool io_sqe_needs_user(const struct io_uring_sqe
*sqe
)
568 u8 opcode
= READ_ONCE(sqe
->opcode
);
570 return !(opcode
== IORING_OP_READ_FIXED
||
571 opcode
== IORING_OP_WRITE_FIXED
);
574 static inline bool io_prep_async_work(struct io_kiocb
*req
,
575 struct io_kiocb
**link
)
577 bool do_hashed
= false;
580 switch (req
->sqe
->opcode
) {
581 case IORING_OP_WRITEV
:
582 case IORING_OP_WRITE_FIXED
:
585 case IORING_OP_READV
:
586 case IORING_OP_READ_FIXED
:
587 case IORING_OP_SENDMSG
:
588 case IORING_OP_RECVMSG
:
589 case IORING_OP_ACCEPT
:
590 case IORING_OP_POLL_ADD
:
591 case IORING_OP_CONNECT
:
593 * We know REQ_F_ISREG is not set on some of these
594 * opcodes, but this enables us to keep the check in
597 if (!(req
->flags
& REQ_F_ISREG
))
598 req
->work
.flags
|= IO_WQ_WORK_UNBOUND
;
601 if (io_sqe_needs_user(req
->sqe
))
602 req
->work
.flags
|= IO_WQ_WORK_NEEDS_USER
;
605 *link
= io_prep_linked_timeout(req
);
609 static inline void io_queue_async_work(struct io_kiocb
*req
)
611 struct io_ring_ctx
*ctx
= req
->ctx
;
612 struct io_kiocb
*link
;
615 do_hashed
= io_prep_async_work(req
, &link
);
617 trace_io_uring_queue_async_work(ctx
, do_hashed
, req
, &req
->work
,
620 io_wq_enqueue(ctx
->io_wq
, &req
->work
);
622 io_wq_enqueue_hashed(ctx
->io_wq
, &req
->work
,
623 file_inode(req
->file
));
627 io_queue_linked_timeout(link
);
630 static void io_kill_timeout(struct io_kiocb
*req
)
634 ret
= hrtimer_try_to_cancel(&req
->io
->timeout
.timer
);
636 atomic_inc(&req
->ctx
->cq_timeouts
);
637 list_del_init(&req
->list
);
638 io_cqring_fill_event(req
, 0);
643 static void io_kill_timeouts(struct io_ring_ctx
*ctx
)
645 struct io_kiocb
*req
, *tmp
;
647 spin_lock_irq(&ctx
->completion_lock
);
648 list_for_each_entry_safe(req
, tmp
, &ctx
->timeout_list
, list
)
649 io_kill_timeout(req
);
650 spin_unlock_irq(&ctx
->completion_lock
);
653 static void io_commit_cqring(struct io_ring_ctx
*ctx
)
655 struct io_kiocb
*req
;
657 while ((req
= io_get_timeout_req(ctx
)) != NULL
)
658 io_kill_timeout(req
);
660 __io_commit_cqring(ctx
);
662 while ((req
= io_get_deferred_req(ctx
)) != NULL
) {
663 req
->flags
|= REQ_F_IO_DRAINED
;
664 io_queue_async_work(req
);
668 static struct io_uring_cqe
*io_get_cqring(struct io_ring_ctx
*ctx
)
670 struct io_rings
*rings
= ctx
->rings
;
673 tail
= ctx
->cached_cq_tail
;
675 * writes to the cq entry need to come after reading head; the
676 * control dependency is enough as we're using WRITE_ONCE to
679 if (tail
- READ_ONCE(rings
->cq
.head
) == rings
->cq_ring_entries
)
682 ctx
->cached_cq_tail
++;
683 return &rings
->cqes
[tail
& ctx
->cq_mask
];
686 static void io_cqring_ev_posted(struct io_ring_ctx
*ctx
)
688 if (waitqueue_active(&ctx
->wait
))
690 if (waitqueue_active(&ctx
->sqo_wait
))
691 wake_up(&ctx
->sqo_wait
);
693 eventfd_signal(ctx
->cq_ev_fd
, 1);
696 /* Returns true if there are no backlogged entries after the flush */
697 static bool io_cqring_overflow_flush(struct io_ring_ctx
*ctx
, bool force
)
699 struct io_rings
*rings
= ctx
->rings
;
700 struct io_uring_cqe
*cqe
;
701 struct io_kiocb
*req
;
706 if (list_empty_careful(&ctx
->cq_overflow_list
))
708 if ((ctx
->cached_cq_tail
- READ_ONCE(rings
->cq
.head
) ==
709 rings
->cq_ring_entries
))
713 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
715 /* if force is set, the ring is going away. always drop after that */
717 ctx
->cq_overflow_flushed
= true;
720 while (!list_empty(&ctx
->cq_overflow_list
)) {
721 cqe
= io_get_cqring(ctx
);
725 req
= list_first_entry(&ctx
->cq_overflow_list
, struct io_kiocb
,
727 list_move(&req
->list
, &list
);
729 WRITE_ONCE(cqe
->user_data
, req
->user_data
);
730 WRITE_ONCE(cqe
->res
, req
->result
);
731 WRITE_ONCE(cqe
->flags
, 0);
733 WRITE_ONCE(ctx
->rings
->cq_overflow
,
734 atomic_inc_return(&ctx
->cached_cq_overflow
));
738 io_commit_cqring(ctx
);
739 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
740 io_cqring_ev_posted(ctx
);
742 while (!list_empty(&list
)) {
743 req
= list_first_entry(&list
, struct io_kiocb
, list
);
744 list_del(&req
->list
);
751 static void io_cqring_fill_event(struct io_kiocb
*req
, long res
)
753 struct io_ring_ctx
*ctx
= req
->ctx
;
754 struct io_uring_cqe
*cqe
;
756 trace_io_uring_complete(ctx
, req
->user_data
, res
);
759 * If we can't get a cq entry, userspace overflowed the
760 * submission (by quite a lot). Increment the overflow count in
763 cqe
= io_get_cqring(ctx
);
765 WRITE_ONCE(cqe
->user_data
, req
->user_data
);
766 WRITE_ONCE(cqe
->res
, res
);
767 WRITE_ONCE(cqe
->flags
, 0);
768 } else if (ctx
->cq_overflow_flushed
) {
769 WRITE_ONCE(ctx
->rings
->cq_overflow
,
770 atomic_inc_return(&ctx
->cached_cq_overflow
));
772 refcount_inc(&req
->refs
);
774 list_add_tail(&req
->list
, &ctx
->cq_overflow_list
);
778 static void io_cqring_add_event(struct io_kiocb
*req
, long res
)
780 struct io_ring_ctx
*ctx
= req
->ctx
;
783 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
784 io_cqring_fill_event(req
, res
);
785 io_commit_cqring(ctx
);
786 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
788 io_cqring_ev_posted(ctx
);
791 static inline bool io_is_fallback_req(struct io_kiocb
*req
)
793 return req
== (struct io_kiocb
*)
794 ((unsigned long) req
->ctx
->fallback_req
& ~1UL);
797 static struct io_kiocb
*io_get_fallback_req(struct io_ring_ctx
*ctx
)
799 struct io_kiocb
*req
;
801 req
= ctx
->fallback_req
;
802 if (!test_and_set_bit_lock(0, (unsigned long *) ctx
->fallback_req
))
808 static struct io_kiocb
*io_get_req(struct io_ring_ctx
*ctx
,
809 struct io_submit_state
*state
)
811 gfp_t gfp
= GFP_KERNEL
| __GFP_NOWARN
;
812 struct io_kiocb
*req
;
814 if (!percpu_ref_tryget(&ctx
->refs
))
818 req
= kmem_cache_alloc(req_cachep
, gfp
);
821 } else if (!state
->free_reqs
) {
825 sz
= min_t(size_t, state
->ios_left
, ARRAY_SIZE(state
->reqs
));
826 ret
= kmem_cache_alloc_bulk(req_cachep
, gfp
, sz
, state
->reqs
);
829 * Bulk alloc is all-or-nothing. If we fail to get a batch,
830 * retry single alloc to be on the safe side.
832 if (unlikely(ret
<= 0)) {
833 state
->reqs
[0] = kmem_cache_alloc(req_cachep
, gfp
);
838 state
->free_reqs
= ret
- 1;
840 req
= state
->reqs
[0];
842 req
= state
->reqs
[state
->cur_req
];
849 req
->ring_file
= NULL
;
853 /* one is dropped after submission, the other at completion */
854 refcount_set(&req
->refs
, 2);
856 INIT_IO_WORK(&req
->work
, io_wq_submit_work
);
859 req
= io_get_fallback_req(ctx
);
862 percpu_ref_put(&ctx
->refs
);
866 static void io_free_req_many(struct io_ring_ctx
*ctx
, void **reqs
, int *nr
)
869 kmem_cache_free_bulk(req_cachep
, *nr
, reqs
);
870 percpu_ref_put_many(&ctx
->refs
, *nr
);
875 static void __io_free_req(struct io_kiocb
*req
)
877 struct io_ring_ctx
*ctx
= req
->ctx
;
881 if (req
->file
&& !(req
->flags
& REQ_F_FIXED_FILE
))
883 if (req
->flags
& REQ_F_INFLIGHT
) {
886 spin_lock_irqsave(&ctx
->inflight_lock
, flags
);
887 list_del(&req
->inflight_entry
);
888 if (waitqueue_active(&ctx
->inflight_wait
))
889 wake_up(&ctx
->inflight_wait
);
890 spin_unlock_irqrestore(&ctx
->inflight_lock
, flags
);
892 percpu_ref_put(&ctx
->refs
);
893 if (likely(!io_is_fallback_req(req
)))
894 kmem_cache_free(req_cachep
, req
);
896 clear_bit_unlock(0, (unsigned long *) ctx
->fallback_req
);
899 static bool io_link_cancel_timeout(struct io_kiocb
*req
)
901 struct io_ring_ctx
*ctx
= req
->ctx
;
904 ret
= hrtimer_try_to_cancel(&req
->io
->timeout
.timer
);
906 io_cqring_fill_event(req
, -ECANCELED
);
907 io_commit_cqring(ctx
);
908 req
->flags
&= ~REQ_F_LINK
;
916 static void io_req_link_next(struct io_kiocb
*req
, struct io_kiocb
**nxtptr
)
918 struct io_ring_ctx
*ctx
= req
->ctx
;
919 bool wake_ev
= false;
921 /* Already got next link */
922 if (req
->flags
& REQ_F_LINK_NEXT
)
926 * The list should never be empty when we are called here. But could
927 * potentially happen if the chain is messed up, check to be on the
930 while (!list_empty(&req
->link_list
)) {
931 struct io_kiocb
*nxt
= list_first_entry(&req
->link_list
,
932 struct io_kiocb
, link_list
);
934 if (unlikely((req
->flags
& REQ_F_LINK_TIMEOUT
) &&
935 (nxt
->flags
& REQ_F_TIMEOUT
))) {
936 list_del_init(&nxt
->link_list
);
937 wake_ev
|= io_link_cancel_timeout(nxt
);
938 req
->flags
&= ~REQ_F_LINK_TIMEOUT
;
942 list_del_init(&req
->link_list
);
943 if (!list_empty(&nxt
->link_list
))
944 nxt
->flags
|= REQ_F_LINK
;
949 req
->flags
|= REQ_F_LINK_NEXT
;
951 io_cqring_ev_posted(ctx
);
955 * Called if REQ_F_LINK is set, and we fail the head request
957 static void io_fail_links(struct io_kiocb
*req
)
959 struct io_ring_ctx
*ctx
= req
->ctx
;
962 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
964 while (!list_empty(&req
->link_list
)) {
965 struct io_kiocb
*link
= list_first_entry(&req
->link_list
,
966 struct io_kiocb
, link_list
);
968 list_del_init(&link
->link_list
);
969 trace_io_uring_fail_link(req
, link
);
971 if ((req
->flags
& REQ_F_LINK_TIMEOUT
) &&
972 link
->sqe
->opcode
== IORING_OP_LINK_TIMEOUT
) {
973 io_link_cancel_timeout(link
);
975 io_cqring_fill_event(link
, -ECANCELED
);
976 __io_double_put_req(link
);
978 req
->flags
&= ~REQ_F_LINK_TIMEOUT
;
981 io_commit_cqring(ctx
);
982 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
983 io_cqring_ev_posted(ctx
);
986 static void io_req_find_next(struct io_kiocb
*req
, struct io_kiocb
**nxt
)
988 if (likely(!(req
->flags
& REQ_F_LINK
)))
992 * If LINK is set, we have dependent requests in this chain. If we
993 * didn't fail this request, queue the first one up, moving any other
994 * dependencies to the next request. In case of failure, fail the rest
997 if (req
->flags
& REQ_F_FAIL_LINK
) {
999 } else if ((req
->flags
& (REQ_F_LINK_TIMEOUT
| REQ_F_COMP_LOCKED
)) ==
1000 REQ_F_LINK_TIMEOUT
) {
1001 struct io_ring_ctx
*ctx
= req
->ctx
;
1002 unsigned long flags
;
1005 * If this is a timeout link, we could be racing with the
1006 * timeout timer. Grab the completion lock for this case to
1007 * protect against that.
1009 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
1010 io_req_link_next(req
, nxt
);
1011 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
1013 io_req_link_next(req
, nxt
);
1017 static void io_free_req(struct io_kiocb
*req
)
1019 struct io_kiocb
*nxt
= NULL
;
1021 io_req_find_next(req
, &nxt
);
1025 io_queue_async_work(nxt
);
1029 * Drop reference to request, return next in chain (if there is one) if this
1030 * was the last reference to this request.
1032 __attribute__((nonnull
))
1033 static void io_put_req_find_next(struct io_kiocb
*req
, struct io_kiocb
**nxtptr
)
1035 io_req_find_next(req
, nxtptr
);
1037 if (refcount_dec_and_test(&req
->refs
))
1041 static void io_put_req(struct io_kiocb
*req
)
1043 if (refcount_dec_and_test(&req
->refs
))
1048 * Must only be used if we don't need to care about links, usually from
1049 * within the completion handling itself.
1051 static void __io_double_put_req(struct io_kiocb
*req
)
1053 /* drop both submit and complete references */
1054 if (refcount_sub_and_test(2, &req
->refs
))
1058 static void io_double_put_req(struct io_kiocb
*req
)
1060 /* drop both submit and complete references */
1061 if (refcount_sub_and_test(2, &req
->refs
))
1065 static unsigned io_cqring_events(struct io_ring_ctx
*ctx
, bool noflush
)
1067 struct io_rings
*rings
= ctx
->rings
;
1070 * noflush == true is from the waitqueue handler, just ensure we wake
1071 * up the task, and the next invocation will flush the entries. We
1072 * cannot safely to it from here.
1074 if (noflush
&& !list_empty(&ctx
->cq_overflow_list
))
1077 io_cqring_overflow_flush(ctx
, false);
1079 /* See comment at the top of this file */
1081 return READ_ONCE(rings
->cq
.tail
) - READ_ONCE(rings
->cq
.head
);
1084 static inline unsigned int io_sqring_entries(struct io_ring_ctx
*ctx
)
1086 struct io_rings
*rings
= ctx
->rings
;
1088 /* make sure SQ entry isn't read before tail */
1089 return smp_load_acquire(&rings
->sq
.tail
) - ctx
->cached_sq_head
;
1093 * Find and free completed poll iocbs
1095 static void io_iopoll_complete(struct io_ring_ctx
*ctx
, unsigned int *nr_events
,
1096 struct list_head
*done
)
1098 void *reqs
[IO_IOPOLL_BATCH
];
1099 struct io_kiocb
*req
;
1103 while (!list_empty(done
)) {
1104 req
= list_first_entry(done
, struct io_kiocb
, list
);
1105 list_del(&req
->list
);
1107 io_cqring_fill_event(req
, req
->result
);
1110 if (refcount_dec_and_test(&req
->refs
)) {
1111 /* If we're not using fixed files, we have to pair the
1112 * completion part with the file put. Use regular
1113 * completions for those, only batch free for fixed
1114 * file and non-linked commands.
1116 if (((req
->flags
& (REQ_F_FIXED_FILE
|REQ_F_LINK
)) ==
1117 REQ_F_FIXED_FILE
) && !io_is_fallback_req(req
) &&
1119 reqs
[to_free
++] = req
;
1120 if (to_free
== ARRAY_SIZE(reqs
))
1121 io_free_req_many(ctx
, reqs
, &to_free
);
1128 io_commit_cqring(ctx
);
1129 io_free_req_many(ctx
, reqs
, &to_free
);
1132 static int io_do_iopoll(struct io_ring_ctx
*ctx
, unsigned int *nr_events
,
1135 struct io_kiocb
*req
, *tmp
;
1141 * Only spin for completions if we don't have multiple devices hanging
1142 * off our complete list, and we're under the requested amount.
1144 spin
= !ctx
->poll_multi_file
&& *nr_events
< min
;
1147 list_for_each_entry_safe(req
, tmp
, &ctx
->poll_list
, list
) {
1148 struct kiocb
*kiocb
= &req
->rw
;
1151 * Move completed entries to our local list. If we find a
1152 * request that requires polling, break out and complete
1153 * the done list first, if we have entries there.
1155 if (req
->flags
& REQ_F_IOPOLL_COMPLETED
) {
1156 list_move_tail(&req
->list
, &done
);
1159 if (!list_empty(&done
))
1162 ret
= kiocb
->ki_filp
->f_op
->iopoll(kiocb
, spin
);
1171 if (!list_empty(&done
))
1172 io_iopoll_complete(ctx
, nr_events
, &done
);
1178 * Poll for a mininum of 'min' events. Note that if min == 0 we consider that a
1179 * non-spinning poll check - we'll still enter the driver poll loop, but only
1180 * as a non-spinning completion check.
1182 static int io_iopoll_getevents(struct io_ring_ctx
*ctx
, unsigned int *nr_events
,
1185 while (!list_empty(&ctx
->poll_list
) && !need_resched()) {
1188 ret
= io_do_iopoll(ctx
, nr_events
, min
);
1191 if (!min
|| *nr_events
>= min
)
1199 * We can't just wait for polled events to come to us, we have to actively
1200 * find and complete them.
1202 static void io_iopoll_reap_events(struct io_ring_ctx
*ctx
)
1204 if (!(ctx
->flags
& IORING_SETUP_IOPOLL
))
1207 mutex_lock(&ctx
->uring_lock
);
1208 while (!list_empty(&ctx
->poll_list
)) {
1209 unsigned int nr_events
= 0;
1211 io_iopoll_getevents(ctx
, &nr_events
, 1);
1214 * Ensure we allow local-to-the-cpu processing to take place,
1215 * in this case we need to ensure that we reap all events.
1219 mutex_unlock(&ctx
->uring_lock
);
1222 static int __io_iopoll_check(struct io_ring_ctx
*ctx
, unsigned *nr_events
,
1225 int iters
= 0, ret
= 0;
1231 * Don't enter poll loop if we already have events pending.
1232 * If we do, we can potentially be spinning for commands that
1233 * already triggered a CQE (eg in error).
1235 if (io_cqring_events(ctx
, false))
1239 * If a submit got punted to a workqueue, we can have the
1240 * application entering polling for a command before it gets
1241 * issued. That app will hold the uring_lock for the duration
1242 * of the poll right here, so we need to take a breather every
1243 * now and then to ensure that the issue has a chance to add
1244 * the poll to the issued list. Otherwise we can spin here
1245 * forever, while the workqueue is stuck trying to acquire the
1248 if (!(++iters
& 7)) {
1249 mutex_unlock(&ctx
->uring_lock
);
1250 mutex_lock(&ctx
->uring_lock
);
1253 if (*nr_events
< min
)
1254 tmin
= min
- *nr_events
;
1256 ret
= io_iopoll_getevents(ctx
, nr_events
, tmin
);
1260 } while (min
&& !*nr_events
&& !need_resched());
1265 static int io_iopoll_check(struct io_ring_ctx
*ctx
, unsigned *nr_events
,
1271 * We disallow the app entering submit/complete with polling, but we
1272 * still need to lock the ring to prevent racing with polled issue
1273 * that got punted to a workqueue.
1275 mutex_lock(&ctx
->uring_lock
);
1276 ret
= __io_iopoll_check(ctx
, nr_events
, min
);
1277 mutex_unlock(&ctx
->uring_lock
);
1281 static void kiocb_end_write(struct io_kiocb
*req
)
1284 * Tell lockdep we inherited freeze protection from submission
1287 if (req
->flags
& REQ_F_ISREG
) {
1288 struct inode
*inode
= file_inode(req
->file
);
1290 __sb_writers_acquired(inode
->i_sb
, SB_FREEZE_WRITE
);
1292 file_end_write(req
->file
);
1295 static void io_complete_rw_common(struct kiocb
*kiocb
, long res
)
1297 struct io_kiocb
*req
= container_of(kiocb
, struct io_kiocb
, rw
);
1299 if (kiocb
->ki_flags
& IOCB_WRITE
)
1300 kiocb_end_write(req
);
1302 if ((req
->flags
& REQ_F_LINK
) && res
!= req
->result
)
1303 req
->flags
|= REQ_F_FAIL_LINK
;
1304 io_cqring_add_event(req
, res
);
1307 static void io_complete_rw(struct kiocb
*kiocb
, long res
, long res2
)
1309 struct io_kiocb
*req
= container_of(kiocb
, struct io_kiocb
, rw
);
1311 io_complete_rw_common(kiocb
, res
);
1315 static struct io_kiocb
*__io_complete_rw(struct kiocb
*kiocb
, long res
)
1317 struct io_kiocb
*req
= container_of(kiocb
, struct io_kiocb
, rw
);
1318 struct io_kiocb
*nxt
= NULL
;
1320 io_complete_rw_common(kiocb
, res
);
1321 io_put_req_find_next(req
, &nxt
);
1326 static void io_complete_rw_iopoll(struct kiocb
*kiocb
, long res
, long res2
)
1328 struct io_kiocb
*req
= container_of(kiocb
, struct io_kiocb
, rw
);
1330 if (kiocb
->ki_flags
& IOCB_WRITE
)
1331 kiocb_end_write(req
);
1333 if ((req
->flags
& REQ_F_LINK
) && res
!= req
->result
)
1334 req
->flags
|= REQ_F_FAIL_LINK
;
1337 req
->flags
|= REQ_F_IOPOLL_COMPLETED
;
1341 * After the iocb has been issued, it's safe to be found on the poll list.
1342 * Adding the kiocb to the list AFTER submission ensures that we don't
1343 * find it from a io_iopoll_getevents() thread before the issuer is done
1344 * accessing the kiocb cookie.
1346 static void io_iopoll_req_issued(struct io_kiocb
*req
)
1348 struct io_ring_ctx
*ctx
= req
->ctx
;
1351 * Track whether we have multiple files in our lists. This will impact
1352 * how we do polling eventually, not spinning if we're on potentially
1353 * different devices.
1355 if (list_empty(&ctx
->poll_list
)) {
1356 ctx
->poll_multi_file
= false;
1357 } else if (!ctx
->poll_multi_file
) {
1358 struct io_kiocb
*list_req
;
1360 list_req
= list_first_entry(&ctx
->poll_list
, struct io_kiocb
,
1362 if (list_req
->rw
.ki_filp
!= req
->rw
.ki_filp
)
1363 ctx
->poll_multi_file
= true;
1367 * For fast devices, IO may have already completed. If it has, add
1368 * it to the front so we find it first.
1370 if (req
->flags
& REQ_F_IOPOLL_COMPLETED
)
1371 list_add(&req
->list
, &ctx
->poll_list
);
1373 list_add_tail(&req
->list
, &ctx
->poll_list
);
1376 static void io_file_put(struct io_submit_state
*state
)
1379 int diff
= state
->has_refs
- state
->used_refs
;
1382 fput_many(state
->file
, diff
);
1388 * Get as many references to a file as we have IOs left in this submission,
1389 * assuming most submissions are for one file, or at least that each file
1390 * has more than one submission.
1392 static struct file
*io_file_get(struct io_submit_state
*state
, int fd
)
1398 if (state
->fd
== fd
) {
1405 state
->file
= fget_many(fd
, state
->ios_left
);
1410 state
->has_refs
= state
->ios_left
;
1411 state
->used_refs
= 1;
1417 * If we tracked the file through the SCM inflight mechanism, we could support
1418 * any file. For now, just ensure that anything potentially problematic is done
1421 static bool io_file_supports_async(struct file
*file
)
1423 umode_t mode
= file_inode(file
)->i_mode
;
1425 if (S_ISBLK(mode
) || S_ISCHR(mode
))
1427 if (S_ISREG(mode
) && file
->f_op
!= &io_uring_fops
)
1433 static int io_prep_rw(struct io_kiocb
*req
, bool force_nonblock
)
1435 const struct io_uring_sqe
*sqe
= req
->sqe
;
1436 struct io_ring_ctx
*ctx
= req
->ctx
;
1437 struct kiocb
*kiocb
= &req
->rw
;
1444 if (S_ISREG(file_inode(req
->file
)->i_mode
))
1445 req
->flags
|= REQ_F_ISREG
;
1447 kiocb
->ki_pos
= READ_ONCE(sqe
->off
);
1448 kiocb
->ki_flags
= iocb_flags(kiocb
->ki_filp
);
1449 kiocb
->ki_hint
= ki_hint_validate(file_write_hint(kiocb
->ki_filp
));
1451 ioprio
= READ_ONCE(sqe
->ioprio
);
1453 ret
= ioprio_check_cap(ioprio
);
1457 kiocb
->ki_ioprio
= ioprio
;
1459 kiocb
->ki_ioprio
= get_current_ioprio();
1461 ret
= kiocb_set_rw_flags(kiocb
, READ_ONCE(sqe
->rw_flags
));
1465 /* don't allow async punt if RWF_NOWAIT was requested */
1466 if ((kiocb
->ki_flags
& IOCB_NOWAIT
) ||
1467 (req
->file
->f_flags
& O_NONBLOCK
))
1468 req
->flags
|= REQ_F_NOWAIT
;
1471 kiocb
->ki_flags
|= IOCB_NOWAIT
;
1473 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
1474 if (!(kiocb
->ki_flags
& IOCB_DIRECT
) ||
1475 !kiocb
->ki_filp
->f_op
->iopoll
)
1478 kiocb
->ki_flags
|= IOCB_HIPRI
;
1479 kiocb
->ki_complete
= io_complete_rw_iopoll
;
1482 if (kiocb
->ki_flags
& IOCB_HIPRI
)
1484 kiocb
->ki_complete
= io_complete_rw
;
1489 static inline void io_rw_done(struct kiocb
*kiocb
, ssize_t ret
)
1495 case -ERESTARTNOINTR
:
1496 case -ERESTARTNOHAND
:
1497 case -ERESTART_RESTARTBLOCK
:
1499 * We can't just restart the syscall, since previously
1500 * submitted sqes may already be in progress. Just fail this
1506 kiocb
->ki_complete(kiocb
, ret
, 0);
1510 static void kiocb_done(struct kiocb
*kiocb
, ssize_t ret
, struct io_kiocb
**nxt
,
1513 if (in_async
&& ret
>= 0 && kiocb
->ki_complete
== io_complete_rw
)
1514 *nxt
= __io_complete_rw(kiocb
, ret
);
1516 io_rw_done(kiocb
, ret
);
1519 static ssize_t
io_import_fixed(struct io_ring_ctx
*ctx
, int rw
,
1520 const struct io_uring_sqe
*sqe
,
1521 struct iov_iter
*iter
)
1523 size_t len
= READ_ONCE(sqe
->len
);
1524 struct io_mapped_ubuf
*imu
;
1525 unsigned index
, buf_index
;
1529 /* attempt to use fixed buffers without having provided iovecs */
1530 if (unlikely(!ctx
->user_bufs
))
1533 buf_index
= READ_ONCE(sqe
->buf_index
);
1534 if (unlikely(buf_index
>= ctx
->nr_user_bufs
))
1537 index
= array_index_nospec(buf_index
, ctx
->nr_user_bufs
);
1538 imu
= &ctx
->user_bufs
[index
];
1539 buf_addr
= READ_ONCE(sqe
->addr
);
1542 if (buf_addr
+ len
< buf_addr
)
1544 /* not inside the mapped region */
1545 if (buf_addr
< imu
->ubuf
|| buf_addr
+ len
> imu
->ubuf
+ imu
->len
)
1549 * May not be a start of buffer, set size appropriately
1550 * and advance us to the beginning.
1552 offset
= buf_addr
- imu
->ubuf
;
1553 iov_iter_bvec(iter
, rw
, imu
->bvec
, imu
->nr_bvecs
, offset
+ len
);
1557 * Don't use iov_iter_advance() here, as it's really slow for
1558 * using the latter parts of a big fixed buffer - it iterates
1559 * over each segment manually. We can cheat a bit here, because
1562 * 1) it's a BVEC iter, we set it up
1563 * 2) all bvecs are PAGE_SIZE in size, except potentially the
1564 * first and last bvec
1566 * So just find our index, and adjust the iterator afterwards.
1567 * If the offset is within the first bvec (or the whole first
1568 * bvec, just use iov_iter_advance(). This makes it easier
1569 * since we can just skip the first segment, which may not
1570 * be PAGE_SIZE aligned.
1572 const struct bio_vec
*bvec
= imu
->bvec
;
1574 if (offset
<= bvec
->bv_len
) {
1575 iov_iter_advance(iter
, offset
);
1577 unsigned long seg_skip
;
1579 /* skip first vec */
1580 offset
-= bvec
->bv_len
;
1581 seg_skip
= 1 + (offset
>> PAGE_SHIFT
);
1583 iter
->bvec
= bvec
+ seg_skip
;
1584 iter
->nr_segs
-= seg_skip
;
1585 iter
->count
-= bvec
->bv_len
+ offset
;
1586 iter
->iov_offset
= offset
& ~PAGE_MASK
;
1593 static ssize_t
io_import_iovec(int rw
, struct io_kiocb
*req
,
1594 struct iovec
**iovec
, struct iov_iter
*iter
)
1596 const struct io_uring_sqe
*sqe
= req
->sqe
;
1597 void __user
*buf
= u64_to_user_ptr(READ_ONCE(sqe
->addr
));
1598 size_t sqe_len
= READ_ONCE(sqe
->len
);
1602 * We're reading ->opcode for the second time, but the first read
1603 * doesn't care whether it's _FIXED or not, so it doesn't matter
1604 * whether ->opcode changes concurrently. The first read does care
1605 * about whether it is a READ or a WRITE, so we don't trust this read
1606 * for that purpose and instead let the caller pass in the read/write
1609 opcode
= READ_ONCE(sqe
->opcode
);
1610 if (opcode
== IORING_OP_READ_FIXED
|| opcode
== IORING_OP_WRITE_FIXED
) {
1612 return io_import_fixed(req
->ctx
, rw
, sqe
, iter
);
1616 struct io_async_rw
*iorw
= &req
->io
->rw
;
1619 iov_iter_init(iter
, rw
, *iovec
, iorw
->nr_segs
, iorw
->size
);
1620 if (iorw
->iov
== iorw
->fast_iov
)
1628 #ifdef CONFIG_COMPAT
1629 if (req
->ctx
->compat
)
1630 return compat_import_iovec(rw
, buf
, sqe_len
, UIO_FASTIOV
,
1634 return import_iovec(rw
, buf
, sqe_len
, UIO_FASTIOV
, iovec
, iter
);
1638 * For files that don't have ->read_iter() and ->write_iter(), handle them
1639 * by looping over ->read() or ->write() manually.
1641 static ssize_t
loop_rw_iter(int rw
, struct file
*file
, struct kiocb
*kiocb
,
1642 struct iov_iter
*iter
)
1647 * Don't support polled IO through this interface, and we can't
1648 * support non-blocking either. For the latter, this just causes
1649 * the kiocb to be handled from an async context.
1651 if (kiocb
->ki_flags
& IOCB_HIPRI
)
1653 if (kiocb
->ki_flags
& IOCB_NOWAIT
)
1656 while (iov_iter_count(iter
)) {
1660 if (!iov_iter_is_bvec(iter
)) {
1661 iovec
= iov_iter_iovec(iter
);
1663 /* fixed buffers import bvec */
1664 iovec
.iov_base
= kmap(iter
->bvec
->bv_page
)
1666 iovec
.iov_len
= min(iter
->count
,
1667 iter
->bvec
->bv_len
- iter
->iov_offset
);
1671 nr
= file
->f_op
->read(file
, iovec
.iov_base
,
1672 iovec
.iov_len
, &kiocb
->ki_pos
);
1674 nr
= file
->f_op
->write(file
, iovec
.iov_base
,
1675 iovec
.iov_len
, &kiocb
->ki_pos
);
1678 if (iov_iter_is_bvec(iter
))
1679 kunmap(iter
->bvec
->bv_page
);
1687 if (nr
!= iovec
.iov_len
)
1689 iov_iter_advance(iter
, nr
);
1695 static void io_req_map_io(struct io_kiocb
*req
, ssize_t io_size
,
1696 struct iovec
*iovec
, struct iovec
*fast_iov
,
1697 struct iov_iter
*iter
)
1699 req
->io
->rw
.nr_segs
= iter
->nr_segs
;
1700 req
->io
->rw
.size
= io_size
;
1701 req
->io
->rw
.iov
= iovec
;
1702 if (!req
->io
->rw
.iov
) {
1703 req
->io
->rw
.iov
= req
->io
->rw
.fast_iov
;
1704 memcpy(req
->io
->rw
.iov
, fast_iov
,
1705 sizeof(struct iovec
) * iter
->nr_segs
);
1709 static int io_setup_async_io(struct io_kiocb
*req
, ssize_t io_size
,
1710 struct iovec
*iovec
, struct iovec
*fast_iov
,
1711 struct iov_iter
*iter
)
1713 req
->io
= kmalloc(sizeof(*req
->io
), GFP_KERNEL
);
1715 io_req_map_io(req
, io_size
, iovec
, fast_iov
, iter
);
1716 memcpy(&req
->io
->sqe
, req
->sqe
, sizeof(req
->io
->sqe
));
1717 req
->sqe
= &req
->io
->sqe
;
1724 static int io_read_prep(struct io_kiocb
*req
, struct iovec
**iovec
,
1725 struct iov_iter
*iter
, bool force_nonblock
)
1729 ret
= io_prep_rw(req
, force_nonblock
);
1733 if (unlikely(!(req
->file
->f_mode
& FMODE_READ
)))
1736 return io_import_iovec(READ
, req
, iovec
, iter
);
1739 static int io_read(struct io_kiocb
*req
, struct io_kiocb
**nxt
,
1740 bool force_nonblock
)
1742 struct iovec inline_vecs
[UIO_FASTIOV
], *iovec
= inline_vecs
;
1743 struct kiocb
*kiocb
= &req
->rw
;
1744 struct iov_iter iter
;
1747 ssize_t io_size
, ret
;
1750 ret
= io_read_prep(req
, &iovec
, &iter
, force_nonblock
);
1754 ret
= io_import_iovec(READ
, req
, &iovec
, &iter
);
1761 if (req
->flags
& REQ_F_LINK
)
1762 req
->result
= io_size
;
1765 * If the file doesn't support async, mark it as REQ_F_MUST_PUNT so
1766 * we know to async punt it even if it was opened O_NONBLOCK
1768 if (force_nonblock
&& !io_file_supports_async(file
)) {
1769 req
->flags
|= REQ_F_MUST_PUNT
;
1773 iov_count
= iov_iter_count(&iter
);
1774 ret
= rw_verify_area(READ
, file
, &kiocb
->ki_pos
, iov_count
);
1778 if (file
->f_op
->read_iter
)
1779 ret2
= call_read_iter(file
, kiocb
, &iter
);
1781 ret2
= loop_rw_iter(READ
, file
, kiocb
, &iter
);
1784 * In case of a short read, punt to async. This can happen
1785 * if we have data partially cached. Alternatively we can
1786 * return the short read, in which case the application will
1787 * need to issue another SQE and wait for it. That SQE will
1788 * need async punt anyway, so it's more efficient to do it
1791 if (force_nonblock
&& !(req
->flags
& REQ_F_NOWAIT
) &&
1792 (req
->flags
& REQ_F_ISREG
) &&
1793 ret2
> 0 && ret2
< io_size
)
1795 /* Catch -EAGAIN return for forced non-blocking submission */
1796 if (!force_nonblock
|| ret2
!= -EAGAIN
) {
1797 kiocb_done(kiocb
, ret2
, nxt
, req
->in_async
);
1800 ret
= io_setup_async_io(req
, io_size
, iovec
,
1801 inline_vecs
, &iter
);
1812 static int io_write_prep(struct io_kiocb
*req
, struct iovec
**iovec
,
1813 struct iov_iter
*iter
, bool force_nonblock
)
1817 ret
= io_prep_rw(req
, force_nonblock
);
1821 if (unlikely(!(req
->file
->f_mode
& FMODE_WRITE
)))
1824 return io_import_iovec(WRITE
, req
, iovec
, iter
);
1827 static int io_write(struct io_kiocb
*req
, struct io_kiocb
**nxt
,
1828 bool force_nonblock
)
1830 struct iovec inline_vecs
[UIO_FASTIOV
], *iovec
= inline_vecs
;
1831 struct kiocb
*kiocb
= &req
->rw
;
1832 struct iov_iter iter
;
1835 ssize_t ret
, io_size
;
1838 ret
= io_write_prep(req
, &iovec
, &iter
, force_nonblock
);
1842 ret
= io_import_iovec(WRITE
, req
, &iovec
, &iter
);
1847 file
= kiocb
->ki_filp
;
1849 if (req
->flags
& REQ_F_LINK
)
1850 req
->result
= io_size
;
1853 * If the file doesn't support async, mark it as REQ_F_MUST_PUNT so
1854 * we know to async punt it even if it was opened O_NONBLOCK
1856 if (force_nonblock
&& !io_file_supports_async(req
->file
)) {
1857 req
->flags
|= REQ_F_MUST_PUNT
;
1861 if (force_nonblock
&& !(kiocb
->ki_flags
& IOCB_DIRECT
))
1864 iov_count
= iov_iter_count(&iter
);
1865 ret
= rw_verify_area(WRITE
, file
, &kiocb
->ki_pos
, iov_count
);
1870 * Open-code file_start_write here to grab freeze protection,
1871 * which will be released by another thread in
1872 * io_complete_rw(). Fool lockdep by telling it the lock got
1873 * released so that it doesn't complain about the held lock when
1874 * we return to userspace.
1876 if (req
->flags
& REQ_F_ISREG
) {
1877 __sb_start_write(file_inode(file
)->i_sb
,
1878 SB_FREEZE_WRITE
, true);
1879 __sb_writers_release(file_inode(file
)->i_sb
,
1882 kiocb
->ki_flags
|= IOCB_WRITE
;
1884 if (file
->f_op
->write_iter
)
1885 ret2
= call_write_iter(file
, kiocb
, &iter
);
1887 ret2
= loop_rw_iter(WRITE
, file
, kiocb
, &iter
);
1888 if (!force_nonblock
|| ret2
!= -EAGAIN
) {
1889 kiocb_done(kiocb
, ret2
, nxt
, req
->in_async
);
1892 ret
= io_setup_async_io(req
, io_size
, iovec
,
1893 inline_vecs
, &iter
);
1905 * IORING_OP_NOP just posts a completion event, nothing else.
1907 static int io_nop(struct io_kiocb
*req
)
1909 struct io_ring_ctx
*ctx
= req
->ctx
;
1911 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
1914 io_cqring_add_event(req
, 0);
1919 static int io_prep_fsync(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
1921 struct io_ring_ctx
*ctx
= req
->ctx
;
1926 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
1928 if (unlikely(sqe
->addr
|| sqe
->ioprio
|| sqe
->buf_index
))
1934 static int io_fsync(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
1935 struct io_kiocb
**nxt
, bool force_nonblock
)
1937 loff_t sqe_off
= READ_ONCE(sqe
->off
);
1938 loff_t sqe_len
= READ_ONCE(sqe
->len
);
1939 loff_t end
= sqe_off
+ sqe_len
;
1940 unsigned fsync_flags
;
1943 fsync_flags
= READ_ONCE(sqe
->fsync_flags
);
1944 if (unlikely(fsync_flags
& ~IORING_FSYNC_DATASYNC
))
1947 ret
= io_prep_fsync(req
, sqe
);
1951 /* fsync always requires a blocking context */
1955 ret
= vfs_fsync_range(req
->rw
.ki_filp
, sqe_off
,
1956 end
> 0 ? end
: LLONG_MAX
,
1957 fsync_flags
& IORING_FSYNC_DATASYNC
);
1959 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
1960 req
->flags
|= REQ_F_FAIL_LINK
;
1961 io_cqring_add_event(req
, ret
);
1962 io_put_req_find_next(req
, nxt
);
1966 static int io_prep_sfr(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
1968 struct io_ring_ctx
*ctx
= req
->ctx
;
1974 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
1976 if (unlikely(sqe
->addr
|| sqe
->ioprio
|| sqe
->buf_index
))
1982 static int io_sync_file_range(struct io_kiocb
*req
,
1983 const struct io_uring_sqe
*sqe
,
1984 struct io_kiocb
**nxt
,
1985 bool force_nonblock
)
1992 ret
= io_prep_sfr(req
, sqe
);
1996 /* sync_file_range always requires a blocking context */
2000 sqe_off
= READ_ONCE(sqe
->off
);
2001 sqe_len
= READ_ONCE(sqe
->len
);
2002 flags
= READ_ONCE(sqe
->sync_range_flags
);
2004 ret
= sync_file_range(req
->rw
.ki_filp
, sqe_off
, sqe_len
, flags
);
2006 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2007 req
->flags
|= REQ_F_FAIL_LINK
;
2008 io_cqring_add_event(req
, ret
);
2009 io_put_req_find_next(req
, nxt
);
2013 static int io_sendmsg_prep(struct io_kiocb
*req
, struct io_async_ctx
*io
)
2015 #if defined(CONFIG_NET)
2016 const struct io_uring_sqe
*sqe
= req
->sqe
;
2017 struct user_msghdr __user
*msg
;
2020 flags
= READ_ONCE(sqe
->msg_flags
);
2021 msg
= (struct user_msghdr __user
*)(unsigned long) READ_ONCE(sqe
->addr
);
2022 return sendmsg_copy_msghdr(&io
->msg
.msg
, msg
, flags
, &io
->msg
.iov
);
2028 static int io_sendmsg(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
2029 struct io_kiocb
**nxt
, bool force_nonblock
)
2031 #if defined(CONFIG_NET)
2032 struct socket
*sock
;
2035 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
2038 sock
= sock_from_file(req
->file
, &ret
);
2040 struct io_async_ctx io
, *copy
;
2041 struct sockaddr_storage addr
;
2042 struct msghdr
*kmsg
;
2045 flags
= READ_ONCE(sqe
->msg_flags
);
2046 if (flags
& MSG_DONTWAIT
)
2047 req
->flags
|= REQ_F_NOWAIT
;
2048 else if (force_nonblock
)
2049 flags
|= MSG_DONTWAIT
;
2052 kmsg
= &req
->io
->msg
.msg
;
2053 kmsg
->msg_name
= &addr
;
2056 kmsg
->msg_name
= &addr
;
2057 io
.msg
.iov
= io
.msg
.fast_iov
;
2058 ret
= io_sendmsg_prep(req
, &io
);
2063 ret
= __sys_sendmsg_sock(sock
, kmsg
, flags
);
2064 if (force_nonblock
&& ret
== -EAGAIN
) {
2065 copy
= kmalloc(sizeof(*copy
), GFP_KERNEL
);
2070 memcpy(©
->msg
, &io
.msg
, sizeof(copy
->msg
));
2072 memcpy(&req
->io
->sqe
, req
->sqe
, sizeof(*req
->sqe
));
2073 req
->sqe
= &req
->io
->sqe
;
2076 if (ret
== -ERESTARTSYS
)
2081 io_cqring_add_event(req
, ret
);
2082 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2083 req
->flags
|= REQ_F_FAIL_LINK
;
2084 io_put_req_find_next(req
, nxt
);
2091 static int io_recvmsg_prep(struct io_kiocb
*req
, struct io_async_ctx
*io
)
2093 #if defined(CONFIG_NET)
2094 const struct io_uring_sqe
*sqe
= req
->sqe
;
2095 struct user_msghdr __user
*msg
;
2098 flags
= READ_ONCE(sqe
->msg_flags
);
2099 msg
= (struct user_msghdr __user
*)(unsigned long) READ_ONCE(sqe
->addr
);
2100 return recvmsg_copy_msghdr(&io
->msg
.msg
, msg
, flags
, &io
->msg
.uaddr
,
2107 static int io_recvmsg(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
2108 struct io_kiocb
**nxt
, bool force_nonblock
)
2110 #if defined(CONFIG_NET)
2111 struct socket
*sock
;
2114 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
2117 sock
= sock_from_file(req
->file
, &ret
);
2119 struct user_msghdr __user
*msg
;
2120 struct io_async_ctx io
, *copy
;
2121 struct sockaddr_storage addr
;
2122 struct msghdr
*kmsg
;
2125 flags
= READ_ONCE(sqe
->msg_flags
);
2126 if (flags
& MSG_DONTWAIT
)
2127 req
->flags
|= REQ_F_NOWAIT
;
2128 else if (force_nonblock
)
2129 flags
|= MSG_DONTWAIT
;
2131 msg
= (struct user_msghdr __user
*) (unsigned long)
2132 READ_ONCE(sqe
->addr
);
2134 kmsg
= &req
->io
->msg
.msg
;
2135 kmsg
->msg_name
= &addr
;
2138 kmsg
->msg_name
= &addr
;
2139 io
.msg
.iov
= io
.msg
.fast_iov
;
2140 ret
= io_recvmsg_prep(req
, &io
);
2145 ret
= __sys_recvmsg_sock(sock
, kmsg
, msg
, io
.msg
.uaddr
, flags
);
2146 if (force_nonblock
&& ret
== -EAGAIN
) {
2147 copy
= kmalloc(sizeof(*copy
), GFP_KERNEL
);
2152 memcpy(copy
, &io
, sizeof(*copy
));
2154 memcpy(&req
->io
->sqe
, req
->sqe
, sizeof(*req
->sqe
));
2155 req
->sqe
= &req
->io
->sqe
;
2158 if (ret
== -ERESTARTSYS
)
2163 io_cqring_add_event(req
, ret
);
2164 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2165 req
->flags
|= REQ_F_FAIL_LINK
;
2166 io_put_req_find_next(req
, nxt
);
2173 static int io_accept(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
2174 struct io_kiocb
**nxt
, bool force_nonblock
)
2176 #if defined(CONFIG_NET)
2177 struct sockaddr __user
*addr
;
2178 int __user
*addr_len
;
2179 unsigned file_flags
;
2182 if (unlikely(req
->ctx
->flags
& (IORING_SETUP_IOPOLL
|IORING_SETUP_SQPOLL
)))
2184 if (sqe
->ioprio
|| sqe
->len
|| sqe
->buf_index
)
2187 addr
= (struct sockaddr __user
*) (unsigned long) READ_ONCE(sqe
->addr
);
2188 addr_len
= (int __user
*) (unsigned long) READ_ONCE(sqe
->addr2
);
2189 flags
= READ_ONCE(sqe
->accept_flags
);
2190 file_flags
= force_nonblock
? O_NONBLOCK
: 0;
2192 ret
= __sys_accept4_file(req
->file
, file_flags
, addr
, addr_len
, flags
);
2193 if (ret
== -EAGAIN
&& force_nonblock
) {
2194 req
->work
.flags
|= IO_WQ_WORK_NEEDS_FILES
;
2197 if (ret
== -ERESTARTSYS
)
2199 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2200 req
->flags
|= REQ_F_FAIL_LINK
;
2201 io_cqring_add_event(req
, ret
);
2202 io_put_req_find_next(req
, nxt
);
2209 static int io_connect_prep(struct io_kiocb
*req
, struct io_async_ctx
*io
)
2211 #if defined(CONFIG_NET)
2212 const struct io_uring_sqe
*sqe
= req
->sqe
;
2213 struct sockaddr __user
*addr
;
2216 addr
= (struct sockaddr __user
*) (unsigned long) READ_ONCE(sqe
->addr
);
2217 addr_len
= READ_ONCE(sqe
->addr2
);
2218 return move_addr_to_kernel(addr
, addr_len
, &io
->connect
.address
);
2224 static int io_connect(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
2225 struct io_kiocb
**nxt
, bool force_nonblock
)
2227 #if defined(CONFIG_NET)
2228 struct io_async_ctx __io
, *io
;
2229 unsigned file_flags
;
2232 if (unlikely(req
->ctx
->flags
& (IORING_SETUP_IOPOLL
|IORING_SETUP_SQPOLL
)))
2234 if (sqe
->ioprio
|| sqe
->len
|| sqe
->buf_index
|| sqe
->rw_flags
)
2237 addr_len
= READ_ONCE(sqe
->addr2
);
2238 file_flags
= force_nonblock
? O_NONBLOCK
: 0;
2243 ret
= io_connect_prep(req
, &__io
);
2249 ret
= __sys_connect_file(req
->file
, &io
->connect
.address
, addr_len
,
2251 if ((ret
== -EAGAIN
|| ret
== -EINPROGRESS
) && force_nonblock
) {
2252 io
= kmalloc(sizeof(*io
), GFP_KERNEL
);
2257 memcpy(&io
->connect
, &__io
.connect
, sizeof(io
->connect
));
2259 memcpy(&io
->sqe
, req
->sqe
, sizeof(*req
->sqe
));
2260 req
->sqe
= &io
->sqe
;
2263 if (ret
== -ERESTARTSYS
)
2266 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2267 req
->flags
|= REQ_F_FAIL_LINK
;
2268 io_cqring_add_event(req
, ret
);
2269 io_put_req_find_next(req
, nxt
);
2276 static void io_poll_remove_one(struct io_kiocb
*req
)
2278 struct io_poll_iocb
*poll
= &req
->poll
;
2280 spin_lock(&poll
->head
->lock
);
2281 WRITE_ONCE(poll
->canceled
, true);
2282 if (!list_empty(&poll
->wait
->entry
)) {
2283 list_del_init(&poll
->wait
->entry
);
2284 io_queue_async_work(req
);
2286 spin_unlock(&poll
->head
->lock
);
2287 hash_del(&req
->hash_node
);
2290 static void io_poll_remove_all(struct io_ring_ctx
*ctx
)
2292 struct hlist_node
*tmp
;
2293 struct io_kiocb
*req
;
2296 spin_lock_irq(&ctx
->completion_lock
);
2297 for (i
= 0; i
< (1U << ctx
->cancel_hash_bits
); i
++) {
2298 struct hlist_head
*list
;
2300 list
= &ctx
->cancel_hash
[i
];
2301 hlist_for_each_entry_safe(req
, tmp
, list
, hash_node
)
2302 io_poll_remove_one(req
);
2304 spin_unlock_irq(&ctx
->completion_lock
);
2307 static int io_poll_cancel(struct io_ring_ctx
*ctx
, __u64 sqe_addr
)
2309 struct hlist_head
*list
;
2310 struct io_kiocb
*req
;
2312 list
= &ctx
->cancel_hash
[hash_long(sqe_addr
, ctx
->cancel_hash_bits
)];
2313 hlist_for_each_entry(req
, list
, hash_node
) {
2314 if (sqe_addr
== req
->user_data
) {
2315 io_poll_remove_one(req
);
2324 * Find a running poll command that matches one specified in sqe->addr,
2325 * and remove it if found.
2327 static int io_poll_remove(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
2329 struct io_ring_ctx
*ctx
= req
->ctx
;
2332 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
2334 if (sqe
->ioprio
|| sqe
->off
|| sqe
->len
|| sqe
->buf_index
||
2338 spin_lock_irq(&ctx
->completion_lock
);
2339 ret
= io_poll_cancel(ctx
, READ_ONCE(sqe
->addr
));
2340 spin_unlock_irq(&ctx
->completion_lock
);
2342 io_cqring_add_event(req
, ret
);
2343 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2344 req
->flags
|= REQ_F_FAIL_LINK
;
2349 static void io_poll_complete(struct io_kiocb
*req
, __poll_t mask
, int error
)
2351 struct io_ring_ctx
*ctx
= req
->ctx
;
2353 req
->poll
.done
= true;
2354 kfree(req
->poll
.wait
);
2356 io_cqring_fill_event(req
, error
);
2358 io_cqring_fill_event(req
, mangle_poll(mask
));
2359 io_commit_cqring(ctx
);
2362 static void io_poll_complete_work(struct io_wq_work
**workptr
)
2364 struct io_wq_work
*work
= *workptr
;
2365 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
2366 struct io_poll_iocb
*poll
= &req
->poll
;
2367 struct poll_table_struct pt
= { ._key
= poll
->events
};
2368 struct io_ring_ctx
*ctx
= req
->ctx
;
2369 struct io_kiocb
*nxt
= NULL
;
2373 if (work
->flags
& IO_WQ_WORK_CANCEL
) {
2374 WRITE_ONCE(poll
->canceled
, true);
2376 } else if (READ_ONCE(poll
->canceled
)) {
2380 if (ret
!= -ECANCELED
)
2381 mask
= vfs_poll(poll
->file
, &pt
) & poll
->events
;
2384 * Note that ->ki_cancel callers also delete iocb from active_reqs after
2385 * calling ->ki_cancel. We need the ctx_lock roundtrip here to
2386 * synchronize with them. In the cancellation case the list_del_init
2387 * itself is not actually needed, but harmless so we keep it in to
2388 * avoid further branches in the fast path.
2390 spin_lock_irq(&ctx
->completion_lock
);
2391 if (!mask
&& ret
!= -ECANCELED
) {
2392 add_wait_queue(poll
->head
, poll
->wait
);
2393 spin_unlock_irq(&ctx
->completion_lock
);
2396 hash_del(&req
->hash_node
);
2397 io_poll_complete(req
, mask
, ret
);
2398 spin_unlock_irq(&ctx
->completion_lock
);
2400 io_cqring_ev_posted(ctx
);
2402 if (ret
< 0 && req
->flags
& REQ_F_LINK
)
2403 req
->flags
|= REQ_F_FAIL_LINK
;
2404 io_put_req_find_next(req
, &nxt
);
2406 *workptr
= &nxt
->work
;
2409 static int io_poll_wake(struct wait_queue_entry
*wait
, unsigned mode
, int sync
,
2412 struct io_poll_iocb
*poll
= wait
->private;
2413 struct io_kiocb
*req
= container_of(poll
, struct io_kiocb
, poll
);
2414 struct io_ring_ctx
*ctx
= req
->ctx
;
2415 __poll_t mask
= key_to_poll(key
);
2416 unsigned long flags
;
2418 /* for instances that support it check for an event match first: */
2419 if (mask
&& !(mask
& poll
->events
))
2422 list_del_init(&poll
->wait
->entry
);
2425 * Run completion inline if we can. We're using trylock here because
2426 * we are violating the completion_lock -> poll wq lock ordering.
2427 * If we have a link timeout we're going to need the completion_lock
2428 * for finalizing the request, mark us as having grabbed that already.
2430 if (mask
&& spin_trylock_irqsave(&ctx
->completion_lock
, flags
)) {
2431 hash_del(&req
->hash_node
);
2432 io_poll_complete(req
, mask
, 0);
2433 req
->flags
|= REQ_F_COMP_LOCKED
;
2435 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
2437 io_cqring_ev_posted(ctx
);
2439 io_queue_async_work(req
);
2445 struct io_poll_table
{
2446 struct poll_table_struct pt
;
2447 struct io_kiocb
*req
;
2451 static void io_poll_queue_proc(struct file
*file
, struct wait_queue_head
*head
,
2452 struct poll_table_struct
*p
)
2454 struct io_poll_table
*pt
= container_of(p
, struct io_poll_table
, pt
);
2456 if (unlikely(pt
->req
->poll
.head
)) {
2457 pt
->error
= -EINVAL
;
2462 pt
->req
->poll
.head
= head
;
2463 add_wait_queue(head
, pt
->req
->poll
.wait
);
2466 static void io_poll_req_insert(struct io_kiocb
*req
)
2468 struct io_ring_ctx
*ctx
= req
->ctx
;
2469 struct hlist_head
*list
;
2471 list
= &ctx
->cancel_hash
[hash_long(req
->user_data
, ctx
->cancel_hash_bits
)];
2472 hlist_add_head(&req
->hash_node
, list
);
2475 static int io_poll_add(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
2476 struct io_kiocb
**nxt
)
2478 struct io_poll_iocb
*poll
= &req
->poll
;
2479 struct io_ring_ctx
*ctx
= req
->ctx
;
2480 struct io_poll_table ipt
;
2481 bool cancel
= false;
2485 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
2487 if (sqe
->addr
|| sqe
->ioprio
|| sqe
->off
|| sqe
->len
|| sqe
->buf_index
)
2492 poll
->wait
= kmalloc(sizeof(*poll
->wait
), GFP_KERNEL
);
2497 INIT_IO_WORK(&req
->work
, io_poll_complete_work
);
2498 events
= READ_ONCE(sqe
->poll_events
);
2499 poll
->events
= demangle_poll(events
) | EPOLLERR
| EPOLLHUP
;
2500 INIT_HLIST_NODE(&req
->hash_node
);
2504 poll
->canceled
= false;
2506 ipt
.pt
._qproc
= io_poll_queue_proc
;
2507 ipt
.pt
._key
= poll
->events
;
2509 ipt
.error
= -EINVAL
; /* same as no support for IOCB_CMD_POLL */
2511 /* initialized the list so that we can do list_empty checks */
2512 INIT_LIST_HEAD(&poll
->wait
->entry
);
2513 init_waitqueue_func_entry(poll
->wait
, io_poll_wake
);
2514 poll
->wait
->private = poll
;
2516 INIT_LIST_HEAD(&req
->list
);
2518 mask
= vfs_poll(poll
->file
, &ipt
.pt
) & poll
->events
;
2520 spin_lock_irq(&ctx
->completion_lock
);
2521 if (likely(poll
->head
)) {
2522 spin_lock(&poll
->head
->lock
);
2523 if (unlikely(list_empty(&poll
->wait
->entry
))) {
2529 if (mask
|| ipt
.error
)
2530 list_del_init(&poll
->wait
->entry
);
2532 WRITE_ONCE(poll
->canceled
, true);
2533 else if (!poll
->done
) /* actually waiting for an event */
2534 io_poll_req_insert(req
);
2535 spin_unlock(&poll
->head
->lock
);
2537 if (mask
) { /* no async, we'd stolen it */
2539 io_poll_complete(req
, mask
, 0);
2541 spin_unlock_irq(&ctx
->completion_lock
);
2544 io_cqring_ev_posted(ctx
);
2545 io_put_req_find_next(req
, nxt
);
2550 static enum hrtimer_restart
io_timeout_fn(struct hrtimer
*timer
)
2552 struct io_timeout_data
*data
= container_of(timer
,
2553 struct io_timeout_data
, timer
);
2554 struct io_kiocb
*req
= data
->req
;
2555 struct io_ring_ctx
*ctx
= req
->ctx
;
2556 unsigned long flags
;
2558 atomic_inc(&ctx
->cq_timeouts
);
2560 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
2562 * We could be racing with timeout deletion. If the list is empty,
2563 * then timeout lookup already found it and will be handling it.
2565 if (!list_empty(&req
->list
)) {
2566 struct io_kiocb
*prev
;
2569 * Adjust the reqs sequence before the current one because it
2570 * will consume a slot in the cq_ring and the the cq_tail
2571 * pointer will be increased, otherwise other timeout reqs may
2572 * return in advance without waiting for enough wait_nr.
2575 list_for_each_entry_continue_reverse(prev
, &ctx
->timeout_list
, list
)
2577 list_del_init(&req
->list
);
2580 io_cqring_fill_event(req
, -ETIME
);
2581 io_commit_cqring(ctx
);
2582 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
2584 io_cqring_ev_posted(ctx
);
2585 if (req
->flags
& REQ_F_LINK
)
2586 req
->flags
|= REQ_F_FAIL_LINK
;
2588 return HRTIMER_NORESTART
;
2591 static int io_timeout_cancel(struct io_ring_ctx
*ctx
, __u64 user_data
)
2593 struct io_kiocb
*req
;
2596 list_for_each_entry(req
, &ctx
->timeout_list
, list
) {
2597 if (user_data
== req
->user_data
) {
2598 list_del_init(&req
->list
);
2607 ret
= hrtimer_try_to_cancel(&req
->io
->timeout
.timer
);
2611 if (req
->flags
& REQ_F_LINK
)
2612 req
->flags
|= REQ_F_FAIL_LINK
;
2613 io_cqring_fill_event(req
, -ECANCELED
);
2619 * Remove or update an existing timeout command
2621 static int io_timeout_remove(struct io_kiocb
*req
,
2622 const struct io_uring_sqe
*sqe
)
2624 struct io_ring_ctx
*ctx
= req
->ctx
;
2628 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
2630 if (sqe
->flags
|| sqe
->ioprio
|| sqe
->buf_index
|| sqe
->len
)
2632 flags
= READ_ONCE(sqe
->timeout_flags
);
2636 spin_lock_irq(&ctx
->completion_lock
);
2637 ret
= io_timeout_cancel(ctx
, READ_ONCE(sqe
->addr
));
2639 io_cqring_fill_event(req
, ret
);
2640 io_commit_cqring(ctx
);
2641 spin_unlock_irq(&ctx
->completion_lock
);
2642 io_cqring_ev_posted(ctx
);
2643 if (ret
< 0 && req
->flags
& REQ_F_LINK
)
2644 req
->flags
|= REQ_F_FAIL_LINK
;
2649 static int io_timeout_prep(struct io_kiocb
*req
, struct io_async_ctx
*io
,
2650 bool is_timeout_link
)
2652 const struct io_uring_sqe
*sqe
= req
->sqe
;
2653 struct io_timeout_data
*data
;
2656 if (unlikely(req
->ctx
->flags
& IORING_SETUP_IOPOLL
))
2658 if (sqe
->ioprio
|| sqe
->buf_index
|| sqe
->len
!= 1)
2660 if (sqe
->off
&& is_timeout_link
)
2662 flags
= READ_ONCE(sqe
->timeout_flags
);
2663 if (flags
& ~IORING_TIMEOUT_ABS
)
2666 data
= &io
->timeout
;
2668 req
->flags
|= REQ_F_TIMEOUT
;
2670 if (get_timespec64(&data
->ts
, u64_to_user_ptr(sqe
->addr
)))
2673 if (flags
& IORING_TIMEOUT_ABS
)
2674 data
->mode
= HRTIMER_MODE_ABS
;
2676 data
->mode
= HRTIMER_MODE_REL
;
2678 hrtimer_init(&data
->timer
, CLOCK_MONOTONIC
, data
->mode
);
2683 static int io_timeout(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
)
2686 struct io_ring_ctx
*ctx
= req
->ctx
;
2687 struct io_timeout_data
*data
;
2688 struct io_async_ctx
*io
;
2689 struct list_head
*entry
;
2696 io
= kmalloc(sizeof(*io
), GFP_KERNEL
);
2699 ret
= io_timeout_prep(req
, io
, false);
2705 data
= &req
->io
->timeout
;
2708 * sqe->off holds how many events that need to occur for this
2709 * timeout event to be satisfied. If it isn't set, then this is
2710 * a pure timeout request, sequence isn't used.
2712 count
= READ_ONCE(sqe
->off
);
2714 req
->flags
|= REQ_F_TIMEOUT_NOSEQ
;
2715 spin_lock_irq(&ctx
->completion_lock
);
2716 entry
= ctx
->timeout_list
.prev
;
2720 req
->sequence
= ctx
->cached_sq_head
+ count
- 1;
2721 data
->seq_offset
= count
;
2724 * Insertion sort, ensuring the first entry in the list is always
2725 * the one we need first.
2727 spin_lock_irq(&ctx
->completion_lock
);
2728 list_for_each_prev(entry
, &ctx
->timeout_list
) {
2729 struct io_kiocb
*nxt
= list_entry(entry
, struct io_kiocb
, list
);
2730 unsigned nxt_sq_head
;
2731 long long tmp
, tmp_nxt
;
2732 u32 nxt_offset
= nxt
->io
->timeout
.seq_offset
;
2734 if (nxt
->flags
& REQ_F_TIMEOUT_NOSEQ
)
2738 * Since cached_sq_head + count - 1 can overflow, use type long
2741 tmp
= (long long)ctx
->cached_sq_head
+ count
- 1;
2742 nxt_sq_head
= nxt
->sequence
- nxt_offset
+ 1;
2743 tmp_nxt
= (long long)nxt_sq_head
+ nxt_offset
- 1;
2746 * cached_sq_head may overflow, and it will never overflow twice
2747 * once there is some timeout req still be valid.
2749 if (ctx
->cached_sq_head
< nxt_sq_head
)
2756 * Sequence of reqs after the insert one and itself should
2757 * be adjusted because each timeout req consumes a slot.
2762 req
->sequence
-= span
;
2764 list_add(&req
->list
, entry
);
2765 data
->timer
.function
= io_timeout_fn
;
2766 hrtimer_start(&data
->timer
, timespec64_to_ktime(data
->ts
), data
->mode
);
2767 spin_unlock_irq(&ctx
->completion_lock
);
2771 static bool io_cancel_cb(struct io_wq_work
*work
, void *data
)
2773 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
2775 return req
->user_data
== (unsigned long) data
;
2778 static int io_async_cancel_one(struct io_ring_ctx
*ctx
, void *sqe_addr
)
2780 enum io_wq_cancel cancel_ret
;
2783 cancel_ret
= io_wq_cancel_cb(ctx
->io_wq
, io_cancel_cb
, sqe_addr
);
2784 switch (cancel_ret
) {
2785 case IO_WQ_CANCEL_OK
:
2788 case IO_WQ_CANCEL_RUNNING
:
2791 case IO_WQ_CANCEL_NOTFOUND
:
2799 static void io_async_find_and_cancel(struct io_ring_ctx
*ctx
,
2800 struct io_kiocb
*req
, __u64 sqe_addr
,
2801 struct io_kiocb
**nxt
, int success_ret
)
2803 unsigned long flags
;
2806 ret
= io_async_cancel_one(ctx
, (void *) (unsigned long) sqe_addr
);
2807 if (ret
!= -ENOENT
) {
2808 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
2812 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
2813 ret
= io_timeout_cancel(ctx
, sqe_addr
);
2816 ret
= io_poll_cancel(ctx
, sqe_addr
);
2820 io_cqring_fill_event(req
, ret
);
2821 io_commit_cqring(ctx
);
2822 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
2823 io_cqring_ev_posted(ctx
);
2825 if (ret
< 0 && (req
->flags
& REQ_F_LINK
))
2826 req
->flags
|= REQ_F_FAIL_LINK
;
2827 io_put_req_find_next(req
, nxt
);
2830 static int io_async_cancel(struct io_kiocb
*req
, const struct io_uring_sqe
*sqe
,
2831 struct io_kiocb
**nxt
)
2833 struct io_ring_ctx
*ctx
= req
->ctx
;
2835 if (unlikely(ctx
->flags
& IORING_SETUP_IOPOLL
))
2837 if (sqe
->flags
|| sqe
->ioprio
|| sqe
->off
|| sqe
->len
||
2841 io_async_find_and_cancel(ctx
, req
, READ_ONCE(sqe
->addr
), nxt
, 0);
2845 static int io_req_defer_prep(struct io_kiocb
*req
, struct io_async_ctx
*io
)
2847 struct iovec inline_vecs
[UIO_FASTIOV
], *iovec
= inline_vecs
;
2848 struct iov_iter iter
;
2851 memcpy(&io
->sqe
, req
->sqe
, sizeof(io
->sqe
));
2852 req
->sqe
= &io
->sqe
;
2854 switch (io
->sqe
.opcode
) {
2855 case IORING_OP_READV
:
2856 case IORING_OP_READ_FIXED
:
2857 ret
= io_read_prep(req
, &iovec
, &iter
, true);
2859 case IORING_OP_WRITEV
:
2860 case IORING_OP_WRITE_FIXED
:
2861 ret
= io_write_prep(req
, &iovec
, &iter
, true);
2863 case IORING_OP_SENDMSG
:
2864 ret
= io_sendmsg_prep(req
, io
);
2866 case IORING_OP_RECVMSG
:
2867 ret
= io_recvmsg_prep(req
, io
);
2869 case IORING_OP_CONNECT
:
2870 ret
= io_connect_prep(req
, io
);
2872 case IORING_OP_TIMEOUT
:
2873 return io_timeout_prep(req
, io
, false);
2874 case IORING_OP_LINK_TIMEOUT
:
2875 return io_timeout_prep(req
, io
, true);
2885 io_req_map_io(req
, ret
, iovec
, inline_vecs
, &iter
);
2889 static int io_req_defer(struct io_kiocb
*req
)
2891 struct io_ring_ctx
*ctx
= req
->ctx
;
2892 struct io_async_ctx
*io
;
2895 /* Still need defer if there is pending req in defer list. */
2896 if (!req_need_defer(req
) && list_empty(&ctx
->defer_list
))
2899 io
= kmalloc(sizeof(*io
), GFP_KERNEL
);
2903 ret
= io_req_defer_prep(req
, io
);
2909 spin_lock_irq(&ctx
->completion_lock
);
2910 if (!req_need_defer(req
) && list_empty(&ctx
->defer_list
)) {
2911 spin_unlock_irq(&ctx
->completion_lock
);
2915 trace_io_uring_defer(ctx
, req
, req
->user_data
);
2916 list_add_tail(&req
->list
, &ctx
->defer_list
);
2917 spin_unlock_irq(&ctx
->completion_lock
);
2918 return -EIOCBQUEUED
;
2921 __attribute__((nonnull
))
2922 static int io_issue_sqe(struct io_kiocb
*req
, struct io_kiocb
**nxt
,
2923 bool force_nonblock
)
2926 struct io_ring_ctx
*ctx
= req
->ctx
;
2928 opcode
= READ_ONCE(req
->sqe
->opcode
);
2933 case IORING_OP_READV
:
2934 if (unlikely(req
->sqe
->buf_index
))
2936 ret
= io_read(req
, nxt
, force_nonblock
);
2938 case IORING_OP_WRITEV
:
2939 if (unlikely(req
->sqe
->buf_index
))
2941 ret
= io_write(req
, nxt
, force_nonblock
);
2943 case IORING_OP_READ_FIXED
:
2944 ret
= io_read(req
, nxt
, force_nonblock
);
2946 case IORING_OP_WRITE_FIXED
:
2947 ret
= io_write(req
, nxt
, force_nonblock
);
2949 case IORING_OP_FSYNC
:
2950 ret
= io_fsync(req
, req
->sqe
, nxt
, force_nonblock
);
2952 case IORING_OP_POLL_ADD
:
2953 ret
= io_poll_add(req
, req
->sqe
, nxt
);
2955 case IORING_OP_POLL_REMOVE
:
2956 ret
= io_poll_remove(req
, req
->sqe
);
2958 case IORING_OP_SYNC_FILE_RANGE
:
2959 ret
= io_sync_file_range(req
, req
->sqe
, nxt
, force_nonblock
);
2961 case IORING_OP_SENDMSG
:
2962 ret
= io_sendmsg(req
, req
->sqe
, nxt
, force_nonblock
);
2964 case IORING_OP_RECVMSG
:
2965 ret
= io_recvmsg(req
, req
->sqe
, nxt
, force_nonblock
);
2967 case IORING_OP_TIMEOUT
:
2968 ret
= io_timeout(req
, req
->sqe
);
2970 case IORING_OP_TIMEOUT_REMOVE
:
2971 ret
= io_timeout_remove(req
, req
->sqe
);
2973 case IORING_OP_ACCEPT
:
2974 ret
= io_accept(req
, req
->sqe
, nxt
, force_nonblock
);
2976 case IORING_OP_CONNECT
:
2977 ret
= io_connect(req
, req
->sqe
, nxt
, force_nonblock
);
2979 case IORING_OP_ASYNC_CANCEL
:
2980 ret
= io_async_cancel(req
, req
->sqe
, nxt
);
2990 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
2991 if (req
->result
== -EAGAIN
)
2994 /* workqueue context doesn't hold uring_lock, grab it now */
2996 mutex_lock(&ctx
->uring_lock
);
2997 io_iopoll_req_issued(req
);
2999 mutex_unlock(&ctx
->uring_lock
);
3005 static void io_link_work_cb(struct io_wq_work
**workptr
)
3007 struct io_wq_work
*work
= *workptr
;
3008 struct io_kiocb
*link
= work
->data
;
3010 io_queue_linked_timeout(link
);
3011 work
->func
= io_wq_submit_work
;
3014 static void io_wq_submit_work(struct io_wq_work
**workptr
)
3016 struct io_wq_work
*work
= *workptr
;
3017 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
3018 struct io_kiocb
*nxt
= NULL
;
3021 /* Ensure we clear previously set non-block flag */
3022 req
->rw
.ki_flags
&= ~IOCB_NOWAIT
;
3024 if (work
->flags
& IO_WQ_WORK_CANCEL
)
3028 req
->has_user
= (work
->flags
& IO_WQ_WORK_HAS_MM
) != 0;
3029 req
->in_async
= true;
3031 ret
= io_issue_sqe(req
, &nxt
, false);
3033 * We can get EAGAIN for polled IO even though we're
3034 * forcing a sync submission from here, since we can't
3035 * wait for request slots on the block side.
3043 /* drop submission reference */
3047 if (req
->flags
& REQ_F_LINK
)
3048 req
->flags
|= REQ_F_FAIL_LINK
;
3049 io_cqring_add_event(req
, ret
);
3053 /* if a dependent link is ready, pass it back */
3055 struct io_kiocb
*link
;
3057 io_prep_async_work(nxt
, &link
);
3058 *workptr
= &nxt
->work
;
3060 nxt
->work
.flags
|= IO_WQ_WORK_CB
;
3061 nxt
->work
.func
= io_link_work_cb
;
3062 nxt
->work
.data
= link
;
3067 static bool io_op_needs_file(const struct io_uring_sqe
*sqe
)
3069 int op
= READ_ONCE(sqe
->opcode
);
3073 case IORING_OP_POLL_REMOVE
:
3074 case IORING_OP_TIMEOUT
:
3075 case IORING_OP_TIMEOUT_REMOVE
:
3076 case IORING_OP_ASYNC_CANCEL
:
3077 case IORING_OP_LINK_TIMEOUT
:
3084 static inline struct file
*io_file_from_index(struct io_ring_ctx
*ctx
,
3087 struct fixed_file_table
*table
;
3089 table
= &ctx
->file_table
[index
>> IORING_FILE_TABLE_SHIFT
];
3090 return table
->files
[index
& IORING_FILE_TABLE_MASK
];
3093 static int io_req_set_file(struct io_submit_state
*state
, struct io_kiocb
*req
)
3095 struct io_ring_ctx
*ctx
= req
->ctx
;
3099 flags
= READ_ONCE(req
->sqe
->flags
);
3100 fd
= READ_ONCE(req
->sqe
->fd
);
3102 if (flags
& IOSQE_IO_DRAIN
)
3103 req
->flags
|= REQ_F_IO_DRAIN
;
3105 if (!io_op_needs_file(req
->sqe
))
3108 if (flags
& IOSQE_FIXED_FILE
) {
3109 if (unlikely(!ctx
->file_table
||
3110 (unsigned) fd
>= ctx
->nr_user_files
))
3112 fd
= array_index_nospec(fd
, ctx
->nr_user_files
);
3113 req
->file
= io_file_from_index(ctx
, fd
);
3116 req
->flags
|= REQ_F_FIXED_FILE
;
3118 if (req
->needs_fixed_file
)
3120 trace_io_uring_file_get(ctx
, fd
);
3121 req
->file
= io_file_get(state
, fd
);
3122 if (unlikely(!req
->file
))
3129 static int io_grab_files(struct io_kiocb
*req
)
3132 struct io_ring_ctx
*ctx
= req
->ctx
;
3135 spin_lock_irq(&ctx
->inflight_lock
);
3137 * We use the f_ops->flush() handler to ensure that we can flush
3138 * out work accessing these files if the fd is closed. Check if
3139 * the fd has changed since we started down this path, and disallow
3140 * this operation if it has.
3142 if (fcheck(req
->ring_fd
) == req
->ring_file
) {
3143 list_add(&req
->inflight_entry
, &ctx
->inflight_list
);
3144 req
->flags
|= REQ_F_INFLIGHT
;
3145 req
->work
.files
= current
->files
;
3148 spin_unlock_irq(&ctx
->inflight_lock
);
3154 static enum hrtimer_restart
io_link_timeout_fn(struct hrtimer
*timer
)
3156 struct io_timeout_data
*data
= container_of(timer
,
3157 struct io_timeout_data
, timer
);
3158 struct io_kiocb
*req
= data
->req
;
3159 struct io_ring_ctx
*ctx
= req
->ctx
;
3160 struct io_kiocb
*prev
= NULL
;
3161 unsigned long flags
;
3163 spin_lock_irqsave(&ctx
->completion_lock
, flags
);
3166 * We don't expect the list to be empty, that will only happen if we
3167 * race with the completion of the linked work.
3169 if (!list_empty(&req
->link_list
)) {
3170 prev
= list_entry(req
->link_list
.prev
, struct io_kiocb
,
3172 if (refcount_inc_not_zero(&prev
->refs
)) {
3173 list_del_init(&req
->link_list
);
3174 prev
->flags
&= ~REQ_F_LINK_TIMEOUT
;
3179 spin_unlock_irqrestore(&ctx
->completion_lock
, flags
);
3182 if (prev
->flags
& REQ_F_LINK
)
3183 prev
->flags
|= REQ_F_FAIL_LINK
;
3184 io_async_find_and_cancel(ctx
, req
, prev
->user_data
, NULL
,
3188 io_cqring_add_event(req
, -ETIME
);
3191 return HRTIMER_NORESTART
;
3194 static void io_queue_linked_timeout(struct io_kiocb
*req
)
3196 struct io_ring_ctx
*ctx
= req
->ctx
;
3199 * If the list is now empty, then our linked request finished before
3200 * we got a chance to setup the timer
3202 spin_lock_irq(&ctx
->completion_lock
);
3203 if (!list_empty(&req
->link_list
)) {
3204 struct io_timeout_data
*data
= &req
->io
->timeout
;
3206 data
->timer
.function
= io_link_timeout_fn
;
3207 hrtimer_start(&data
->timer
, timespec64_to_ktime(data
->ts
),
3210 spin_unlock_irq(&ctx
->completion_lock
);
3212 /* drop submission reference */
3216 static struct io_kiocb
*io_prep_linked_timeout(struct io_kiocb
*req
)
3218 struct io_kiocb
*nxt
;
3220 if (!(req
->flags
& REQ_F_LINK
))
3223 nxt
= list_first_entry_or_null(&req
->link_list
, struct io_kiocb
,
3225 if (!nxt
|| nxt
->sqe
->opcode
!= IORING_OP_LINK_TIMEOUT
)
3228 req
->flags
|= REQ_F_LINK_TIMEOUT
;
3232 static void __io_queue_sqe(struct io_kiocb
*req
)
3234 struct io_kiocb
*linked_timeout
= io_prep_linked_timeout(req
);
3235 struct io_kiocb
*nxt
= NULL
;
3238 ret
= io_issue_sqe(req
, &nxt
, true);
3240 io_queue_async_work(nxt
);
3243 * We async punt it if the file wasn't marked NOWAIT, or if the file
3244 * doesn't support non-blocking read/write attempts
3246 if (ret
== -EAGAIN
&& (!(req
->flags
& REQ_F_NOWAIT
) ||
3247 (req
->flags
& REQ_F_MUST_PUNT
))) {
3248 if (req
->work
.flags
& IO_WQ_WORK_NEEDS_FILES
) {
3249 ret
= io_grab_files(req
);
3255 * Queued up for async execution, worker will release
3256 * submit reference when the iocb is actually submitted.
3258 io_queue_async_work(req
);
3263 /* drop submission reference */
3266 if (linked_timeout
) {
3268 io_queue_linked_timeout(linked_timeout
);
3270 io_put_req(linked_timeout
);
3273 /* and drop final reference, if we failed */
3275 io_cqring_add_event(req
, ret
);
3276 if (req
->flags
& REQ_F_LINK
)
3277 req
->flags
|= REQ_F_FAIL_LINK
;
3282 static void io_queue_sqe(struct io_kiocb
*req
)
3286 if (unlikely(req
->ctx
->drain_next
)) {
3287 req
->flags
|= REQ_F_IO_DRAIN
;
3288 req
->ctx
->drain_next
= false;
3290 req
->ctx
->drain_next
= (req
->flags
& REQ_F_DRAIN_LINK
);
3292 ret
= io_req_defer(req
);
3294 if (ret
!= -EIOCBQUEUED
) {
3295 io_cqring_add_event(req
, ret
);
3296 if (req
->flags
& REQ_F_LINK
)
3297 req
->flags
|= REQ_F_FAIL_LINK
;
3298 io_double_put_req(req
);
3301 __io_queue_sqe(req
);
3304 static inline void io_queue_link_head(struct io_kiocb
*req
)
3306 if (unlikely(req
->flags
& REQ_F_FAIL_LINK
)) {
3307 io_cqring_add_event(req
, -ECANCELED
);
3308 io_double_put_req(req
);
3314 #define SQE_VALID_FLAGS (IOSQE_FIXED_FILE|IOSQE_IO_DRAIN|IOSQE_IO_LINK)
3316 static bool io_submit_sqe(struct io_kiocb
*req
, struct io_submit_state
*state
,
3317 struct io_kiocb
**link
)
3319 struct io_ring_ctx
*ctx
= req
->ctx
;
3322 req
->user_data
= req
->sqe
->user_data
;
3324 /* enforce forwards compatibility on users */
3325 if (unlikely(req
->sqe
->flags
& ~SQE_VALID_FLAGS
)) {
3330 ret
= io_req_set_file(state
, req
);
3331 if (unlikely(ret
)) {
3333 io_cqring_add_event(req
, ret
);
3334 io_double_put_req(req
);
3339 * If we already have a head request, queue this one for async
3340 * submittal once the head completes. If we don't have a head but
3341 * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be
3342 * submitted sync once the chain is complete. If none of those
3343 * conditions are true (normal request), then just queue it.
3346 struct io_kiocb
*prev
= *link
;
3347 struct io_async_ctx
*io
;
3349 if (req
->sqe
->flags
& IOSQE_IO_DRAIN
)
3350 (*link
)->flags
|= REQ_F_DRAIN_LINK
| REQ_F_IO_DRAIN
;
3352 io
= kmalloc(sizeof(*io
), GFP_KERNEL
);
3358 ret
= io_req_defer_prep(req
, io
);
3361 prev
->flags
|= REQ_F_FAIL_LINK
;
3364 trace_io_uring_link(ctx
, req
, prev
);
3365 list_add_tail(&req
->link_list
, &prev
->link_list
);
3366 } else if (req
->sqe
->flags
& IOSQE_IO_LINK
) {
3367 req
->flags
|= REQ_F_LINK
;
3369 INIT_LIST_HEAD(&req
->link_list
);
3379 * Batched submission is done, ensure local IO is flushed out.
3381 static void io_submit_state_end(struct io_submit_state
*state
)
3383 blk_finish_plug(&state
->plug
);
3385 if (state
->free_reqs
)
3386 kmem_cache_free_bulk(req_cachep
, state
->free_reqs
,
3387 &state
->reqs
[state
->cur_req
]);
3391 * Start submission side cache.
3393 static void io_submit_state_start(struct io_submit_state
*state
,
3394 unsigned int max_ios
)
3396 blk_start_plug(&state
->plug
);
3397 state
->free_reqs
= 0;
3399 state
->ios_left
= max_ios
;
3402 static void io_commit_sqring(struct io_ring_ctx
*ctx
)
3404 struct io_rings
*rings
= ctx
->rings
;
3406 if (ctx
->cached_sq_head
!= READ_ONCE(rings
->sq
.head
)) {
3408 * Ensure any loads from the SQEs are done at this point,
3409 * since once we write the new head, the application could
3410 * write new data to them.
3412 smp_store_release(&rings
->sq
.head
, ctx
->cached_sq_head
);
3417 * Fetch an sqe, if one is available. Note that s->sqe will point to memory
3418 * that is mapped by userspace. This means that care needs to be taken to
3419 * ensure that reads are stable, as we cannot rely on userspace always
3420 * being a good citizen. If members of the sqe are validated and then later
3421 * used, it's important that those reads are done through READ_ONCE() to
3422 * prevent a re-load down the line.
3424 static bool io_get_sqring(struct io_ring_ctx
*ctx
, struct io_kiocb
*req
)
3426 struct io_rings
*rings
= ctx
->rings
;
3427 u32
*sq_array
= ctx
->sq_array
;
3431 * The cached sq head (or cq tail) serves two purposes:
3433 * 1) allows us to batch the cost of updating the user visible
3435 * 2) allows the kernel side to track the head on its own, even
3436 * though the application is the one updating it.
3438 head
= ctx
->cached_sq_head
;
3439 /* make sure SQ entry isn't read before tail */
3440 if (unlikely(head
== smp_load_acquire(&rings
->sq
.tail
)))
3443 head
= READ_ONCE(sq_array
[head
& ctx
->sq_mask
]);
3444 if (likely(head
< ctx
->sq_entries
)) {
3446 * All io need record the previous position, if LINK vs DARIN,
3447 * it can be used to mark the position of the first IO in the
3450 req
->sequence
= ctx
->cached_sq_head
;
3451 req
->sqe
= &ctx
->sq_sqes
[head
];
3452 ctx
->cached_sq_head
++;
3456 /* drop invalid entries */
3457 ctx
->cached_sq_head
++;
3458 ctx
->cached_sq_dropped
++;
3459 WRITE_ONCE(rings
->sq_dropped
, ctx
->cached_sq_dropped
);
3463 static int io_submit_sqes(struct io_ring_ctx
*ctx
, unsigned int nr
,
3464 struct file
*ring_file
, int ring_fd
,
3465 struct mm_struct
**mm
, bool async
)
3467 struct io_submit_state state
, *statep
= NULL
;
3468 struct io_kiocb
*link
= NULL
;
3469 int i
, submitted
= 0;
3470 bool mm_fault
= false;
3472 /* if we have a backlog and couldn't flush it all, return BUSY */
3473 if (!list_empty(&ctx
->cq_overflow_list
) &&
3474 !io_cqring_overflow_flush(ctx
, false))
3477 if (nr
> IO_PLUG_THRESHOLD
) {
3478 io_submit_state_start(&state
, nr
);
3482 for (i
= 0; i
< nr
; i
++) {
3483 struct io_kiocb
*req
;
3484 unsigned int sqe_flags
;
3486 req
= io_get_req(ctx
, statep
);
3487 if (unlikely(!req
)) {
3489 submitted
= -EAGAIN
;
3492 if (!io_get_sqring(ctx
, req
)) {
3497 if (io_sqe_needs_user(req
->sqe
) && !*mm
) {
3498 mm_fault
= mm_fault
|| !mmget_not_zero(ctx
->sqo_mm
);
3500 use_mm(ctx
->sqo_mm
);
3506 sqe_flags
= req
->sqe
->flags
;
3508 req
->ring_file
= ring_file
;
3509 req
->ring_fd
= ring_fd
;
3510 req
->has_user
= *mm
!= NULL
;
3511 req
->in_async
= async
;
3512 req
->needs_fixed_file
= async
;
3513 trace_io_uring_submit_sqe(ctx
, req
->sqe
->user_data
,
3515 if (!io_submit_sqe(req
, statep
, &link
))
3518 * If previous wasn't linked and we have a linked command,
3519 * that's the end of the chain. Submit the previous link.
3521 if (!(sqe_flags
& IOSQE_IO_LINK
) && link
) {
3522 io_queue_link_head(link
);
3528 io_queue_link_head(link
);
3530 io_submit_state_end(&state
);
3532 /* Commit SQ ring head once we've consumed and submitted all SQEs */
3533 io_commit_sqring(ctx
);
3538 static int io_sq_thread(void *data
)
3540 struct io_ring_ctx
*ctx
= data
;
3541 struct mm_struct
*cur_mm
= NULL
;
3542 const struct cred
*old_cred
;
3543 mm_segment_t old_fs
;
3546 unsigned long timeout
;
3549 complete(&ctx
->completions
[1]);
3553 old_cred
= override_creds(ctx
->creds
);
3555 ret
= timeout
= inflight
= 0;
3556 while (!kthread_should_park()) {
3557 unsigned int to_submit
;
3560 unsigned nr_events
= 0;
3562 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
3564 * inflight is the count of the maximum possible
3565 * entries we submitted, but it can be smaller
3566 * if we dropped some of them. If we don't have
3567 * poll entries available, then we know that we
3568 * have nothing left to poll for. Reset the
3569 * inflight count to zero in that case.
3571 mutex_lock(&ctx
->uring_lock
);
3572 if (!list_empty(&ctx
->poll_list
))
3573 __io_iopoll_check(ctx
, &nr_events
, 0);
3576 mutex_unlock(&ctx
->uring_lock
);
3579 * Normal IO, just pretend everything completed.
3580 * We don't have to poll completions for that.
3582 nr_events
= inflight
;
3585 inflight
-= nr_events
;
3587 timeout
= jiffies
+ ctx
->sq_thread_idle
;
3590 to_submit
= io_sqring_entries(ctx
);
3593 * If submit got -EBUSY, flag us as needing the application
3594 * to enter the kernel to reap and flush events.
3596 if (!to_submit
|| ret
== -EBUSY
) {
3598 * We're polling. If we're within the defined idle
3599 * period, then let us spin without work before going
3600 * to sleep. The exception is if we got EBUSY doing
3601 * more IO, we should wait for the application to
3602 * reap events and wake us up.
3605 (!time_after(jiffies
, timeout
) && ret
!= -EBUSY
)) {
3611 * Drop cur_mm before scheduling, we can't hold it for
3612 * long periods (or over schedule()). Do this before
3613 * adding ourselves to the waitqueue, as the unuse/drop
3622 prepare_to_wait(&ctx
->sqo_wait
, &wait
,
3623 TASK_INTERRUPTIBLE
);
3625 /* Tell userspace we may need a wakeup call */
3626 ctx
->rings
->sq_flags
|= IORING_SQ_NEED_WAKEUP
;
3627 /* make sure to read SQ tail after writing flags */
3630 to_submit
= io_sqring_entries(ctx
);
3631 if (!to_submit
|| ret
== -EBUSY
) {
3632 if (kthread_should_park()) {
3633 finish_wait(&ctx
->sqo_wait
, &wait
);
3636 if (signal_pending(current
))
3637 flush_signals(current
);
3639 finish_wait(&ctx
->sqo_wait
, &wait
);
3641 ctx
->rings
->sq_flags
&= ~IORING_SQ_NEED_WAKEUP
;
3644 finish_wait(&ctx
->sqo_wait
, &wait
);
3646 ctx
->rings
->sq_flags
&= ~IORING_SQ_NEED_WAKEUP
;
3649 to_submit
= min(to_submit
, ctx
->sq_entries
);
3650 ret
= io_submit_sqes(ctx
, to_submit
, NULL
, -1, &cur_mm
, true);
3660 revert_creds(old_cred
);
3667 struct io_wait_queue
{
3668 struct wait_queue_entry wq
;
3669 struct io_ring_ctx
*ctx
;
3671 unsigned nr_timeouts
;
3674 static inline bool io_should_wake(struct io_wait_queue
*iowq
, bool noflush
)
3676 struct io_ring_ctx
*ctx
= iowq
->ctx
;
3679 * Wake up if we have enough events, or if a timeout occured since we
3680 * started waiting. For timeouts, we always want to return to userspace,
3681 * regardless of event count.
3683 return io_cqring_events(ctx
, noflush
) >= iowq
->to_wait
||
3684 atomic_read(&ctx
->cq_timeouts
) != iowq
->nr_timeouts
;
3687 static int io_wake_function(struct wait_queue_entry
*curr
, unsigned int mode
,
3688 int wake_flags
, void *key
)
3690 struct io_wait_queue
*iowq
= container_of(curr
, struct io_wait_queue
,
3693 /* use noflush == true, as we can't safely rely on locking context */
3694 if (!io_should_wake(iowq
, true))
3697 return autoremove_wake_function(curr
, mode
, wake_flags
, key
);
3701 * Wait until events become available, if we don't already have some. The
3702 * application must reap them itself, as they reside on the shared cq ring.
3704 static int io_cqring_wait(struct io_ring_ctx
*ctx
, int min_events
,
3705 const sigset_t __user
*sig
, size_t sigsz
)
3707 struct io_wait_queue iowq
= {
3710 .func
= io_wake_function
,
3711 .entry
= LIST_HEAD_INIT(iowq
.wq
.entry
),
3714 .to_wait
= min_events
,
3716 struct io_rings
*rings
= ctx
->rings
;
3719 if (io_cqring_events(ctx
, false) >= min_events
)
3723 #ifdef CONFIG_COMPAT
3724 if (in_compat_syscall())
3725 ret
= set_compat_user_sigmask((const compat_sigset_t __user
*)sig
,
3729 ret
= set_user_sigmask(sig
, sigsz
);
3735 iowq
.nr_timeouts
= atomic_read(&ctx
->cq_timeouts
);
3736 trace_io_uring_cqring_wait(ctx
, min_events
);
3738 prepare_to_wait_exclusive(&ctx
->wait
, &iowq
.wq
,
3739 TASK_INTERRUPTIBLE
);
3740 if (io_should_wake(&iowq
, false))
3743 if (signal_pending(current
)) {
3748 finish_wait(&ctx
->wait
, &iowq
.wq
);
3750 restore_saved_sigmask_unless(ret
== -EINTR
);
3752 return READ_ONCE(rings
->cq
.head
) == READ_ONCE(rings
->cq
.tail
) ? ret
: 0;
3755 static void __io_sqe_files_unregister(struct io_ring_ctx
*ctx
)
3757 #if defined(CONFIG_UNIX)
3758 if (ctx
->ring_sock
) {
3759 struct sock
*sock
= ctx
->ring_sock
->sk
;
3760 struct sk_buff
*skb
;
3762 while ((skb
= skb_dequeue(&sock
->sk_receive_queue
)) != NULL
)
3768 for (i
= 0; i
< ctx
->nr_user_files
; i
++) {
3771 file
= io_file_from_index(ctx
, i
);
3778 static int io_sqe_files_unregister(struct io_ring_ctx
*ctx
)
3780 unsigned nr_tables
, i
;
3782 if (!ctx
->file_table
)
3785 __io_sqe_files_unregister(ctx
);
3786 nr_tables
= DIV_ROUND_UP(ctx
->nr_user_files
, IORING_MAX_FILES_TABLE
);
3787 for (i
= 0; i
< nr_tables
; i
++)
3788 kfree(ctx
->file_table
[i
].files
);
3789 kfree(ctx
->file_table
);
3790 ctx
->file_table
= NULL
;
3791 ctx
->nr_user_files
= 0;
3795 static void io_sq_thread_stop(struct io_ring_ctx
*ctx
)
3797 if (ctx
->sqo_thread
) {
3798 wait_for_completion(&ctx
->completions
[1]);
3800 * The park is a bit of a work-around, without it we get
3801 * warning spews on shutdown with SQPOLL set and affinity
3802 * set to a single CPU.
3804 kthread_park(ctx
->sqo_thread
);
3805 kthread_stop(ctx
->sqo_thread
);
3806 ctx
->sqo_thread
= NULL
;
3810 static void io_finish_async(struct io_ring_ctx
*ctx
)
3812 io_sq_thread_stop(ctx
);
3815 io_wq_destroy(ctx
->io_wq
);
3820 #if defined(CONFIG_UNIX)
3821 static void io_destruct_skb(struct sk_buff
*skb
)
3823 struct io_ring_ctx
*ctx
= skb
->sk
->sk_user_data
;
3826 io_wq_flush(ctx
->io_wq
);
3828 unix_destruct_scm(skb
);
3832 * Ensure the UNIX gc is aware of our file set, so we are certain that
3833 * the io_uring can be safely unregistered on process exit, even if we have
3834 * loops in the file referencing.
3836 static int __io_sqe_files_scm(struct io_ring_ctx
*ctx
, int nr
, int offset
)
3838 struct sock
*sk
= ctx
->ring_sock
->sk
;
3839 struct scm_fp_list
*fpl
;
3840 struct sk_buff
*skb
;
3843 if (!capable(CAP_SYS_RESOURCE
) && !capable(CAP_SYS_ADMIN
)) {
3844 unsigned long inflight
= ctx
->user
->unix_inflight
+ nr
;
3846 if (inflight
> task_rlimit(current
, RLIMIT_NOFILE
))
3850 fpl
= kzalloc(sizeof(*fpl
), GFP_KERNEL
);
3854 skb
= alloc_skb(0, GFP_KERNEL
);
3863 fpl
->user
= get_uid(ctx
->user
);
3864 for (i
= 0; i
< nr
; i
++) {
3865 struct file
*file
= io_file_from_index(ctx
, i
+ offset
);
3869 fpl
->fp
[nr_files
] = get_file(file
);
3870 unix_inflight(fpl
->user
, fpl
->fp
[nr_files
]);
3875 fpl
->max
= SCM_MAX_FD
;
3876 fpl
->count
= nr_files
;
3877 UNIXCB(skb
).fp
= fpl
;
3878 skb
->destructor
= io_destruct_skb
;
3879 refcount_add(skb
->truesize
, &sk
->sk_wmem_alloc
);
3880 skb_queue_head(&sk
->sk_receive_queue
, skb
);
3882 for (i
= 0; i
< nr_files
; i
++)
3893 * If UNIX sockets are enabled, fd passing can cause a reference cycle which
3894 * causes regular reference counting to break down. We rely on the UNIX
3895 * garbage collection to take care of this problem for us.
3897 static int io_sqe_files_scm(struct io_ring_ctx
*ctx
)
3899 unsigned left
, total
;
3903 left
= ctx
->nr_user_files
;
3905 unsigned this_files
= min_t(unsigned, left
, SCM_MAX_FD
);
3907 ret
= __io_sqe_files_scm(ctx
, this_files
, total
);
3911 total
+= this_files
;
3917 while (total
< ctx
->nr_user_files
) {
3918 struct file
*file
= io_file_from_index(ctx
, total
);
3928 static int io_sqe_files_scm(struct io_ring_ctx
*ctx
)
3934 static int io_sqe_alloc_file_tables(struct io_ring_ctx
*ctx
, unsigned nr_tables
,
3939 for (i
= 0; i
< nr_tables
; i
++) {
3940 struct fixed_file_table
*table
= &ctx
->file_table
[i
];
3941 unsigned this_files
;
3943 this_files
= min(nr_files
, IORING_MAX_FILES_TABLE
);
3944 table
->files
= kcalloc(this_files
, sizeof(struct file
*),
3948 nr_files
-= this_files
;
3954 for (i
= 0; i
< nr_tables
; i
++) {
3955 struct fixed_file_table
*table
= &ctx
->file_table
[i
];
3956 kfree(table
->files
);
3961 static int io_sqe_files_register(struct io_ring_ctx
*ctx
, void __user
*arg
,
3964 __s32 __user
*fds
= (__s32 __user
*) arg
;
3969 if (ctx
->file_table
)
3973 if (nr_args
> IORING_MAX_FIXED_FILES
)
3976 nr_tables
= DIV_ROUND_UP(nr_args
, IORING_MAX_FILES_TABLE
);
3977 ctx
->file_table
= kcalloc(nr_tables
, sizeof(struct fixed_file_table
),
3979 if (!ctx
->file_table
)
3982 if (io_sqe_alloc_file_tables(ctx
, nr_tables
, nr_args
)) {
3983 kfree(ctx
->file_table
);
3984 ctx
->file_table
= NULL
;
3988 for (i
= 0; i
< nr_args
; i
++, ctx
->nr_user_files
++) {
3989 struct fixed_file_table
*table
;
3993 if (copy_from_user(&fd
, &fds
[i
], sizeof(fd
)))
3995 /* allow sparse sets */
4001 table
= &ctx
->file_table
[i
>> IORING_FILE_TABLE_SHIFT
];
4002 index
= i
& IORING_FILE_TABLE_MASK
;
4003 table
->files
[index
] = fget(fd
);
4006 if (!table
->files
[index
])
4009 * Don't allow io_uring instances to be registered. If UNIX
4010 * isn't enabled, then this causes a reference cycle and this
4011 * instance can never get freed. If UNIX is enabled we'll
4012 * handle it just fine, but there's still no point in allowing
4013 * a ring fd as it doesn't support regular read/write anyway.
4015 if (table
->files
[index
]->f_op
== &io_uring_fops
) {
4016 fput(table
->files
[index
]);
4023 for (i
= 0; i
< ctx
->nr_user_files
; i
++) {
4026 file
= io_file_from_index(ctx
, i
);
4030 for (i
= 0; i
< nr_tables
; i
++)
4031 kfree(ctx
->file_table
[i
].files
);
4033 kfree(ctx
->file_table
);
4034 ctx
->file_table
= NULL
;
4035 ctx
->nr_user_files
= 0;
4039 ret
= io_sqe_files_scm(ctx
);
4041 io_sqe_files_unregister(ctx
);
4046 static void io_sqe_file_unregister(struct io_ring_ctx
*ctx
, int index
)
4048 #if defined(CONFIG_UNIX)
4049 struct file
*file
= io_file_from_index(ctx
, index
);
4050 struct sock
*sock
= ctx
->ring_sock
->sk
;
4051 struct sk_buff_head list
, *head
= &sock
->sk_receive_queue
;
4052 struct sk_buff
*skb
;
4055 __skb_queue_head_init(&list
);
4058 * Find the skb that holds this file in its SCM_RIGHTS. When found,
4059 * remove this entry and rearrange the file array.
4061 skb
= skb_dequeue(head
);
4063 struct scm_fp_list
*fp
;
4065 fp
= UNIXCB(skb
).fp
;
4066 for (i
= 0; i
< fp
->count
; i
++) {
4069 if (fp
->fp
[i
] != file
)
4072 unix_notinflight(fp
->user
, fp
->fp
[i
]);
4073 left
= fp
->count
- 1 - i
;
4075 memmove(&fp
->fp
[i
], &fp
->fp
[i
+ 1],
4076 left
* sizeof(struct file
*));
4083 __skb_queue_tail(&list
, skb
);
4093 __skb_queue_tail(&list
, skb
);
4095 skb
= skb_dequeue(head
);
4098 if (skb_peek(&list
)) {
4099 spin_lock_irq(&head
->lock
);
4100 while ((skb
= __skb_dequeue(&list
)) != NULL
)
4101 __skb_queue_tail(head
, skb
);
4102 spin_unlock_irq(&head
->lock
);
4105 fput(io_file_from_index(ctx
, index
));
4109 static int io_sqe_file_register(struct io_ring_ctx
*ctx
, struct file
*file
,
4112 #if defined(CONFIG_UNIX)
4113 struct sock
*sock
= ctx
->ring_sock
->sk
;
4114 struct sk_buff_head
*head
= &sock
->sk_receive_queue
;
4115 struct sk_buff
*skb
;
4118 * See if we can merge this file into an existing skb SCM_RIGHTS
4119 * file set. If there's no room, fall back to allocating a new skb
4120 * and filling it in.
4122 spin_lock_irq(&head
->lock
);
4123 skb
= skb_peek(head
);
4125 struct scm_fp_list
*fpl
= UNIXCB(skb
).fp
;
4127 if (fpl
->count
< SCM_MAX_FD
) {
4128 __skb_unlink(skb
, head
);
4129 spin_unlock_irq(&head
->lock
);
4130 fpl
->fp
[fpl
->count
] = get_file(file
);
4131 unix_inflight(fpl
->user
, fpl
->fp
[fpl
->count
]);
4133 spin_lock_irq(&head
->lock
);
4134 __skb_queue_head(head
, skb
);
4139 spin_unlock_irq(&head
->lock
);
4146 return __io_sqe_files_scm(ctx
, 1, index
);
4152 static int io_sqe_files_update(struct io_ring_ctx
*ctx
, void __user
*arg
,
4155 struct io_uring_files_update up
;
4160 if (!ctx
->file_table
)
4164 if (copy_from_user(&up
, arg
, sizeof(up
)))
4166 if (check_add_overflow(up
.offset
, nr_args
, &done
))
4168 if (done
> ctx
->nr_user_files
)
4172 fds
= (__s32 __user
*) up
.fds
;
4174 struct fixed_file_table
*table
;
4178 if (copy_from_user(&fd
, &fds
[done
], sizeof(fd
))) {
4182 i
= array_index_nospec(up
.offset
, ctx
->nr_user_files
);
4183 table
= &ctx
->file_table
[i
>> IORING_FILE_TABLE_SHIFT
];
4184 index
= i
& IORING_FILE_TABLE_MASK
;
4185 if (table
->files
[index
]) {
4186 io_sqe_file_unregister(ctx
, i
);
4187 table
->files
[index
] = NULL
;
4198 * Don't allow io_uring instances to be registered. If
4199 * UNIX isn't enabled, then this causes a reference
4200 * cycle and this instance can never get freed. If UNIX
4201 * is enabled we'll handle it just fine, but there's
4202 * still no point in allowing a ring fd as it doesn't
4203 * support regular read/write anyway.
4205 if (file
->f_op
== &io_uring_fops
) {
4210 table
->files
[index
] = file
;
4211 err
= io_sqe_file_register(ctx
, file
, i
);
4220 return done
? done
: err
;
4223 static void io_put_work(struct io_wq_work
*work
)
4225 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
4230 static void io_get_work(struct io_wq_work
*work
)
4232 struct io_kiocb
*req
= container_of(work
, struct io_kiocb
, work
);
4234 refcount_inc(&req
->refs
);
4237 static int io_sq_offload_start(struct io_ring_ctx
*ctx
,
4238 struct io_uring_params
*p
)
4240 struct io_wq_data data
;
4241 unsigned concurrency
;
4244 init_waitqueue_head(&ctx
->sqo_wait
);
4245 mmgrab(current
->mm
);
4246 ctx
->sqo_mm
= current
->mm
;
4248 if (ctx
->flags
& IORING_SETUP_SQPOLL
) {
4250 if (!capable(CAP_SYS_ADMIN
))
4253 ctx
->sq_thread_idle
= msecs_to_jiffies(p
->sq_thread_idle
);
4254 if (!ctx
->sq_thread_idle
)
4255 ctx
->sq_thread_idle
= HZ
;
4257 if (p
->flags
& IORING_SETUP_SQ_AFF
) {
4258 int cpu
= p
->sq_thread_cpu
;
4261 if (cpu
>= nr_cpu_ids
)
4263 if (!cpu_online(cpu
))
4266 ctx
->sqo_thread
= kthread_create_on_cpu(io_sq_thread
,
4270 ctx
->sqo_thread
= kthread_create(io_sq_thread
, ctx
,
4273 if (IS_ERR(ctx
->sqo_thread
)) {
4274 ret
= PTR_ERR(ctx
->sqo_thread
);
4275 ctx
->sqo_thread
= NULL
;
4278 wake_up_process(ctx
->sqo_thread
);
4279 } else if (p
->flags
& IORING_SETUP_SQ_AFF
) {
4280 /* Can't have SQ_AFF without SQPOLL */
4285 data
.mm
= ctx
->sqo_mm
;
4286 data
.user
= ctx
->user
;
4287 data
.creds
= ctx
->creds
;
4288 data
.get_work
= io_get_work
;
4289 data
.put_work
= io_put_work
;
4291 /* Do QD, or 4 * CPUS, whatever is smallest */
4292 concurrency
= min(ctx
->sq_entries
, 4 * num_online_cpus());
4293 ctx
->io_wq
= io_wq_create(concurrency
, &data
);
4294 if (IS_ERR(ctx
->io_wq
)) {
4295 ret
= PTR_ERR(ctx
->io_wq
);
4302 io_finish_async(ctx
);
4303 mmdrop(ctx
->sqo_mm
);
4308 static void io_unaccount_mem(struct user_struct
*user
, unsigned long nr_pages
)
4310 atomic_long_sub(nr_pages
, &user
->locked_vm
);
4313 static int io_account_mem(struct user_struct
*user
, unsigned long nr_pages
)
4315 unsigned long page_limit
, cur_pages
, new_pages
;
4317 /* Don't allow more pages than we can safely lock */
4318 page_limit
= rlimit(RLIMIT_MEMLOCK
) >> PAGE_SHIFT
;
4321 cur_pages
= atomic_long_read(&user
->locked_vm
);
4322 new_pages
= cur_pages
+ nr_pages
;
4323 if (new_pages
> page_limit
)
4325 } while (atomic_long_cmpxchg(&user
->locked_vm
, cur_pages
,
4326 new_pages
) != cur_pages
);
4331 static void io_mem_free(void *ptr
)
4338 page
= virt_to_head_page(ptr
);
4339 if (put_page_testzero(page
))
4340 free_compound_page(page
);
4343 static void *io_mem_alloc(size_t size
)
4345 gfp_t gfp_flags
= GFP_KERNEL
| __GFP_ZERO
| __GFP_NOWARN
| __GFP_COMP
|
4348 return (void *) __get_free_pages(gfp_flags
, get_order(size
));
4351 static unsigned long rings_size(unsigned sq_entries
, unsigned cq_entries
,
4354 struct io_rings
*rings
;
4355 size_t off
, sq_array_size
;
4357 off
= struct_size(rings
, cqes
, cq_entries
);
4358 if (off
== SIZE_MAX
)
4362 off
= ALIGN(off
, SMP_CACHE_BYTES
);
4367 sq_array_size
= array_size(sizeof(u32
), sq_entries
);
4368 if (sq_array_size
== SIZE_MAX
)
4371 if (check_add_overflow(off
, sq_array_size
, &off
))
4380 static unsigned long ring_pages(unsigned sq_entries
, unsigned cq_entries
)
4384 pages
= (size_t)1 << get_order(
4385 rings_size(sq_entries
, cq_entries
, NULL
));
4386 pages
+= (size_t)1 << get_order(
4387 array_size(sizeof(struct io_uring_sqe
), sq_entries
));
4392 static int io_sqe_buffer_unregister(struct io_ring_ctx
*ctx
)
4396 if (!ctx
->user_bufs
)
4399 for (i
= 0; i
< ctx
->nr_user_bufs
; i
++) {
4400 struct io_mapped_ubuf
*imu
= &ctx
->user_bufs
[i
];
4402 for (j
= 0; j
< imu
->nr_bvecs
; j
++)
4403 put_user_page(imu
->bvec
[j
].bv_page
);
4405 if (ctx
->account_mem
)
4406 io_unaccount_mem(ctx
->user
, imu
->nr_bvecs
);
4411 kfree(ctx
->user_bufs
);
4412 ctx
->user_bufs
= NULL
;
4413 ctx
->nr_user_bufs
= 0;
4417 static int io_copy_iov(struct io_ring_ctx
*ctx
, struct iovec
*dst
,
4418 void __user
*arg
, unsigned index
)
4420 struct iovec __user
*src
;
4422 #ifdef CONFIG_COMPAT
4424 struct compat_iovec __user
*ciovs
;
4425 struct compat_iovec ciov
;
4427 ciovs
= (struct compat_iovec __user
*) arg
;
4428 if (copy_from_user(&ciov
, &ciovs
[index
], sizeof(ciov
)))
4431 dst
->iov_base
= (void __user
*) (unsigned long) ciov
.iov_base
;
4432 dst
->iov_len
= ciov
.iov_len
;
4436 src
= (struct iovec __user
*) arg
;
4437 if (copy_from_user(dst
, &src
[index
], sizeof(*dst
)))
4442 static int io_sqe_buffer_register(struct io_ring_ctx
*ctx
, void __user
*arg
,
4445 struct vm_area_struct
**vmas
= NULL
;
4446 struct page
**pages
= NULL
;
4447 int i
, j
, got_pages
= 0;
4452 if (!nr_args
|| nr_args
> UIO_MAXIOV
)
4455 ctx
->user_bufs
= kcalloc(nr_args
, sizeof(struct io_mapped_ubuf
),
4457 if (!ctx
->user_bufs
)
4460 for (i
= 0; i
< nr_args
; i
++) {
4461 struct io_mapped_ubuf
*imu
= &ctx
->user_bufs
[i
];
4462 unsigned long off
, start
, end
, ubuf
;
4467 ret
= io_copy_iov(ctx
, &iov
, arg
, i
);
4472 * Don't impose further limits on the size and buffer
4473 * constraints here, we'll -EINVAL later when IO is
4474 * submitted if they are wrong.
4477 if (!iov
.iov_base
|| !iov
.iov_len
)
4480 /* arbitrary limit, but we need something */
4481 if (iov
.iov_len
> SZ_1G
)
4484 ubuf
= (unsigned long) iov
.iov_base
;
4485 end
= (ubuf
+ iov
.iov_len
+ PAGE_SIZE
- 1) >> PAGE_SHIFT
;
4486 start
= ubuf
>> PAGE_SHIFT
;
4487 nr_pages
= end
- start
;
4489 if (ctx
->account_mem
) {
4490 ret
= io_account_mem(ctx
->user
, nr_pages
);
4496 if (!pages
|| nr_pages
> got_pages
) {
4499 pages
= kvmalloc_array(nr_pages
, sizeof(struct page
*),
4501 vmas
= kvmalloc_array(nr_pages
,
4502 sizeof(struct vm_area_struct
*),
4504 if (!pages
|| !vmas
) {
4506 if (ctx
->account_mem
)
4507 io_unaccount_mem(ctx
->user
, nr_pages
);
4510 got_pages
= nr_pages
;
4513 imu
->bvec
= kvmalloc_array(nr_pages
, sizeof(struct bio_vec
),
4517 if (ctx
->account_mem
)
4518 io_unaccount_mem(ctx
->user
, nr_pages
);
4523 down_read(¤t
->mm
->mmap_sem
);
4524 pret
= get_user_pages(ubuf
, nr_pages
,
4525 FOLL_WRITE
| FOLL_LONGTERM
,
4527 if (pret
== nr_pages
) {
4528 /* don't support file backed memory */
4529 for (j
= 0; j
< nr_pages
; j
++) {
4530 struct vm_area_struct
*vma
= vmas
[j
];
4533 !is_file_hugepages(vma
->vm_file
)) {
4539 ret
= pret
< 0 ? pret
: -EFAULT
;
4541 up_read(¤t
->mm
->mmap_sem
);
4544 * if we did partial map, or found file backed vmas,
4545 * release any pages we did get
4548 put_user_pages(pages
, pret
);
4549 if (ctx
->account_mem
)
4550 io_unaccount_mem(ctx
->user
, nr_pages
);
4555 off
= ubuf
& ~PAGE_MASK
;
4557 for (j
= 0; j
< nr_pages
; j
++) {
4560 vec_len
= min_t(size_t, size
, PAGE_SIZE
- off
);
4561 imu
->bvec
[j
].bv_page
= pages
[j
];
4562 imu
->bvec
[j
].bv_len
= vec_len
;
4563 imu
->bvec
[j
].bv_offset
= off
;
4567 /* store original address for later verification */
4569 imu
->len
= iov
.iov_len
;
4570 imu
->nr_bvecs
= nr_pages
;
4572 ctx
->nr_user_bufs
++;
4580 io_sqe_buffer_unregister(ctx
);
4584 static int io_eventfd_register(struct io_ring_ctx
*ctx
, void __user
*arg
)
4586 __s32 __user
*fds
= arg
;
4592 if (copy_from_user(&fd
, fds
, sizeof(*fds
)))
4595 ctx
->cq_ev_fd
= eventfd_ctx_fdget(fd
);
4596 if (IS_ERR(ctx
->cq_ev_fd
)) {
4597 int ret
= PTR_ERR(ctx
->cq_ev_fd
);
4598 ctx
->cq_ev_fd
= NULL
;
4605 static int io_eventfd_unregister(struct io_ring_ctx
*ctx
)
4607 if (ctx
->cq_ev_fd
) {
4608 eventfd_ctx_put(ctx
->cq_ev_fd
);
4609 ctx
->cq_ev_fd
= NULL
;
4616 static void io_ring_ctx_free(struct io_ring_ctx
*ctx
)
4618 io_finish_async(ctx
);
4620 mmdrop(ctx
->sqo_mm
);
4622 io_iopoll_reap_events(ctx
);
4623 io_sqe_buffer_unregister(ctx
);
4624 io_sqe_files_unregister(ctx
);
4625 io_eventfd_unregister(ctx
);
4627 #if defined(CONFIG_UNIX)
4628 if (ctx
->ring_sock
) {
4629 ctx
->ring_sock
->file
= NULL
; /* so that iput() is called */
4630 sock_release(ctx
->ring_sock
);
4634 io_mem_free(ctx
->rings
);
4635 io_mem_free(ctx
->sq_sqes
);
4637 percpu_ref_exit(&ctx
->refs
);
4638 if (ctx
->account_mem
)
4639 io_unaccount_mem(ctx
->user
,
4640 ring_pages(ctx
->sq_entries
, ctx
->cq_entries
));
4641 free_uid(ctx
->user
);
4642 put_cred(ctx
->creds
);
4643 kfree(ctx
->completions
);
4644 kfree(ctx
->cancel_hash
);
4645 kmem_cache_free(req_cachep
, ctx
->fallback_req
);
4649 static __poll_t
io_uring_poll(struct file
*file
, poll_table
*wait
)
4651 struct io_ring_ctx
*ctx
= file
->private_data
;
4654 poll_wait(file
, &ctx
->cq_wait
, wait
);
4656 * synchronizes with barrier from wq_has_sleeper call in
4660 if (READ_ONCE(ctx
->rings
->sq
.tail
) - ctx
->cached_sq_head
!=
4661 ctx
->rings
->sq_ring_entries
)
4662 mask
|= EPOLLOUT
| EPOLLWRNORM
;
4663 if (READ_ONCE(ctx
->rings
->cq
.head
) != ctx
->cached_cq_tail
)
4664 mask
|= EPOLLIN
| EPOLLRDNORM
;
4669 static int io_uring_fasync(int fd
, struct file
*file
, int on
)
4671 struct io_ring_ctx
*ctx
= file
->private_data
;
4673 return fasync_helper(fd
, file
, on
, &ctx
->cq_fasync
);
4676 static void io_ring_ctx_wait_and_kill(struct io_ring_ctx
*ctx
)
4678 mutex_lock(&ctx
->uring_lock
);
4679 percpu_ref_kill(&ctx
->refs
);
4680 mutex_unlock(&ctx
->uring_lock
);
4682 io_kill_timeouts(ctx
);
4683 io_poll_remove_all(ctx
);
4686 io_wq_cancel_all(ctx
->io_wq
);
4688 io_iopoll_reap_events(ctx
);
4689 /* if we failed setting up the ctx, we might not have any rings */
4691 io_cqring_overflow_flush(ctx
, true);
4692 wait_for_completion(&ctx
->completions
[0]);
4693 io_ring_ctx_free(ctx
);
4696 static int io_uring_release(struct inode
*inode
, struct file
*file
)
4698 struct io_ring_ctx
*ctx
= file
->private_data
;
4700 file
->private_data
= NULL
;
4701 io_ring_ctx_wait_and_kill(ctx
);
4705 static void io_uring_cancel_files(struct io_ring_ctx
*ctx
,
4706 struct files_struct
*files
)
4708 struct io_kiocb
*req
;
4711 while (!list_empty_careful(&ctx
->inflight_list
)) {
4712 struct io_kiocb
*cancel_req
= NULL
;
4714 spin_lock_irq(&ctx
->inflight_lock
);
4715 list_for_each_entry(req
, &ctx
->inflight_list
, inflight_entry
) {
4716 if (req
->work
.files
!= files
)
4718 /* req is being completed, ignore */
4719 if (!refcount_inc_not_zero(&req
->refs
))
4725 prepare_to_wait(&ctx
->inflight_wait
, &wait
,
4726 TASK_UNINTERRUPTIBLE
);
4727 spin_unlock_irq(&ctx
->inflight_lock
);
4729 /* We need to keep going until we don't find a matching req */
4733 io_wq_cancel_work(ctx
->io_wq
, &cancel_req
->work
);
4734 io_put_req(cancel_req
);
4737 finish_wait(&ctx
->inflight_wait
, &wait
);
4740 static int io_uring_flush(struct file
*file
, void *data
)
4742 struct io_ring_ctx
*ctx
= file
->private_data
;
4744 io_uring_cancel_files(ctx
, data
);
4745 if (fatal_signal_pending(current
) || (current
->flags
& PF_EXITING
)) {
4746 io_cqring_overflow_flush(ctx
, true);
4747 io_wq_cancel_all(ctx
->io_wq
);
4752 static void *io_uring_validate_mmap_request(struct file
*file
,
4753 loff_t pgoff
, size_t sz
)
4755 struct io_ring_ctx
*ctx
= file
->private_data
;
4756 loff_t offset
= pgoff
<< PAGE_SHIFT
;
4761 case IORING_OFF_SQ_RING
:
4762 case IORING_OFF_CQ_RING
:
4765 case IORING_OFF_SQES
:
4769 return ERR_PTR(-EINVAL
);
4772 page
= virt_to_head_page(ptr
);
4773 if (sz
> page_size(page
))
4774 return ERR_PTR(-EINVAL
);
4781 static int io_uring_mmap(struct file
*file
, struct vm_area_struct
*vma
)
4783 size_t sz
= vma
->vm_end
- vma
->vm_start
;
4787 ptr
= io_uring_validate_mmap_request(file
, vma
->vm_pgoff
, sz
);
4789 return PTR_ERR(ptr
);
4791 pfn
= virt_to_phys(ptr
) >> PAGE_SHIFT
;
4792 return remap_pfn_range(vma
, vma
->vm_start
, pfn
, sz
, vma
->vm_page_prot
);
4795 #else /* !CONFIG_MMU */
4797 static int io_uring_mmap(struct file
*file
, struct vm_area_struct
*vma
)
4799 return vma
->vm_flags
& (VM_SHARED
| VM_MAYSHARE
) ? 0 : -EINVAL
;
4802 static unsigned int io_uring_nommu_mmap_capabilities(struct file
*file
)
4804 return NOMMU_MAP_DIRECT
| NOMMU_MAP_READ
| NOMMU_MAP_WRITE
;
4807 static unsigned long io_uring_nommu_get_unmapped_area(struct file
*file
,
4808 unsigned long addr
, unsigned long len
,
4809 unsigned long pgoff
, unsigned long flags
)
4813 ptr
= io_uring_validate_mmap_request(file
, pgoff
, len
);
4815 return PTR_ERR(ptr
);
4817 return (unsigned long) ptr
;
4820 #endif /* !CONFIG_MMU */
4822 SYSCALL_DEFINE6(io_uring_enter
, unsigned int, fd
, u32
, to_submit
,
4823 u32
, min_complete
, u32
, flags
, const sigset_t __user
*, sig
,
4826 struct io_ring_ctx
*ctx
;
4831 if (flags
& ~(IORING_ENTER_GETEVENTS
| IORING_ENTER_SQ_WAKEUP
))
4839 if (f
.file
->f_op
!= &io_uring_fops
)
4843 ctx
= f
.file
->private_data
;
4844 if (!percpu_ref_tryget(&ctx
->refs
))
4848 * For SQ polling, the thread will do all submissions and completions.
4849 * Just return the requested submit count, and wake the thread if
4853 if (ctx
->flags
& IORING_SETUP_SQPOLL
) {
4854 if (!list_empty_careful(&ctx
->cq_overflow_list
))
4855 io_cqring_overflow_flush(ctx
, false);
4856 if (flags
& IORING_ENTER_SQ_WAKEUP
)
4857 wake_up(&ctx
->sqo_wait
);
4858 submitted
= to_submit
;
4859 } else if (to_submit
) {
4860 struct mm_struct
*cur_mm
;
4862 to_submit
= min(to_submit
, ctx
->sq_entries
);
4863 mutex_lock(&ctx
->uring_lock
);
4864 /* already have mm, so io_submit_sqes() won't try to grab it */
4865 cur_mm
= ctx
->sqo_mm
;
4866 submitted
= io_submit_sqes(ctx
, to_submit
, f
.file
, fd
,
4868 mutex_unlock(&ctx
->uring_lock
);
4870 if (flags
& IORING_ENTER_GETEVENTS
) {
4871 unsigned nr_events
= 0;
4873 min_complete
= min(min_complete
, ctx
->cq_entries
);
4875 if (ctx
->flags
& IORING_SETUP_IOPOLL
) {
4876 ret
= io_iopoll_check(ctx
, &nr_events
, min_complete
);
4878 ret
= io_cqring_wait(ctx
, min_complete
, sig
, sigsz
);
4882 percpu_ref_put(&ctx
->refs
);
4885 return submitted
? submitted
: ret
;
4888 static const struct file_operations io_uring_fops
= {
4889 .release
= io_uring_release
,
4890 .flush
= io_uring_flush
,
4891 .mmap
= io_uring_mmap
,
4893 .get_unmapped_area
= io_uring_nommu_get_unmapped_area
,
4894 .mmap_capabilities
= io_uring_nommu_mmap_capabilities
,
4896 .poll
= io_uring_poll
,
4897 .fasync
= io_uring_fasync
,
4900 static int io_allocate_scq_urings(struct io_ring_ctx
*ctx
,
4901 struct io_uring_params
*p
)
4903 struct io_rings
*rings
;
4904 size_t size
, sq_array_offset
;
4906 size
= rings_size(p
->sq_entries
, p
->cq_entries
, &sq_array_offset
);
4907 if (size
== SIZE_MAX
)
4910 rings
= io_mem_alloc(size
);
4915 ctx
->sq_array
= (u32
*)((char *)rings
+ sq_array_offset
);
4916 rings
->sq_ring_mask
= p
->sq_entries
- 1;
4917 rings
->cq_ring_mask
= p
->cq_entries
- 1;
4918 rings
->sq_ring_entries
= p
->sq_entries
;
4919 rings
->cq_ring_entries
= p
->cq_entries
;
4920 ctx
->sq_mask
= rings
->sq_ring_mask
;
4921 ctx
->cq_mask
= rings
->cq_ring_mask
;
4922 ctx
->sq_entries
= rings
->sq_ring_entries
;
4923 ctx
->cq_entries
= rings
->cq_ring_entries
;
4925 size
= array_size(sizeof(struct io_uring_sqe
), p
->sq_entries
);
4926 if (size
== SIZE_MAX
) {
4927 io_mem_free(ctx
->rings
);
4932 ctx
->sq_sqes
= io_mem_alloc(size
);
4933 if (!ctx
->sq_sqes
) {
4934 io_mem_free(ctx
->rings
);
4943 * Allocate an anonymous fd, this is what constitutes the application
4944 * visible backing of an io_uring instance. The application mmaps this
4945 * fd to gain access to the SQ/CQ ring details. If UNIX sockets are enabled,
4946 * we have to tie this fd to a socket for file garbage collection purposes.
4948 static int io_uring_get_fd(struct io_ring_ctx
*ctx
)
4953 #if defined(CONFIG_UNIX)
4954 ret
= sock_create_kern(&init_net
, PF_UNIX
, SOCK_RAW
, IPPROTO_IP
,
4960 ret
= get_unused_fd_flags(O_RDWR
| O_CLOEXEC
);
4964 file
= anon_inode_getfile("[io_uring]", &io_uring_fops
, ctx
,
4965 O_RDWR
| O_CLOEXEC
);
4968 ret
= PTR_ERR(file
);
4972 #if defined(CONFIG_UNIX)
4973 ctx
->ring_sock
->file
= file
;
4974 ctx
->ring_sock
->sk
->sk_user_data
= ctx
;
4976 fd_install(ret
, file
);
4979 #if defined(CONFIG_UNIX)
4980 sock_release(ctx
->ring_sock
);
4981 ctx
->ring_sock
= NULL
;
4986 static int io_uring_create(unsigned entries
, struct io_uring_params
*p
)
4988 struct user_struct
*user
= NULL
;
4989 struct io_ring_ctx
*ctx
;
4993 if (!entries
|| entries
> IORING_MAX_ENTRIES
)
4997 * Use twice as many entries for the CQ ring. It's possible for the
4998 * application to drive a higher depth than the size of the SQ ring,
4999 * since the sqes are only used at submission time. This allows for
5000 * some flexibility in overcommitting a bit. If the application has
5001 * set IORING_SETUP_CQSIZE, it will have passed in the desired number
5002 * of CQ ring entries manually.
5004 p
->sq_entries
= roundup_pow_of_two(entries
);
5005 if (p
->flags
& IORING_SETUP_CQSIZE
) {
5007 * If IORING_SETUP_CQSIZE is set, we do the same roundup
5008 * to a power-of-two, if it isn't already. We do NOT impose
5009 * any cq vs sq ring sizing.
5011 if (p
->cq_entries
< p
->sq_entries
|| p
->cq_entries
> IORING_MAX_CQ_ENTRIES
)
5013 p
->cq_entries
= roundup_pow_of_two(p
->cq_entries
);
5015 p
->cq_entries
= 2 * p
->sq_entries
;
5018 user
= get_uid(current_user());
5019 account_mem
= !capable(CAP_IPC_LOCK
);
5022 ret
= io_account_mem(user
,
5023 ring_pages(p
->sq_entries
, p
->cq_entries
));
5030 ctx
= io_ring_ctx_alloc(p
);
5033 io_unaccount_mem(user
, ring_pages(p
->sq_entries
,
5038 ctx
->compat
= in_compat_syscall();
5039 ctx
->account_mem
= account_mem
;
5041 ctx
->creds
= get_current_cred();
5043 ret
= io_allocate_scq_urings(ctx
, p
);
5047 ret
= io_sq_offload_start(ctx
, p
);
5051 memset(&p
->sq_off
, 0, sizeof(p
->sq_off
));
5052 p
->sq_off
.head
= offsetof(struct io_rings
, sq
.head
);
5053 p
->sq_off
.tail
= offsetof(struct io_rings
, sq
.tail
);
5054 p
->sq_off
.ring_mask
= offsetof(struct io_rings
, sq_ring_mask
);
5055 p
->sq_off
.ring_entries
= offsetof(struct io_rings
, sq_ring_entries
);
5056 p
->sq_off
.flags
= offsetof(struct io_rings
, sq_flags
);
5057 p
->sq_off
.dropped
= offsetof(struct io_rings
, sq_dropped
);
5058 p
->sq_off
.array
= (char *)ctx
->sq_array
- (char *)ctx
->rings
;
5060 memset(&p
->cq_off
, 0, sizeof(p
->cq_off
));
5061 p
->cq_off
.head
= offsetof(struct io_rings
, cq
.head
);
5062 p
->cq_off
.tail
= offsetof(struct io_rings
, cq
.tail
);
5063 p
->cq_off
.ring_mask
= offsetof(struct io_rings
, cq_ring_mask
);
5064 p
->cq_off
.ring_entries
= offsetof(struct io_rings
, cq_ring_entries
);
5065 p
->cq_off
.overflow
= offsetof(struct io_rings
, cq_overflow
);
5066 p
->cq_off
.cqes
= offsetof(struct io_rings
, cqes
);
5069 * Install ring fd as the very last thing, so we don't risk someone
5070 * having closed it before we finish setup
5072 ret
= io_uring_get_fd(ctx
);
5076 p
->features
= IORING_FEAT_SINGLE_MMAP
| IORING_FEAT_NODROP
|
5077 IORING_FEAT_SUBMIT_STABLE
;
5078 trace_io_uring_create(ret
, ctx
, p
->sq_entries
, p
->cq_entries
, p
->flags
);
5081 io_ring_ctx_wait_and_kill(ctx
);
5086 * Sets up an aio uring context, and returns the fd. Applications asks for a
5087 * ring size, we return the actual sq/cq ring sizes (among other things) in the
5088 * params structure passed in.
5090 static long io_uring_setup(u32 entries
, struct io_uring_params __user
*params
)
5092 struct io_uring_params p
;
5096 if (copy_from_user(&p
, params
, sizeof(p
)))
5098 for (i
= 0; i
< ARRAY_SIZE(p
.resv
); i
++) {
5103 if (p
.flags
& ~(IORING_SETUP_IOPOLL
| IORING_SETUP_SQPOLL
|
5104 IORING_SETUP_SQ_AFF
| IORING_SETUP_CQSIZE
))
5107 ret
= io_uring_create(entries
, &p
);
5111 if (copy_to_user(params
, &p
, sizeof(p
)))
5117 SYSCALL_DEFINE2(io_uring_setup
, u32
, entries
,
5118 struct io_uring_params __user
*, params
)
5120 return io_uring_setup(entries
, params
);
5123 static int __io_uring_register(struct io_ring_ctx
*ctx
, unsigned opcode
,
5124 void __user
*arg
, unsigned nr_args
)
5125 __releases(ctx
->uring_lock
)
5126 __acquires(ctx
->uring_lock
)
5131 * We're inside the ring mutex, if the ref is already dying, then
5132 * someone else killed the ctx or is already going through
5133 * io_uring_register().
5135 if (percpu_ref_is_dying(&ctx
->refs
))
5138 percpu_ref_kill(&ctx
->refs
);
5141 * Drop uring mutex before waiting for references to exit. If another
5142 * thread is currently inside io_uring_enter() it might need to grab
5143 * the uring_lock to make progress. If we hold it here across the drain
5144 * wait, then we can deadlock. It's safe to drop the mutex here, since
5145 * no new references will come in after we've killed the percpu ref.
5147 mutex_unlock(&ctx
->uring_lock
);
5148 wait_for_completion(&ctx
->completions
[0]);
5149 mutex_lock(&ctx
->uring_lock
);
5152 case IORING_REGISTER_BUFFERS
:
5153 ret
= io_sqe_buffer_register(ctx
, arg
, nr_args
);
5155 case IORING_UNREGISTER_BUFFERS
:
5159 ret
= io_sqe_buffer_unregister(ctx
);
5161 case IORING_REGISTER_FILES
:
5162 ret
= io_sqe_files_register(ctx
, arg
, nr_args
);
5164 case IORING_UNREGISTER_FILES
:
5168 ret
= io_sqe_files_unregister(ctx
);
5170 case IORING_REGISTER_FILES_UPDATE
:
5171 ret
= io_sqe_files_update(ctx
, arg
, nr_args
);
5173 case IORING_REGISTER_EVENTFD
:
5177 ret
= io_eventfd_register(ctx
, arg
);
5179 case IORING_UNREGISTER_EVENTFD
:
5183 ret
= io_eventfd_unregister(ctx
);
5190 /* bring the ctx back to life */
5191 reinit_completion(&ctx
->completions
[0]);
5192 percpu_ref_reinit(&ctx
->refs
);
5196 SYSCALL_DEFINE4(io_uring_register
, unsigned int, fd
, unsigned int, opcode
,
5197 void __user
*, arg
, unsigned int, nr_args
)
5199 struct io_ring_ctx
*ctx
;
5208 if (f
.file
->f_op
!= &io_uring_fops
)
5211 ctx
= f
.file
->private_data
;
5213 mutex_lock(&ctx
->uring_lock
);
5214 ret
= __io_uring_register(ctx
, opcode
, arg
, nr_args
);
5215 mutex_unlock(&ctx
->uring_lock
);
5216 trace_io_uring_register(ctx
, opcode
, ctx
->nr_user_files
, ctx
->nr_user_bufs
,
5217 ctx
->cq_ev_fd
!= NULL
, ret
);
5223 static int __init
io_uring_init(void)
5225 req_cachep
= KMEM_CACHE(io_kiocb
, SLAB_HWCACHE_ALIGN
| SLAB_PANIC
);
5228 __initcall(io_uring_init
);