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