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 <net/compat.h>
  48 #include <linux/refcount.h>
  49 #include <linux/uio.h>
  50 #include <linux/bits.h>
  51
  52 #include <linux/sched/signal.h>
  53 #include <linux/fs.h>
  54 #include <linux/file.h>
  55 #include <linux/fdtable.h>
  56 #include <linux/mm.h>
  57 #include <linux/mman.h>
  58 #include <linux/percpu.h>
  59 #include <linux/slab.h>
  60 #include <linux/kthread.h>
  61 #include <linux/blkdev.h>
  62 #include <linux/bvec.h>
  63 #include <linux/net.h>
  64 #include <net/sock.h>
  65 #include <net/af_unix.h>
  66 #include <net/scm.h>
  67 #include <linux/anon_inodes.h>
  68 #include <linux/sched/mm.h>
  69 #include <linux/uaccess.h>
  70 #include <linux/nospec.h>
  71 #include <linux/sizes.h>
  72 #include <linux/hugetlb.h>
  73 #include <linux/highmem.h>
  74 #include <linux/namei.h>
  75 #include <linux/fsnotify.h>
  76 #include <linux/fadvise.h>
  77 #include <linux/eventpoll.h>
  78 #include <linux/fs_struct.h>
  79 #include <linux/splice.h>
  80 #include <linux/task_work.h>
  81 #include <linux/pagemap.h>
  82 #include <linux/io_uring.h>
  83 #include <linux/blk-cgroup.h>
  84 #include <linux/audit.h>
  85
  86 #define CREATE_TRACE_POINTS
  87 #include <trace/events/io_uring.h>
  88
  89 #include <uapi/linux/io_uring.h>
  90
  91 #include "internal.h"
  92 #include "io-wq.h"
  93
  94 #define IORING_MAX_ENTRIES      32768
  95 #define IORING_MAX_CQ_ENTRIES   (2 * IORING_MAX_ENTRIES)
  96
  97 /*
  98  * Shift of 9 is 512 entries, or exactly one page on 64-bit archs
  99  */
 100 #define IORING_FILE_TABLE_SHIFT 9
 101 #define IORING_MAX_FILES_TABLE  (1U << IORING_FILE_TABLE_SHIFT)
 102 #define IORING_FILE_TABLE_MASK  (IORING_MAX_FILES_TABLE - 1)
 103 #define IORING_MAX_FIXED_FILES  (64 * IORING_MAX_FILES_TABLE)
 104 #define IORING_MAX_RESTRICTIONS (IORING_RESTRICTION_LAST + \
 105                                  IORING_REGISTER_LAST + IORING_OP_LAST)
 106
 107 struct io_uring {
 108         u32 head ____cacheline_aligned_in_smp;
 109         u32 tail ____cacheline_aligned_in_smp;
 110 };
 111
 112 /*
 113  * This data is shared with the application through the mmap at offsets
 114  * IORING_OFF_SQ_RING and IORING_OFF_CQ_RING.
 115  *
 116  * The offsets to the member fields are published through struct
 117  * io_sqring_offsets when calling io_uring_setup.
 118  */
 119 struct io_rings {
 120         /*
 121          * Head and tail offsets into the ring; the offsets need to be
 122          * masked to get valid indices.
 123          *
 124          * The kernel controls head of the sq ring and the tail of the cq ring,
 125          * and the application controls tail of the sq ring and the head of the
 126          * cq ring.
 127          */
 128         struct io_uring         sq, cq;
 129         /*
 130          * Bitmasks to apply to head and tail offsets (constant, equals
 131          * ring_entries - 1)
 132          */
 133         u32                     sq_ring_mask, cq_ring_mask;
 134         /* Ring sizes (constant, power of 2) */
 135         u32                     sq_ring_entries, cq_ring_entries;
 136         /*
 137          * Number of invalid entries dropped by the kernel due to
 138          * invalid index stored in array
 139          *
 140          * Written by the kernel, shouldn't be modified by the
 141          * application (i.e. get number of "new events" by comparing to
 142          * cached value).
 143          *
 144          * After a new SQ head value was read by the application this
 145          * counter includes all submissions that were dropped reaching
 146          * the new SQ head (and possibly more).
 147          */
 148         u32                     sq_dropped;
 149         /*
 150          * Runtime SQ flags
 151          *
 152          * Written by the kernel, shouldn't be modified by the
 153          * application.
 154          *
 155          * The application needs a full memory barrier before checking
 156          * for IORING_SQ_NEED_WAKEUP after updating the sq tail.
 157          */
 158         u32                     sq_flags;
 159         /*
 160          * Runtime CQ flags
 161          *
 162          * Written by the application, shouldn't be modified by the
 163          * kernel.
 164          */
 165         u32                     cq_flags;
 166         /*
 167          * Number of completion events lost because the queue was full;
 168          * this should be avoided by the application by making sure
 169          * there are not more requests pending than there is space in
 170          * the completion queue.
 171          *
 172          * Written by the kernel, shouldn't be modified by the
 173          * application (i.e. get number of "new events" by comparing to
 174          * cached value).
 175          *
 176          * As completion events come in out of order this counter is not
 177          * ordered with any other data.
 178          */
 179         u32                     cq_overflow;
 180         /*
 181          * Ring buffer of completion events.
 182          *
 183          * The kernel writes completion events fresh every time they are
 184          * produced, so the application is allowed to modify pending
 185          * entries.
 186          */
 187         struct io_uring_cqe     cqes[] ____cacheline_aligned_in_smp;
 188 };
 189
 190 struct io_mapped_ubuf {
 191         u64             ubuf;
 192         size_t          len;
 193         struct          bio_vec *bvec;
 194         unsigned int    nr_bvecs;
 195         unsigned long   acct_pages;
 196 };
 197
 198 struct fixed_file_table {
 199         struct file             **files;
 200 };
 201
 202 struct fixed_file_ref_node {
 203         struct percpu_ref               refs;
 204         struct list_head                node;
 205         struct list_head                file_list;
 206         struct fixed_file_data          *file_data;
 207         struct llist_node               llist;
 208 };
 209
 210 struct fixed_file_data {
 211         struct fixed_file_table         *table;
 212         struct io_ring_ctx              *ctx;
 213
 214         struct fixed_file_ref_node      *node;
 215         struct percpu_ref               refs;
 216         struct completion               done;
 217         struct list_head                ref_list;
 218         spinlock_t                      lock;
 219 };
 220
 221 struct io_buffer {
 222         struct list_head list;
 223         __u64 addr;
 224         __s32 len;
 225         __u16 bid;
 226 };
 227
 228 struct io_restriction {
 229         DECLARE_BITMAP(register_op, IORING_REGISTER_LAST);
 230         DECLARE_BITMAP(sqe_op, IORING_OP_LAST);
 231         u8 sqe_flags_allowed;
 232         u8 sqe_flags_required;
 233         bool registered;
 234 };
 235
 236 struct io_sq_data {
 237         refcount_t              refs;
 238         struct mutex            lock;
 239
 240         /* ctx's that are using this sqd */
 241         struct list_head        ctx_list;
 242         struct list_head        ctx_new_list;
 243         struct mutex            ctx_lock;
 244
 245         struct task_struct      *thread;
 246         struct wait_queue_head  wait;
 247 };
 248
 249 struct io_ring_ctx {
 250         struct {
 251                 struct percpu_ref       refs;
 252         } ____cacheline_aligned_in_smp;
 253
 254         struct {
 255                 unsigned int            flags;
 256                 unsigned int            compat: 1;
 257                 unsigned int            limit_mem: 1;
 258                 unsigned int            cq_overflow_flushed: 1;
 259                 unsigned int            drain_next: 1;
 260                 unsigned int            eventfd_async: 1;
 261                 unsigned int            restricted: 1;
 262
 263                 /*
 264                  * Ring buffer of indices into array of io_uring_sqe, which is
 265                  * mmapped by the application using the IORING_OFF_SQES offset.
 266                  *
 267                  * This indirection could e.g. be used to assign fixed
 268                  * io_uring_sqe entries to operations and only submit them to
 269                  * the queue when needed.
 270                  *
 271                  * The kernel modifies neither the indices array nor the entries
 272                  * array.
 273                  */
 274                 u32                     *sq_array;
 275                 unsigned                cached_sq_head;
 276                 unsigned                sq_entries;
 277                 unsigned                sq_mask;
 278                 unsigned                sq_thread_idle;
 279                 unsigned                cached_sq_dropped;
 280                 unsigned                cached_cq_overflow;
 281                 unsigned long           sq_check_overflow;
 282
 283                 struct list_head        defer_list;
 284                 struct list_head        timeout_list;
 285                 struct list_head        cq_overflow_list;
 286
 287                 wait_queue_head_t       inflight_wait;
 288                 struct io_uring_sqe     *sq_sqes;
 289         } ____cacheline_aligned_in_smp;
 290
 291         struct io_rings *rings;
 292
 293         /* IO offload */
 294         struct io_wq            *io_wq;
 295
 296         /*
 297          * For SQPOLL usage - we hold a reference to the parent task, so we
 298          * have access to the ->files
 299          */
 300         struct task_struct      *sqo_task;
 301
 302         /* Only used for accounting purposes */
 303         struct mm_struct        *mm_account;
 304
 305 #ifdef CONFIG_BLK_CGROUP
 306         struct cgroup_subsys_state      *sqo_blkcg_css;
 307 #endif
 308
 309         struct io_sq_data       *sq_data;       /* if using sq thread polling */
 310
 311         struct wait_queue_head  sqo_sq_wait;
 312         struct wait_queue_entry sqo_wait_entry;
 313         struct list_head        sqd_list;
 314
 315         /*
 316          * If used, fixed file set. Writers must ensure that ->refs is dead,
 317          * readers must ensure that ->refs is alive as long as the file* is
 318          * used. Only updated through io_uring_register(2).
 319          */
 320         struct fixed_file_data  *file_data;
 321         unsigned                nr_user_files;
 322
 323         /* if used, fixed mapped user buffers */
 324         unsigned                nr_user_bufs;
 325         struct io_mapped_ubuf   *user_bufs;
 326
 327         struct user_struct      *user;
 328
 329         const struct cred       *creds;
 330
 331 #ifdef CONFIG_AUDIT
 332         kuid_t                  loginuid;
 333         unsigned int            sessionid;
 334 #endif
 335
 336         struct completion       ref_comp;
 337         struct completion       sq_thread_comp;
 338
 339         /* if all else fails... */
 340         struct io_kiocb         *fallback_req;
 341
 342 #if defined(CONFIG_UNIX)
 343         struct socket           *ring_sock;
 344 #endif
 345
 346         struct idr              io_buffer_idr;
 347
 348         struct idr              personality_idr;
 349
 350         struct {
 351                 unsigned                cached_cq_tail;
 352                 unsigned                cq_entries;
 353                 unsigned                cq_mask;
 354                 atomic_t                cq_timeouts;
 355                 unsigned long           cq_check_overflow;
 356                 struct wait_queue_head  cq_wait;
 357                 struct fasync_struct    *cq_fasync;
 358                 struct eventfd_ctx      *cq_ev_fd;
 359         } ____cacheline_aligned_in_smp;
 360
 361         struct {
 362                 struct mutex            uring_lock;
 363                 wait_queue_head_t       wait;
 364         } ____cacheline_aligned_in_smp;
 365
 366         struct {
 367                 spinlock_t              completion_lock;
 368
 369                 /*
 370                  * ->iopoll_list is protected by the ctx->uring_lock for
 371                  * io_uring instances that don't use IORING_SETUP_SQPOLL.
 372                  * For SQPOLL, only the single threaded io_sq_thread() will
 373                  * manipulate the list, hence no extra locking is needed there.
 374                  */
 375                 struct list_head        iopoll_list;
 376                 struct hlist_head       *cancel_hash;
 377                 unsigned                cancel_hash_bits;
 378                 bool                    poll_multi_file;
 379
 380                 spinlock_t              inflight_lock;
 381                 struct list_head        inflight_list;
 382         } ____cacheline_aligned_in_smp;
 383
 384         struct delayed_work             file_put_work;
 385         struct llist_head               file_put_llist;
 386
 387         struct work_struct              exit_work;
 388         struct io_restriction           restrictions;
 389 };
 390
 391 /*
 392  * First field must be the file pointer in all the
 393  * iocb unions! See also 'struct kiocb' in <linux/fs.h>
 394  */
 395 struct io_poll_iocb {
 396         struct file                     *file;
 397         union {
 398                 struct wait_queue_head  *head;
 399                 u64                     addr;
 400         };
 401         __poll_t                        events;
 402         bool                            done;
 403         bool                            canceled;
 404         struct wait_queue_entry         wait;
 405 };
 406
 407 struct io_close {
 408         struct file                     *file;
 409         struct file                     *put_file;
 410         int                             fd;
 411 };
 412
 413 struct io_timeout_data {
 414         struct io_kiocb                 *req;
 415         struct hrtimer                  timer;
 416         struct timespec64               ts;
 417         enum hrtimer_mode               mode;
 418 };
 419
 420 struct io_accept {
 421         struct file                     *file;
 422         struct sockaddr __user          *addr;
 423         int __user                      *addr_len;
 424         int                             flags;
 425         unsigned long                   nofile;
 426 };
 427
 428 struct io_sync {
 429         struct file                     *file;
 430         loff_t                          len;
 431         loff_t                          off;
 432         int                             flags;
 433         int                             mode;
 434 };
 435
 436 struct io_cancel {
 437         struct file                     *file;
 438         u64                             addr;
 439 };
 440
 441 struct io_timeout {
 442         struct file                     *file;
 443         u32                             off;
 444         u32                             target_seq;
 445         struct list_head                list;
 446 };
 447
 448 struct io_timeout_rem {
 449         struct file                     *file;
 450         u64                             addr;
 451 };
 452
 453 struct io_rw {
 454         /* NOTE: kiocb has the file as the first member, so don't do it here */
 455         struct kiocb                    kiocb;
 456         u64                             addr;
 457         u64                             len;
 458 };
 459
 460 struct io_connect {
 461         struct file                     *file;
 462         struct sockaddr __user          *addr;
 463         int                             addr_len;
 464 };
 465
 466 struct io_sr_msg {
 467         struct file                     *file;
 468         union {
 469                 struct user_msghdr __user *umsg;
 470                 void __user             *buf;
 471         };
 472         int                             msg_flags;
 473         int                             bgid;
 474         size_t                          len;
 475         struct io_buffer                *kbuf;
 476 };
 477
 478 struct io_open {
 479         struct file                     *file;
 480         int                             dfd;
 481         struct filename                 *filename;
 482         struct open_how                 how;
 483         unsigned long                   nofile;
 484 };
 485
 486 struct io_files_update {
 487         struct file                     *file;
 488         u64                             arg;
 489         u32                             nr_args;
 490         u32                             offset;
 491 };
 492
 493 struct io_fadvise {
 494         struct file                     *file;
 495         u64                             offset;
 496         u32                             len;
 497         u32                             advice;
 498 };
 499
 500 struct io_madvise {
 501         struct file                     *file;
 502         u64                             addr;
 503         u32                             len;
 504         u32                             advice;
 505 };
 506
 507 struct io_epoll {
 508         struct file                     *file;
 509         int                             epfd;
 510         int                             op;
 511         int                             fd;
 512         struct epoll_event              event;
 513 };
 514
 515 struct io_splice {
 516         struct file                     *file_out;
 517         struct file                     *file_in;
 518         loff_t                          off_out;
 519         loff_t                          off_in;
 520         u64                             len;
 521         unsigned int                    flags;
 522 };
 523
 524 struct io_provide_buf {
 525         struct file                     *file;
 526         __u64                           addr;
 527         __s32                           len;
 528         __u32                           bgid;
 529         __u16                           nbufs;
 530         __u16                           bid;
 531 };
 532
 533 struct io_statx {
 534         struct file                     *file;
 535         int                             dfd;
 536         unsigned int                    mask;
 537         unsigned int                    flags;
 538         const char __user               *filename;
 539         struct statx __user             *buffer;
 540 };
 541
 542 struct io_completion {
 543         struct file                     *file;
 544         struct list_head                list;
 545         int                             cflags;
 546 };
 547
 548 struct io_async_connect {
 549         struct sockaddr_storage         address;
 550 };
 551
 552 struct io_async_msghdr {
 553         struct iovec                    fast_iov[UIO_FASTIOV];
 554         struct iovec                    *iov;
 555         struct sockaddr __user          *uaddr;
 556         struct msghdr                   msg;
 557         struct sockaddr_storage         addr;
 558 };
 559
 560 struct io_async_rw {
 561         struct iovec                    fast_iov[UIO_FASTIOV];
 562         const struct iovec              *free_iovec;
 563         struct iov_iter                 iter;
 564         size_t                          bytes_done;
 565         struct wait_page_queue          wpq;
 566 };
 567
 568 enum {
 569         REQ_F_FIXED_FILE_BIT    = IOSQE_FIXED_FILE_BIT,
 570         REQ_F_IO_DRAIN_BIT      = IOSQE_IO_DRAIN_BIT,
 571         REQ_F_LINK_BIT          = IOSQE_IO_LINK_BIT,
 572         REQ_F_HARDLINK_BIT      = IOSQE_IO_HARDLINK_BIT,
 573         REQ_F_FORCE_ASYNC_BIT   = IOSQE_ASYNC_BIT,
 574         REQ_F_BUFFER_SELECT_BIT = IOSQE_BUFFER_SELECT_BIT,
 575
 576         REQ_F_LINK_HEAD_BIT,
 577         REQ_F_FAIL_LINK_BIT,
 578         REQ_F_INFLIGHT_BIT,
 579         REQ_F_CUR_POS_BIT,
 580         REQ_F_NOWAIT_BIT,
 581         REQ_F_LINK_TIMEOUT_BIT,
 582         REQ_F_ISREG_BIT,
 583         REQ_F_NEED_CLEANUP_BIT,
 584         REQ_F_POLLED_BIT,
 585         REQ_F_BUFFER_SELECTED_BIT,
 586         REQ_F_NO_FILE_TABLE_BIT,
 587         REQ_F_WORK_INITIALIZED_BIT,
 588         REQ_F_LTIMEOUT_ACTIVE_BIT,
 589
 590         /* not a real bit, just to check we're not overflowing the space */
 591         __REQ_F_LAST_BIT,
 592 };
 593
 594 enum {
 595         /* ctx owns file */
 596         REQ_F_FIXED_FILE        = BIT(REQ_F_FIXED_FILE_BIT),
 597         /* drain existing IO first */
 598         REQ_F_IO_DRAIN          = BIT(REQ_F_IO_DRAIN_BIT),
 599         /* linked sqes */
 600         REQ_F_LINK              = BIT(REQ_F_LINK_BIT),
 601         /* doesn't sever on completion < 0 */
 602         REQ_F_HARDLINK          = BIT(REQ_F_HARDLINK_BIT),
 603         /* IOSQE_ASYNC */
 604         REQ_F_FORCE_ASYNC       = BIT(REQ_F_FORCE_ASYNC_BIT),
 605         /* IOSQE_BUFFER_SELECT */
 606         REQ_F_BUFFER_SELECT     = BIT(REQ_F_BUFFER_SELECT_BIT),
 607
 608         /* head of a link */
 609         REQ_F_LINK_HEAD         = BIT(REQ_F_LINK_HEAD_BIT),
 610         /* fail rest of links */
 611         REQ_F_FAIL_LINK         = BIT(REQ_F_FAIL_LINK_BIT),
 612         /* on inflight list */
 613         REQ_F_INFLIGHT          = BIT(REQ_F_INFLIGHT_BIT),
 614         /* read/write uses file position */
 615         REQ_F_CUR_POS           = BIT(REQ_F_CUR_POS_BIT),
 616         /* must not punt to workers */
 617         REQ_F_NOWAIT            = BIT(REQ_F_NOWAIT_BIT),
 618         /* has or had linked timeout */
 619         REQ_F_LINK_TIMEOUT      = BIT(REQ_F_LINK_TIMEOUT_BIT),
 620         /* regular file */
 621         REQ_F_ISREG             = BIT(REQ_F_ISREG_BIT),
 622         /* needs cleanup */
 623         REQ_F_NEED_CLEANUP      = BIT(REQ_F_NEED_CLEANUP_BIT),
 624         /* already went through poll handler */
 625         REQ_F_POLLED            = BIT(REQ_F_POLLED_BIT),
 626         /* buffer already selected */
 627         REQ_F_BUFFER_SELECTED   = BIT(REQ_F_BUFFER_SELECTED_BIT),
 628         /* doesn't need file table for this request */
 629         REQ_F_NO_FILE_TABLE     = BIT(REQ_F_NO_FILE_TABLE_BIT),
 630         /* io_wq_work is initialized */
 631         REQ_F_WORK_INITIALIZED  = BIT(REQ_F_WORK_INITIALIZED_BIT),
 632         /* linked timeout is active, i.e. prepared by link's head */
 633         REQ_F_LTIMEOUT_ACTIVE   = BIT(REQ_F_LTIMEOUT_ACTIVE_BIT),
 634 };
 635
 636 struct async_poll {
 637         struct io_poll_iocb     poll;
 638         struct io_poll_iocb     *double_poll;
 639 };
 640
 641 /*
 642  * NOTE! Each of the iocb union members has the file pointer
 643  * as the first entry in their struct definition. So you can
 644  * access the file pointer through any of the sub-structs,
 645  * or directly as just 'ki_filp' in this struct.
 646  */
 647 struct io_kiocb {
 648         union {
 649                 struct file             *file;
 650                 struct io_rw            rw;
 651                 struct io_poll_iocb     poll;
 652                 struct io_accept        accept;
 653                 struct io_sync          sync;
 654                 struct io_cancel        cancel;
 655                 struct io_timeout       timeout;
 656                 struct io_timeout_rem   timeout_rem;
 657                 struct io_connect       connect;
 658                 struct io_sr_msg        sr_msg;
 659                 struct io_open          open;
 660                 struct io_close         close;
 661                 struct io_files_update  files_update;
 662                 struct io_fadvise       fadvise;
 663                 struct io_madvise       madvise;
 664                 struct io_epoll         epoll;
 665                 struct io_splice        splice;
 666                 struct io_provide_buf   pbuf;
 667                 struct io_statx         statx;
 668                 /* use only after cleaning per-op data, see io_clean_op() */
 669                 struct io_completion    compl;
 670         };
 671
 672         /* opcode allocated if it needs to store data for async defer */
 673         void                            *async_data;
 674         u8                              opcode;
 675         /* polled IO has completed */
 676         u8                              iopoll_completed;
 677
 678         u16                             buf_index;
 679         u32                             result;
 680
 681         struct io_ring_ctx              *ctx;
 682         unsigned int                    flags;
 683         refcount_t                      refs;
 684         struct task_struct              *task;
 685         u64                             user_data;
 686
 687         struct list_head                link_list;
 688
 689         /*
 690          * 1. used with ctx->iopoll_list with reads/writes
 691          * 2. to track reqs with ->files (see io_op_def::file_table)
 692          */
 693         struct list_head                inflight_entry;
 694
 695         struct percpu_ref               *fixed_file_refs;
 696         struct callback_head            task_work;
 697         /* for polled requests, i.e. IORING_OP_POLL_ADD and async armed poll */
 698         struct hlist_node               hash_node;
 699         struct async_poll               *apoll;
 700         struct io_wq_work               work;
 701 };
 702
 703 struct io_defer_entry {
 704         struct list_head        list;
 705         struct io_kiocb         *req;
 706         u32                     seq;
 707 };
 708
 709 #define IO_IOPOLL_BATCH                 8
 710
 711 struct io_comp_state {
 712         unsigned int            nr;
 713         struct list_head        list;
 714         struct io_ring_ctx      *ctx;
 715 };
 716
 717 struct io_submit_state {
 718         struct blk_plug         plug;
 719
 720         /*
 721          * io_kiocb alloc cache
 722          */
 723         void                    *reqs[IO_IOPOLL_BATCH];
 724         unsigned int            free_reqs;
 725
 726         /*
 727          * Batch completion logic
 728          */
 729         struct io_comp_state    comp;
 730
 731         /*
 732          * File reference cache
 733          */
 734         struct file             *file;
 735         unsigned int            fd;
 736         unsigned int            has_refs;
 737         unsigned int            ios_left;
 738 };
 739
 740 struct io_op_def {
 741         /* needs req->file assigned */
 742         unsigned                needs_file : 1;
 743         /* don't fail if file grab fails */
 744         unsigned                needs_file_no_error : 1;
 745         /* hash wq insertion if file is a regular file */
 746         unsigned                hash_reg_file : 1;
 747         /* unbound wq insertion if file is a non-regular file */
 748         unsigned                unbound_nonreg_file : 1;
 749         /* opcode is not supported by this kernel */
 750         unsigned                not_supported : 1;
 751         /* set if opcode supports polled "wait" */
 752         unsigned                pollin : 1;
 753         unsigned                pollout : 1;
 754         /* op supports buffer selection */
 755         unsigned                buffer_select : 1;
 756         /* must always have async data allocated */
 757         unsigned                needs_async_data : 1;
 758         /* size of async data needed, if any */
 759         unsigned short          async_size;
 760         unsigned                work_flags;
 761 };
 762
 763 static const struct io_op_def io_op_defs[] = {
 764         [IORING_OP_NOP] = {},
 765         [IORING_OP_READV] = {
 766                 .needs_file             = 1,
 767                 .unbound_nonreg_file    = 1,
 768                 .pollin                 = 1,
 769                 .buffer_select          = 1,
 770                 .needs_async_data       = 1,
 771                 .async_size             = sizeof(struct io_async_rw),
 772                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG,
 773         },
 774         [IORING_OP_WRITEV] = {
 775                 .needs_file             = 1,
 776                 .hash_reg_file          = 1,
 777                 .unbound_nonreg_file    = 1,
 778                 .pollout                = 1,
 779                 .needs_async_data       = 1,
 780                 .async_size             = sizeof(struct io_async_rw),
 781                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG |
 782                                                 IO_WQ_WORK_FSIZE,
 783         },
 784         [IORING_OP_FSYNC] = {
 785                 .needs_file             = 1,
 786                 .work_flags             = IO_WQ_WORK_BLKCG,
 787         },
 788         [IORING_OP_READ_FIXED] = {
 789                 .needs_file             = 1,
 790                 .unbound_nonreg_file    = 1,
 791                 .pollin                 = 1,
 792                 .async_size             = sizeof(struct io_async_rw),
 793                 .work_flags             = IO_WQ_WORK_BLKCG | IO_WQ_WORK_MM,
 794         },
 795         [IORING_OP_WRITE_FIXED] = {
 796                 .needs_file             = 1,
 797                 .hash_reg_file          = 1,
 798                 .unbound_nonreg_file    = 1,
 799                 .pollout                = 1,
 800                 .async_size             = sizeof(struct io_async_rw),
 801                 .work_flags             = IO_WQ_WORK_BLKCG | IO_WQ_WORK_FSIZE |
 802                                                 IO_WQ_WORK_MM,
 803         },
 804         [IORING_OP_POLL_ADD] = {
 805                 .needs_file             = 1,
 806                 .unbound_nonreg_file    = 1,
 807         },
 808         [IORING_OP_POLL_REMOVE] = {},
 809         [IORING_OP_SYNC_FILE_RANGE] = {
 810                 .needs_file             = 1,
 811                 .work_flags             = IO_WQ_WORK_BLKCG,
 812         },
 813         [IORING_OP_SENDMSG] = {
 814                 .needs_file             = 1,
 815                 .unbound_nonreg_file    = 1,
 816                 .pollout                = 1,
 817                 .needs_async_data       = 1,
 818                 .async_size             = sizeof(struct io_async_msghdr),
 819                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG |
 820                                                 IO_WQ_WORK_FS,
 821         },
 822         [IORING_OP_RECVMSG] = {
 823                 .needs_file             = 1,
 824                 .unbound_nonreg_file    = 1,
 825                 .pollin                 = 1,
 826                 .buffer_select          = 1,
 827                 .needs_async_data       = 1,
 828                 .async_size             = sizeof(struct io_async_msghdr),
 829                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG |
 830                                                 IO_WQ_WORK_FS,
 831         },
 832         [IORING_OP_TIMEOUT] = {
 833                 .needs_async_data       = 1,
 834                 .async_size             = sizeof(struct io_timeout_data),
 835                 .work_flags             = IO_WQ_WORK_MM,
 836         },
 837         [IORING_OP_TIMEOUT_REMOVE] = {},
 838         [IORING_OP_ACCEPT] = {
 839                 .needs_file             = 1,
 840                 .unbound_nonreg_file    = 1,
 841                 .pollin                 = 1,
 842                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_FILES,
 843         },
 844         [IORING_OP_ASYNC_CANCEL] = {},
 845         [IORING_OP_LINK_TIMEOUT] = {
 846                 .needs_async_data       = 1,
 847                 .async_size             = sizeof(struct io_timeout_data),
 848                 .work_flags             = IO_WQ_WORK_MM,
 849         },
 850         [IORING_OP_CONNECT] = {
 851                 .needs_file             = 1,
 852                 .unbound_nonreg_file    = 1,
 853                 .pollout                = 1,
 854                 .needs_async_data       = 1,
 855                 .async_size             = sizeof(struct io_async_connect),
 856                 .work_flags             = IO_WQ_WORK_MM,
 857         },
 858         [IORING_OP_FALLOCATE] = {
 859                 .needs_file             = 1,
 860                 .work_flags             = IO_WQ_WORK_BLKCG | IO_WQ_WORK_FSIZE,
 861         },
 862         [IORING_OP_OPENAT] = {
 863                 .work_flags             = IO_WQ_WORK_FILES | IO_WQ_WORK_BLKCG |
 864                                                 IO_WQ_WORK_FS,
 865         },
 866         [IORING_OP_CLOSE] = {
 867                 .needs_file             = 1,
 868                 .needs_file_no_error    = 1,
 869                 .work_flags             = IO_WQ_WORK_FILES | IO_WQ_WORK_BLKCG,
 870         },
 871         [IORING_OP_FILES_UPDATE] = {
 872                 .work_flags             = IO_WQ_WORK_FILES | IO_WQ_WORK_MM,
 873         },
 874         [IORING_OP_STATX] = {
 875                 .work_flags             = IO_WQ_WORK_FILES | IO_WQ_WORK_MM |
 876                                                 IO_WQ_WORK_FS | IO_WQ_WORK_BLKCG,
 877         },
 878         [IORING_OP_READ] = {
 879                 .needs_file             = 1,
 880                 .unbound_nonreg_file    = 1,
 881                 .pollin                 = 1,
 882                 .buffer_select          = 1,
 883                 .async_size             = sizeof(struct io_async_rw),
 884                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG,
 885         },
 886         [IORING_OP_WRITE] = {
 887                 .needs_file             = 1,
 888                 .unbound_nonreg_file    = 1,
 889                 .pollout                = 1,
 890                 .async_size             = sizeof(struct io_async_rw),
 891                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG |
 892                                                 IO_WQ_WORK_FSIZE,
 893         },
 894         [IORING_OP_FADVISE] = {
 895                 .needs_file             = 1,
 896                 .work_flags             = IO_WQ_WORK_BLKCG,
 897         },
 898         [IORING_OP_MADVISE] = {
 899                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG,
 900         },
 901         [IORING_OP_SEND] = {
 902                 .needs_file             = 1,
 903                 .unbound_nonreg_file    = 1,
 904                 .pollout                = 1,
 905                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG,
 906         },
 907         [IORING_OP_RECV] = {
 908                 .needs_file             = 1,
 909                 .unbound_nonreg_file    = 1,
 910                 .pollin                 = 1,
 911                 .buffer_select          = 1,
 912                 .work_flags             = IO_WQ_WORK_MM | IO_WQ_WORK_BLKCG,
 913         },
 914         [IORING_OP_OPENAT2] = {
 915                 .work_flags             = IO_WQ_WORK_FILES | IO_WQ_WORK_FS |
 916                                                 IO_WQ_WORK_BLKCG,
 917         },
 918         [IORING_OP_EPOLL_CTL] = {
 919                 .unbound_nonreg_file    = 1,
 920                 .work_flags             = IO_WQ_WORK_FILES,
 921         },
 922         [IORING_OP_SPLICE] = {
 923                 .needs_file             = 1,
 924                 .hash_reg_file          = 1,
 925                 .unbound_nonreg_file    = 1,
 926                 .work_flags             = IO_WQ_WORK_BLKCG,
 927         },
 928         [IORING_OP_PROVIDE_BUFFERS] = {},
 929         [IORING_OP_REMOVE_BUFFERS] = {},
 930         [IORING_OP_TEE] = {
 931                 .needs_file             = 1,
 932                 .hash_reg_file          = 1,
 933                 .unbound_nonreg_file    = 1,
 934         },
 935 };
 936
 937 enum io_mem_account {
 938         ACCT_LOCKED,
 939         ACCT_PINNED,
 940 };
 941
 942 static void __io_complete_rw(struct io_kiocb *req, long res, long res2,
 943                              struct io_comp_state *cs);
 944 static void io_cqring_fill_event(struct io_kiocb *req, long res);
 945 static void io_put_req(struct io_kiocb *req);
 946 static void io_put_req_deferred(struct io_kiocb *req, int nr);
 947 static void io_double_put_req(struct io_kiocb *req);
 948 static struct io_kiocb *io_prep_linked_timeout(struct io_kiocb *req);
 949 static void __io_queue_linked_timeout(struct io_kiocb *req);
 950 static void io_queue_linked_timeout(struct io_kiocb *req);
 951 static int __io_sqe_files_update(struct io_ring_ctx *ctx,
 952                                  struct io_uring_files_update *ip,
 953                                  unsigned nr_args);
 954 static void __io_clean_op(struct io_kiocb *req);
 955 static struct file *io_file_get(struct io_submit_state *state,
 956                                 struct io_kiocb *req, int fd, bool fixed);
 957 static void __io_queue_sqe(struct io_kiocb *req, struct io_comp_state *cs);
 958 static void io_file_put_work(struct work_struct *work);
 959
 960 static ssize_t io_import_iovec(int rw, struct io_kiocb *req,
 961                                struct iovec **iovec, struct iov_iter *iter,
 962                                bool needs_lock);
 963 static int io_setup_async_rw(struct io_kiocb *req, const struct iovec *iovec,
 964                              const struct iovec *fast_iov,
 965                              struct iov_iter *iter, bool force);
 966
 967 static struct kmem_cache *req_cachep;
 968
 969 static const struct file_operations io_uring_fops;
 970
 971 struct sock *io_uring_get_socket(struct file *file)
 972 {
 973 #if defined(CONFIG_UNIX)
 974         if (file->f_op == &io_uring_fops) {
 975                 struct io_ring_ctx *ctx = file->private_data;
 976
 977                 return ctx->ring_sock->sk;
 978         }
 979 #endif
 980         return NULL;
 981 }
 982 EXPORT_SYMBOL(io_uring_get_socket);
 983
 984 static inline void io_clean_op(struct io_kiocb *req)
 985 {
 986         if (req->flags & (REQ_F_NEED_CLEANUP | REQ_F_BUFFER_SELECTED |
 987                           REQ_F_INFLIGHT))
 988                 __io_clean_op(req);
 989 }
 990
 991 static void io_sq_thread_drop_mm(void)
 992 {
 993         struct mm_struct *mm = current->mm;
 994
 995         if (mm) {
 996                 kthread_unuse_mm(mm);
 997                 mmput(mm);
 998                 current->mm = NULL;
 999         }
1000 }
1001
1002 static int __io_sq_thread_acquire_mm(struct io_ring_ctx *ctx)
1003 {
1004         struct mm_struct *mm;
1005
1006         if (current->mm)
1007                 return 0;
1008
1009         /* Should never happen */
1010         if (unlikely(!(ctx->flags & IORING_SETUP_SQPOLL)))
1011                 return -EFAULT;
1012
1013         task_lock(ctx->sqo_task);
1014         mm = ctx->sqo_task->mm;
1015         if (unlikely(!mm || !mmget_not_zero(mm)))
1016                 mm = NULL;
1017         task_unlock(ctx->sqo_task);
1018
1019         if (mm) {
1020                 kthread_use_mm(mm);
1021                 return 0;
1022         }
1023
1024         return -EFAULT;
1025 }
1026
1027 static int io_sq_thread_acquire_mm(struct io_ring_ctx *ctx,
1028                                    struct io_kiocb *req)
1029 {
1030         if (!(io_op_defs[req->opcode].work_flags & IO_WQ_WORK_MM))
1031                 return 0;
1032         return __io_sq_thread_acquire_mm(ctx);
1033 }
1034
1035 static void io_sq_thread_associate_blkcg(struct io_ring_ctx *ctx,
1036                                          struct cgroup_subsys_state **cur_css)
1037
1038 {
1039 #ifdef CONFIG_BLK_CGROUP
1040         /* puts the old one when swapping */
1041         if (*cur_css != ctx->sqo_blkcg_css) {
1042                 kthread_associate_blkcg(ctx->sqo_blkcg_css);
1043                 *cur_css = ctx->sqo_blkcg_css;
1044         }
1045 #endif
1046 }
1047
1048 static void io_sq_thread_unassociate_blkcg(void)
1049 {
1050 #ifdef CONFIG_BLK_CGROUP
1051         kthread_associate_blkcg(NULL);
1052 #endif
1053 }
1054
1055 static inline void req_set_fail_links(struct io_kiocb *req)
1056 {
1057         if ((req->flags & (REQ_F_LINK | REQ_F_HARDLINK)) == REQ_F_LINK)
1058                 req->flags |= REQ_F_FAIL_LINK;
1059 }
1060
1061 /*
1062  * None of these are dereferenced, they are simply used to check if any of
1063  * them have changed. If we're under current and check they are still the
1064  * same, we're fine to grab references to them for actual out-of-line use.
1065  */
1066 static void io_init_identity(struct io_identity *id)
1067 {
1068         id->files = current->files;
1069         id->mm = current->mm;
1070 #ifdef CONFIG_BLK_CGROUP
1071         rcu_read_lock();
1072         id->blkcg_css = blkcg_css();
1073         rcu_read_unlock();
1074 #endif
1075         id->creds = current_cred();
1076         id->nsproxy = current->nsproxy;
1077         id->fs = current->fs;
1078         id->fsize = rlimit(RLIMIT_FSIZE);
1079 #ifdef CONFIG_AUDIT
1080         id->loginuid = current->loginuid;
1081         id->sessionid = current->sessionid;
1082 #endif
1083         refcount_set(&id->count, 1);
1084 }
1085
1086 static inline void __io_req_init_async(struct io_kiocb *req)
1087 {
1088         memset(&req->work, 0, sizeof(req->work));
1089         req->flags |= REQ_F_WORK_INITIALIZED;
1090 }
1091
1092 /*
1093  * Note: must call io_req_init_async() for the first time you
1094  * touch any members of io_wq_work.
1095  */
1096 static inline void io_req_init_async(struct io_kiocb *req)
1097 {
1098         struct io_uring_task *tctx = current->io_uring;
1099
1100         if (req->flags & REQ_F_WORK_INITIALIZED)
1101                 return;
1102
1103         __io_req_init_async(req);
1104
1105         /* Grab a ref if this isn't our static identity */
1106         req->work.identity = tctx->identity;
1107         if (tctx->identity != &tctx->__identity)
1108                 refcount_inc(&req->work.identity->count);
1109 }
1110
1111 static inline bool io_async_submit(struct io_ring_ctx *ctx)
1112 {
1113         return ctx->flags & IORING_SETUP_SQPOLL;
1114 }
1115
1116 static void io_ring_ctx_ref_free(struct percpu_ref *ref)
1117 {
1118         struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);
1119
1120         complete(&ctx->ref_comp);
1121 }
1122
1123 static inline bool io_is_timeout_noseq(struct io_kiocb *req)
1124 {
1125         return !req->timeout.off;
1126 }
1127
1128 static struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
1129 {
1130         struct io_ring_ctx *ctx;
1131         int hash_bits;
1132
1133         ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
1134         if (!ctx)
1135                 return NULL;
1136
1137         ctx->fallback_req = kmem_cache_alloc(req_cachep, GFP_KERNEL);
1138         if (!ctx->fallback_req)
1139                 goto err;
1140
1141         /*
1142          * Use 5 bits less than the max cq entries, that should give us around
1143          * 32 entries per hash list if totally full and uniformly spread.
1144          */
1145         hash_bits = ilog2(p->cq_entries);
1146         hash_bits -= 5;
1147         if (hash_bits <= 0)
1148                 hash_bits = 1;
1149         ctx->cancel_hash_bits = hash_bits;
1150         ctx->cancel_hash = kmalloc((1U << hash_bits) * sizeof(struct hlist_head),
1151                                         GFP_KERNEL);
1152         if (!ctx->cancel_hash)
1153                 goto err;
1154         __hash_init(ctx->cancel_hash, 1U << hash_bits);
1155
1156         if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free,
1157                             PERCPU_REF_ALLOW_REINIT, GFP_KERNEL))
1158                 goto err;
1159
1160         ctx->flags = p->flags;
1161         init_waitqueue_head(&ctx->sqo_sq_wait);
1162         INIT_LIST_HEAD(&ctx->sqd_list);
1163         init_waitqueue_head(&ctx->cq_wait);
1164         INIT_LIST_HEAD(&ctx->cq_overflow_list);
1165         init_completion(&ctx->ref_comp);
1166         init_completion(&ctx->sq_thread_comp);
1167         idr_init(&ctx->io_buffer_idr);
1168         idr_init(&ctx->personality_idr);
1169         mutex_init(&ctx->uring_lock);
1170         init_waitqueue_head(&ctx->wait);
1171         spin_lock_init(&ctx->completion_lock);
1172         INIT_LIST_HEAD(&ctx->iopoll_list);
1173         INIT_LIST_HEAD(&ctx->defer_list);
1174         INIT_LIST_HEAD(&ctx->timeout_list);
1175         init_waitqueue_head(&ctx->inflight_wait);
1176         spin_lock_init(&ctx->inflight_lock);
1177         INIT_LIST_HEAD(&ctx->inflight_list);
1178         INIT_DELAYED_WORK(&ctx->file_put_work, io_file_put_work);
1179         init_llist_head(&ctx->file_put_llist);
1180         return ctx;
1181 err:
1182         if (ctx->fallback_req)
1183                 kmem_cache_free(req_cachep, ctx->fallback_req);
1184         kfree(ctx->cancel_hash);
1185         kfree(ctx);
1186         return NULL;
1187 }
1188
1189 static bool req_need_defer(struct io_kiocb *req, u32 seq)
1190 {
1191         if (unlikely(req->flags & REQ_F_IO_DRAIN)) {
1192                 struct io_ring_ctx *ctx = req->ctx;
1193
1194                 return seq != ctx->cached_cq_tail
1195                                 + READ_ONCE(ctx->cached_cq_overflow);
1196         }
1197
1198         return false;
1199 }
1200
1201 static void __io_commit_cqring(struct io_ring_ctx *ctx)
1202 {
1203         struct io_rings *rings = ctx->rings;
1204
1205         /* order cqe stores with ring update */
1206         smp_store_release(&rings->cq.tail, ctx->cached_cq_tail);
1207
1208         if (wq_has_sleeper(&ctx->cq_wait)) {
1209                 wake_up_interruptible(&ctx->cq_wait);
1210                 kill_fasync(&ctx->cq_fasync, SIGIO, POLL_IN);
1211         }
1212 }
1213
1214 static void io_put_identity(struct io_uring_task *tctx, struct io_kiocb *req)
1215 {
1216         if (req->work.identity == &tctx->__identity)
1217                 return;
1218         if (refcount_dec_and_test(&req->work.identity->count))
1219                 kfree(req->work.identity);
1220 }
1221
1222 static void io_req_clean_work(struct io_kiocb *req)
1223 {
1224         if (!(req->flags & REQ_F_WORK_INITIALIZED))
1225                 return;
1226
1227         req->flags &= ~REQ_F_WORK_INITIALIZED;
1228
1229         if (req->work.flags & IO_WQ_WORK_MM) {
1230                 mmdrop(req->work.identity->mm);
1231                 req->work.flags &= ~IO_WQ_WORK_MM;
1232         }
1233 #ifdef CONFIG_BLK_CGROUP
1234         if (req->work.flags & IO_WQ_WORK_BLKCG) {
1235                 css_put(req->work.identity->blkcg_css);
1236                 req->work.flags &= ~IO_WQ_WORK_BLKCG;
1237         }
1238 #endif
1239         if (req->work.flags & IO_WQ_WORK_CREDS) {
1240                 put_cred(req->work.identity->creds);
1241                 req->work.flags &= ~IO_WQ_WORK_CREDS;
1242         }
1243         if (req->work.flags & IO_WQ_WORK_FS) {
1244                 struct fs_struct *fs = req->work.identity->fs;
1245
1246                 spin_lock(&req->work.identity->fs->lock);
1247                 if (--fs->users)
1248                         fs = NULL;
1249                 spin_unlock(&req->work.identity->fs->lock);
1250                 if (fs)
1251                         free_fs_struct(fs);
1252                 req->work.flags &= ~IO_WQ_WORK_FS;
1253         }
1254
1255         io_put_identity(req->task->io_uring, req);
1256 }
1257
1258 /*
1259  * Create a private copy of io_identity, since some fields don't match
1260  * the current context.
1261  */
1262 static bool io_identity_cow(struct io_kiocb *req)
1263 {
1264         struct io_uring_task *tctx = current->io_uring;
1265         const struct cred *creds = NULL;
1266         struct io_identity *id;
1267
1268         if (req->work.flags & IO_WQ_WORK_CREDS)
1269                 creds = req->work.identity->creds;
1270
1271         id = kmemdup(req->work.identity, sizeof(*id), GFP_KERNEL);
1272         if (unlikely(!id)) {
1273                 req->work.flags |= IO_WQ_WORK_CANCEL;
1274                 return false;
1275         }
1276
1277         /*
1278          * We can safely just re-init the creds we copied  Either the field
1279          * matches the current one, or we haven't grabbed it yet. The only
1280          * exception is ->creds, through registered personalities, so handle
1281          * that one separately.
1282          */
1283         io_init_identity(id);
1284         if (creds)
1285                 req->work.identity->creds = creds;
1286
1287         /* add one for this request */
1288         refcount_inc(&id->count);
1289
1290         /* drop tctx and req identity references, if needed */
1291         if (tctx->identity != &tctx->__identity &&
1292             refcount_dec_and_test(&tctx->identity->count))
1293                 kfree(tctx->identity);
1294         if (req->work.identity != &tctx->__identity &&
1295             refcount_dec_and_test(&req->work.identity->count))
1296                 kfree(req->work.identity);
1297
1298         req->work.identity = id;
1299         tctx->identity = id;
1300         return true;
1301 }
1302
1303 static bool io_grab_identity(struct io_kiocb *req)
1304 {
1305         const struct io_op_def *def = &io_op_defs[req->opcode];
1306         struct io_identity *id = req->work.identity;
1307         struct io_ring_ctx *ctx = req->ctx;
1308
1309         if (def->work_flags & IO_WQ_WORK_FSIZE) {
1310                 if (id->fsize != rlimit(RLIMIT_FSIZE))
1311                         return false;
1312                 req->work.flags |= IO_WQ_WORK_FSIZE;
1313         }
1314
1315         if (!(req->work.flags & IO_WQ_WORK_FILES) &&
1316             (def->work_flags & IO_WQ_WORK_FILES) &&
1317             !(req->flags & REQ_F_NO_FILE_TABLE)) {
1318                 if (id->files != current->files ||
1319                     id->nsproxy != current->nsproxy)
1320                         return false;
1321                 atomic_inc(&id->files->count);
1322                 get_nsproxy(id->nsproxy);
1323                 req->flags |= REQ_F_INFLIGHT;
1324
1325                 spin_lock_irq(&ctx->inflight_lock);
1326                 list_add(&req->inflight_entry, &ctx->inflight_list);
1327                 spin_unlock_irq(&ctx->inflight_lock);
1328                 req->work.flags |= IO_WQ_WORK_FILES;
1329         }
1330 #ifdef CONFIG_BLK_CGROUP
1331         if (!(req->work.flags & IO_WQ_WORK_BLKCG) &&
1332             (def->work_flags & IO_WQ_WORK_BLKCG)) {
1333                 rcu_read_lock();
1334                 if (id->blkcg_css != blkcg_css()) {
1335                         rcu_read_unlock();
1336                         return false;
1337                 }
1338                 /*
1339                  * This should be rare, either the cgroup is dying or the task
1340                  * is moving cgroups. Just punt to root for the handful of ios.
1341                  */
1342                 if (css_tryget_online(id->blkcg_css))
1343                         req->work.flags |= IO_WQ_WORK_BLKCG;
1344                 rcu_read_unlock();
1345         }
1346 #endif
1347         if (!(req->work.flags & IO_WQ_WORK_CREDS)) {
1348                 if (id->creds != current_cred())
1349                         return false;
1350                 get_cred(id->creds);
1351                 req->work.flags |= IO_WQ_WORK_CREDS;
1352         }
1353 #ifdef CONFIG_AUDIT
1354         if (!uid_eq(current->loginuid, id->loginuid) ||
1355             current->sessionid != id->sessionid)
1356                 return false;
1357 #endif
1358         if (!(req->work.flags & IO_WQ_WORK_FS) &&
1359             (def->work_flags & IO_WQ_WORK_FS)) {
1360                 if (current->fs != id->fs)
1361                         return false;
1362                 spin_lock(&id->fs->lock);
1363                 if (!id->fs->in_exec) {
1364                         id->fs->users++;
1365                         req->work.flags |= IO_WQ_WORK_FS;
1366                 } else {
1367                         req->work.flags |= IO_WQ_WORK_CANCEL;
1368                 }
1369                 spin_unlock(&current->fs->lock);
1370         }
1371
1372         return true;
1373 }
1374
1375 static void io_prep_async_work(struct io_kiocb *req)
1376 {
1377         const struct io_op_def *def = &io_op_defs[req->opcode];
1378         struct io_ring_ctx *ctx = req->ctx;
1379         struct io_identity *id;
1380
1381         io_req_init_async(req);
1382         id = req->work.identity;
1383
1384         if (req->flags & REQ_F_FORCE_ASYNC)
1385                 req->work.flags |= IO_WQ_WORK_CONCURRENT;
1386
1387         if (req->flags & REQ_F_ISREG) {
1388                 if (def->hash_reg_file || (ctx->flags & IORING_SETUP_IOPOLL))
1389                         io_wq_hash_work(&req->work, file_inode(req->file));
1390         } else {
1391                 if (def->unbound_nonreg_file)
1392                         req->work.flags |= IO_WQ_WORK_UNBOUND;
1393         }
1394
1395         /* ->mm can never change on us */
1396         if (!(req->work.flags & IO_WQ_WORK_MM) &&
1397             (def->work_flags & IO_WQ_WORK_MM)) {
1398                 mmgrab(id->mm);
1399                 req->work.flags |= IO_WQ_WORK_MM;
1400         }
1401
1402         /* if we fail grabbing identity, we must COW, regrab, and retry */
1403         if (io_grab_identity(req))
1404                 return;
1405
1406         if (!io_identity_cow(req))
1407                 return;
1408
1409         /* can't fail at this point */
1410         if (!io_grab_identity(req))
1411                 WARN_ON(1);
1412 }
1413
1414 static void io_prep_async_link(struct io_kiocb *req)
1415 {
1416         struct io_kiocb *cur;
1417
1418         io_prep_async_work(req);
1419         if (req->flags & REQ_F_LINK_HEAD)
1420                 list_for_each_entry(cur, &req->link_list, link_list)
1421                         io_prep_async_work(cur);
1422 }
1423
1424 static struct io_kiocb *__io_queue_async_work(struct io_kiocb *req)
1425 {
1426         struct io_ring_ctx *ctx = req->ctx;
1427         struct io_kiocb *link = io_prep_linked_timeout(req);
1428
1429         trace_io_uring_queue_async_work(ctx, io_wq_is_hashed(&req->work), req,
1430                                         &req->work, req->flags);
1431         io_wq_enqueue(ctx->io_wq, &req->work);
1432         return link;
1433 }
1434
1435 static void io_queue_async_work(struct io_kiocb *req)
1436 {
1437         struct io_kiocb *link;
1438
1439         /* init ->work of the whole link before punting */
1440         io_prep_async_link(req);
1441         link = __io_queue_async_work(req);
1442
1443         if (link)
1444                 io_queue_linked_timeout(link);
1445 }
1446
1447 static void io_kill_timeout(struct io_kiocb *req)
1448 {
1449         struct io_timeout_data *io = req->async_data;
1450         int ret;
1451
1452         ret = hrtimer_try_to_cancel(&io->timer);
1453         if (ret != -1) {
1454                 atomic_set(&req->ctx->cq_timeouts,
1455                         atomic_read(&req->ctx->cq_timeouts) + 1);
1456                 list_del_init(&req->timeout.list);
1457                 io_cqring_fill_event(req, 0);
1458                 io_put_req_deferred(req, 1);
1459         }
1460 }
1461
1462 static bool io_task_match(struct io_kiocb *req, struct task_struct *tsk)
1463 {
1464         struct io_ring_ctx *ctx = req->ctx;
1465
1466         if (!tsk || req->task == tsk)
1467                 return true;
1468         if (ctx->flags & IORING_SETUP_SQPOLL) {
1469                 if (ctx->sq_data && req->task == ctx->sq_data->thread)
1470                         return true;
1471         }
1472         return false;
1473 }
1474
1475 /*
1476  * Returns true if we found and killed one or more timeouts
1477  */
1478 static bool io_kill_timeouts(struct io_ring_ctx *ctx, struct task_struct *tsk)
1479 {
1480         struct io_kiocb *req, *tmp;
1481         int canceled = 0;
1482
1483         spin_lock_irq(&ctx->completion_lock);
1484         list_for_each_entry_safe(req, tmp, &ctx->timeout_list, timeout.list) {
1485                 if (io_task_match(req, tsk)) {
1486                         io_kill_timeout(req);
1487                         canceled++;
1488                 }
1489         }
1490         spin_unlock_irq(&ctx->completion_lock);
1491         return canceled != 0;
1492 }
1493
1494 static void __io_queue_deferred(struct io_ring_ctx *ctx)
1495 {
1496         do {
1497                 struct io_defer_entry *de = list_first_entry(&ctx->defer_list,
1498                                                 struct io_defer_entry, list);
1499                 struct io_kiocb *link;
1500
1501                 if (req_need_defer(de->req, de->seq))
1502                         break;
1503                 list_del_init(&de->list);
1504                 /* punt-init is done before queueing for defer */
1505                 link = __io_queue_async_work(de->req);
1506                 if (link) {
1507                         __io_queue_linked_timeout(link);
1508                         /* drop submission reference */
1509                         io_put_req_deferred(link, 1);
1510                 }
1511                 kfree(de);
1512         } while (!list_empty(&ctx->defer_list));
1513 }
1514
1515 static void io_flush_timeouts(struct io_ring_ctx *ctx)
1516 {
1517         while (!list_empty(&ctx->timeout_list)) {
1518                 struct io_kiocb *req = list_first_entry(&ctx->timeout_list,
1519                                                 struct io_kiocb, timeout.list);
1520
1521                 if (io_is_timeout_noseq(req))
1522                         break;
1523                 if (req->timeout.target_seq != ctx->cached_cq_tail
1524                                         - atomic_read(&ctx->cq_timeouts))
1525                         break;
1526
1527                 list_del_init(&req->timeout.list);
1528                 io_kill_timeout(req);
1529         }
1530 }
1531
1532 static void io_commit_cqring(struct io_ring_ctx *ctx)
1533 {
1534         io_flush_timeouts(ctx);
1535         __io_commit_cqring(ctx);
1536
1537         if (unlikely(!list_empty(&ctx->defer_list)))
1538                 __io_queue_deferred(ctx);
1539 }
1540
1541 static inline bool io_sqring_full(struct io_ring_ctx *ctx)
1542 {
1543         struct io_rings *r = ctx->rings;
1544
1545         return READ_ONCE(r->sq.tail) - ctx->cached_sq_head == r->sq_ring_entries;
1546 }
1547
1548 static struct io_uring_cqe *io_get_cqring(struct io_ring_ctx *ctx)
1549 {
1550         struct io_rings *rings = ctx->rings;
1551         unsigned tail;
1552
1553         tail = ctx->cached_cq_tail;
1554         /*
1555          * writes to the cq entry need to come after reading head; the
1556          * control dependency is enough as we're using WRITE_ONCE to
1557          * fill the cq entry
1558          */
1559         if (tail - READ_ONCE(rings->cq.head) == rings->cq_ring_entries)
1560                 return NULL;
1561
1562         ctx->cached_cq_tail++;
1563         return &rings->cqes[tail & ctx->cq_mask];
1564 }
1565
1566 static inline bool io_should_trigger_evfd(struct io_ring_ctx *ctx)
1567 {
1568         if (!ctx->cq_ev_fd)
1569                 return false;
1570         if (READ_ONCE(ctx->rings->cq_flags) & IORING_CQ_EVENTFD_DISABLED)
1571                 return false;
1572         if (!ctx->eventfd_async)
1573                 return true;
1574         return io_wq_current_is_worker();
1575 }
1576
1577 static void io_cqring_ev_posted(struct io_ring_ctx *ctx)
1578 {
1579         if (waitqueue_active(&ctx->wait))
1580                 wake_up(&ctx->wait);
1581         if (ctx->sq_data && waitqueue_active(&ctx->sq_data->wait))
1582                 wake_up(&ctx->sq_data->wait);
1583         if (io_should_trigger_evfd(ctx))
1584                 eventfd_signal(ctx->cq_ev_fd, 1);
1585 }
1586
1587 static void io_cqring_mark_overflow(struct io_ring_ctx *ctx)
1588 {
1589         if (list_empty(&ctx->cq_overflow_list)) {
1590                 clear_bit(0, &ctx->sq_check_overflow);
1591                 clear_bit(0, &ctx->cq_check_overflow);
1592                 ctx->rings->sq_flags &= ~IORING_SQ_CQ_OVERFLOW;
1593         }
1594 }
1595
1596 static inline bool __io_match_files(struct io_kiocb *req,
1597                                     struct files_struct *files)
1598 {
1599         return ((req->flags & REQ_F_WORK_INITIALIZED) &&
1600                 (req->work.flags & IO_WQ_WORK_FILES)) &&
1601                 req->work.identity->files == files;
1602 }
1603
1604 static bool io_match_files(struct io_kiocb *req,
1605                            struct files_struct *files)
1606 {
1607         struct io_kiocb *link;
1608
1609         if (!files)
1610                 return true;
1611         if (__io_match_files(req, files))
1612                 return true;
1613         if (req->flags & REQ_F_LINK_HEAD) {
1614                 list_for_each_entry(link, &req->link_list, link_list) {
1615                         if (__io_match_files(link, files))
1616                                 return true;
1617                 }
1618         }
1619         return false;
1620 }
1621
1622 /* Returns true if there are no backlogged entries after the flush */
1623 static bool io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force,
1624                                      struct task_struct *tsk,
1625                                      struct files_struct *files)
1626 {
1627         struct io_rings *rings = ctx->rings;
1628         struct io_kiocb *req, *tmp;
1629         struct io_uring_cqe *cqe;
1630         unsigned long flags;
1631         LIST_HEAD(list);
1632
1633         if (!force) {
1634                 if (list_empty_careful(&ctx->cq_overflow_list))
1635                         return true;
1636                 if ((ctx->cached_cq_tail - READ_ONCE(rings->cq.head) ==
1637                     rings->cq_ring_entries))
1638                         return false;
1639         }
1640
1641         spin_lock_irqsave(&ctx->completion_lock, flags);
1642
1643         /* if force is set, the ring is going away. always drop after that */
1644         if (force)
1645                 ctx->cq_overflow_flushed = 1;
1646
1647         cqe = NULL;
1648         list_for_each_entry_safe(req, tmp, &ctx->cq_overflow_list, compl.list) {
1649                 if (tsk && req->task != tsk)
1650                         continue;
1651                 if (!io_match_files(req, files))
1652                         continue;
1653
1654                 cqe = io_get_cqring(ctx);
1655                 if (!cqe && !force)
1656                         break;
1657
1658                 list_move(&req->compl.list, &list);
1659                 if (cqe) {
1660                         WRITE_ONCE(cqe->user_data, req->user_data);
1661                         WRITE_ONCE(cqe->res, req->result);
1662                         WRITE_ONCE(cqe->flags, req->compl.cflags);
1663                 } else {
1664                         ctx->cached_cq_overflow++;
1665                         WRITE_ONCE(ctx->rings->cq_overflow,
1666                                    ctx->cached_cq_overflow);
1667                 }
1668         }
1669
1670         io_commit_cqring(ctx);
1671         io_cqring_mark_overflow(ctx);
1672
1673         spin_unlock_irqrestore(&ctx->completion_lock, flags);
1674         io_cqring_ev_posted(ctx);
1675
1676         while (!list_empty(&list)) {
1677                 req = list_first_entry(&list, struct io_kiocb, compl.list);
1678                 list_del(&req->compl.list);
1679                 io_put_req(req);
1680         }
1681
1682         return cqe != NULL;
1683 }
1684
1685 static void __io_cqring_fill_event(struct io_kiocb *req, long res, long cflags)
1686 {
1687         struct io_ring_ctx *ctx = req->ctx;
1688         struct io_uring_cqe *cqe;
1689
1690         trace_io_uring_complete(ctx, req->user_data, res);
1691
1692         /*
1693          * If we can't get a cq entry, userspace overflowed the
1694          * submission (by quite a lot). Increment the overflow count in
1695          * the ring.
1696          */
1697         cqe = io_get_cqring(ctx);
1698         if (likely(cqe)) {
1699                 WRITE_ONCE(cqe->user_data, req->user_data);
1700                 WRITE_ONCE(cqe->res, res);
1701                 WRITE_ONCE(cqe->flags, cflags);
1702         } else if (ctx->cq_overflow_flushed ||
1703                    atomic_read(&req->task->io_uring->in_idle)) {
1704                 /*
1705                  * If we're in ring overflow flush mode, or in task cancel mode,
1706                  * then we cannot store the request for later flushing, we need
1707                  * to drop it on the floor.
1708                  */
1709                 ctx->cached_cq_overflow++;
1710                 WRITE_ONCE(ctx->rings->cq_overflow, ctx->cached_cq_overflow);
1711         } else {
1712                 if (list_empty(&ctx->cq_overflow_list)) {
1713                         set_bit(0, &ctx->sq_check_overflow);
1714                         set_bit(0, &ctx->cq_check_overflow);
1715                         ctx->rings->sq_flags |= IORING_SQ_CQ_OVERFLOW;
1716                 }
1717                 io_clean_op(req);
1718                 req->result = res;
1719                 req->compl.cflags = cflags;
1720                 refcount_inc(&req->refs);
1721                 list_add_tail(&req->compl.list, &ctx->cq_overflow_list);
1722         }
1723 }
1724
1725 static void io_cqring_fill_event(struct io_kiocb *req, long res)
1726 {
1727         __io_cqring_fill_event(req, res, 0);
1728 }
1729
1730 static void io_cqring_add_event(struct io_kiocb *req, long res, long cflags)
1731 {
1732         struct io_ring_ctx *ctx = req->ctx;
1733         unsigned long flags;
1734
1735         spin_lock_irqsave(&ctx->completion_lock, flags);
1736         __io_cqring_fill_event(req, res, cflags);
1737         io_commit_cqring(ctx);
1738         spin_unlock_irqrestore(&ctx->completion_lock, flags);
1739
1740         io_cqring_ev_posted(ctx);
1741 }
1742
1743 static void io_submit_flush_completions(struct io_comp_state *cs)
1744 {
1745         struct io_ring_ctx *ctx = cs->ctx;
1746
1747         spin_lock_irq(&ctx->completion_lock);
1748         while (!list_empty(&cs->list)) {
1749                 struct io_kiocb *req;
1750
1751                 req = list_first_entry(&cs->list, struct io_kiocb, compl.list);
1752                 list_del(&req->compl.list);
1753                 __io_cqring_fill_event(req, req->result, req->compl.cflags);
1754
1755                 /*
1756                  * io_free_req() doesn't care about completion_lock unless one
1757                  * of these flags is set. REQ_F_WORK_INITIALIZED is in the list
1758                  * because of a potential deadlock with req->work.fs->lock
1759                  */
1760                 if (req->flags & (REQ_F_FAIL_LINK|REQ_F_LINK_TIMEOUT
1761                                  |REQ_F_WORK_INITIALIZED)) {
1762                         spin_unlock_irq(&ctx->completion_lock);
1763                         io_put_req(req);
1764                         spin_lock_irq(&ctx->completion_lock);
1765                 } else {
1766                         io_put_req(req);
1767                 }
1768         }
1769         io_commit_cqring(ctx);
1770         spin_unlock_irq(&ctx->completion_lock);
1771
1772         io_cqring_ev_posted(ctx);
1773         cs->nr = 0;
1774 }
1775
1776 static void __io_req_complete(struct io_kiocb *req, long res, unsigned cflags,
1777                               struct io_comp_state *cs)
1778 {
1779         if (!cs) {
1780                 io_cqring_add_event(req, res, cflags);
1781                 io_put_req(req);
1782         } else {
1783                 io_clean_op(req);
1784                 req->result = res;
1785                 req->compl.cflags = cflags;
1786                 list_add_tail(&req->compl.list, &cs->list);
1787                 if (++cs->nr >= 32)
1788                         io_submit_flush_completions(cs);
1789         }
1790 }
1791
1792 static void io_req_complete(struct io_kiocb *req, long res)
1793 {
1794         __io_req_complete(req, res, 0, NULL);
1795 }
1796
1797 static inline bool io_is_fallback_req(struct io_kiocb *req)
1798 {
1799         return req == (struct io_kiocb *)
1800                         ((unsigned long) req->ctx->fallback_req & ~1UL);
1801 }
1802
1803 static struct io_kiocb *io_get_fallback_req(struct io_ring_ctx *ctx)
1804 {
1805         struct io_kiocb *req;
1806
1807         req = ctx->fallback_req;
1808         if (!test_and_set_bit_lock(0, (unsigned long *) &ctx->fallback_req))
1809                 return req;
1810
1811         return NULL;
1812 }
1813
1814 static struct io_kiocb *io_alloc_req(struct io_ring_ctx *ctx,
1815                                      struct io_submit_state *state)
1816 {
1817         if (!state->free_reqs) {
1818                 gfp_t gfp = GFP_KERNEL | __GFP_NOWARN;
1819                 size_t sz;
1820                 int ret;
1821
1822                 sz = min_t(size_t, state->ios_left, ARRAY_SIZE(state->reqs));
1823                 ret = kmem_cache_alloc_bulk(req_cachep, gfp, sz, state->reqs);
1824
1825                 /*
1826                  * Bulk alloc is all-or-nothing. If we fail to get a batch,
1827                  * retry single alloc to be on the safe side.
1828                  */
1829                 if (unlikely(ret <= 0)) {
1830                         state->reqs[0] = kmem_cache_alloc(req_cachep, gfp);
1831                         if (!state->reqs[0])
1832                                 goto fallback;
1833                         ret = 1;
1834                 }
1835                 state->free_reqs = ret;
1836         }
1837
1838         state->free_reqs--;
1839         return state->reqs[state->free_reqs];
1840 fallback:
1841         return io_get_fallback_req(ctx);
1842 }
1843
1844 static inline void io_put_file(struct io_kiocb *req, struct file *file,
1845                           bool fixed)
1846 {
1847         if (fixed)
1848                 percpu_ref_put(req->fixed_file_refs);
1849         else
1850                 fput(file);
1851 }
1852
1853 static void io_dismantle_req(struct io_kiocb *req)
1854 {
1855         io_clean_op(req);
1856
1857         if (req->async_data)
1858                 kfree(req->async_data);
1859         if (req->file)
1860                 io_put_file(req, req->file, (req->flags & REQ_F_FIXED_FILE));
1861
1862         io_req_clean_work(req);
1863 }
1864
1865 static void __io_free_req(struct io_kiocb *req)
1866 {
1867         struct io_uring_task *tctx = req->task->io_uring;
1868         struct io_ring_ctx *ctx = req->ctx;
1869
1870         io_dismantle_req(req);
1871
1872         percpu_counter_dec(&tctx->inflight);
1873         if (atomic_read(&tctx->in_idle))
1874                 wake_up(&tctx->wait);
1875         put_task_struct(req->task);
1876
1877         if (likely(!io_is_fallback_req(req)))
1878                 kmem_cache_free(req_cachep, req);
1879         else
1880                 clear_bit_unlock(0, (unsigned long *) &ctx->fallback_req);
1881         percpu_ref_put(&ctx->refs);
1882 }
1883
1884 static void io_kill_linked_timeout(struct io_kiocb *req)
1885 {
1886         struct io_ring_ctx *ctx = req->ctx;
1887         struct io_kiocb *link;
1888         bool cancelled = false;
1889         unsigned long flags;
1890
1891         spin_lock_irqsave(&ctx->completion_lock, flags);
1892         link = list_first_entry_or_null(&req->link_list, struct io_kiocb,
1893                                         link_list);
1894         /*
1895          * Can happen if a linked timeout fired and link had been like
1896          * req -> link t-out -> link t-out [-> ...]
1897          */
1898         if (link && (link->flags & REQ_F_LTIMEOUT_ACTIVE)) {
1899                 struct io_timeout_data *io = link->async_data;
1900                 int ret;
1901
1902                 list_del_init(&link->link_list);
1903                 ret = hrtimer_try_to_cancel(&io->timer);
1904                 if (ret != -1) {
1905                         io_cqring_fill_event(link, -ECANCELED);
1906                         io_commit_cqring(ctx);
1907                         cancelled = true;
1908                 }
1909         }
1910         req->flags &= ~REQ_F_LINK_TIMEOUT;
1911         spin_unlock_irqrestore(&ctx->completion_lock, flags);
1912
1913         if (cancelled) {
1914                 io_cqring_ev_posted(ctx);
1915                 io_put_req(link);
1916         }
1917 }
1918
1919 static struct io_kiocb *io_req_link_next(struct io_kiocb *req)
1920 {
1921         struct io_kiocb *nxt;
1922
1923         /*
1924          * The list should never be empty when we are called here. But could
1925          * potentially happen if the chain is messed up, check to be on the
1926          * safe side.
1927          */
1928         if (unlikely(list_empty(&req->link_list)))
1929                 return NULL;
1930
1931         nxt = list_first_entry(&req->link_list, struct io_kiocb, link_list);
1932         list_del_init(&req->link_list);
1933         if (!list_empty(&nxt->link_list))
1934                 nxt->flags |= REQ_F_LINK_HEAD;
1935         return nxt;
1936 }
1937
1938 /*
1939  * Called if REQ_F_LINK_HEAD is set, and we fail the head request
1940  */
1941 static void io_fail_links(struct io_kiocb *req)
1942 {
1943         struct io_ring_ctx *ctx = req->ctx;
1944         unsigned long flags;
1945
1946         spin_lock_irqsave(&ctx->completion_lock, flags);
1947         while (!list_empty(&req->link_list)) {
1948                 struct io_kiocb *link = list_first_entry(&req->link_list,
1949                                                 struct io_kiocb, link_list);
1950
1951                 list_del_init(&link->link_list);
1952                 trace_io_uring_fail_link(req, link);
1953
1954                 io_cqring_fill_event(link, -ECANCELED);
1955
1956                 /*
1957                  * It's ok to free under spinlock as they're not linked anymore,
1958                  * but avoid REQ_F_WORK_INITIALIZED because it may deadlock on
1959                  * work.fs->lock.
1960                  */
1961                 if (link->flags & REQ_F_WORK_INITIALIZED)
1962                         io_put_req_deferred(link, 2);
1963                 else
1964                         io_double_put_req(link);
1965         }
1966
1967         io_commit_cqring(ctx);
1968         spin_unlock_irqrestore(&ctx->completion_lock, flags);
1969
1970         io_cqring_ev_posted(ctx);
1971 }
1972
1973 static struct io_kiocb *__io_req_find_next(struct io_kiocb *req)
1974 {
1975         req->flags &= ~REQ_F_LINK_HEAD;
1976         if (req->flags & REQ_F_LINK_TIMEOUT)
1977                 io_kill_linked_timeout(req);
1978
1979         /*
1980          * If LINK is set, we have dependent requests in this chain. If we
1981          * didn't fail this request, queue the first one up, moving any other
1982          * dependencies to the next request. In case of failure, fail the rest
1983          * of the chain.
1984          */
1985         if (likely(!(req->flags & REQ_F_FAIL_LINK)))
1986                 return io_req_link_next(req);
1987         io_fail_links(req);
1988         return NULL;
1989 }
1990
1991 static struct io_kiocb *io_req_find_next(struct io_kiocb *req)
1992 {
1993         if (likely(!(req->flags & REQ_F_LINK_HEAD)))
1994                 return NULL;
1995         return __io_req_find_next(req);
1996 }
1997
1998 static int io_req_task_work_add(struct io_kiocb *req, bool twa_signal_ok)
1999 {
2000         struct task_struct *tsk = req->task;
2001         struct io_ring_ctx *ctx = req->ctx;
2002         enum task_work_notify_mode notify;
2003         int ret;
2004
2005         if (tsk->flags & PF_EXITING)
2006                 return -ESRCH;
2007
2008         /*
2009          * SQPOLL kernel thread doesn't need notification, just a wakeup. For
2010          * all other cases, use TWA_SIGNAL unconditionally to ensure we're
2011          * processing task_work. There's no reliable way to tell if TWA_RESUME
2012          * will do the job.
2013          */
2014         notify = TWA_NONE;
2015         if (!(ctx->flags & IORING_SETUP_SQPOLL) && twa_signal_ok)
2016                 notify = TWA_SIGNAL;
2017
2018         ret = task_work_add(tsk, &req->task_work, notify);
2019         if (!ret)
2020                 wake_up_process(tsk);
2021
2022         return ret;
2023 }
2024
2025 static void __io_req_task_cancel(struct io_kiocb *req, int error)
2026 {
2027         struct io_ring_ctx *ctx = req->ctx;
2028
2029         spin_lock_irq(&ctx->completion_lock);
2030         io_cqring_fill_event(req, error);
2031         io_commit_cqring(ctx);
2032         spin_unlock_irq(&ctx->completion_lock);
2033
2034         io_cqring_ev_posted(ctx);
2035         req_set_fail_links(req);
2036         io_double_put_req(req);
2037 }
2038
2039 static void io_req_task_cancel(struct callback_head *cb)
2040 {
2041         struct io_kiocb *req = container_of(cb, struct io_kiocb, task_work);
2042         struct io_ring_ctx *ctx = req->ctx;
2043
2044         __io_req_task_cancel(req, -ECANCELED);
2045         percpu_ref_put(&ctx->refs);
2046 }
2047
2048 static void __io_req_task_submit(struct io_kiocb *req)
2049 {
2050         struct io_ring_ctx *ctx = req->ctx;
2051
2052         if (!__io_sq_thread_acquire_mm(ctx)) {
2053                 mutex_lock(&ctx->uring_lock);
2054                 __io_queue_sqe(req, NULL);
2055                 mutex_unlock(&ctx->uring_lock);
2056         } else {
2057                 __io_req_task_cancel(req, -EFAULT);
2058         }
2059 }
2060
2061 static void io_req_task_submit(struct callback_head *cb)
2062 {
2063         struct io_kiocb *req = container_of(cb, struct io_kiocb, task_work);
2064         struct io_ring_ctx *ctx = req->ctx;
2065
2066         __io_req_task_submit(req);
2067         percpu_ref_put(&ctx->refs);
2068 }
2069
2070 static void io_req_task_queue(struct io_kiocb *req)
2071 {
2072         int ret;
2073
2074         init_task_work(&req->task_work, io_req_task_submit);
2075         percpu_ref_get(&req->ctx->refs);
2076
2077         ret = io_req_task_work_add(req, true);
2078         if (unlikely(ret)) {
2079                 struct task_struct *tsk;
2080
2081                 init_task_work(&req->task_work, io_req_task_cancel);
2082                 tsk = io_wq_get_task(req->ctx->io_wq);
2083                 task_work_add(tsk, &req->task_work, TWA_NONE);
2084                 wake_up_process(tsk);
2085         }
2086 }
2087
2088 static void io_queue_next(struct io_kiocb *req)
2089 {
2090         struct io_kiocb *nxt = io_req_find_next(req);
2091
2092         if (nxt)
2093                 io_req_task_queue(nxt);
2094 }
2095
2096 static void io_free_req(struct io_kiocb *req)
2097 {
2098         io_queue_next(req);
2099         __io_free_req(req);
2100 }
2101
2102 struct req_batch {
2103         void *reqs[IO_IOPOLL_BATCH];
2104         int to_free;
2105
2106         struct task_struct      *task;
2107         int                     task_refs;
2108 };
2109
2110 static inline void io_init_req_batch(struct req_batch *rb)
2111 {
2112         rb->to_free = 0;
2113         rb->task_refs = 0;
2114         rb->task = NULL;
2115 }
2116
2117 static void __io_req_free_batch_flush(struct io_ring_ctx *ctx,
2118                                       struct req_batch *rb)
2119 {
2120         kmem_cache_free_bulk(req_cachep, rb->to_free, rb->reqs);
2121         percpu_ref_put_many(&ctx->refs, rb->to_free);
2122         rb->to_free = 0;
2123 }
2124
2125 static void io_req_free_batch_finish(struct io_ring_ctx *ctx,
2126                                      struct req_batch *rb)
2127 {
2128         if (rb->to_free)
2129                 __io_req_free_batch_flush(ctx, rb);
2130         if (rb->task) {
2131                 struct io_uring_task *tctx = rb->task->io_uring;
2132
2133                 percpu_counter_sub(&tctx->inflight, rb->task_refs);
2134                 put_task_struct_many(rb->task, rb->task_refs);
2135                 rb->task = NULL;
2136         }
2137 }
2138
2139 static void io_req_free_batch(struct req_batch *rb, struct io_kiocb *req)
2140 {
2141         if (unlikely(io_is_fallback_req(req))) {
2142                 io_free_req(req);
2143                 return;
2144         }
2145         if (req->flags & REQ_F_LINK_HEAD)
2146                 io_queue_next(req);
2147
2148         if (req->task != rb->task) {
2149                 if (rb->task) {
2150                         struct io_uring_task *tctx = rb->task->io_uring;
2151
2152                         percpu_counter_sub(&tctx->inflight, rb->task_refs);
2153                         put_task_struct_many(rb->task, rb->task_refs);
2154                 }
2155                 rb->task = req->task;
2156                 rb->task_refs = 0;
2157         }
2158         rb->task_refs++;
2159
2160         io_dismantle_req(req);
2161         rb->reqs[rb->to_free++] = req;
2162         if (unlikely(rb->to_free == ARRAY_SIZE(rb->reqs)))
2163                 __io_req_free_batch_flush(req->ctx, rb);
2164 }
2165
2166 /*
2167  * Drop reference to request, return next in chain (if there is one) if this
2168  * was the last reference to this request.
2169  */
2170 static struct io_kiocb *io_put_req_find_next(struct io_kiocb *req)
2171 {
2172         struct io_kiocb *nxt = NULL;
2173
2174         if (refcount_dec_and_test(&req->refs)) {
2175                 nxt = io_req_find_next(req);
2176                 __io_free_req(req);
2177         }
2178         return nxt;
2179 }
2180
2181 static void io_put_req(struct io_kiocb *req)
2182 {
2183         if (refcount_dec_and_test(&req->refs))
2184                 io_free_req(req);
2185 }
2186
2187 static void io_put_req_deferred_cb(struct callback_head *cb)
2188 {
2189         struct io_kiocb *req = container_of(cb, struct io_kiocb, task_work);
2190
2191         io_free_req(req);
2192 }
2193
2194 static void io_free_req_deferred(struct io_kiocb *req)
2195 {
2196         int ret;
2197
2198         init_task_work(&req->task_work, io_put_req_deferred_cb);
2199         ret = io_req_task_work_add(req, true);
2200         if (unlikely(ret)) {
2201                 struct task_struct *tsk;
2202
2203                 tsk = io_wq_get_task(req->ctx->io_wq);
2204                 task_work_add(tsk, &req->task_work, TWA_NONE);
2205                 wake_up_process(tsk);
2206         }
2207 }
2208
2209 static inline void io_put_req_deferred(struct io_kiocb *req, int refs)
2210 {
2211         if (refcount_sub_and_test(refs, &req->refs))
2212                 io_free_req_deferred(req);
2213 }
2214
2215 static struct io_wq_work *io_steal_work(struct io_kiocb *req)
2216 {
2217         struct io_kiocb *nxt;
2218
2219         /*
2220          * A ref is owned by io-wq in which context we're. So, if that's the
2221          * last one, it's safe to steal next work. False negatives are Ok,
2222          * it just will be re-punted async in io_put_work()
2223          */
2224         if (refcount_read(&req->refs) != 1)
2225                 return NULL;
2226
2227         nxt = io_req_find_next(req);
2228         return nxt ? &nxt->work : NULL;
2229 }
2230
2231 static void io_double_put_req(struct io_kiocb *req)
2232 {
2233         /* drop both submit and complete references */
2234         if (refcount_sub_and_test(2, &req->refs))
2235                 io_free_req(req);
2236 }
2237
2238 static unsigned io_cqring_events(struct io_ring_ctx *ctx, bool noflush)
2239 {
2240         struct io_rings *rings = ctx->rings;
2241
2242         if (test_bit(0, &ctx->cq_check_overflow)) {
2243                 /*
2244                  * noflush == true is from the waitqueue handler, just ensure
2245                  * we wake up the task, and the next invocation will flush the
2246                  * entries. We cannot safely to it from here.
2247                  */
2248                 if (noflush && !list_empty(&ctx->cq_overflow_list))
2249                         return -1U;
2250
2251                 io_cqring_overflow_flush(ctx, false, NULL, NULL);
2252         }
2253
2254         /* See comment at the top of this file */
2255         smp_rmb();
2256         return ctx->cached_cq_tail - READ_ONCE(rings->cq.head);
2257 }
2258
2259 static inline unsigned int io_sqring_entries(struct io_ring_ctx *ctx)
2260 {
2261         struct io_rings *rings = ctx->rings;
2262
2263         /* make sure SQ entry isn't read before tail */
2264         return smp_load_acquire(&rings->sq.tail) - ctx->cached_sq_head;
2265 }
2266
2267 static unsigned int io_put_kbuf(struct io_kiocb *req, struct io_buffer *kbuf)
2268 {
2269         unsigned int cflags;
2270
2271         cflags = kbuf->bid << IORING_CQE_BUFFER_SHIFT;
2272         cflags |= IORING_CQE_F_BUFFER;
2273         req->flags &= ~REQ_F_BUFFER_SELECTED;
2274         kfree(kbuf);
2275         return cflags;
2276 }
2277
2278 static inline unsigned int io_put_rw_kbuf(struct io_kiocb *req)
2279 {
2280         struct io_buffer *kbuf;
2281
2282         kbuf = (struct io_buffer *) (unsigned long) req->rw.addr;
2283         return io_put_kbuf(req, kbuf);
2284 }
2285
2286 static inline bool io_run_task_work(void)
2287 {
2288         /*
2289          * Not safe to run on exiting task, and the task_work handling will
2290          * not add work to such a task.
2291          */
2292         if (unlikely(current->flags & PF_EXITING))
2293                 return false;
2294         if (current->task_works) {
2295                 __set_current_state(TASK_RUNNING);
2296                 task_work_run();
2297                 return true;
2298         }
2299
2300         return false;
2301 }
2302
2303 static void io_iopoll_queue(struct list_head *again)
2304 {
2305         struct io_kiocb *req;
2306
2307         do {
2308                 req = list_first_entry(again, struct io_kiocb, inflight_entry);
2309                 list_del(&req->inflight_entry);
2310                 __io_complete_rw(req, -EAGAIN, 0, NULL);
2311         } while (!list_empty(again));
2312 }
2313
2314 /*
2315  * Find and free completed poll iocbs
2316  */
2317 static void io_iopoll_complete(struct io_ring_ctx *ctx, unsigned int *nr_events,
2318                                struct list_head *done)
2319 {
2320         struct req_batch rb;
2321         struct io_kiocb *req;
2322         LIST_HEAD(again);
2323
2324         /* order with ->result store in io_complete_rw_iopoll() */
2325         smp_rmb();
2326
2327         io_init_req_batch(&rb);
2328         while (!list_empty(done)) {
2329                 int cflags = 0;
2330
2331                 req = list_first_entry(done, struct io_kiocb, inflight_entry);
2332                 if (READ_ONCE(req->result) == -EAGAIN) {
2333                         req->result = 0;
2334                         req->iopoll_completed = 0;
2335                         list_move_tail(&req->inflight_entry, &again);
2336                         continue;
2337                 }
2338                 list_del(&req->inflight_entry);
2339
2340                 if (req->flags & REQ_F_BUFFER_SELECTED)
2341                         cflags = io_put_rw_kbuf(req);
2342
2343                 __io_cqring_fill_event(req, req->result, cflags);
2344                 (*nr_events)++;
2345
2346                 if (refcount_dec_and_test(&req->refs))
2347                         io_req_free_batch(&rb, req);
2348         }
2349
2350         io_commit_cqring(ctx);
2351         if (ctx->flags & IORING_SETUP_SQPOLL)
2352                 io_cqring_ev_posted(ctx);
2353         io_req_free_batch_finish(ctx, &rb);
2354
2355         if (!list_empty(&again))
2356                 io_iopoll_queue(&again);
2357 }
2358
2359 static int io_do_iopoll(struct io_ring_ctx *ctx, unsigned int *nr_events,
2360                         long min)
2361 {
2362         struct io_kiocb *req, *tmp;
2363         LIST_HEAD(done);
2364         bool spin;
2365         int ret;
2366
2367         /*
2368          * Only spin for completions if we don't have multiple devices hanging
2369          * off our complete list, and we're under the requested amount.
2370          */
2371         spin = !ctx->poll_multi_file && *nr_events < min;
2372
2373         ret = 0;
2374         list_for_each_entry_safe(req, tmp, &ctx->iopoll_list, inflight_entry) {
2375                 struct kiocb *kiocb = &req->rw.kiocb;
2376
2377                 /*
2378                  * Move completed and retryable entries to our local lists.
2379                  * If we find a request that requires polling, break out
2380                  * and complete those lists first, if we have entries there.
2381                  */
2382                 if (READ_ONCE(req->iopoll_completed)) {
2383                         list_move_tail(&req->inflight_entry, &done);
2384                         continue;
2385                 }
2386                 if (!list_empty(&done))
2387                         break;
2388
2389                 ret = kiocb->ki_filp->f_op->iopoll(kiocb, spin);
2390                 if (ret < 0)
2391                         break;
2392
2393                 /* iopoll may have completed current req */
2394                 if (READ_ONCE(req->iopoll_completed))
2395                         list_move_tail(&req->inflight_entry, &done);
2396
2397                 if (ret && spin)
2398                         spin = false;
2399                 ret = 0;
2400         }
2401
2402         if (!list_empty(&done))
2403                 io_iopoll_complete(ctx, nr_events, &done);
2404
2405         return ret;
2406 }
2407
2408 /*
2409  * Poll for a minimum of 'min' events. Note that if min == 0 we consider that a
2410  * non-spinning poll check - we'll still enter the driver poll loop, but only
2411  * as a non-spinning completion check.
2412  */
2413 static int io_iopoll_getevents(struct io_ring_ctx *ctx, unsigned int *nr_events,
2414                                 long min)
2415 {
2416         while (!list_empty(&ctx->iopoll_list) && !need_resched()) {
2417                 int ret;
2418
2419                 ret = io_do_iopoll(ctx, nr_events, min);
2420                 if (ret < 0)
2421                         return ret;
2422                 if (*nr_events >= min)
2423                         return 0;
2424         }
2425
2426         return 1;
2427 }
2428
2429 /*
2430  * We can't just wait for polled events to come to us, we have to actively
2431  * find and complete them.
2432  */
2433 static void io_iopoll_try_reap_events(struct io_ring_ctx *ctx)
2434 {
2435         if (!(ctx->flags & IORING_SETUP_IOPOLL))
2436                 return;
2437
2438         mutex_lock(&ctx->uring_lock);
2439         while (!list_empty(&ctx->iopoll_list)) {
2440                 unsigned int nr_events = 0;
2441
2442                 io_do_iopoll(ctx, &nr_events, 0);
2443
2444                 /* let it sleep and repeat later if can't complete a request */
2445                 if (nr_events == 0)
2446                         break;
2447                 /*
2448                  * Ensure we allow local-to-the-cpu processing to take place,
2449                  * in this case we need to ensure that we reap all events.
2450                  * Also let task_work, etc. to progress by releasing the mutex
2451                  */
2452                 if (need_resched()) {
2453                         mutex_unlock(&ctx->uring_lock);
2454                         cond_resched();
2455                         mutex_lock(&ctx->uring_lock);
2456                 }
2457         }
2458         mutex_unlock(&ctx->uring_lock);
2459 }
2460
2461 static int io_iopoll_check(struct io_ring_ctx *ctx, long min)
2462 {
2463         unsigned int nr_events = 0;
2464         int iters = 0, ret = 0;
2465
2466         /*
2467          * We disallow the app entering submit/complete with polling, but we
2468          * still need to lock the ring to prevent racing with polled issue
2469          * that got punted to a workqueue.
2470          */
2471         mutex_lock(&ctx->uring_lock);
2472         do {
2473                 /*
2474                  * Don't enter poll loop if we already have events pending.
2475                  * If we do, we can potentially be spinning for commands that
2476                  * already triggered a CQE (eg in error).
2477                  */
2478                 if (io_cqring_events(ctx, false))
2479                         break;
2480
2481                 /*
2482                  * If a submit got punted to a workqueue, we can have the
2483                  * application entering polling for a command before it gets
2484                  * issued. That app will hold the uring_lock for the duration
2485                  * of the poll right here, so we need to take a breather every
2486                  * now and then to ensure that the issue has a chance to add
2487                  * the poll to the issued list. Otherwise we can spin here
2488                  * forever, while the workqueue is stuck trying to acquire the
2489                  * very same mutex.
2490                  */
2491                 if (!(++iters & 7)) {
2492                         mutex_unlock(&ctx->uring_lock);
2493                         io_run_task_work();
2494                         mutex_lock(&ctx->uring_lock);
2495                 }
2496
2497                 ret = io_iopoll_getevents(ctx, &nr_events, min);
2498                 if (ret <= 0)
2499                         break;
2500                 ret = 0;
2501         } while (min && !nr_events && !need_resched());
2502
2503         mutex_unlock(&ctx->uring_lock);
2504         return ret;
2505 }
2506
2507 static void kiocb_end_write(struct io_kiocb *req)
2508 {
2509         /*
2510          * Tell lockdep we inherited freeze protection from submission
2511          * thread.
2512          */
2513         if (req->flags & REQ_F_ISREG) {
2514                 struct inode *inode = file_inode(req->file);
2515
2516                 __sb_writers_acquired(inode->i_sb, SB_FREEZE_WRITE);
2517         }
2518         file_end_write(req->file);
2519 }
2520
2521 static void io_complete_rw_common(struct kiocb *kiocb, long res,
2522                                   struct io_comp_state *cs)
2523 {
2524         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
2525         int cflags = 0;
2526
2527         if (kiocb->ki_flags & IOCB_WRITE)
2528                 kiocb_end_write(req);
2529
2530         if (res != req->result)
2531                 req_set_fail_links(req);
2532         if (req->flags & REQ_F_BUFFER_SELECTED)
2533                 cflags = io_put_rw_kbuf(req);
2534         __io_req_complete(req, res, cflags, cs);
2535 }
2536
2537 #ifdef CONFIG_BLOCK
2538 static bool io_resubmit_prep(struct io_kiocb *req, int error)
2539 {
2540         struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
2541         ssize_t ret = -ECANCELED;
2542         struct iov_iter iter;
2543         int rw;
2544
2545         if (error) {
2546                 ret = error;
2547                 goto end_req;
2548         }
2549
2550         switch (req->opcode) {
2551         case IORING_OP_READV:
2552         case IORING_OP_READ_FIXED:
2553         case IORING_OP_READ:
2554                 rw = READ;
2555                 break;
2556         case IORING_OP_WRITEV:
2557         case IORING_OP_WRITE_FIXED:
2558         case IORING_OP_WRITE:
2559                 rw = WRITE;
2560                 break;
2561         default:
2562                 printk_once(KERN_WARNING "io_uring: bad opcode in resubmit %d\n",
2563                                 req->opcode);
2564                 goto end_req;
2565         }
2566
2567         if (!req->async_data) {
2568                 ret = io_import_iovec(rw, req, &iovec, &iter, false);
2569                 if (ret < 0)
2570                         goto end_req;
2571                 ret = io_setup_async_rw(req, iovec, inline_vecs, &iter, false);
2572                 if (!ret)
2573                         return true;
2574                 kfree(iovec);
2575         } else {
2576                 return true;
2577         }
2578 end_req:
2579         req_set_fail_links(req);
2580         io_req_complete(req, ret);
2581         return false;
2582 }
2583 #endif
2584
2585 static bool io_rw_reissue(struct io_kiocb *req, long res)
2586 {
2587 #ifdef CONFIG_BLOCK
2588         umode_t mode = file_inode(req->file)->i_mode;
2589         int ret;
2590
2591         if (!S_ISBLK(mode) && !S_ISREG(mode))
2592                 return false;
2593         if ((res != -EAGAIN && res != -EOPNOTSUPP) || io_wq_current_is_worker())
2594                 return false;
2595
2596         ret = io_sq_thread_acquire_mm(req->ctx, req);
2597
2598         if (io_resubmit_prep(req, ret)) {
2599                 refcount_inc(&req->refs);
2600                 io_queue_async_work(req);
2601                 return true;
2602         }
2603
2604 #endif
2605         return false;
2606 }
2607
2608 static void __io_complete_rw(struct io_kiocb *req, long res, long res2,
2609                              struct io_comp_state *cs)
2610 {
2611         if (!io_rw_reissue(req, res))
2612                 io_complete_rw_common(&req->rw.kiocb, res, cs);
2613 }
2614
2615 static void io_complete_rw(struct kiocb *kiocb, long res, long res2)
2616 {
2617         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
2618
2619         __io_complete_rw(req, res, res2, NULL);
2620 }
2621
2622 static void io_complete_rw_iopoll(struct kiocb *kiocb, long res, long res2)
2623 {
2624         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
2625
2626         if (kiocb->ki_flags & IOCB_WRITE)
2627                 kiocb_end_write(req);
2628
2629         if (res != -EAGAIN && res != req->result)
2630                 req_set_fail_links(req);
2631
2632         WRITE_ONCE(req->result, res);
2633         /* order with io_poll_complete() checking ->result */
2634         smp_wmb();
2635         WRITE_ONCE(req->iopoll_completed, 1);
2636 }
2637
2638 /*
2639  * After the iocb has been issued, it's safe to be found on the poll list.
2640  * Adding the kiocb to the list AFTER submission ensures that we don't
2641  * find it from a io_iopoll_getevents() thread before the issuer is done
2642  * accessing the kiocb cookie.
2643  */
2644 static void io_iopoll_req_issued(struct io_kiocb *req)
2645 {
2646         struct io_ring_ctx *ctx = req->ctx;
2647
2648         /*
2649          * Track whether we have multiple files in our lists. This will impact
2650          * how we do polling eventually, not spinning if we're on potentially
2651          * different devices.
2652          */
2653         if (list_empty(&ctx->iopoll_list)) {
2654                 ctx->poll_multi_file = false;
2655         } else if (!ctx->poll_multi_file) {
2656                 struct io_kiocb *list_req;
2657
2658                 list_req = list_first_entry(&ctx->iopoll_list, struct io_kiocb,
2659                                                 inflight_entry);
2660                 if (list_req->file != req->file)
2661                         ctx->poll_multi_file = true;
2662         }
2663
2664         /*
2665          * For fast devices, IO may have already completed. If it has, add
2666          * it to the front so we find it first.
2667          */
2668         if (READ_ONCE(req->iopoll_completed))
2669                 list_add(&req->inflight_entry, &ctx->iopoll_list);
2670         else
2671                 list_add_tail(&req->inflight_entry, &ctx->iopoll_list);
2672
2673         if ((ctx->flags & IORING_SETUP_SQPOLL) &&
2674             wq_has_sleeper(&ctx->sq_data->wait))
2675                 wake_up(&ctx->sq_data->wait);
2676 }
2677
2678 static void __io_state_file_put(struct io_submit_state *state)
2679 {
2680         if (state->has_refs)
2681                 fput_many(state->file, state->has_refs);
2682         state->file = NULL;
2683 }
2684
2685 static inline void io_state_file_put(struct io_submit_state *state)
2686 {
2687         if (state->file)
2688                 __io_state_file_put(state);
2689 }
2690
2691 /*
2692  * Get as many references to a file as we have IOs left in this submission,
2693  * assuming most submissions are for one file, or at least that each file
2694  * has more than one submission.
2695  */
2696 static struct file *__io_file_get(struct io_submit_state *state, int fd)
2697 {
2698         if (!state)
2699                 return fget(fd);
2700
2701         if (state->file) {
2702                 if (state->fd == fd) {
2703                         state->has_refs--;
2704                         return state->file;
2705                 }
2706                 __io_state_file_put(state);
2707         }
2708         state->file = fget_many(fd, state->ios_left);
2709         if (!state->file)
2710                 return NULL;
2711
2712         state->fd = fd;
2713         state->has_refs = state->ios_left - 1;
2714         return state->file;
2715 }
2716
2717 static bool io_bdev_nowait(struct block_device *bdev)
2718 {
2719         return !bdev || blk_queue_nowait(bdev_get_queue(bdev));
2720 }
2721
2722 /*
2723  * If we tracked the file through the SCM inflight mechanism, we could support
2724  * any file. For now, just ensure that anything potentially problematic is done
2725  * inline.
2726  */
2727 static bool io_file_supports_async(struct file *file, int rw)
2728 {
2729         umode_t mode = file_inode(file)->i_mode;
2730
2731         if (S_ISBLK(mode)) {
2732                 if (IS_ENABLED(CONFIG_BLOCK) &&
2733                     io_bdev_nowait(I_BDEV(file->f_mapping->host)))
2734                         return true;
2735                 return false;
2736         }
2737         if (S_ISCHR(mode) || S_ISSOCK(mode))
2738                 return true;
2739         if (S_ISREG(mode)) {
2740                 if (IS_ENABLED(CONFIG_BLOCK) &&
2741                     io_bdev_nowait(file->f_inode->i_sb->s_bdev) &&
2742                     file->f_op != &io_uring_fops)
2743                         return true;
2744                 return false;
2745         }
2746
2747         /* any ->read/write should understand O_NONBLOCK */
2748         if (file->f_flags & O_NONBLOCK)
2749                 return true;
2750
2751         if (!(file->f_mode & FMODE_NOWAIT))
2752                 return false;
2753
2754         if (rw == READ)
2755                 return file->f_op->read_iter != NULL;
2756
2757         return file->f_op->write_iter != NULL;
2758 }
2759
2760 static int io_prep_rw(struct io_kiocb *req, const struct io_uring_sqe *sqe)
2761 {
2762         struct io_ring_ctx *ctx = req->ctx;
2763         struct kiocb *kiocb = &req->rw.kiocb;
2764         unsigned ioprio;
2765         int ret;
2766
2767         if (S_ISREG(file_inode(req->file)->i_mode))
2768                 req->flags |= REQ_F_ISREG;
2769
2770         kiocb->ki_pos = READ_ONCE(sqe->off);
2771         if (kiocb->ki_pos == -1 && !(req->file->f_mode & FMODE_STREAM)) {
2772                 req->flags |= REQ_F_CUR_POS;
2773                 kiocb->ki_pos = req->file->f_pos;
2774         }
2775         kiocb->ki_hint = ki_hint_validate(file_write_hint(kiocb->ki_filp));
2776         kiocb->ki_flags = iocb_flags(kiocb->ki_filp);
2777         ret = kiocb_set_rw_flags(kiocb, READ_ONCE(sqe->rw_flags));
2778         if (unlikely(ret))
2779                 return ret;
2780
2781         ioprio = READ_ONCE(sqe->ioprio);
2782         if (ioprio) {
2783                 ret = ioprio_check_cap(ioprio);
2784                 if (ret)
2785                         return ret;
2786
2787                 kiocb->ki_ioprio = ioprio;
2788         } else
2789                 kiocb->ki_ioprio = get_current_ioprio();
2790
2791         /* don't allow async punt if RWF_NOWAIT was requested */
2792         if (kiocb->ki_flags & IOCB_NOWAIT)
2793                 req->flags |= REQ_F_NOWAIT;
2794
2795         if (ctx->flags & IORING_SETUP_IOPOLL) {
2796                 if (!(kiocb->ki_flags & IOCB_DIRECT) ||
2797                     !kiocb->ki_filp->f_op->iopoll)
2798                         return -EOPNOTSUPP;
2799
2800                 kiocb->ki_flags |= IOCB_HIPRI;
2801                 kiocb->ki_complete = io_complete_rw_iopoll;
2802                 req->iopoll_completed = 0;
2803         } else {
2804                 if (kiocb->ki_flags & IOCB_HIPRI)
2805                         return -EINVAL;
2806                 kiocb->ki_complete = io_complete_rw;
2807         }
2808
2809         req->rw.addr = READ_ONCE(sqe->addr);
2810         req->rw.len = READ_ONCE(sqe->len);
2811         req->buf_index = READ_ONCE(sqe->buf_index);
2812         return 0;
2813 }
2814
2815 static inline void io_rw_done(struct kiocb *kiocb, ssize_t ret)
2816 {
2817         switch (ret) {
2818         case -EIOCBQUEUED:
2819                 break;
2820         case -ERESTARTSYS:
2821         case -ERESTARTNOINTR:
2822         case -ERESTARTNOHAND:
2823         case -ERESTART_RESTARTBLOCK:
2824                 /*
2825                  * We can't just restart the syscall, since previously
2826                  * submitted sqes may already be in progress. Just fail this
2827                  * IO with EINTR.
2828                  */
2829                 ret = -EINTR;
2830                 fallthrough;
2831         default:
2832                 kiocb->ki_complete(kiocb, ret, 0);
2833         }
2834 }
2835
2836 static void kiocb_done(struct kiocb *kiocb, ssize_t ret,
2837                        struct io_comp_state *cs)
2838 {
2839         struct io_kiocb *req = container_of(kiocb, struct io_kiocb, rw.kiocb);
2840         struct io_async_rw *io = req->async_data;
2841
2842         /* add previously done IO, if any */
2843         if (io && io->bytes_done > 0) {
2844                 if (ret < 0)
2845                         ret = io->bytes_done;
2846                 else
2847                         ret += io->bytes_done;
2848         }
2849
2850         if (req->flags & REQ_F_CUR_POS)
2851                 req->file->f_pos = kiocb->ki_pos;
2852         if (ret >= 0 && kiocb->ki_complete == io_complete_rw)
2853                 __io_complete_rw(req, ret, 0, cs);
2854         else
2855                 io_rw_done(kiocb, ret);
2856 }
2857
2858 static ssize_t io_import_fixed(struct io_kiocb *req, int rw,
2859                                struct iov_iter *iter)
2860 {
2861         struct io_ring_ctx *ctx = req->ctx;
2862         size_t len = req->rw.len;
2863         struct io_mapped_ubuf *imu;
2864         u16 index, buf_index = req->buf_index;
2865         size_t offset;
2866         u64 buf_addr;
2867
2868         if (unlikely(buf_index >= ctx->nr_user_bufs))
2869                 return -EFAULT;
2870         index = array_index_nospec(buf_index, ctx->nr_user_bufs);
2871         imu = &ctx->user_bufs[index];
2872         buf_addr = req->rw.addr;
2873
2874         /* overflow */
2875         if (buf_addr + len < buf_addr)
2876                 return -EFAULT;
2877         /* not inside the mapped region */
2878         if (buf_addr < imu->ubuf || buf_addr + len > imu->ubuf + imu->len)
2879                 return -EFAULT;
2880
2881         /*
2882          * May not be a start of buffer, set size appropriately
2883          * and advance us to the beginning.
2884          */
2885         offset = buf_addr - imu->ubuf;
2886         iov_iter_bvec(iter, rw, imu->bvec, imu->nr_bvecs, offset + len);
2887
2888         if (offset) {
2889                 /*
2890                  * Don't use iov_iter_advance() here, as it's really slow for
2891                  * using the latter parts of a big fixed buffer - it iterates
2892                  * over each segment manually. We can cheat a bit here, because
2893                  * we know that:
2894                  *
2895                  * 1) it's a BVEC iter, we set it up
2896                  * 2) all bvecs are PAGE_SIZE in size, except potentially the
2897                  *    first and last bvec
2898                  *
2899                  * So just find our index, and adjust the iterator afterwards.
2900                  * If the offset is within the first bvec (or the whole first
2901                  * bvec, just use iov_iter_advance(). This makes it easier
2902                  * since we can just skip the first segment, which may not
2903                  * be PAGE_SIZE aligned.
2904                  */
2905                 const struct bio_vec *bvec = imu->bvec;
2906
2907                 if (offset <= bvec->bv_len) {
2908                         iov_iter_advance(iter, offset);
2909                 } else {
2910                         unsigned long seg_skip;
2911
2912                         /* skip first vec */
2913                         offset -= bvec->bv_len;
2914                         seg_skip = 1 + (offset >> PAGE_SHIFT);
2915
2916                         iter->bvec = bvec + seg_skip;
2917                         iter->nr_segs -= seg_skip;
2918                         iter->count -= bvec->bv_len + offset;
2919                         iter->iov_offset = offset & ~PAGE_MASK;
2920                 }
2921         }
2922
2923         return len;
2924 }
2925
2926 static void io_ring_submit_unlock(struct io_ring_ctx *ctx, bool needs_lock)
2927 {
2928         if (needs_lock)
2929                 mutex_unlock(&ctx->uring_lock);
2930 }
2931
2932 static void io_ring_submit_lock(struct io_ring_ctx *ctx, bool needs_lock)
2933 {
2934         /*
2935          * "Normal" inline submissions always hold the uring_lock, since we
2936          * grab it from the system call. Same is true for the SQPOLL offload.
2937          * The only exception is when we've detached the request and issue it
2938          * from an async worker thread, grab the lock for that case.
2939          */
2940         if (needs_lock)
2941                 mutex_lock(&ctx->uring_lock);
2942 }
2943
2944 static struct io_buffer *io_buffer_select(struct io_kiocb *req, size_t *len,
2945                                           int bgid, struct io_buffer *kbuf,
2946                                           bool needs_lock)
2947 {
2948         struct io_buffer *head;
2949
2950         if (req->flags & REQ_F_BUFFER_SELECTED)
2951                 return kbuf;
2952
2953         io_ring_submit_lock(req->ctx, needs_lock);
2954
2955         lockdep_assert_held(&req->ctx->uring_lock);
2956
2957         head = idr_find(&req->ctx->io_buffer_idr, bgid);
2958         if (head) {
2959                 if (!list_empty(&head->list)) {
2960                         kbuf = list_last_entry(&head->list, struct io_buffer,
2961                                                         list);
2962                         list_del(&kbuf->list);
2963                 } else {
2964                         kbuf = head;
2965                         idr_remove(&req->ctx->io_buffer_idr, bgid);
2966                 }
2967                 if (*len > kbuf->len)
2968                         *len = kbuf->len;
2969         } else {
2970                 kbuf = ERR_PTR(-ENOBUFS);
2971         }
2972
2973         io_ring_submit_unlock(req->ctx, needs_lock);
2974
2975         return kbuf;
2976 }
2977
2978 static void __user *io_rw_buffer_select(struct io_kiocb *req, size_t *len,
2979                                         bool needs_lock)
2980 {
2981         struct io_buffer *kbuf;
2982         u16 bgid;
2983
2984         kbuf = (struct io_buffer *) (unsigned long) req->rw.addr;
2985         bgid = req->buf_index;
2986         kbuf = io_buffer_select(req, len, bgid, kbuf, needs_lock);
2987         if (IS_ERR(kbuf))
2988                 return kbuf;
2989         req->rw.addr = (u64) (unsigned long) kbuf;
2990         req->flags |= REQ_F_BUFFER_SELECTED;
2991         return u64_to_user_ptr(kbuf->addr);
2992 }
2993
2994 #ifdef CONFIG_COMPAT
2995 static ssize_t io_compat_import(struct io_kiocb *req, struct iovec *iov,
2996                                 bool needs_lock)
2997 {
2998         struct compat_iovec __user *uiov;
2999         compat_ssize_t clen;
3000         void __user *buf;
3001         ssize_t len;
3002
3003         uiov = u64_to_user_ptr(req->rw.addr);
3004         if (!access_ok(uiov, sizeof(*uiov)))
3005                 return -EFAULT;
3006         if (__get_user(clen, &uiov->iov_len))
3007                 return -EFAULT;
3008         if (clen < 0)
3009                 return -EINVAL;
3010
3011         len = clen;
3012         buf = io_rw_buffer_select(req, &len, needs_lock);
3013         if (IS_ERR(buf))
3014                 return PTR_ERR(buf);
3015         iov[0].iov_base = buf;
3016         iov[0].iov_len = (compat_size_t) len;
3017         return 0;
3018 }
3019 #endif
3020
3021 static ssize_t __io_iov_buffer_select(struct io_kiocb *req, struct iovec *iov,
3022                                       bool needs_lock)
3023 {
3024         struct iovec __user *uiov = u64_to_user_ptr(req->rw.addr);
3025         void __user *buf;
3026         ssize_t len;
3027
3028         if (copy_from_user(iov, uiov, sizeof(*uiov)))
3029                 return -EFAULT;
3030
3031         len = iov[0].iov_len;
3032         if (len < 0)
3033                 return -EINVAL;
3034         buf = io_rw_buffer_select(req, &len, needs_lock);
3035         if (IS_ERR(buf))
3036                 return PTR_ERR(buf);
3037         iov[0].iov_base = buf;
3038         iov[0].iov_len = len;
3039         return 0;
3040 }
3041
3042 static ssize_t io_iov_buffer_select(struct io_kiocb *req, struct iovec *iov,
3043                                     bool needs_lock)
3044 {
3045         if (req->flags & REQ_F_BUFFER_SELECTED) {
3046                 struct io_buffer *kbuf;
3047
3048                 kbuf = (struct io_buffer *) (unsigned long) req->rw.addr;
3049                 iov[0].iov_base = u64_to_user_ptr(kbuf->addr);
3050                 iov[0].iov_len = kbuf->len;
3051                 return 0;
3052         }
3053         if (!req->rw.len)
3054                 return 0;
3055         else if (req->rw.len > 1)
3056                 return -EINVAL;
3057
3058 #ifdef CONFIG_COMPAT
3059         if (req->ctx->compat)
3060                 return io_compat_import(req, iov, needs_lock);
3061 #endif
3062
3063         return __io_iov_buffer_select(req, iov, needs_lock);
3064 }
3065
3066 static ssize_t __io_import_iovec(int rw, struct io_kiocb *req,
3067                                  struct iovec **iovec, struct iov_iter *iter,
3068                                  bool needs_lock)
3069 {
3070         void __user *buf = u64_to_user_ptr(req->rw.addr);
3071         size_t sqe_len = req->rw.len;
3072         ssize_t ret;
3073         u8 opcode;
3074
3075         opcode = req->opcode;
3076         if (opcode == IORING_OP_READ_FIXED || opcode == IORING_OP_WRITE_FIXED) {
3077                 *iovec = NULL;
3078                 return io_import_fixed(req, rw, iter);
3079         }
3080
3081         /* buffer index only valid with fixed read/write, or buffer select  */
3082         if (req->buf_index && !(req->flags & REQ_F_BUFFER_SELECT))
3083                 return -EINVAL;
3084
3085         if (opcode == IORING_OP_READ || opcode == IORING_OP_WRITE) {
3086                 if (req->flags & REQ_F_BUFFER_SELECT) {
3087                         buf = io_rw_buffer_select(req, &sqe_len, needs_lock);
3088                         if (IS_ERR(buf))
3089                                 return PTR_ERR(buf);
3090                         req->rw.len = sqe_len;
3091                 }
3092
3093                 ret = import_single_range(rw, buf, sqe_len, *iovec, iter);
3094                 *iovec = NULL;
3095                 return ret < 0 ? ret : sqe_len;
3096         }
3097
3098         if (req->flags & REQ_F_BUFFER_SELECT) {
3099                 ret = io_iov_buffer_select(req, *iovec, needs_lock);
3100                 if (!ret) {
3101                         ret = (*iovec)->iov_len;
3102                         iov_iter_init(iter, rw, *iovec, 1, ret);
3103                 }
3104                 *iovec = NULL;
3105                 return ret;
3106         }
3107
3108         return __import_iovec(rw, buf, sqe_len, UIO_FASTIOV, iovec, iter,
3109                               req->ctx->compat);
3110 }
3111
3112 static ssize_t io_import_iovec(int rw, struct io_kiocb *req,
3113                                struct iovec **iovec, struct iov_iter *iter,
3114                                bool needs_lock)
3115 {
3116         struct io_async_rw *iorw = req->async_data;
3117
3118         if (!iorw)
3119                 return __io_import_iovec(rw, req, iovec, iter, needs_lock);
3120         *iovec = NULL;
3121         return iov_iter_count(&iorw->iter);
3122 }
3123
3124 static inline loff_t *io_kiocb_ppos(struct kiocb *kiocb)
3125 {
3126         return (kiocb->ki_filp->f_mode & FMODE_STREAM) ? NULL : &kiocb->ki_pos;
3127 }
3128
3129 /*
3130  * For files that don't have ->read_iter() and ->write_iter(), handle them
3131  * by looping over ->read() or ->write() manually.
3132  */
3133 static ssize_t loop_rw_iter(int rw, struct io_kiocb *req, struct iov_iter *iter)
3134 {
3135         struct kiocb *kiocb = &req->rw.kiocb;
3136         struct file *file = req->file;
3137         ssize_t ret = 0;
3138
3139         /*
3140          * Don't support polled IO through this interface, and we can't
3141          * support non-blocking either. For the latter, this just causes
3142          * the kiocb to be handled from an async context.
3143          */
3144         if (kiocb->ki_flags & IOCB_HIPRI)
3145                 return -EOPNOTSUPP;
3146         if (kiocb->ki_flags & IOCB_NOWAIT)
3147                 return -EAGAIN;
3148
3149         while (iov_iter_count(iter)) {
3150                 struct iovec iovec;
3151                 ssize_t nr;
3152
3153                 if (!iov_iter_is_bvec(iter)) {
3154                         iovec = iov_iter_iovec(iter);
3155                 } else {
3156                         iovec.iov_base = u64_to_user_ptr(req->rw.addr);
3157                         iovec.iov_len = req->rw.len;
3158                 }
3159
3160                 if (rw == READ) {
3161                         nr = file->f_op->read(file, iovec.iov_base,
3162                                               iovec.iov_len, io_kiocb_ppos(kiocb));
3163                 } else {
3164                         nr = file->f_op->write(file, iovec.iov_base,
3165                                                iovec.iov_len, io_kiocb_ppos(kiocb));
3166                 }
3167
3168                 if (nr < 0) {
3169                         if (!ret)
3170                                 ret = nr;
3171                         break;
3172                 }
3173                 ret += nr;
3174                 if (nr != iovec.iov_len)
3175                         break;
3176                 req->rw.len -= nr;
3177                 req->rw.addr += nr;
3178                 iov_iter_advance(iter, nr);
3179         }
3180
3181         return ret;
3182 }
3183
3184 static void io_req_map_rw(struct io_kiocb *req, const struct iovec *iovec,
3185                           const struct iovec *fast_iov, struct iov_iter *iter)
3186 {
3187         struct io_async_rw *rw = req->async_data;
3188
3189         memcpy(&rw->iter, iter, sizeof(*iter));
3190         rw->free_iovec = iovec;
3191         rw->bytes_done = 0;
3192         /* can only be fixed buffers, no need to do anything */
3193         if (iter->type == ITER_BVEC)
3194                 return;
3195         if (!iovec) {
3196                 unsigned iov_off = 0;
3197
3198                 rw->iter.iov = rw->fast_iov;
3199                 if (iter->iov != fast_iov) {
3200                         iov_off = iter->iov - fast_iov;
3201                         rw->iter.iov += iov_off;
3202                 }
3203                 if (rw->fast_iov != fast_iov)
3204                         memcpy(rw->fast_iov + iov_off, fast_iov + iov_off,
3205                                sizeof(struct iovec) * iter->nr_segs);
3206         } else {
3207                 req->flags |= REQ_F_NEED_CLEANUP;
3208         }
3209 }
3210
3211 static inline int __io_alloc_async_data(struct io_kiocb *req)
3212 {
3213         WARN_ON_ONCE(!io_op_defs[req->opcode].async_size);
3214         req->async_data = kmalloc(io_op_defs[req->opcode].async_size, GFP_KERNEL);
3215         return req->async_data == NULL;
3216 }
3217
3218 static int io_alloc_async_data(struct io_kiocb *req)
3219 {
3220         if (!io_op_defs[req->opcode].needs_async_data)
3221                 return 0;
3222
3223         return  __io_alloc_async_data(req);
3224 }
3225
3226 static int io_setup_async_rw(struct io_kiocb *req, const struct iovec *iovec,
3227                              const struct iovec *fast_iov,
3228                              struct iov_iter *iter, bool force)
3229 {
3230         if (!force && !io_op_defs[req->opcode].needs_async_data)
3231                 return 0;
3232         if (!req->async_data) {
3233                 if (__io_alloc_async_data(req))
3234                         return -ENOMEM;
3235
3236                 io_req_map_rw(req, iovec, fast_iov, iter);
3237         }
3238         return 0;
3239 }
3240
3241 static inline int io_rw_prep_async(struct io_kiocb *req, int rw)
3242 {
3243         struct io_async_rw *iorw = req->async_data;
3244         struct iovec *iov = iorw->fast_iov;
3245         ssize_t ret;
3246
3247         ret = __io_import_iovec(rw, req, &iov, &iorw->iter, false);
3248         if (unlikely(ret < 0))
3249                 return ret;
3250
3251         iorw->bytes_done = 0;
3252         iorw->free_iovec = iov;
3253         if (iov)
3254                 req->flags |= REQ_F_NEED_CLEANUP;
3255         return 0;
3256 }
3257
3258 static int io_read_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3259 {
3260         ssize_t ret;
3261
3262         ret = io_prep_rw(req, sqe);
3263         if (ret)
3264                 return ret;
3265
3266         if (unlikely(!(req->file->f_mode & FMODE_READ)))
3267                 return -EBADF;
3268
3269         /* either don't need iovec imported or already have it */
3270         if (!req->async_data)
3271                 return 0;
3272         return io_rw_prep_async(req, READ);
3273 }
3274
3275 /*
3276  * This is our waitqueue callback handler, registered through lock_page_async()
3277  * when we initially tried to do the IO with the iocb armed our waitqueue.
3278  * This gets called when the page is unlocked, and we generally expect that to
3279  * happen when the page IO is completed and the page is now uptodate. This will
3280  * queue a task_work based retry of the operation, attempting to copy the data
3281  * again. If the latter fails because the page was NOT uptodate, then we will
3282  * do a thread based blocking retry of the operation. That's the unexpected
3283  * slow path.
3284  */
3285 static int io_async_buf_func(struct wait_queue_entry *wait, unsigned mode,
3286                              int sync, void *arg)
3287 {
3288         struct wait_page_queue *wpq;
3289         struct io_kiocb *req = wait->private;
3290         struct wait_page_key *key = arg;
3291         int ret;
3292
3293         wpq = container_of(wait, struct wait_page_queue, wait);
3294
3295         if (!wake_page_match(wpq, key))
3296                 return 0;
3297
3298         req->rw.kiocb.ki_flags &= ~IOCB_WAITQ;
3299         list_del_init(&wait->entry);
3300
3301         init_task_work(&req->task_work, io_req_task_submit);
3302         percpu_ref_get(&req->ctx->refs);
3303
3304         /* submit ref gets dropped, acquire a new one */
3305         refcount_inc(&req->refs);
3306         ret = io_req_task_work_add(req, true);
3307         if (unlikely(ret)) {
3308                 struct task_struct *tsk;
3309
3310                 /* queue just for cancelation */
3311                 init_task_work(&req->task_work, io_req_task_cancel);
3312                 tsk = io_wq_get_task(req->ctx->io_wq);
3313                 task_work_add(tsk, &req->task_work, TWA_NONE);
3314                 wake_up_process(tsk);
3315         }
3316         return 1;
3317 }
3318
3319 /*
3320  * This controls whether a given IO request should be armed for async page
3321  * based retry. If we return false here, the request is handed to the async
3322  * worker threads for retry. If we're doing buffered reads on a regular file,
3323  * we prepare a private wait_page_queue entry and retry the operation. This
3324  * will either succeed because the page is now uptodate and unlocked, or it
3325  * will register a callback when the page is unlocked at IO completion. Through
3326  * that callback, io_uring uses task_work to setup a retry of the operation.
3327  * That retry will attempt the buffered read again. The retry will generally
3328  * succeed, or in rare cases where it fails, we then fall back to using the
3329  * async worker threads for a blocking retry.
3330  */
3331 static bool io_rw_should_retry(struct io_kiocb *req)
3332 {
3333         struct io_async_rw *rw = req->async_data;
3334         struct wait_page_queue *wait = &rw->wpq;
3335         struct kiocb *kiocb = &req->rw.kiocb;
3336
3337         /* never retry for NOWAIT, we just complete with -EAGAIN */
3338         if (req->flags & REQ_F_NOWAIT)
3339                 return false;
3340
3341         /* Only for buffered IO */
3342         if (kiocb->ki_flags & (IOCB_DIRECT | IOCB_HIPRI))
3343                 return false;
3344
3345         /*
3346          * just use poll if we can, and don't attempt if the fs doesn't
3347          * support callback based unlocks
3348          */
3349         if (file_can_poll(req->file) || !(req->file->f_mode & FMODE_BUF_RASYNC))
3350                 return false;
3351
3352         wait->wait.func = io_async_buf_func;
3353         wait->wait.private = req;
3354         wait->wait.flags = 0;
3355         INIT_LIST_HEAD(&wait->wait.entry);
3356         kiocb->ki_flags |= IOCB_WAITQ;
3357         kiocb->ki_flags &= ~IOCB_NOWAIT;
3358         kiocb->ki_waitq = wait;
3359         return true;
3360 }
3361
3362 static int io_iter_do_read(struct io_kiocb *req, struct iov_iter *iter)
3363 {
3364         if (req->file->f_op->read_iter)
3365                 return call_read_iter(req->file, &req->rw.kiocb, iter);
3366         else if (req->file->f_op->read)
3367                 return loop_rw_iter(READ, req, iter);
3368         else
3369                 return -EINVAL;
3370 }
3371
3372 static int io_read(struct io_kiocb *req, bool force_nonblock,
3373                    struct io_comp_state *cs)
3374 {
3375         struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
3376         struct kiocb *kiocb = &req->rw.kiocb;
3377         struct iov_iter __iter, *iter = &__iter;
3378         struct io_async_rw *rw = req->async_data;
3379         ssize_t io_size, ret, ret2;
3380         size_t iov_count;
3381         bool no_async;
3382
3383         if (rw)
3384                 iter = &rw->iter;
3385
3386         ret = io_import_iovec(READ, req, &iovec, iter, !force_nonblock);
3387         if (ret < 0)
3388                 return ret;
3389         iov_count = iov_iter_count(iter);
3390         io_size = ret;
3391         req->result = io_size;
3392         ret = 0;
3393
3394         /* Ensure we clear previously set non-block flag */
3395         if (!force_nonblock)
3396                 kiocb->ki_flags &= ~IOCB_NOWAIT;
3397         else
3398                 kiocb->ki_flags |= IOCB_NOWAIT;
3399
3400
3401         /* If the file doesn't support async, just async punt */
3402         no_async = force_nonblock && !io_file_supports_async(req->file, READ);
3403         if (no_async)
3404                 goto copy_iov;
3405
3406         ret = rw_verify_area(READ, req->file, io_kiocb_ppos(kiocb), iov_count);
3407         if (unlikely(ret))
3408                 goto out_free;
3409
3410         ret = io_iter_do_read(req, iter);
3411
3412         if (!ret) {
3413                 goto done;
3414         } else if (ret == -EIOCBQUEUED) {
3415                 ret = 0;
3416                 goto out_free;
3417         } else if (ret == -EAGAIN) {
3418                 /* IOPOLL retry should happen for io-wq threads */
3419                 if (!force_nonblock && !(req->ctx->flags & IORING_SETUP_IOPOLL))
3420                         goto done;
3421                 /* no retry on NONBLOCK marked file */
3422                 if (req->file->f_flags & O_NONBLOCK)
3423                         goto done;
3424                 /* some cases will consume bytes even on error returns */
3425                 iov_iter_revert(iter, iov_count - iov_iter_count(iter));
3426                 ret = 0;
3427                 goto copy_iov;
3428         } else if (ret < 0) {
3429                 /* make sure -ERESTARTSYS -> -EINTR is done */
3430                 goto done;
3431         }
3432
3433         /* read it all, or we did blocking attempt. no retry. */
3434         if (!iov_iter_count(iter) || !force_nonblock ||
3435             (req->file->f_flags & O_NONBLOCK))
3436                 goto done;
3437
3438         io_size -= ret;
3439 copy_iov:
3440         ret2 = io_setup_async_rw(req, iovec, inline_vecs, iter, true);
3441         if (ret2) {
3442                 ret = ret2;
3443                 goto out_free;
3444         }
3445         if (no_async)
3446                 return -EAGAIN;
3447         rw = req->async_data;
3448         /* it's copied and will be cleaned with ->io */
3449         iovec = NULL;
3450         /* now use our persistent iterator, if we aren't already */
3451         iter = &rw->iter;
3452 retry:
3453         rw->bytes_done += ret;
3454         /* if we can retry, do so with the callbacks armed */
3455         if (!io_rw_should_retry(req)) {
3456                 kiocb->ki_flags &= ~IOCB_WAITQ;
3457                 return -EAGAIN;
3458         }
3459
3460         /*
3461          * Now retry read with the IOCB_WAITQ parts set in the iocb. If we
3462          * get -EIOCBQUEUED, then we'll get a notification when the desired
3463          * page gets unlocked. We can also get a partial read here, and if we
3464          * do, then just retry at the new offset.
3465          */
3466         ret = io_iter_do_read(req, iter);
3467         if (ret == -EIOCBQUEUED) {
3468                 ret = 0;
3469                 goto out_free;
3470         } else if (ret > 0 && ret < io_size) {
3471                 /* we got some bytes, but not all. retry. */
3472                 goto retry;
3473         }
3474 done:
3475         kiocb_done(kiocb, ret, cs);
3476         ret = 0;
3477 out_free:
3478         /* it's reportedly faster than delegating the null check to kfree() */
3479         if (iovec)
3480                 kfree(iovec);
3481         return ret;
3482 }
3483
3484 static int io_write_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3485 {
3486         ssize_t ret;
3487
3488         ret = io_prep_rw(req, sqe);
3489         if (ret)
3490                 return ret;
3491
3492         if (unlikely(!(req->file->f_mode & FMODE_WRITE)))
3493                 return -EBADF;
3494
3495         /* either don't need iovec imported or already have it */
3496         if (!req->async_data)
3497                 return 0;
3498         return io_rw_prep_async(req, WRITE);
3499 }
3500
3501 static int io_write(struct io_kiocb *req, bool force_nonblock,
3502                     struct io_comp_state *cs)
3503 {
3504         struct iovec inline_vecs[UIO_FASTIOV], *iovec = inline_vecs;
3505         struct kiocb *kiocb = &req->rw.kiocb;
3506         struct iov_iter __iter, *iter = &__iter;
3507         struct io_async_rw *rw = req->async_data;
3508         size_t iov_count;
3509         ssize_t ret, ret2, io_size;
3510
3511         if (rw)
3512                 iter = &rw->iter;
3513
3514         ret = io_import_iovec(WRITE, req, &iovec, iter, !force_nonblock);
3515         if (ret < 0)
3516                 return ret;
3517         iov_count = iov_iter_count(iter);
3518         io_size = ret;
3519         req->result = io_size;
3520
3521         /* Ensure we clear previously set non-block flag */
3522         if (!force_nonblock)
3523                 kiocb->ki_flags &= ~IOCB_NOWAIT;
3524         else
3525                 kiocb->ki_flags |= IOCB_NOWAIT;
3526
3527         /* If the file doesn't support async, just async punt */
3528         if (force_nonblock && !io_file_supports_async(req->file, WRITE))
3529                 goto copy_iov;
3530
3531         /* file path doesn't support NOWAIT for non-direct_IO */
3532         if (force_nonblock && !(kiocb->ki_flags & IOCB_DIRECT) &&
3533             (req->flags & REQ_F_ISREG))
3534                 goto copy_iov;
3535
3536         ret = rw_verify_area(WRITE, req->file, io_kiocb_ppos(kiocb), iov_count);
3537         if (unlikely(ret))
3538                 goto out_free;
3539
3540         /*
3541          * Open-code file_start_write here to grab freeze protection,
3542          * which will be released by another thread in
3543          * io_complete_rw().  Fool lockdep by telling it the lock got
3544          * released so that it doesn't complain about the held lock when
3545          * we return to userspace.
3546          */
3547         if (req->flags & REQ_F_ISREG) {
3548                 sb_start_write(file_inode(req->file)->i_sb);
3549                 __sb_writers_release(file_inode(req->file)->i_sb,
3550                                         SB_FREEZE_WRITE);
3551         }
3552         kiocb->ki_flags |= IOCB_WRITE;
3553
3554         if (req->file->f_op->write_iter)
3555                 ret2 = call_write_iter(req->file, kiocb, iter);
3556         else if (req->file->f_op->write)
3557                 ret2 = loop_rw_iter(WRITE, req, iter);
3558         else
3559                 ret2 = -EINVAL;
3560
3561         /*
3562          * Raw bdev writes will return -EOPNOTSUPP for IOCB_NOWAIT. Just
3563          * retry them without IOCB_NOWAIT.
3564          */
3565         if (ret2 == -EOPNOTSUPP && (kiocb->ki_flags & IOCB_NOWAIT))
3566                 ret2 = -EAGAIN;
3567         /* no retry on NONBLOCK marked file */
3568         if (ret2 == -EAGAIN && (req->file->f_flags & O_NONBLOCK))
3569                 goto done;
3570         if (!force_nonblock || ret2 != -EAGAIN) {
3571                 /* IOPOLL retry should happen for io-wq threads */
3572                 if ((req->ctx->flags & IORING_SETUP_IOPOLL) && ret2 == -EAGAIN)
3573                         goto copy_iov;
3574 done:
3575                 kiocb_done(kiocb, ret2, cs);
3576         } else {
3577 copy_iov:
3578                 /* some cases will consume bytes even on error returns */
3579                 iov_iter_revert(iter, iov_count - iov_iter_count(iter));
3580                 ret = io_setup_async_rw(req, iovec, inline_vecs, iter, false);
3581                 if (!ret)
3582                         return -EAGAIN;
3583         }
3584 out_free:
3585         /* it's reportedly faster than delegating the null check to kfree() */
3586         if (iovec)
3587                 kfree(iovec);
3588         return ret;
3589 }
3590
3591 static int __io_splice_prep(struct io_kiocb *req,
3592                             const struct io_uring_sqe *sqe)
3593 {
3594         struct io_splice* sp = &req->splice;
3595         unsigned int valid_flags = SPLICE_F_FD_IN_FIXED | SPLICE_F_ALL;
3596
3597         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
3598                 return -EINVAL;
3599
3600         sp->file_in = NULL;
3601         sp->len = READ_ONCE(sqe->len);
3602         sp->flags = READ_ONCE(sqe->splice_flags);
3603
3604         if (unlikely(sp->flags & ~valid_flags))
3605                 return -EINVAL;
3606
3607         sp->file_in = io_file_get(NULL, req, READ_ONCE(sqe->splice_fd_in),
3608                                   (sp->flags & SPLICE_F_FD_IN_FIXED));
3609         if (!sp->file_in)
3610                 return -EBADF;
3611         req->flags |= REQ_F_NEED_CLEANUP;
3612
3613         if (!S_ISREG(file_inode(sp->file_in)->i_mode)) {
3614                 /*
3615                  * Splice operation will be punted aync, and here need to
3616                  * modify io_wq_work.flags, so initialize io_wq_work firstly.
3617                  */
3618                 io_req_init_async(req);
3619                 req->work.flags |= IO_WQ_WORK_UNBOUND;
3620         }
3621
3622         return 0;
3623 }
3624
3625 static int io_tee_prep(struct io_kiocb *req,
3626                        const struct io_uring_sqe *sqe)
3627 {
3628         if (READ_ONCE(sqe->splice_off_in) || READ_ONCE(sqe->off))
3629                 return -EINVAL;
3630         return __io_splice_prep(req, sqe);
3631 }
3632
3633 static int io_tee(struct io_kiocb *req, bool force_nonblock)
3634 {
3635         struct io_splice *sp = &req->splice;
3636         struct file *in = sp->file_in;
3637         struct file *out = sp->file_out;
3638         unsigned int flags = sp->flags & ~SPLICE_F_FD_IN_FIXED;
3639         long ret = 0;
3640
3641         if (force_nonblock)
3642                 return -EAGAIN;
3643         if (sp->len)
3644                 ret = do_tee(in, out, sp->len, flags);
3645
3646         io_put_file(req, in, (sp->flags & SPLICE_F_FD_IN_FIXED));
3647         req->flags &= ~REQ_F_NEED_CLEANUP;
3648
3649         if (ret != sp->len)
3650                 req_set_fail_links(req);
3651         io_req_complete(req, ret);
3652         return 0;
3653 }
3654
3655 static int io_splice_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3656 {
3657         struct io_splice* sp = &req->splice;
3658
3659         sp->off_in = READ_ONCE(sqe->splice_off_in);
3660         sp->off_out = READ_ONCE(sqe->off);
3661         return __io_splice_prep(req, sqe);
3662 }
3663
3664 static int io_splice(struct io_kiocb *req, bool force_nonblock)
3665 {
3666         struct io_splice *sp = &req->splice;
3667         struct file *in = sp->file_in;
3668         struct file *out = sp->file_out;
3669         unsigned int flags = sp->flags & ~SPLICE_F_FD_IN_FIXED;
3670         loff_t *poff_in, *poff_out;
3671         long ret = 0;
3672
3673         if (force_nonblock)
3674                 return -EAGAIN;
3675
3676         poff_in = (sp->off_in == -1) ? NULL : &sp->off_in;
3677         poff_out = (sp->off_out == -1) ? NULL : &sp->off_out;
3678
3679         if (sp->len)
3680                 ret = do_splice(in, poff_in, out, poff_out, sp->len, flags);
3681
3682         io_put_file(req, in, (sp->flags & SPLICE_F_FD_IN_FIXED));
3683         req->flags &= ~REQ_F_NEED_CLEANUP;
3684
3685         if (ret != sp->len)
3686                 req_set_fail_links(req);
3687         io_req_complete(req, ret);
3688         return 0;
3689 }
3690
3691 /*
3692  * IORING_OP_NOP just posts a completion event, nothing else.
3693  */
3694 static int io_nop(struct io_kiocb *req, struct io_comp_state *cs)
3695 {
3696         struct io_ring_ctx *ctx = req->ctx;
3697
3698         if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
3699                 return -EINVAL;
3700
3701         __io_req_complete(req, 0, 0, cs);
3702         return 0;
3703 }
3704
3705 static int io_prep_fsync(struct io_kiocb *req, const struct io_uring_sqe *sqe)
3706 {
3707         struct io_ring_ctx *ctx = req->ctx;
3708
3709         if (!req->file)
3710                 return -EBADF;
3711
3712         if (unlikely(ctx->flags & IORING_SETUP_IOPOLL))
3713                 return -EINVAL;
3714         if (unlikely(sqe->addr || sqe->ioprio || sqe->buf_index))
3715                 return -EINVAL;
3716
3717         req->sync.flags = READ_ONCE(sqe->fsync_flags);
3718         if (unlikely(req->sync.flags & ~IORING_FSYNC_DATASYNC))
3719                 return -EINVAL;
3720
3721         req->sync.off = READ_ONCE(sqe->off);
3722         req->sync.len = READ_ONCE(sqe->len);
3723         return 0;
3724 }
3725
3726 static int io_fsync(struct io_kiocb *req, bool force_nonblock)
3727 {
3728         loff_t end = req->sync.off + req->sync.len;
3729         int ret;
3730
3731         /* fsync always requires a blocking context */
3732         if (force_nonblock)
3733                 return -EAGAIN;
3734
3735         ret = vfs_fsync_range(req->file, req->sync.off,
3736                                 end > 0 ? end : LLONG_MAX,
3737                                 req->sync.flags & IORING_FSYNC_DATASYNC);
3738         if (ret < 0)
3739                 req_set_fail_links(req);
3740         io_req_complete(req, ret);
3741         return 0;
3742 }
3743
3744 static int io_fallocate_prep(struct io_kiocb *req,
3745                              const struct io_uring_sqe *sqe)
3746 {
3747         if (sqe->ioprio || sqe->buf_index || sqe->rw_flags)
3748                 return -EINVAL;
3749         if (unlikely(req->ctx->flags & IORING_SETUP_IOPOLL))
3750                 return -EINVAL;
3751
3752         req->sync.off = READ_ONCE(sqe->off);
3753         req->sync.len = READ_ONCE(sqe->addr);
3754         req->sync.mode = READ_ONCE(sqe->len);
3755         return 0;
3756 }
3757
3758 static int io_fallocate(struct io_kiocb *req, bool force_nonblock)
3759 {
3760         int ret;
3761
3762         /* fallocate always requiring blocking context */
3763         if (force_nonblock)
3764