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