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