1// SPDX-License-Identifier: GPL-2.0
2/*
3 * Shared application/kernel submission and completion ring pairs, for
4 * supporting fast/efficient IO.
5 *
6 * A note on the read/write ordering memory barriers that are matched between
7 * the application and kernel side. When the application reads the CQ ring
8 * tail, it must use an appropriate smp_rmb() to order with the smp_wmb()
9 * the kernel uses after writing the tail. Failure to do so could cause a
10 * delay in when the application notices that completion events available.
11 * This isn't a fatal condition. Likewise, the application must use an
12 * appropriate smp_wmb() both before writing the SQ tail, and after writing
13 * the SQ tail. The first one orders the sqe writes with the tail write, and
14 * the latter is paired with the smp_rmb() the kernel will issue before
15 * reading the SQ tail on submission.
16 *
17 * Also see the examples in the liburing library:
18 *
19 * git://git.kernel.dk/liburing
20 *
21 * io_uring also uses READ/WRITE_ONCE() for _any_ store or load that happens
22 * from data shared between the kernel and application. This is done both
23 * for ordering purposes, but also to ensure that once a value is loaded from
24 * data that the application could potentially modify, it remains stable.
25 *
26 * Copyright (C) 2018-2019 Jens Axboe
27 * Copyright (c) 2018-2019 Christoph Hellwig
28 */
29#include <linux/kernel.h>
30#include <linux/init.h>
31#include <linux/errno.h>
32#include <linux/syscalls.h>
33#include <linux/compat.h>
34#include <linux/refcount.h>
35#include <linux/uio.h>
36
37#include <linux/sched/signal.h>
38#include <linux/fs.h>
39#include <linux/file.h>
40#include <linux/fdtable.h>
41#include <linux/mm.h>
42#include <linux/mman.h>
43#include <linux/mmu_context.h>
44#include <linux/percpu.h>
45#include <linux/slab.h>
46#include <linux/workqueue.h>
47#include <linux/kthread.h>
48#include <linux/blkdev.h>
49#include <linux/bvec.h>
50#include <linux/net.h>
51#include <net/sock.h>
52#include <net/af_unix.h>
53#include <net/scm.h>
54#include <linux/anon_inodes.h>
55#include <linux/sched/mm.h>
56#include <linux/uaccess.h>
57#include <linux/nospec.h>
58#include <linux/sizes.h>
59#include <linux/hugetlb.h>
60
61#include <uapi/linux/io_uring.h>
62
63#include "internal.h"
64
65#define IORING_MAX_ENTRIES 4096
66#define IORING_MAX_FIXED_FILES 1024
67
68struct io_uring {
69 u32 head ____cacheline_aligned_in_smp;
70 u32 tail ____cacheline_aligned_in_smp;
71};
72
73struct io_sq_ring {
74 struct io_uring r;
75 u32 ring_mask;
76 u32 ring_entries;
77 u32 dropped;
78 u32 flags;
79 u32 array[];
80};
81
82struct io_cq_ring {
83 struct io_uring r;
84 u32 ring_mask;
85 u32 ring_entries;
86 u32 overflow;
87 struct io_uring_cqe cqes[];
88};
89
90struct io_mapped_ubuf {
91 u64 ubuf;
92 size_t len;
93 struct bio_vec *bvec;
94 unsigned int nr_bvecs;
95};
96
97struct async_list {
98 spinlock_t lock;
99 atomic_t cnt;
100 struct list_head list;
101
102 struct file *file;
103 off_t io_end;
104 size_t io_pages;
105};
106
107struct io_ring_ctx {
108 struct {
109 struct percpu_ref refs;
110 } ____cacheline_aligned_in_smp;
111
112 struct {
113 unsigned int flags;
114 bool compat;
115 bool account_mem;
116
117 /* SQ ring */
118 struct io_sq_ring *sq_ring;
119 unsigned cached_sq_head;
120 unsigned sq_entries;
121 unsigned sq_mask;
122 unsigned sq_thread_idle;
123 struct io_uring_sqe *sq_sqes;
124 } ____cacheline_aligned_in_smp;
125
126 /* IO offload */
127 struct workqueue_struct *sqo_wq;
128 struct task_struct *sqo_thread; /* if using sq thread polling */
129 struct mm_struct *sqo_mm;
130 wait_queue_head_t sqo_wait;
131 unsigned sqo_stop;
132
133 struct {
134 /* CQ ring */
135 struct io_cq_ring *cq_ring;
136 unsigned cached_cq_tail;
137 unsigned cq_entries;
138 unsigned cq_mask;
139 struct wait_queue_head cq_wait;
140 struct fasync_struct *cq_fasync;
141 } ____cacheline_aligned_in_smp;
142
143 /*
144 * If used, fixed file set. Writers must ensure that ->refs is dead,
145 * readers must ensure that ->refs is alive as long as the file* is
146 * used. Only updated through io_uring_register(2).
147 */
148 struct file **user_files;
149 unsigned nr_user_files;
150
151 /* if used, fixed mapped user buffers */
152 unsigned nr_user_bufs;
153 struct io_mapped_ubuf *user_bufs;
154
155 struct user_struct *user;
156
157 struct completion ctx_done;
158
159 struct {
160 struct mutex uring_lock;
161 wait_queue_head_t wait;
162 } ____cacheline_aligned_in_smp;
163
164 struct {
165 spinlock_t completion_lock;
166 bool poll_multi_file;
167 /*
168 * ->poll_list is protected by the ctx->uring_lock for
169 * io_uring instances that don't use IORING_SETUP_SQPOLL.
170 * For SQPOLL, only the single threaded io_sq_thread() will
171 * manipulate the list, hence no extra locking is needed there.
172 */
173 struct list_head poll_list;
174 struct list_head cancel_list;
175 } ____cacheline_aligned_in_smp;
176
177 struct async_list pending_async[2];
178
179#if defined(CONFIG_UNIX)
180 struct socket *ring_sock;
181#endif
182};
183
184struct sqe_submit {
185 const struct io_uring_sqe *sqe;
186 unsigned short index;
187 bool has_user;
188 bool needs_lock;
189 bool needs_fixed_file;
190};
191
192/*
193 * First field must be the file pointer in all the
194 * iocb unions! See also 'struct kiocb' in <linux/fs.h>
195 */
196struct io_poll_iocb {
197 struct file *file;
198 struct wait_queue_head *head;
199 __poll_t events;
200 bool done;
201 bool canceled;
202 struct wait_queue_entry wait;
203};
204
205/*
206 * NOTE! Each of the iocb union members has the file pointer
207 * as the first entry in their struct definition. So you can
208 * access the file pointer through any of the sub-structs,
209 * or directly as just 'ki_filp' in this struct.
210 */
211struct io_kiocb {
212 union {
213 struct file *file;
214 struct kiocb rw;
215 struct io_poll_iocb poll;
216 };
217
218 struct sqe_submit submit;
219
220 struct io_ring_ctx *ctx;
221 struct list_head list;
222 unsigned int flags;
223 refcount_t refs;
224#define REQ_F_FORCE_NONBLOCK 1 /* inline submission attempt */
225#define REQ_F_IOPOLL_COMPLETED 2 /* polled IO has completed */
226#define REQ_F_FIXED_FILE 4 /* ctx owns file */
227#define REQ_F_SEQ_PREV 8 /* sequential with previous */
228#define REQ_F_PREPPED 16 /* prep already done */
229 u64 user_data;
230 u64 error;
231
232 struct work_struct work;
233};
234
235#define IO_PLUG_THRESHOLD 2
236#define IO_IOPOLL_BATCH 8
237
238struct io_submit_state {
239 struct blk_plug plug;
240
241 /*
242 * io_kiocb alloc cache
243 */
244 void *reqs[IO_IOPOLL_BATCH];
245 unsigned int free_reqs;
246 unsigned int cur_req;
247
248 /*
249 * File reference cache
250 */
251 struct file *file;
252 unsigned int fd;
253 unsigned int has_refs;
254 unsigned int used_refs;
255 unsigned int ios_left;
256};
257
258static struct kmem_cache *req_cachep;
259
260static const struct file_operations io_uring_fops;
261
262struct sock *io_uring_get_socket(struct file *file)
263{
264#if defined(CONFIG_UNIX)
265 if (file->f_op == &io_uring_fops) {
266 struct io_ring_ctx *ctx = file->private_data;
267
268 return ctx->ring_sock->sk;
269 }
270#endif
271 return NULL;
272}
273EXPORT_SYMBOL(io_uring_get_socket);
274
275static void io_ring_ctx_ref_free(struct percpu_ref *ref)
276{
277 struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);
278
279 complete(&ctx->ctx_done);
280}
281
282static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
283{
284 struct io_ring_ctx *ctx;
285 int i;
286
287 ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
288 if (!ctx)
289 return NULL;
290
291 if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free, 0, GFP_KERNEL)) {
292 kfree(ctx);
293 return NULL;
294 }
295
296 ctx->flags = p->flags;
297 init_waitqueue_head(&ctx->cq_wait);
298 init_completion(&ctx->ctx_done);
299 mutex_init(&ctx->uring_lock);
300 init_waitqueue_head(&ctx->wait);
301 for (i = 0; i < ARRAY_SIZE(ctx->pending_async); i++) {
302 spin_lock_init(&ctx->pending_async[i].lock);
303 INIT_LIST_HEAD(&ctx->pending_async[i].list);
304 atomic_set(&ctx->pending_async[i].cnt, 0);
305 }
306 spin_lock_init(&ctx->completion_lock);
307 INIT_LIST_HEAD(&ctx->poll_list);
308 INIT_LIST_HEAD(&ctx->cancel_list);
309 return ctx;
310}
311
312static void io_commit_cqring(struct io_ring_ctx *ctx)
313{
314 struct io_cq_ring *ring = ctx->cq_ring;
315
316 if (ctx->cached_cq_tail != READ_ONCE(ring->r.tail)) {
317 /* order cqe stores with ring update */
318 smp_store_release(&ring->r.tail, ctx->cached_cq_tail);
319
320 /*
321 * Write sider barrier of tail update, app has read side. See
322 * comment at the top of this file.
323 */
324 smp_wmb();
325
326 if (wq_has_sleeper(&ctx->cq_wait)) {
327 wake_up_interruptible(&ctx->cq_wait);
328 kill_fasync(&ctx->cq_fasync, SIGIO, POLL_IN);
329 }
330 }
331}
332
333static struct io_uring_cqe *io_get_cqring(struct io_ring_ctx *ctx)
334{
335 struct io_cq_ring *ring = ctx->cq_ring;
336 unsigned tail;
337
338 tail = ctx->cached_cq_tail;
339 /* See comment at the top of the file */
340 smp_rmb();
341 if (tail + 1 == READ_ONCE(ring->r.head))
342 return NULL;
343
344 ctx->cached_cq_tail++;
345 return &ring->cqes[tail & ctx->cq_mask];
346}
347
348static void io_cqring_fill_event(struct io_ring_ctx *ctx, u64 ki_user_data,
349 long res, unsigned ev_flags)
350{
351 struct io_uring_cqe *cqe;
352
353 /*
354 * If we can't get a cq entry, userspace overflowed the
355 * submission (by quite a lot). Increment the overflow count in
356 * the ring.
357 */
358 cqe = io_get_cqring(ctx);
359 if (cqe) {
360 WRITE_ONCE(cqe->user_data, ki_user_data);
361 WRITE_ONCE(cqe->res, res);
362 WRITE_ONCE(cqe->flags, ev_flags);
363 } else {
364 unsigned overflow = READ_ONCE(ctx->cq_ring->overflow);
365
366 WRITE_ONCE(ctx->cq_ring->overflow, overflow + 1);
367 }
368}
369
370static void io_cqring_ev_posted(struct io_ring_ctx *ctx)
371{
372 if (waitqueue_active(&ctx->wait))
373 wake_up(&ctx->wait);
374 if (waitqueue_active(&ctx->sqo_wait))
375 wake_up(&ctx->sqo_wait);
376}
377
378static void io_cqring_add_event(struct io_ring_ctx *ctx, u64 user_data,
379 long res, unsigned ev_flags)
380{
381 unsigned long flags;
382
383 spin_lock_irqsave(&ctx->completion_lock, flags);
384 io_cqring_fill_event(ctx, user_data, res, ev_flags);
385 io_commit_cqring(ctx);
386 spin_unlock_irqrestore(&ctx->completion_lock, flags);
387
388 io_cqring_ev_posted(ctx);
389}
390
391static void io_ring_drop_ctx_refs(struct io_ring_ctx *ctx, unsigned refs)
392{
393 percpu_ref_put_many(&ctx->refs, refs);
394
395 if (waitqueue_active(&ctx->wait))
396 wake_up(&ctx->wait);
397}
398
399static struct io_kiocb *io_get_req(struct io_ring_ctx *ctx,
400 struct io_submit_state *state)
401{
402 gfp_t gfp = GFP_KERNEL | __GFP_NOWARN;
403 struct io_kiocb *req;
404
405 if (!percpu_ref_tryget(&ctx->refs))
406 return NULL;
407
408 if (!state) {
409 req = kmem_cache_alloc(req_cachep, gfp);
410 if (unlikely(!req))
411 goto out;
412 } else if (!state->free_reqs) {
413 size_t sz;
414 int ret;
415
416 sz = min_t(size_t, state->ios_left, ARRAY_SIZE(state->reqs));
417 ret = kmem_cache_alloc_bulk(req_cachep, gfp, sz, state->reqs);
418
419 /*
420 * Bulk alloc is all-or-nothing. If we fail to get a batch,
421 * retry single alloc to be on the safe side.
422 */
423 if (unlikely(ret <= 0)) {
424 state->reqs[0] = kmem_cache_alloc(req_cachep, gfp);
425 if (!state->reqs[0])
426 goto out;
427 ret = 1;
428 }
429 state->free_reqs = ret - 1;
430 state->cur_req = 1;
431 req = state->reqs[0];
432 } else {
433 req = state->reqs[state->cur_req];
434 state->free_reqs--;
435 state->cur_req++;
436 }
437
438 req->ctx = ctx;
439 req->flags = 0;
440 /* one is dropped after submission, the other at completion */
441 refcount_set(&req->refs, 2);
442 return req;
443out:
444 io_ring_drop_ctx_refs(ctx, 1);
445 return NULL;
446}
447
448static void io_free_req_many(struct io_ring_ctx *ctx, void **reqs, int *nr)
449{
450 if (*nr) {
451 kmem_cache_free_bulk(req_cachep, *nr, reqs);
452 io_ring_drop_ctx_refs(ctx, *nr);
453 *nr = 0;
454 }
455}
456
457static void io_free_req(struct io_kiocb *req)
458{
459 if (req->file && !(req->flags & REQ_F_FIXED_FILE))
460 fput(req->file);
461 io_ring_drop_ctx_refs(req->ctx, 1);
462 kmem_cache_free(req_cachep, req);
463}
464
465static void io_put_req(struct io_kiocb *req)
466{
467 if (refcount_dec_and_test(&req->refs))
468 io_free_req(req);
469}
470
471/*
472 * Find and free completed poll iocbs
473 */
474static void io_iopoll_complete(struct io_ring_ctx *ctx, unsigned int *nr_events,
475 struct list_head *done)
476{
477 void *reqs[IO_IOPOLL_BATCH];
478 struct io_kiocb *req;
479 int to_free;
480
481 to_free = 0;
482 while (!list_empty(done)) {
483 req = list_first_entry(done, struct io_kiocb, list);
484 list_del(&req->list);
485
486 io_cqring_fill_event(ctx, req->user_data, req->error, 0);
487 (*nr_events)++;
488
489 if (refcount_dec_and_test(&req->refs)) {
490 /* If we're not using fixed files, we have to pair the
491 * completion part with the file put. Use regular
492 * completions for those, only batch free for fixed
493 * file.
494 */
495 if (req->flags & REQ_F_FIXED_FILE) {
496 reqs[to_free++] = req;
497 if (to_free == ARRAY_SIZE(reqs))
498 io_free_req_many(ctx, reqs, &to_free);
499 } else {
500 io_free_req(req);
501 }
502 }
503 }
504
505 io_commit_cqring(ctx);
506 io_free_req_many(ctx, reqs, &to_free);
507}
508
509static int io_do_iopoll(struct io_ring_ctx *ctx, unsigned int *nr_events,
510 long min)
511{
512 struct io_kiocb *req, *tmp;
513 LIST_HEAD(done);
514 bool spin;
515 int ret;
516
517 /*
518 * Only spin for completions if we don't have multiple devices hanging
519 * off our complete list, and we're under the requested amount.
520 */
521 spin = !ctx->poll_multi_file && *nr_events < min;
522
523 ret = 0;
524 list_for_each_entry_safe(req, tmp, &ctx->poll_list, list) {
525 struct kiocb *kiocb = &req->rw;
526
527 /*
528 * Move completed entries to our local list. If we find a
529 * request that requires polling, break out and complete
530 * the done list first, if we have entries there.
531 */
532 if (req->flags & REQ_F_IOPOLL_COMPLETED) {
533 list_move_tail(&req->list, &done);
534 continue;
535 }
536 if (!list_empty(&done))
537 break;
538
539 ret = kiocb->ki_filp->f_op->iopoll(kiocb, spin);
540 if (ret < 0)
541 break;
542
543 if (ret && spin)
544 spin = false;
545 ret = 0;
546 }
547
548 if (!list_empty(&done))
549 io_iopoll_complete(ctx, nr_events, &done);
550
551 return ret;
552}
553
554/*
555 * Poll for a mininum of 'min' events. Note that if min == 0 we consider that a
556 * non-spinning poll check - we'll still enter the driver poll loop, but only
557 * as a non-spinning completion check.
558 */
559static int io_iopoll_getevents(struct io_ring_ctx *ctx, unsigned int *nr_events,
560 long min)
561{
562 while (!list_empty(&ctx->poll_list)) {
563 int ret;
564
565 ret = io_do_iopoll(ctx, nr_events, min);
566 if (ret < 0)
567 return ret;
568 if (!min || *nr_events >= min)
569 return 0;
570 }
571
572 return 1;
573}
574
575/*
576 * We can't just wait for polled events to come to us, we have to actively
577 * find and complete them.
578 */
579static void io_iopoll_reap_events(struct io_ring_ctx *ctx)
580{
581 if (!(ctx->flags & IORING_SETUP_IOPOLL))
582 return;
583
584 mutex_lock(&ctx->uring_lock);
585 while (!list_empty(&ctx->poll_list)) {
586 unsigned int nr_events = 0;
587
588 io_iopoll_getevents(ctx, &nr_events, 1);
589 }
590 mutex_unlock(&ctx->uring_lock);
591}
592
593static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned *nr_events,
594 long min)
595{
596 int ret = 0;
597
598 do {
599 int tmin = 0;
600
601 if (*nr_events < min)
602 tmin = min - *nr_events;
603
604 ret = io_iopoll_getevents(ctx, nr_events, tmin);
605 if (ret <= 0)
606 break;
607 ret = 0;
608 } while (min && !*nr_events && !need_resched());
609
610 return ret;
611}
612
613static void kiocb_end_write(struct kiocb *kiocb)
614{
615 if (kiocb->ki_flags & IOCB_WRITE) {
616 struct inode *inode = file_inode(kiocb->ki_filp);
617
618 /*
619 * Tell lockdep we inherited freeze protection from submission
620 * thread.
621 */
622 if (S_ISREG(inode->i_mode))
623 __sb_writers_acquired(inode->i_sb, SB_FREEZE_WRITE);
624 file_end_write(kiocb->ki_filp);
625 }
626}
627
628static void io_complete_rw(struct kiocb *kiocb, long res, long res2)
629{
630 struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw);
631
632 kiocb_end_write(kiocb);
633
634 io_cqring_add_event(req->ctx, req->user_data, res, 0);
635 io_put_req(req);
636}
637
638static void io_complete_rw_iopoll(struct kiocb *kiocb, long res, long res2)
639{
640 struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw);
641
642 kiocb_end_write(kiocb);
643
644 req->error = res;
645 if (res != -EAGAIN)
646 req->flags |= REQ_F_IOPOLL_COMPLETED;
647}
648
649/*
650 * After the iocb has been issued, it's safe to be found on the poll list.
651 * Adding the kiocb to the list AFTER submission ensures that we don't
652 * find it from a io_iopoll_getevents() thread before the issuer is done
653 * accessing the kiocb cookie.
654 */
655static void io_iopoll_req_issued(struct io_kiocb *req)
656{
657 struct io_ring_ctx *ctx = req->ctx;
658
659 /*
660 * Track whether we have multiple files in our lists. This will impact
661 * how we do polling eventually, not spinning if we're on potentially
662 * different devices.
663 */
664 if (list_empty(&ctx->poll_list)) {
665 ctx->poll_multi_file = false;
666 } else if (!ctx->poll_multi_file) {
667 struct io_kiocb *list_req;
668
669 list_req = list_first_entry(&ctx->poll_list, struct io_kiocb,
670 list);
671 if (list_req->rw.ki_filp != req->rw.ki_filp)
672 ctx->poll_multi_file = true;
673 }
674
675 /*
676 * For fast devices, IO may have already completed. If it has, add
677 * it to the front so we find it first.
678 */
679 if (req->flags & REQ_F_IOPOLL_COMPLETED)
680 list_add(&req->list, &ctx->poll_list);
681 else
682 list_add_tail(&req->list, &ctx->poll_list);
683}
684
685static void io_file_put(struct io_submit_state *state, struct file *file)
686{
687 if (!state) {
688 fput(file);
689 } else if (state->file) {
690 int diff = state->has_refs - state->used_refs;
691
692 if (diff)
693 fput_many(state->file, diff);
694 state->file = NULL;
695 }
696}
697
698/*
699 * Get as many references to a file as we have IOs left in this submission,
700 * assuming most submissions are for one file, or at least that each file
701 * has more than one submission.
702 */
703static struct file *io_file_get(struct io_submit_state *state, int fd)
704{
705 if (!state)
706 return fget(fd);
707
708 if (state->file) {
709 if (state->fd == fd) {
710 state->used_refs++;
711 state->ios_left--;
712 return state->file;
713 }
714 io_file_put(state, NULL);
715 }
716 state->file = fget_many(fd, state->ios_left);
717 if (!state->file)
718 return NULL;
719
720 state->fd = fd;
721 state->has_refs = state->ios_left;
722 state->used_refs = 1;
723 state->ios_left--;
724 return state->file;
725}
726
727/*
728 * If we tracked the file through the SCM inflight mechanism, we could support
729 * any file. For now, just ensure that anything potentially problematic is done
730 * inline.
731 */
732static bool io_file_supports_async(struct file *file)
733{
734 umode_t mode = file_inode(file)->i_mode;
735
736 if (S_ISBLK(mode) || S_ISCHR(mode))
737 return true;
738 if (S_ISREG(mode) && file->f_op != &io_uring_fops)
739 return true;
740
741 return false;
742}
743
744static int io_prep_rw(struct io_kiocb *req, const struct sqe_submit *s,
745 bool force_nonblock, struct io_submit_state *state)
746{
747 const struct io_uring_sqe *sqe = s->sqe;
748 struct io_ring_ctx *ctx = req->ctx;
749 struct kiocb *kiocb = &req->rw;
750 unsigned ioprio;
751 int ret;
752
753 if (!req->file)
754 return -EBADF;
755 /* For -EAGAIN retry, everything is already prepped */
756 if (req->flags & REQ_F_PREPPED)
757 return 0;
758
759 if (force_nonblock && !io_file_supports_async(req->file))
760 force_nonblock = false;
761
762 kiocb->ki_pos = READ_ONCE(sqe->off);
763 kiocb->ki_flags = iocb_flags(kiocb->ki_filp);
764 kiocb->ki_hint = ki_hint_validate(file_write_hint(kiocb->ki_filp));
765
766 ioprio = READ_ONCE(sqe->ioprio);
767 if (ioprio) {
768 ret = ioprio_check_cap(ioprio);
769 if (ret)
770 return ret;
771
772 kiocb->ki_ioprio = ioprio;
773 } else
774 kiocb->ki_ioprio = get_current_ioprio();
775
776 ret = kiocb_set_rw_flags(kiocb, READ_ONCE(sqe->rw_flags));
777 if (unlikely(ret))
778 return ret;
779 if (force_nonblock) {
780 kiocb->ki_flags |= IOCB_NOWAIT;
781 req->flags |= REQ_F_FORCE_NONBLOCK;
782 }
783 if (ctx->flags & IORING_SETUP_IOPOLL) {
784 if (!(kiocb->ki_flags & IOCB_DIRECT) ||
785 !kiocb->ki_filp->f_op->iopoll)
786 return -EOPNOTSUPP;
787
788 req->error = 0;
789 kiocb->ki_flags |= IOCB_HIPRI;
790 kiocb->ki_complete = io_complete_rw_iopoll;
791 } else {
792 if (kiocb->ki_flags & IOCB_HIPRI)
793 return -EINVAL;
794 kiocb->ki_complete = io_complete_rw;
795 }
796 req->flags |= REQ_F_PREPPED;
797 return 0;
798}
799
800static inline void io_rw_done(struct kiocb *kiocb, ssize_t ret)
801{
802 switch (ret) {
803 case -EIOCBQUEUED:
804 break;
805 case -ERESTARTSYS:
806 case -ERESTARTNOINTR:
807 case -ERESTARTNOHAND:
808 case -ERESTART_RESTARTBLOCK:
809 /*
810 * We can't just restart the syscall, since previously
811 * submitted sqes may already be in progress. Just fail this
812 * IO with EINTR.
813 */
814 ret = -EINTR;
815 /* fall through */
816 default:
817 kiocb->ki_complete(kiocb, ret, 0);
818 }
819}
820
821static int io_import_fixed(struct io_ring_ctx *ctx, int rw,
822 const struct io_uring_sqe *sqe,
823 struct iov_iter *iter)
824{
825 size_t len = READ_ONCE(sqe->len);
826 struct io_mapped_ubuf *imu;
827 unsigned index, buf_index;
828 size_t offset;
829 u64 buf_addr;
830
831 /* attempt to use fixed buffers without having provided iovecs */
832 if (unlikely(!ctx->user_bufs))
833 return -EFAULT;
834
835 buf_index = READ_ONCE(sqe->buf_index);
836 if (unlikely(buf_index >= ctx->nr_user_bufs))
837 return -EFAULT;
838
839 index = array_index_nospec(buf_index, ctx->nr_user_bufs);
840 imu = &ctx->user_bufs[index];
841 buf_addr = READ_ONCE(sqe->addr);
842
843 /* overflow */
844 if (buf_addr + len < buf_addr)
845 return -EFAULT;
846 /* not inside the mapped region */
847 if (buf_addr < imu->ubuf || buf_addr + len > imu->ubuf + imu->len)
848 return -EFAULT;
849
850 /*
851 * May not be a start of buffer, set size appropriately
852 * and advance us to the beginning.
853 */
854 offset = buf_addr - imu->ubuf;
855 iov_iter_bvec(iter, rw, imu->bvec, imu->nr_bvecs, offset + len);
856 if (offset)
857 iov_iter_advance(iter, offset);
858
859 /* don't drop a reference to these pages */
860 iter->type |= ITER_BVEC_FLAG_NO_REF;
861 return 0;
862}
863
864static int io_import_iovec(struct io_ring_ctx *ctx, int rw,
865 const struct sqe_submit *s, struct iovec **iovec,
866 struct iov_iter *iter)
867{
868 const struct io_uring_sqe *sqe = s->sqe;
869 void __user *buf = u64_to_user_ptr(READ_ONCE(sqe->addr));
870 size_t sqe_len = READ_ONCE(sqe->len);
871 u8 opcode;
872
873 /*
874 * We're reading ->opcode for the second time, but the first read
875 * doesn't care whether it's _FIXED or not, so it doesn't matter
876 * whether ->opcode changes concurrently. The first read does care
877 * about whether it is a READ or a WRITE, so we don't trust this read
878 * for that purpose and instead let the caller pass in the read/write
879 * flag.
880 */
881 opcode = READ_ONCE(sqe->opcode);
882 if (opcode == IORING_OP_READ_FIXED ||
883 opcode == IORING_OP_WRITE_FIXED) {
884 int ret = io_import_fixed(ctx, rw, sqe, iter);
885 *iovec = NULL;
886 return ret;
887 }
888
889 if (!s->has_user)
890 return -EFAULT;
891
892#ifdef CONFIG_COMPAT
893 if (ctx->compat)
894 return compat_import_iovec(rw, buf, sqe_len, UIO_FASTIOV,
895 iovec, iter);
896#endif
897
898 return import_iovec(rw, buf, sqe_len, UIO_FASTIOV, iovec, iter);
899}
900
901/*
902 * Make a note of the last file/offset/direction we punted to async
903 * context. We'll use this information to see if we can piggy back a
904 * sequential request onto the previous one, if it's still hasn't been
905 * completed by the async worker.
906 */
907static void io_async_list_note(int rw, struct io_kiocb *req, size_t len)
908{
909 struct async_list *async_list = &req->ctx->pending_async[rw];
910 struct kiocb *kiocb = &req->rw;
911 struct file *filp = kiocb->ki_filp;
912 off_t io_end = kiocb->ki_pos + len;
913
914 if (filp == async_list->file && kiocb->ki_pos == async_list->io_end) {
915 unsigned long max_pages;
916
917 /* Use 8x RA size as a decent limiter for both reads/writes */
918 max_pages = filp->f_ra.ra_pages;
919 if (!max_pages)
920 max_pages = VM_READAHEAD_PAGES;
921 max_pages *= 8;
922
923 /* If max pages are exceeded, reset the state */
924 len >>= PAGE_SHIFT;
925 if (async_list->io_pages + len <= max_pages) {
926 req->flags |= REQ_F_SEQ_PREV;
927 async_list->io_pages += len;
928 } else {
929 io_end = 0;
930 async_list->io_pages = 0;
931 }
932 }
933
934 /* New file? Reset state. */
935 if (async_list->file != filp) {
936 async_list->io_pages = 0;
937 async_list->file = filp;
938 }
939 async_list->io_end = io_end;
940}
941
942static int io_read(struct io_kiocb *req, const struct sqe_submit *s,
943 bool force_nonblock, struct io_submit_state *state)
944{
945 struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
946 struct kiocb *kiocb = &req->rw;
947 struct iov_iter iter;
948 struct file *file;
949 size_t iov_count;
950 int ret;
951
952 ret = io_prep_rw(req, s, force_nonblock, state);
953 if (ret)
954 return ret;
955 file = kiocb->ki_filp;
956
957 if (unlikely(!(file->f_mode & FMODE_READ)))
958 return -EBADF;
959 if (unlikely(!file->f_op->read_iter))
960 return -EINVAL;
961
962 ret = io_import_iovec(req->ctx, READ, s, &iovec, &iter);
963 if (ret)
964 return ret;
965
966 iov_count = iov_iter_count(&iter);
967 ret = rw_verify_area(READ, file, &kiocb->ki_pos, iov_count);
968 if (!ret) {
969 ssize_t ret2;
970
971 /* Catch -EAGAIN return for forced non-blocking submission */
972 ret2 = call_read_iter(file, kiocb, &iter);
973 if (!force_nonblock || ret2 != -EAGAIN) {
974 io_rw_done(kiocb, ret2);
975 } else {
976 /*
977 * If ->needs_lock is true, we're already in async
978 * context.
979 */
980 if (!s->needs_lock)
981 io_async_list_note(READ, req, iov_count);
982 ret = -EAGAIN;
983 }
984 }
985 kfree(iovec);
986 return ret;
987}
988
989static int io_write(struct io_kiocb *req, const struct sqe_submit *s,
990 bool force_nonblock, struct io_submit_state *state)
991{
992 struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
993 struct kiocb *kiocb = &req->rw;
994 struct iov_iter iter;
995 struct file *file;
996 size_t iov_count;
997 int ret;
998
999 ret = io_prep_rw(req, s, force_nonblock, state);
1000 if (ret)
1001 return ret;
1002
1003 file = kiocb->ki_filp;
1004 if (unlikely(!(file->f_mode & FMODE_WRITE)))
1005 return -EBADF;
1006 if (unlikely(!file->f_op->write_iter))
1007 return -EINVAL;
1008
1009 ret = io_import_iovec(req->ctx, WRITE, s, &iovec, &iter);
1010 if (ret)
1011 return ret;
1012
1013 iov_count = iov_iter_count(&iter);
1014
1015 ret = -EAGAIN;
1016 if (force_nonblock && !(kiocb->ki_flags & IOCB_DIRECT)) {
1017 /* If ->needs_lock is true, we're already in async context. */
1018 if (!s->needs_lock)
1019 io_async_list_note(WRITE, req, iov_count);
1020 goto out_free;
1021 }
1022
1023 ret = rw_verify_area(WRITE, file, &kiocb->ki_pos, iov_count);
1024 if (!ret) {
1025 /*
1026 * Open-code file_start_write here to grab freeze protection,
1027 * which will be released by another thread in
1028 * io_complete_rw(). Fool lockdep by telling it the lock got
1029 * released so that it doesn't complain about the held lock when
1030 * we return to userspace.
1031 */
1032 if (S_ISREG(file_inode(file)->i_mode)) {
1033 __sb_start_write(file_inode(file)->i_sb,
1034 SB_FREEZE_WRITE, true);
1035 __sb_writers_release(file_inode(file)->i_sb,
1036 SB_FREEZE_WRITE);
1037 }
1038 kiocb->ki_flags |= IOCB_WRITE;
1039 io_rw_done(kiocb, call_write_iter(file, kiocb, &iter));
1040 }
1041out_free:
1042 kfree(iovec);
1043 return ret;
1044}
1045
1046/*
1047 * IORING_OP_NOP just posts a completion event, nothing else.
1048 */
1049static int io_nop(struct io_kiocb *req, u64 user_data)
1050{
1051 struct io_ring_ctx *ctx = req->ctx;
1052 long err = 0;
1053
1054 if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
1055 return -EINVAL;
1056
1057 io_cqring_add_event(ctx, user_data, err, 0);
1058 io_put_req(req);
1059 return 0;
1060}
1061
1062static int io_prep_fsync(struct io_kiocb *req, const struct io_uring_sqe *sqe)
1063{
1064 struct io_ring_ctx *ctx = req->ctx;
1065
1066 if (!req->file)
1067 return -EBADF;
1068 /* Prep already done (EAGAIN retry) */
1069 if (req->flags & REQ_F_PREPPED)
1070 return 0;
1071
1072 if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
1073 return -EINVAL;
1074 if (unlikely(sqe->addr || sqe->ioprio || sqe->buf_index))
1075 return -EINVAL;
1076
1077 req->flags |= REQ_F_PREPPED;
1078 return 0;
1079}
1080
1081static int io_fsync(struct io_kiocb *req, const struct io_uring_sqe *sqe,
1082 bool force_nonblock)
1083{
1084 loff_t sqe_off = READ_ONCE(sqe->off);
1085 loff_t sqe_len = READ_ONCE(sqe->len);
1086 loff_t end = sqe_off + sqe_len;
1087 unsigned fsync_flags;
1088 int ret;
1089
1090 fsync_flags = READ_ONCE(sqe->fsync_flags);
1091 if (unlikely(fsync_flags & ~IORING_FSYNC_DATASYNC))
1092 return -EINVAL;
1093
1094 ret = io_prep_fsync(req, sqe);
1095 if (ret)
1096 return ret;
1097
1098 /* fsync always requires a blocking context */
1099 if (force_nonblock)
1100 return -EAGAIN;
1101
1102 ret = vfs_fsync_range(req->rw.ki_filp, sqe_off,
1103 end > 0 ? end : LLONG_MAX,
1104 fsync_flags & IORING_FSYNC_DATASYNC);
1105
1106 io_cqring_add_event(req->ctx, sqe->user_data, ret, 0);
1107 io_put_req(req);
1108 return 0;
1109}
1110
1111static void io_poll_remove_one(struct io_kiocb *req)
1112{
1113 struct io_poll_iocb *poll = &req->poll;
1114
1115 spin_lock(&poll->head->lock);
1116 WRITE_ONCE(poll->canceled, true);
1117 if (!list_empty(&poll->wait.entry)) {
1118 list_del_init(&poll->wait.entry);
1119 queue_work(req->ctx->sqo_wq, &req->work);
1120 }
1121 spin_unlock(&poll->head->lock);
1122
1123 list_del_init(&req->list);
1124}
1125
1126static void io_poll_remove_all(struct io_ring_ctx *ctx)
1127{
1128 struct io_kiocb *req;
1129
1130 spin_lock_irq(&ctx->completion_lock);
1131 while (!list_empty(&ctx->cancel_list)) {
1132 req = list_first_entry(&ctx->cancel_list, struct io_kiocb,list);
1133 io_poll_remove_one(req);
1134 }
1135 spin_unlock_irq(&ctx->completion_lock);
1136}
1137
1138/*
1139 * Find a running poll command that matches one specified in sqe->addr,
1140 * and remove it if found.
1141 */
1142static int io_poll_remove(struct io_kiocb *req, const struct io_uring_sqe *sqe)
1143{
1144 struct io_ring_ctx *ctx = req->ctx;
1145 struct io_kiocb *poll_req, *next;
1146 int ret = -ENOENT;
1147
1148 if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
1149 return -EINVAL;
1150 if (sqe->ioprio || sqe->off || sqe->len || sqe->buf_index ||
1151 sqe->poll_events)
1152 return -EINVAL;
1153
1154 spin_lock_irq(&ctx->completion_lock);
1155 list_for_each_entry_safe(poll_req, next, &ctx->cancel_list, list) {
1156 if (READ_ONCE(sqe->addr) == poll_req->user_data) {
1157 io_poll_remove_one(poll_req);
1158 ret = 0;
1159 break;
1160 }
1161 }
1162 spin_unlock_irq(&ctx->completion_lock);
1163
1164 io_cqring_add_event(req->ctx, sqe->user_data, ret, 0);
1165 io_put_req(req);
1166 return 0;
1167}
1168
1169static void io_poll_complete(struct io_ring_ctx *ctx, struct io_kiocb *req,
1170 __poll_t mask)
1171{
1172 req->poll.done = true;
1173 io_cqring_fill_event(ctx, req->user_data, mangle_poll(mask), 0);
1174 io_commit_cqring(ctx);
1175}
1176
1177static void io_poll_complete_work(struct work_struct *work)
1178{
1179 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1180 struct io_poll_iocb *poll = &req->poll;
1181 struct poll_table_struct pt = { ._key = poll->events };
1182 struct io_ring_ctx *ctx = req->ctx;
1183 __poll_t mask = 0;
1184
1185 if (!READ_ONCE(poll->canceled))
1186 mask = vfs_poll(poll->file, &pt) & poll->events;
1187
1188 /*
1189 * Note that ->ki_cancel callers also delete iocb from active_reqs after
1190 * calling ->ki_cancel. We need the ctx_lock roundtrip here to
1191 * synchronize with them. In the cancellation case the list_del_init
1192 * itself is not actually needed, but harmless so we keep it in to
1193 * avoid further branches in the fast path.
1194 */
1195 spin_lock_irq(&ctx->completion_lock);
1196 if (!mask && !READ_ONCE(poll->canceled)) {
1197 add_wait_queue(poll->head, &poll->wait);
1198 spin_unlock_irq(&ctx->completion_lock);
1199 return;
1200 }
1201 list_del_init(&req->list);
1202 io_poll_complete(ctx, req, mask);
1203 spin_unlock_irq(&ctx->completion_lock);
1204
1205 io_cqring_ev_posted(ctx);
1206 io_put_req(req);
1207}
1208
1209static int io_poll_wake(struct wait_queue_entry *wait, unsigned mode, int sync,
1210 void *key)
1211{
1212 struct io_poll_iocb *poll = container_of(wait, struct io_poll_iocb,
1213 wait);
1214 struct io_kiocb *req = container_of(poll, struct io_kiocb, poll);
1215 struct io_ring_ctx *ctx = req->ctx;
1216 __poll_t mask = key_to_poll(key);
1217 unsigned long flags;
1218
1219 /* for instances that support it check for an event match first: */
1220 if (mask && !(mask & poll->events))
1221 return 0;
1222
1223 list_del_init(&poll->wait.entry);
1224
1225 if (mask && spin_trylock_irqsave(&ctx->completion_lock, flags)) {
1226 list_del(&req->list);
1227 io_poll_complete(ctx, req, mask);
1228 spin_unlock_irqrestore(&ctx->completion_lock, flags);
1229
1230 io_cqring_ev_posted(ctx);
1231 io_put_req(req);
1232 } else {
1233 queue_work(ctx->sqo_wq, &req->work);
1234 }
1235
1236 return 1;
1237}
1238
1239struct io_poll_table {
1240 struct poll_table_struct pt;
1241 struct io_kiocb *req;
1242 int error;
1243};
1244
1245static void io_poll_queue_proc(struct file *file, struct wait_queue_head *head,
1246 struct poll_table_struct *p)
1247{
1248 struct io_poll_table *pt = container_of(p, struct io_poll_table, pt);
1249
1250 if (unlikely(pt->req->poll.head)) {
1251 pt->error = -EINVAL;
1252 return;
1253 }
1254
1255 pt->error = 0;
1256 pt->req->poll.head = head;
1257 add_wait_queue(head, &pt->req->poll.wait);
1258}
1259
1260static int io_poll_add(struct io_kiocb *req, const struct io_uring_sqe *sqe)
1261{
1262 struct io_poll_iocb *poll = &req->poll;
1263 struct io_ring_ctx *ctx = req->ctx;
1264 struct io_poll_table ipt;
1265 bool cancel = false;
1266 __poll_t mask;
1267 u16 events;
1268
1269 if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
1270 return -EINVAL;
1271 if (sqe->addr || sqe->ioprio || sqe->off || sqe->len || sqe->buf_index)
1272 return -EINVAL;
1273 if (!poll->file)
1274 return -EBADF;
1275
1276 INIT_WORK(&req->work, io_poll_complete_work);
1277 events = READ_ONCE(sqe->poll_events);
1278 poll->events = demangle_poll(events) | EPOLLERR | EPOLLHUP;
1279
1280 poll->head = NULL;
1281 poll->done = false;
1282 poll->canceled = false;
1283
1284 ipt.pt._qproc = io_poll_queue_proc;
1285 ipt.pt._key = poll->events;
1286 ipt.req = req;
1287 ipt.error = -EINVAL; /* same as no support for IOCB_CMD_POLL */
1288
1289 /* initialized the list so that we can do list_empty checks */
1290 INIT_LIST_HEAD(&poll->wait.entry);
1291 init_waitqueue_func_entry(&poll->wait, io_poll_wake);
1292
1293 mask = vfs_poll(poll->file, &ipt.pt) & poll->events;
1294
1295 spin_lock_irq(&ctx->completion_lock);
1296 if (likely(poll->head)) {
1297 spin_lock(&poll->head->lock);
1298 if (unlikely(list_empty(&poll->wait.entry))) {
1299 if (ipt.error)
1300 cancel = true;
1301 ipt.error = 0;
1302 mask = 0;
1303 }
1304 if (mask || ipt.error)
1305 list_del_init(&poll->wait.entry);
1306 else if (cancel)
1307 WRITE_ONCE(poll->canceled, true);
1308 else if (!poll->done) /* actually waiting for an event */
1309 list_add_tail(&req->list, &ctx->cancel_list);
1310 spin_unlock(&poll->head->lock);
1311 }
1312 if (mask) { /* no async, we'd stolen it */
1313 req->error = mangle_poll(mask);
1314 ipt.error = 0;
1315 io_poll_complete(ctx, req, mask);
1316 }
1317 spin_unlock_irq(&ctx->completion_lock);
1318
1319 if (mask) {
1320 io_cqring_ev_posted(ctx);
1321 io_put_req(req);
1322 }
1323 return ipt.error;
1324}
1325
1326static int __io_submit_sqe(struct io_ring_ctx *ctx, struct io_kiocb *req,
1327 const struct sqe_submit *s, bool force_nonblock,
1328 struct io_submit_state *state)
1329{
1330 int ret, opcode;
1331
1332 if (unlikely(s->index >= ctx->sq_entries))
1333 return -EINVAL;
1334 req->user_data = READ_ONCE(s->sqe->user_data);
1335
1336 opcode = READ_ONCE(s->sqe->opcode);
1337 switch (opcode) {
1338 case IORING_OP_NOP:
1339 ret = io_nop(req, req->user_data);
1340 break;
1341 case IORING_OP_READV:
1342 if (unlikely(s->sqe->buf_index))
1343 return -EINVAL;
1344 ret = io_read(req, s, force_nonblock, state);
1345 break;
1346 case IORING_OP_WRITEV:
1347 if (unlikely(s->sqe->buf_index))
1348 return -EINVAL;
1349 ret = io_write(req, s, force_nonblock, state);
1350 break;
1351 case IORING_OP_READ_FIXED:
1352 ret = io_read(req, s, force_nonblock, state);
1353 break;
1354 case IORING_OP_WRITE_FIXED:
1355 ret = io_write(req, s, force_nonblock, state);
1356 break;
1357 case IORING_OP_FSYNC:
1358 ret = io_fsync(req, s->sqe, force_nonblock);
1359 break;
1360 case IORING_OP_POLL_ADD:
1361 ret = io_poll_add(req, s->sqe);
1362 break;
1363 case IORING_OP_POLL_REMOVE:
1364 ret = io_poll_remove(req, s->sqe);
1365 break;
1366 default:
1367 ret = -EINVAL;
1368 break;
1369 }
1370
1371 if (ret)
1372 return ret;
1373
1374 if (ctx->flags & IORING_SETUP_IOPOLL) {
1375 if (req->error == -EAGAIN)
1376 return -EAGAIN;
1377
1378 /* workqueue context doesn't hold uring_lock, grab it now */
1379 if (s->needs_lock)
1380 mutex_lock(&ctx->uring_lock);
1381 io_iopoll_req_issued(req);
1382 if (s->needs_lock)
1383 mutex_unlock(&ctx->uring_lock);
1384 }
1385
1386 return 0;
1387}
1388
1389static struct async_list *io_async_list_from_sqe(struct io_ring_ctx *ctx,
1390 const struct io_uring_sqe *sqe)
1391{
1392 switch (sqe->opcode) {
1393 case IORING_OP_READV:
1394 case IORING_OP_READ_FIXED:
1395 return &ctx->pending_async[READ];
1396 case IORING_OP_WRITEV:
1397 case IORING_OP_WRITE_FIXED:
1398 return &ctx->pending_async[WRITE];
1399 default:
1400 return NULL;
1401 }
1402}
1403
1404static inline bool io_sqe_needs_user(const struct io_uring_sqe *sqe)
1405{
1406 u8 opcode = READ_ONCE(sqe->opcode);
1407
1408 return !(opcode == IORING_OP_READ_FIXED ||
1409 opcode == IORING_OP_WRITE_FIXED);
1410}
1411
1412static void io_sq_wq_submit_work(struct work_struct *work)
1413{
1414 struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1415 struct io_ring_ctx *ctx = req->ctx;
1416 struct mm_struct *cur_mm = NULL;
1417 struct async_list *async_list;
1418 LIST_HEAD(req_list);
1419 mm_segment_t old_fs;
1420 int ret;
1421
1422 async_list = io_async_list_from_sqe(ctx, req->submit.sqe);
1423restart:
1424 do {
1425 struct sqe_submit *s = &req->submit;
1426 const struct io_uring_sqe *sqe = s->sqe;
1427
1428 /* Ensure we clear previously set forced non-block flag */
1429 req->flags &= ~REQ_F_FORCE_NONBLOCK;
1430 req->rw.ki_flags &= ~IOCB_NOWAIT;
1431
1432 ret = 0;
1433 if (io_sqe_needs_user(sqe) && !cur_mm) {
1434 if (!mmget_not_zero(ctx->sqo_mm)) {
1435 ret = -EFAULT;
1436 } else {
1437 cur_mm = ctx->sqo_mm;
1438 use_mm(cur_mm);
1439 old_fs = get_fs();
1440 set_fs(USER_DS);
1441 }
1442 }
1443
1444 if (!ret) {
1445 s->has_user = cur_mm != NULL;
1446 s->needs_lock = true;
1447 do {
1448 ret = __io_submit_sqe(ctx, req, s, false, NULL);
1449 /*
1450 * We can get EAGAIN for polled IO even though
1451 * we're forcing a sync submission from here,
1452 * since we can't wait for request slots on the
1453 * block side.
1454 */
1455 if (ret != -EAGAIN)
1456 break;
1457 cond_resched();
1458 } while (1);
1459
1460 /* drop submission reference */
1461 io_put_req(req);
1462 }
1463 if (ret) {
1464 io_cqring_add_event(ctx, sqe->user_data, ret, 0);
1465 io_put_req(req);
1466 }
1467
1468 /* async context always use a copy of the sqe */
1469 kfree(sqe);
1470
1471 if (!async_list)
1472 break;
1473 if (!list_empty(&req_list)) {
1474 req = list_first_entry(&req_list, struct io_kiocb,
1475 list);
1476 list_del(&req->list);
1477 continue;
1478 }
1479 if (list_empty(&async_list->list))
1480 break;
1481
1482 req = NULL;
1483 spin_lock(&async_list->lock);
1484 if (list_empty(&async_list->list)) {
1485 spin_unlock(&async_list->lock);
1486 break;
1487 }
1488 list_splice_init(&async_list->list, &req_list);
1489 spin_unlock(&async_list->lock);
1490
1491 req = list_first_entry(&req_list, struct io_kiocb, list);
1492 list_del(&req->list);
1493 } while (req);
1494
1495 /*
1496 * Rare case of racing with a submitter. If we find the count has
1497 * dropped to zero AND we have pending work items, then restart
1498 * the processing. This is a tiny race window.
1499 */
1500 if (async_list) {
1501 ret = atomic_dec_return(&async_list->cnt);
1502 while (!ret && !list_empty(&async_list->list)) {
1503 spin_lock(&async_list->lock);
1504 atomic_inc(&async_list->cnt);
1505 list_splice_init(&async_list->list, &req_list);
1506 spin_unlock(&async_list->lock);
1507
1508 if (!list_empty(&req_list)) {
1509 req = list_first_entry(&req_list,
1510 struct io_kiocb, list);
1511 list_del(&req->list);
1512 goto restart;
1513 }
1514 ret = atomic_dec_return(&async_list->cnt);
1515 }
1516 }
1517
1518 if (cur_mm) {
1519 set_fs(old_fs);
1520 unuse_mm(cur_mm);
1521 mmput(cur_mm);
1522 }
1523}
1524
1525/*
1526 * See if we can piggy back onto previously submitted work, that is still
1527 * running. We currently only allow this if the new request is sequential
1528 * to the previous one we punted.
1529 */
1530static bool io_add_to_prev_work(struct async_list *list, struct io_kiocb *req)
1531{
1532 bool ret = false;
1533
1534 if (!list)
1535 return false;
1536 if (!(req->flags & REQ_F_SEQ_PREV))
1537 return false;
1538 if (!atomic_read(&list->cnt))
1539 return false;
1540
1541 ret = true;
1542 spin_lock(&list->lock);
1543 list_add_tail(&req->list, &list->list);
1544 if (!atomic_read(&list->cnt)) {
1545 list_del_init(&req->list);
1546 ret = false;
1547 }
1548 spin_unlock(&list->lock);
1549 return ret;
1550}
1551
1552static bool io_op_needs_file(const struct io_uring_sqe *sqe)
1553{
1554 int op = READ_ONCE(sqe->opcode);
1555
1556 switch (op) {
1557 case IORING_OP_NOP:
1558 case IORING_OP_POLL_REMOVE:
1559 return false;
1560 default:
1561 return true;
1562 }
1563}
1564
1565static int io_req_set_file(struct io_ring_ctx *ctx, const struct sqe_submit *s,
1566 struct io_submit_state *state, struct io_kiocb *req)
1567{
1568 unsigned flags;
1569 int fd;
1570
1571 flags = READ_ONCE(s->sqe->flags);
1572 fd = READ_ONCE(s->sqe->fd);
1573
1574 if (!io_op_needs_file(s->sqe)) {
1575 req->file = NULL;
1576 return 0;
1577 }
1578
1579 if (flags & IOSQE_FIXED_FILE) {
1580 if (unlikely(!ctx->user_files ||
1581 (unsigned) fd >= ctx->nr_user_files))
1582 return -EBADF;
1583 req->file = ctx->user_files[fd];
1584 req->flags |= REQ_F_FIXED_FILE;
1585 } else {
1586 if (s->needs_fixed_file)
1587 return -EBADF;
1588 req->file = io_file_get(state, fd);
1589 if (unlikely(!req->file))
1590 return -EBADF;
1591 }
1592
1593 return 0;
1594}
1595
1596static int io_submit_sqe(struct io_ring_ctx *ctx, struct sqe_submit *s,
1597 struct io_submit_state *state)
1598{
1599 struct io_kiocb *req;
1600 int ret;
1601
1602 /* enforce forwards compatibility on users */
1603 if (unlikely(s->sqe->flags & ~IOSQE_FIXED_FILE))
1604 return -EINVAL;
1605
1606 req = io_get_req(ctx, state);
1607 if (unlikely(!req))
1608 return -EAGAIN;
1609
1610 ret = io_req_set_file(ctx, s, state, req);
1611 if (unlikely(ret))
1612 goto out;
1613
1614 ret = __io_submit_sqe(ctx, req, s, true, state);
1615 if (ret == -EAGAIN) {
1616 struct io_uring_sqe *sqe_copy;
1617
1618 sqe_copy = kmalloc(sizeof(*sqe_copy), GFP_KERNEL);
1619 if (sqe_copy) {
1620 struct async_list *list;
1621
1622 memcpy(sqe_copy, s->sqe, sizeof(*sqe_copy));
1623 s->sqe = sqe_copy;
1624
1625 memcpy(&req->submit, s, sizeof(*s));
1626 list = io_async_list_from_sqe(ctx, s->sqe);
1627 if (!io_add_to_prev_work(list, req)) {
1628 if (list)
1629 atomic_inc(&list->cnt);
1630 INIT_WORK(&req->work, io_sq_wq_submit_work);
1631 queue_work(ctx->sqo_wq, &req->work);
1632 }
1633
1634 /*
1635 * Queued up for async execution, worker will release
1636 * submit reference when the iocb is actually
1637 * submitted.
1638 */
1639 return 0;
1640 }
1641 }
1642
1643out:
1644 /* drop submission reference */
1645 io_put_req(req);
1646
1647 /* and drop final reference, if we failed */
1648 if (ret)
1649 io_put_req(req);
1650
1651 return ret;
1652}
1653
1654/*
1655 * Batched submission is done, ensure local IO is flushed out.
1656 */
1657static void io_submit_state_end(struct io_submit_state *state)
1658{
1659 blk_finish_plug(&state->plug);
1660 io_file_put(state, NULL);
1661 if (state->free_reqs)
1662 kmem_cache_free_bulk(req_cachep, state->free_reqs,
1663 &state->reqs[state->cur_req]);
1664}
1665
1666/*
1667 * Start submission side cache.
1668 */
1669static void io_submit_state_start(struct io_submit_state *state,
1670 struct io_ring_ctx *ctx, unsigned max_ios)
1671{
1672 blk_start_plug(&state->plug);
1673 state->free_reqs = 0;
1674 state->file = NULL;
1675 state->ios_left = max_ios;
1676}
1677
1678static void io_commit_sqring(struct io_ring_ctx *ctx)
1679{
1680 struct io_sq_ring *ring = ctx->sq_ring;
1681
1682 if (ctx->cached_sq_head != READ_ONCE(ring->r.head)) {
1683 /*
1684 * Ensure any loads from the SQEs are done at this point,
1685 * since once we write the new head, the application could
1686 * write new data to them.
1687 */
1688 smp_store_release(&ring->r.head, ctx->cached_sq_head);
1689
1690 /*
1691 * write side barrier of head update, app has read side. See
1692 * comment at the top of this file
1693 */
1694 smp_wmb();
1695 }
1696}
1697
1698/*
1699 * Undo last io_get_sqring()
1700 */
1701static void io_drop_sqring(struct io_ring_ctx *ctx)
1702{
1703 ctx->cached_sq_head--;
1704}
1705
1706/*
1707 * Fetch an sqe, if one is available. Note that s->sqe will point to memory
1708 * that is mapped by userspace. This means that care needs to be taken to
1709 * ensure that reads are stable, as we cannot rely on userspace always
1710 * being a good citizen. If members of the sqe are validated and then later
1711 * used, it's important that those reads are done through READ_ONCE() to
1712 * prevent a re-load down the line.
1713 */
1714static bool io_get_sqring(struct io_ring_ctx *ctx, struct sqe_submit *s)
1715{
1716 struct io_sq_ring *ring = ctx->sq_ring;
1717 unsigned head;
1718
1719 /*
1720 * The cached sq head (or cq tail) serves two purposes:
1721 *
1722 * 1) allows us to batch the cost of updating the user visible
1723 * head updates.
1724 * 2) allows the kernel side to track the head on its own, even
1725 * though the application is the one updating it.
1726 */
1727 head = ctx->cached_sq_head;
1728 /* See comment at the top of this file */
1729 smp_rmb();
1730 if (head == READ_ONCE(ring->r.tail))
1731 return false;
1732
1733 head = READ_ONCE(ring->array[head & ctx->sq_mask]);
1734 if (head < ctx->sq_entries) {
1735 s->index = head;
1736 s->sqe = &ctx->sq_sqes[head];
1737 ctx->cached_sq_head++;
1738 return true;
1739 }
1740
1741 /* drop invalid entries */
1742 ctx->cached_sq_head++;
1743 ring->dropped++;
1744 /* See comment at the top of this file */
1745 smp_wmb();
1746 return false;
1747}
1748
1749static int io_submit_sqes(struct io_ring_ctx *ctx, struct sqe_submit *sqes,
1750 unsigned int nr, bool has_user, bool mm_fault)
1751{
1752 struct io_submit_state state, *statep = NULL;
1753 int ret, i, submitted = 0;
1754
1755 if (nr > IO_PLUG_THRESHOLD) {
1756 io_submit_state_start(&state, ctx, nr);
1757 statep = &state;
1758 }
1759
1760 for (i = 0; i < nr; i++) {
1761 if (unlikely(mm_fault)) {
1762 ret = -EFAULT;
1763 } else {
1764 sqes[i].has_user = has_user;
1765 sqes[i].needs_lock = true;
1766 sqes[i].needs_fixed_file = true;
1767 ret = io_submit_sqe(ctx, &sqes[i], statep);
1768 }
1769 if (!ret) {
1770 submitted++;
1771 continue;
1772 }
1773
1774 io_cqring_add_event(ctx, sqes[i].sqe->user_data, ret, 0);
1775 }
1776
1777 if (statep)
1778 io_submit_state_end(&state);
1779
1780 return submitted;
1781}
1782
1783static int io_sq_thread(void *data)
1784{
1785 struct sqe_submit sqes[IO_IOPOLL_BATCH];
1786 struct io_ring_ctx *ctx = data;
1787 struct mm_struct *cur_mm = NULL;
1788 mm_segment_t old_fs;
1789 DEFINE_WAIT(wait);
1790 unsigned inflight;
1791 unsigned long timeout;
1792
1793 old_fs = get_fs();
1794 set_fs(USER_DS);
1795
1796 timeout = inflight = 0;
1797 while (!kthread_should_stop() && !ctx->sqo_stop) {
1798 bool all_fixed, mm_fault = false;
1799 int i;
1800
1801 if (inflight) {
1802 unsigned nr_events = 0;
1803
1804 if (ctx->flags & IORING_SETUP_IOPOLL) {
1805 /*
1806 * We disallow the app entering submit/complete
1807 * with polling, but we still need to lock the
1808 * ring to prevent racing with polled issue
1809 * that got punted to a workqueue.
1810 */
1811 mutex_lock(&ctx->uring_lock);
1812 io_iopoll_check(ctx, &nr_events, 0);
1813 mutex_unlock(&ctx->uring_lock);
1814 } else {
1815 /*
1816 * Normal IO, just pretend everything completed.
1817 * We don't have to poll completions for that.
1818 */
1819 nr_events = inflight;
1820 }
1821
1822 inflight -= nr_events;
1823 if (!inflight)
1824 timeout = jiffies + ctx->sq_thread_idle;
1825 }
1826
1827 if (!io_get_sqring(ctx, &sqes[0])) {
1828 /*
1829 * We're polling. If we're within the defined idle
1830 * period, then let us spin without work before going
1831 * to sleep.
1832 */
1833 if (inflight || !time_after(jiffies, timeout)) {
1834 cpu_relax();
1835 continue;
1836 }
1837
1838 /*
1839 * Drop cur_mm before scheduling, we can't hold it for
1840 * long periods (or over schedule()). Do this before
1841 * adding ourselves to the waitqueue, as the unuse/drop
1842 * may sleep.
1843 */
1844 if (cur_mm) {
1845 unuse_mm(cur_mm);
1846 mmput(cur_mm);
1847 cur_mm = NULL;
1848 }
1849
1850 prepare_to_wait(&ctx->sqo_wait, &wait,
1851 TASK_INTERRUPTIBLE);
1852
1853 /* Tell userspace we may need a wakeup call */
1854 ctx->sq_ring->flags |= IORING_SQ_NEED_WAKEUP;
1855 smp_wmb();
1856
1857 if (!io_get_sqring(ctx, &sqes[0])) {
1858 if (kthread_should_stop()) {
1859 finish_wait(&ctx->sqo_wait, &wait);
1860 break;
1861 }
1862 if (signal_pending(current))
1863 flush_signals(current);
1864 schedule();
1865 finish_wait(&ctx->sqo_wait, &wait);
1866
1867 ctx->sq_ring->flags &= ~IORING_SQ_NEED_WAKEUP;
1868 smp_wmb();
1869 continue;
1870 }
1871 finish_wait(&ctx->sqo_wait, &wait);
1872
1873 ctx->sq_ring->flags &= ~IORING_SQ_NEED_WAKEUP;
1874 smp_wmb();
1875 }
1876
1877 i = 0;
1878 all_fixed = true;
1879 do {
1880 if (all_fixed && io_sqe_needs_user(sqes[i].sqe))
1881 all_fixed = false;
1882
1883 i++;
1884 if (i == ARRAY_SIZE(sqes))
1885 break;
1886 } while (io_get_sqring(ctx, &sqes[i]));
1887
1888 /* Unless all new commands are FIXED regions, grab mm */
1889 if (!all_fixed && !cur_mm) {
1890 mm_fault = !mmget_not_zero(ctx->sqo_mm);
1891 if (!mm_fault) {
1892 use_mm(ctx->sqo_mm);
1893 cur_mm = ctx->sqo_mm;
1894 }
1895 }
1896
1897 inflight += io_submit_sqes(ctx, sqes, i, cur_mm != NULL,
1898 mm_fault);
1899
1900 /* Commit SQ ring head once we've consumed all SQEs */
1901 io_commit_sqring(ctx);
1902 }
1903
1904 set_fs(old_fs);
1905 if (cur_mm) {
1906 unuse_mm(cur_mm);
1907 mmput(cur_mm);
1908 }
1909 return 0;
1910}
1911
1912static int io_ring_submit(struct io_ring_ctx *ctx, unsigned int to_submit)
1913{
1914 struct io_submit_state state, *statep = NULL;
1915 int i, ret = 0, submit = 0;
1916
1917 if (to_submit > IO_PLUG_THRESHOLD) {
1918 io_submit_state_start(&state, ctx, to_submit);
1919 statep = &state;
1920 }
1921
1922 for (i = 0; i < to_submit; i++) {
1923 struct sqe_submit s;
1924
1925 if (!io_get_sqring(ctx, &s))
1926 break;
1927
1928 s.has_user = true;
1929 s.needs_lock = false;
1930 s.needs_fixed_file = false;
1931
1932 ret = io_submit_sqe(ctx, &s, statep);
1933 if (ret) {
1934 io_drop_sqring(ctx);
1935 break;
1936 }
1937
1938 submit++;
1939 }
1940 io_commit_sqring(ctx);
1941
1942 if (statep)
1943 io_submit_state_end(statep);
1944
1945 return submit ? submit : ret;
1946}
1947
1948static unsigned io_cqring_events(struct io_cq_ring *ring)
1949{
1950 return READ_ONCE(ring->r.tail) - READ_ONCE(ring->r.head);
1951}
1952
1953/*
1954 * Wait until events become available, if we don't already have some. The
1955 * application must reap them itself, as they reside on the shared cq ring.
1956 */
1957static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
1958 const sigset_t __user *sig, size_t sigsz)
1959{
1960 struct io_cq_ring *ring = ctx->cq_ring;
1961 sigset_t ksigmask, sigsaved;
1962 DEFINE_WAIT(wait);
1963 int ret;
1964
1965 /* See comment at the top of this file */
1966 smp_rmb();
1967 if (io_cqring_events(ring) >= min_events)
1968 return 0;
1969
1970 if (sig) {
1971 ret = set_user_sigmask(sig, &ksigmask, &sigsaved, sigsz);
1972 if (ret)
1973 return ret;
1974 }
1975
1976 do {
1977 prepare_to_wait(&ctx->wait, &wait, TASK_INTERRUPTIBLE);
1978
1979 ret = 0;
1980 /* See comment at the top of this file */
1981 smp_rmb();
1982 if (io_cqring_events(ring) >= min_events)
1983 break;
1984
1985 schedule();
1986
1987 ret = -EINTR;
1988 if (signal_pending(current))
1989 break;
1990 } while (1);
1991
1992 finish_wait(&ctx->wait, &wait);
1993
1994 if (sig)
1995 restore_user_sigmask(sig, &sigsaved);
1996
1997 return READ_ONCE(ring->r.head) == READ_ONCE(ring->r.tail) ? ret : 0;
1998}
1999
2000static void __io_sqe_files_unregister(struct io_ring_ctx *ctx)
2001{
2002#if defined(CONFIG_UNIX)
2003 if (ctx->ring_sock) {
2004 struct sock *sock = ctx->ring_sock->sk;
2005 struct sk_buff *skb;
2006
2007 while ((skb = skb_dequeue(&sock->sk_receive_queue)) != NULL)
2008 kfree_skb(skb);
2009 }
2010#else
2011 int i;
2012
2013 for (i = 0; i < ctx->nr_user_files; i++)
2014 fput(ctx->user_files[i]);
2015#endif
2016}
2017
2018static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
2019{
2020 if (!ctx->user_files)
2021 return -ENXIO;
2022
2023 __io_sqe_files_unregister(ctx);
2024 kfree(ctx->user_files);
2025 ctx->user_files = NULL;
2026 ctx->nr_user_files = 0;
2027 return 0;
2028}
2029
2030static void io_sq_thread_stop(struct io_ring_ctx *ctx)
2031{
2032 if (ctx->sqo_thread) {
2033 ctx->sqo_stop = 1;
2034 mb();
2035 kthread_stop(ctx->sqo_thread);
2036 ctx->sqo_thread = NULL;
2037 }
2038}
2039
2040static void io_finish_async(struct io_ring_ctx *ctx)
2041{
2042 io_sq_thread_stop(ctx);
2043
2044 if (ctx->sqo_wq) {
2045 destroy_workqueue(ctx->sqo_wq);
2046 ctx->sqo_wq = NULL;
2047 }
2048}
2049
2050#if defined(CONFIG_UNIX)
2051static void io_destruct_skb(struct sk_buff *skb)
2052{
2053 struct io_ring_ctx *ctx = skb->sk->sk_user_data;
2054
2055 io_finish_async(ctx);
2056 unix_destruct_scm(skb);
2057}
2058
2059/*
2060 * Ensure the UNIX gc is aware of our file set, so we are certain that
2061 * the io_uring can be safely unregistered on process exit, even if we have
2062 * loops in the file referencing.
2063 */
2064static int __io_sqe_files_scm(struct io_ring_ctx *ctx, int nr, int offset)
2065{
2066 struct sock *sk = ctx->ring_sock->sk;
2067 struct scm_fp_list *fpl;
2068 struct sk_buff *skb;
2069 int i;
2070
2071 if (!capable(CAP_SYS_RESOURCE) && !capable(CAP_SYS_ADMIN)) {
2072 unsigned long inflight = ctx->user->unix_inflight + nr;
2073
2074 if (inflight > task_rlimit(current, RLIMIT_NOFILE))
2075 return -EMFILE;
2076 }
2077
2078 fpl = kzalloc(sizeof(*fpl), GFP_KERNEL);
2079 if (!fpl)
2080 return -ENOMEM;
2081
2082 skb = alloc_skb(0, GFP_KERNEL);
2083 if (!skb) {
2084 kfree(fpl);
2085 return -ENOMEM;
2086 }
2087
2088 skb->sk = sk;
2089 skb->destructor = io_destruct_skb;
2090
2091 fpl->user = get_uid(ctx->user);
2092 for (i = 0; i < nr; i++) {
2093 fpl->fp[i] = get_file(ctx->user_files[i + offset]);
2094 unix_inflight(fpl->user, fpl->fp[i]);
2095 }
2096
2097 fpl->max = fpl->count = nr;
2098 UNIXCB(skb).fp = fpl;
2099 refcount_add(skb->truesize, &sk->sk_wmem_alloc);
2100 skb_queue_head(&sk->sk_receive_queue, skb);
2101
2102 for (i = 0; i < nr; i++)
2103 fput(fpl->fp[i]);
2104
2105 return 0;
2106}
2107
2108/*
2109 * If UNIX sockets are enabled, fd passing can cause a reference cycle which
2110 * causes regular reference counting to break down. We rely on the UNIX
2111 * garbage collection to take care of this problem for us.
2112 */
2113static int io_sqe_files_scm(struct io_ring_ctx *ctx)
2114{
2115 unsigned left, total;
2116 int ret = 0;
2117
2118 total = 0;
2119 left = ctx->nr_user_files;
2120 while (left) {
2121 unsigned this_files = min_t(unsigned, left, SCM_MAX_FD);
2122 int ret;
2123
2124 ret = __io_sqe_files_scm(ctx, this_files, total);
2125 if (ret)
2126 break;
2127 left -= this_files;
2128 total += this_files;
2129 }
2130
2131 if (!ret)
2132 return 0;
2133
2134 while (total < ctx->nr_user_files) {
2135 fput(ctx->user_files[total]);
2136 total++;
2137 }
2138
2139 return ret;
2140}
2141#else
2142static int io_sqe_files_scm(struct io_ring_ctx *ctx)
2143{
2144 return 0;
2145}
2146#endif
2147
2148static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
2149 unsigned nr_args)
2150{
2151 __s32 __user *fds = (__s32 __user *) arg;
2152 int fd, ret = 0;
2153 unsigned i;
2154
2155 if (ctx->user_files)
2156 return -EBUSY;
2157 if (!nr_args)
2158 return -EINVAL;
2159 if (nr_args > IORING_MAX_FIXED_FILES)
2160 return -EMFILE;
2161
2162 ctx->user_files = kcalloc(nr_args, sizeof(struct file *), GFP_KERNEL);
2163 if (!ctx->user_files)
2164 return -ENOMEM;
2165
2166 for (i = 0; i < nr_args; i++) {
2167 ret = -EFAULT;
2168 if (copy_from_user(&fd, &fds[i], sizeof(fd)))
2169 break;
2170
2171 ctx->user_files[i] = fget(fd);
2172
2173 ret = -EBADF;
2174 if (!ctx->user_files[i])
2175 break;
2176 /*
2177 * Don't allow io_uring instances to be registered. If UNIX
2178 * isn't enabled, then this causes a reference cycle and this
2179 * instance can never get freed. If UNIX is enabled we'll
2180 * handle it just fine, but there's still no point in allowing
2181 * a ring fd as it doesn't support regular read/write anyway.
2182 */
2183 if (ctx->user_files[i]->f_op == &io_uring_fops) {
2184 fput(ctx->user_files[i]);
2185 break;
2186 }
2187 ctx->nr_user_files++;
2188 ret = 0;
2189 }
2190
2191 if (ret) {
2192 for (i = 0; i < ctx->nr_user_files; i++)
2193 fput(ctx->user_files[i]);
2194
2195 kfree(ctx->user_files);
2196 ctx->nr_user_files = 0;
2197 return ret;
2198 }
2199
2200 ret = io_sqe_files_scm(ctx);
2201 if (ret)
2202 io_sqe_files_unregister(ctx);
2203
2204 return ret;
2205}
2206
2207static int io_sq_offload_start(struct io_ring_ctx *ctx,
2208 struct io_uring_params *p)
2209{
2210 int ret;
2211
2212 init_waitqueue_head(&ctx->sqo_wait);
2213 mmgrab(current->mm);
2214 ctx->sqo_mm = current->mm;
2215
2216 ctx->sq_thread_idle = msecs_to_jiffies(p->sq_thread_idle);
2217 if (!ctx->sq_thread_idle)
2218 ctx->sq_thread_idle = HZ;
2219
2220 ret = -EINVAL;
2221 if (!cpu_possible(p->sq_thread_cpu))
2222 goto err;
2223
2224 if (ctx->flags & IORING_SETUP_SQPOLL) {
2225 if (p->flags & IORING_SETUP_SQ_AFF) {
2226 int cpu;
2227
2228 cpu = array_index_nospec(p->sq_thread_cpu, NR_CPUS);
2229 ctx->sqo_thread = kthread_create_on_cpu(io_sq_thread,
2230 ctx, cpu,
2231 "io_uring-sq");
2232 } else {
2233 ctx->sqo_thread = kthread_create(io_sq_thread, ctx,
2234 "io_uring-sq");
2235 }
2236 if (IS_ERR(ctx->sqo_thread)) {
2237 ret = PTR_ERR(ctx->sqo_thread);
2238 ctx->sqo_thread = NULL;
2239 goto err;
2240 }
2241 wake_up_process(ctx->sqo_thread);
2242 } else if (p->flags & IORING_SETUP_SQ_AFF) {
2243 /* Can't have SQ_AFF without SQPOLL */
2244 ret = -EINVAL;
2245 goto err;
2246 }
2247
2248 /* Do QD, or 2 * CPUS, whatever is smallest */
2249 ctx->sqo_wq = alloc_workqueue("io_ring-wq", WQ_UNBOUND | WQ_FREEZABLE,
2250 min(ctx->sq_entries - 1, 2 * num_online_cpus()));
2251 if (!ctx->sqo_wq) {
2252 ret = -ENOMEM;
2253 goto err;
2254 }
2255
2256 return 0;
2257err:
2258 io_sq_thread_stop(ctx);
2259 mmdrop(ctx->sqo_mm);
2260 ctx->sqo_mm = NULL;
2261 return ret;
2262}
2263
2264static void io_unaccount_mem(struct user_struct *user, unsigned long nr_pages)
2265{
2266 atomic_long_sub(nr_pages, &user->locked_vm);
2267}
2268
2269static int io_account_mem(struct user_struct *user, unsigned long nr_pages)
2270{
2271 unsigned long page_limit, cur_pages, new_pages;
2272
2273 /* Don't allow more pages than we can safely lock */
2274 page_limit = rlimit(RLIMIT_MEMLOCK) >> PAGE_SHIFT;
2275
2276 do {
2277 cur_pages = atomic_long_read(&user->locked_vm);
2278 new_pages = cur_pages + nr_pages;
2279 if (new_pages > page_limit)
2280 return -ENOMEM;
2281 } while (atomic_long_cmpxchg(&user->locked_vm, cur_pages,
2282 new_pages) != cur_pages);
2283
2284 return 0;
2285}
2286
2287static void io_mem_free(void *ptr)
2288{
2289 struct page *page = virt_to_head_page(ptr);
2290
2291 if (put_page_testzero(page))
2292 free_compound_page(page);
2293}
2294
2295static void *io_mem_alloc(size_t size)
2296{
2297 gfp_t gfp_flags = GFP_KERNEL | __GFP_ZERO | __GFP_NOWARN | __GFP_COMP |
2298 __GFP_NORETRY;
2299
2300 return (void *) __get_free_pages(gfp_flags, get_order(size));
2301}
2302
2303static unsigned long ring_pages(unsigned sq_entries, unsigned cq_entries)
2304{
2305 struct io_sq_ring *sq_ring;
2306 struct io_cq_ring *cq_ring;
2307 size_t bytes;
2308
2309 bytes = struct_size(sq_ring, array, sq_entries);
2310 bytes += array_size(sizeof(struct io_uring_sqe), sq_entries);
2311 bytes += struct_size(cq_ring, cqes, cq_entries);
2312
2313 return (bytes + PAGE_SIZE - 1) / PAGE_SIZE;
2314}
2315
2316static int io_sqe_buffer_unregister(struct io_ring_ctx *ctx)
2317{
2318 int i, j;
2319
2320 if (!ctx->user_bufs)
2321 return -ENXIO;
2322
2323 for (i = 0; i < ctx->nr_user_bufs; i++) {
2324 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
2325
2326 for (j = 0; j < imu->nr_bvecs; j++)
2327 put_page(imu->bvec[j].bv_page);
2328
2329 if (ctx->account_mem)
2330 io_unaccount_mem(ctx->user, imu->nr_bvecs);
2331 kfree(imu->bvec);
2332 imu->nr_bvecs = 0;
2333 }
2334
2335 kfree(ctx->user_bufs);
2336 ctx->user_bufs = NULL;
2337 ctx->nr_user_bufs = 0;
2338 return 0;
2339}
2340
2341static int io_copy_iov(struct io_ring_ctx *ctx, struct iovec *dst,
2342 void __user *arg, unsigned index)
2343{
2344 struct iovec __user *src;
2345
2346#ifdef CONFIG_COMPAT
2347 if (ctx->compat) {
2348 struct compat_iovec __user *ciovs;
2349 struct compat_iovec ciov;
2350
2351 ciovs = (struct compat_iovec __user *) arg;
2352 if (copy_from_user(&ciov, &ciovs[index], sizeof(ciov)))
2353 return -EFAULT;
2354
2355 dst->iov_base = (void __user *) (unsigned long) ciov.iov_base;
2356 dst->iov_len = ciov.iov_len;
2357 return 0;
2358 }
2359#endif
2360 src = (struct iovec __user *) arg;
2361 if (copy_from_user(dst, &src[index], sizeof(*dst)))
2362 return -EFAULT;
2363 return 0;
2364}
2365
2366static int io_sqe_buffer_register(struct io_ring_ctx *ctx, void __user *arg,
2367 unsigned nr_args)
2368{
2369 struct vm_area_struct **vmas = NULL;
2370 struct page **pages = NULL;
2371 int i, j, got_pages = 0;
2372 int ret = -EINVAL;
2373
2374 if (ctx->user_bufs)
2375 return -EBUSY;
2376 if (!nr_args || nr_args > UIO_MAXIOV)
2377 return -EINVAL;
2378
2379 ctx->user_bufs = kcalloc(nr_args, sizeof(struct io_mapped_ubuf),
2380 GFP_KERNEL);
2381 if (!ctx->user_bufs)
2382 return -ENOMEM;
2383
2384 for (i = 0; i < nr_args; i++) {
2385 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
2386 unsigned long off, start, end, ubuf;
2387 int pret, nr_pages;
2388 struct iovec iov;
2389 size_t size;
2390
2391 ret = io_copy_iov(ctx, &iov, arg, i);
2392 if (ret)
2393 break;
2394
2395 /*
2396 * Don't impose further limits on the size and buffer
2397 * constraints here, we'll -EINVAL later when IO is
2398 * submitted if they are wrong.
2399 */
2400 ret = -EFAULT;
2401 if (!iov.iov_base || !iov.iov_len)
2402 goto err;
2403
2404 /* arbitrary limit, but we need something */
2405 if (iov.iov_len > SZ_1G)
2406 goto err;
2407
2408 ubuf = (unsigned long) iov.iov_base;
2409 end = (ubuf + iov.iov_len + PAGE_SIZE - 1) >> PAGE_SHIFT;
2410 start = ubuf >> PAGE_SHIFT;
2411 nr_pages = end - start;
2412
2413 if (ctx->account_mem) {
2414 ret = io_account_mem(ctx->user, nr_pages);
2415 if (ret)
2416 goto err;
2417 }
2418
2419 ret = 0;
2420 if (!pages || nr_pages > got_pages) {
2421 kfree(vmas);
2422 kfree(pages);
2423 pages = kmalloc_array(nr_pages, sizeof(struct page *),
2424 GFP_KERNEL);
2425 vmas = kmalloc_array(nr_pages,
2426 sizeof(struct vm_area_struct *),
2427 GFP_KERNEL);
2428 if (!pages || !vmas) {
2429 ret = -ENOMEM;
2430 if (ctx->account_mem)
2431 io_unaccount_mem(ctx->user, nr_pages);
2432 goto err;
2433 }
2434 got_pages = nr_pages;
2435 }
2436
2437 imu->bvec = kmalloc_array(nr_pages, sizeof(struct bio_vec),
2438 GFP_KERNEL);
2439 ret = -ENOMEM;
2440 if (!imu->bvec) {
2441 if (ctx->account_mem)
2442 io_unaccount_mem(ctx->user, nr_pages);
2443 goto err;
2444 }
2445
2446 ret = 0;
2447 down_read(&current->mm->mmap_sem);
2448 pret = get_user_pages_longterm(ubuf, nr_pages, FOLL_WRITE,
2449 pages, vmas);
2450 if (pret == nr_pages) {
2451 /* don't support file backed memory */
2452 for (j = 0; j < nr_pages; j++) {
2453 struct vm_area_struct *vma = vmas[j];
2454
2455 if (vma->vm_file &&
2456 !is_file_hugepages(vma->vm_file)) {
2457 ret = -EOPNOTSUPP;
2458 break;
2459 }
2460 }
2461 } else {
2462 ret = pret < 0 ? pret : -EFAULT;
2463 }
2464 up_read(&current->mm->mmap_sem);
2465 if (ret) {
2466 /*
2467 * if we did partial map, or found file backed vmas,
2468 * release any pages we did get
2469 */
2470 if (pret > 0) {
2471 for (j = 0; j < pret; j++)
2472 put_page(pages[j]);
2473 }
2474 if (ctx->account_mem)
2475 io_unaccount_mem(ctx->user, nr_pages);
2476 goto err;
2477 }
2478
2479 off = ubuf & ~PAGE_MASK;
2480 size = iov.iov_len;
2481 for (j = 0; j < nr_pages; j++) {
2482 size_t vec_len;
2483
2484 vec_len = min_t(size_t, size, PAGE_SIZE - off);
2485 imu->bvec[j].bv_page = pages[j];
2486 imu->bvec[j].bv_len = vec_len;
2487 imu->bvec[j].bv_offset = off;
2488 off = 0;
2489 size -= vec_len;
2490 }
2491 /* store original address for later verification */
2492 imu->ubuf = ubuf;
2493 imu->len = iov.iov_len;
2494 imu->nr_bvecs = nr_pages;
2495
2496 ctx->nr_user_bufs++;
2497 }
2498 kfree(pages);
2499 kfree(vmas);
2500 return 0;
2501err:
2502 kfree(pages);
2503 kfree(vmas);
2504 io_sqe_buffer_unregister(ctx);
2505 return ret;
2506}
2507
2508static void io_ring_ctx_free(struct io_ring_ctx *ctx)
2509{
2510 io_finish_async(ctx);
2511 if (ctx->sqo_mm)
2512 mmdrop(ctx->sqo_mm);
2513
2514 io_iopoll_reap_events(ctx);
2515 io_sqe_buffer_unregister(ctx);
2516 io_sqe_files_unregister(ctx);
2517
2518#if defined(CONFIG_UNIX)
2519 if (ctx->ring_sock)
2520 sock_release(ctx->ring_sock);
2521#endif
2522
2523 io_mem_free(ctx->sq_ring);
2524 io_mem_free(ctx->sq_sqes);
2525 io_mem_free(ctx->cq_ring);
2526
2527 percpu_ref_exit(&ctx->refs);
2528 if (ctx->account_mem)
2529 io_unaccount_mem(ctx->user,
2530 ring_pages(ctx->sq_entries, ctx->cq_entries));
2531 free_uid(ctx->user);
2532 kfree(ctx);
2533}
2534
2535static __poll_t io_uring_poll(struct file *file, poll_table *wait)
2536{
2537 struct io_ring_ctx *ctx = file->private_data;
2538 __poll_t mask = 0;
2539
2540 poll_wait(file, &ctx->cq_wait, wait);
2541 /* See comment at the top of this file */
2542 smp_rmb();
2543 if (READ_ONCE(ctx->sq_ring->r.tail) + 1 != ctx->cached_sq_head)
2544 mask |= EPOLLOUT | EPOLLWRNORM;
2545 if (READ_ONCE(ctx->cq_ring->r.head) != ctx->cached_cq_tail)
2546 mask |= EPOLLIN | EPOLLRDNORM;
2547
2548 return mask;
2549}
2550
2551static int io_uring_fasync(int fd, struct file *file, int on)
2552{
2553 struct io_ring_ctx *ctx = file->private_data;
2554
2555 return fasync_helper(fd, file, on, &ctx->cq_fasync);
2556}
2557
2558static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
2559{
2560 mutex_lock(&ctx->uring_lock);
2561 percpu_ref_kill(&ctx->refs);
2562 mutex_unlock(&ctx->uring_lock);
2563
2564 io_poll_remove_all(ctx);
2565 io_iopoll_reap_events(ctx);
2566 wait_for_completion(&ctx->ctx_done);
2567 io_ring_ctx_free(ctx);
2568}
2569
2570static int io_uring_release(struct inode *inode, struct file *file)
2571{
2572 struct io_ring_ctx *ctx = file->private_data;
2573
2574 file->private_data = NULL;
2575 io_ring_ctx_wait_and_kill(ctx);
2576 return 0;
2577}
2578
2579static int io_uring_mmap(struct file *file, struct vm_area_struct *vma)
2580{
2581 loff_t offset = (loff_t) vma->vm_pgoff << PAGE_SHIFT;
2582 unsigned long sz = vma->vm_end - vma->vm_start;
2583 struct io_ring_ctx *ctx = file->private_data;
2584 unsigned long pfn;
2585 struct page *page;
2586 void *ptr;
2587
2588 switch (offset) {
2589 case IORING_OFF_SQ_RING:
2590 ptr = ctx->sq_ring;
2591 break;
2592 case IORING_OFF_SQES:
2593 ptr = ctx->sq_sqes;
2594 break;
2595 case IORING_OFF_CQ_RING:
2596 ptr = ctx->cq_ring;
2597 break;
2598 default:
2599 return -EINVAL;
2600 }
2601
2602 page = virt_to_head_page(ptr);
2603 if (sz > (PAGE_SIZE << compound_order(page)))
2604 return -EINVAL;
2605
2606 pfn = virt_to_phys(ptr) >> PAGE_SHIFT;
2607 return remap_pfn_range(vma, vma->vm_start, pfn, sz, vma->vm_page_prot);
2608}
2609
2610SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
2611 u32, min_complete, u32, flags, const sigset_t __user *, sig,
2612 size_t, sigsz)
2613{
2614 struct io_ring_ctx *ctx;
2615 long ret = -EBADF;
2616 int submitted = 0;
2617 struct fd f;
2618
2619 if (flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP))
2620 return -EINVAL;
2621
2622 f = fdget(fd);
2623 if (!f.file)
2624 return -EBADF;
2625
2626 ret = -EOPNOTSUPP;
2627 if (f.file->f_op != &io_uring_fops)
2628 goto out_fput;
2629
2630 ret = -ENXIO;
2631 ctx = f.file->private_data;
2632 if (!percpu_ref_tryget(&ctx->refs))
2633 goto out_fput;
2634
2635 /*
2636 * For SQ polling, the thread will do all submissions and completions.
2637 * Just return the requested submit count, and wake the thread if
2638 * we were asked to.
2639 */
2640 if (ctx->flags & IORING_SETUP_SQPOLL) {
2641 if (flags & IORING_ENTER_SQ_WAKEUP)
2642 wake_up(&ctx->sqo_wait);
2643 submitted = to_submit;
2644 goto out_ctx;
2645 }
2646
2647 ret = 0;
2648 if (to_submit) {
2649 to_submit = min(to_submit, ctx->sq_entries);
2650
2651 mutex_lock(&ctx->uring_lock);
2652 submitted = io_ring_submit(ctx, to_submit);
2653 mutex_unlock(&ctx->uring_lock);
2654
2655 if (submitted < 0)
2656 goto out_ctx;
2657 }
2658 if (flags & IORING_ENTER_GETEVENTS) {
2659 unsigned nr_events = 0;
2660
2661 min_complete = min(min_complete, ctx->cq_entries);
2662
2663 /*
2664 * The application could have included the 'to_submit' count
2665 * in how many events it wanted to wait for. If we failed to
2666 * submit the desired count, we may need to adjust the number
2667 * of events to poll/wait for.
2668 */
2669 if (submitted < to_submit)
2670 min_complete = min_t(unsigned, submitted, min_complete);
2671
2672 if (ctx->flags & IORING_SETUP_IOPOLL) {
2673 mutex_lock(&ctx->uring_lock);
2674 ret = io_iopoll_check(ctx, &nr_events, min_complete);
2675 mutex_unlock(&ctx->uring_lock);
2676 } else {
2677 ret = io_cqring_wait(ctx, min_complete, sig, sigsz);
2678 }
2679 }
2680
2681out_ctx:
2682 io_ring_drop_ctx_refs(ctx, 1);
2683out_fput:
2684 fdput(f);
2685 return submitted ? submitted : ret;
2686}
2687
2688static const struct file_operations io_uring_fops = {
2689 .release = io_uring_release,
2690 .mmap = io_uring_mmap,
2691 .poll = io_uring_poll,
2692 .fasync = io_uring_fasync,
2693};
2694
2695static int io_allocate_scq_urings(struct io_ring_ctx *ctx,
2696 struct io_uring_params *p)
2697{
2698 struct io_sq_ring *sq_ring;
2699 struct io_cq_ring *cq_ring;
2700 size_t size;
2701
2702 sq_ring = io_mem_alloc(struct_size(sq_ring, array, p->sq_entries));
2703 if (!sq_ring)
2704 return -ENOMEM;
2705
2706 ctx->sq_ring = sq_ring;
2707 sq_ring->ring_mask = p->sq_entries - 1;
2708 sq_ring->ring_entries = p->sq_entries;
2709 ctx->sq_mask = sq_ring->ring_mask;
2710 ctx->sq_entries = sq_ring->ring_entries;
2711
2712 size = array_size(sizeof(struct io_uring_sqe), p->sq_entries);
2713 if (size == SIZE_MAX)
2714 return -EOVERFLOW;
2715
2716 ctx->sq_sqes = io_mem_alloc(size);
2717 if (!ctx->sq_sqes) {
2718 io_mem_free(ctx->sq_ring);
2719 return -ENOMEM;
2720 }
2721
2722 cq_ring = io_mem_alloc(struct_size(cq_ring, cqes, p->cq_entries));
2723 if (!cq_ring) {
2724 io_mem_free(ctx->sq_ring);
2725 io_mem_free(ctx->sq_sqes);
2726 return -ENOMEM;
2727 }
2728
2729 ctx->cq_ring = cq_ring;
2730 cq_ring->ring_mask = p->cq_entries - 1;
2731 cq_ring->ring_entries = p->cq_entries;
2732 ctx->cq_mask = cq_ring->ring_mask;
2733 ctx->cq_entries = cq_ring->ring_entries;
2734 return 0;
2735}
2736
2737/*
2738 * Allocate an anonymous fd, this is what constitutes the application
2739 * visible backing of an io_uring instance. The application mmaps this
2740 * fd to gain access to the SQ/CQ ring details. If UNIX sockets are enabled,
2741 * we have to tie this fd to a socket for file garbage collection purposes.
2742 */
2743static int io_uring_get_fd(struct io_ring_ctx *ctx)
2744{
2745 struct file *file;
2746 int ret;
2747
2748#if defined(CONFIG_UNIX)
2749 ret = sock_create_kern(&init_net, PF_UNIX, SOCK_RAW, IPPROTO_IP,
2750 &ctx->ring_sock);
2751 if (ret)
2752 return ret;
2753#endif
2754
2755 ret = get_unused_fd_flags(O_RDWR | O_CLOEXEC);
2756 if (ret < 0)
2757 goto err;
2758
2759 file = anon_inode_getfile("[io_uring]", &io_uring_fops, ctx,
2760 O_RDWR | O_CLOEXEC);
2761 if (IS_ERR(file)) {
2762 put_unused_fd(ret);
2763 ret = PTR_ERR(file);
2764 goto err;
2765 }
2766
2767#if defined(CONFIG_UNIX)
2768 ctx->ring_sock->file = file;
2769 ctx->ring_sock->sk->sk_user_data = ctx;
2770#endif
2771 fd_install(ret, file);
2772 return ret;
2773err:
2774#if defined(CONFIG_UNIX)
2775 sock_release(ctx->ring_sock);
2776 ctx->ring_sock = NULL;
2777#endif
2778 return ret;
2779}
2780
2781static int io_uring_create(unsigned entries, struct io_uring_params *p)
2782{
2783 struct user_struct *user = NULL;
2784 struct io_ring_ctx *ctx;
2785 bool account_mem;
2786 int ret;
2787
2788 if (!entries || entries > IORING_MAX_ENTRIES)
2789 return -EINVAL;
2790
2791 /*
2792 * Use twice as many entries for the CQ ring. It's possible for the
2793 * application to drive a higher depth than the size of the SQ ring,
2794 * since the sqes are only used at submission time. This allows for
2795 * some flexibility in overcommitting a bit.
2796 */
2797 p->sq_entries = roundup_pow_of_two(entries);
2798 p->cq_entries = 2 * p->sq_entries;
2799
2800 user = get_uid(current_user());
2801 account_mem = !capable(CAP_IPC_LOCK);
2802
2803 if (account_mem) {
2804 ret = io_account_mem(user,
2805 ring_pages(p->sq_entries, p->cq_entries));
2806 if (ret) {
2807 free_uid(user);
2808 return ret;
2809 }
2810 }
2811
2812 ctx = io_ring_ctx_alloc(p);
2813 if (!ctx) {
2814 if (account_mem)
2815 io_unaccount_mem(user, ring_pages(p->sq_entries,
2816 p->cq_entries));
2817 free_uid(user);
2818 return -ENOMEM;
2819 }
2820 ctx->compat = in_compat_syscall();
2821 ctx->account_mem = account_mem;
2822 ctx->user = user;
2823
2824 ret = io_allocate_scq_urings(ctx, p);
2825 if (ret)
2826 goto err;
2827
2828 ret = io_sq_offload_start(ctx, p);
2829 if (ret)
2830 goto err;
2831
2832 ret = io_uring_get_fd(ctx);
2833 if (ret < 0)
2834 goto err;
2835
2836 memset(&p->sq_off, 0, sizeof(p->sq_off));
2837 p->sq_off.head = offsetof(struct io_sq_ring, r.head);
2838 p->sq_off.tail = offsetof(struct io_sq_ring, r.tail);
2839 p->sq_off.ring_mask = offsetof(struct io_sq_ring, ring_mask);
2840 p->sq_off.ring_entries = offsetof(struct io_sq_ring, ring_entries);
2841 p->sq_off.flags = offsetof(struct io_sq_ring, flags);
2842 p->sq_off.dropped = offsetof(struct io_sq_ring, dropped);
2843 p->sq_off.array = offsetof(struct io_sq_ring, array);
2844
2845 memset(&p->cq_off, 0, sizeof(p->cq_off));
2846 p->cq_off.head = offsetof(struct io_cq_ring, r.head);
2847 p->cq_off.tail = offsetof(struct io_cq_ring, r.tail);
2848 p->cq_off.ring_mask = offsetof(struct io_cq_ring, ring_mask);
2849 p->cq_off.ring_entries = offsetof(struct io_cq_ring, ring_entries);
2850 p->cq_off.overflow = offsetof(struct io_cq_ring, overflow);
2851 p->cq_off.cqes = offsetof(struct io_cq_ring, cqes);
2852 return ret;
2853err:
2854 io_ring_ctx_wait_and_kill(ctx);
2855 return ret;
2856}
2857
2858/*
2859 * Sets up an aio uring context, and returns the fd. Applications asks for a
2860 * ring size, we return the actual sq/cq ring sizes (among other things) in the
2861 * params structure passed in.
2862 */
2863static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
2864{
2865 struct io_uring_params p;
2866 long ret;
2867 int i;
2868
2869 if (copy_from_user(&p, params, sizeof(p)))
2870 return -EFAULT;
2871 for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
2872 if (p.resv[i])
2873 return -EINVAL;
2874 }
2875
2876 if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
2877 IORING_SETUP_SQ_AFF))
2878 return -EINVAL;
2879
2880 ret = io_uring_create(entries, &p);
2881 if (ret < 0)
2882 return ret;
2883
2884 if (copy_to_user(params, &p, sizeof(p)))
2885 return -EFAULT;
2886
2887 return ret;
2888}
2889
2890SYSCALL_DEFINE2(io_uring_setup, u32, entries,
2891 struct io_uring_params __user *, params)
2892{
2893 return io_uring_setup(entries, params);
2894}
2895
2896static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
2897 void __user *arg, unsigned nr_args)
2898{
2899 int ret;
2900
2901 percpu_ref_kill(&ctx->refs);
2902 wait_for_completion(&ctx->ctx_done);
2903
2904 switch (opcode) {
2905 case IORING_REGISTER_BUFFERS:
2906 ret = io_sqe_buffer_register(ctx, arg, nr_args);
2907 break;
2908 case IORING_UNREGISTER_BUFFERS:
2909 ret = -EINVAL;
2910 if (arg || nr_args)
2911 break;
2912 ret = io_sqe_buffer_unregister(ctx);
2913 break;
2914 case IORING_REGISTER_FILES:
2915 ret = io_sqe_files_register(ctx, arg, nr_args);
2916 break;
2917 case IORING_UNREGISTER_FILES:
2918 ret = -EINVAL;
2919 if (arg || nr_args)
2920 break;
2921 ret = io_sqe_files_unregister(ctx);
2922 break;
2923 default:
2924 ret = -EINVAL;
2925 break;
2926 }
2927
2928 /* bring the ctx back to life */
2929 reinit_completion(&ctx->ctx_done);
2930 percpu_ref_reinit(&ctx->refs);
2931 return ret;
2932}
2933
2934SYSCALL_DEFINE4(io_uring_register, unsigned int, fd, unsigned int, opcode,
2935 void __user *, arg, unsigned int, nr_args)
2936{
2937 struct io_ring_ctx *ctx;
2938 long ret = -EBADF;
2939 struct fd f;
2940
2941 f = fdget(fd);
2942 if (!f.file)
2943 return -EBADF;
2944
2945 ret = -EOPNOTSUPP;
2946 if (f.file->f_op != &io_uring_fops)
2947 goto out_fput;
2948
2949 ctx = f.file->private_data;
2950
2951 mutex_lock(&ctx->uring_lock);
2952 ret = __io_uring_register(ctx, opcode, arg, nr_args);
2953 mutex_unlock(&ctx->uring_lock);
2954out_fput:
2955 fdput(f);
2956 return ret;
2957}
2958
2959static int __init io_uring_init(void)
2960{
2961 req_cachep = KMEM_CACHE(io_kiocb, SLAB_HWCACHE_ALIGN | SLAB_PANIC);
2962 return 0;
2963};
2964__initcall(io_uring_init);
2965