Merge tag 'linux-kselftest-5.3-rc3' 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
1842                 /* Ensure we clear previously set non-block flag */
1843                 req->rw.ki_flags &= ~IOCB_NOWAIT;
1844
1845                 ret = 0;
1846                 if (io_sqe_needs_user(sqe) && !cur_mm) {
1847                         if (!mmget_not_zero(ctx->sqo_mm)) {
1848                                 ret = -EFAULT;
1849                         } else {
1850                                 cur_mm = ctx->sqo_mm;
1851                                 use_mm(cur_mm);
1852                                 old_fs = get_fs();
1853                                 set_fs(USER_DS);
1854                         }
1855                 }
1856
1857                 if (!ret) {
1858                         s->has_user = cur_mm != NULL;
1859                         s->needs_lock = true;
1860                         do {
1861                                 ret = __io_submit_sqe(ctx, req, s, false);
1862                                 /*
1863                                  * We can get EAGAIN for polled IO even though
1864                                  * we're forcing a sync submission from here,
1865                                  * since we can't wait for request slots on the
1866                                  * block side.
1867                                  */
1868                                 if (ret != -EAGAIN)
1869                                         break;
1870                                 cond_resched();
1871                         } while (1);
1872                 }
1873
1874                 /* drop submission reference */
1875                 io_put_req(req);
1876
1877                 if (ret) {
1878                         io_cqring_add_event(ctx, sqe->user_data, ret);
1879                         io_put_req(req);
1880                 }
1881
1882                 /* async context always use a copy of the sqe */
1883                 kfree(sqe);
1884
1885                 /* req from defer and link list needn't decrease async cnt */
1886                 if (req->flags & (REQ_F_IO_DRAINED | REQ_F_LINK_DONE))
1887                         goto out;
1888
1889                 if (!async_list)
1890                         break;
1891                 if (!list_empty(&req_list)) {
1892                         req = list_first_entry(&req_list, struct io_kiocb,
1893                                                 list);
1894                         list_del(&req->list);
1895                         continue;
1896                 }
1897                 if (list_empty(&async_list->list))
1898                         break;
1899
1900                 req = NULL;
1901                 spin_lock(&async_list->lock);
1902                 if (list_empty(&async_list->list)) {
1903                         spin_unlock(&async_list->lock);
1904                         break;
1905                 }
1906                 list_splice_init(&async_list->list, &req_list);
1907                 spin_unlock(&async_list->lock);
1908
1909                 req = list_first_entry(&req_list, struct io_kiocb, list);
1910                 list_del(&req->list);
1911         } while (req);
1912
1913         /*
1914          * Rare case of racing with a submitter. If we find the count has
1915          * dropped to zero AND we have pending work items, then restart
1916          * the processing. This is a tiny race window.
1917          */
1918         if (async_list) {
1919                 ret = atomic_dec_return(&async_list->cnt);
1920                 while (!ret && !list_empty(&async_list->list)) {
1921                         spin_lock(&async_list->lock);
1922                         atomic_inc(&async_list->cnt);
1923                         list_splice_init(&async_list->list, &req_list);
1924                         spin_unlock(&async_list->lock);
1925
1926                         if (!list_empty(&req_list)) {
1927                                 req = list_first_entry(&req_list,
1928                                                         struct io_kiocb, list);
1929                                 list_del(&req->list);
1930                                 goto restart;
1931                         }
1932                         ret = atomic_dec_return(&async_list->cnt);
1933                 }
1934         }
1935
1936 out:
1937         if (cur_mm) {
1938                 set_fs(old_fs);
1939                 unuse_mm(cur_mm);
1940                 mmput(cur_mm);
1941         }
1942 }
1943
1944 /*
1945  * See if we can piggy back onto previously submitted work, that is still
1946  * running. We currently only allow this if the new request is sequential
1947  * to the previous one we punted.
1948  */
1949 static bool io_add_to_prev_work(struct async_list *list, struct io_kiocb *req)
1950 {
1951         bool ret = false;
1952
1953         if (!list)
1954                 return false;
1955         if (!(req->flags & REQ_F_SEQ_PREV))
1956                 return false;
1957         if (!atomic_read(&list->cnt))
1958                 return false;
1959
1960         ret = true;
1961         spin_lock(&list->lock);
1962         list_add_tail(&req->list, &list->list);
1963         /*
1964          * Ensure we see a simultaneous modification from io_sq_wq_submit_work()
1965          */
1966         smp_mb();
1967         if (!atomic_read(&list->cnt)) {
1968                 list_del_init(&req->list);
1969                 ret = false;
1970         }
1971         spin_unlock(&list->lock);
1972         return ret;
1973 }
1974
1975 static bool io_op_needs_file(const struct io_uring_sqe *sqe)
1976 {
1977         int op = READ_ONCE(sqe->opcode);
1978
1979         switch (op) {
1980         case IORING_OP_NOP:
1981         case IORING_OP_POLL_REMOVE:
1982                 return false;
1983         default:
1984                 return true;
1985         }
1986 }
1987
1988 static int io_req_set_file(struct io_ring_ctx *ctx, const struct sqe_submit *s,
1989                            struct io_submit_state *state, struct io_kiocb *req)
1990 {
1991         unsigned flags;
1992         int fd;
1993
1994         flags = READ_ONCE(s->sqe->flags);
1995         fd = READ_ONCE(s->sqe->fd);
1996
1997         if (flags & IOSQE_IO_DRAIN) {
1998                 req->flags |= REQ_F_IO_DRAIN;
1999                 req->sequence = ctx->cached_sq_head - 1;
2000         }
2001
2002         if (!io_op_needs_file(s->sqe))
2003                 return 0;
2004
2005         if (flags & IOSQE_FIXED_FILE) {
2006                 if (unlikely(!ctx->user_files ||
2007                     (unsigned) fd >= ctx->nr_user_files))
2008                         return -EBADF;
2009                 req->file = ctx->user_files[fd];
2010                 req->flags |= REQ_F_FIXED_FILE;
2011         } else {
2012                 if (s->needs_fixed_file)
2013                         return -EBADF;
2014                 req->file = io_file_get(state, fd);
2015                 if (unlikely(!req->file))
2016                         return -EBADF;
2017         }
2018
2019         return 0;
2020 }
2021
2022 static int io_queue_sqe(struct io_ring_ctx *ctx, struct io_kiocb *req,
2023                         struct sqe_submit *s)
2024 {
2025         int ret;
2026
2027         ret = __io_submit_sqe(ctx, req, s, true);
2028         if (ret == -EAGAIN && !(req->flags & REQ_F_NOWAIT)) {
2029                 struct io_uring_sqe *sqe_copy;
2030
2031                 sqe_copy = kmalloc(sizeof(*sqe_copy), GFP_KERNEL);
2032                 if (sqe_copy) {
2033                         struct async_list *list;
2034
2035                         memcpy(sqe_copy, s->sqe, sizeof(*sqe_copy));
2036                         s->sqe = sqe_copy;
2037
2038                         memcpy(&req->submit, s, sizeof(*s));
2039                         list = io_async_list_from_sqe(ctx, s->sqe);
2040                         if (!io_add_to_prev_work(list, req)) {
2041                                 if (list)
2042                                         atomic_inc(&list->cnt);
2043                                 INIT_WORK(&req->work, io_sq_wq_submit_work);
2044                                 queue_work(ctx->sqo_wq, &req->work);
2045                         }
2046
2047                         /*
2048                          * Queued up for async execution, worker will release
2049                          * submit reference when the iocb is actually submitted.
2050                          */
2051                         return 0;
2052                 }
2053         }
2054
2055         /* drop submission reference */
2056         io_put_req(req);
2057
2058         /* and drop final reference, if we failed */
2059         if (ret) {
2060                 io_cqring_add_event(ctx, req->user_data, ret);
2061                 if (req->flags & REQ_F_LINK)
2062                         req->flags |= REQ_F_FAIL_LINK;
2063                 io_put_req(req);
2064         }
2065
2066         return ret;
2067 }
2068
2069 #define SQE_VALID_FLAGS (IOSQE_FIXED_FILE|IOSQE_IO_DRAIN|IOSQE_IO_LINK)
2070
2071 static void io_submit_sqe(struct io_ring_ctx *ctx, struct sqe_submit *s,
2072                           struct io_submit_state *state, struct io_kiocb **link)
2073 {
2074         struct io_uring_sqe *sqe_copy;
2075         struct io_kiocb *req;
2076         int ret;
2077
2078         /* enforce forwards compatibility on users */
2079         if (unlikely(s->sqe->flags & ~SQE_VALID_FLAGS)) {
2080                 ret = -EINVAL;
2081                 goto err;
2082         }
2083
2084         req = io_get_req(ctx, state);
2085         if (unlikely(!req)) {
2086                 ret = -EAGAIN;
2087                 goto err;
2088         }
2089
2090         ret = io_req_set_file(ctx, s, state, req);
2091         if (unlikely(ret)) {
2092 err_req:
2093                 io_free_req(req);
2094 err:
2095                 io_cqring_add_event(ctx, s->sqe->user_data, ret);
2096                 return;
2097         }
2098
2099         ret = io_req_defer(ctx, req, s->sqe);
2100         if (ret) {
2101                 if (ret != -EIOCBQUEUED)
2102                         goto err_req;
2103                 return;
2104         }
2105
2106         /*
2107          * If we already have a head request, queue this one for async
2108          * submittal once the head completes. If we don't have a head but
2109          * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be
2110          * submitted sync once the chain is complete. If none of those
2111          * conditions are true (normal request), then just queue it.
2112          */
2113         if (*link) {
2114                 struct io_kiocb *prev = *link;
2115
2116                 sqe_copy = kmemdup(s->sqe, sizeof(*sqe_copy), GFP_KERNEL);
2117                 if (!sqe_copy) {
2118                         ret = -EAGAIN;
2119                         goto err_req;
2120                 }
2121
2122                 s->sqe = sqe_copy;
2123                 memcpy(&req->submit, s, sizeof(*s));
2124                 list_add_tail(&req->list, &prev->link_list);
2125         } else if (s->sqe->flags & IOSQE_IO_LINK) {
2126                 req->flags |= REQ_F_LINK;
2127
2128                 memcpy(&req->submit, s, sizeof(*s));
2129                 INIT_LIST_HEAD(&req->link_list);
2130                 *link = req;
2131         } else {
2132                 io_queue_sqe(ctx, req, s);
2133         }
2134 }
2135
2136 /*
2137  * Batched submission is done, ensure local IO is flushed out.
2138  */
2139 static void io_submit_state_end(struct io_submit_state *state)
2140 {
2141         blk_finish_plug(&state->plug);
2142         io_file_put(state);
2143         if (state->free_reqs)
2144                 kmem_cache_free_bulk(req_cachep, state->free_reqs,
2145                                         &state->reqs[state->cur_req]);
2146 }
2147
2148 /*
2149  * Start submission side cache.
2150  */
2151 static void io_submit_state_start(struct io_submit_state *state,
2152                                   struct io_ring_ctx *ctx, unsigned max_ios)
2153 {
2154         blk_start_plug(&state->plug);
2155         state->free_reqs = 0;
2156         state->file = NULL;
2157         state->ios_left = max_ios;
2158 }
2159
2160 static void io_commit_sqring(struct io_ring_ctx *ctx)
2161 {
2162         struct io_sq_ring *ring = ctx->sq_ring;
2163
2164         if (ctx->cached_sq_head != READ_ONCE(ring->r.head)) {
2165                 /*
2166                  * Ensure any loads from the SQEs are done at this point,
2167                  * since once we write the new head, the application could
2168                  * write new data to them.
2169                  */
2170                 smp_store_release(&ring->r.head, ctx->cached_sq_head);
2171         }
2172 }
2173
2174 /*
2175  * Fetch an sqe, if one is available. Note that s->sqe will point to memory
2176  * that is mapped by userspace. This means that care needs to be taken to
2177  * ensure that reads are stable, as we cannot rely on userspace always
2178  * being a good citizen. If members of the sqe are validated and then later
2179  * used, it's important that those reads are done through READ_ONCE() to
2180  * prevent a re-load down the line.
2181  */
2182 static bool io_get_sqring(struct io_ring_ctx *ctx, struct sqe_submit *s)
2183 {
2184         struct io_sq_ring *ring = ctx->sq_ring;
2185         unsigned head;
2186
2187         /*
2188          * The cached sq head (or cq tail) serves two purposes:
2189          *
2190          * 1) allows us to batch the cost of updating the user visible
2191          *    head updates.
2192          * 2) allows the kernel side to track the head on its own, even
2193          *    though the application is the one updating it.
2194          */
2195         head = ctx->cached_sq_head;
2196         /* make sure SQ entry isn't read before tail */
2197         if (head == smp_load_acquire(&ring->r.tail))
2198                 return false;
2199
2200         head = READ_ONCE(ring->array[head & ctx->sq_mask]);
2201         if (head < ctx->sq_entries) {
2202                 s->index = head;
2203                 s->sqe = &ctx->sq_sqes[head];
2204                 ctx->cached_sq_head++;
2205                 return true;
2206         }
2207
2208         /* drop invalid entries */
2209         ctx->cached_sq_head++;
2210         ring->dropped++;
2211         return false;
2212 }
2213
2214 static int io_submit_sqes(struct io_ring_ctx *ctx, struct sqe_submit *sqes,
2215                           unsigned int nr, bool has_user, bool mm_fault)
2216 {
2217         struct io_submit_state state, *statep = NULL;
2218         struct io_kiocb *link = NULL;
2219         bool prev_was_link = false;
2220         int i, submitted = 0;
2221
2222         if (nr > IO_PLUG_THRESHOLD) {
2223                 io_submit_state_start(&state, ctx, nr);
2224                 statep = &state;
2225         }
2226
2227         for (i = 0; i < nr; i++) {
2228                 /*
2229                  * If previous wasn't linked and we have a linked command,
2230                  * that's the end of the chain. Submit the previous link.
2231                  */
2232                 if (!prev_was_link && link) {
2233                         io_queue_sqe(ctx, link, &link->submit);
2234                         link = NULL;
2235                 }
2236                 prev_was_link = (sqes[i].sqe->flags & IOSQE_IO_LINK) != 0;
2237
2238                 if (unlikely(mm_fault)) {
2239                         io_cqring_add_event(ctx, sqes[i].sqe->user_data,
2240                                                 -EFAULT);
2241                 } else {
2242                         sqes[i].has_user = has_user;
2243                         sqes[i].needs_lock = true;
2244                         sqes[i].needs_fixed_file = true;
2245                         io_submit_sqe(ctx, &sqes[i], statep, &link);
2246                         submitted++;
2247                 }
2248         }
2249
2250         if (link)
2251                 io_queue_sqe(ctx, link, &link->submit);
2252         if (statep)
2253                 io_submit_state_end(&state);
2254
2255         return submitted;
2256 }
2257
2258 static int io_sq_thread(void *data)
2259 {
2260         struct sqe_submit sqes[IO_IOPOLL_BATCH];
2261         struct io_ring_ctx *ctx = data;
2262         struct mm_struct *cur_mm = NULL;
2263         mm_segment_t old_fs;
2264         DEFINE_WAIT(wait);
2265         unsigned inflight;
2266         unsigned long timeout;
2267
2268         complete(&ctx->sqo_thread_started);
2269
2270         old_fs = get_fs();
2271         set_fs(USER_DS);
2272
2273         timeout = inflight = 0;
2274         while (!kthread_should_park()) {
2275                 bool all_fixed, mm_fault = false;
2276                 int i;
2277
2278                 if (inflight) {
2279                         unsigned nr_events = 0;
2280
2281                         if (ctx->flags & IORING_SETUP_IOPOLL) {
2282                                 /*
2283                                  * We disallow the app entering submit/complete
2284                                  * with polling, but we still need to lock the
2285                                  * ring to prevent racing with polled issue
2286                                  * that got punted to a workqueue.
2287                                  */
2288                                 mutex_lock(&ctx->uring_lock);
2289                                 io_iopoll_check(ctx, &nr_events, 0);
2290                                 mutex_unlock(&ctx->uring_lock);
2291                         } else {
2292                                 /*
2293                                  * Normal IO, just pretend everything completed.
2294                                  * We don't have to poll completions for that.
2295                                  */
2296                                 nr_events = inflight;
2297                         }
2298
2299                         inflight -= nr_events;
2300                         if (!inflight)
2301                                 timeout = jiffies + ctx->sq_thread_idle;
2302                 }
2303
2304                 if (!io_get_sqring(ctx, &sqes[0])) {
2305                         /*
2306                          * We're polling. If we're within the defined idle
2307                          * period, then let us spin without work before going
2308                          * to sleep.
2309                          */
2310                         if (inflight || !time_after(jiffies, timeout)) {
2311                                 cpu_relax();
2312                                 continue;
2313                         }
2314
2315                         /*
2316                          * Drop cur_mm before scheduling, we can't hold it for
2317                          * long periods (or over schedule()). Do this before
2318                          * adding ourselves to the waitqueue, as the unuse/drop
2319                          * may sleep.
2320                          */
2321                         if (cur_mm) {
2322                                 unuse_mm(cur_mm);
2323                                 mmput(cur_mm);
2324                                 cur_mm = NULL;
2325                         }
2326
2327                         prepare_to_wait(&ctx->sqo_wait, &wait,
2328                                                 TASK_INTERRUPTIBLE);
2329
2330                         /* Tell userspace we may need a wakeup call */
2331                         ctx->sq_ring->flags |= IORING_SQ_NEED_WAKEUP;
2332                         /* make sure to read SQ tail after writing flags */
2333                         smp_mb();
2334
2335                         if (!io_get_sqring(ctx, &sqes[0])) {
2336                                 if (kthread_should_park()) {
2337                                         finish_wait(&ctx->sqo_wait, &wait);
2338                                         break;
2339                                 }
2340                                 if (signal_pending(current))
2341                                         flush_signals(current);
2342                                 schedule();
2343                                 finish_wait(&ctx->sqo_wait, &wait);
2344
2345                                 ctx->sq_ring->flags &= ~IORING_SQ_NEED_WAKEUP;
2346                                 continue;
2347                         }
2348                         finish_wait(&ctx->sqo_wait, &wait);
2349
2350                         ctx->sq_ring->flags &= ~IORING_SQ_NEED_WAKEUP;
2351                 }
2352
2353                 i = 0;
2354                 all_fixed = true;
2355                 do {
2356                         if (all_fixed && io_sqe_needs_user(sqes[i].sqe))
2357                                 all_fixed = false;
2358
2359                         i++;
2360                         if (i == ARRAY_SIZE(sqes))
2361                                 break;
2362                 } while (io_get_sqring(ctx, &sqes[i]));
2363
2364                 /* Unless all new commands are FIXED regions, grab mm */
2365                 if (!all_fixed && !cur_mm) {
2366                         mm_fault = !mmget_not_zero(ctx->sqo_mm);
2367                         if (!mm_fault) {
2368                                 use_mm(ctx->sqo_mm);
2369                                 cur_mm = ctx->sqo_mm;
2370                         }
2371                 }
2372
2373                 inflight += io_submit_sqes(ctx, sqes, i, cur_mm != NULL,
2374                                                 mm_fault);
2375
2376                 /* Commit SQ ring head once we've consumed all SQEs */
2377                 io_commit_sqring(ctx);
2378         }
2379
2380         set_fs(old_fs);
2381         if (cur_mm) {
2382                 unuse_mm(cur_mm);
2383                 mmput(cur_mm);
2384         }
2385
2386         kthread_parkme();
2387
2388         return 0;
2389 }
2390
2391 static int io_ring_submit(struct io_ring_ctx *ctx, unsigned int to_submit)
2392 {
2393         struct io_submit_state state, *statep = NULL;
2394         struct io_kiocb *link = NULL;
2395         bool prev_was_link = false;
2396         int i, submit = 0;
2397
2398         if (to_submit > IO_PLUG_THRESHOLD) {
2399                 io_submit_state_start(&state, ctx, to_submit);
2400                 statep = &state;
2401         }
2402
2403         for (i = 0; i < to_submit; i++) {
2404                 struct sqe_submit s;
2405
2406                 if (!io_get_sqring(ctx, &s))
2407                         break;
2408
2409                 /*
2410                  * If previous wasn't linked and we have a linked command,
2411                  * that's the end of the chain. Submit the previous link.
2412                  */
2413                 if (!prev_was_link && link) {
2414                         io_queue_sqe(ctx, link, &link->submit);
2415                         link = NULL;
2416                 }
2417                 prev_was_link = (s.sqe->flags & IOSQE_IO_LINK) != 0;
2418
2419                 s.has_user = true;
2420                 s.needs_lock = false;
2421                 s.needs_fixed_file = false;
2422                 submit++;
2423                 io_submit_sqe(ctx, &s, statep, &link);
2424         }
2425         io_commit_sqring(ctx);
2426
2427         if (link)
2428                 io_queue_sqe(ctx, link, &link->submit);
2429         if (statep)
2430                 io_submit_state_end(statep);
2431
2432         return submit;
2433 }
2434
2435 static unsigned io_cqring_events(struct io_cq_ring *ring)
2436 {
2437         /* See comment at the top of this file */
2438         smp_rmb();
2439         return READ_ONCE(ring->r.tail) - READ_ONCE(ring->r.head);
2440 }
2441
2442 /*
2443  * Wait until events become available, if we don't already have some. The
2444  * application must reap them itself, as they reside on the shared cq ring.
2445  */
2446 static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
2447                           const sigset_t __user *sig, size_t sigsz)
2448 {
2449         struct io_cq_ring *ring = ctx->cq_ring;
2450         int ret;
2451
2452         if (io_cqring_events(ring) >= min_events)
2453                 return 0;
2454
2455         if (sig) {
2456 #ifdef CONFIG_COMPAT
2457                 if (in_compat_syscall())
2458                         ret = set_compat_user_sigmask((const compat_sigset_t __user *)sig,
2459                                                       sigsz);
2460                 else
2461 #endif
2462                         ret = set_user_sigmask(sig, sigsz);
2463
2464                 if (ret)
2465                         return ret;
2466         }
2467
2468         ret = wait_event_interruptible(ctx->wait, io_cqring_events(ring) >= min_events);
2469         restore_saved_sigmask_unless(ret == -ERESTARTSYS);
2470         if (ret == -ERESTARTSYS)
2471                 ret = -EINTR;
2472
2473         return READ_ONCE(ring->r.head) == READ_ONCE(ring->r.tail) ? ret : 0;
2474 }
2475
2476 static void __io_sqe_files_unregister(struct io_ring_ctx *ctx)
2477 {
2478 #if defined(CONFIG_UNIX)
2479         if (ctx->ring_sock) {
2480                 struct sock *sock = ctx->ring_sock->sk;
2481                 struct sk_buff *skb;
2482
2483                 while ((skb = skb_dequeue(&sock->sk_receive_queue)) != NULL)
2484                         kfree_skb(skb);
2485         }
2486 #else
2487         int i;
2488
2489         for (i = 0; i < ctx->nr_user_files; i++)
2490                 fput(ctx->user_files[i]);
2491 #endif
2492 }
2493
2494 static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
2495 {
2496         if (!ctx->user_files)
2497                 return -ENXIO;
2498
2499         __io_sqe_files_unregister(ctx);
2500         kfree(ctx->user_files);
2501         ctx->user_files = NULL;
2502         ctx->nr_user_files = 0;
2503         return 0;
2504 }
2505
2506 static void io_sq_thread_stop(struct io_ring_ctx *ctx)
2507 {
2508         if (ctx->sqo_thread) {
2509                 wait_for_completion(&ctx->sqo_thread_started);
2510                 /*
2511                  * The park is a bit of a work-around, without it we get
2512                  * warning spews on shutdown with SQPOLL set and affinity
2513                  * set to a single CPU.
2514                  */
2515                 kthread_park(ctx->sqo_thread);
2516                 kthread_stop(ctx->sqo_thread);
2517                 ctx->sqo_thread = NULL;
2518         }
2519 }
2520
2521 static void io_finish_async(struct io_ring_ctx *ctx)
2522 {
2523         io_sq_thread_stop(ctx);
2524
2525         if (ctx->sqo_wq) {
2526                 destroy_workqueue(ctx->sqo_wq);
2527                 ctx->sqo_wq = NULL;
2528         }
2529 }
2530
2531 #if defined(CONFIG_UNIX)
2532 static void io_destruct_skb(struct sk_buff *skb)
2533 {
2534         struct io_ring_ctx *ctx = skb->sk->sk_user_data;
2535
2536         io_finish_async(ctx);
2537         unix_destruct_scm(skb);
2538 }
2539
2540 /*
2541  * Ensure the UNIX gc is aware of our file set, so we are certain that
2542  * the io_uring can be safely unregistered on process exit, even if we have
2543  * loops in the file referencing.
2544  */
2545 static int __io_sqe_files_scm(struct io_ring_ctx *ctx, int nr, int offset)
2546 {
2547         struct sock *sk = ctx->ring_sock->sk;
2548         struct scm_fp_list *fpl;
2549         struct sk_buff *skb;
2550         int i;
2551
2552         if (!capable(CAP_SYS_RESOURCE) && !capable(CAP_SYS_ADMIN)) {
2553                 unsigned long inflight = ctx->user->unix_inflight + nr;
2554
2555                 if (inflight > task_rlimit(current, RLIMIT_NOFILE))
2556                         return -EMFILE;
2557         }
2558
2559         fpl = kzalloc(sizeof(*fpl), GFP_KERNEL);
2560         if (!fpl)
2561                 return -ENOMEM;
2562
2563         skb = alloc_skb(0, GFP_KERNEL);
2564         if (!skb) {
2565                 kfree(fpl);
2566                 return -ENOMEM;
2567         }
2568
2569         skb->sk = sk;
2570         skb->destructor = io_destruct_skb;
2571
2572         fpl->user = get_uid(ctx->user);
2573         for (i = 0; i < nr; i++) {
2574                 fpl->fp[i] = get_file(ctx->user_files[i + offset]);
2575                 unix_inflight(fpl->user, fpl->fp[i]);
2576         }
2577
2578         fpl->max = fpl->count = nr;
2579         UNIXCB(skb).fp = fpl;
2580         refcount_add(skb->truesize, &sk->sk_wmem_alloc);
2581         skb_queue_head(&sk->sk_receive_queue, skb);
2582
2583         for (i = 0; i < nr; i++)
2584                 fput(fpl->fp[i]);
2585
2586         return 0;
2587 }
2588
2589 /*
2590  * If UNIX sockets are enabled, fd passing can cause a reference cycle which
2591  * causes regular reference counting to break down. We rely on the UNIX
2592  * garbage collection to take care of this problem for us.
2593  */
2594 static int io_sqe_files_scm(struct io_ring_ctx *ctx)
2595 {
2596         unsigned left, total;
2597         int ret = 0;
2598
2599         total = 0;
2600         left = ctx->nr_user_files;
2601         while (left) {
2602                 unsigned this_files = min_t(unsigned, left, SCM_MAX_FD);
2603
2604                 ret = __io_sqe_files_scm(ctx, this_files, total);
2605                 if (ret)
2606                         break;
2607                 left -= this_files;
2608                 total += this_files;
2609         }
2610
2611         if (!ret)
2612                 return 0;
2613
2614         while (total < ctx->nr_user_files) {
2615                 fput(ctx->user_files[total]);
2616                 total++;
2617         }
2618
2619         return ret;
2620 }
2621 #else
2622 static int io_sqe_files_scm(struct io_ring_ctx *ctx)
2623 {
2624         return 0;
2625 }
2626 #endif
2627
2628 static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
2629                                  unsigned nr_args)
2630 {
2631         __s32 __user *fds = (__s32 __user *) arg;
2632         int fd, ret = 0;
2633         unsigned i;
2634
2635         if (ctx->user_files)
2636                 return -EBUSY;
2637         if (!nr_args)
2638                 return -EINVAL;
2639         if (nr_args > IORING_MAX_FIXED_FILES)
2640                 return -EMFILE;
2641
2642         ctx->user_files = kcalloc(nr_args, sizeof(struct file *), GFP_KERNEL);
2643         if (!ctx->user_files)
2644                 return -ENOMEM;
2645
2646         for (i = 0; i < nr_args; i++) {
2647                 ret = -EFAULT;
2648                 if (copy_from_user(&fd, &fds[i], sizeof(fd)))
2649                         break;
2650
2651                 ctx->user_files[i] = fget(fd);
2652
2653                 ret = -EBADF;
2654                 if (!ctx->user_files[i])
2655                         break;
2656                 /*
2657                  * Don't allow io_uring instances to be registered. If UNIX
2658                  * isn't enabled, then this causes a reference cycle and this
2659                  * instance can never get freed. If UNIX is enabled we'll
2660                  * handle it just fine, but there's still no point in allowing
2661                  * a ring fd as it doesn't support regular read/write anyway.
2662                  */
2663                 if (ctx->user_files[i]->f_op == &io_uring_fops) {
2664                         fput(ctx->user_files[i]);
2665                         break;
2666                 }
2667                 ctx->nr_user_files++;
2668                 ret = 0;
2669         }
2670
2671         if (ret) {
2672                 for (i = 0; i < ctx->nr_user_files; i++)
2673                         fput(ctx->user_files[i]);
2674
2675                 kfree(ctx->user_files);
2676                 ctx->user_files = NULL;
2677                 ctx->nr_user_files = 0;
2678                 return ret;
2679         }
2680
2681         ret = io_sqe_files_scm(ctx);
2682         if (ret)
2683                 io_sqe_files_unregister(ctx);
2684
2685         return ret;
2686 }
2687
2688 static int io_sq_offload_start(struct io_ring_ctx *ctx,
2689                                struct io_uring_params *p)
2690 {
2691         int ret;
2692
2693         init_waitqueue_head(&ctx->sqo_wait);
2694         mmgrab(current->mm);
2695         ctx->sqo_mm = current->mm;
2696
2697         if (ctx->flags & IORING_SETUP_SQPOLL) {
2698                 ret = -EPERM;
2699                 if (!capable(CAP_SYS_ADMIN))
2700                         goto err;
2701
2702                 ctx->sq_thread_idle = msecs_to_jiffies(p->sq_thread_idle);
2703                 if (!ctx->sq_thread_idle)
2704                         ctx->sq_thread_idle = HZ;
2705
2706                 if (p->flags & IORING_SETUP_SQ_AFF) {
2707                         int cpu = p->sq_thread_cpu;
2708
2709                         ret = -EINVAL;
2710                         if (cpu >= nr_cpu_ids)
2711                                 goto err;
2712                         if (!cpu_online(cpu))
2713                                 goto err;
2714
2715                         ctx->sqo_thread = kthread_create_on_cpu(io_sq_thread,
2716                                                         ctx, cpu,
2717                                                         "io_uring-sq");
2718                 } else {
2719                         ctx->sqo_thread = kthread_create(io_sq_thread, ctx,
2720                                                         "io_uring-sq");
2721                 }
2722                 if (IS_ERR(ctx->sqo_thread)) {
2723                         ret = PTR_ERR(ctx->sqo_thread);
2724                         ctx->sqo_thread = NULL;
2725                         goto err;
2726                 }
2727                 wake_up_process(ctx->sqo_thread);
2728         } else if (p->flags & IORING_SETUP_SQ_AFF) {
2729                 /* Can't have SQ_AFF without SQPOLL */
2730                 ret = -EINVAL;
2731                 goto err;
2732         }
2733
2734         /* Do QD, or 2 * CPUS, whatever is smallest */
2735         ctx->sqo_wq = alloc_workqueue("io_ring-wq", WQ_UNBOUND | WQ_FREEZABLE,
2736                         min(ctx->sq_entries - 1, 2 * num_online_cpus()));
2737         if (!ctx->sqo_wq) {
2738                 ret = -ENOMEM;
2739                 goto err;
2740         }
2741
2742         return 0;
2743 err:
2744         io_sq_thread_stop(ctx);
2745         mmdrop(ctx->sqo_mm);
2746         ctx->sqo_mm = NULL;
2747         return ret;
2748 }
2749
2750 static void io_unaccount_mem(struct user_struct *user, unsigned long nr_pages)
2751 {
2752         atomic_long_sub(nr_pages, &user->locked_vm);
2753 }
2754
2755 static int io_account_mem(struct user_struct *user, unsigned long nr_pages)
2756 {
2757         unsigned long page_limit, cur_pages, new_pages;
2758
2759         /* Don't allow more pages than we can safely lock */
2760         page_limit = rlimit(RLIMIT_MEMLOCK) >> PAGE_SHIFT;
2761
2762         do {
2763                 cur_pages = atomic_long_read(&user->locked_vm);
2764                 new_pages = cur_pages + nr_pages;
2765                 if (new_pages > page_limit)
2766                         return -ENOMEM;
2767         } while (atomic_long_cmpxchg(&user->locked_vm, cur_pages,
2768                                         new_pages) != cur_pages);
2769
2770         return 0;
2771 }
2772
2773 static void io_mem_free(void *ptr)
2774 {
2775         struct page *page;
2776
2777         if (!ptr)
2778                 return;
2779
2780         page = virt_to_head_page(ptr);
2781         if (put_page_testzero(page))
2782                 free_compound_page(page);
2783 }
2784
2785 static void *io_mem_alloc(size_t size)
2786 {
2787         gfp_t gfp_flags = GFP_KERNEL | __GFP_ZERO | __GFP_NOWARN | __GFP_COMP |
2788                                 __GFP_NORETRY;
2789
2790         return (void *) __get_free_pages(gfp_flags, get_order(size));
2791 }
2792
2793 static unsigned long ring_pages(unsigned sq_entries, unsigned cq_entries)
2794 {
2795         struct io_sq_ring *sq_ring;
2796         struct io_cq_ring *cq_ring;
2797         size_t bytes;
2798
2799         bytes = struct_size(sq_ring, array, sq_entries);
2800         bytes += array_size(sizeof(struct io_uring_sqe), sq_entries);
2801         bytes += struct_size(cq_ring, cqes, cq_entries);
2802
2803         return (bytes + PAGE_SIZE - 1) / PAGE_SIZE;
2804 }
2805
2806 static int io_sqe_buffer_unregister(struct io_ring_ctx *ctx)
2807 {
2808         int i, j;
2809
2810         if (!ctx->user_bufs)
2811                 return -ENXIO;
2812
2813         for (i = 0; i < ctx->nr_user_bufs; i++) {
2814                 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
2815
2816                 for (j = 0; j < imu->nr_bvecs; j++)
2817                         put_page(imu->bvec[j].bv_page);
2818
2819                 if (ctx->account_mem)
2820                         io_unaccount_mem(ctx->user, imu->nr_bvecs);
2821                 kvfree(imu->bvec);
2822                 imu->nr_bvecs = 0;
2823         }
2824
2825         kfree(ctx->user_bufs);
2826         ctx->user_bufs = NULL;
2827         ctx->nr_user_bufs = 0;
2828         return 0;
2829 }
2830
2831 static int io_copy_iov(struct io_ring_ctx *ctx, struct iovec *dst,
2832                        void __user *arg, unsigned index)
2833 {
2834         struct iovec __user *src;
2835
2836 #ifdef CONFIG_COMPAT
2837         if (ctx->compat) {
2838                 struct compat_iovec __user *ciovs;
2839                 struct compat_iovec ciov;
2840
2841                 ciovs = (struct compat_iovec __user *) arg;
2842                 if (copy_from_user(&ciov, &ciovs[index], sizeof(ciov)))
2843                         return -EFAULT;
2844
2845                 dst->iov_base = (void __user *) (unsigned long) ciov.iov_base;
2846                 dst->iov_len = ciov.iov_len;
2847                 return 0;
2848         }
2849 #endif
2850         src = (struct iovec __user *) arg;
2851         if (copy_from_user(dst, &src[index], sizeof(*dst)))
2852                 return -EFAULT;
2853         return 0;
2854 }
2855
2856 static int io_sqe_buffer_register(struct io_ring_ctx *ctx, void __user *arg,
2857                                   unsigned nr_args)
2858 {
2859         struct vm_area_struct **vmas = NULL;
2860         struct page **pages = NULL;
2861         int i, j, got_pages = 0;
2862         int ret = -EINVAL;
2863
2864         if (ctx->user_bufs)
2865                 return -EBUSY;
2866         if (!nr_args || nr_args > UIO_MAXIOV)
2867                 return -EINVAL;
2868
2869         ctx->user_bufs = kcalloc(nr_args, sizeof(struct io_mapped_ubuf),
2870                                         GFP_KERNEL);
2871         if (!ctx->user_bufs)
2872                 return -ENOMEM;
2873
2874         for (i = 0; i < nr_args; i++) {
2875                 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
2876                 unsigned long off, start, end, ubuf;
2877                 int pret, nr_pages;
2878                 struct iovec iov;
2879                 size_t size;
2880
2881                 ret = io_copy_iov(ctx, &iov, arg, i);
2882                 if (ret)
2883                         goto err;
2884
2885                 /*
2886                  * Don't impose further limits on the size and buffer
2887                  * constraints here, we'll -EINVAL later when IO is
2888                  * submitted if they are wrong.
2889                  */
2890                 ret = -EFAULT;
2891                 if (!iov.iov_base || !iov.iov_len)
2892                         goto err;
2893
2894                 /* arbitrary limit, but we need something */
2895                 if (iov.iov_len > SZ_1G)
2896                         goto err;
2897
2898                 ubuf = (unsigned long) iov.iov_base;
2899                 end = (ubuf + iov.iov_len + PAGE_SIZE - 1) >> PAGE_SHIFT;
2900                 start = ubuf >> PAGE_SHIFT;
2901                 nr_pages = end - start;
2902
2903                 if (ctx->account_mem) {
2904                         ret = io_account_mem(ctx->user, nr_pages);
2905                         if (ret)
2906                                 goto err;
2907                 }
2908
2909                 ret = 0;
2910                 if (!pages || nr_pages > got_pages) {
2911                         kfree(vmas);
2912                         kfree(pages);
2913                         pages = kvmalloc_array(nr_pages, sizeof(struct page *),
2914                                                 GFP_KERNEL);
2915                         vmas = kvmalloc_array(nr_pages,
2916                                         sizeof(struct vm_area_struct *),
2917                                         GFP_KERNEL);
2918                         if (!pages || !vmas) {
2919                                 ret = -ENOMEM;
2920                                 if (ctx->account_mem)
2921                                         io_unaccount_mem(ctx->user, nr_pages);
2922                                 goto err;
2923                         }
2924                         got_pages = nr_pages;
2925                 }
2926
2927                 imu->bvec = kvmalloc_array(nr_pages, sizeof(struct bio_vec),
2928                                                 GFP_KERNEL);
2929                 ret = -ENOMEM;
2930                 if (!imu->bvec) {
2931                         if (ctx->account_mem)
2932                                 io_unaccount_mem(ctx->user, nr_pages);
2933                         goto err;
2934                 }
2935
2936                 ret = 0;
2937                 down_read(&current->mm->mmap_sem);
2938                 pret = get_user_pages(ubuf, nr_pages,
2939                                       FOLL_WRITE | FOLL_LONGTERM,
2940                                       pages, vmas);
2941                 if (pret == nr_pages) {
2942                         /* don't support file backed memory */
2943                         for (j = 0; j < nr_pages; j++) {
2944                                 struct vm_area_struct *vma = vmas[j];
2945
2946                                 if (vma->vm_file &&
2947                                     !is_file_hugepages(vma->vm_file)) {
2948                                         ret = -EOPNOTSUPP;
2949                                         break;
2950                                 }
2951                         }
2952                 } else {
2953                         ret = pret < 0 ? pret : -EFAULT;
2954                 }
2955                 up_read(&current->mm->mmap_sem);
2956                 if (ret) {
2957                         /*
2958                          * if we did partial map, or found file backed vmas,
2959                          * release any pages we did get
2960                          */
2961                         if (pret > 0) {
2962                                 for (j = 0; j < pret; j++)
2963                                         put_page(pages[j]);
2964                         }
2965                         if (ctx->account_mem)
2966                                 io_unaccount_mem(ctx->user, nr_pages);
2967                         kvfree(imu->bvec);
2968                         goto err;
2969                 }
2970
2971                 off = ubuf & ~PAGE_MASK;
2972                 size = iov.iov_len;
2973                 for (j = 0; j < nr_pages; j++) {
2974                         size_t vec_len;
2975
2976                         vec_len = min_t(size_t, size, PAGE_SIZE - off);
2977                         imu->bvec[j].bv_page = pages[j];
2978                         imu->bvec[j].bv_len = vec_len;
2979                         imu->bvec[j].bv_offset = off;
2980                         off = 0;
2981                         size -= vec_len;
2982                 }
2983                 /* store original address for later verification */
2984                 imu->ubuf = ubuf;
2985                 imu->len = iov.iov_len;
2986                 imu->nr_bvecs = nr_pages;
2987
2988                 ctx->nr_user_bufs++;
2989         }
2990         kvfree(pages);
2991         kvfree(vmas);
2992         return 0;
2993 err:
2994         kvfree(pages);
2995         kvfree(vmas);
2996         io_sqe_buffer_unregister(ctx);
2997         return ret;
2998 }
2999
3000 static int io_eventfd_register(struct io_ring_ctx *ctx, void __user *arg)
3001 {
3002         __s32 __user *fds = arg;
3003         int fd;
3004
3005         if (ctx->cq_ev_fd)
3006                 return -EBUSY;
3007
3008         if (copy_from_user(&fd, fds, sizeof(*fds)))
3009                 return -EFAULT;
3010
3011         ctx->cq_ev_fd = eventfd_ctx_fdget(fd);
3012         if (IS_ERR(ctx->cq_ev_fd)) {
3013                 int ret = PTR_ERR(ctx->cq_ev_fd);
3014                 ctx->cq_ev_fd = NULL;
3015                 return ret;
3016         }
3017
3018         return 0;
3019 }
3020
3021 static int io_eventfd_unregister(struct io_ring_ctx *ctx)
3022 {
3023         if (ctx->cq_ev_fd) {
3024                 eventfd_ctx_put(ctx->cq_ev_fd);
3025                 ctx->cq_ev_fd = NULL;
3026                 return 0;
3027         }
3028
3029         return -ENXIO;
3030 }
3031
3032 static void io_ring_ctx_free(struct io_ring_ctx *ctx)
3033 {
3034         io_finish_async(ctx);
3035         if (ctx->sqo_mm)
3036                 mmdrop(ctx->sqo_mm);
3037
3038         io_iopoll_reap_events(ctx);
3039         io_sqe_buffer_unregister(ctx);
3040         io_sqe_files_unregister(ctx);
3041         io_eventfd_unregister(ctx);
3042
3043 #if defined(CONFIG_UNIX)
3044         if (ctx->ring_sock) {
3045                 ctx->ring_sock->file = NULL; /* so that iput() is called */
3046                 sock_release(ctx->ring_sock);
3047         }
3048 #endif
3049
3050         io_mem_free(ctx->sq_ring);
3051         io_mem_free(ctx->sq_sqes);
3052         io_mem_free(ctx->cq_ring);
3053
3054         percpu_ref_exit(&ctx->refs);
3055         if (ctx->account_mem)
3056                 io_unaccount_mem(ctx->user,
3057                                 ring_pages(ctx->sq_entries, ctx->cq_entries));
3058         free_uid(ctx->user);
3059         kfree(ctx);
3060 }
3061
3062 static __poll_t io_uring_poll(struct file *file, poll_table *wait)
3063 {
3064         struct io_ring_ctx *ctx = file->private_data;
3065         __poll_t mask = 0;
3066
3067         poll_wait(file, &ctx->cq_wait, wait);
3068         /*
3069          * synchronizes with barrier from wq_has_sleeper call in
3070          * io_commit_cqring
3071          */
3072         smp_rmb();
3073         if (READ_ONCE(ctx->sq_ring->r.tail) - ctx->cached_sq_head !=
3074             ctx->sq_ring->ring_entries)
3075                 mask |= EPOLLOUT | EPOLLWRNORM;
3076         if (READ_ONCE(ctx->cq_ring->r.head) != ctx->cached_cq_tail)
3077                 mask |= EPOLLIN | EPOLLRDNORM;
3078
3079         return mask;
3080 }
3081
3082 static int io_uring_fasync(int fd, struct file *file, int on)
3083 {
3084         struct io_ring_ctx *ctx = file->private_data;
3085
3086         return fasync_helper(fd, file, on, &ctx->cq_fasync);
3087 }
3088
3089 static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
3090 {
3091         mutex_lock(&ctx->uring_lock);
3092         percpu_ref_kill(&ctx->refs);
3093         mutex_unlock(&ctx->uring_lock);
3094
3095         io_poll_remove_all(ctx);
3096         io_iopoll_reap_events(ctx);
3097         wait_for_completion(&ctx->ctx_done);
3098         io_ring_ctx_free(ctx);
3099 }
3100
3101 static int io_uring_release(struct inode *inode, struct file *file)
3102 {
3103         struct io_ring_ctx *ctx = file->private_data;
3104
3105         file->private_data = NULL;
3106         io_ring_ctx_wait_and_kill(ctx);
3107         return 0;
3108 }
3109
3110 static int io_uring_mmap(struct file *file, struct vm_area_struct *vma)
3111 {
3112         loff_t offset = (loff_t) vma->vm_pgoff << PAGE_SHIFT;
3113         unsigned long sz = vma->vm_end - vma->vm_start;
3114         struct io_ring_ctx *ctx = file->private_data;
3115         unsigned long pfn;
3116         struct page *page;
3117         void *ptr;
3118
3119         switch (offset) {
3120         case IORING_OFF_SQ_RING:
3121                 ptr = ctx->sq_ring;
3122                 break;
3123         case IORING_OFF_SQES:
3124                 ptr = ctx->sq_sqes;
3125                 break;
3126         case IORING_OFF_CQ_RING:
3127                 ptr = ctx->cq_ring;
3128                 break;
3129         default:
3130                 return -EINVAL;
3131         }
3132
3133         page = virt_to_head_page(ptr);
3134         if (sz > (PAGE_SIZE << compound_order(page)))
3135                 return -EINVAL;
3136
3137         pfn = virt_to_phys(ptr) >> PAGE_SHIFT;
3138         return remap_pfn_range(vma, vma->vm_start, pfn, sz, vma->vm_page_prot);
3139 }
3140
3141 SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
3142                 u32, min_complete, u32, flags, const sigset_t __user *, sig,
3143                 size_t, sigsz)
3144 {
3145         struct io_ring_ctx *ctx;
3146         long ret = -EBADF;
3147         int submitted = 0;
3148         struct fd f;
3149
3150         if (flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP))
3151                 return -EINVAL;
3152
3153         f = fdget(fd);
3154         if (!f.file)
3155                 return -EBADF;
3156
3157         ret = -EOPNOTSUPP;
3158         if (f.file->f_op != &io_uring_fops)
3159                 goto out_fput;
3160
3161         ret = -ENXIO;
3162         ctx = f.file->private_data;
3163         if (!percpu_ref_tryget(&ctx->refs))
3164                 goto out_fput;
3165
3166         /*
3167          * For SQ polling, the thread will do all submissions and completions.
3168          * Just return the requested submit count, and wake the thread if
3169          * we were asked to.
3170          */
3171         if (ctx->flags & IORING_SETUP_SQPOLL) {
3172                 if (flags & IORING_ENTER_SQ_WAKEUP)
3173                         wake_up(&ctx->sqo_wait);
3174                 submitted = to_submit;
3175                 goto out_ctx;
3176         }
3177
3178         ret = 0;
3179         if (to_submit) {
3180                 to_submit = min(to_submit, ctx->sq_entries);
3181
3182                 mutex_lock(&ctx->uring_lock);
3183                 submitted = io_ring_submit(ctx, to_submit);
3184                 mutex_unlock(&ctx->uring_lock);
3185         }
3186         if (flags & IORING_ENTER_GETEVENTS) {
3187                 unsigned nr_events = 0;
3188
3189                 min_complete = min(min_complete, ctx->cq_entries);
3190
3191                 if (ctx->flags & IORING_SETUP_IOPOLL) {
3192                         mutex_lock(&ctx->uring_lock);
3193                         ret = io_iopoll_check(ctx, &nr_events, min_complete);
3194                         mutex_unlock(&ctx->uring_lock);
3195                 } else {
3196                         ret = io_cqring_wait(ctx, min_complete, sig, sigsz);
3197                 }
3198         }
3199
3200 out_ctx:
3201         io_ring_drop_ctx_refs(ctx, 1);
3202 out_fput:
3203         fdput(f);
3204         return submitted ? submitted : ret;
3205 }
3206
3207 static const struct file_operations io_uring_fops = {
3208         .release        = io_uring_release,
3209         .mmap           = io_uring_mmap,
3210         .poll           = io_uring_poll,
3211         .fasync         = io_uring_fasync,
3212 };
3213
3214 static int io_allocate_scq_urings(struct io_ring_ctx *ctx,
3215                                   struct io_uring_params *p)
3216 {
3217         struct io_sq_ring *sq_ring;
3218         struct io_cq_ring *cq_ring;
3219         size_t size;
3220
3221         sq_ring = io_mem_alloc(struct_size(sq_ring, array, p->sq_entries));
3222         if (!sq_ring)
3223                 return -ENOMEM;
3224
3225         ctx->sq_ring = sq_ring;
3226         sq_ring->ring_mask = p->sq_entries - 1;
3227         sq_ring->ring_entries = p->sq_entries;
3228         ctx->sq_mask = sq_ring->ring_mask;
3229         ctx->sq_entries = sq_ring->ring_entries;
3230
3231         size = array_size(sizeof(struct io_uring_sqe), p->sq_entries);
3232         if (size == SIZE_MAX)
3233                 return -EOVERFLOW;
3234
3235         ctx->sq_sqes = io_mem_alloc(size);
3236         if (!ctx->sq_sqes)
3237                 return -ENOMEM;
3238
3239         cq_ring = io_mem_alloc(struct_size(cq_ring, cqes, p->cq_entries));
3240         if (!cq_ring)
3241                 return -ENOMEM;
3242
3243         ctx->cq_ring = cq_ring;
3244         cq_ring->ring_mask = p->cq_entries - 1;
3245         cq_ring->ring_entries = p->cq_entries;
3246         ctx->cq_mask = cq_ring->ring_mask;
3247         ctx->cq_entries = cq_ring->ring_entries;
3248         return 0;
3249 }
3250
3251 /*
3252  * Allocate an anonymous fd, this is what constitutes the application
3253  * visible backing of an io_uring instance. The application mmaps this
3254  * fd to gain access to the SQ/CQ ring details. If UNIX sockets are enabled,
3255  * we have to tie this fd to a socket for file garbage collection purposes.
3256  */
3257 static int io_uring_get_fd(struct io_ring_ctx *ctx)
3258 {
3259         struct file *file;
3260         int ret;
3261
3262 #if defined(CONFIG_UNIX)
3263         ret = sock_create_kern(&init_net, PF_UNIX, SOCK_RAW, IPPROTO_IP,
3264                                 &ctx->ring_sock);
3265         if (ret)
3266                 return ret;
3267 #endif
3268
3269         ret = get_unused_fd_flags(O_RDWR | O_CLOEXEC);
3270         if (ret < 0)
3271                 goto err;
3272
3273         file = anon_inode_getfile("[io_uring]", &io_uring_fops, ctx,
3274                                         O_RDWR | O_CLOEXEC);
3275         if (IS_ERR(file)) {
3276                 put_unused_fd(ret);
3277                 ret = PTR_ERR(file);
3278                 goto err;
3279         }
3280
3281 #if defined(CONFIG_UNIX)
3282         ctx->ring_sock->file = file;
3283         ctx->ring_sock->sk->sk_user_data = ctx;
3284 #endif
3285         fd_install(ret, file);
3286         return ret;
3287 err:
3288 #if defined(CONFIG_UNIX)
3289         sock_release(ctx->ring_sock);
3290         ctx->ring_sock = NULL;
3291 #endif
3292         return ret;
3293 }
3294
3295 static int io_uring_create(unsigned entries, struct io_uring_params *p)
3296 {
3297         struct user_struct *user = NULL;
3298         struct io_ring_ctx *ctx;
3299         bool account_mem;
3300         int ret;
3301
3302         if (!entries || entries > IORING_MAX_ENTRIES)
3303                 return -EINVAL;
3304
3305         /*
3306          * Use twice as many entries for the CQ ring. It's possible for the
3307          * application to drive a higher depth than the size of the SQ ring,
3308          * since the sqes are only used at submission time. This allows for
3309          * some flexibility in overcommitting a bit.
3310          */
3311         p->sq_entries = roundup_pow_of_two(entries);
3312         p->cq_entries = 2 * p->sq_entries;
3313
3314         user = get_uid(current_user());
3315         account_mem = !capable(CAP_IPC_LOCK);
3316
3317         if (account_mem) {
3318                 ret = io_account_mem(user,
3319                                 ring_pages(p->sq_entries, p->cq_entries));
3320                 if (ret) {
3321                         free_uid(user);
3322                         return ret;
3323                 }
3324         }
3325
3326         ctx = io_ring_ctx_alloc(p);
3327         if (!ctx) {
3328                 if (account_mem)
3329                         io_unaccount_mem(user, ring_pages(p->sq_entries,
3330                                                                 p->cq_entries));
3331                 free_uid(user);
3332                 return -ENOMEM;
3333         }
3334         ctx->compat = in_compat_syscall();
3335         ctx->account_mem = account_mem;
3336         ctx->user = user;
3337
3338         ret = io_allocate_scq_urings(ctx, p);
3339         if (ret)
3340                 goto err;
3341
3342         ret = io_sq_offload_start(ctx, p);
3343         if (ret)
3344                 goto err;
3345
3346         ret = io_uring_get_fd(ctx);
3347         if (ret < 0)
3348                 goto err;
3349
3350         memset(&p->sq_off, 0, sizeof(p->sq_off));
3351         p->sq_off.head = offsetof(struct io_sq_ring, r.head);
3352         p->sq_off.tail = offsetof(struct io_sq_ring, r.tail);
3353         p->sq_off.ring_mask = offsetof(struct io_sq_ring, ring_mask);
3354         p->sq_off.ring_entries = offsetof(struct io_sq_ring, ring_entries);
3355         p->sq_off.flags = offsetof(struct io_sq_ring, flags);
3356         p->sq_off.dropped = offsetof(struct io_sq_ring, dropped);
3357         p->sq_off.array = offsetof(struct io_sq_ring, array);
3358
3359         memset(&p->cq_off, 0, sizeof(p->cq_off));
3360         p->cq_off.head = offsetof(struct io_cq_ring, r.head);
3361         p->cq_off.tail = offsetof(struct io_cq_ring, r.tail);
3362         p->cq_off.ring_mask = offsetof(struct io_cq_ring, ring_mask);
3363         p->cq_off.ring_entries = offsetof(struct io_cq_ring, ring_entries);
3364         p->cq_off.overflow = offsetof(struct io_cq_ring, overflow);
3365         p->cq_off.cqes = offsetof(struct io_cq_ring, cqes);
3366         return ret;
3367 err:
3368         io_ring_ctx_wait_and_kill(ctx);
3369         return ret;
3370 }
3371
3372 /*
3373  * Sets up an aio uring context, and returns the fd. Applications asks for a
3374  * ring size, we return the actual sq/cq ring sizes (among other things) in the
3375  * params structure passed in.
3376  */
3377 static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
3378 {
3379         struct io_uring_params p;
3380         long ret;
3381         int i;
3382
3383         if (copy_from_user(&p, params, sizeof(p)))
3384                 return -EFAULT;
3385         for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
3386                 if (p.resv[i])
3387                         return -EINVAL;
3388         }
3389
3390         if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
3391                         IORING_SETUP_SQ_AFF))
3392                 return -EINVAL;
3393
3394         ret = io_uring_create(entries, &p);
3395         if (ret < 0)
3396                 return ret;
3397
3398         if (copy_to_user(params, &p, sizeof(p)))
3399                 return -EFAULT;
3400
3401         return ret;
3402 }
3403
3404 SYSCALL_DEFINE2(io_uring_setup, u32, entries,
3405                 struct io_uring_params __user *, params)
3406 {
3407         return io_uring_setup(entries, params);
3408 }
3409
3410 static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
3411                                void __user *arg, unsigned nr_args)
3412         __releases(ctx->uring_lock)
3413         __acquires(ctx->uring_lock)
3414 {
3415         int ret;
3416
3417         /*
3418          * We're inside the ring mutex, if the ref is already dying, then
3419          * someone else killed the ctx or is already going through
3420          * io_uring_register().
3421          */
3422         if (percpu_ref_is_dying(&ctx->refs))
3423                 return -ENXIO;
3424
3425         percpu_ref_kill(&ctx->refs);
3426
3427         /*
3428          * Drop uring mutex before waiting for references to exit. If another
3429          * thread is currently inside io_uring_enter() it might need to grab
3430          * the uring_lock to make progress. If we hold it here across the drain
3431          * wait, then we can deadlock. It's safe to drop the mutex here, since
3432          * no new references will come in after we've killed the percpu ref.
3433          */
3434         mutex_unlock(&ctx->uring_lock);
3435         wait_for_completion(&ctx->ctx_done);
3436         mutex_lock(&ctx->uring_lock);
3437
3438         switch (opcode) {
3439         case IORING_REGISTER_BUFFERS:
3440                 ret = io_sqe_buffer_register(ctx, arg, nr_args);
3441                 break;
3442         case IORING_UNREGISTER_BUFFERS:
3443                 ret = -EINVAL;
3444                 if (arg || nr_args)
3445                         break;
3446                 ret = io_sqe_buffer_unregister(ctx);
3447                 break;
3448         case IORING_REGISTER_FILES:
3449                 ret = io_sqe_files_register(ctx, arg, nr_args);
3450                 break;
3451         case IORING_UNREGISTER_FILES:
3452                 ret = -EINVAL;
3453                 if (arg || nr_args)
3454                         break;
3455                 ret = io_sqe_files_unregister(ctx);
3456                 break;
3457         case IORING_REGISTER_EVENTFD:
3458                 ret = -EINVAL;
3459                 if (nr_args != 1)
3460                         break;
3461                 ret = io_eventfd_register(ctx, arg);
3462                 break;
3463         case IORING_UNREGISTER_EVENTFD:
3464                 ret = -EINVAL;
3465                 if (arg || nr_args)
3466                         break;
3467                 ret = io_eventfd_unregister(ctx);
3468                 break;
3469         default:
3470                 ret = -EINVAL;
3471                 break;
3472         }
3473
3474         /* bring the ctx back to life */
3475         reinit_completion(&ctx->ctx_done);
3476         percpu_ref_reinit(&ctx->refs);
3477         return ret;
3478 }
3479
3480 SYSCALL_DEFINE4(io_uring_register, unsigned int, fd, unsigned int, opcode,
3481                 void __user *, arg, unsigned int, nr_args)
3482 {
3483         struct io_ring_ctx *ctx;
3484         long ret = -EBADF;
3485         struct fd f;
3486
3487         f = fdget(fd);
3488         if (!f.file)
3489                 return -EBADF;
3490
3491         ret = -EOPNOTSUPP;
3492         if (f.file->f_op != &io_uring_fops)
3493                 goto out_fput;
3494
3495         ctx = f.file->private_data;
3496
3497         mutex_lock(&ctx->uring_lock);
3498         ret = __io_uring_register(ctx, opcode, arg, nr_args);
3499         mutex_unlock(&ctx->uring_lock);
3500 out_fput:
3501         fdput(f);
3502         return ret;
3503 }
3504
3505 static int __init io_uring_init(void)
3506 {
3507         req_cachep = KMEM_CACHE(io_kiocb, SLAB_HWCACHE_ALIGN | SLAB_PANIC);
3508         return 0;
3509 };
3510 __initcall(io_uring_init);