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