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