fs/io_uring.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 /*
   3  * Shared application/kernel submission and completion ring pairs, for
   4  * supporting fast/efficient IO.
   5  *
   6  * A note on the read/write ordering memory barriers that are matched between
   7  * the application and kernel side.
   8  *
   9  * After the application reads the CQ ring tail, it must use an
  10  * appropriate smp_rmb() to pair with the smp_wmb() the kernel uses
  11  * before writing the tail (using smp_load_acquire to read the tail will
  12  * do). It also needs a smp_mb() before updating CQ head (ordering the
  13  * entry load(s) with the head store), pairing with an implicit barrier
  14  * through a control-dependency in io_get_cqring (smp_store_release to
  15  * store head will do). Failure to do so could lead to reading invalid
  16  * CQ entries.
  17  *
  18  * Likewise, the application must use an appropriate smp_wmb() before
  19  * writing the SQ tail (ordering SQ entry stores with the tail store),
  20  * which pairs with smp_load_acquire in io_get_sqring (smp_store_release
  21  * to store the tail will do). And it needs a barrier ordering the SQ
  22  * head load before writing new SQ entries (smp_load_acquire to read
  23  * head will do).
  24  *
  25  * When using the SQ poll thread (IORING_SETUP_SQPOLL), the application
  26  * needs to check the SQ flags for IORING_SQ_NEED_WAKEUP *after*
  27  * updating the SQ tail; a full memory barrier smp_mb() is needed
  28  * between.
  29  *
  30  * Also see the examples in the liburing library:
  31  *
  32  *      git://git.kernel.dk/liburing
  33  *
  34  * io_uring also uses READ/WRITE_ONCE() for _any_ store or load that happens
  35  * from data shared between the kernel and application. This is done both
  36  * for ordering purposes, but also to ensure that once a value is loaded from
  37  * data that the application could potentially modify, it remains stable.
  38  *
  39  * Copyright (C) 2018-2019 Jens Axboe
  40  * Copyright (c) 2018-2019 Christoph Hellwig
  41  */
  42 #include <linux/kernel.h>
  43 #include <linux/init.h>
  44 #include <linux/errno.h>
  45 #include <linux/syscalls.h>
  46 #include <linux/compat.h>
  47 #include <linux/refcount.h>
  48 #include <linux/uio.h>
  49
  50 #include <linux/sched/signal.h>
  51 #include <linux/fs.h>
  52 #include <linux/file.h>
  53 #include <linux/fdtable.h>
  54 #include <linux/mm.h>
  55 #include <linux/mman.h>
  56 #include <linux/mmu_context.h>
  57 #include <linux/percpu.h>
  58 #include <linux/slab.h>
  59 #include <linux/kthread.h>
  60 #include <linux/blkdev.h>
  61 #include <linux/bvec.h>
  62 #include <linux/net.h>
  63 #include <net/sock.h>
  64 #include <net/af_unix.h>
  65 #include <net/scm.h>
  66 #include <linux/anon_inodes.h>
  67 #include <linux/sched/mm.h>
  68 #include <linux/uaccess.h>
  69 #include <linux/nospec.h>
  70 #include <linux/sizes.h>
  71 #include <linux/hugetlb.h>
  72 #include <linux/highmem.h>
  73
  74 #define CREATE_TRACE_POINTS
  75 #include <trace/events/io_uring.h>
  76
  77 #include <uapi/linux/io_uring.h>
  78
  79 #include "internal.h"
  80 #include "io-wq.h"
  81
  82 #define IORING_MAX_ENTRIES      32768
  83 #define IORING_MAX_CQ_ENTRIES   (2 * IORING_MAX_ENTRIES)
  84
  85 /*
  86  * Shift of 9 is 512 entries, or exactly one page on 64-bit archs
  87  */
  88 #define IORING_FILE_TABLE_SHIFT 9
  89 #define IORING_MAX_FILES_TABLE  (1U << IORING_FILE_TABLE_SHIFT)
  90 #define IORING_FILE_TABLE_MASK  (IORING_MAX_FILES_TABLE - 1)
  91 #define IORING_MAX_FIXED_FILES  (64 * IORING_MAX_FILES_TABLE)
  92
  93 struct io_uring {
  94         u32 head ____cacheline_aligned_in_smp;
  95         u32 tail ____cacheline_aligned_in_smp;
  96 };
  97
  98 /*
  99  * This data is shared with the application through the mmap at offsets
 100  * IORING_OFF_SQ_RING and IORING_OFF_CQ_RING.
 101  *
 102  * The offsets to the member fields are published through struct
 103  * io_sqring_offsets when calling io_uring_setup.
 104  */
 105 struct io_rings {
 106         /*
 107          * Head and tail offsets into the ring; the offsets need to be
 108          * masked to get valid indices.
 109          *
 110          * The kernel controls head of the sq ring and the tail of the cq ring,
 111          * and the application controls tail of the sq ring and the head of the
 112          * cq ring.
 113          */
 114         struct io_uring         sq, cq;
 115         /*
 116          * Bitmasks to apply to head and tail offsets (constant, equals
 117          * ring_entries - 1)
 118          */
 119         u32                     sq_ring_mask, cq_ring_mask;
 120         /* Ring sizes (constant, power of 2) */
 121         u32                     sq_ring_entries, cq_ring_entries;
 122         /*
 123          * Number of invalid entries dropped by the kernel due to
 124          * invalid index stored in array
 125          *
 126          * Written by the kernel, shouldn't be modified by the
 127          * application (i.e. get number of "new events" by comparing to
 128          * cached value).
 129          *
 130          * After a new SQ head value was read by the application this
 131          * counter includes all submissions that were dropped reaching
 132          * the new SQ head (and possibly more).
 133          */
 134         u32                     sq_dropped;
 135         /*
 136          * Runtime flags
 137          *
 138          * Written by the kernel, shouldn't be modified by the
 139          * application.
 140          *
 141          * The application needs a full memory barrier before checking
 142          * for IORING_SQ_NEED_WAKEUP after updating the sq tail.
 143          */
 144         u32                     sq_flags;
 145         /*
 146          * Number of completion events lost because the queue was full;
 147          * this should be avoided by the application by making sure
 148          * there are not more requests pending than there is space in
 149          * the completion queue.
 150          *
 151          * Written by the kernel, shouldn't be modified by the
 152          * application (i.e. get number of "new events" by comparing to
 153          * cached value).
 154          *
 155          * As completion events come in out of order this counter is not
 156          * ordered with any other data.
 157          */
 158         u32                     cq_overflow;
 159         /*
 160          * Ring buffer of completion events.
 161          *
 162          * The kernel writes completion events fresh every time they are
 163          * produced, so the application is allowed to modify pending
 164          * entries.
 165          */
 166         struct io_uring_cqe     cqes[] ____cacheline_aligned_in_smp;
 167 };
 168
 169 struct io_mapped_ubuf {
 170         u64             ubuf;
 171         size_t          len;
 172         struct          bio_vec *bvec;
 173         unsigned int    nr_bvecs;
 174 };
 175
 176 struct fixed_file_table {
 177         struct file             **files;
 178 };
 179
 180 struct io_ring_ctx {
 181         struct {
 182                 struct percpu_ref       refs;
 183         } ____cacheline_aligned_in_smp;
 184
 185         struct {
 186                 unsigned int            flags;
 187                 bool                    compat;
 188                 bool                    account_mem;
 189                 bool                    cq_overflow_flushed;
 190                 bool                    drain_next;
 191
 192                 /*
 193                  * Ring buffer of indices into array of io_uring_sqe, which is
 194                  * mmapped by the application using the IORING_OFF_SQES offset.
 195                  *
 196                  * This indirection could e.g. be used to assign fixed
 197                  * io_uring_sqe entries to operations and only submit them to
 198                  * the queue when needed.
 199                  *
 200                  * The kernel modifies neither the indices array nor the entries
 201                  * array.
 202                  */
 203                 u32                     *sq_array;
 204                 unsigned                cached_sq_head;
 205                 unsigned                sq_entries;
 206                 unsigned                sq_mask;
 207                 unsigned                sq_thread_idle;
 208                 unsigned                cached_sq_dropped;
 209                 atomic_t                cached_cq_overflow;
 210                 struct io_uring_sqe     *sq_sqes;
 211
 212                 struct list_head        defer_list;
 213                 struct list_head        timeout_list;
 214                 struct list_head        cq_overflow_list;
 215
 216                 wait_queue_head_t       inflight_wait;
 217         } ____cacheline_aligned_in_smp;
 218
 219         struct io_rings *rings;
 220
 221         /* IO offload */
 222         struct io_wq            *io_wq;
 223         struct task_struct      *sqo_thread;    /* if using sq thread polling */
 224         struct mm_struct        *sqo_mm;
 225         wait_queue_head_t       sqo_wait;
 226
 227         /*
 228          * If used, fixed file set. Writers must ensure that ->refs is dead,
 229          * readers must ensure that ->refs is alive as long as the file* is
 230          * used. Only updated through io_uring_register(2).
 231          */
 232         struct fixed_file_table *file_table;
 233         unsigned                nr_user_files;
 234
 235         /* if used, fixed mapped user buffers */
 236         unsigned                nr_user_bufs;
 237         struct io_mapped_ubuf   *user_bufs;
 238
 239         struct user_struct      *user;
 240
 241         const struct cred       *creds;
 242
 243         /* 0 is for ctx quiesce/reinit/free, 1 is for sqo_thread started */
 244         struct completion       *completions;
 245
 246         /* if all else fails... */
 247         struct io_kiocb         *fallback_req;
 248
 249 #if defined(CONFIG_UNIX)
 250         struct socket           *ring_sock;
 251 #endif
 252
 253         struct {
 254                 unsigned                cached_cq_tail;
 255                 unsigned                cq_entries;
 256                 unsigned                cq_mask;
 257                 atomic_t                cq_timeouts;
 258                 struct wait_queue_head  cq_wait;
 259                 struct fasync_struct    *cq_fasync;
 260                 struct eventfd_ctx      *cq_ev_fd;
 261         } ____cacheline_aligned_in_smp;
 262
 263         struct {
 264                 struct mutex            uring_lock;
 265                 wait_queue_head_t       wait;
 266         } ____cacheline_aligned_in_smp;
 267
 268         struct {
 269                 spinlock_t              completion_lock;
 270                 bool                    poll_multi_file;
 271                 /*
 272                  * ->poll_list is protected by the ctx->uring_lock for
 273                  * io_uring instances that don't use IORING_SETUP_SQPOLL.
 274                  * For SQPOLL, only the single threaded io_sq_thread() will
 275                  * manipulate the list, hence no extra locking is needed there.
 276                  */
 277                 struct list_head        poll_list;
 278                 struct hlist_head       *cancel_hash;
 279                 unsigned                cancel_hash_bits;
 280
 281                 spinlock_t              inflight_lock;
 282                 struct list_head        inflight_list;
 283         } ____cacheline_aligned_in_smp;
 284 };
 285
 286 /*
 287  * First field must be the file pointer in all the
 288  * iocb unions! See also 'struct kiocb' in <linux/fs.h>
 289  */
 290 struct io_poll_iocb {
 291         struct file                     *file;
 292         union {
 293                 struct wait_queue_head  *head;
 294                 u64                     addr;
 295         };
 296         __poll_t                        events;
 297         bool                            done;
 298         bool                            canceled;
 299         struct wait_queue_entry         wait;
 300 };
 301
 302 struct io_timeout_data {
 303         struct io_kiocb                 *req;
 304         struct hrtimer                  timer;
 305         struct timespec64               ts;
 306         enum hrtimer_mode               mode;
 307         u32                             seq_offset;
 308 };
 309
 310 struct io_accept {
 311         struct file                     *file;
 312         struct sockaddr __user          *addr;
 313         int __user                      *addr_len;
 314         int                             flags;
 315 };
 316
 317 struct io_sync {
 318         struct file                     *file;
 319         loff_t                          len;
 320         loff_t                          off;
 321         int                             flags;
 322 };
 323
 324 struct io_cancel {
 325         struct file                     *file;
 326         u64                             addr;
 327 };
 328
 329 struct io_timeout {
 330         struct file                     *file;
 331         u64                             addr;
 332         int                             flags;
 333         unsigned                        count;
 334 };
 335
 336 struct io_rw {
 337         /* NOTE: kiocb has the file as the first member, so don't do it here */
 338         struct kiocb                    kiocb;
 339         u64                             addr;
 340         u64                             len;
 341 };
 342
 343 struct io_connect {
 344         struct file                     *file;
 345         struct sockaddr __user          *addr;
 346         int                             addr_len;
 347 };
 348
 349 struct io_sr_msg {
 350         struct file                     *file;
 351         struct user_msghdr __user       *msg;
 352         int                             msg_flags;
 353 };
 354
 355 struct io_async_connect {
 356         struct sockaddr_storage         address;
 357 };
 358
 359 struct io_async_msghdr {
 360         struct iovec                    fast_iov[UIO_FASTIOV];
 361         struct iovec                    *iov;
 362         struct sockaddr __user          *uaddr;
 363         struct msghdr                   msg;
 364 };
 365
 366 struct io_async_rw {
 367         struct iovec                    fast_iov[UIO_FASTIOV];
 368         struct iovec                    *iov;
 369         ssize_t                         nr_segs;
 370         ssize_t                         size;
 371 };
 372
 373 struct io_async_ctx {
 374         union {
 375                 struct io_async_rw      rw;
 376                 struct io_async_msghdr  msg;
 377                 struct io_async_connect connect;
 378                 struct io_timeout_data  timeout;
 379         };
 380 };
 381
 382 /*
 383  * NOTE! Each of the iocb union members has the file pointer
 384  * as the first entry in their struct definition. So you can
 385  * access the file pointer through any of the sub-structs,
 386  * or directly as just 'ki_filp' in this struct.
 387  */
 388 struct io_kiocb {
 389         union {
 390                 struct file             *file;
 391                 struct io_rw            rw;
 392                 struct io_poll_iocb     poll;
 393                 struct io_accept        accept;
 394                 struct io_sync          sync;
 395                 struct io_cancel        cancel;
 396                 struct io_timeout       timeout;
 397                 struct io_connect       connect;
 398                 struct io_sr_msg        sr_msg;
 399         };
 400
 401         struct io_async_ctx             *io;
 402         struct file                     *ring_file;
 403         int                             ring_fd;
 404         bool                            has_user;
 405         bool                            in_async;
 406         bool                            needs_fixed_file;
 407         u8                              opcode;
 408
 409         struct io_ring_ctx      *ctx;
 410         union {
 411                 struct list_head        list;
 412                 struct hlist_node       hash_node;
 413         };
 414         struct list_head        link_list;
 415         unsigned int            flags;
 416         refcount_t              refs;
 417 #define REQ_F_NOWAIT            1       /* must not punt to workers */
 418 #define REQ_F_IOPOLL_COMPLETED  2       /* polled IO has completed */
 419 #define REQ_F_FIXED_FILE        4       /* ctx owns file */
 420 #define REQ_F_LINK_NEXT         8       /* already grabbed next link */
 421 #define REQ_F_IO_DRAIN          16      /* drain existing IO first */
 422 #define REQ_F_IO_DRAINED        32      /* drain done */
 423 #define REQ_F_LINK              64      /* linked sqes */
 424 #define REQ_F_LINK_TIMEOUT      128     /* has linked timeout */
 425 #define REQ_F_FAIL_LINK         256     /* fail rest of links */
 426 #define REQ_F_DRAIN_LINK        512     /* link should be fully drained */
 427 #define REQ_F_TIMEOUT           1024    /* timeout request */
 428 #define REQ_F_ISREG             2048    /* regular file */
 429 #define REQ_F_MUST_PUNT         4096    /* must be punted even for NONBLOCK */
 430 #define REQ_F_TIMEOUT_NOSEQ     8192    /* no timeout sequence */
 431 #define REQ_F_INFLIGHT          16384   /* on inflight list */
 432 #define REQ_F_COMP_LOCKED       32768   /* completion under lock */
 433 #define REQ_F_HARDLINK          65536   /* doesn't sever on completion < 0 */
 434         u64                     user_data;
 435         u32                     result;
 436         u32                     sequence;
 437
 438         struct list_head        inflight_entry;
 439
 440         struct io_wq_work       work;
 441 };
 442
 443 #define IO_PLUG_THRESHOLD               2
 444 #define IO_IOPOLL_BATCH                 8
 445
 446 struct io_submit_state {
 447         struct blk_plug         plug;
 448
 449         /*
 450          * io_kiocb alloc cache
 451          */
 452         void                    *reqs[IO_IOPOLL_BATCH];
 453         unsigned                int free_reqs;
 454         unsigned                int cur_req;
 455
 456         /*
 457          * File reference cache
 458          */
 459         struct file             *file;
 460         unsigned int            fd;
 461         unsigned int            has_refs;
 462         unsigned int            used_refs;
 463         unsigned int            ios_left;
 464 };
 465
 466 static void io_wq_submit_work(struct io_wq_work **workptr);
 467 static void io_cqring_fill_event(struct io_kiocb *req, long res);
 468 static void __io_free_req(struct io_kiocb *req);
 469 static void io_put_req(struct io_kiocb *req);
 470 static void io_double_put_req(struct io_kiocb *req);
 471 static void __io_double_put_req(struct io_kiocb *req);
 472 static struct io_kiocb *io_prep_linked_timeout(struct io_kiocb *req);
 473 static void io_queue_linked_timeout(struct io_kiocb *req);
 474
 475 static struct kmem_cache *req_cachep;
 476
 477 static const struct file_operations io_uring_fops;
 478
 479 struct sock *io_uring_get_socket(struct file *file)
 480 {
 481 #if defined(CONFIG_UNIX)
 482         if (file->f_op == &io_uring_fops) {
 483                 struct io_ring_ctx *ctx = file->private_data;
 484
 485                 return ctx->ring_sock->sk;
 486         }
 487 #endif
 488         return NULL;
 489 }
 490 EXPORT_SYMBOL(io_uring_get_socket);
 491
 492 static void io_ring_ctx_ref_free(struct percpu_ref *ref)
 493 {
 494         struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);
 495
 496         complete(&ctx->completions[0]);
 497 }
 498
 499 static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
 500 {
 501         struct io_ring_ctx *ctx;
 502         int hash_bits;
 503
 504         ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
 505         if (!ctx)
 506                 return NULL;
 507
 508         ctx->fallback_req = kmem_cache_alloc(req_cachep, GFP_KERNEL);
 509         if (!ctx->fallback_req)
 510                 goto err;
 511
 512         ctx->completions = kmalloc(2 * sizeof(struct completion), GFP_KERNEL);
 513         if (!ctx->completions)
 514                 goto err;
 515
 516         /*
 517          * Use 5 bits less than the max cq entries, that should give us around
 518          * 32 entries per hash list if totally full and uniformly spread.
 519          */
 520         hash_bits = ilog2(p->cq_entries);
 521         hash_bits -= 5;
 522         if (hash_bits <= 0)
 523                 hash_bits = 1;
 524         ctx->cancel_hash_bits = hash_bits;
 525         ctx->cancel_hash = kmalloc((1U << hash_bits) * sizeof(struct hlist_head),
 526                                         GFP_KERNEL);
 527         if (!ctx->cancel_hash)
 528                 goto err;
 529         __hash_init(ctx->cancel_hash, 1U << hash_bits);
 530
 531         if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free,
 532                             PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
 533                 goto err;
 534
 535         ctx->flags = p->flags;
 536         init_waitqueue_head(&ctx->cq_wait);
 537         INIT_LIST_HEAD(&ctx->cq_overflow_list);
 538         init_completion(&ctx->completions[0]);
 539         init_completion(&ctx->completions[1]);
 540         mutex_init(&ctx->uring_lock);
 541         init_waitqueue_head(&ctx->wait);
 542         spin_lock_init(&ctx->completion_lock);
 543         INIT_LIST_HEAD(&ctx->poll_list);
 544         INIT_LIST_HEAD(&ctx->defer_list);
 545         INIT_LIST_HEAD(&ctx->timeout_list);
 546         init_waitqueue_head(&ctx->inflight_wait);
 547         spin_lock_init(&ctx->inflight_lock);
 548         INIT_LIST_HEAD(&ctx->inflight_list);
 549         return ctx;
 550 err:
 551         if (ctx->fallback_req)
 552                 kmem_cache_free(req_cachep, ctx->fallback_req);
 553         kfree(ctx->completions);
 554         kfree(ctx->cancel_hash);
 555         kfree(ctx);
 556         return NULL;
 557 }
 558
 559 static inline bool __req_need_defer(struct io_kiocb *req)
 560 {
 561         struct io_ring_ctx *ctx = req->ctx;
 562
 563         return req->sequence != ctx->cached_cq_tail + ctx->cached_sq_dropped
 564                                         + atomic_read(&ctx->cached_cq_overflow);
 565 }
 566
 567 static inline bool req_need_defer(struct io_kiocb *req)
 568 {
 569         if ((req->flags & (REQ_F_IO_DRAIN|REQ_F_IO_DRAINED)) == REQ_F_IO_DRAIN)
 570                 return __req_need_defer(req);
 571
 572         return false;
 573 }
 574
 575 static struct io_kiocb *io_get_deferred_req(struct io_ring_ctx *ctx)
 576 {
 577         struct io_kiocb *req;
 578
 579         req = list_first_entry_or_null(&ctx->defer_list, struct io_kiocb, list);
 580         if (req && !req_need_defer(req)) {
 581                 list_del_init(&req->list);
 582                 return req;
 583         }
 584
 585         return NULL;
 586 }
 587
 588 static struct io_kiocb *io_get_timeout_req(struct io_ring_ctx *ctx)
 589 {
 590         struct io_kiocb *req;
 591
 592         req = list_first_entry_or_null(&ctx->timeout_list, struct io_kiocb, list);
 593         if (req) {
 594                 if (req->flags & REQ_F_TIMEOUT_NOSEQ)
 595                         return NULL;
 596                 if (!__req_need_defer(req)) {
 597                         list_del_init(&req->list);
 598                         return req;
 599                 }
 600         }
 601
 602         return NULL;
 603 }
 604
 605 static void __io_commit_cqring(struct io_ring_ctx *ctx)
 606 {
 607         struct io_rings *rings = ctx->rings;
 608
 609         if (ctx->cached_cq_tail != READ_ONCE(rings->cq.tail)) {
 610                 /* order cqe stores with ring update */
 611                 smp_store_release(&rings->cq.tail, ctx->cached_cq_tail);
 612
 613                 if (wq_has_sleeper(&ctx->cq_wait)) {
 614                         wake_up_interruptible(&ctx->cq_wait);
 615                         kill_fasync(&ctx->cq_fasync, SIGIO, POLL_IN);
 616                 }
 617         }
 618 }
 619
 620 static inline bool io_req_needs_user(struct io_kiocb *req)
 621 {
 622         return !(req->opcode == IORING_OP_READ_FIXED ||
 623                  req->opcode == IORING_OP_WRITE_FIXED);
 624 }
 625
 626 static inline bool io_prep_async_work(struct io_kiocb *req,
 627                                       struct io_kiocb **link)
 628 {
 629         bool do_hashed = false;
 630
 631         switch (req->opcode) {
 632         case IORING_OP_WRITEV:
 633         case IORING_OP_WRITE_FIXED:
 634                 /* only regular files should be hashed for writes */
 635                 if (req->flags & REQ_F_ISREG)
 636                         do_hashed = true;
 637                 /* fall-through */
 638         case IORING_OP_READV:
 639         case IORING_OP_READ_FIXED:
 640         case IORING_OP_SENDMSG:
 641         case IORING_OP_RECVMSG:
 642         case IORING_OP_ACCEPT:
 643         case IORING_OP_POLL_ADD:
 644         case IORING_OP_CONNECT:
 645                 /*
 646                  * We know REQ_F_ISREG is not set on some of these
 647                  * opcodes, but this enables us to keep the check in
 648                  * just one place.
 649                  */
 650                 if (!(req->flags & REQ_F_ISREG))
 651                         req->work.flags |= IO_WQ_WORK_UNBOUND;
 652                 break;
 653         }
 654         if (io_req_needs_user(req))
 655                 req->work.flags |= IO_WQ_WORK_NEEDS_USER;
 656
 657         *link = io_prep_linked_timeout(req);
 658         return do_hashed;
 659 }
 660
 661 static inline void io_queue_async_work(struct io_kiocb *req)
 662 {
 663         struct io_ring_ctx *ctx = req->ctx;
 664         struct io_kiocb *link;
 665         bool do_hashed;
 666
 667         do_hashed = io_prep_async_work(req, &link);
 668
 669         trace_io_uring_queue_async_work(ctx, do_hashed, req, &req->work,
 670                                         req->flags);
 671         if (!do_hashed) {
 672                 io_wq_enqueue(ctx->io_wq, &req->work);
 673         } else {
 674                 io_wq_enqueue_hashed(ctx->io_wq, &req->work,
 675                                         file_inode(req->file));
 676         }
 677
 678         if (link)
 679                 io_queue_linked_timeout(link);
 680 }
 681
 682 static void io_kill_timeout(struct io_kiocb *req)
 683 {
 684         int ret;
 685
 686         ret = hrtimer_try_to_cancel(&req->io->timeout.timer);
 687         if (ret != -1) {
 688                 atomic_inc(&req->ctx->cq_timeouts);
 689                 list_del_init(&req->list);
 690                 io_cqring_fill_event(req, 0);
 691                 io_put_req(req);
 692         }
 693 }
 694
 695 static void io_kill_timeouts(struct io_ring_ctx *ctx)
 696 {
 697         struct io_kiocb *req, *tmp;
 698
 699         spin_lock_irq(&ctx->completion_lock);
 700         list_for_each_entry_safe(req, tmp, &ctx->timeout_list, list)
 701                 io_kill_timeout(req);
 702         spin_unlock_irq(&ctx->completion_lock);
 703 }
 704
 705 static void io_commit_cqring(struct io_ring_ctx *ctx)
 706 {
 707         struct io_kiocb *req;
 708
 709         while ((req = io_get_timeout_req(ctx)) != NULL)
 710                 io_kill_timeout(req);
 711
 712         __io_commit_cqring(ctx);
 713
 714         while ((req = io_get_deferred_req(ctx)) != NULL) {
 715                 req->flags |= REQ_F_IO_DRAINED;
 716                 io_queue_async_work(req);
 717         }
 718 }
 719
 720 static struct io_uring_cqe *io_get_cqring(struct io_ring_ctx *ctx)
 721 {
 722         struct io_rings *rings = ctx->rings;
 723         unsigned tail;
 724
 725         tail = ctx->cached_cq_tail;
 726         /*
 727          * writes to the cq entry need to come after reading head; the
 728          * control dependency is enough as we're using WRITE_ONCE to
 729          * fill the cq entry
 730          */
 731         if (tail - READ_ONCE(rings->cq.head) == rings->cq_ring_entries)
 732                 return NULL;
 733
 734         ctx->cached_cq_tail++;
 735         return &rings->cqes[tail & ctx->cq_mask];
 736 }
 737
 738 static void io_cqring_ev_posted(struct io_ring_ctx *ctx)
 739 {
 740         if (waitqueue_active(&ctx->wait))
 741                 wake_up(&ctx->wait);
 742         if (waitqueue_active(&ctx->sqo_wait))
 743                 wake_up(&ctx->sqo_wait);
 744         if (ctx->cq_ev_fd)
 745                 eventfd_signal(ctx->cq_ev_fd, 1);
 746 }
 747
 748 /* Returns true if there are no backlogged entries after the flush */
 749 static bool io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force)
 750 {
 751         struct io_rings *rings = ctx->rings;
 752         struct io_uring_cqe *cqe;
 753         struct io_kiocb *req;
 754         unsigned long flags;
 755         LIST_HEAD(list);
 756
 757         if (!force) {
 758                 if (list_empty_careful(&ctx->cq_overflow_list))
 759                         return true;
 760                 if ((ctx->cached_cq_tail - READ_ONCE(rings->cq.head) ==
 761                     rings->cq_ring_entries))
 762                         return false;
 763         }
 764
 765         spin_lock_irqsave(&ctx->completion_lock, flags);
 766
 767         /* if force is set, the ring is going away. always drop after that */
 768         if (force)
 769                 ctx->cq_overflow_flushed = true;
 770
 771         cqe = NULL;
 772         while (!list_empty(&ctx->cq_overflow_list)) {
 773                 cqe = io_get_cqring(ctx);
 774                 if (!cqe && !force)
 775                         break;
 776
 777                 req = list_first_entry(&ctx->cq_overflow_list, struct io_kiocb,
 778                                                 list);
 779                 list_move(&req->list, &list);
 780                 if (cqe) {
 781                         WRITE_ONCE(cqe->user_data, req->user_data);
 782                         WRITE_ONCE(cqe->res, req->result);
 783                         WRITE_ONCE(cqe->flags, 0);
 784                 } else {
 785                         WRITE_ONCE(ctx->rings->cq_overflow,
 786                                 atomic_inc_return(&ctx->cached_cq_overflow));
 787                 }
 788         }
 789
 790         io_commit_cqring(ctx);
 791         spin_unlock_irqrestore(&ctx->completion_lock, flags);
 792         io_cqring_ev_posted(ctx);
 793
 794         while (!list_empty(&list)) {
 795                 req = list_first_entry(&list, struct io_kiocb, list);
 796                 list_del(&req->list);
 797                 io_put_req(req);
 798         }
 799
 800         return cqe != NULL;
 801 }
 802
 803 static void io_cqring_fill_event(struct io_kiocb *req, long res)
 804 {
 805         struct io_ring_ctx *ctx = req->ctx;
 806         struct io_uring_cqe *cqe;
 807
 808         trace_io_uring_complete(ctx, req->user_data, res);
 809
 810         /*
 811          * If we can't get a cq entry, userspace overflowed the
 812          * submission (by quite a lot). Increment the overflow count in
 813          * the ring.
 814          */
 815         cqe = io_get_cqring(ctx);
 816         if (likely(cqe)) {
 817                 WRITE_ONCE(cqe->user_data, req->user_data);
 818                 WRITE_ONCE(cqe->res, res);
 819                 WRITE_ONCE(cqe->flags, 0);
 820         } else if (ctx->cq_overflow_flushed) {
 821                 WRITE_ONCE(ctx->rings->cq_overflow,
 822                                 atomic_inc_return(&ctx->cached_cq_overflow));
 823         } else {
 824                 refcount_inc(&req->refs);
 825                 req->result = res;
 826                 list_add_tail(&req->list, &ctx->cq_overflow_list);
 827         }
 828 }
 829
 830 static void io_cqring_add_event(struct io_kiocb *req, long res)
 831 {
 832         struct io_ring_ctx *ctx = req->ctx;
 833         unsigned long flags;
 834
 835         spin_lock_irqsave(&ctx->completion_lock, flags);
 836         io_cqring_fill_event(req, res);
 837         io_commit_cqring(ctx);
 838         spin_unlock_irqrestore(&ctx->completion_lock, flags);
 839
 840         io_cqring_ev_posted(ctx);
 841 }
 842
 843 static inline bool io_is_fallback_req(struct io_kiocb *req)
 844 {
 845         return req == (struct io_kiocb *)
 846                         ((unsigned long) req->ctx->fallback_req & ~1UL);
 847 }
 848
 849 static struct io_kiocb *io_get_fallback_req(struct io_ring_ctx *ctx)
 850 {
 851         struct io_kiocb *req;
 852
 853         req = ctx->fallback_req;
 854         if (!test_and_set_bit_lock(0, (unsigned long *) ctx->fallback_req))
 855                 return req;
 856
 857         return NULL;
 858 }
 859
 860 static struct io_kiocb *io_get_req(struct io_ring_ctx *ctx,
 861                                    struct io_submit_state *state)
 862 {
 863         gfp_t gfp = GFP_KERNEL | __GFP_NOWARN;
 864         struct io_kiocb *req;
 865
 866         if (!percpu_ref_tryget(&ctx->refs))
 867                 return NULL;
 868
 869         if (!state) {
 870                 req = kmem_cache_alloc(req_cachep, gfp);
 871                 if (unlikely(!req))
 872                         goto fallback;
 873         } else if (!state->free_reqs) {
 874                 size_t sz;
 875                 int ret;
 876
 877                 sz = min_t(size_t, state->ios_left, ARRAY_SIZE(state->reqs));
 878                 ret = kmem_cache_alloc_bulk(req_cachep, gfp, sz, state->reqs);
 879
 880                 /*
 881                  * Bulk alloc is all-or-nothing. If we fail to get a batch,
 882                  * retry single alloc to be on the safe side.
 883                  */
 884                 if (unlikely(ret <= 0)) {
 885                         state->reqs[0] = kmem_cache_alloc(req_cachep, gfp);
 886                         if (!state->reqs[0])
 887                                 goto fallback;
 888                         ret = 1;
 889                 }
 890                 state->free_reqs = ret - 1;
 891                 state->cur_req = 1;
 892                 req = state->reqs[0];
 893         } else {
 894                 req = state->reqs[state->cur_req];
 895                 state->free_reqs--;
 896                 state->cur_req++;
 897         }
 898
 899 got_it:
 900         req->io = NULL;
 901         req->ring_file = NULL;
 902         req->file = NULL;
 903         req->ctx = ctx;
 904         req->flags = 0;
 905         /* one is dropped after submission, the other at completion */
 906         refcount_set(&req->refs, 2);
 907         req->result = 0;
 908         INIT_IO_WORK(&req->work, io_wq_submit_work);
 909         return req;
 910 fallback:
 911         req = io_get_fallback_req(ctx);
 912         if (req)
 913                 goto got_it;
 914         percpu_ref_put(&ctx->refs);
 915         return NULL;
 916 }
 917
 918 static void io_free_req_many(struct io_ring_ctx *ctx, void **reqs, int *nr)
 919 {
 920         if (*nr) {
 921                 kmem_cache_free_bulk(req_cachep, *nr, reqs);
 922                 percpu_ref_put_many(&ctx->refs, *nr);
 923                 *nr = 0;
 924         }
 925 }
 926
 927 static void __io_free_req(struct io_kiocb *req)
 928 {
 929         struct io_ring_ctx *ctx = req->ctx;
 930
 931         if (req->io)
 932                 kfree(req->io);
 933         if (req->file && !(req->flags & REQ_F_FIXED_FILE))
 934                 fput(req->file);
 935         if (req->flags & REQ_F_INFLIGHT) {
 936                 unsigned long flags;
 937
 938                 spin_lock_irqsave(&ctx->inflight_lock, flags);
 939                 list_del(&req->inflight_entry);
 940                 if (waitqueue_active(&ctx->inflight_wait))
 941                         wake_up(&ctx->inflight_wait);
 942                 spin_unlock_irqrestore(&ctx->inflight_lock, flags);
 943         }
 944         percpu_ref_put(&ctx->refs);
 945         if (likely(!io_is_fallback_req(req)))
 946                 kmem_cache_free(req_cachep, req);
 947         else
 948                 clear_bit_unlock(0, (unsigned long *) ctx->fallback_req);
 949 }
 950
 951 static bool io_link_cancel_timeout(struct io_kiocb *req)
 952 {
 953         struct io_ring_ctx *ctx = req->ctx;
 954         int ret;
 955
 956         ret = hrtimer_try_to_cancel(&req->io->timeout.timer);
 957         if (ret != -1) {
 958                 io_cqring_fill_event(req, -ECANCELED);
 959                 io_commit_cqring(ctx);
 960                 req->flags &= ~REQ_F_LINK;
 961                 io_put_req(req);
 962                 return true;
 963         }
 964
 965         return false;
 966 }
 967
 968 static void io_req_link_next(struct io_kiocb *req, struct io_kiocb **nxtptr)
 969 {
 970         struct io_ring_ctx *ctx = req->ctx;
 971         bool wake_ev = false;
 972
 973         /* Already got next link */
 974         if (req->flags & REQ_F_LINK_NEXT)
 975                 return;
 976
 977         /*
 978          * The list should never be empty when we are called here. But could
 979          * potentially happen if the chain is messed up, check to be on the
 980          * safe side.
 981          */
 982         while (!list_empty(&req->link_list)) {
 983                 struct io_kiocb *nxt = list_first_entry(&req->link_list,
 984                                                 struct io_kiocb, link_list);
 985
 986                 if (unlikely((req->flags & REQ_F_LINK_TIMEOUT) &&
 987                              (nxt->flags & REQ_F_TIMEOUT))) {
 988                         list_del_init(&nxt->link_list);
 989                         wake_ev |= io_link_cancel_timeout(nxt);
 990                         req->flags &= ~REQ_F_LINK_TIMEOUT;
 991                         continue;
 992                 }
 993
 994                 list_del_init(&req->link_list);
 995                 if (!list_empty(&nxt->link_list))
 996                         nxt->flags |= REQ_F_LINK;
 997                 *nxtptr = nxt;
 998                 break;
 999         }
1000
1001         req->flags |= REQ_F_LINK_NEXT;
1002         if (wake_ev)
1003                 io_cqring_ev_posted(ctx);
1004 }
1005
1006 /*
1007  * Called if REQ_F_LINK is set, and we fail the head request
1008  */
1009 static void io_fail_links(struct io_kiocb *req)
1010 {
1011         struct io_ring_ctx *ctx = req->ctx;
1012         unsigned long flags;
1013
1014         spin_lock_irqsave(&ctx->completion_lock, flags);
1015
1016         while (!list_empty(&req->link_list)) {
1017                 struct io_kiocb *link = list_first_entry(&req->link_list,
1018                                                 struct io_kiocb, link_list);
1019
1020                 list_del_init(&link->link_list);
1021                 trace_io_uring_fail_link(req, link);
1022
1023                 if ((req->flags & REQ_F_LINK_TIMEOUT) &&
1024                     link->opcode == IORING_OP_LINK_TIMEOUT) {
1025                         io_link_cancel_timeout(link);
1026                 } else {
1027                         io_cqring_fill_event(link, -ECANCELED);
1028                         __io_double_put_req(link);
1029                 }
1030                 req->flags &= ~REQ_F_LINK_TIMEOUT;
1031         }
1032
1033         io_commit_cqring(ctx);
1034         spin_unlock_irqrestore(&ctx->completion_lock, flags);
1035         io_cqring_ev_posted(ctx);
1036 }
1037
1038 static void io_req_find_next(struct io_kiocb *req, struct io_kiocb **nxt)
1039 {
1040         if (likely(!(req->flags & REQ_F_LINK)))
1041                 return;
1042
1043         /*
1044          * If LINK is set, we have dependent requests in this chain. If we
1045          * didn't fail this request, queue the first one up, moving any other
1046          * dependencies to the next request. In case of failure, fail the rest
1047          * of the chain.
1048          */
1049         if (req->flags & REQ_F_FAIL_LINK) {
1050                 io_fail_links(req);
1051         } else if ((req->flags & (REQ_F_LINK_TIMEOUT | REQ_F_COMP_LOCKED)) ==
1052                         REQ_F_LINK_TIMEOUT) {
1053                 struct io_ring_ctx *ctx = req->ctx;
1054                 unsigned long flags;
1055
1056                 /*
1057                  * If this is a timeout link, we could be racing with the
1058                  * timeout timer. Grab the completion lock for this case to
1059                  * protect against that.
1060                  */
1061                 spin_lock_irqsave(&ctx->completion_lock, flags);
1062                 io_req_link_next(req, nxt);
1063                 spin_unlock_irqrestore(&ctx->completion_lock, flags);
1064         } else {
1065                 io_req_link_next(req, nxt);
1066         }
1067 }
1068
1069 static void io_free_req(struct io_kiocb *req)
1070 {
1071         struct io_kiocb *nxt = NULL;
1072
1073         io_req_find_next(req, &nxt);
1074         __io_free_req(req);
1075
1076         if (nxt)
1077                 io_queue_async_work(nxt);
1078 }
1079
1080 /*
1081  * Drop reference to request, return next in chain (if there is one) if this
1082  * was the last reference to this request.
1083  */
1084 __attribute__((nonnull))
1085 static void io_put_req_find_next(struct io_kiocb *req, struct io_kiocb **nxtptr)
1086 {
1087         io_req_find_next(req, nxtptr);
1088
1089         if (refcount_dec_and_test(&req->refs))
1090                 __io_free_req(req);
1091 }
1092
1093 static void io_put_req(struct io_kiocb *req)
1094 {
1095         if (refcount_dec_and_test(&req->refs))
1096                 io_free_req(req);
1097 }
1098
1099 /*
1100  * Must only be used if we don't need to care about links, usually from
1101  * within the completion handling itself.
1102  */
1103 static void __io_double_put_req(struct io_kiocb *req)
1104 {
1105         /* drop both submit and complete references */
1106         if (refcount_sub_and_test(2, &req->refs))
1107                 __io_free_req(req);
1108 }
1109
1110 static void io_double_put_req(struct io_kiocb *req)
1111 {
1112         /* drop both submit and complete references */
1113         if (refcount_sub_and_test(2, &req->refs))
1114                 io_free_req(req);
1115 }
1116
1117 static unsigned io_cqring_events(struct io_ring_ctx *ctx, bool noflush)
1118 {
1119         struct io_rings *rings = ctx->rings;
1120
1121         /*
1122          * noflush == true is from the waitqueue handler, just ensure we wake
1123          * up the task, and the next invocation will flush the entries. We
1124          * cannot safely to it from here.
1125          */
1126         if (noflush && !list_empty(&ctx->cq_overflow_list))
1127                 return -1U;
1128
1129         io_cqring_overflow_flush(ctx, false);
1130
1131         /* See comment at the top of this file */
1132         smp_rmb();
1133         return READ_ONCE(rings->cq.tail) - READ_ONCE(rings->cq.head);
1134 }
1135
1136 static inline unsigned int io_sqring_entries(struct io_ring_ctx *ctx)
1137 {
1138         struct io_rings *rings = ctx->rings;
1139
1140         /* make sure SQ entry isn't read before tail */
1141         return smp_load_acquire(&rings->sq.tail) - ctx->cached_sq_head;
1142 }
1143
1144 /*
1145  * Find and free completed poll iocbs
1146  */
1147 static void io_iopoll_complete(struct io_ring_ctx *ctx, unsigned int *nr_events,
1148                                struct list_head *done)
1149 {
1150         void *reqs[IO_IOPOLL_BATCH];
1151         struct io_kiocb *req;
1152         int to_free;
1153
1154         to_free = 0;
1155         while (!list_empty(done)) {
1156                 req = list_first_entry(done, struct io_kiocb, list);
1157                 list_del(&req->list);
1158
1159                 io_cqring_fill_event(req, req->result);
1160                 (*nr_events)++;
1161
1162                 if (refcount_dec_and_test(&req->refs)) {
1163                         /* If we're not using fixed files, we have to pair the
1164                          * completion part with the file put. Use regular
1165                          * completions for those, only batch free for fixed
1166                          * file and non-linked commands.
1167                          */
1168                         if (((req->flags & (REQ_F_FIXED_FILE|REQ_F_LINK)) ==
1169                             REQ_F_FIXED_FILE) && !io_is_fallback_req(req) &&
1170                             !req->io) {
1171                                 reqs[to_free++] = req;
1172                                 if (to_free == ARRAY_SIZE(reqs))
1173                                         io_free_req_many(ctx, reqs, &to_free);
1174                         } else {
1175                                 io_free_req(req);
1176                         }
1177                 }
1178         }
1179
1180         io_commit_cqring(ctx);
1181         io_free_req_many(ctx, reqs, &to_free);
1182 }
1183
1184 static int io_do_iopoll(struct io_ring_ctx *ctx, unsigned int *nr_events,
1185                         long min)
1186 {
1187         struct io_kiocb *req, *tmp;
1188         LIST_HEAD(done);
1189         bool spin;
1190         int ret;
1191
1192         /*
1193          * Only spin for completions if we don't have multiple devices hanging
1194          * off our complete list, and we're under the requested amount.
1195          */
1196         spin = !ctx->poll_multi_file && *nr_events < min;
1197
1198         ret = 0;
1199         list_for_each_entry_safe(req, tmp, &ctx->poll_list, list) {
1200                 struct kiocb *kiocb = &req->rw.kiocb;
1201
1202                 /*
1203                  * Move completed entries to our local list. If we find a
1204                  * request that requires polling, break out and complete
1205                  * the done list first, if we have entries there.
1206                  */
1207                 if (req->flags & REQ_F_IOPOLL_COMPLETED) {
1208                         list_move_tail(&req->list, &done);
1209                         continue;
1210                 }
1211                 if (!list_empty(&done))
1212                         break;
1213
1214                 ret = kiocb->ki_filp->f_op->iopoll(kiocb, spin);
1215                 if (ret < 0)
1216                         break;
1217
1218                 if (ret && spin)
1219                         spin = false;
1220                 ret = 0;
1221         }
1222
1223         if (!list_empty(&done))
1224                 io_iopoll_complete(ctx, nr_events, &done);
1225
1226         return ret;
1227 }
1228
1229 /*
1230  * Poll for a minimum of 'min' events. Note that if min == 0 we consider that a
1231  * non-spinning poll check - we'll still enter the driver poll loop, but only
1232  * as a non-spinning completion check.
1233  */
1234 static int io_iopoll_getevents(struct io_ring_ctx *ctx, unsigned int *nr_events,
1235                                 long min)
1236 {
1237         while (!list_empty(&ctx->poll_list) && !need_resched()) {
1238                 int ret;
1239
1240                 ret = io_do_iopoll(ctx, nr_events, min);
1241                 if (ret < 0)
1242                         return ret;
1243                 if (!min || *nr_events >= min)
1244                         return 0;
1245         }
1246
1247         return 1;
1248 }
1249
1250 /*
1251  * We can't just wait for polled events to come to us, we have to actively
1252  * find and complete them.
1253  */
1254 static void io_iopoll_reap_events(struct io_ring_ctx *ctx)
1255 {
1256         if (!(ctx->flags & IORING_SETUP_IOPOLL))
1257                 return;
1258
1259         mutex_lock(&ctx->uring_lock);
1260         while (!list_empty(&ctx->poll_list)) {
1261                 unsigned int nr_events = 0;
1262
1263                 io_iopoll_getevents(ctx, &nr_events, 1);
1264
1265                 /*
1266                  * Ensure we allow local-to-the-cpu processing to take place,
1267                  * in this case we need to ensure that we reap all events.
1268                  */
1269                 cond_resched();
1270         }
1271         mutex_unlock(&ctx->uring_lock);
1272 }
1273
1274 static int __io_iopoll_check(struct io_ring_ctx *ctx, unsigned *nr_events,
1275                             long min)
1276 {
1277         int iters = 0, ret = 0;
1278
1279         do {
1280                 int tmin = 0;
1281
1282                 /*
1283                  * Don't enter poll loop if we already have events pending.
1284                  * If we do, we can potentially be spinning for commands that
1285                  * already triggered a CQE (eg in error).
1286                  */
1287                 if (io_cqring_events(ctx, false))
1288                         break;
1289
1290                 /*
1291                  * If a submit got punted to a workqueue, we can have the
1292                  * application entering polling for a command before it gets
1293                  * issued. That app will hold the uring_lock for the duration
1294                  * of the poll right here, so we need to take a breather every
1295                  * now and then to ensure that the issue has a chance to add
1296                  * the poll to the issued list. Otherwise we can spin here
1297                  * forever, while the workqueue is stuck trying to acquire the
1298                  * very same mutex.
1299                  */
1300                 if (!(++iters & 7)) {
1301                         mutex_unlock(&ctx->uring_lock);
1302                         mutex_lock(&ctx->uring_lock);
1303                 }
1304
1305                 if (*nr_events < min)
1306                         tmin = min - *nr_events;
1307
1308                 ret = io_iopoll_getevents(ctx, nr_events, tmin);
1309                 if (ret <= 0)
1310                         break;
1311                 ret = 0;
1312         } while (min && !*nr_events && !need_resched());
1313
1314         return ret;
1315 }
1316
1317 static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned *nr_events,
1318                            long min)
1319 {
1320         int ret;
1321
1322         /*
1323          * We disallow the app entering submit/complete with polling, but we
1324          * still need to lock the ring to prevent racing with polled issue
1325          * that got punted to a workqueue.
1326          */
1327         mutex_lock(&ctx->uring_lock);
1328         ret = __io_iopoll_check(ctx, nr_events, min);
1329         mutex_unlock(&ctx->uring_lock);
1330         return ret;
1331 }
1332
1333 static void kiocb_end_write(struct io_kiocb *req)
1334 {
1335         /*
1336          * Tell lockdep we inherited freeze protection from submission
1337          * thread.
1338          */
1339         if (req->flags & REQ_F_ISREG) {
1340                 struct inode *inode = file_inode(req->file);
1341
1342                 __sb_writers_acquired(inode->i_sb, SB_FREEZE_WRITE);
1343         }
1344         file_end_write(req->file);
1345 }
1346
1347 static inline void req_set_fail_links(struct io_kiocb *req)
1348 {
1349         if ((req->flags & (REQ_F_LINK | REQ_F_HARDLINK)) == REQ_F_LINK)
1350                 req->flags |= REQ_F_FAIL_LINK;
1351 }
1352
1353 static void io_complete_rw_common(struct kiocb *kiocb, long res)
1354 {
1355         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
1356
1357         if (kiocb->ki_flags & IOCB_WRITE)
1358                 kiocb_end_write(req);
1359
1360         if (res != req->result)
1361                 req_set_fail_links(req);
1362         io_cqring_add_event(req, res);
1363 }
1364
1365 static void io_complete_rw(struct kiocb *kiocb, long res, long res2)
1366 {
1367         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
1368
1369         io_complete_rw_common(kiocb, res);
1370         io_put_req(req);
1371 }
1372
1373 static struct io_kiocb *__io_complete_rw(struct kiocb *kiocb, long res)
1374 {
1375         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
1376         struct io_kiocb *nxt = NULL;
1377
1378         io_complete_rw_common(kiocb, res);
1379         io_put_req_find_next(req, &nxt);
1380
1381         return nxt;
1382 }
1383
1384 static void io_complete_rw_iopoll(struct kiocb *kiocb, long res, long res2)
1385 {
1386         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
1387
1388         if (kiocb->ki_flags & IOCB_WRITE)
1389                 kiocb_end_write(req);
1390
1391         if (res != req->result)
1392                 req_set_fail_links(req);
1393         req->result = res;
1394         if (res != -EAGAIN)
1395                 req->flags |= REQ_F_IOPOLL_COMPLETED;
1396 }
1397
1398 /*
1399  * After the iocb has been issued, it's safe to be found on the poll list.
1400  * Adding the kiocb to the list AFTER submission ensures that we don't
1401  * find it from a io_iopoll_getevents() thread before the issuer is done
1402  * accessing the kiocb cookie.
1403  */
1404 static void io_iopoll_req_issued(struct io_kiocb *req)
1405 {
1406         struct io_ring_ctx *ctx = req->ctx;
1407
1408         /*
1409          * Track whether we have multiple files in our lists. This will impact
1410          * how we do polling eventually, not spinning if we're on potentially
1411          * different devices.
1412          */
1413         if (list_empty(&ctx->poll_list)) {
1414                 ctx->poll_multi_file = false;
1415         } else if (!ctx->poll_multi_file) {
1416                 struct io_kiocb *list_req;
1417
1418                 list_req = list_first_entry(&ctx->poll_list, struct io_kiocb,
1419                                                 list);
1420                 if (list_req->file != req->file)
1421                         ctx->poll_multi_file = true;
1422         }
1423
1424         /*
1425          * For fast devices, IO may have already completed. If it has, add
1426          * it to the front so we find it first.
1427          */
1428         if (req->flags & REQ_F_IOPOLL_COMPLETED)
1429                 list_add(&req->list, &ctx->poll_list);
1430         else
1431                 list_add_tail(&req->list, &ctx->poll_list);
1432 }
1433
1434 static void io_file_put(struct io_submit_state *state)
1435 {
1436         if (state->file) {
1437                 int diff = state->has_refs - state->used_refs;
1438
1439                 if (diff)
1440                         fput_many(state->file, diff);
1441                 state->file = NULL;
1442         }
1443 }
1444
1445 /*
1446  * Get as many references to a file as we have IOs left in this submission,
1447  * assuming most submissions are for one file, or at least that each file
1448  * has more than one submission.
1449  */
1450 static struct file *io_file_get(struct io_submit_state *state, int fd)
1451 {
1452         if (!state)
1453                 return fget(fd);
1454
1455         if (state->file) {
1456                 if (state->fd == fd) {
1457                         state->used_refs++;
1458                         state->ios_left--;
1459                         return state->file;
1460                 }
1461                 io_file_put(state);
1462         }
1463         state->file = fget_many(fd, state->ios_left);
1464         if (!state->file)
1465                 return NULL;
1466
1467         state->fd = fd;
1468         state->has_refs = state->ios_left;
1469         state->used_refs = 1;
1470         state->ios_left--;
1471         return state->file;
1472 }
1473
1474 /*
1475  * If we tracked the file through the SCM inflight mechanism, we could support
1476  * any file. For now, just ensure that anything potentially problematic is done
1477  * inline.
1478  */
1479 static bool io_file_supports_async(struct file *file)
1480 {
1481         umode_t mode = file_inode(file)->i_mode;
1482
1483         if (S_ISBLK(mode) || S_ISCHR(mode) || S_ISSOCK(mode))
1484                 return true;
1485         if (S_ISREG(mode) && file->f_op != &io_uring_fops)
1486                 return true;
1487
1488         return false;
1489 }
1490
1491 static int io_prep_rw(struct io_kiocb *req, const struct io_uring_sqe *sqe,
1492                       bool force_nonblock)
1493 {
1494         struct io_ring_ctx *ctx = req->ctx;
1495         struct kiocb *kiocb = &req->rw.kiocb;
1496         unsigned ioprio;
1497         int ret;
1498
1499         if (!req->file)
1500                 return -EBADF;
1501
1502         if (S_ISREG(file_inode(req->file)->i_mode))
1503                 req->flags |= REQ_F_ISREG;
1504
1505         kiocb->ki_pos = READ_ONCE(sqe->off);
1506         kiocb->ki_flags = iocb_flags(kiocb->ki_filp);
1507         kiocb->ki_hint = ki_hint_validate(file_write_hint(kiocb->ki_filp));
1508
1509         ioprio = READ_ONCE(sqe->ioprio);
1510         if (ioprio) {
1511                 ret = ioprio_check_cap(ioprio);
1512                 if (ret)
1513                         return ret;
1514
1515                 kiocb->ki_ioprio = ioprio;
1516         } else
1517                 kiocb->ki_ioprio = get_current_ioprio();
1518
1519         ret = kiocb_set_rw_flags(kiocb, READ_ONCE(sqe->rw_flags));
1520         if (unlikely(ret))
1521                 return ret;
1522
1523         /* don't allow async punt if RWF_NOWAIT was requested */
1524         if ((kiocb->ki_flags & IOCB_NOWAIT) ||
1525             (req->file->f_flags & O_NONBLOCK))
1526                 req->flags |= REQ_F_NOWAIT;
1527
1528         if (force_nonblock)
1529                 kiocb->ki_flags |= IOCB_NOWAIT;
1530
1531         if (ctx->flags & IORING_SETUP_IOPOLL) {
1532                 if (!(kiocb->ki_flags & IOCB_DIRECT) ||
1533                     !kiocb->ki_filp->f_op->iopoll)
1534                         return -EOPNOTSUPP;
1535
1536                 kiocb->ki_flags |= IOCB_HIPRI;
1537                 kiocb->ki_complete = io_complete_rw_iopoll;
1538                 req->result = 0;
1539         } else {
1540                 if (kiocb->ki_flags & IOCB_HIPRI)
1541                         return -EINVAL;
1542                 kiocb->ki_complete = io_complete_rw;
1543         }
1544
1545         req->rw.addr = READ_ONCE(sqe->addr);
1546         req->rw.len = READ_ONCE(sqe->len);
1547         /* we own ->private, reuse it for the buffer index */
1548         req->rw.kiocb.private = (void *) (unsigned long)
1549                                         READ_ONCE(sqe->buf_index);
1550         return 0;
1551 }
1552
1553 static inline void io_rw_done(struct kiocb *kiocb, ssize_t ret)
1554 {
1555         switch (ret) {
1556         case -EIOCBQUEUED:
1557                 break;
1558         case -ERESTARTSYS:
1559         case -ERESTARTNOINTR:
1560         case -ERESTARTNOHAND:
1561         case -ERESTART_RESTARTBLOCK:
1562                 /*
1563                  * We can't just restart the syscall, since previously
1564                  * submitted sqes may already be in progress. Just fail this
1565                  * IO with EINTR.
1566                  */
1567                 ret = -EINTR;
1568                 /* fall through */
1569         default:
1570                 kiocb->ki_complete(kiocb, ret, 0);
1571         }
1572 }
1573
1574 static void kiocb_done(struct kiocb *kiocb, ssize_t ret, struct io_kiocb **nxt,
1575                        bool in_async)
1576 {
1577         if (in_async && ret >= 0 && kiocb->ki_complete == io_complete_rw)
1578                 *nxt = __io_complete_rw(kiocb, ret);
1579         else
1580                 io_rw_done(kiocb, ret);
1581 }
1582
1583 static ssize_t io_import_fixed(struct io_kiocb *req, int rw,
1584                                struct iov_iter *iter)
1585 {
1586         struct io_ring_ctx *ctx = req->ctx;
1587         size_t len = req->rw.len;
1588         struct io_mapped_ubuf *imu;
1589         unsigned index, buf_index;
1590         size_t offset;
1591         u64 buf_addr;
1592
1593         /* attempt to use fixed buffers without having provided iovecs */
1594         if (unlikely(!ctx->user_bufs))
1595                 return -EFAULT;
1596
1597         buf_index = (unsigned long) req->rw.kiocb.private;
1598         if (unlikely(buf_index >= ctx->nr_user_bufs))
1599                 return -EFAULT;
1600
1601         index = array_index_nospec(buf_index, ctx->nr_user_bufs);
1602         imu = &ctx->user_bufs[index];
1603         buf_addr = req->rw.addr;
1604
1605         /* overflow */
1606         if (buf_addr + len < buf_addr)
1607                 return -EFAULT;
1608         /* not inside the mapped region */
1609         if (buf_addr < imu->ubuf || buf_addr + len > imu->ubuf + imu->len)
1610                 return -EFAULT;
1611
1612         /*
1613          * May not be a start of buffer, set size appropriately
1614          * and advance us to the beginning.
1615          */
1616         offset = buf_addr - imu->ubuf;
1617         iov_iter_bvec(iter, rw, imu->bvec, imu->nr_bvecs, offset + len);
1618
1619         if (offset) {
1620                 /*
1621                  * Don't use iov_iter_advance() here, as it's really slow for
1622                  * using the latter parts of a big fixed buffer - it iterates
1623                  * over each segment manually. We can cheat a bit here, because
1624                  * we know that:
1625                  *
1626                  * 1) it's a BVEC iter, we set it up
1627                  * 2) all bvecs are PAGE_SIZE in size, except potentially the
1628                  *    first and last bvec
1629                  *
1630                  * So just find our index, and adjust the iterator afterwards.
1631                  * If the offset is within the first bvec (or the whole first
1632                  * bvec, just use iov_iter_advance(). This makes it easier
1633                  * since we can just skip the first segment, which may not
1634                  * be PAGE_SIZE aligned.
1635                  */
1636                 const struct bio_vec *bvec = imu->bvec;
1637
1638                 if (offset <= bvec->bv_len) {
1639                         iov_iter_advance(iter, offset);
1640                 } else {
1641                         unsigned long seg_skip;
1642
1643                         /* skip first vec */
1644                         offset -= bvec->bv_len;
1645                         seg_skip = 1 + (offset >> PAGE_SHIFT);
1646
1647                         iter->bvec = bvec + seg_skip;
1648                         iter->nr_segs -= seg_skip;
1649                         iter->count -= bvec->bv_len + offset;
1650                         iter->iov_offset = offset & ~PAGE_MASK;
1651                 }
1652         }
1653
1654         return len;
1655 }
1656
1657 static ssize_t io_import_iovec(int rw, struct io_kiocb *req,
1658                                struct iovec **iovec, struct iov_iter *iter)
1659 {
1660         void __user *buf = u64_to_user_ptr(req->rw.addr);
1661         size_t sqe_len = req->rw.len;
1662         u8 opcode;
1663
1664         opcode = req->opcode;
1665         if (opcode == IORING_OP_READ_FIXED || opcode == IORING_OP_WRITE_FIXED) {
1666                 *iovec = NULL;
1667                 return io_import_fixed(req, rw, iter);
1668         }
1669
1670         /* buffer index only valid with fixed read/write */
1671         if (req->rw.kiocb.private)
1672                 return -EINVAL;
1673
1674         if (req->io) {
1675                 struct io_async_rw *iorw = &req->io->rw;
1676
1677                 *iovec = iorw->iov;
1678                 iov_iter_init(iter, rw, *iovec, iorw->nr_segs, iorw->size);
1679                 if (iorw->iov == iorw->fast_iov)
1680                         *iovec = NULL;
1681                 return iorw->size;
1682         }
1683
1684         if (!req->has_user)
1685                 return -EFAULT;
1686
1687 #ifdef CONFIG_COMPAT
1688         if (req->ctx->compat)
1689                 return compat_import_iovec(rw, buf, sqe_len, UIO_FASTIOV,
1690                                                 iovec, iter);
1691 #endif
1692
1693         return import_iovec(rw, buf, sqe_len, UIO_FASTIOV, iovec, iter);
1694 }
1695
1696 /*
1697  * For files that don't have ->read_iter() and ->write_iter(), handle them
1698  * by looping over ->read() or ->write() manually.
1699  */
1700 static ssize_t loop_rw_iter(int rw, struct file *file, struct kiocb *kiocb,
1701                            struct iov_iter *iter)
1702 {
1703         ssize_t ret = 0;
1704
1705         /*
1706          * Don't support polled IO through this interface, and we can't
1707          * support non-blocking either. For the latter, this just causes
1708          * the kiocb to be handled from an async context.
1709          */
1710         if (kiocb->ki_flags & IOCB_HIPRI)
1711                 return -EOPNOTSUPP;
1712         if (kiocb->ki_flags & IOCB_NOWAIT)
1713                 return -EAGAIN;
1714
1715         while (iov_iter_count(iter)) {
1716                 struct iovec iovec;
1717                 ssize_t nr;
1718
1719                 if (!iov_iter_is_bvec(iter)) {
1720                         iovec = iov_iter_iovec(iter);
1721                 } else {
1722                         /* fixed buffers import bvec */
1723                         iovec.iov_base = kmap(iter->bvec->bv_page)
1724                                                 + iter->iov_offset;
1725                         iovec.iov_len = min(iter->count,
1726                                         iter->bvec->bv_len - iter->iov_offset);
1727                 }
1728
1729                 if (rw == READ) {
1730                         nr = file->f_op->read(file, iovec.iov_base,
1731                                               iovec.iov_len, &kiocb->ki_pos);
1732                 } else {
1733                         nr = file->f_op->write(file, iovec.iov_base,
1734                                                iovec.iov_len, &kiocb->ki_pos);
1735                 }
1736
1737                 if (iov_iter_is_bvec(iter))
1738                         kunmap(iter->bvec->bv_page);
1739
1740                 if (nr < 0) {
1741                         if (!ret)
1742                                 ret = nr;
1743                         break;
1744                 }
1745                 ret += nr;
1746                 if (nr != iovec.iov_len)
1747                         break;
1748                 iov_iter_advance(iter, nr);
1749         }
1750
1751         return ret;
1752 }
1753
1754 static void io_req_map_rw(struct io_kiocb *req, ssize_t io_size,
1755                           struct iovec *iovec, struct iovec *fast_iov,
1756                           struct iov_iter *iter)
1757 {
1758         req->io->rw.nr_segs = iter->nr_segs;
1759         req->io->rw.size = io_size;
1760         req->io->rw.iov = iovec;
1761         if (!req->io->rw.iov) {
1762                 req->io->rw.iov = req->io->rw.fast_iov;
1763                 memcpy(req->io->rw.iov, fast_iov,
1764                         sizeof(struct iovec) * iter->nr_segs);
1765         }
1766 }
1767
1768 static int io_alloc_async_ctx(struct io_kiocb *req)
1769 {
1770         req->io = kmalloc(sizeof(*req->io), GFP_KERNEL);
1771         return req->io == NULL;
1772 }
1773
1774 static void io_rw_async(struct io_wq_work **workptr)
1775 {
1776         struct io_kiocb *req = container_of(*workptr, struct io_kiocb, work);
1777         struct iovec *iov = NULL;
1778
1779         if (req->io->rw.iov != req->io->rw.fast_iov)
1780                 iov = req->io->rw.iov;
1781         io_wq_submit_work(workptr);
1782         kfree(iov);
1783 }
1784
1785 static int io_setup_async_rw(struct io_kiocb *req, ssize_t io_size,
1786                              struct iovec *iovec, struct iovec *fast_iov,
1787                              struct iov_iter *iter)
1788 {
1789         if (req->opcode == IORING_OP_READ_FIXED ||
1790             req->opcode == IORING_OP_WRITE_FIXED)
1791                 return 0;
1792         if (!req->io && io_alloc_async_ctx(req))
1793                 return -ENOMEM;
1794
1795         io_req_map_rw(req, io_size, iovec, fast_iov, iter);
1796         req->work.func = io_rw_async;
1797         return 0;
1798 }
1799
1800 static int io_read_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe,
1801                         bool force_nonblock)
1802 {
1803         struct io_async_ctx *io;
1804         struct iov_iter iter;
1805         ssize_t ret;
1806
1807         ret = io_prep_rw(req, sqe, force_nonblock);
1808         if (ret)
1809                 return ret;
1810
1811         if (unlikely(!(req->file->f_mode & FMODE_READ)))
1812                 return -EBADF;
1813
1814         if (!req->io)
1815                 return 0;
1816
1817         io = req->io;
1818         io->rw.iov = io->rw.fast_iov;
1819         req->io = NULL;
1820         ret = io_import_iovec(READ, req, &io->rw.iov, &iter);
1821         req->io = io;
1822         if (ret < 0)
1823                 return ret;
1824
1825         io_req_map_rw(req, ret, io->rw.iov, io->rw.fast_iov, &iter);
1826         return 0;
1827 }
1828
1829 static int io_read(struct io_kiocb *req, struct io_kiocb **nxt,
1830                    bool force_nonblock)
1831 {
1832         struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
1833         struct kiocb *kiocb = &req->rw.kiocb;
1834         struct iov_iter iter;
1835         size_t iov_count;
1836         ssize_t io_size, ret;
1837
1838         ret = io_import_iovec(READ, req, &iovec, &iter);
1839         if (ret < 0)
1840                 return ret;
1841
1842         /* Ensure we clear previously set non-block flag */
1843         if (!force_nonblock)
1844                 req->rw.kiocb.ki_flags &= ~IOCB_NOWAIT;
1845
1846         req->result = 0;
1847         io_size = ret;
1848         if (req->flags & REQ_F_LINK)
1849                 req->result = io_size;
1850
1851         /*
1852          * If the file doesn't support async, mark it as REQ_F_MUST_PUNT so
1853          * we know to async punt it even if it was opened O_NONBLOCK
1854          */
1855         if (force_nonblock && !io_file_supports_async(req->file)) {
1856                 req->flags |= REQ_F_MUST_PUNT;
1857                 goto copy_iov;
1858         }
1859
1860         iov_count = iov_iter_count(&iter);
1861         ret = rw_verify_area(READ, req->file, &kiocb->ki_pos, iov_count);
1862         if (!ret) {
1863                 ssize_t ret2;
1864
1865                 if (req->file->f_op->read_iter)
1866                         ret2 = call_read_iter(req->file, kiocb, &iter);
1867                 else
1868                         ret2 = loop_rw_iter(READ, req->file, kiocb, &iter);
1869
1870                 /* Catch -EAGAIN return for forced non-blocking submission */
1871                 if (!force_nonblock || ret2 != -EAGAIN) {
1872                         kiocb_done(kiocb, ret2, nxt, req->in_async);
1873                 } else {
1874 copy_iov:
1875                         ret = io_setup_async_rw(req, io_size, iovec,
1876                                                 inline_vecs, &iter);
1877                         if (ret)
1878                                 goto out_free;
1879                         return -EAGAIN;
1880                 }
1881         }
1882 out_free:
1883         if (!io_wq_current_is_worker())
1884                 kfree(iovec);
1885         return ret;
1886 }
1887
1888 static int io_write_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe,
1889                          bool force_nonblock)
1890 {
1891         struct io_async_ctx *io;
1892         struct iov_iter iter;
1893         ssize_t ret;
1894
1895         ret = io_prep_rw(req, sqe, force_nonblock);
1896         if (ret)
1897                 return ret;
1898
1899         if (unlikely(!(req->file->f_mode & FMODE_WRITE)))
1900                 return -EBADF;
1901
1902         if (!req->io)
1903                 return 0;
1904
1905         io = req->io;
1906         io->rw.iov = io->rw.fast_iov;
1907         req->io = NULL;
1908         ret = io_import_iovec(WRITE, req, &io->rw.iov, &iter);
1909         req->io = io;
1910         if (ret < 0)
1911                 return ret;
1912
1913         io_req_map_rw(req, ret, io->rw.iov, io->rw.fast_iov, &iter);
1914         return 0;
1915 }
1916
1917 static int io_write(struct io_kiocb *req, struct io_kiocb **nxt,
1918                     bool force_nonblock)
1919 {
1920         struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
1921         struct kiocb *kiocb = &req->rw.kiocb;
1922         struct iov_iter iter;
1923         size_t iov_count;
1924         ssize_t ret, io_size;
1925
1926         ret = io_import_iovec(WRITE, req, &iovec, &iter);
1927         if (ret < 0)
1928                 return ret;
1929
1930         /* Ensure we clear previously set non-block flag */
1931         if (!force_nonblock)
1932                 req->rw.kiocb.ki_flags &= ~IOCB_NOWAIT;
1933
1934         req->result = 0;
1935         io_size = ret;
1936         if (req->flags & REQ_F_LINK)
1937                 req->result = io_size;
1938
1939         /*
1940          * If the file doesn't support async, mark it as REQ_F_MUST_PUNT so
1941          * we know to async punt it even if it was opened O_NONBLOCK
1942          */
1943         if (force_nonblock && !io_file_supports_async(req->file)) {
1944                 req->flags |= REQ_F_MUST_PUNT;
1945                 goto copy_iov;
1946         }
1947
1948         /* file path doesn't support NOWAIT for non-direct_IO */
1949         if (force_nonblock && !(kiocb->ki_flags & IOCB_DIRECT) &&
1950             (req->flags & REQ_F_ISREG))
1951                 goto copy_iov;
1952
1953         iov_count = iov_iter_count(&iter);
1954         ret = rw_verify_area(WRITE, req->file, &kiocb->ki_pos, iov_count);
1955         if (!ret) {
1956                 ssize_t ret2;
1957
1958                 /*
1959                  * Open-code file_start_write here to grab freeze protection,
1960                  * which will be released by another thread in
1961                  * io_complete_rw().  Fool lockdep by telling it the lock got
1962                  * released so that it doesn't complain about the held lock when
1963                  * we return to userspace.
1964                  */
1965                 if (req->flags & REQ_F_ISREG) {
1966                         __sb_start_write(file_inode(req->file)->i_sb,
1967                                                 SB_FREEZE_WRITE, true);
1968                         __sb_writers_release(file_inode(req->file)->i_sb,
1969                                                 SB_FREEZE_WRITE);
1970                 }
1971                 kiocb->ki_flags |= IOCB_WRITE;
1972
1973                 if (req->file->f_op->write_iter)
1974                         ret2 = call_write_iter(req->file, kiocb, &iter);
1975                 else
1976                         ret2 = loop_rw_iter(WRITE, req->file, kiocb, &iter);
1977                 if (!force_nonblock || ret2 != -EAGAIN) {
1978                         kiocb_done(kiocb, ret2, nxt, req->in_async);
1979                 } else {
1980 copy_iov:
1981                         ret = io_setup_async_rw(req, io_size, iovec,
1982                                                 inline_vecs, &iter);
1983                         if (ret)
1984                                 goto out_free;
1985                         return -EAGAIN;
1986                 }
1987         }
1988 out_free:
1989         if (!io_wq_current_is_worker())
1990                 kfree(iovec);
1991         return ret;
1992 }
1993
1994 /*
1995  * IORING_OP_NOP just posts a completion event, nothing else.
1996  */
1997 static int io_nop(struct io_kiocb *req)
1998 {
1999         struct io_ring_ctx *ctx = req->ctx;
2000
2001         if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
2002                 return -EINVAL;
2003
2004         io_cqring_add_event(req, 0);
2005         io_put_req(req);
2006         return 0;
2007 }
2008
2009 static int io_prep_fsync(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2010 {
2011         struct io_ring_ctx *ctx = req->ctx;
2012
2013         if (!req->file)
2014                 return -EBADF;
2015
2016         if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
2017                 return -EINVAL;
2018         if (unlikely(sqe->addr || sqe->ioprio || sqe->buf_index))
2019                 return -EINVAL;
2020
2021         req->sync.flags = READ_ONCE(sqe->fsync_flags);
2022         if (unlikely(req->sync.flags & ~IORING_FSYNC_DATASYNC))
2023                 return -EINVAL;
2024
2025         req->sync.off = READ_ONCE(sqe->off);
2026         req->sync.len = READ_ONCE(sqe->len);
2027         return 0;
2028 }
2029
2030 static bool io_req_cancelled(struct io_kiocb *req)
2031 {
2032         if (req->work.flags & IO_WQ_WORK_CANCEL) {
2033                 req_set_fail_links(req);
2034                 io_cqring_add_event(req, -ECANCELED);
2035                 io_put_req(req);
2036                 return true;
2037         }
2038
2039         return false;
2040 }
2041
2042 static void io_link_work_cb(struct io_wq_work **workptr)
2043 {
2044         struct io_wq_work *work = *workptr;
2045         struct io_kiocb *link = work->data;
2046
2047         io_queue_linked_timeout(link);
2048         work->func = io_wq_submit_work;
2049 }
2050
2051 static void io_wq_assign_next(struct io_wq_work **workptr, struct io_kiocb *nxt)
2052 {
2053         struct io_kiocb *link;
2054
2055         io_prep_async_work(nxt, &link);
2056         *workptr = &nxt->work;
2057         if (link) {
2058                 nxt->work.flags |= IO_WQ_WORK_CB;
2059                 nxt->work.func = io_link_work_cb;
2060                 nxt->work.data = link;
2061         }
2062 }
2063
2064 static void io_fsync_finish(struct io_wq_work **workptr)
2065 {
2066         struct io_kiocb *req = container_of(*workptr, struct io_kiocb, work);
2067         loff_t end = req->sync.off + req->sync.len;
2068         struct io_kiocb *nxt = NULL;
2069         int ret;
2070
2071         if (io_req_cancelled(req))
2072                 return;
2073
2074         ret = vfs_fsync_range(req->file, req->sync.off,
2075                                 end > 0 ? end : LLONG_MAX,
2076                                 req->sync.flags & IORING_FSYNC_DATASYNC);
2077         if (ret < 0)
2078                 req_set_fail_links(req);
2079         io_cqring_add_event(req, ret);
2080         io_put_req_find_next(req, &nxt);
2081         if (nxt)
2082                 io_wq_assign_next(workptr, nxt);
2083 }
2084
2085 static int io_fsync(struct io_kiocb *req, struct io_kiocb **nxt,
2086                     bool force_nonblock)
2087 {
2088         struct io_wq_work *work, *old_work;
2089
2090         /* fsync always requires a blocking context */
2091         if (force_nonblock) {
2092                 io_put_req(req);
2093                 req->work.func = io_fsync_finish;
2094                 return -EAGAIN;
2095         }
2096
2097         work = old_work = &req->work;
2098         io_fsync_finish(&work);
2099         if (work && work != old_work)
2100                 *nxt = container_of(work, struct io_kiocb, work);
2101         return 0;
2102 }
2103
2104 static int io_prep_sfr(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2105 {
2106         struct io_ring_ctx *ctx = req->ctx;
2107
2108         if (!req->file)
2109                 return -EBADF;
2110
2111         if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
2112                 return -EINVAL;
2113         if (unlikely(sqe->addr || sqe->ioprio || sqe->buf_index))
2114                 return -EINVAL;
2115
2116         req->sync.off = READ_ONCE(sqe->off);
2117         req->sync.len = READ_ONCE(sqe->len);
2118         req->sync.flags = READ_ONCE(sqe->sync_range_flags);
2119         return 0;
2120 }
2121
2122 static void io_sync_file_range_finish(struct io_wq_work **workptr)
2123 {
2124         struct io_kiocb *req = container_of(*workptr, struct io_kiocb, work);
2125         struct io_kiocb *nxt = NULL;
2126         int ret;
2127
2128         if (io_req_cancelled(req))
2129                 return;
2130
2131         ret = sync_file_range(req->file, req->sync.off, req->sync.len,
2132                                 req->sync.flags);
2133         if (ret < 0)
2134                 req_set_fail_links(req);
2135         io_cqring_add_event(req, ret);
2136         io_put_req_find_next(req, &nxt);
2137         if (nxt)
2138                 io_wq_assign_next(workptr, nxt);
2139 }
2140
2141 static int io_sync_file_range(struct io_kiocb *req, struct io_kiocb **nxt,
2142                               bool force_nonblock)
2143 {
2144         struct io_wq_work *work, *old_work;
2145
2146         /* sync_file_range always requires a blocking context */
2147         if (force_nonblock) {
2148                 io_put_req(req);
2149                 req->work.func = io_sync_file_range_finish;
2150                 return -EAGAIN;
2151         }
2152
2153         work = old_work = &req->work;
2154         io_sync_file_range_finish(&work);
2155         if (work && work != old_work)
2156                 *nxt = container_of(work, struct io_kiocb, work);
2157         return 0;
2158 }
2159
2160 #if defined(CONFIG_NET)
2161 static void io_sendrecv_async(struct io_wq_work **workptr)
2162 {
2163         struct io_kiocb *req = container_of(*workptr, struct io_kiocb, work);
2164         struct iovec *iov = NULL;
2165
2166         if (req->io->rw.iov != req->io->rw.fast_iov)
2167                 iov = req->io->msg.iov;
2168         io_wq_submit_work(workptr);
2169         kfree(iov);
2170 }
2171 #endif
2172
2173 static int io_sendmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2174 {
2175 #if defined(CONFIG_NET)
2176         struct io_sr_msg *sr = &req->sr_msg;
2177         struct io_async_ctx *io = req->io;
2178
2179         sr->msg_flags = READ_ONCE(sqe->msg_flags);
2180         sr->msg = u64_to_user_ptr(READ_ONCE(sqe->addr));
2181
2182         if (!io)
2183                 return 0;
2184
2185         io->msg.iov = io->msg.fast_iov;
2186         return sendmsg_copy_msghdr(&io->msg.msg, sr->msg, sr->msg_flags,
2187                                         &io->msg.iov);
2188 #else
2189         return -EOPNOTSUPP;
2190 #endif
2191 }
2192
2193 static int io_sendmsg(struct io_kiocb *req, struct io_kiocb **nxt,
2194                       bool force_nonblock)
2195 {
2196 #if defined(CONFIG_NET)
2197         struct io_async_msghdr *kmsg = NULL;
2198         struct socket *sock;
2199         int ret;
2200
2201         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
2202                 return -EINVAL;
2203
2204         sock = sock_from_file(req->file, &ret);
2205         if (sock) {
2206                 struct io_async_ctx io;
2207                 struct sockaddr_storage addr;
2208                 unsigned flags;
2209
2210                 if (req->io) {
2211                         kmsg = &req->io->msg;
2212                         kmsg->msg.msg_name = &addr;
2213                         /* if iov is set, it's allocated already */
2214                         if (!kmsg->iov)
2215                                 kmsg->iov = kmsg->fast_iov;
2216                         kmsg->msg.msg_iter.iov = kmsg->iov;
2217                 } else {
2218                         struct io_sr_msg *sr = &req->sr_msg;
2219
2220                         kmsg = &io.msg;
2221                         kmsg->msg.msg_name = &addr;
2222
2223                         io.msg.iov = io.msg.fast_iov;
2224                         ret = sendmsg_copy_msghdr(&io.msg.msg, sr->msg,
2225                                         sr->msg_flags, &io.msg.iov);
2226                         if (ret)
2227                                 return ret;
2228                 }
2229
2230                 flags = req->sr_msg.msg_flags;
2231                 if (flags & MSG_DONTWAIT)
2232                         req->flags |= REQ_F_NOWAIT;
2233                 else if (force_nonblock)
2234                         flags |= MSG_DONTWAIT;
2235
2236                 ret = __sys_sendmsg_sock(sock, &kmsg->msg, flags);
2237                 if (force_nonblock && ret == -EAGAIN) {
2238                         if (req->io)
2239                                 return -EAGAIN;
2240                         if (io_alloc_async_ctx(req))
2241                                 return -ENOMEM;
2242                         memcpy(&req->io->msg, &io.msg, sizeof(io.msg));
2243                         req->work.func = io_sendrecv_async;
2244                         return -EAGAIN;
2245                 }
2246                 if (ret == -ERESTARTSYS)
2247                         ret = -EINTR;
2248         }
2249
2250         if (!io_wq_current_is_worker() && kmsg && kmsg->iov != kmsg->fast_iov)
2251                 kfree(kmsg->iov);
2252         io_cqring_add_event(req, ret);
2253         if (ret < 0)
2254                 req_set_fail_links(req);
2255         io_put_req_find_next(req, nxt);
2256         return 0;
2257 #else
2258         return -EOPNOTSUPP;
2259 #endif
2260 }
2261
2262 static int io_recvmsg_prep(struct io_kiocb *req,
2263                            const struct io_uring_sqe *sqe)
2264 {
2265 #if defined(CONFIG_NET)
2266         struct io_sr_msg *sr = &req->sr_msg;
2267         struct io_async_ctx *io = req->io;
2268
2269         sr->msg_flags = READ_ONCE(sqe->msg_flags);
2270         sr->msg = u64_to_user_ptr(READ_ONCE(sqe->addr));
2271
2272         if (!io)
2273                 return 0;
2274
2275         io->msg.iov = io->msg.fast_iov;
2276         return recvmsg_copy_msghdr(&io->msg.msg, sr->msg, sr->msg_flags,
2277                                         &io->msg.uaddr, &io->msg.iov);
2278 #else
2279         return -EOPNOTSUPP;
2280 #endif
2281 }
2282
2283 static int io_recvmsg(struct io_kiocb *req, struct io_kiocb **nxt,
2284                       bool force_nonblock)
2285 {
2286 #if defined(CONFIG_NET)
2287         struct io_async_msghdr *kmsg = NULL;
2288         struct socket *sock;
2289         int ret;
2290
2291         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
2292                 return -EINVAL;
2293
2294         sock = sock_from_file(req->file, &ret);
2295         if (sock) {
2296                 struct io_async_ctx io;
2297                 struct sockaddr_storage addr;
2298                 unsigned flags;
2299
2300                 if (req->io) {
2301                         kmsg = &req->io->msg;
2302                         kmsg->msg.msg_name = &addr;
2303                         /* if iov is set, it's allocated already */
2304                         if (!kmsg->iov)
2305                                 kmsg->iov = kmsg->fast_iov;
2306                         kmsg->msg.msg_iter.iov = kmsg->iov;
2307                 } else {
2308                         struct io_sr_msg *sr = &req->sr_msg;
2309
2310                         kmsg = &io.msg;
2311                         kmsg->msg.msg_name = &addr;
2312
2313                         io.msg.iov = io.msg.fast_iov;
2314                         ret = recvmsg_copy_msghdr(&io.msg.msg, sr->msg,
2315                                         sr->msg_flags, &io.msg.uaddr,
2316                                         &io.msg.iov);
2317                         if (ret)
2318                                 return ret;
2319                 }
2320
2321                 flags = req->sr_msg.msg_flags;
2322                 if (flags & MSG_DONTWAIT)
2323                         req->flags |= REQ_F_NOWAIT;
2324                 else if (force_nonblock)
2325                         flags |= MSG_DONTWAIT;
2326
2327                 ret = __sys_recvmsg_sock(sock, &kmsg->msg, req->sr_msg.msg,
2328                                                 kmsg->uaddr, flags);
2329                 if (force_nonblock && ret == -EAGAIN) {
2330                         if (req->io)
2331                                 return -EAGAIN;
2332                         if (io_alloc_async_ctx(req))
2333                                 return -ENOMEM;
2334                         memcpy(&req->io->msg, &io.msg, sizeof(io.msg));
2335                         req->work.func = io_sendrecv_async;
2336                         return -EAGAIN;
2337                 }
2338                 if (ret == -ERESTARTSYS)
2339                         ret = -EINTR;
2340         }
2341
2342         if (!io_wq_current_is_worker() && kmsg && kmsg->iov != kmsg->fast_iov)
2343                 kfree(kmsg->iov);
2344         io_cqring_add_event(req, ret);
2345         if (ret < 0)
2346                 req_set_fail_links(req);
2347         io_put_req_find_next(req, nxt);
2348         return 0;
2349 #else
2350         return -EOPNOTSUPP;
2351 #endif
2352 }
2353
2354 static int io_accept_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2355 {
2356 #if defined(CONFIG_NET)
2357         struct io_accept *accept = &req->accept;
2358
2359         if (unlikely(req->ctx->flags & (IORING_SETUP_IOPOLL|IORING_SETUP_SQPOLL)))
2360                 return -EINVAL;
2361         if (sqe->ioprio || sqe->len || sqe->buf_index)
2362                 return -EINVAL;
2363
2364         accept->addr = u64_to_user_ptr(READ_ONCE(sqe->addr));
2365         accept->addr_len = u64_to_user_ptr(READ_ONCE(sqe->addr2));
2366         accept->flags = READ_ONCE(sqe->accept_flags);
2367         return 0;
2368 #else
2369         return -EOPNOTSUPP;
2370 #endif
2371 }
2372
2373 #if defined(CONFIG_NET)
2374 static int __io_accept(struct io_kiocb *req, struct io_kiocb **nxt,
2375                        bool force_nonblock)
2376 {
2377         struct io_accept *accept = &req->accept;
2378         unsigned file_flags;
2379         int ret;
2380
2381         file_flags = force_nonblock ? O_NONBLOCK : 0;
2382         ret = __sys_accept4_file(req->file, file_flags, accept->addr,
2383                                         accept->addr_len, accept->flags);
2384         if (ret == -EAGAIN && force_nonblock)
2385                 return -EAGAIN;
2386         if (ret == -ERESTARTSYS)
2387                 ret = -EINTR;
2388         if (ret < 0)
2389                 req_set_fail_links(req);
2390         io_cqring_add_event(req, ret);
2391         io_put_req_find_next(req, nxt);
2392         return 0;
2393 }
2394
2395 static void io_accept_finish(struct io_wq_work **workptr)
2396 {
2397         struct io_kiocb *req = container_of(*workptr, struct io_kiocb, work);
2398         struct io_kiocb *nxt = NULL;
2399
2400         if (io_req_cancelled(req))
2401                 return;
2402         __io_accept(req, &nxt, false);
2403         if (nxt)
2404                 io_wq_assign_next(workptr, nxt);
2405 }
2406 #endif
2407
2408 static int io_accept(struct io_kiocb *req, struct io_kiocb **nxt,
2409                      bool force_nonblock)
2410 {
2411 #if defined(CONFIG_NET)
2412         int ret;
2413
2414         ret = __io_accept(req, nxt, force_nonblock);
2415         if (ret == -EAGAIN && force_nonblock) {
2416                 req->work.func = io_accept_finish;
2417                 req->work.flags |= IO_WQ_WORK_NEEDS_FILES;
2418                 io_put_req(req);
2419                 return -EAGAIN;
2420         }
2421         return 0;
2422 #else
2423         return -EOPNOTSUPP;
2424 #endif
2425 }
2426
2427 static int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2428 {
2429 #if defined(CONFIG_NET)
2430         struct io_connect *conn = &req->connect;
2431         struct io_async_ctx *io = req->io;
2432
2433         if (unlikely(req->ctx->flags & (IORING_SETUP_IOPOLL|IORING_SETUP_SQPOLL)))
2434                 return -EINVAL;
2435         if (sqe->ioprio || sqe->len || sqe->buf_index || sqe->rw_flags)
2436                 return -EINVAL;
2437
2438         conn->addr = u64_to_user_ptr(READ_ONCE(sqe->addr));
2439         conn->addr_len =  READ_ONCE(sqe->addr2);
2440
2441         if (!io)
2442                 return 0;
2443
2444         return move_addr_to_kernel(conn->addr, conn->addr_len,
2445                                         &io->connect.address);
2446 #else
2447         return -EOPNOTSUPP;
2448 #endif
2449 }
2450
2451 static int io_connect(struct io_kiocb *req, struct io_kiocb **nxt,
2452                       bool force_nonblock)
2453 {
2454 #if defined(CONFIG_NET)
2455         struct io_async_ctx __io, *io;
2456         unsigned file_flags;
2457         int ret;
2458
2459         if (req->io) {
2460                 io = req->io;
2461         } else {
2462                 ret = move_addr_to_kernel(req->connect.addr,
2463                                                 req->connect.addr_len,
2464                                                 &__io.connect.address);
2465                 if (ret)
2466                         goto out;
2467                 io = &__io;
2468         }
2469
2470         file_flags = force_nonblock ? O_NONBLOCK : 0;
2471
2472         ret = __sys_connect_file(req->file, &io->connect.address,
2473                                         req->connect.addr_len, file_flags);
2474         if ((ret == -EAGAIN || ret == -EINPROGRESS) && force_nonblock) {
2475                 if (req->io)
2476                         return -EAGAIN;
2477                 if (io_alloc_async_ctx(req)) {
2478                         ret = -ENOMEM;
2479                         goto out;
2480                 }
2481                 memcpy(&req->io->connect, &__io.connect, sizeof(__io.connect));
2482                 return -EAGAIN;
2483         }
2484         if (ret == -ERESTARTSYS)
2485                 ret = -EINTR;
2486 out:
2487         if (ret < 0)
2488                 req_set_fail_links(req);
2489         io_cqring_add_event(req, ret);
2490         io_put_req_find_next(req, nxt);
2491         return 0;
2492 #else
2493         return -EOPNOTSUPP;
2494 #endif
2495 }
2496
2497 static void io_poll_remove_one(struct io_kiocb *req)
2498 {
2499         struct io_poll_iocb *poll = &req->poll;
2500
2501         spin_lock(&poll->head->lock);
2502         WRITE_ONCE(poll->canceled, true);
2503         if (!list_empty(&poll->wait.entry)) {
2504                 list_del_init(&poll->wait.entry);
2505                 io_queue_async_work(req);
2506         }
2507         spin_unlock(&poll->head->lock);
2508         hash_del(&req->hash_node);
2509 }
2510
2511 static void io_poll_remove_all(struct io_ring_ctx *ctx)
2512 {
2513         struct hlist_node *tmp;
2514         struct io_kiocb *req;
2515         int i;
2516
2517         spin_lock_irq(&ctx->completion_lock);
2518         for (i = 0; i < (1U << ctx->cancel_hash_bits); i++) {
2519                 struct hlist_head *list;
2520
2521                 list = &ctx->cancel_hash[i];
2522                 hlist_for_each_entry_safe(req, tmp, list, hash_node)
2523                         io_poll_remove_one(req);
2524         }
2525         spin_unlock_irq(&ctx->completion_lock);
2526 }
2527
2528 static int io_poll_cancel(struct io_ring_ctx *ctx, __u64 sqe_addr)
2529 {
2530         struct hlist_head *list;
2531         struct io_kiocb *req;
2532
2533         list = &ctx->cancel_hash[hash_long(sqe_addr, ctx->cancel_hash_bits)];
2534         hlist_for_each_entry(req, list, hash_node) {
2535                 if (sqe_addr == req->user_data) {
2536                         io_poll_remove_one(req);
2537                         return 0;
2538                 }
2539         }
2540
2541         return -ENOENT;
2542 }
2543
2544 static int io_poll_remove_prep(struct io_kiocb *req,
2545                                const struct io_uring_sqe *sqe)
2546 {
2547         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
2548                 return -EINVAL;
2549         if (sqe->ioprio || sqe->off || sqe->len || sqe->buf_index ||
2550             sqe->poll_events)
2551                 return -EINVAL;
2552
2553         req->poll.addr = READ_ONCE(sqe->addr);
2554         return 0;
2555 }
2556
2557 /*
2558  * Find a running poll command that matches one specified in sqe->addr,
2559  * and remove it if found.
2560  */
2561 static int io_poll_remove(struct io_kiocb *req)
2562 {
2563         struct io_ring_ctx *ctx = req->ctx;
2564         u64 addr;
2565         int ret;
2566
2567         addr = req->poll.addr;
2568         spin_lock_irq(&ctx->completion_lock);
2569         ret = io_poll_cancel(ctx, addr);
2570         spin_unlock_irq(&ctx->completion_lock);
2571
2572         io_cqring_add_event(req, ret);
2573         if (ret < 0)
2574                 req_set_fail_links(req);
2575         io_put_req(req);
2576         return 0;
2577 }
2578
2579 static void io_poll_complete(struct io_kiocb *req, __poll_t mask, int error)
2580 {
2581         struct io_ring_ctx *ctx = req->ctx;
2582
2583         req->poll.done = true;
2584         if (error)
2585                 io_cqring_fill_event(req, error);
2586         else
2587                 io_cqring_fill_event(req, mangle_poll(mask));
2588         io_commit_cqring(ctx);
2589 }
2590
2591 static void io_poll_complete_work(struct io_wq_work **workptr)
2592 {
2593         struct io_wq_work *work = *workptr;
2594         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
2595         struct io_poll_iocb *poll = &req->poll;
2596         struct poll_table_struct pt = { ._key = poll->events };
2597         struct io_ring_ctx *ctx = req->ctx;
2598         struct io_kiocb *nxt = NULL;
2599         __poll_t mask = 0;
2600         int ret = 0;
2601
2602         if (work->flags & IO_WQ_WORK_CANCEL) {
2603                 WRITE_ONCE(poll->canceled, true);
2604                 ret = -ECANCELED;
2605         } else if (READ_ONCE(poll->canceled)) {
2606                 ret = -ECANCELED;
2607         }
2608
2609         if (ret != -ECANCELED)
2610                 mask = vfs_poll(poll->file, &pt) & poll->events;
2611
2612         /*
2613          * Note that ->ki_cancel callers also delete iocb from active_reqs after
2614          * calling ->ki_cancel.  We need the ctx_lock roundtrip here to
2615          * synchronize with them.  In the cancellation case the list_del_init
2616          * itself is not actually needed, but harmless so we keep it in to
2617          * avoid further branches in the fast path.
2618          */
2619         spin_lock_irq(&ctx->completion_lock);
2620         if (!mask && ret != -ECANCELED) {
2621                 add_wait_queue(poll->head, &poll->wait);
2622                 spin_unlock_irq(&ctx->completion_lock);
2623                 return;
2624         }
2625         hash_del(&req->hash_node);
2626         io_poll_complete(req, mask, ret);
2627         spin_unlock_irq(&ctx->completion_lock);
2628
2629         io_cqring_ev_posted(ctx);
2630
2631         if (ret < 0)
2632                 req_set_fail_links(req);
2633         io_put_req_find_next(req, &nxt);
2634         if (nxt)
2635                 io_wq_assign_next(workptr, nxt);
2636 }
2637
2638 static int io_poll_wake(struct wait_queue_entry *wait, unsigned mode, int sync,
2639                         void *key)
2640 {
2641         struct io_poll_iocb *poll = wait->private;
2642         struct io_kiocb *req = container_of(poll, struct io_kiocb, poll);
2643         struct io_ring_ctx *ctx = req->ctx;
2644         __poll_t mask = key_to_poll(key);
2645         unsigned long flags;
2646
2647         /* for instances that support it check for an event match first: */
2648         if (mask && !(mask & poll->events))
2649                 return 0;
2650
2651         list_del_init(&poll->wait.entry);
2652
2653         /*
2654          * Run completion inline if we can. We're using trylock here because
2655          * we are violating the completion_lock -> poll wq lock ordering.
2656          * If we have a link timeout we're going to need the completion_lock
2657          * for finalizing the request, mark us as having grabbed that already.
2658          */
2659         if (mask && spin_trylock_irqsave(&ctx->completion_lock, flags)) {
2660                 hash_del(&req->hash_node);
2661                 io_poll_complete(req, mask, 0);
2662                 req->flags |= REQ_F_COMP_LOCKED;
2663                 io_put_req(req);
2664                 spin_unlock_irqrestore(&ctx->completion_lock, flags);
2665
2666                 io_cqring_ev_posted(ctx);
2667         } else {
2668                 io_queue_async_work(req);
2669         }
2670
2671         return 1;
2672 }
2673
2674 struct io_poll_table {
2675         struct poll_table_struct pt;
2676         struct io_kiocb *req;
2677         int error;
2678 };
2679
2680 static void io_poll_queue_proc(struct file *file, struct wait_queue_head *head,
2681                                struct poll_table_struct *p)
2682 {
2683         struct io_poll_table *pt = container_of(p, struct io_poll_table, pt);
2684
2685         if (unlikely(pt->req->poll.head)) {
2686                 pt->error = -EINVAL;
2687                 return;
2688         }
2689
2690         pt->error = 0;
2691         pt->req->poll.head = head;
2692         add_wait_queue(head, &pt->req->poll.wait);
2693 }
2694
2695 static void io_poll_req_insert(struct io_kiocb *req)
2696 {
2697         struct io_ring_ctx *ctx = req->ctx;
2698         struct hlist_head *list;
2699
2700         list = &ctx->cancel_hash[hash_long(req->user_data, ctx->cancel_hash_bits)];
2701         hlist_add_head(&req->hash_node, list);
2702 }
2703
2704 static int io_poll_add_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2705 {
2706         struct io_poll_iocb *poll = &req->poll;
2707         u16 events;
2708
2709         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
2710                 return -EINVAL;
2711         if (sqe->addr || sqe->ioprio || sqe->off || sqe->len || sqe->buf_index)
2712                 return -EINVAL;
2713         if (!poll->file)
2714                 return -EBADF;
2715
2716         events = READ_ONCE(sqe->poll_events);
2717         poll->events = demangle_poll(events) | EPOLLERR | EPOLLHUP;
2718         return 0;
2719 }
2720
2721 static int io_poll_add(struct io_kiocb *req, struct io_kiocb **nxt)
2722 {
2723         struct io_poll_iocb *poll = &req->poll;
2724         struct io_ring_ctx *ctx = req->ctx;
2725         struct io_poll_table ipt;
2726         bool cancel = false;
2727         __poll_t mask;
2728
2729         INIT_IO_WORK(&req->work, io_poll_complete_work);
2730         INIT_HLIST_NODE(&req->hash_node);
2731
2732         poll->head = NULL;
2733         poll->done = false;
2734         poll->canceled = false;
2735
2736         ipt.pt._qproc = io_poll_queue_proc;
2737         ipt.pt._key = poll->events;
2738         ipt.req = req;
2739         ipt.error = -EINVAL; /* same as no support for IOCB_CMD_POLL */
2740
2741         /* initialized the list so that we can do list_empty checks */
2742         INIT_LIST_HEAD(&poll->wait.entry);
2743         init_waitqueue_func_entry(&poll->wait, io_poll_wake);
2744         poll->wait.private = poll;
2745
2746         INIT_LIST_HEAD(&req->list);
2747
2748         mask = vfs_poll(poll->file, &ipt.pt) & poll->events;
2749
2750         spin_lock_irq(&ctx->completion_lock);
2751         if (likely(poll->head)) {
2752                 spin_lock(&poll->head->lock);
2753                 if (unlikely(list_empty(&poll->wait.entry))) {
2754                         if (ipt.error)
2755                                 cancel = true;
2756                         ipt.error = 0;
2757                         mask = 0;
2758                 }
2759                 if (mask || ipt.error)
2760                         list_del_init(&poll->wait.entry);
2761                 else if (cancel)
2762                         WRITE_ONCE(poll->canceled, true);
2763                 else if (!poll->done) /* actually waiting for an event */
2764                         io_poll_req_insert(req);
2765                 spin_unlock(&poll->head->lock);
2766         }
2767         if (mask) { /* no async, we'd stolen it */
2768                 ipt.error = 0;
2769                 io_poll_complete(req, mask, 0);
2770         }
2771         spin_unlock_irq(&ctx->completion_lock);
2772
2773         if (mask) {
2774                 io_cqring_ev_posted(ctx);
2775                 io_put_req_find_next(req, nxt);
2776         }
2777         return ipt.error;
2778 }
2779
2780 static enum hrtimer_restart io_timeout_fn(struct hrtimer *timer)
2781 {
2782         struct io_timeout_data *data = container_of(timer,
2783                                                 struct io_timeout_data, timer);
2784         struct io_kiocb *req = data->req;
2785         struct io_ring_ctx *ctx = req->ctx;
2786         unsigned long flags;
2787
2788         atomic_inc(&ctx->cq_timeouts);
2789
2790         spin_lock_irqsave(&ctx->completion_lock, flags);
2791         /*
2792          * We could be racing with timeout deletion. If the list is empty,
2793          * then timeout lookup already found it and will be handling it.
2794          */
2795         if (!list_empty(&req->list)) {
2796                 struct io_kiocb *prev;
2797
2798                 /*
2799                  * Adjust the reqs sequence before the current one because it
2800                  * will consume a slot in the cq_ring and the cq_tail
2801                  * pointer will be increased, otherwise other timeout reqs may
2802                  * return in advance without waiting for enough wait_nr.
2803                  */
2804                 prev = req;
2805                 list_for_each_entry_continue_reverse(prev, &ctx->timeout_list, list)
2806                         prev->sequence++;
2807                 list_del_init(&req->list);
2808         }
2809
2810         io_cqring_fill_event(req, -ETIME);
2811         io_commit_cqring(ctx);
2812         spin_unlock_irqrestore(&ctx->completion_lock, flags);
2813
2814         io_cqring_ev_posted(ctx);
2815         req_set_fail_links(req);
2816         io_put_req(req);
2817         return HRTIMER_NORESTART;
2818 }
2819
2820 static int io_timeout_cancel(struct io_ring_ctx *ctx, __u64 user_data)
2821 {
2822         struct io_kiocb *req;
2823         int ret = -ENOENT;
2824
2825         list_for_each_entry(req, &ctx->timeout_list, list) {
2826                 if (user_data == req->user_data) {
2827                         list_del_init(&req->list);
2828                         ret = 0;
2829                         break;
2830                 }
2831         }
2832
2833         if (ret == -ENOENT)
2834                 return ret;
2835
2836         ret = hrtimer_try_to_cancel(&req->io->timeout.timer);
2837         if (ret == -1)
2838                 return -EALREADY;
2839
2840         req_set_fail_links(req);
2841         io_cqring_fill_event(req, -ECANCELED);
2842         io_put_req(req);
2843         return 0;
2844 }
2845
2846 static int io_timeout_remove_prep(struct io_kiocb *req,
2847                                   const struct io_uring_sqe *sqe)
2848 {
2849         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
2850                 return -EINVAL;
2851         if (sqe->flags || sqe->ioprio || sqe->buf_index || sqe->len)
2852                 return -EINVAL;
2853
2854         req->timeout.addr = READ_ONCE(sqe->addr);
2855         req->timeout.flags = READ_ONCE(sqe->timeout_flags);
2856         if (req->timeout.flags)
2857                 return -EINVAL;
2858
2859         return 0;
2860 }
2861
2862 /*
2863  * Remove or update an existing timeout command
2864  */
2865 static int io_timeout_remove(struct io_kiocb *req)
2866 {
2867         struct io_ring_ctx *ctx = req->ctx;
2868         int ret;
2869
2870         spin_lock_irq(&ctx->completion_lock);
2871         ret = io_timeout_cancel(ctx, req->timeout.addr);
2872
2873         io_cqring_fill_event(req, ret);
2874         io_commit_cqring(ctx);
2875         spin_unlock_irq(&ctx->completion_lock);
2876         io_cqring_ev_posted(ctx);
2877         if (ret < 0)
2878                 req_set_fail_links(req);
2879         io_put_req(req);
2880         return 0;
2881 }
2882
2883 static int io_timeout_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe,
2884                            bool is_timeout_link)
2885 {
2886         struct io_timeout_data *data;
2887         unsigned flags;
2888
2889         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
2890                 return -EINVAL;
2891         if (sqe->ioprio || sqe->buf_index || sqe->len != 1)
2892                 return -EINVAL;
2893         if (sqe->off && is_timeout_link)
2894                 return -EINVAL;
2895         flags = READ_ONCE(sqe->timeout_flags);
2896         if (flags & ~IORING_TIMEOUT_ABS)
2897                 return -EINVAL;
2898
2899         req->timeout.count = READ_ONCE(sqe->off);
2900
2901         if (!req->io && io_alloc_async_ctx(req))
2902                 return -ENOMEM;
2903
2904         data = &req->io->timeout;
2905         data->req = req;
2906         req->flags |= REQ_F_TIMEOUT;
2907
2908         if (get_timespec64(&data->ts, u64_to_user_ptr(sqe->addr)))
2909                 return -EFAULT;
2910
2911         if (flags & IORING_TIMEOUT_ABS)
2912                 data->mode = HRTIMER_MODE_ABS;
2913         else
2914                 data->mode = HRTIMER_MODE_REL;
2915
2916         hrtimer_init(&data->timer, CLOCK_MONOTONIC, data->mode);
2917         return 0;
2918 }
2919
2920 static int io_timeout(struct io_kiocb *req)
2921 {
2922         unsigned count;
2923         struct io_ring_ctx *ctx = req->ctx;
2924         struct io_timeout_data *data;
2925         struct list_head *entry;
2926         unsigned span = 0;
2927
2928         data = &req->io->timeout;
2929
2930         /*
2931          * sqe->off holds how many events that need to occur for this
2932          * timeout event to be satisfied. If it isn't set, then this is
2933          * a pure timeout request, sequence isn't used.
2934          */
2935         count = req->timeout.count;
2936         if (!count) {
2937                 req->flags |= REQ_F_TIMEOUT_NOSEQ;
2938                 spin_lock_irq(&ctx->completion_lock);
2939                 entry = ctx->timeout_list.prev;
2940                 goto add;
2941         }
2942
2943         req->sequence = ctx->cached_sq_head + count - 1;
2944         data->seq_offset = count;
2945
2946         /*
2947          * Insertion sort, ensuring the first entry in the list is always
2948          * the one we need first.
2949          */
2950         spin_lock_irq(&ctx->completion_lock);
2951         list_for_each_prev(entry, &ctx->timeout_list) {
2952                 struct io_kiocb *nxt = list_entry(entry, struct io_kiocb, list);
2953                 unsigned nxt_sq_head;
2954                 long long tmp, tmp_nxt;
2955                 u32 nxt_offset = nxt->io->timeout.seq_offset;
2956
2957                 if (nxt->flags & REQ_F_TIMEOUT_NOSEQ)
2958                         continue;
2959
2960                 /*
2961                  * Since cached_sq_head + count - 1 can overflow, use type long
2962                  * long to store it.
2963                  */
2964                 tmp = (long long)ctx->cached_sq_head + count - 1;
2965                 nxt_sq_head = nxt->sequence - nxt_offset + 1;
2966                 tmp_nxt = (long long)nxt_sq_head + nxt_offset - 1;
2967
2968                 /*
2969                  * cached_sq_head may overflow, and it will never overflow twice
2970                  * once there is some timeout req still be valid.
2971                  */
2972                 if (ctx->cached_sq_head < nxt_sq_head)
2973                         tmp += UINT_MAX;
2974
2975                 if (tmp > tmp_nxt)
2976                         break;
2977
2978                 /*
2979                  * Sequence of reqs after the insert one and itself should
2980                  * be adjusted because each timeout req consumes a slot.
2981                  */
2982                 span++;
2983                 nxt->sequence++;
2984         }
2985         req->sequence -= span;
2986 add:
2987         list_add(&req->list, entry);
2988         data->timer.function = io_timeout_fn;
2989         hrtimer_start(&data->timer, timespec64_to_ktime(data->ts), data->mode);
2990         spin_unlock_irq(&ctx->completion_lock);
2991         return 0;
2992 }
2993
2994 static bool io_cancel_cb(struct io_wq_work *work, void *data)
2995 {
2996         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
2997
2998         return req->user_data == (unsigned long) data;
2999 }
3000
3001 static int io_async_cancel_one(struct io_ring_ctx *ctx, void *sqe_addr)
3002 {
3003         enum io_wq_cancel cancel_ret;
3004         int ret = 0;
3005
3006         cancel_ret = io_wq_cancel_cb(ctx->io_wq, io_cancel_cb, sqe_addr);
3007         switch (cancel_ret) {
3008         case IO_WQ_CANCEL_OK:
3009                 ret = 0;
3010                 break;
3011         case IO_WQ_CANCEL_RUNNING:
3012                 ret = -EALREADY;
3013                 break;
3014         case IO_WQ_CANCEL_NOTFOUND:
3015                 ret = -ENOENT;
3016                 break;
3017         }
3018
3019         return ret;
3020 }
3021
3022 static void io_async_find_and_cancel(struct io_ring_ctx *ctx,
3023                                      struct io_kiocb *req, __u64 sqe_addr,
3024                                      struct io_kiocb **nxt, int success_ret)
3025 {
3026         unsigned long flags;
3027         int ret;
3028
3029         ret = io_async_cancel_one(ctx, (void *) (unsigned long) sqe_addr);
3030         if (ret != -ENOENT) {
3031                 spin_lock_irqsave(&ctx->completion_lock, flags);
3032                 goto done;
3033         }
3034
3035         spin_lock_irqsave(&ctx->completion_lock, flags);
3036         ret = io_timeout_cancel(ctx, sqe_addr);
3037         if (ret != -ENOENT)
3038                 goto done;
3039         ret = io_poll_cancel(ctx, sqe_addr);
3040 done:
3041         if (!ret)
3042                 ret = success_ret;
3043         io_cqring_fill_event(req, ret);
3044         io_commit_cqring(ctx);
3045         spin_unlock_irqrestore(&ctx->completion_lock, flags);
3046         io_cqring_ev_posted(ctx);
3047
3048         if (ret < 0)
3049                 req_set_fail_links(req);
3050         io_put_req_find_next(req, nxt);
3051 }
3052
3053 static int io_async_cancel_prep(struct io_kiocb *req,
3054                                 const struct io_uring_sqe *sqe)
3055 {
3056         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
3057                 return -EINVAL;
3058         if (sqe->flags || sqe->ioprio || sqe->off || sqe->len ||
3059             sqe->cancel_flags)
3060                 return -EINVAL;
3061
3062         req->cancel.addr = READ_ONCE(sqe->addr);
3063         return 0;
3064 }
3065
3066 static int io_async_cancel(struct io_kiocb *req, struct io_kiocb **nxt)
3067 {
3068         struct io_ring_ctx *ctx = req->ctx;
3069
3070         io_async_find_and_cancel(ctx, req, req->cancel.addr, nxt, 0);
3071         return 0;
3072 }
3073
3074 static int io_req_defer_prep(struct io_kiocb *req,
3075                              const struct io_uring_sqe *sqe)
3076 {
3077         ssize_t ret = 0;
3078
3079         switch (req->opcode) {
3080         case IORING_OP_NOP:
3081                 break;
3082         case IORING_OP_READV:
3083         case IORING_OP_READ_FIXED:
3084                 ret = io_read_prep(req, sqe, true);
3085                 break;
3086         case IORING_OP_WRITEV:
3087         case IORING_OP_WRITE_FIXED:
3088                 ret = io_write_prep(req, sqe, true);
3089                 break;
3090         case IORING_OP_POLL_ADD:
3091                 ret = io_poll_add_prep(req, sqe);
3092                 break;
3093         case IORING_OP_POLL_REMOVE:
3094                 ret = io_poll_remove_prep(req, sqe);
3095                 break;
3096         case IORING_OP_FSYNC:
3097                 ret = io_prep_fsync(req, sqe);
3098                 break;
3099         case IORING_OP_SYNC_FILE_RANGE:
3100                 ret = io_prep_sfr(req, sqe);
3101                 break;
3102         case IORING_OP_SENDMSG:
3103                 ret = io_sendmsg_prep(req, sqe);
3104                 break;
3105         case IORING_OP_RECVMSG:
3106                 ret = io_recvmsg_prep(req, sqe);
3107                 break;
3108         case IORING_OP_CONNECT:
3109                 ret = io_connect_prep(req, sqe);
3110                 break;
3111         case IORING_OP_TIMEOUT:
3112                 ret = io_timeout_prep(req, sqe, false);
3113                 break;
3114         case IORING_OP_TIMEOUT_REMOVE:
3115                 ret = io_timeout_remove_prep(req, sqe);
3116                 break;
3117         case IORING_OP_ASYNC_CANCEL:
3118                 ret = io_async_cancel_prep(req, sqe);
3119                 break;
3120         case IORING_OP_LINK_TIMEOUT:
3121                 ret = io_timeout_prep(req, sqe, true);
3122                 break;
3123         case IORING_OP_ACCEPT:
3124                 ret = io_accept_prep(req, sqe);
3125                 break;
3126         default:
3127                 printk_once(KERN_WARNING "io_uring: unhandled opcode %d\n",
3128                                 req->opcode);
3129                 ret = -EINVAL;
3130                 break;
3131         }
3132
3133         return ret;
3134 }
3135
3136 static int io_req_defer(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3137 {
3138         struct io_ring_ctx *ctx = req->ctx;
3139         int ret;
3140
3141         /* Still need defer if there is pending req in defer list. */
3142         if (!req_need_defer(req) && list_empty(&ctx->defer_list))
3143                 return 0;
3144
3145         if (!req->io && io_alloc_async_ctx(req))
3146                 return -EAGAIN;
3147
3148         ret = io_req_defer_prep(req, sqe);
3149         if (ret < 0)
3150                 return ret;
3151
3152         spin_lock_irq(&ctx->completion_lock);
3153         if (!req_need_defer(req) && list_empty(&ctx->defer_list)) {
3154                 spin_unlock_irq(&ctx->completion_lock);
3155                 return 0;
3156         }
3157
3158         trace_io_uring_defer(ctx, req, req->user_data);
3159         list_add_tail(&req->list, &ctx->defer_list);
3160         spin_unlock_irq(&ctx->completion_lock);
3161         return -EIOCBQUEUED;
3162 }
3163
3164 static int io_issue_sqe(struct io_kiocb *req, const struct io_uring_sqe *sqe,
3165                         struct io_kiocb **nxt, bool force_nonblock)
3166 {
3167         struct io_ring_ctx *ctx = req->ctx;
3168         int ret;
3169
3170         switch (req->opcode) {
3171         case IORING_OP_NOP:
3172                 ret = io_nop(req);
3173                 break;
3174         case IORING_OP_READV:
3175         case IORING_OP_READ_FIXED:
3176                 if (sqe) {
3177                         ret = io_read_prep(req, sqe, force_nonblock);
3178                         if (ret < 0)
3179                                 break;
3180                 }
3181                 ret = io_read(req, nxt, force_nonblock);
3182                 break;
3183         case IORING_OP_WRITEV:
3184         case IORING_OP_WRITE_FIXED:
3185                 if (sqe) {
3186                         ret = io_write_prep(req, sqe, force_nonblock);
3187                         if (ret < 0)
3188                                 break;
3189                 }
3190                 ret = io_write(req, nxt, force_nonblock);
3191                 break;
3192         case IORING_OP_FSYNC:
3193                 if (sqe) {
3194                         ret = io_prep_fsync(req, sqe);
3195                         if (ret < 0)
3196                                 break;
3197                 }
3198                 ret = io_fsync(req, nxt, force_nonblock);
3199                 break;
3200         case IORING_OP_POLL_ADD:
3201                 if (sqe) {
3202                         ret = io_poll_add_prep(req, sqe);
3203                         if (ret)
3204                                 break;
3205                 }
3206                 ret = io_poll_add(req, nxt);
3207                 break;
3208         case IORING_OP_POLL_REMOVE:
3209                 if (sqe) {
3210                         ret = io_poll_remove_prep(req, sqe);
3211                         if (ret < 0)
3212                                 break;
3213                 }
3214                 ret = io_poll_remove(req);
3215                 break;
3216         case IORING_OP_SYNC_FILE_RANGE:
3217                 if (sqe) {
3218                         ret = io_prep_sfr(req, sqe);
3219                         if (ret < 0)
3220                                 break;
3221                 }
3222                 ret = io_sync_file_range(req, nxt, force_nonblock);
3223                 break;
3224         case IORING_OP_SENDMSG:
3225                 if (sqe) {
3226                         ret = io_sendmsg_prep(req, sqe);
3227                         if (ret < 0)
3228                                 break;
3229                 }
3230                 ret = io_sendmsg(req, nxt, force_nonblock);
3231                 break;
3232         case IORING_OP_RECVMSG:
3233                 if (sqe) {
3234                         ret = io_recvmsg_prep(req, sqe);
3235                         if (ret)
3236                                 break;
3237                 }
3238                 ret = io_recvmsg(req, nxt, force_nonblock);
3239                 break;
3240         case IORING_OP_TIMEOUT:
3241                 if (sqe) {
3242                         ret = io_timeout_prep(req, sqe, false);
3243                         if (ret)
3244                                 break;
3245                 }
3246                 ret = io_timeout(req);
3247                 break;
3248         case IORING_OP_TIMEOUT_REMOVE:
3249                 if (sqe) {
3250                         ret = io_timeout_remove_prep(req, sqe);
3251                         if (ret)
3252                                 break;
3253                 }
3254                 ret = io_timeout_remove(req);
3255                 break;
3256         case IORING_OP_ACCEPT:
3257                 if (sqe) {
3258                         ret = io_accept_prep(req, sqe);
3259                         if (ret)
3260                                 break;
3261                 }
3262                 ret = io_accept(req, nxt, force_nonblock);
3263                 break;
3264         case IORING_OP_CONNECT:
3265                 if (sqe) {
3266                         ret = io_connect_prep(req, sqe);
3267                         if (ret)
3268                                 break;
3269                 }
3270                 ret = io_connect(req, nxt, force_nonblock);
3271                 break;
3272         case IORING_OP_ASYNC_CANCEL:
3273                 if (sqe) {
3274                         ret = io_async_cancel_prep(req, sqe);
3275                         if (ret)
3276                                 break;
3277                 }
3278                 ret = io_async_cancel(req, nxt);
3279                 break;
3280         default:
3281                 ret = -EINVAL;
3282                 break;
3283         }
3284
3285         if (ret)
3286                 return ret;
3287
3288         if (ctx->flags & IORING_SETUP_IOPOLL) {
3289                 const bool in_async = io_wq_current_is_worker();
3290
3291                 if (req->result == -EAGAIN)
3292                         return -EAGAIN;
3293
3294                 /* workqueue context doesn't hold uring_lock, grab it now */
3295                 if (in_async)
3296                         mutex_lock(&ctx->uring_lock);
3297
3298                 io_iopoll_req_issued(req);
3299
3300                 if (in_async)
3301                         mutex_unlock(&ctx->uring_lock);
3302         }
3303
3304         return 0;
3305 }
3306
3307 static void io_wq_submit_work(struct io_wq_work **workptr)
3308 {
3309         struct io_wq_work *work = *workptr;
3310         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
3311         struct io_kiocb *nxt = NULL;
3312         int ret = 0;
3313
3314         if (work->flags & IO_WQ_WORK_CANCEL)
3315                 ret = -ECANCELED;
3316
3317         if (!ret) {
3318                 req->has_user = (work->flags & IO_WQ_WORK_HAS_MM) != 0;
3319                 req->in_async = true;
3320                 do {
3321                         ret = io_issue_sqe(req, NULL, &nxt, false);
3322                         /*
3323                          * We can get EAGAIN for polled IO even though we're
3324                          * forcing a sync submission from here, since we can't
3325                          * wait for request slots on the block side.
3326                          */
3327                         if (ret != -EAGAIN)
3328                                 break;
3329                         cond_resched();
3330                 } while (1);
3331         }
3332
3333         /* drop submission reference */
3334         io_put_req(req);
3335
3336         if (ret) {
3337                 req_set_fail_links(req);
3338                 io_cqring_add_event(req, ret);
3339                 io_put_req(req);
3340         }
3341
3342         /* if a dependent link is ready, pass it back */
3343         if (!ret && nxt)
3344                 io_wq_assign_next(workptr, nxt);
3345 }
3346
3347 static bool io_req_op_valid(int op)
3348 {
3349         return op >= IORING_OP_NOP && op < IORING_OP_LAST;
3350 }
3351
3352 static int io_req_needs_file(struct io_kiocb *req)
3353 {
3354         switch (req->opcode) {
3355         case IORING_OP_NOP:
3356         case IORING_OP_POLL_REMOVE:
3357         case IORING_OP_TIMEOUT:
3358         case IORING_OP_TIMEOUT_REMOVE:
3359         case IORING_OP_ASYNC_CANCEL:
3360         case IORING_OP_LINK_TIMEOUT:
3361                 return 0;
3362         default:
3363                 if (io_req_op_valid(req->opcode))
3364                         return 1;
3365                 return -EINVAL;
3366         }
3367 }
3368
3369 static inline struct file *io_file_from_index(struct io_ring_ctx *ctx,
3370                                               int index)
3371 {
3372         struct fixed_file_table *table;
3373
3374         table = &ctx->file_table[index >> IORING_FILE_TABLE_SHIFT];
3375         return table->files[index & IORING_FILE_TABLE_MASK];
3376 }
3377
3378 static int io_req_set_file(struct io_submit_state *state, struct io_kiocb *req,
3379                            const struct io_uring_sqe *sqe)
3380 {
3381         struct io_ring_ctx *ctx = req->ctx;
3382         unsigned flags;
3383         int fd, ret;
3384
3385         flags = READ_ONCE(sqe->flags);
3386         fd = READ_ONCE(sqe->fd);
3387
3388         if (flags & IOSQE_IO_DRAIN)
3389                 req->flags |= REQ_F_IO_DRAIN;
3390
3391         ret = io_req_needs_file(req);
3392         if (ret <= 0)
3393                 return ret;
3394
3395         if (flags & IOSQE_FIXED_FILE) {
3396                 if (unlikely(!ctx->file_table ||
3397                     (unsigned) fd >= ctx->nr_user_files))
3398                         return -EBADF;
3399                 fd = array_index_nospec(fd, ctx->nr_user_files);
3400                 req->file = io_file_from_index(ctx, fd);
3401                 if (!req->file)
3402                         return -EBADF;
3403                 req->flags |= REQ_F_FIXED_FILE;
3404         } else {
3405                 if (req->needs_fixed_file)
3406                         return -EBADF;
3407                 trace_io_uring_file_get(ctx, fd);
3408                 req->file = io_file_get(state, fd);
3409                 if (unlikely(!req->file))
3410                         return -EBADF;
3411         }
3412
3413         return 0;
3414 }
3415
3416 static int io_grab_files(struct io_kiocb *req)
3417 {
3418         int ret = -EBADF;
3419         struct io_ring_ctx *ctx = req->ctx;
3420
3421         rcu_read_lock();
3422         spin_lock_irq(&ctx->inflight_lock);
3423         /*
3424          * We use the f_ops->flush() handler to ensure that we can flush
3425          * out work accessing these files if the fd is closed. Check if
3426          * the fd has changed since we started down this path, and disallow
3427          * this operation if it has.
3428          */
3429         if (fcheck(req->ring_fd) == req->ring_file) {
3430                 list_add(&req->inflight_entry, &ctx->inflight_list);
3431                 req->flags |= REQ_F_INFLIGHT;
3432                 req->work.files = current->files;
3433                 ret = 0;
3434         }
3435         spin_unlock_irq(&ctx->inflight_lock);
3436         rcu_read_unlock();
3437
3438         return ret;
3439 }
3440
3441 static enum hrtimer_restart io_link_timeout_fn(struct hrtimer *timer)
3442 {
3443         struct io_timeout_data *data = container_of(timer,
3444                                                 struct io_timeout_data, timer);
3445         struct io_kiocb *req = data->req;
3446         struct io_ring_ctx *ctx = req->ctx;
3447         struct io_kiocb *prev = NULL;
3448         unsigned long flags;
3449
3450         spin_lock_irqsave(&ctx->completion_lock, flags);
3451
3452         /*
3453          * We don't expect the list to be empty, that will only happen if we
3454          * race with the completion of the linked work.
3455          */
3456         if (!list_empty(&req->link_list)) {
3457                 prev = list_entry(req->link_list.prev, struct io_kiocb,
3458                                   link_list);
3459                 if (refcount_inc_not_zero(&prev->refs)) {
3460                         list_del_init(&req->link_list);
3461                         prev->flags &= ~REQ_F_LINK_TIMEOUT;
3462                 } else
3463                         prev = NULL;
3464         }
3465
3466         spin_unlock_irqrestore(&ctx->completion_lock, flags);
3467
3468         if (prev) {
3469                 req_set_fail_links(prev);
3470                 io_async_find_and_cancel(ctx, req, prev->user_data, NULL,
3471                                                 -ETIME);
3472                 io_put_req(prev);
3473         } else {
3474                 io_cqring_add_event(req, -ETIME);
3475                 io_put_req(req);
3476         }
3477         return HRTIMER_NORESTART;
3478 }
3479
3480 static void io_queue_linked_timeout(struct io_kiocb *req)
3481 {
3482         struct io_ring_ctx *ctx = req->ctx;
3483
3484         /*
3485          * If the list is now empty, then our linked request finished before
3486          * we got a chance to setup the timer
3487          */
3488         spin_lock_irq(&ctx->completion_lock);
3489         if (!list_empty(&req->link_list)) {
3490                 struct io_timeout_data *data = &req->io->timeout;
3491
3492                 data->timer.function = io_link_timeout_fn;
3493                 hrtimer_start(&data->timer, timespec64_to_ktime(data->ts),
3494                                 data->mode);
3495         }
3496         spin_unlock_irq(&ctx->completion_lock);
3497
3498         /* drop submission reference */
3499         io_put_req(req);
3500 }
3501
3502 static struct io_kiocb *io_prep_linked_timeout(struct io_kiocb *req)
3503 {
3504         struct io_kiocb *nxt;
3505
3506         if (!(req->flags & REQ_F_LINK))
3507                 return NULL;
3508
3509         nxt = list_first_entry_or_null(&req->link_list, struct io_kiocb,
3510                                         link_list);
3511         if (!nxt || nxt->opcode != IORING_OP_LINK_TIMEOUT)
3512                 return NULL;
3513
3514         req->flags |= REQ_F_LINK_TIMEOUT;
3515         return nxt;
3516 }
3517
3518 static void __io_queue_sqe(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3519 {
3520         struct io_kiocb *linked_timeout;
3521         struct io_kiocb *nxt = NULL;
3522         int ret;
3523
3524 again:
3525         linked_timeout = io_prep_linked_timeout(req);
3526
3527         ret = io_issue_sqe(req, sqe, &nxt, true);
3528
3529         /*
3530          * We async punt it if the file wasn't marked NOWAIT, or if the file
3531          * doesn't support non-blocking read/write attempts
3532          */
3533         if (ret == -EAGAIN && (!(req->flags & REQ_F_NOWAIT) ||
3534             (req->flags & REQ_F_MUST_PUNT))) {
3535                 if (req->work.flags & IO_WQ_WORK_NEEDS_FILES) {
3536                         ret = io_grab_files(req);
3537                         if (ret)
3538                                 goto err;
3539                 }
3540
3541                 /*
3542                  * Queued up for async execution, worker will release
3543                  * submit reference when the iocb is actually submitted.
3544                  */
3545                 io_queue_async_work(req);
3546                 goto done_req;
3547         }
3548
3549 err:
3550         /* drop submission reference */
3551         io_put_req(req);
3552
3553         if (linked_timeout) {
3554                 if (!ret)
3555                         io_queue_linked_timeout(linked_timeout);
3556                 else
3557                         io_put_req(linked_timeout);
3558         }
3559
3560         /* and drop final reference, if we failed */
3561         if (ret) {
3562                 io_cqring_add_event(req, ret);
3563                 req_set_fail_links(req);
3564                 io_put_req(req);
3565         }
3566 done_req:
3567         if (nxt) {
3568                 req = nxt;
3569                 nxt = NULL;
3570                 goto again;
3571         }
3572 }
3573
3574 static void io_queue_sqe(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3575 {
3576         int ret;
3577
3578         if (unlikely(req->ctx->drain_next)) {
3579                 req->flags |= REQ_F_IO_DRAIN;
3580                 req->ctx->drain_next = false;
3581         }
3582         req->ctx->drain_next = (req->flags & REQ_F_DRAIN_LINK);
3583
3584         ret = io_req_defer(req, sqe);
3585         if (ret) {
3586                 if (ret != -EIOCBQUEUED) {
3587                         io_cqring_add_event(req, ret);
3588                         req_set_fail_links(req);
3589                         io_double_put_req(req);
3590                 }
3591         } else
3592                 __io_queue_sqe(req, sqe);
3593 }
3594
3595 static inline void io_queue_link_head(struct io_kiocb *req)
3596 {
3597         if (unlikely(req->flags & REQ_F_FAIL_LINK)) {
3598                 io_cqring_add_event(req, -ECANCELED);
3599                 io_double_put_req(req);
3600         } else
3601                 io_queue_sqe(req, NULL);
3602 }
3603
3604 #define SQE_VALID_FLAGS (IOSQE_FIXED_FILE|IOSQE_IO_DRAIN|IOSQE_IO_LINK| \
3605                                 IOSQE_IO_HARDLINK)
3606
3607 static bool io_submit_sqe(struct io_kiocb *req, const struct io_uring_sqe *sqe,
3608                           struct io_submit_state *state, struct io_kiocb **link)
3609 {
3610         struct io_ring_ctx *ctx = req->ctx;
3611         int ret;
3612
3613         /* enforce forwards compatibility on users */
3614         if (unlikely(sqe->flags & ~SQE_VALID_FLAGS)) {
3615                 ret = -EINVAL;
3616                 goto err_req;
3617         }
3618
3619         ret = io_req_set_file(state, req, sqe);
3620         if (unlikely(ret)) {
3621 err_req:
3622                 io_cqring_add_event(req, ret);
3623                 io_double_put_req(req);
3624                 return false;
3625         }
3626
3627         /*
3628          * If we already have a head request, queue this one for async
3629          * submittal once the head completes. If we don't have a head but
3630          * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be
3631          * submitted sync once the chain is complete. If none of those
3632          * conditions are true (normal request), then just queue it.
3633          */
3634         if (*link) {
3635                 struct io_kiocb *prev = *link;
3636
3637                 if (sqe->flags & IOSQE_IO_DRAIN)
3638                         (*link)->flags |= REQ_F_DRAIN_LINK | REQ_F_IO_DRAIN;
3639
3640                 if (sqe->flags & IOSQE_IO_HARDLINK)
3641                         req->flags |= REQ_F_HARDLINK;
3642
3643                 if (io_alloc_async_ctx(req)) {
3644                         ret = -EAGAIN;
3645                         goto err_req;
3646                 }
3647
3648                 ret = io_req_defer_prep(req, sqe);
3649                 if (ret) {
3650                         /* fail even hard links since we don't submit */
3651                         prev->flags |= REQ_F_FAIL_LINK;
3652                         goto err_req;
3653                 }
3654                 trace_io_uring_link(ctx, req, prev);
3655                 list_add_tail(&req->link_list, &prev->link_list);
3656         } else if (sqe->flags & (IOSQE_IO_LINK|IOSQE_IO_HARDLINK)) {
3657                 req->flags |= REQ_F_LINK;
3658                 if (sqe->flags & IOSQE_IO_HARDLINK)
3659                         req->flags |= REQ_F_HARDLINK;
3660
3661                 INIT_LIST_HEAD(&req->link_list);
3662                 ret = io_req_defer_prep(req, sqe);
3663                 if (ret)
3664                         req->flags |= REQ_F_FAIL_LINK;
3665                 *link = req;
3666         } else {
3667                 io_queue_sqe(req, sqe);
3668         }
3669
3670         return true;
3671 }
3672
3673 /*
3674  * Batched submission is done, ensure local IO is flushed out.
3675  */
3676 static void io_submit_state_end(struct io_submit_state *state)
3677 {
3678         blk_finish_plug(&state->plug);
3679         io_file_put(state);
3680         if (state->free_reqs)
3681                 kmem_cache_free_bulk(req_cachep, state->free_reqs,
3682                                         &state->reqs[state->cur_req]);
3683 }
3684
3685 /*
3686  * Start submission side cache.
3687  */
3688 static void io_submit_state_start(struct io_submit_state *state,
3689                                   unsigned int max_ios)
3690 {
3691         blk_start_plug(&state->plug);
3692         state->free_reqs = 0;
3693         state->file = NULL;
3694         state->ios_left = max_ios;
3695 }
3696
3697 static void io_commit_sqring(struct io_ring_ctx *ctx)
3698 {
3699         struct io_rings *rings = ctx->rings;
3700
3701         if (ctx->cached_sq_head != READ_ONCE(rings->sq.head)) {
3702                 /*
3703                  * Ensure any loads from the SQEs are done at this point,
3704                  * since once we write the new head, the application could
3705                  * write new data to them.
3706                  */
3707                 smp_store_release(&rings->sq.head, ctx->cached_sq_head);
3708         }
3709 }
3710
3711 /*
3712  * Fetch an sqe, if one is available. Note that sqe_ptr will point to memory
3713  * that is mapped by userspace. This means that care needs to be taken to
3714  * ensure that reads are stable, as we cannot rely on userspace always
3715  * being a good citizen. If members of the sqe are validated and then later
3716  * used, it's important that those reads are done through READ_ONCE() to
3717  * prevent a re-load down the line.
3718  */
3719 static bool io_get_sqring(struct io_ring_ctx *ctx, struct io_kiocb *req,
3720                           const struct io_uring_sqe **sqe_ptr)
3721 {
3722         struct io_rings *rings = ctx->rings;
3723         u32 *sq_array = ctx->sq_array;
3724         unsigned head;
3725
3726         /*
3727          * The cached sq head (or cq tail) serves two purposes:
3728          *
3729          * 1) allows us to batch the cost of updating the user visible
3730          *    head updates.
3731          * 2) allows the kernel side to track the head on its own, even
3732          *    though the application is the one updating it.
3733          */
3734         head = ctx->cached_sq_head;
3735         /* make sure SQ entry isn't read before tail */
3736         if (unlikely(head == smp_load_acquire(&rings->sq.tail)))
3737                 return false;
3738
3739         head = READ_ONCE(sq_array[head & ctx->sq_mask]);
3740         if (likely(head < ctx->sq_entries)) {
3741                 /*
3742                  * All io need record the previous position, if LINK vs DARIN,
3743                  * it can be used to mark the position of the first IO in the
3744                  * link list.
3745                  */
3746                 req->sequence = ctx->cached_sq_head;
3747                 *sqe_ptr = &ctx->sq_sqes[head];
3748                 req->opcode = READ_ONCE((*sqe_ptr)->opcode);
3749                 req->user_data = READ_ONCE((*sqe_ptr)->user_data);
3750                 ctx->cached_sq_head++;
3751                 return true;
3752         }
3753
3754         /* drop invalid entries */
3755         ctx->cached_sq_head++;
3756         ctx->cached_sq_dropped++;
3757         WRITE_ONCE(rings->sq_dropped, ctx->cached_sq_dropped);
3758         return false;
3759 }
3760
3761 static int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr,
3762                           struct file *ring_file, int ring_fd,
3763                           struct mm_struct **mm, bool async)
3764 {
3765         struct io_submit_state state, *statep = NULL;
3766         struct io_kiocb *link = NULL;
3767         int i, submitted = 0;
3768         bool mm_fault = false;
3769
3770         /* if we have a backlog and couldn't flush it all, return BUSY */
3771         if (!list_empty(&ctx->cq_overflow_list) &&
3772             !io_cqring_overflow_flush(ctx, false))
3773                 return -EBUSY;
3774
3775         if (nr > IO_PLUG_THRESHOLD) {
3776                 io_submit_state_start(&state, nr);
3777                 statep = &state;
3778         }
3779
3780         for (i = 0; i < nr; i++) {
3781                 const struct io_uring_sqe *sqe;
3782                 struct io_kiocb *req;
3783                 unsigned int sqe_flags;
3784
3785                 req = io_get_req(ctx, statep);
3786                 if (unlikely(!req)) {
3787                         if (!submitted)
3788                                 submitted = -EAGAIN;
3789                         break;
3790                 }
3791                 if (!io_get_sqring(ctx, req, &sqe)) {
3792                         __io_free_req(req);
3793                         break;
3794                 }
3795
3796                 if (io_req_needs_user(req) && !*mm) {
3797                         mm_fault = mm_fault || !mmget_not_zero(ctx->sqo_mm);
3798                         if (!mm_fault) {
3799                                 use_mm(ctx->sqo_mm);
3800                                 *mm = ctx->sqo_mm;
3801                         }
3802                 }
3803
3804                 submitted++;
3805                 sqe_flags = sqe->flags;
3806
3807                 req->ring_file = ring_file;
3808                 req->ring_fd = ring_fd;
3809                 req->has_user = *mm != NULL;
3810                 req->in_async = async;
3811                 req->needs_fixed_file = async;
3812                 trace_io_uring_submit_sqe(ctx, req->user_data, true, async);
3813                 if (!io_submit_sqe(req, sqe, statep, &link))
3814                         break;
3815                 /*
3816                  * If previous wasn't linked and we have a linked command,
3817                  * that's the end of the chain. Submit the previous link.
3818                  */
3819                 if (!(sqe_flags & (IOSQE_IO_LINK|IOSQE_IO_HARDLINK)) && link) {
3820                         io_queue_link_head(link);
3821                         link = NULL;
3822                 }
3823         }
3824
3825         if (link)
3826                 io_queue_link_head(link);
3827         if (statep)
3828                 io_submit_state_end(&state);
3829
3830          /* Commit SQ ring head once we've consumed and submitted all SQEs */
3831         io_commit_sqring(ctx);
3832
3833         return submitted;
3834 }
3835
3836 static int io_sq_thread(void *data)
3837 {
3838         struct io_ring_ctx *ctx = data;
3839         struct mm_struct *cur_mm = NULL;
3840         const struct cred *old_cred;
3841         mm_segment_t old_fs;
3842         DEFINE_WAIT(wait);
3843         unsigned inflight;
3844         unsigned long timeout;
3845         int ret;
3846
3847         complete(&ctx->completions[1]);
3848
3849         old_fs = get_fs();
3850         set_fs(USER_DS);
3851         old_cred = override_creds(ctx->creds);
3852
3853         ret = timeout = inflight = 0;
3854         while (!kthread_should_park()) {
3855                 unsigned int to_submit;
3856
3857                 if (inflight) {
3858                         unsigned nr_events = 0;
3859
3860                         if (ctx->flags & IORING_SETUP_IOPOLL) {
3861                                 /*
3862                                  * inflight is the count of the maximum possible
3863                                  * entries we submitted, but it can be smaller
3864                                  * if we dropped some of them. If we don't have
3865                                  * poll entries available, then we know that we
3866                                  * have nothing left to poll for. Reset the
3867                                  * inflight count to zero in that case.
3868                                  */
3869                                 mutex_lock(&ctx->uring_lock);
3870                                 if (!list_empty(&ctx->poll_list))
3871                                         __io_iopoll_check(ctx, &nr_events, 0);
3872                                 else
3873                                         inflight = 0;
3874                                 mutex_unlock(&ctx->uring_lock);
3875                         } else {
3876                                 /*
3877                                  * Normal IO, just pretend everything completed.
3878                                  * We don't have to poll completions for that.
3879                                  */
3880                                 nr_events = inflight;
3881                         }
3882
3883                         inflight -= nr_events;
3884                         if (!inflight)
3885                                 timeout = jiffies + ctx->sq_thread_idle;
3886                 }
3887
3888                 to_submit = io_sqring_entries(ctx);
3889
3890                 /*
3891                  * If submit got -EBUSY, flag us as needing the application
3892                  * to enter the kernel to reap and flush events.
3893                  */
3894                 if (!to_submit || ret == -EBUSY) {
3895                         /*
3896                          * We're polling. If we're within the defined idle
3897                          * period, then let us spin without work before going
3898                          * to sleep. The exception is if we got EBUSY doing
3899                          * more IO, we should wait for the application to
3900                          * reap events and wake us up.
3901                          */
3902                         if (inflight ||
3903                             (!time_after(jiffies, timeout) && ret != -EBUSY)) {
3904                                 cond_resched();
3905                                 continue;
3906                         }
3907
3908                         /*
3909                          * Drop cur_mm before scheduling, we can't hold it for
3910                          * long periods (or over schedule()). Do this before
3911                          * adding ourselves to the waitqueue, as the unuse/drop
3912                          * may sleep.
3913                          */
3914                         if (cur_mm) {
3915                                 unuse_mm(cur_mm);
3916                                 mmput(cur_mm);
3917                                 cur_mm = NULL;
3918                         }
3919
3920                         prepare_to_wait(&ctx->sqo_wait, &wait,
3921                                                 TASK_INTERRUPTIBLE);
3922
3923                         /* Tell userspace we may need a wakeup call */
3924                         ctx->rings->sq_flags |= IORING_SQ_NEED_WAKEUP;
3925                         /* make sure to read SQ tail after writing flags */
3926                         smp_mb();
3927
3928                         to_submit = io_sqring_entries(ctx);
3929                         if (!to_submit || ret == -EBUSY) {
3930                                 if (kthread_should_park()) {
3931                                         finish_wait(&ctx->sqo_wait, &wait);
3932                                         break;
3933                                 }
3934                                 if (signal_pending(current))
3935                                         flush_signals(current);
3936                                 schedule();
3937                                 finish_wait(&ctx->sqo_wait, &wait);
3938
3939                                 ctx->rings->sq_flags &= ~IORING_SQ_NEED_WAKEUP;
3940                                 continue;
3941                         }
3942                         finish_wait(&ctx->sqo_wait, &wait);
3943
3944                         ctx->rings->sq_flags &= ~IORING_SQ_NEED_WAKEUP;
3945                 }
3946
3947                 to_submit = min(to_submit, ctx->sq_entries);
3948                 mutex_lock(&ctx->uring_lock);
3949                 ret = io_submit_sqes(ctx, to_submit, NULL, -1, &cur_mm, true);
3950                 mutex_unlock(&ctx->uring_lock);
3951                 if (ret > 0)
3952                         inflight += ret;
3953         }
3954
3955         set_fs(old_fs);
3956         if (cur_mm) {
3957                 unuse_mm(cur_mm);
3958                 mmput(cur_mm);
3959         }
3960         revert_creds(old_cred);
3961
3962         kthread_parkme();
3963
3964         return 0;
3965 }
3966
3967 struct io_wait_queue {
3968         struct wait_queue_entry wq;
3969         struct io_ring_ctx *ctx;
3970         unsigned to_wait;
3971         unsigned nr_timeouts;
3972 };
3973
3974 static inline bool io_should_wake(struct io_wait_queue *iowq, bool noflush)
3975 {
3976         struct io_ring_ctx *ctx = iowq->ctx;
3977
3978         /*
3979          * Wake up if we have enough events, or if a timeout occurred since we
3980          * started waiting. For timeouts, we always want to return to userspace,
3981          * regardless of event count.
3982          */
3983         return io_cqring_events(ctx, noflush) >= iowq->to_wait ||
3984                         atomic_read(&ctx->cq_timeouts) != iowq->nr_timeouts;
3985 }
3986
3987 static int io_wake_function(struct wait_queue_entry *curr, unsigned int mode,
3988                             int wake_flags, void *key)
3989 {
3990         struct io_wait_queue *iowq = container_of(curr, struct io_wait_queue,
3991                                                         wq);
3992
3993         /* use noflush == true, as we can't safely rely on locking context */
3994         if (!io_should_wake(iowq, true))
3995                 return -1;
3996
3997         return autoremove_wake_function(curr, mode, wake_flags, key);
3998 }
3999
4000 /*
4001  * Wait until events become available, if we don't already have some. The
4002  * application must reap them itself, as they reside on the shared cq ring.
4003  */
4004 static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
4005                           const sigset_t __user *sig, size_t sigsz)
4006 {
4007         struct io_wait_queue iowq = {
4008                 .wq = {
4009                         .private        = current,
4010                         .func           = io_wake_function,
4011                         .entry          = LIST_HEAD_INIT(iowq.wq.entry),
4012                 },
4013                 .ctx            = ctx,
4014                 .to_wait        = min_events,
4015         };
4016         struct io_rings *rings = ctx->rings;
4017         int ret = 0;
4018
4019         if (io_cqring_events(ctx, false) >= min_events)
4020                 return 0;
4021
4022         if (sig) {
4023 #ifdef CONFIG_COMPAT
4024                 if (in_compat_syscall())
4025                         ret = set_compat_user_sigmask((const compat_sigset_t __user *)sig,
4026                                                       sigsz);
4027                 else
4028 #endif
4029                         ret = set_user_sigmask(sig, sigsz);
4030
4031                 if (ret)
4032                         return ret;
4033         }
4034
4035         iowq.nr_timeouts = atomic_read(&ctx->cq_timeouts);
4036         trace_io_uring_cqring_wait(ctx, min_events);
4037         do {
4038                 prepare_to_wait_exclusive(&ctx->wait, &iowq.wq,
4039                                                 TASK_INTERRUPTIBLE);
4040                 if (io_should_wake(&iowq, false))
4041                         break;
4042                 schedule();
4043                 if (signal_pending(current)) {
4044                         ret = -EINTR;
4045                         break;
4046                 }
4047         } while (1);
4048         finish_wait(&ctx->wait, &iowq.wq);
4049
4050         restore_saved_sigmask_unless(ret == -EINTR);
4051
4052         return READ_ONCE(rings->cq.head) == READ_ONCE(rings->cq.tail) ? ret : 0;
4053 }
4054
4055 static void __io_sqe_files_unregister(struct io_ring_ctx *ctx)
4056 {
4057 #if defined(CONFIG_UNIX)
4058         if (ctx->ring_sock) {
4059                 struct sock *sock = ctx->ring_sock->sk;
4060                 struct sk_buff *skb;
4061
4062                 while ((skb = skb_dequeue(&sock->sk_receive_queue)) != NULL)
4063                         kfree_skb(skb);
4064         }
4065 #else
4066         int i;
4067
4068         for (i = 0; i < ctx->nr_user_files; i++) {
4069                 struct file *file;
4070
4071                 file = io_file_from_index(ctx, i);
4072                 if (file)
4073                         fput(file);
4074         }
4075 #endif
4076 }
4077
4078 static int io_sqe_files_unregister(struct io_ring_ctx *ctx)
4079 {
4080         unsigned nr_tables, i;
4081
4082         if (!ctx->file_table)
4083                 return -ENXIO;
4084
4085         __io_sqe_files_unregister(ctx);
4086         nr_tables = DIV_ROUND_UP(ctx->nr_user_files, IORING_MAX_FILES_TABLE);
4087         for (i = 0; i < nr_tables; i++)
4088                 kfree(ctx->file_table[i].files);
4089         kfree(ctx->file_table);
4090         ctx->file_table = NULL;
4091         ctx->nr_user_files = 0;
4092         return 0;
4093 }
4094
4095 static void io_sq_thread_stop(struct io_ring_ctx *ctx)
4096 {
4097         if (ctx->sqo_thread) {
4098                 wait_for_completion(&ctx->completions[1]);
4099                 /*
4100                  * The park is a bit of a work-around, without it we get
4101                  * warning spews on shutdown with SQPOLL set and affinity
4102                  * set to a single CPU.
4103                  */
4104                 kthread_park(ctx->sqo_thread);
4105                 kthread_stop(ctx->sqo_thread);
4106                 ctx->sqo_thread = NULL;
4107         }
4108 }
4109
4110 static void io_finish_async(struct io_ring_ctx *ctx)
4111 {
4112         io_sq_thread_stop(ctx);
4113
4114         if (ctx->io_wq) {
4115                 io_wq_destroy(ctx->io_wq);
4116                 ctx->io_wq = NULL;
4117         }
4118 }
4119
4120 #if defined(CONFIG_UNIX)
4121 static void io_destruct_skb(struct sk_buff *skb)
4122 {
4123         struct io_ring_ctx *ctx = skb->sk->sk_user_data;
4124
4125         if (ctx->io_wq)
4126                 io_wq_flush(ctx->io_wq);
4127
4128         unix_destruct_scm(skb);
4129 }
4130
4131 /*
4132  * Ensure the UNIX gc is aware of our file set, so we are certain that
4133  * the io_uring can be safely unregistered on process exit, even if we have
4134  * loops in the file referencing.
4135  */
4136 static int __io_sqe_files_scm(struct io_ring_ctx *ctx, int nr, int offset)
4137 {
4138         struct sock *sk = ctx->ring_sock->sk;
4139         struct scm_fp_list *fpl;
4140         struct sk_buff *skb;
4141         int i, nr_files;
4142
4143         if (!capable(CAP_SYS_RESOURCE) && !capable(CAP_SYS_ADMIN)) {
4144                 unsigned long inflight = ctx->user->unix_inflight + nr;
4145
4146                 if (inflight > task_rlimit(current, RLIMIT_NOFILE))
4147                         return -EMFILE;
4148         }
4149
4150         fpl = kzalloc(sizeof(*fpl), GFP_KERNEL);
4151         if (!fpl)
4152                 return -ENOMEM;
4153
4154         skb = alloc_skb(0, GFP_KERNEL);
4155         if (!skb) {
4156                 kfree(fpl);
4157                 return -ENOMEM;
4158         }
4159
4160         skb->sk = sk;
4161
4162         nr_files = 0;
4163         fpl->user = get_uid(ctx->user);
4164         for (i = 0; i < nr; i++) {
4165                 struct file *file = io_file_from_index(ctx, i + offset);
4166
4167                 if (!file)
4168                         continue;
4169                 fpl->fp[nr_files] = get_file(file);
4170                 unix_inflight(fpl->user, fpl->fp[nr_files]);
4171                 nr_files++;
4172         }
4173
4174         if (nr_files) {
4175                 fpl->max = SCM_MAX_FD;
4176                 fpl->count = nr_files;
4177                 UNIXCB(skb).fp = fpl;
4178                 skb->destructor = io_destruct_skb;
4179                 refcount_add(skb->truesize, &sk->sk_wmem_alloc);
4180                 skb_queue_head(&sk->sk_receive_queue, skb);
4181
4182                 for (i = 0; i < nr_files; i++)
4183                         fput(fpl->fp[i]);
4184         } else {
4185                 kfree_skb(skb);
4186                 kfree(fpl);
4187         }
4188
4189         return 0;
4190 }
4191
4192 /*
4193  * If UNIX sockets are enabled, fd passing can cause a reference cycle which
4194  * causes regular reference counting to break down. We rely on the UNIX
4195  * garbage collection to take care of this problem for us.
4196  */
4197 static int io_sqe_files_scm(struct io_ring_ctx *ctx)
4198 {
4199         unsigned left, total;
4200         int ret = 0;
4201
4202         total = 0;
4203         left = ctx->nr_user_files;
4204         while (left) {
4205                 unsigned this_files = min_t(unsigned, left, SCM_MAX_FD);
4206
4207                 ret = __io_sqe_files_scm(ctx, this_files, total);
4208                 if (ret)
4209                         break;
4210                 left -= this_files;
4211                 total += this_files;
4212         }
4213
4214         if (!ret)
4215                 return 0;
4216
4217         while (total < ctx->nr_user_files) {
4218                 struct file *file = io_file_from_index(ctx, total);
4219
4220                 if (file)
4221                         fput(file);
4222                 total++;
4223         }
4224
4225         return ret;
4226 }
4227 #else
4228 static int io_sqe_files_scm(struct io_ring_ctx *ctx)
4229 {
4230         return 0;
4231 }
4232 #endif
4233
4234 static int io_sqe_alloc_file_tables(struct io_ring_ctx *ctx, unsigned nr_tables,
4235                                     unsigned nr_files)
4236 {
4237         int i;
4238
4239         for (i = 0; i < nr_tables; i++) {
4240                 struct fixed_file_table *table = &ctx->file_table[i];
4241                 unsigned this_files;
4242
4243                 this_files = min(nr_files, IORING_MAX_FILES_TABLE);
4244                 table->files = kcalloc(this_files, sizeof(struct file *),
4245                                         GFP_KERNEL);
4246                 if (!table->files)
4247                         break;
4248                 nr_files -= this_files;
4249         }
4250
4251         if (i == nr_tables)
4252                 return 0;
4253
4254         for (i = 0; i < nr_tables; i++) {
4255                 struct fixed_file_table *table = &ctx->file_table[i];
4256                 kfree(table->files);
4257         }
4258         return 1;
4259 }
4260
4261 static int io_sqe_files_register(struct io_ring_ctx *ctx, void __user *arg,
4262                                  unsigned nr_args)
4263 {
4264         __s32 __user *fds = (__s32 __user *) arg;
4265         unsigned nr_tables;
4266         int fd, ret = 0;
4267         unsigned i;
4268
4269         if (ctx->file_table)
4270                 return -EBUSY;
4271         if (!nr_args)
4272                 return -EINVAL;
4273         if (nr_args > IORING_MAX_FIXED_FILES)
4274                 return -EMFILE;
4275
4276         nr_tables = DIV_ROUND_UP(nr_args, IORING_MAX_FILES_TABLE);
4277         ctx->file_table = kcalloc(nr_tables, sizeof(struct fixed_file_table),
4278                                         GFP_KERNEL);
4279         if (!ctx->file_table)
4280                 return -ENOMEM;
4281
4282         if (io_sqe_alloc_file_tables(ctx, nr_tables, nr_args)) {
4283                 kfree(ctx->file_table);
4284                 ctx->file_table = NULL;
4285                 return -ENOMEM;
4286         }
4287
4288         for (i = 0; i < nr_args; i++, ctx->nr_user_files++) {
4289                 struct fixed_file_table *table;
4290                 unsigned index;
4291
4292                 ret = -EFAULT;
4293                 if (copy_from_user(&fd, &fds[i], sizeof(fd)))
4294                         break;
4295                 /* allow sparse sets */
4296                 if (fd == -1) {
4297                         ret = 0;
4298                         continue;
4299                 }
4300
4301                 table = &ctx->file_table[i >> IORING_FILE_TABLE_SHIFT];
4302                 index = i & IORING_FILE_TABLE_MASK;
4303                 table->files[index] = fget(fd);
4304
4305                 ret = -EBADF;
4306                 if (!table->files[index])
4307                         break;
4308                 /*
4309                  * Don't allow io_uring instances to be registered. If UNIX
4310                  * isn't enabled, then this causes a reference cycle and this
4311                  * instance can never get freed. If UNIX is enabled we'll
4312                  * handle it just fine, but there's still no point in allowing
4313                  * a ring fd as it doesn't support regular read/write anyway.
4314                  */
4315                 if (table->files[index]->f_op == &io_uring_fops) {
4316                         fput(table->files[index]);
4317                         break;
4318                 }
4319                 ret = 0;
4320         }
4321
4322         if (ret) {
4323                 for (i = 0; i < ctx->nr_user_files; i++) {
4324                         struct file *file;
4325
4326                         file = io_file_from_index(ctx, i);
4327                         if (file)
4328                                 fput(file);
4329                 }
4330                 for (i = 0; i < nr_tables; i++)
4331                         kfree(ctx->file_table[i].files);
4332
4333                 kfree(ctx->file_table);
4334                 ctx->file_table = NULL;
4335                 ctx->nr_user_files = 0;
4336                 return ret;
4337         }
4338
4339         ret = io_sqe_files_scm(ctx);
4340         if (ret)
4341                 io_sqe_files_unregister(ctx);
4342
4343         return ret;
4344 }
4345
4346 static void io_sqe_file_unregister(struct io_ring_ctx *ctx, int index)
4347 {
4348 #if defined(CONFIG_UNIX)
4349         struct file *file = io_file_from_index(ctx, index);
4350         struct sock *sock = ctx->ring_sock->sk;
4351         struct sk_buff_head list, *head = &sock->sk_receive_queue;
4352         struct sk_buff *skb;
4353         int i;
4354
4355         __skb_queue_head_init(&list);
4356
4357         /*
4358          * Find the skb that holds this file in its SCM_RIGHTS. When found,
4359          * remove this entry and rearrange the file array.
4360          */
4361         skb = skb_dequeue(head);
4362         while (skb) {
4363                 struct scm_fp_list *fp;
4364
4365                 fp = UNIXCB(skb).fp;
4366                 for (i = 0; i < fp->count; i++) {
4367                         int left;
4368
4369                         if (fp->fp[i] != file)
4370                                 continue;
4371
4372                         unix_notinflight(fp->user, fp->fp[i]);
4373                         left = fp->count - 1 - i;
4374                         if (left) {
4375                                 memmove(&fp->fp[i], &fp->fp[i + 1],
4376                                                 left * sizeof(struct file *));
4377                         }
4378                         fp->count--;
4379                         if (!fp->count) {
4380                                 kfree_skb(skb);
4381                                 skb = NULL;
4382                         } else {
4383                                 __skb_queue_tail(&list, skb);
4384                         }
4385                         fput(file);
4386                         file = NULL;
4387                         break;
4388                 }
4389
4390                 if (!file)
4391                         break;
4392
4393                 __skb_queue_tail(&list, skb);
4394
4395                 skb = skb_dequeue(head);
4396         }
4397
4398         if (skb_peek(&list)) {
4399                 spin_lock_irq(&head->lock);
4400                 while ((skb = __skb_dequeue(&list)) != NULL)
4401                         __skb_queue_tail(head, skb);
4402                 spin_unlock_irq(&head->lock);
4403         }
4404 #else
4405         fput(io_file_from_index(ctx, index));
4406 #endif
4407 }
4408
4409 static int io_sqe_file_register(struct io_ring_ctx *ctx, struct file *file,
4410                                 int index)
4411 {
4412 #if defined(CONFIG_UNIX)
4413         struct sock *sock = ctx->ring_sock->sk;
4414         struct sk_buff_head *head = &sock->sk_receive_queue;
4415         struct sk_buff *skb;
4416
4417         /*
4418          * See if we can merge this file into an existing skb SCM_RIGHTS
4419          * file set. If there's no room, fall back to allocating a new skb
4420          * and filling it in.
4421          */
4422         spin_lock_irq(&head->lock);
4423         skb = skb_peek(head);
4424         if (skb) {
4425                 struct scm_fp_list *fpl = UNIXCB(skb).fp;
4426
4427                 if (fpl->count < SCM_MAX_FD) {
4428                         __skb_unlink(skb, head);
4429                         spin_unlock_irq(&head->lock);
4430                         fpl->fp[fpl->count] = get_file(file);
4431                         unix_inflight(fpl->user, fpl->fp[fpl->count]);
4432                         fpl->count++;
4433                         spin_lock_irq(&head->lock);
4434                         __skb_queue_head(head, skb);
4435                 } else {
4436                         skb = NULL;
4437                 }
4438         }
4439         spin_unlock_irq(&head->lock);
4440
4441         if (skb) {
4442                 fput(file);
4443                 return 0;
4444         }
4445
4446         return __io_sqe_files_scm(ctx, 1, index);
4447 #else
4448         return 0;
4449 #endif
4450 }
4451
4452 static int io_sqe_files_update(struct io_ring_ctx *ctx, void __user *arg,
4453                                unsigned nr_args)
4454 {
4455         struct io_uring_files_update up;
4456         __s32 __user *fds;
4457         int fd, i, err;
4458         __u32 done;
4459
4460         if (!ctx->file_table)
4461                 return -ENXIO;
4462         if (!nr_args)
4463                 return -EINVAL;
4464         if (copy_from_user(&up, arg, sizeof(up)))
4465                 return -EFAULT;
4466         if (check_add_overflow(up.offset, nr_args, &done))
4467                 return -EOVERFLOW;
4468         if (done > ctx->nr_user_files)
4469                 return -EINVAL;
4470
4471         done = 0;
4472         fds = (__s32 __user *) up.fds;
4473         while (nr_args) {
4474                 struct fixed_file_table *table;
4475                 unsigned index;
4476
4477                 err = 0;
4478                 if (copy_from_user(&fd, &fds[done], sizeof(fd))) {
4479                         err = -EFAULT;
4480                         break;
4481                 }
4482                 i = array_index_nospec(up.offset, ctx->nr_user_files);
4483                 table = &ctx->file_table[i >> IORING_FILE_TABLE_SHIFT];
4484                 index = i & IORING_FILE_TABLE_MASK;
4485                 if (table->files[index]) {
4486                         io_sqe_file_unregister(ctx, i);
4487                         table->files[index] = NULL;
4488                 }
4489                 if (fd != -1) {
4490                         struct file *file;
4491
4492                         file = fget(fd);
4493                         if (!file) {
4494                                 err = -EBADF;
4495                                 break;
4496                         }
4497                         /*
4498                          * Don't allow io_uring instances to be registered. If
4499                          * UNIX isn't enabled, then this causes a reference
4500                          * cycle and this instance can never get freed. If UNIX
4501                          * is enabled we'll handle it just fine, but there's
4502                          * still no point in allowing a ring fd as it doesn't
4503                          * support regular read/write anyway.
4504                          */
4505                         if (file->f_op == &io_uring_fops) {
4506                                 fput(file);
4507                                 err = -EBADF;
4508                                 break;
4509                         }
4510                         table->files[index] = file;
4511                         err = io_sqe_file_register(ctx, file, i);
4512                         if (err)
4513                                 break;
4514                 }
4515                 nr_args--;
4516                 done++;
4517                 up.offset++;
4518         }
4519
4520         return done ? done : err;
4521 }
4522
4523 static void io_put_work(struct io_wq_work *work)
4524 {
4525         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
4526
4527         io_put_req(req);
4528 }
4529
4530 static void io_get_work(struct io_wq_work *work)
4531 {
4532         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
4533
4534         refcount_inc(&req->refs);
4535 }
4536
4537 static int io_sq_offload_start(struct io_ring_ctx *ctx,
4538                                struct io_uring_params *p)
4539 {
4540         struct io_wq_data data;
4541         unsigned concurrency;
4542         int ret;
4543
4544         init_waitqueue_head(&ctx->sqo_wait);
4545         mmgrab(current->mm);
4546         ctx->sqo_mm = current->mm;
4547
4548         if (ctx->flags & IORING_SETUP_SQPOLL) {
4549                 ret = -EPERM;
4550                 if (!capable(CAP_SYS_ADMIN))
4551                         goto err;
4552
4553                 ctx->sq_thread_idle = msecs_to_jiffies(p->sq_thread_idle);
4554                 if (!ctx->sq_thread_idle)
4555                         ctx->sq_thread_idle = HZ;
4556
4557                 if (p->flags & IORING_SETUP_SQ_AFF) {
4558                         int cpu = p->sq_thread_cpu;
4559
4560                         ret = -EINVAL;
4561                         if (cpu >= nr_cpu_ids)
4562                                 goto err;
4563                         if (!cpu_online(cpu))
4564                                 goto err;
4565
4566                         ctx->sqo_thread = kthread_create_on_cpu(io_sq_thread,
4567                                                         ctx, cpu,
4568                                                         "io_uring-sq");
4569                 } else {
4570                         ctx->sqo_thread = kthread_create(io_sq_thread, ctx,
4571                                                         "io_uring-sq");
4572                 }
4573                 if (IS_ERR(ctx->sqo_thread)) {
4574                         ret = PTR_ERR(ctx->sqo_thread);
4575                         ctx->sqo_thread = NULL;
4576                         goto err;
4577                 }
4578                 wake_up_process(ctx->sqo_thread);
4579         } else if (p->flags & IORING_SETUP_SQ_AFF) {
4580                 /* Can't have SQ_AFF without SQPOLL */
4581                 ret = -EINVAL;
4582                 goto err;
4583         }
4584
4585         data.mm = ctx->sqo_mm;
4586         data.user = ctx->user;
4587         data.creds = ctx->creds;
4588         data.get_work = io_get_work;
4589         data.put_work = io_put_work;
4590
4591         /* Do QD, or 4 * CPUS, whatever is smallest */
4592         concurrency = min(ctx->sq_entries, 4 * num_online_cpus());
4593         ctx->io_wq = io_wq_create(concurrency, &data);
4594         if (IS_ERR(ctx->io_wq)) {
4595                 ret = PTR_ERR(ctx->io_wq);
4596                 ctx->io_wq = NULL;
4597                 goto err;
4598         }
4599
4600         return 0;
4601 err:
4602         io_finish_async(ctx);
4603         mmdrop(ctx->sqo_mm);
4604         ctx->sqo_mm = NULL;
4605         return ret;
4606 }
4607
4608 static void io_unaccount_mem(struct user_struct *user, unsigned long nr_pages)
4609 {
4610         atomic_long_sub(nr_pages, &user->locked_vm);
4611 }
4612
4613 static int io_account_mem(struct user_struct *user, unsigned long nr_pages)
4614 {
4615         unsigned long page_limit, cur_pages, new_pages;
4616
4617         /* Don't allow more pages than we can safely lock */
4618         page_limit = rlimit(RLIMIT_MEMLOCK) >> PAGE_SHIFT;
4619
4620         do {
4621                 cur_pages = atomic_long_read(&user->locked_vm);
4622                 new_pages = cur_pages + nr_pages;
4623                 if (new_pages > page_limit)
4624                         return -ENOMEM;
4625         } while (atomic_long_cmpxchg(&user->locked_vm, cur_pages,
4626                                         new_pages) != cur_pages);
4627
4628         return 0;
4629 }
4630
4631 static void io_mem_free(void *ptr)
4632 {
4633         struct page *page;
4634
4635         if (!ptr)
4636                 return;
4637
4638         page = virt_to_head_page(ptr);
4639         if (put_page_testzero(page))
4640                 free_compound_page(page);
4641 }
4642
4643 static void *io_mem_alloc(size_t size)
4644 {
4645         gfp_t gfp_flags = GFP_KERNEL | __GFP_ZERO | __GFP_NOWARN | __GFP_COMP |
4646                                 __GFP_NORETRY;
4647
4648         return (void *) __get_free_pages(gfp_flags, get_order(size));
4649 }
4650
4651 static unsigned long rings_size(unsigned sq_entries, unsigned cq_entries,
4652                                 size_t *sq_offset)
4653 {
4654         struct io_rings *rings;
4655         size_t off, sq_array_size;
4656
4657         off = struct_size(rings, cqes, cq_entries);
4658         if (off == SIZE_MAX)
4659                 return SIZE_MAX;
4660
4661 #ifdef CONFIG_SMP
4662         off = ALIGN(off, SMP_CACHE_BYTES);
4663         if (off == 0)
4664                 return SIZE_MAX;
4665 #endif
4666
4667         sq_array_size = array_size(sizeof(u32), sq_entries);
4668         if (sq_array_size == SIZE_MAX)
4669                 return SIZE_MAX;
4670
4671         if (check_add_overflow(off, sq_array_size, &off))
4672                 return SIZE_MAX;
4673
4674         if (sq_offset)
4675                 *sq_offset = off;
4676
4677         return off;
4678 }
4679
4680 static unsigned long ring_pages(unsigned sq_entries, unsigned cq_entries)
4681 {
4682         size_t pages;
4683
4684         pages = (size_t)1 << get_order(
4685                 rings_size(sq_entries, cq_entries, NULL));
4686         pages += (size_t)1 << get_order(
4687                 array_size(sizeof(struct io_uring_sqe), sq_entries));
4688
4689         return pages;
4690 }
4691
4692 static int io_sqe_buffer_unregister(struct io_ring_ctx *ctx)
4693 {
4694         int i, j;
4695
4696         if (!ctx->user_bufs)
4697                 return -ENXIO;
4698
4699         for (i = 0; i < ctx->nr_user_bufs; i++) {
4700                 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
4701
4702                 for (j = 0; j < imu->nr_bvecs; j++)
4703                         put_user_page(imu->bvec[j].bv_page);
4704
4705                 if (ctx->account_mem)
4706                         io_unaccount_mem(ctx->user, imu->nr_bvecs);
4707                 kvfree(imu->bvec);
4708                 imu->nr_bvecs = 0;
4709         }
4710
4711         kfree(ctx->user_bufs);
4712         ctx->user_bufs = NULL;
4713         ctx->nr_user_bufs = 0;
4714         return 0;
4715 }
4716
4717 static int io_copy_iov(struct io_ring_ctx *ctx, struct iovec *dst,
4718                        void __user *arg, unsigned index)
4719 {
4720         struct iovec __user *src;
4721
4722 #ifdef CONFIG_COMPAT
4723         if (ctx->compat) {
4724                 struct compat_iovec __user *ciovs;
4725                 struct compat_iovec ciov;
4726
4727                 ciovs = (struct compat_iovec __user *) arg;
4728                 if (copy_from_user(&ciov, &ciovs[index], sizeof(ciov)))
4729                         return -EFAULT;
4730
4731                 dst->iov_base = u64_to_user_ptr((u64)ciov.iov_base);
4732                 dst->iov_len = ciov.iov_len;
4733                 return 0;
4734         }
4735 #endif
4736         src = (struct iovec __user *) arg;
4737         if (copy_from_user(dst, &src[index], sizeof(*dst)))
4738                 return -EFAULT;
4739         return 0;
4740 }
4741
4742 static int io_sqe_buffer_register(struct io_ring_ctx *ctx, void __user *arg,
4743                                   unsigned nr_args)
4744 {
4745         struct vm_area_struct **vmas = NULL;
4746         struct page **pages = NULL;
4747         int i, j, got_pages = 0;
4748         int ret = -EINVAL;
4749
4750         if (ctx->user_bufs)
4751                 return -EBUSY;
4752         if (!nr_args || nr_args > UIO_MAXIOV)
4753                 return -EINVAL;
4754
4755         ctx->user_bufs = kcalloc(nr_args, sizeof(struct io_mapped_ubuf),
4756                                         GFP_KERNEL);
4757         if (!ctx->user_bufs)
4758                 return -ENOMEM;
4759
4760         for (i = 0; i < nr_args; i++) {
4761                 struct io_mapped_ubuf *imu = &ctx->user_bufs[i];
4762                 unsigned long off, start, end, ubuf;
4763                 int pret, nr_pages;
4764                 struct iovec iov;
4765                 size_t size;
4766
4767                 ret = io_copy_iov(ctx, &iov, arg, i);
4768                 if (ret)
4769                         goto err;
4770
4771                 /*
4772                  * Don't impose further limits on the size and buffer
4773                  * constraints here, we'll -EINVAL later when IO is
4774                  * submitted if they are wrong.
4775                  */
4776                 ret = -EFAULT;
4777                 if (!iov.iov_base || !iov.iov_len)
4778                         goto err;
4779
4780                 /* arbitrary limit, but we need something */
4781                 if (iov.iov_len > SZ_1G)
4782                         goto err;
4783
4784                 ubuf = (unsigned long) iov.iov_base;
4785                 end = (ubuf + iov.iov_len + PAGE_SIZE - 1) >> PAGE_SHIFT;
4786                 start = ubuf >> PAGE_SHIFT;
4787                 nr_pages = end - start;
4788
4789                 if (ctx->account_mem) {
4790                         ret = io_account_mem(ctx->user, nr_pages);
4791                         if (ret)
4792                                 goto err;
4793                 }
4794
4795                 ret = 0;
4796                 if (!pages || nr_pages > got_pages) {
4797                         kfree(vmas);
4798                         kfree(pages);
4799                         pages = kvmalloc_array(nr_pages, sizeof(struct page *),
4800                                                 GFP_KERNEL);
4801                         vmas = kvmalloc_array(nr_pages,
4802                                         sizeof(struct vm_area_struct *),
4803                                         GFP_KERNEL);
4804                         if (!pages || !vmas) {
4805                                 ret = -ENOMEM;
4806                                 if (ctx->account_mem)
4807                                         io_unaccount_mem(ctx->user, nr_pages);
4808                                 goto err;
4809                         }
4810                         got_pages = nr_pages;
4811                 }
4812
4813                 imu->bvec = kvmalloc_array(nr_pages, sizeof(struct bio_vec),
4814                                                 GFP_KERNEL);
4815                 ret = -ENOMEM;
4816                 if (!imu->bvec) {
4817                         if (ctx->account_mem)
4818                                 io_unaccount_mem(ctx->user, nr_pages);
4819                         goto err;
4820                 }
4821
4822                 ret = 0;
4823                 down_read(&current->mm->mmap_sem);
4824                 pret = get_user_pages(ubuf, nr_pages,
4825                                       FOLL_WRITE | FOLL_LONGTERM,
4826                                       pages, vmas);
4827                 if (pret == nr_pages) {
4828                         /* don't support file backed memory */
4829                         for (j = 0; j < nr_pages; j++) {
4830                                 struct vm_area_struct *vma = vmas[j];
4831
4832                                 if (vma->vm_file &&
4833                                     !is_file_hugepages(vma->vm_file)) {
4834                                         ret = -EOPNOTSUPP;
4835                                         break;
4836                                 }
4837                         }
4838                 } else {
4839                         ret = pret < 0 ? pret : -EFAULT;
4840                 }
4841                 up_read(&current->mm->mmap_sem);
4842                 if (ret) {
4843                         /*
4844                          * if we did partial map, or found file backed vmas,
4845                          * release any pages we did get
4846                          */
4847                         if (pret > 0)
4848                                 put_user_pages(pages, pret);
4849                         if (ctx->account_mem)
4850                                 io_unaccount_mem(ctx->user, nr_pages);
4851                         kvfree(imu->bvec);
4852                         goto err;
4853                 }
4854
4855                 off = ubuf & ~PAGE_MASK;
4856                 size = iov.iov_len;
4857                 for (j = 0; j < nr_pages; j++) {
4858                         size_t vec_len;
4859
4860                         vec_len = min_t(size_t, size, PAGE_SIZE - off);
4861                         imu->bvec[j].bv_page = pages[j];
4862                         imu->bvec[j].bv_len = vec_len;
4863                         imu->bvec[j].bv_offset = off;
4864                         off = 0;
4865                         size -= vec_len;
4866                 }
4867                 /* store original address for later verification */
4868                 imu->ubuf = ubuf;
4869                 imu->len = iov.iov_len;
4870                 imu->nr_bvecs = nr_pages;
4871
4872                 ctx->nr_user_bufs++;
4873         }
4874         kvfree(pages);
4875         kvfree(vmas);
4876         return 0;
4877 err:
4878         kvfree(pages);
4879         kvfree(vmas);
4880         io_sqe_buffer_unregister(ctx);
4881         return ret;
4882 }
4883
4884 static int io_eventfd_register(struct io_ring_ctx *ctx, void __user *arg)
4885 {
4886         __s32 __user *fds = arg;
4887         int fd;
4888
4889         if (ctx->cq_ev_fd)
4890                 return -EBUSY;
4891
4892         if (copy_from_user(&fd, fds, sizeof(*fds)))
4893                 return -EFAULT;
4894
4895         ctx->cq_ev_fd = eventfd_ctx_fdget(fd);
4896         if (IS_ERR(ctx->cq_ev_fd)) {
4897                 int ret = PTR_ERR(ctx->cq_ev_fd);
4898                 ctx->cq_ev_fd = NULL;
4899                 return ret;
4900         }
4901
4902         return 0;
4903 }
4904
4905 static int io_eventfd_unregister(struct io_ring_ctx *ctx)
4906 {
4907         if (ctx->cq_ev_fd) {
4908                 eventfd_ctx_put(ctx->cq_ev_fd);
4909                 ctx->cq_ev_fd = NULL;
4910                 return 0;
4911         }
4912
4913         return -ENXIO;
4914 }
4915
4916 static void io_ring_ctx_free(struct io_ring_ctx *ctx)
4917 {
4918         io_finish_async(ctx);
4919         if (ctx->sqo_mm)
4920                 mmdrop(ctx->sqo_mm);
4921
4922         io_iopoll_reap_events(ctx);
4923         io_sqe_buffer_unregister(ctx);
4924         io_sqe_files_unregister(ctx);
4925         io_eventfd_unregister(ctx);
4926
4927 #if defined(CONFIG_UNIX)
4928         if (ctx->ring_sock) {
4929                 ctx->ring_sock->file = NULL; /* so that iput() is called */
4930                 sock_release(ctx->ring_sock);
4931         }
4932 #endif
4933
4934         io_mem_free(ctx->rings);
4935         io_mem_free(ctx->sq_sqes);
4936
4937         percpu_ref_exit(&ctx->refs);
4938         if (ctx->account_mem)
4939                 io_unaccount_mem(ctx->user,
4940                                 ring_pages(ctx->sq_entries, ctx->cq_entries));
4941         free_uid(ctx->user);
4942         put_cred(ctx->creds);
4943         kfree(ctx->completions);
4944         kfree(ctx->cancel_hash);
4945         kmem_cache_free(req_cachep, ctx->fallback_req);
4946         kfree(ctx);
4947 }
4948
4949 static __poll_t io_uring_poll(struct file *file, poll_table *wait)
4950 {
4951         struct io_ring_ctx *ctx = file->private_data;
4952         __poll_t mask = 0;
4953
4954         poll_wait(file, &ctx->cq_wait, wait);
4955         /*
4956          * synchronizes with barrier from wq_has_sleeper call in
4957          * io_commit_cqring
4958          */
4959         smp_rmb();
4960         if (READ_ONCE(ctx->rings->sq.tail) - ctx->cached_sq_head !=
4961             ctx->rings->sq_ring_entries)
4962                 mask |= EPOLLOUT | EPOLLWRNORM;
4963         if (READ_ONCE(ctx->rings->cq.head) != ctx->cached_cq_tail)
4964                 mask |= EPOLLIN | EPOLLRDNORM;
4965
4966         return mask;
4967 }
4968
4969 static int io_uring_fasync(int fd, struct file *file, int on)
4970 {
4971         struct io_ring_ctx *ctx = file->private_data;
4972
4973         return fasync_helper(fd, file, on, &ctx->cq_fasync);
4974 }
4975
4976 static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
4977 {
4978         mutex_lock(&ctx->uring_lock);
4979         percpu_ref_kill(&ctx->refs);
4980         mutex_unlock(&ctx->uring_lock);
4981
4982         io_kill_timeouts(ctx);
4983         io_poll_remove_all(ctx);
4984
4985         if (ctx->io_wq)
4986                 io_wq_cancel_all(ctx->io_wq);
4987
4988         io_iopoll_reap_events(ctx);
4989         /* if we failed setting up the ctx, we might not have any rings */
4990         if (ctx->rings)
4991                 io_cqring_overflow_flush(ctx, true);
4992         wait_for_completion(&ctx->completions[0]);
4993         io_ring_ctx_free(ctx);
4994 }
4995
4996 static int io_uring_release(struct inode *inode, struct file *file)
4997 {
4998         struct io_ring_ctx *ctx = file->private_data;
4999
5000         file->private_data = NULL;
5001         io_ring_ctx_wait_and_kill(ctx);
5002         return 0;
5003 }
5004
5005 static void io_uring_cancel_files(struct io_ring_ctx *ctx,
5006                                   struct files_struct *files)
5007 {
5008         struct io_kiocb *req;
5009         DEFINE_WAIT(wait);
5010
5011         while (!list_empty_careful(&ctx->inflight_list)) {
5012                 struct io_kiocb *cancel_req = NULL;
5013
5014                 spin_lock_irq(&ctx->inflight_lock);
5015                 list_for_each_entry(req, &ctx->inflight_list, inflight_entry) {
5016                         if (req->work.files != files)
5017                                 continue;
5018                         /* req is being completed, ignore */
5019                         if (!refcount_inc_not_zero(&req->refs))
5020                                 continue;
5021                         cancel_req = req;
5022                         break;
5023                 }
5024                 if (cancel_req)
5025                         prepare_to_wait(&ctx->inflight_wait, &wait,
5026                                                 TASK_UNINTERRUPTIBLE);
5027                 spin_unlock_irq(&ctx->inflight_lock);
5028
5029                 /* We need to keep going until we don't find a matching req */
5030                 if (!cancel_req)
5031                         break;
5032
5033                 io_wq_cancel_work(ctx->io_wq, &cancel_req->work);
5034                 io_put_req(cancel_req);
5035                 schedule();
5036         }
5037         finish_wait(&ctx->inflight_wait, &wait);
5038 }
5039
5040 static int io_uring_flush(struct file *file, void *data)
5041 {
5042         struct io_ring_ctx *ctx = file->private_data;
5043
5044         io_uring_cancel_files(ctx, data);
5045         if (fatal_signal_pending(current) || (current->flags & PF_EXITING)) {
5046                 io_cqring_overflow_flush(ctx, true);
5047                 io_wq_cancel_all(ctx->io_wq);
5048         }
5049         return 0;
5050 }
5051
5052 static void *io_uring_validate_mmap_request(struct file *file,
5053                                             loff_t pgoff, size_t sz)
5054 {
5055         struct io_ring_ctx *ctx = file->private_data;
5056         loff_t offset = pgoff << PAGE_SHIFT;
5057         struct page *page;
5058         void *ptr;
5059
5060         switch (offset) {
5061         case IORING_OFF_SQ_RING:
5062         case IORING_OFF_CQ_RING:
5063                 ptr = ctx->rings;
5064                 break;
5065         case IORING_OFF_SQES:
5066                 ptr = ctx->sq_sqes;
5067                 break;
5068         default:
5069                 return ERR_PTR(-EINVAL);
5070         }
5071
5072         page = virt_to_head_page(ptr);
5073         if (sz > page_size(page))
5074                 return ERR_PTR(-EINVAL);
5075
5076         return ptr;
5077 }
5078
5079 #ifdef CONFIG_MMU
5080
5081 static int io_uring_mmap(struct file *file, struct vm_area_struct *vma)
5082 {
5083         size_t sz = vma->vm_end - vma->vm_start;
5084         unsigned long pfn;
5085         void *ptr;
5086
5087         ptr = io_uring_validate_mmap_request(file, vma->vm_pgoff, sz);
5088         if (IS_ERR(ptr))
5089                 return PTR_ERR(ptr);
5090
5091         pfn = virt_to_phys(ptr) >> PAGE_SHIFT;
5092         return remap_pfn_range(vma, vma->vm_start, pfn, sz, vma->vm_page_prot);
5093 }
5094
5095 #else /* !CONFIG_MMU */
5096
5097 static int io_uring_mmap(struct file *file, struct vm_area_struct *vma)
5098 {
5099         return vma->vm_flags & (VM_SHARED | VM_MAYSHARE) ? 0 : -EINVAL;
5100 }
5101
5102 static unsigned int io_uring_nommu_mmap_capabilities(struct file *file)
5103 {
5104         return NOMMU_MAP_DIRECT | NOMMU_MAP_READ | NOMMU_MAP_WRITE;
5105 }
5106
5107 static unsigned long io_uring_nommu_get_unmapped_area(struct file *file,
5108         unsigned long addr, unsigned long len,
5109         unsigned long pgoff, unsigned long flags)
5110 {
5111         void *ptr;
5112
5113         ptr = io_uring_validate_mmap_request(file, pgoff, len);
5114         if (IS_ERR(ptr))
5115                 return PTR_ERR(ptr);
5116
5117         return (unsigned long) ptr;
5118 }
5119
5120 #endif /* !CONFIG_MMU */
5121
5122 SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
5123                 u32, min_complete, u32, flags, const sigset_t __user *, sig,
5124                 size_t, sigsz)
5125 {
5126         struct io_ring_ctx *ctx;
5127         long ret = -EBADF;
5128         int submitted = 0;
5129         struct fd f;
5130
5131         if (flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP))
5132                 return -EINVAL;
5133
5134         f = fdget(fd);
5135         if (!f.file)
5136                 return -EBADF;
5137
5138         ret = -EOPNOTSUPP;
5139         if (f.file->f_op != &io_uring_fops)
5140                 goto out_fput;
5141
5142         ret = -ENXIO;
5143         ctx = f.file->private_data;
5144         if (!percpu_ref_tryget(&ctx->refs))
5145                 goto out_fput;
5146
5147         /*
5148          * For SQ polling, the thread will do all submissions and completions.
5149          * Just return the requested submit count, and wake the thread if
5150          * we were asked to.
5151          */
5152         ret = 0;
5153         if (ctx->flags & IORING_SETUP_SQPOLL) {
5154                 if (!list_empty_careful(&ctx->cq_overflow_list))
5155                         io_cqring_overflow_flush(ctx, false);
5156                 if (flags & IORING_ENTER_SQ_WAKEUP)
5157                         wake_up(&ctx->sqo_wait);
5158                 submitted = to_submit;
5159         } else if (to_submit) {
5160                 struct mm_struct *cur_mm;
5161
5162                 if (current->mm != ctx->sqo_mm ||
5163                     current_cred() != ctx->creds) {
5164                         ret = -EPERM;
5165                         goto out;
5166                 }
5167
5168                 to_submit = min(to_submit, ctx->sq_entries);
5169                 mutex_lock(&ctx->uring_lock);
5170                 /* already have mm, so io_submit_sqes() won't try to grab it */
5171                 cur_mm = ctx->sqo_mm;
5172                 submitted = io_submit_sqes(ctx, to_submit, f.file, fd,
5173                                            &cur_mm, false);
5174                 mutex_unlock(&ctx->uring_lock);
5175
5176                 if (submitted != to_submit)
5177                         goto out;
5178         }
5179         if (flags & IORING_ENTER_GETEVENTS) {
5180                 unsigned nr_events = 0;
5181
5182                 min_complete = min(min_complete, ctx->cq_entries);
5183
5184                 if (ctx->flags & IORING_SETUP_IOPOLL) {
5185                         ret = io_iopoll_check(ctx, &nr_events, min_complete);
5186                 } else {
5187                         ret = io_cqring_wait(ctx, min_complete, sig, sigsz);
5188                 }
5189         }
5190
5191 out:
5192         percpu_ref_put(&ctx->refs);
5193 out_fput:
5194         fdput(f);
5195         return submitted ? submitted : ret;
5196 }
5197
5198 static const struct file_operations io_uring_fops = {
5199         .release        = io_uring_release,
5200         .flush          = io_uring_flush,
5201         .mmap           = io_uring_mmap,
5202 #ifndef CONFIG_MMU
5203         .get_unmapped_area = io_uring_nommu_get_unmapped_area,
5204         .mmap_capabilities = io_uring_nommu_mmap_capabilities,
5205 #endif
5206         .poll           = io_uring_poll,
5207         .fasync         = io_uring_fasync,
5208 };
5209
5210 static int io_allocate_scq_urings(struct io_ring_ctx *ctx,
5211                                   struct io_uring_params *p)
5212 {
5213         struct io_rings *rings;
5214         size_t size, sq_array_offset;
5215
5216         size = rings_size(p->sq_entries, p->cq_entries, &sq_array_offset);
5217         if (size == SIZE_MAX)
5218                 return -EOVERFLOW;
5219
5220         rings = io_mem_alloc(size);
5221         if (!rings)
5222                 return -ENOMEM;
5223
5224         ctx->rings = rings;
5225         ctx->sq_array = (u32 *)((char *)rings + sq_array_offset);
5226         rings->sq_ring_mask = p->sq_entries - 1;
5227         rings->cq_ring_mask = p->cq_entries - 1;
5228         rings->sq_ring_entries = p->sq_entries;
5229         rings->cq_ring_entries = p->cq_entries;
5230         ctx->sq_mask = rings->sq_ring_mask;
5231         ctx->cq_mask = rings->cq_ring_mask;
5232         ctx->sq_entries = rings->sq_ring_entries;
5233         ctx->cq_entries = rings->cq_ring_entries;
5234
5235         size = array_size(sizeof(struct io_uring_sqe), p->sq_entries);
5236         if (size == SIZE_MAX) {
5237                 io_mem_free(ctx->rings);
5238                 ctx->rings = NULL;
5239                 return -EOVERFLOW;
5240         }
5241
5242         ctx->sq_sqes = io_mem_alloc(size);
5243         if (!ctx->sq_sqes) {
5244                 io_mem_free(ctx->rings);
5245                 ctx->rings = NULL;
5246                 return -ENOMEM;
5247         }
5248
5249         return 0;
5250 }
5251
5252 /*
5253  * Allocate an anonymous fd, this is what constitutes the application
5254  * visible backing of an io_uring instance. The application mmaps this
5255  * fd to gain access to the SQ/CQ ring details. If UNIX sockets are enabled,
5256  * we have to tie this fd to a socket for file garbage collection purposes.
5257  */
5258 static int io_uring_get_fd(struct io_ring_ctx *ctx)
5259 {
5260         struct file *file;
5261         int ret;
5262
5263 #if defined(CONFIG_UNIX)
5264         ret = sock_create_kern(&init_net, PF_UNIX, SOCK_RAW, IPPROTO_IP,
5265                                 &ctx->ring_sock);
5266         if (ret)
5267                 return ret;
5268 #endif
5269
5270         ret = get_unused_fd_flags(O_RDWR | O_CLOEXEC);
5271         if (ret < 0)
5272                 goto err;
5273
5274         file = anon_inode_getfile("[io_uring]", &io_uring_fops, ctx,
5275                                         O_RDWR | O_CLOEXEC);
5276         if (IS_ERR(file)) {
5277                 put_unused_fd(ret);
5278                 ret = PTR_ERR(file);
5279                 goto err;
5280         }
5281
5282 #if defined(CONFIG_UNIX)
5283         ctx->ring_sock->file = file;
5284         ctx->ring_sock->sk->sk_user_data = ctx;
5285 #endif
5286         fd_install(ret, file);
5287         return ret;
5288 err:
5289 #if defined(CONFIG_UNIX)
5290         sock_release(ctx->ring_sock);
5291         ctx->ring_sock = NULL;
5292 #endif
5293         return ret;
5294 }
5295
5296 static int io_uring_create(unsigned entries, struct io_uring_params *p)
5297 {
5298         struct user_struct *user = NULL;
5299         struct io_ring_ctx *ctx;
5300         bool account_mem;
5301         int ret;
5302
5303         if (!entries || entries > IORING_MAX_ENTRIES)
5304                 return -EINVAL;
5305
5306         /*
5307          * Use twice as many entries for the CQ ring. It's possible for the
5308          * application to drive a higher depth than the size of the SQ ring,
5309          * since the sqes are only used at submission time. This allows for
5310          * some flexibility in overcommitting a bit. If the application has
5311          * set IORING_SETUP_CQSIZE, it will have passed in the desired number
5312          * of CQ ring entries manually.
5313          */
5314         p->sq_entries = roundup_pow_of_two(entries);
5315         if (p->flags & IORING_SETUP_CQSIZE) {
5316                 /*
5317                  * If IORING_SETUP_CQSIZE is set, we do the same roundup
5318                  * to a power-of-two, if it isn't already. We do NOT impose
5319                  * any cq vs sq ring sizing.
5320                  */
5321                 if (p->cq_entries < p->sq_entries || p->cq_entries > IORING_MAX_CQ_ENTRIES)
5322                         return -EINVAL;
5323                 p->cq_entries = roundup_pow_of_two(p->cq_entries);
5324         } else {
5325                 p->cq_entries = 2 * p->sq_entries;
5326         }
5327
5328         user = get_uid(current_user());
5329         account_mem = !capable(CAP_IPC_LOCK);
5330
5331         if (account_mem) {
5332                 ret = io_account_mem(user,
5333                                 ring_pages(p->sq_entries, p->cq_entries));
5334                 if (ret) {
5335                         free_uid(user);
5336                         return ret;
5337                 }
5338         }
5339
5340         ctx = io_ring_ctx_alloc(p);
5341         if (!ctx) {
5342                 if (account_mem)
5343                         io_unaccount_mem(user, ring_pages(p->sq_entries,
5344                                                                 p->cq_entries));
5345                 free_uid(user);
5346                 return -ENOMEM;
5347         }
5348         ctx->compat = in_compat_syscall();
5349         ctx->account_mem = account_mem;
5350         ctx->user = user;
5351         ctx->creds = get_current_cred();
5352
5353         ret = io_allocate_scq_urings(ctx, p);
5354         if (ret)
5355                 goto err;
5356
5357         ret = io_sq_offload_start(ctx, p);
5358         if (ret)
5359                 goto err;
5360
5361         memset(&p->sq_off, 0, sizeof(p->sq_off));
5362         p->sq_off.head = offsetof(struct io_rings, sq.head);
5363         p->sq_off.tail = offsetof(struct io_rings, sq.tail);
5364         p->sq_off.ring_mask = offsetof(struct io_rings, sq_ring_mask);
5365         p->sq_off.ring_entries = offsetof(struct io_rings, sq_ring_entries);
5366         p->sq_off.flags = offsetof(struct io_rings, sq_flags);
5367         p->sq_off.dropped = offsetof(struct io_rings, sq_dropped);
5368         p->sq_off.array = (char *)ctx->sq_array - (char *)ctx->rings;
5369
5370         memset(&p->cq_off, 0, sizeof(p->cq_off));
5371         p->cq_off.head = offsetof(struct io_rings, cq.head);
5372         p->cq_off.tail = offsetof(struct io_rings, cq.tail);
5373         p->cq_off.ring_mask = offsetof(struct io_rings, cq_ring_mask);
5374         p->cq_off.ring_entries = offsetof(struct io_rings, cq_ring_entries);
5375         p->cq_off.overflow = offsetof(struct io_rings, cq_overflow);
5376         p->cq_off.cqes = offsetof(struct io_rings, cqes);
5377
5378         /*
5379          * Install ring fd as the very last thing, so we don't risk someone
5380          * having closed it before we finish setup
5381          */
5382         ret = io_uring_get_fd(ctx);
5383         if (ret < 0)
5384                 goto err;
5385
5386         p->features = IORING_FEAT_SINGLE_MMAP | IORING_FEAT_NODROP |
5387                         IORING_FEAT_SUBMIT_STABLE;
5388         trace_io_uring_create(ret, ctx, p->sq_entries, p->cq_entries, p->flags);
5389         return ret;
5390 err:
5391         io_ring_ctx_wait_and_kill(ctx);
5392         return ret;
5393 }
5394
5395 /*
5396  * Sets up an aio uring context, and returns the fd. Applications asks for a
5397  * ring size, we return the actual sq/cq ring sizes (among other things) in the
5398  * params structure passed in.
5399  */
5400 static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
5401 {
5402         struct io_uring_params p;
5403         long ret;
5404         int i;
5405
5406         if (copy_from_user(&p, params, sizeof(p)))
5407                 return -EFAULT;
5408         for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
5409                 if (p.resv[i])
5410                         return -EINVAL;
5411         }
5412
5413         if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
5414                         IORING_SETUP_SQ_AFF | IORING_SETUP_CQSIZE))
5415                 return -EINVAL;
5416
5417         ret = io_uring_create(entries, &p);
5418         if (ret < 0)
5419                 return ret;
5420
5421         if (copy_to_user(params, &p, sizeof(p)))
5422                 return -EFAULT;
5423
5424         return ret;
5425 }
5426
5427 SYSCALL_DEFINE2(io_uring_setup, u32, entries,
5428                 struct io_uring_params __user *, params)
5429 {
5430         return io_uring_setup(entries, params);
5431 }
5432
5433 static int __io_uring_register(struct io_ring_ctx *ctx, unsigned opcode,
5434                                void __user *arg, unsigned nr_args)
5435         __releases(ctx->uring_lock)
5436         __acquires(ctx->uring_lock)
5437 {
5438         int ret;
5439
5440         /*
5441          * We're inside the ring mutex, if the ref is already dying, then
5442          * someone else killed the ctx or is already going through
5443          * io_uring_register().
5444          */
5445         if (percpu_ref_is_dying(&ctx->refs))
5446                 return -ENXIO;
5447
5448         percpu_ref_kill(&ctx->refs);
5449
5450         /*
5451          * Drop uring mutex before waiting for references to exit. If another
5452          * thread is currently inside io_uring_enter() it might need to grab
5453          * the uring_lock to make progress. If we hold it here across the drain
5454          * wait, then we can deadlock. It's safe to drop the mutex here, since
5455          * no new references will come in after we've killed the percpu ref.
5456          */
5457         mutex_unlock(&ctx->uring_lock);
5458         wait_for_completion(&ctx->completions[0]);
5459         mutex_lock(&ctx->uring_lock);
5460
5461         switch (opcode) {
5462         case IORING_REGISTER_BUFFERS:
5463                 ret = io_sqe_buffer_register(ctx, arg, nr_args);
5464                 break;
5465         case IORING_UNREGISTER_BUFFERS:
5466                 ret = -EINVAL;
5467                 if (arg || nr_args)
5468                         break;
5469                 ret = io_sqe_buffer_unregister(ctx);
5470                 break;
5471         case IORING_REGISTER_FILES:
5472                 ret = io_sqe_files_register(ctx, arg, nr_args);
5473                 break;
5474         case IORING_UNREGISTER_FILES:
5475                 ret = -EINVAL;
5476                 if (arg || nr_args)
5477                         break;
5478                 ret = io_sqe_files_unregister(ctx);
5479                 break;
5480         case IORING_REGISTER_FILES_UPDATE:
5481                 ret = io_sqe_files_update(ctx, arg, nr_args);
5482                 break;
5483         case IORING_REGISTER_EVENTFD:
5484                 ret = -EINVAL;
5485                 if (nr_args != 1)
5486                         break;
5487                 ret = io_eventfd_register(ctx, arg);
5488                 break;
5489         case IORING_UNREGISTER_EVENTFD:
5490                 ret = -EINVAL;
5491                 if (arg || nr_args)
5492                         break;
5493                 ret = io_eventfd_unregister(ctx);
5494                 break;
5495         default:
5496                 ret = -EINVAL;
5497                 break;
5498         }
5499
5500         /* bring the ctx back to life */
5501         reinit_completion(&ctx->completions[0]);
5502         percpu_ref_reinit(&ctx->refs);
5503         return ret;
5504 }
5505
5506 SYSCALL_DEFINE4(io_uring_register, unsigned int, fd, unsigned int, opcode,
5507                 void __user *, arg, unsigned int, nr_args)
5508 {
5509         struct io_ring_ctx *ctx;
5510         long ret = -EBADF;
5511         struct fd f;
5512
5513         f = fdget(fd);
5514         if (!f.file)
5515                 return -EBADF;
5516
5517         ret = -EOPNOTSUPP;
5518         if (f.file->f_op != &io_uring_fops)
5519                 goto out_fput;
5520
5521         ctx = f.file->private_data;
5522
5523         mutex_lock(&ctx->uring_lock);
5524         ret = __io_uring_register(ctx, opcode, arg, nr_args);
5525         mutex_unlock(&ctx->uring_lock);
5526         trace_io_uring_register(ctx, opcode, ctx->nr_user_files, ctx->nr_user_bufs,
5527                                                         ctx->cq_ev_fd != NULL, ret);
5528 out_fput:
5529         fdput(f);
5530         return ret;
5531 }
5532
5533 static int __init io_uring_init(void)
5534 {
5535         req_cachep = KMEM_CACHE(io_kiocb, SLAB_HWCACHE_ALIGN | SLAB_PANIC);
5536         return 0;
5537 };
5538 __initcall(io_uring_init);