io_uring/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_cqe (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 <net/compat.h>
  47 #include <linux/refcount.h>
  48 #include <linux/uio.h>
  49 #include <linux/bits.h>
  50
  51 #include <linux/sched/signal.h>
  52 #include <linux/fs.h>
  53 #include <linux/file.h>
  54 #include <linux/mm.h>
  55 #include <linux/mman.h>
  56 #include <linux/percpu.h>
  57 #include <linux/slab.h>
  58 #include <linux/bvec.h>
  59 #include <linux/net.h>
  60 #include <net/sock.h>
  61 #include <linux/anon_inodes.h>
  62 #include <linux/sched/mm.h>
  63 #include <linux/uaccess.h>
  64 #include <linux/nospec.h>
  65 #include <linux/fsnotify.h>
  66 #include <linux/fadvise.h>
  67 #include <linux/task_work.h>
  68 #include <linux/io_uring.h>
  69 #include <linux/io_uring/cmd.h>
  70 #include <linux/audit.h>
  71 #include <linux/security.h>
  72 #include <linux/jump_label.h>
  73 #include <asm/shmparam.h>
  74
  75 #define CREATE_TRACE_POINTS
  76 #include <trace/events/io_uring.h>
  77
  78 #include <uapi/linux/io_uring.h>
  79
  80 #include "io-wq.h"
  81
  82 #include "io_uring.h"
  83 #include "opdef.h"
  84 #include "refs.h"
  85 #include "tctx.h"
  86 #include "register.h"
  87 #include "sqpoll.h"
  88 #include "fdinfo.h"
  89 #include "kbuf.h"
  90 #include "rsrc.h"
  91 #include "cancel.h"
  92 #include "net.h"
  93 #include "notif.h"
  94 #include "waitid.h"
  95 #include "futex.h"
  96 #include "napi.h"
  97 #include "uring_cmd.h"
  98 #include "msg_ring.h"
  99 #include "memmap.h"
 100 #include "zcrx.h"
 101
 102 #include "timeout.h"
 103 #include "poll.h"
 104 #include "rw.h"
 105 #include "alloc_cache.h"
 106 #include "eventfd.h"
 107
 108 #define SQE_COMMON_FLAGS (IOSQE_FIXED_FILE | IOSQE_IO_LINK | \
 109                           IOSQE_IO_HARDLINK | IOSQE_ASYNC)
 110
 111 #define SQE_VALID_FLAGS (SQE_COMMON_FLAGS | IOSQE_BUFFER_SELECT | \
 112                         IOSQE_IO_DRAIN | IOSQE_CQE_SKIP_SUCCESS)
 113
 114 #define IO_REQ_LINK_FLAGS (REQ_F_LINK | REQ_F_HARDLINK)
 115
 116 #define IO_REQ_CLEAN_FLAGS (REQ_F_BUFFER_SELECTED | REQ_F_NEED_CLEANUP | \
 117                                 REQ_F_POLLED | REQ_F_INFLIGHT | REQ_F_CREDS | \
 118                                 REQ_F_ASYNC_DATA)
 119
 120 #define IO_REQ_CLEAN_SLOW_FLAGS (REQ_F_REFCOUNT | IO_REQ_LINK_FLAGS | \
 121                                  REQ_F_REISSUE | IO_REQ_CLEAN_FLAGS)
 122
 123 #define IO_TCTX_REFS_CACHE_NR   (1U << 10)
 124
 125 #define IO_COMPL_BATCH                  32
 126 #define IO_REQ_ALLOC_BATCH              8
 127 #define IO_LOCAL_TW_DEFAULT_MAX         20
 128
 129 struct io_defer_entry {
 130         struct list_head        list;
 131         struct io_kiocb         *req;
 132 };
 133
 134 /* requests with any of those set should undergo io_disarm_next() */
 135 #define IO_DISARM_MASK (REQ_F_ARM_LTIMEOUT | REQ_F_LINK_TIMEOUT | REQ_F_FAIL)
 136
 137 /*
 138  * No waiters. It's larger than any valid value of the tw counter
 139  * so that tests against ->cq_wait_nr would fail and skip wake_up().
 140  */
 141 #define IO_CQ_WAKE_INIT         (-1U)
 142 /* Forced wake up if there is a waiter regardless of ->cq_wait_nr */
 143 #define IO_CQ_WAKE_FORCE        (IO_CQ_WAKE_INIT >> 1)
 144
 145 static bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
 146                                          struct io_uring_task *tctx,
 147                                          bool cancel_all,
 148                                          bool is_sqpoll_thread);
 149
 150 static void io_queue_sqe(struct io_kiocb *req);
 151 static void __io_req_caches_free(struct io_ring_ctx *ctx);
 152
 153 static __read_mostly DEFINE_STATIC_KEY_FALSE(io_key_has_sqarray);
 154
 155 struct kmem_cache *req_cachep;
 156 static struct workqueue_struct *iou_wq __ro_after_init;
 157
 158 static int __read_mostly sysctl_io_uring_disabled;
 159 static int __read_mostly sysctl_io_uring_group = -1;
 160
 161 #ifdef CONFIG_SYSCTL
 162 static const struct ctl_table kernel_io_uring_disabled_table[] = {
 163         {
 164                 .procname       = "io_uring_disabled",
 165                 .data           = &sysctl_io_uring_disabled,
 166                 .maxlen         = sizeof(sysctl_io_uring_disabled),
 167                 .mode           = 0644,
 168                 .proc_handler   = proc_dointvec_minmax,
 169                 .extra1         = SYSCTL_ZERO,
 170                 .extra2         = SYSCTL_TWO,
 171         },
 172         {
 173                 .procname       = "io_uring_group",
 174                 .data           = &sysctl_io_uring_group,
 175                 .maxlen         = sizeof(gid_t),
 176                 .mode           = 0644,
 177                 .proc_handler   = proc_dointvec,
 178         },
 179 };
 180 #endif
 181
 182 static inline unsigned int __io_cqring_events(struct io_ring_ctx *ctx)
 183 {
 184         return ctx->cached_cq_tail - READ_ONCE(ctx->rings->cq.head);
 185 }
 186
 187 static inline unsigned int __io_cqring_events_user(struct io_ring_ctx *ctx)
 188 {
 189         return READ_ONCE(ctx->rings->cq.tail) - READ_ONCE(ctx->rings->cq.head);
 190 }
 191
 192 static bool io_match_linked(struct io_kiocb *head)
 193 {
 194         struct io_kiocb *req;
 195
 196         io_for_each_link(req, head) {
 197                 if (req->flags & REQ_F_INFLIGHT)
 198                         return true;
 199         }
 200         return false;
 201 }
 202
 203 /*
 204  * As io_match_task() but protected against racing with linked timeouts.
 205  * User must not hold timeout_lock.
 206  */
 207 bool io_match_task_safe(struct io_kiocb *head, struct io_uring_task *tctx,
 208                         bool cancel_all)
 209 {
 210         bool matched;
 211
 212         if (tctx && head->tctx != tctx)
 213                 return false;
 214         if (cancel_all)
 215                 return true;
 216
 217         if (head->flags & REQ_F_LINK_TIMEOUT) {
 218                 struct io_ring_ctx *ctx = head->ctx;
 219
 220                 /* protect against races with linked timeouts */
 221                 raw_spin_lock_irq(&ctx->timeout_lock);
 222                 matched = io_match_linked(head);
 223                 raw_spin_unlock_irq(&ctx->timeout_lock);
 224         } else {
 225                 matched = io_match_linked(head);
 226         }
 227         return matched;
 228 }
 229
 230 static inline void req_fail_link_node(struct io_kiocb *req, int res)
 231 {
 232         req_set_fail(req);
 233         io_req_set_res(req, res, 0);
 234 }
 235
 236 static inline void io_req_add_to_cache(struct io_kiocb *req, struct io_ring_ctx *ctx)
 237 {
 238         wq_stack_add_head(&req->comp_list, &ctx->submit_state.free_list);
 239 }
 240
 241 static __cold void io_ring_ctx_ref_free(struct percpu_ref *ref)
 242 {
 243         struct io_ring_ctx *ctx = container_of(ref, struct io_ring_ctx, refs);
 244
 245         complete(&ctx->ref_comp);
 246 }
 247
 248 static __cold void io_fallback_req_func(struct work_struct *work)
 249 {
 250         struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx,
 251                                                 fallback_work.work);
 252         struct llist_node *node = llist_del_all(&ctx->fallback_llist);
 253         struct io_kiocb *req, *tmp;
 254         struct io_tw_state ts = {};
 255
 256         percpu_ref_get(&ctx->refs);
 257         mutex_lock(&ctx->uring_lock);
 258         llist_for_each_entry_safe(req, tmp, node, io_task_work.node)
 259                 req->io_task_work.func(req, ts);
 260         io_submit_flush_completions(ctx);
 261         mutex_unlock(&ctx->uring_lock);
 262         percpu_ref_put(&ctx->refs);
 263 }
 264
 265 static int io_alloc_hash_table(struct io_hash_table *table, unsigned bits)
 266 {
 267         unsigned int hash_buckets;
 268         int i;
 269
 270         do {
 271                 hash_buckets = 1U << bits;
 272                 table->hbs = kvmalloc_array(hash_buckets, sizeof(table->hbs[0]),
 273                                                 GFP_KERNEL_ACCOUNT);
 274                 if (table->hbs)
 275                         break;
 276                 if (bits == 1)
 277                         return -ENOMEM;
 278                 bits--;
 279         } while (1);
 280
 281         table->hash_bits = bits;
 282         for (i = 0; i < hash_buckets; i++)
 283                 INIT_HLIST_HEAD(&table->hbs[i].list);
 284         return 0;
 285 }
 286
 287 static void io_free_alloc_caches(struct io_ring_ctx *ctx)
 288 {
 289         io_alloc_cache_free(&ctx->apoll_cache, kfree);
 290         io_alloc_cache_free(&ctx->netmsg_cache, io_netmsg_cache_free);
 291         io_alloc_cache_free(&ctx->rw_cache, io_rw_cache_free);
 292         io_alloc_cache_free(&ctx->cmd_cache, io_cmd_cache_free);
 293         io_alloc_cache_free(&ctx->msg_cache, kfree);
 294         io_futex_cache_free(ctx);
 295         io_rsrc_cache_free(ctx);
 296 }
 297
 298 static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
 299 {
 300         struct io_ring_ctx *ctx;
 301         int hash_bits;
 302         bool ret;
 303
 304         ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
 305         if (!ctx)
 306                 return NULL;
 307
 308         xa_init(&ctx->io_bl_xa);
 309
 310         /*
 311          * Use 5 bits less than the max cq entries, that should give us around
 312          * 32 entries per hash list if totally full and uniformly spread, but
 313          * don't keep too many buckets to not overconsume memory.
 314          */
 315         hash_bits = ilog2(p->cq_entries) - 5;
 316         hash_bits = clamp(hash_bits, 1, 8);
 317         if (io_alloc_hash_table(&ctx->cancel_table, hash_bits))
 318                 goto err;
 319         if (percpu_ref_init(&ctx->refs, io_ring_ctx_ref_free,
 320                             0, GFP_KERNEL))
 321                 goto err;
 322
 323         ctx->flags = p->flags;
 324         ctx->hybrid_poll_time = LLONG_MAX;
 325         atomic_set(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT);
 326         init_waitqueue_head(&ctx->sqo_sq_wait);
 327         INIT_LIST_HEAD(&ctx->sqd_list);
 328         INIT_LIST_HEAD(&ctx->cq_overflow_list);
 329         ret = io_alloc_cache_init(&ctx->apoll_cache, IO_POLL_ALLOC_CACHE_MAX,
 330                             sizeof(struct async_poll), 0);
 331         ret |= io_alloc_cache_init(&ctx->netmsg_cache, IO_ALLOC_CACHE_MAX,
 332                             sizeof(struct io_async_msghdr),
 333                             offsetof(struct io_async_msghdr, clear));
 334         ret |= io_alloc_cache_init(&ctx->rw_cache, IO_ALLOC_CACHE_MAX,
 335                             sizeof(struct io_async_rw),
 336                             offsetof(struct io_async_rw, clear));
 337         ret |= io_alloc_cache_init(&ctx->cmd_cache, IO_ALLOC_CACHE_MAX,
 338                             sizeof(struct io_async_cmd),
 339                             sizeof(struct io_async_cmd));
 340         spin_lock_init(&ctx->msg_lock);
 341         ret |= io_alloc_cache_init(&ctx->msg_cache, IO_ALLOC_CACHE_MAX,
 342                             sizeof(struct io_kiocb), 0);
 343         ret |= io_futex_cache_init(ctx);
 344         ret |= io_rsrc_cache_init(ctx);
 345         if (ret)
 346                 goto free_ref;
 347         init_completion(&ctx->ref_comp);
 348         xa_init_flags(&ctx->personalities, XA_FLAGS_ALLOC1);
 349         mutex_init(&ctx->uring_lock);
 350         init_waitqueue_head(&ctx->cq_wait);
 351         init_waitqueue_head(&ctx->poll_wq);
 352         spin_lock_init(&ctx->completion_lock);
 353         raw_spin_lock_init(&ctx->timeout_lock);
 354         INIT_WQ_LIST(&ctx->iopoll_list);
 355         INIT_LIST_HEAD(&ctx->defer_list);
 356         INIT_LIST_HEAD(&ctx->timeout_list);
 357         INIT_LIST_HEAD(&ctx->ltimeout_list);
 358         init_llist_head(&ctx->work_llist);
 359         INIT_LIST_HEAD(&ctx->tctx_list);
 360         ctx->submit_state.free_list.next = NULL;
 361         INIT_HLIST_HEAD(&ctx->waitid_list);
 362         xa_init_flags(&ctx->zcrx_ctxs, XA_FLAGS_ALLOC);
 363 #ifdef CONFIG_FUTEX
 364         INIT_HLIST_HEAD(&ctx->futex_list);
 365 #endif
 366         INIT_DELAYED_WORK(&ctx->fallback_work, io_fallback_req_func);
 367         INIT_WQ_LIST(&ctx->submit_state.compl_reqs);
 368         INIT_HLIST_HEAD(&ctx->cancelable_uring_cmd);
 369         io_napi_init(ctx);
 370         mutex_init(&ctx->mmap_lock);
 371
 372         return ctx;
 373
 374 free_ref:
 375         percpu_ref_exit(&ctx->refs);
 376 err:
 377         io_free_alloc_caches(ctx);
 378         kvfree(ctx->cancel_table.hbs);
 379         xa_destroy(&ctx->io_bl_xa);
 380         kfree(ctx);
 381         return NULL;
 382 }
 383
 384 static void io_clean_op(struct io_kiocb *req)
 385 {
 386         if (unlikely(req->flags & REQ_F_BUFFER_SELECTED))
 387                 io_kbuf_drop_legacy(req);
 388
 389         if (req->flags & REQ_F_NEED_CLEANUP) {
 390                 const struct io_cold_def *def = &io_cold_defs[req->opcode];
 391
 392                 if (def->cleanup)
 393                         def->cleanup(req);
 394         }
 395         if ((req->flags & REQ_F_POLLED) && req->apoll) {
 396                 kfree(req->apoll->double_poll);
 397                 kfree(req->apoll);
 398                 req->apoll = NULL;
 399         }
 400         if (req->flags & REQ_F_INFLIGHT)
 401                 atomic_dec(&req->tctx->inflight_tracked);
 402         if (req->flags & REQ_F_CREDS)
 403                 put_cred(req->creds);
 404         if (req->flags & REQ_F_ASYNC_DATA) {
 405                 kfree(req->async_data);
 406                 req->async_data = NULL;
 407         }
 408         req->flags &= ~IO_REQ_CLEAN_FLAGS;
 409 }
 410
 411 /*
 412  * Mark the request as inflight, so that file cancelation will find it.
 413  * Can be used if the file is an io_uring instance, or if the request itself
 414  * relies on ->mm being alive for the duration of the request.
 415  */
 416 inline void io_req_track_inflight(struct io_kiocb *req)
 417 {
 418         if (!(req->flags & REQ_F_INFLIGHT)) {
 419                 req->flags |= REQ_F_INFLIGHT;
 420                 atomic_inc(&req->tctx->inflight_tracked);
 421         }
 422 }
 423
 424 static struct io_kiocb *__io_prep_linked_timeout(struct io_kiocb *req)
 425 {
 426         if (WARN_ON_ONCE(!req->link))
 427                 return NULL;
 428
 429         req->flags &= ~REQ_F_ARM_LTIMEOUT;
 430         req->flags |= REQ_F_LINK_TIMEOUT;
 431
 432         /* linked timeouts should have two refs once prep'ed */
 433         io_req_set_refcount(req);
 434         __io_req_set_refcount(req->link, 2);
 435         return req->link;
 436 }
 437
 438 static void io_prep_async_work(struct io_kiocb *req)
 439 {
 440         const struct io_issue_def *def = &io_issue_defs[req->opcode];
 441         struct io_ring_ctx *ctx = req->ctx;
 442
 443         if (!(req->flags & REQ_F_CREDS)) {
 444                 req->flags |= REQ_F_CREDS;
 445                 req->creds = get_current_cred();
 446         }
 447
 448         req->work.list.next = NULL;
 449         atomic_set(&req->work.flags, 0);
 450         if (req->flags & REQ_F_FORCE_ASYNC)
 451                 atomic_or(IO_WQ_WORK_CONCURRENT, &req->work.flags);
 452
 453         if (req->file && !(req->flags & REQ_F_FIXED_FILE))
 454                 req->flags |= io_file_get_flags(req->file);
 455
 456         if (req->file && (req->flags & REQ_F_ISREG)) {
 457                 bool should_hash = def->hash_reg_file;
 458
 459                 /* don't serialize this request if the fs doesn't need it */
 460                 if (should_hash && (req->file->f_flags & O_DIRECT) &&
 461                     (req->file->f_op->fop_flags & FOP_DIO_PARALLEL_WRITE))
 462                         should_hash = false;
 463                 if (should_hash || (ctx->flags & IORING_SETUP_IOPOLL))
 464                         io_wq_hash_work(&req->work, file_inode(req->file));
 465         } else if (!req->file || !S_ISBLK(file_inode(req->file)->i_mode)) {
 466                 if (def->unbound_nonreg_file)
 467                         atomic_or(IO_WQ_WORK_UNBOUND, &req->work.flags);
 468         }
 469 }
 470
 471 static void io_prep_async_link(struct io_kiocb *req)
 472 {
 473         struct io_kiocb *cur;
 474
 475         if (req->flags & REQ_F_LINK_TIMEOUT) {
 476                 struct io_ring_ctx *ctx = req->ctx;
 477
 478                 raw_spin_lock_irq(&ctx->timeout_lock);
 479                 io_for_each_link(cur, req)
 480                         io_prep_async_work(cur);
 481                 raw_spin_unlock_irq(&ctx->timeout_lock);
 482         } else {
 483                 io_for_each_link(cur, req)
 484                         io_prep_async_work(cur);
 485         }
 486 }
 487
 488 static void io_queue_iowq(struct io_kiocb *req)
 489 {
 490         struct io_uring_task *tctx = req->tctx;
 491
 492         BUG_ON(!tctx);
 493
 494         if ((current->flags & PF_KTHREAD) || !tctx->io_wq) {
 495                 io_req_task_queue_fail(req, -ECANCELED);
 496                 return;
 497         }
 498
 499         /* init ->work of the whole link before punting */
 500         io_prep_async_link(req);
 501
 502         /*
 503          * Not expected to happen, but if we do have a bug where this _can_
 504          * happen, catch it here and ensure the request is marked as
 505          * canceled. That will make io-wq go through the usual work cancel
 506          * procedure rather than attempt to run this request (or create a new
 507          * worker for it).
 508          */
 509         if (WARN_ON_ONCE(!same_thread_group(tctx->task, current)))
 510                 atomic_or(IO_WQ_WORK_CANCEL, &req->work.flags);
 511
 512         trace_io_uring_queue_async_work(req, io_wq_is_hashed(&req->work));
 513         io_wq_enqueue(tctx->io_wq, &req->work);
 514 }
 515
 516 static void io_req_queue_iowq_tw(struct io_kiocb *req, io_tw_token_t tw)
 517 {
 518         io_queue_iowq(req);
 519 }
 520
 521 void io_req_queue_iowq(struct io_kiocb *req)
 522 {
 523         req->io_task_work.func = io_req_queue_iowq_tw;
 524         io_req_task_work_add(req);
 525 }
 526
 527 static unsigned io_linked_nr(struct io_kiocb *req)
 528 {
 529         struct io_kiocb *tmp;
 530         unsigned nr = 0;
 531
 532         io_for_each_link(tmp, req)
 533                 nr++;
 534         return nr;
 535 }
 536
 537 static __cold noinline void io_queue_deferred(struct io_ring_ctx *ctx)
 538 {
 539         bool drain_seen = false, first = true;
 540
 541         lockdep_assert_held(&ctx->uring_lock);
 542         __io_req_caches_free(ctx);
 543
 544         while (!list_empty(&ctx->defer_list)) {
 545                 struct io_defer_entry *de = list_first_entry(&ctx->defer_list,
 546                                                 struct io_defer_entry, list);
 547
 548                 drain_seen |= de->req->flags & REQ_F_IO_DRAIN;
 549                 if ((drain_seen || first) && ctx->nr_req_allocated != ctx->nr_drained)
 550                         return;
 551
 552                 list_del_init(&de->list);
 553                 ctx->nr_drained -= io_linked_nr(de->req);
 554                 io_req_task_queue(de->req);
 555                 kfree(de);
 556                 first = false;
 557         }
 558 }
 559
 560 void __io_commit_cqring_flush(struct io_ring_ctx *ctx)
 561 {
 562         if (ctx->poll_activated)
 563                 io_poll_wq_wake(ctx);
 564         if (ctx->off_timeout_used)
 565                 io_flush_timeouts(ctx);
 566         if (ctx->has_evfd)
 567                 io_eventfd_signal(ctx, true);
 568 }
 569
 570 static inline void __io_cq_lock(struct io_ring_ctx *ctx)
 571 {
 572         if (!ctx->lockless_cq)
 573                 spin_lock(&ctx->completion_lock);
 574 }
 575
 576 static inline void io_cq_lock(struct io_ring_ctx *ctx)
 577         __acquires(ctx->completion_lock)
 578 {
 579         spin_lock(&ctx->completion_lock);
 580 }
 581
 582 static inline void __io_cq_unlock_post(struct io_ring_ctx *ctx)
 583 {
 584         io_commit_cqring(ctx);
 585         if (!ctx->task_complete) {
 586                 if (!ctx->lockless_cq)
 587                         spin_unlock(&ctx->completion_lock);
 588                 /* IOPOLL rings only need to wake up if it's also SQPOLL */
 589                 if (!ctx->syscall_iopoll)
 590                         io_cqring_wake(ctx);
 591         }
 592         io_commit_cqring_flush(ctx);
 593 }
 594
 595 static void io_cq_unlock_post(struct io_ring_ctx *ctx)
 596         __releases(ctx->completion_lock)
 597 {
 598         io_commit_cqring(ctx);
 599         spin_unlock(&ctx->completion_lock);
 600         io_cqring_wake(ctx);
 601         io_commit_cqring_flush(ctx);
 602 }
 603
 604 static void __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool dying)
 605 {
 606         size_t cqe_size = sizeof(struct io_uring_cqe);
 607
 608         lockdep_assert_held(&ctx->uring_lock);
 609
 610         /* don't abort if we're dying, entries must get freed */
 611         if (!dying && __io_cqring_events(ctx) == ctx->cq_entries)
 612                 return;
 613
 614         if (ctx->flags & IORING_SETUP_CQE32)
 615                 cqe_size <<= 1;
 616
 617         io_cq_lock(ctx);
 618         while (!list_empty(&ctx->cq_overflow_list)) {
 619                 struct io_uring_cqe *cqe;
 620                 struct io_overflow_cqe *ocqe;
 621
 622                 ocqe = list_first_entry(&ctx->cq_overflow_list,
 623                                         struct io_overflow_cqe, list);
 624
 625                 if (!dying) {
 626                         if (!io_get_cqe_overflow(ctx, &cqe, true))
 627                                 break;
 628                         memcpy(cqe, &ocqe->cqe, cqe_size);
 629                 }
 630                 list_del(&ocqe->list);
 631                 kfree(ocqe);
 632
 633                 /*
 634                  * For silly syzbot cases that deliberately overflow by huge
 635                  * amounts, check if we need to resched and drop and
 636                  * reacquire the locks if so. Nothing real would ever hit this.
 637                  * Ideally we'd have a non-posting unlock for this, but hard
 638                  * to care for a non-real case.
 639                  */
 640                 if (need_resched()) {
 641                         ctx->cqe_sentinel = ctx->cqe_cached;
 642                         io_cq_unlock_post(ctx);
 643                         mutex_unlock(&ctx->uring_lock);
 644                         cond_resched();
 645                         mutex_lock(&ctx->uring_lock);
 646                         io_cq_lock(ctx);
 647                 }
 648         }
 649
 650         if (list_empty(&ctx->cq_overflow_list)) {
 651                 clear_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);
 652                 atomic_andnot(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);
 653         }
 654         io_cq_unlock_post(ctx);
 655 }
 656
 657 static void io_cqring_overflow_kill(struct io_ring_ctx *ctx)
 658 {
 659         if (ctx->rings)
 660                 __io_cqring_overflow_flush(ctx, true);
 661 }
 662
 663 static void io_cqring_do_overflow_flush(struct io_ring_ctx *ctx)
 664 {
 665         mutex_lock(&ctx->uring_lock);
 666         __io_cqring_overflow_flush(ctx, false);
 667         mutex_unlock(&ctx->uring_lock);
 668 }
 669
 670 /* must to be called somewhat shortly after putting a request */
 671 static inline void io_put_task(struct io_kiocb *req)
 672 {
 673         struct io_uring_task *tctx = req->tctx;
 674
 675         if (likely(tctx->task == current)) {
 676                 tctx->cached_refs++;
 677         } else {
 678                 percpu_counter_sub(&tctx->inflight, 1);
 679                 if (unlikely(atomic_read(&tctx->in_cancel)))
 680                         wake_up(&tctx->wait);
 681                 put_task_struct(tctx->task);
 682         }
 683 }
 684
 685 void io_task_refs_refill(struct io_uring_task *tctx)
 686 {
 687         unsigned int refill = -tctx->cached_refs + IO_TCTX_REFS_CACHE_NR;
 688
 689         percpu_counter_add(&tctx->inflight, refill);
 690         refcount_add(refill, &current->usage);
 691         tctx->cached_refs += refill;
 692 }
 693
 694 static __cold void io_uring_drop_tctx_refs(struct task_struct *task)
 695 {
 696         struct io_uring_task *tctx = task->io_uring;
 697         unsigned int refs = tctx->cached_refs;
 698
 699         if (refs) {
 700                 tctx->cached_refs = 0;
 701                 percpu_counter_sub(&tctx->inflight, refs);
 702                 put_task_struct_many(task, refs);
 703         }
 704 }
 705
 706 static __cold bool io_cqring_add_overflow(struct io_ring_ctx *ctx,
 707                                           struct io_overflow_cqe *ocqe)
 708 {
 709         lockdep_assert_held(&ctx->completion_lock);
 710
 711         if (!ocqe) {
 712                 struct io_rings *r = ctx->rings;
 713
 714                 /*
 715                  * If we're in ring overflow flush mode, or in task cancel mode,
 716                  * or cannot allocate an overflow entry, then we need to drop it
 717                  * on the floor.
 718                  */
 719                 WRITE_ONCE(r->cq_overflow, READ_ONCE(r->cq_overflow) + 1);
 720                 set_bit(IO_CHECK_CQ_DROPPED_BIT, &ctx->check_cq);
 721                 return false;
 722         }
 723         if (list_empty(&ctx->cq_overflow_list)) {
 724                 set_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);
 725                 atomic_or(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);
 726
 727         }
 728         list_add_tail(&ocqe->list, &ctx->cq_overflow_list);
 729         return true;
 730 }
 731
 732 static struct io_overflow_cqe *io_alloc_ocqe(struct io_ring_ctx *ctx,
 733                                              struct io_cqe *cqe,
 734                                              struct io_big_cqe *big_cqe, gfp_t gfp)
 735 {
 736         struct io_overflow_cqe *ocqe;
 737         size_t ocq_size = sizeof(struct io_overflow_cqe);
 738         bool is_cqe32 = (ctx->flags & IORING_SETUP_CQE32);
 739
 740         if (is_cqe32)
 741                 ocq_size += sizeof(struct io_uring_cqe);
 742
 743         ocqe = kzalloc(ocq_size, gfp | __GFP_ACCOUNT);
 744         trace_io_uring_cqe_overflow(ctx, cqe->user_data, cqe->res, cqe->flags, ocqe);
 745         if (ocqe) {
 746                 ocqe->cqe.user_data = cqe->user_data;
 747                 ocqe->cqe.res = cqe->res;
 748                 ocqe->cqe.flags = cqe->flags;
 749                 if (is_cqe32 && big_cqe) {
 750                         ocqe->cqe.big_cqe[0] = big_cqe->extra1;
 751                         ocqe->cqe.big_cqe[1] = big_cqe->extra2;
 752                 }
 753         }
 754         if (big_cqe)
 755                 big_cqe->extra1 = big_cqe->extra2 = 0;
 756         return ocqe;
 757 }
 758
 759 /*
 760  * writes to the cq entry need to come after reading head; the
 761  * control dependency is enough as we're using WRITE_ONCE to
 762  * fill the cq entry
 763  */
 764 bool io_cqe_cache_refill(struct io_ring_ctx *ctx, bool overflow)
 765 {
 766         struct io_rings *rings = ctx->rings;
 767         unsigned int off = ctx->cached_cq_tail & (ctx->cq_entries - 1);
 768         unsigned int free, queued, len;
 769
 770         /*
 771          * Posting into the CQ when there are pending overflowed CQEs may break
 772          * ordering guarantees, which will affect links, F_MORE users and more.
 773          * Force overflow the completion.
 774          */
 775         if (!overflow && (ctx->check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT)))
 776                 return false;
 777
 778         /* userspace may cheat modifying the tail, be safe and do min */
 779         queued = min(__io_cqring_events(ctx), ctx->cq_entries);
 780         free = ctx->cq_entries - queued;
 781         /* we need a contiguous range, limit based on the current array offset */
 782         len = min(free, ctx->cq_entries - off);
 783         if (!len)
 784                 return false;
 785
 786         if (ctx->flags & IORING_SETUP_CQE32) {
 787                 off <<= 1;
 788                 len <<= 1;
 789         }
 790
 791         ctx->cqe_cached = &rings->cqes[off];
 792         ctx->cqe_sentinel = ctx->cqe_cached + len;
 793         return true;
 794 }
 795
 796 static bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res,
 797                               u32 cflags)
 798 {
 799         struct io_uring_cqe *cqe;
 800
 801         if (likely(io_get_cqe(ctx, &cqe))) {
 802                 WRITE_ONCE(cqe->user_data, user_data);
 803                 WRITE_ONCE(cqe->res, res);
 804                 WRITE_ONCE(cqe->flags, cflags);
 805
 806                 if (ctx->flags & IORING_SETUP_CQE32) {
 807                         WRITE_ONCE(cqe->big_cqe[0], 0);
 808                         WRITE_ONCE(cqe->big_cqe[1], 0);
 809                 }
 810
 811                 trace_io_uring_complete(ctx, NULL, cqe);
 812                 return true;
 813         }
 814         return false;
 815 }
 816
 817 static inline struct io_cqe io_init_cqe(u64 user_data, s32 res, u32 cflags)
 818 {
 819         return (struct io_cqe) { .user_data = user_data, .res = res, .flags = cflags };
 820 }
 821
 822 static __cold void io_cqe_overflow(struct io_ring_ctx *ctx, struct io_cqe *cqe,
 823                                    struct io_big_cqe *big_cqe)
 824 {
 825         struct io_overflow_cqe *ocqe;
 826
 827         ocqe = io_alloc_ocqe(ctx, cqe, big_cqe, GFP_KERNEL);
 828         spin_lock(&ctx->completion_lock);
 829         io_cqring_add_overflow(ctx, ocqe);
 830         spin_unlock(&ctx->completion_lock);
 831 }
 832
 833 static __cold bool io_cqe_overflow_locked(struct io_ring_ctx *ctx,
 834                                           struct io_cqe *cqe,
 835                                           struct io_big_cqe *big_cqe)
 836 {
 837         struct io_overflow_cqe *ocqe;
 838
 839         ocqe = io_alloc_ocqe(ctx, cqe, big_cqe, GFP_ATOMIC);
 840         return io_cqring_add_overflow(ctx, ocqe);
 841 }
 842
 843 bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
 844 {
 845         bool filled;
 846
 847         io_cq_lock(ctx);
 848         filled = io_fill_cqe_aux(ctx, user_data, res, cflags);
 849         if (unlikely(!filled)) {
 850                 struct io_cqe cqe = io_init_cqe(user_data, res, cflags);
 851
 852                 filled = io_cqe_overflow_locked(ctx, &cqe, NULL);
 853         }
 854         io_cq_unlock_post(ctx);
 855         return filled;
 856 }
 857
 858 /*
 859  * Must be called from inline task_work so we now a flush will happen later,
 860  * and obviously with ctx->uring_lock held (tw always has that).
 861  */
 862 void io_add_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
 863 {
 864         lockdep_assert_held(&ctx->uring_lock);
 865         lockdep_assert(ctx->lockless_cq);
 866
 867         if (!io_fill_cqe_aux(ctx, user_data, res, cflags)) {
 868                 struct io_cqe cqe = io_init_cqe(user_data, res, cflags);
 869
 870                 io_cqe_overflow(ctx, &cqe, NULL);
 871         }
 872         ctx->submit_state.cq_flush = true;
 873 }
 874
 875 /*
 876  * A helper for multishot requests posting additional CQEs.
 877  * Should only be used from a task_work including IO_URING_F_MULTISHOT.
 878  */
 879 bool io_req_post_cqe(struct io_kiocb *req, s32 res, u32 cflags)
 880 {
 881         struct io_ring_ctx *ctx = req->ctx;
 882         bool posted;
 883
 884         /*
 885          * If multishot has already posted deferred completions, ensure that
 886          * those are flushed first before posting this one. If not, CQEs
 887          * could get reordered.
 888          */
 889         if (!wq_list_empty(&ctx->submit_state.compl_reqs))
 890                 __io_submit_flush_completions(ctx);
 891
 892         lockdep_assert(!io_wq_current_is_worker());
 893         lockdep_assert_held(&ctx->uring_lock);
 894
 895         if (!ctx->lockless_cq) {
 896                 spin_lock(&ctx->completion_lock);
 897                 posted = io_fill_cqe_aux(ctx, req->cqe.user_data, res, cflags);
 898                 spin_unlock(&ctx->completion_lock);
 899         } else {
 900                 posted = io_fill_cqe_aux(ctx, req->cqe.user_data, res, cflags);
 901         }
 902
 903         ctx->submit_state.cq_flush = true;
 904         return posted;
 905 }
 906
 907 static void io_req_complete_post(struct io_kiocb *req, unsigned issue_flags)
 908 {
 909         struct io_ring_ctx *ctx = req->ctx;
 910         bool completed = true;
 911
 912         /*
 913          * All execution paths but io-wq use the deferred completions by
 914          * passing IO_URING_F_COMPLETE_DEFER and thus should not end up here.
 915          */
 916         if (WARN_ON_ONCE(!(issue_flags & IO_URING_F_IOWQ)))
 917                 return;
 918
 919         /*
 920          * Handle special CQ sync cases via task_work. DEFER_TASKRUN requires
 921          * the submitter task context, IOPOLL protects with uring_lock.
 922          */
 923         if (ctx->lockless_cq || (req->flags & REQ_F_REISSUE)) {
 924 defer_complete:
 925                 req->io_task_work.func = io_req_task_complete;
 926                 io_req_task_work_add(req);
 927                 return;
 928         }
 929
 930         io_cq_lock(ctx);
 931         if (!(req->flags & REQ_F_CQE_SKIP))
 932                 completed = io_fill_cqe_req(ctx, req);
 933         io_cq_unlock_post(ctx);
 934
 935         if (!completed)
 936                 goto defer_complete;
 937
 938         /*
 939          * We don't free the request here because we know it's called from
 940          * io-wq only, which holds a reference, so it cannot be the last put.
 941          */
 942         req_ref_put(req);
 943 }
 944
 945 void io_req_defer_failed(struct io_kiocb *req, s32 res)
 946         __must_hold(&ctx->uring_lock)
 947 {
 948         const struct io_cold_def *def = &io_cold_defs[req->opcode];
 949
 950         lockdep_assert_held(&req->ctx->uring_lock);
 951
 952         req_set_fail(req);
 953         io_req_set_res(req, res, io_put_kbuf(req, res, IO_URING_F_UNLOCKED));
 954         if (def->fail)
 955                 def->fail(req);
 956         io_req_complete_defer(req);
 957 }
 958
 959 /*
 960  * A request might get retired back into the request caches even before opcode
 961  * handlers and io_issue_sqe() are done with it, e.g. inline completion path.
 962  * Because of that, io_alloc_req() should be called only under ->uring_lock
 963  * and with extra caution to not get a request that is still worked on.
 964  */
 965 __cold bool __io_alloc_req_refill(struct io_ring_ctx *ctx)
 966         __must_hold(&ctx->uring_lock)
 967 {
 968         gfp_t gfp = GFP_KERNEL | __GFP_NOWARN | __GFP_ZERO;
 969         void *reqs[IO_REQ_ALLOC_BATCH];
 970         int ret;
 971
 972         ret = kmem_cache_alloc_bulk(req_cachep, gfp, ARRAY_SIZE(reqs), reqs);
 973
 974         /*
 975          * Bulk alloc is all-or-nothing. If we fail to get a batch,
 976          * retry single alloc to be on the safe side.
 977          */
 978         if (unlikely(ret <= 0)) {
 979                 reqs[0] = kmem_cache_alloc(req_cachep, gfp);
 980                 if (!reqs[0])
 981                         return false;
 982                 ret = 1;
 983         }
 984
 985         percpu_ref_get_many(&ctx->refs, ret);
 986         ctx->nr_req_allocated += ret;
 987
 988         while (ret--) {
 989                 struct io_kiocb *req = reqs[ret];
 990
 991                 io_req_add_to_cache(req, ctx);
 992         }
 993         return true;
 994 }
 995
 996 __cold void io_free_req(struct io_kiocb *req)
 997 {
 998         /* refs were already put, restore them for io_req_task_complete() */
 999         req->flags &= ~REQ_F_REFCOUNT;
1000         /* we only want to free it, don't post CQEs */
1001         req->flags |= REQ_F_CQE_SKIP;
1002         req->io_task_work.func = io_req_task_complete;
1003         io_req_task_work_add(req);
1004 }
1005
1006 static void __io_req_find_next_prep(struct io_kiocb *req)
1007 {
1008         struct io_ring_ctx *ctx = req->ctx;
1009
1010         spin_lock(&ctx->completion_lock);
1011         io_disarm_next(req);
1012         spin_unlock(&ctx->completion_lock);
1013 }
1014
1015 static inline struct io_kiocb *io_req_find_next(struct io_kiocb *req)
1016 {
1017         struct io_kiocb *nxt;
1018
1019         /*
1020          * If LINK is set, we have dependent requests in this chain. If we
1021          * didn't fail this request, queue the first one up, moving any other
1022          * dependencies to the next request. In case of failure, fail the rest
1023          * of the chain.
1024          */
1025         if (unlikely(req->flags & IO_DISARM_MASK))
1026                 __io_req_find_next_prep(req);
1027         nxt = req->link;
1028         req->link = NULL;
1029         return nxt;
1030 }
1031
1032 static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw)
1033 {
1034         if (!ctx)
1035                 return;
1036         if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1037                 atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
1038
1039         io_submit_flush_completions(ctx);
1040         mutex_unlock(&ctx->uring_lock);
1041         percpu_ref_put(&ctx->refs);
1042 }
1043
1044 /*
1045  * Run queued task_work, returning the number of entries processed in *count.
1046  * If more entries than max_entries are available, stop processing once this
1047  * is reached and return the rest of the list.
1048  */
1049 struct llist_node *io_handle_tw_list(struct llist_node *node,
1050                                      unsigned int *count,
1051                                      unsigned int max_entries)
1052 {
1053         struct io_ring_ctx *ctx = NULL;
1054         struct io_tw_state ts = { };
1055
1056         do {
1057                 struct llist_node *next = node->next;
1058                 struct io_kiocb *req = container_of(node, struct io_kiocb,
1059                                                     io_task_work.node);
1060
1061                 if (req->ctx != ctx) {
1062                         ctx_flush_and_put(ctx, ts);
1063                         ctx = req->ctx;
1064                         mutex_lock(&ctx->uring_lock);
1065                         percpu_ref_get(&ctx->refs);
1066                 }
1067                 INDIRECT_CALL_2(req->io_task_work.func,
1068                                 io_poll_task_func, io_req_rw_complete,
1069                                 req, ts);
1070                 node = next;
1071                 (*count)++;
1072                 if (unlikely(need_resched())) {
1073                         ctx_flush_and_put(ctx, ts);
1074                         ctx = NULL;
1075                         cond_resched();
1076                 }
1077         } while (node && *count < max_entries);
1078
1079         ctx_flush_and_put(ctx, ts);
1080         return node;
1081 }
1082
1083 static __cold void __io_fallback_tw(struct llist_node *node, bool sync)
1084 {
1085         struct io_ring_ctx *last_ctx = NULL;
1086         struct io_kiocb *req;
1087
1088         while (node) {
1089                 req = container_of(node, struct io_kiocb, io_task_work.node);
1090                 node = node->next;
1091                 if (last_ctx != req->ctx) {
1092                         if (last_ctx) {
1093                                 if (sync)
1094                                         flush_delayed_work(&last_ctx->fallback_work);
1095                                 percpu_ref_put(&last_ctx->refs);
1096                         }
1097                         last_ctx = req->ctx;
1098                         percpu_ref_get(&last_ctx->refs);
1099                 }
1100                 if (llist_add(&req->io_task_work.node, &last_ctx->fallback_llist))
1101                         schedule_delayed_work(&last_ctx->fallback_work, 1);
1102         }
1103
1104         if (last_ctx) {
1105                 if (sync)
1106                         flush_delayed_work(&last_ctx->fallback_work);
1107                 percpu_ref_put(&last_ctx->refs);
1108         }
1109 }
1110
1111 static void io_fallback_tw(struct io_uring_task *tctx, bool sync)
1112 {
1113         struct llist_node *node = llist_del_all(&tctx->task_list);
1114
1115         __io_fallback_tw(node, sync);
1116 }
1117
1118 struct llist_node *tctx_task_work_run(struct io_uring_task *tctx,
1119                                       unsigned int max_entries,
1120                                       unsigned int *count)
1121 {
1122         struct llist_node *node;
1123
1124         if (unlikely(current->flags & PF_EXITING)) {
1125                 io_fallback_tw(tctx, true);
1126                 return NULL;
1127         }
1128
1129         node = llist_del_all(&tctx->task_list);
1130         if (node) {
1131                 node = llist_reverse_order(node);
1132                 node = io_handle_tw_list(node, count, max_entries);
1133         }
1134
1135         /* relaxed read is enough as only the task itself sets ->in_cancel */
1136         if (unlikely(atomic_read(&tctx->in_cancel)))
1137                 io_uring_drop_tctx_refs(current);
1138
1139         trace_io_uring_task_work_run(tctx, *count);
1140         return node;
1141 }
1142
1143 void tctx_task_work(struct callback_head *cb)
1144 {
1145         struct io_uring_task *tctx;
1146         struct llist_node *ret;
1147         unsigned int count = 0;
1148
1149         tctx = container_of(cb, struct io_uring_task, task_work);
1150         ret = tctx_task_work_run(tctx, UINT_MAX, &count);
1151         /* can't happen */
1152         WARN_ON_ONCE(ret);
1153 }
1154
1155 static void io_req_local_work_add(struct io_kiocb *req, unsigned flags)
1156 {
1157         struct io_ring_ctx *ctx = req->ctx;
1158         unsigned nr_wait, nr_tw, nr_tw_prev;
1159         struct llist_node *head;
1160
1161         /* See comment above IO_CQ_WAKE_INIT */
1162         BUILD_BUG_ON(IO_CQ_WAKE_FORCE <= IORING_MAX_CQ_ENTRIES);
1163
1164         /*
1165          * We don't know how many reuqests is there in the link and whether
1166          * they can even be queued lazily, fall back to non-lazy.
1167          */
1168         if (req->flags & IO_REQ_LINK_FLAGS)
1169                 flags &= ~IOU_F_TWQ_LAZY_WAKE;
1170
1171         guard(rcu)();
1172
1173         head = READ_ONCE(ctx->work_llist.first);
1174         do {
1175                 nr_tw_prev = 0;
1176                 if (head) {
1177                         struct io_kiocb *first_req = container_of(head,
1178                                                         struct io_kiocb,
1179                                                         io_task_work.node);
1180                         /*
1181                          * Might be executed at any moment, rely on
1182                          * SLAB_TYPESAFE_BY_RCU to keep it alive.
1183                          */
1184                         nr_tw_prev = READ_ONCE(first_req->nr_tw);
1185                 }
1186
1187                 /*
1188                  * Theoretically, it can overflow, but that's fine as one of
1189                  * previous adds should've tried to wake the task.
1190                  */
1191                 nr_tw = nr_tw_prev + 1;
1192                 if (!(flags & IOU_F_TWQ_LAZY_WAKE))
1193                         nr_tw = IO_CQ_WAKE_FORCE;
1194
1195                 req->nr_tw = nr_tw;
1196                 req->io_task_work.node.next = head;
1197         } while (!try_cmpxchg(&ctx->work_llist.first, &head,
1198                               &req->io_task_work.node));
1199
1200         /*
1201          * cmpxchg implies a full barrier, which pairs with the barrier
1202          * in set_current_state() on the io_cqring_wait() side. It's used
1203          * to ensure that either we see updated ->cq_wait_nr, or waiters
1204          * going to sleep will observe the work added to the list, which
1205          * is similar to the wait/wawke task state sync.
1206          */
1207
1208         if (!head) {
1209                 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1210                         atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
1211                 if (ctx->has_evfd)
1212                         io_eventfd_signal(ctx, false);
1213         }
1214
1215         nr_wait = atomic_read(&ctx->cq_wait_nr);
1216         /* not enough or no one is waiting */
1217         if (nr_tw < nr_wait)
1218                 return;
1219         /* the previous add has already woken it up */
1220         if (nr_tw_prev >= nr_wait)
1221                 return;
1222         wake_up_state(ctx->submitter_task, TASK_INTERRUPTIBLE);
1223 }
1224
1225 static void io_req_normal_work_add(struct io_kiocb *req)
1226 {
1227         struct io_uring_task *tctx = req->tctx;
1228         struct io_ring_ctx *ctx = req->ctx;
1229
1230         /* task_work already pending, we're done */
1231         if (!llist_add(&req->io_task_work.node, &tctx->task_list))
1232                 return;
1233
1234         if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1235                 atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
1236
1237         /* SQPOLL doesn't need the task_work added, it'll run it itself */
1238         if (ctx->flags & IORING_SETUP_SQPOLL) {
1239                 __set_notify_signal(tctx->task);
1240                 return;
1241         }
1242
1243         if (likely(!task_work_add(tctx->task, &tctx->task_work, ctx->notify_method)))
1244                 return;
1245
1246         io_fallback_tw(tctx, false);
1247 }
1248
1249 void __io_req_task_work_add(struct io_kiocb *req, unsigned flags)
1250 {
1251         if (req->ctx->flags & IORING_SETUP_DEFER_TASKRUN)
1252                 io_req_local_work_add(req, flags);
1253         else
1254                 io_req_normal_work_add(req);
1255 }
1256
1257 void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags)
1258 {
1259         if (WARN_ON_ONCE(!(req->ctx->flags & IORING_SETUP_DEFER_TASKRUN)))
1260                 return;
1261         __io_req_task_work_add(req, flags);
1262 }
1263
1264 static void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx)
1265 {
1266         struct llist_node *node = llist_del_all(&ctx->work_llist);
1267
1268         __io_fallback_tw(node, false);
1269         node = llist_del_all(&ctx->retry_llist);
1270         __io_fallback_tw(node, false);
1271 }
1272
1273 static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events,
1274                                        int min_events)
1275 {
1276         if (!io_local_work_pending(ctx))
1277                 return false;
1278         if (events < min_events)
1279                 return true;
1280         if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1281                 atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
1282         return false;
1283 }
1284
1285 static int __io_run_local_work_loop(struct llist_node **node,
1286                                     io_tw_token_t tw,
1287                                     int events)
1288 {
1289         int ret = 0;
1290
1291         while (*node) {
1292                 struct llist_node *next = (*node)->next;
1293                 struct io_kiocb *req = container_of(*node, struct io_kiocb,
1294                                                     io_task_work.node);
1295                 INDIRECT_CALL_2(req->io_task_work.func,
1296                                 io_poll_task_func, io_req_rw_complete,
1297                                 req, tw);
1298                 *node = next;
1299                 if (++ret >= events)
1300                         break;
1301         }
1302
1303         return ret;
1304 }
1305
1306 static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw,
1307                                int min_events, int max_events)
1308 {
1309         struct llist_node *node;
1310         unsigned int loops = 0;
1311         int ret = 0;
1312
1313         if (WARN_ON_ONCE(ctx->submitter_task != current))
1314                 return -EEXIST;
1315         if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
1316                 atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
1317 again:
1318         min_events -= ret;
1319         ret = __io_run_local_work_loop(&ctx->retry_llist.first, tw, max_events);
1320         if (ctx->retry_llist.first)
1321                 goto retry_done;
1322
1323         /*
1324          * llists are in reverse order, flip it back the right way before
1325          * running the pending items.
1326          */
1327         node = llist_reverse_order(llist_del_all(&ctx->work_llist));
1328         ret += __io_run_local_work_loop(&node, tw, max_events - ret);
1329         ctx->retry_llist.first = node;
1330         loops++;
1331
1332         if (io_run_local_work_continue(ctx, ret, min_events))
1333                 goto again;
1334 retry_done:
1335         io_submit_flush_completions(ctx);
1336         if (io_run_local_work_continue(ctx, ret, min_events))
1337                 goto again;
1338
1339         trace_io_uring_local_work_run(ctx, ret, loops);
1340         return ret;
1341 }
1342
1343 static inline int io_run_local_work_locked(struct io_ring_ctx *ctx,
1344                                            int min_events)
1345 {
1346         struct io_tw_state ts = {};
1347
1348         if (!io_local_work_pending(ctx))
1349                 return 0;
1350         return __io_run_local_work(ctx, ts, min_events,
1351                                         max(IO_LOCAL_TW_DEFAULT_MAX, min_events));
1352 }
1353
1354 static int io_run_local_work(struct io_ring_ctx *ctx, int min_events,
1355                              int max_events)
1356 {
1357         struct io_tw_state ts = {};
1358         int ret;
1359
1360         mutex_lock(&ctx->uring_lock);
1361         ret = __io_run_local_work(ctx, ts, min_events, max_events);
1362         mutex_unlock(&ctx->uring_lock);
1363         return ret;
1364 }
1365
1366 static void io_req_task_cancel(struct io_kiocb *req, io_tw_token_t tw)
1367 {
1368         io_tw_lock(req->ctx, tw);
1369         io_req_defer_failed(req, req->cqe.res);
1370 }
1371
1372 void io_req_task_submit(struct io_kiocb *req, io_tw_token_t tw)
1373 {
1374         io_tw_lock(req->ctx, tw);
1375         if (unlikely(io_should_terminate_tw()))
1376                 io_req_defer_failed(req, -EFAULT);
1377         else if (req->flags & REQ_F_FORCE_ASYNC)
1378                 io_queue_iowq(req);
1379         else
1380                 io_queue_sqe(req);
1381 }
1382
1383 void io_req_task_queue_fail(struct io_kiocb *req, int ret)
1384 {
1385         io_req_set_res(req, ret, 0);
1386         req->io_task_work.func = io_req_task_cancel;
1387         io_req_task_work_add(req);
1388 }
1389
1390 void io_req_task_queue(struct io_kiocb *req)
1391 {
1392         req->io_task_work.func = io_req_task_submit;
1393         io_req_task_work_add(req);
1394 }
1395
1396 void io_queue_next(struct io_kiocb *req)
1397 {
1398         struct io_kiocb *nxt = io_req_find_next(req);
1399
1400         if (nxt)
1401                 io_req_task_queue(nxt);
1402 }
1403
1404 static inline void io_req_put_rsrc_nodes(struct io_kiocb *req)
1405 {
1406         if (req->file_node) {
1407                 io_put_rsrc_node(req->ctx, req->file_node);
1408                 req->file_node = NULL;
1409         }
1410         if (req->flags & REQ_F_BUF_NODE)
1411                 io_put_rsrc_node(req->ctx, req->buf_node);
1412 }
1413
1414 static void io_free_batch_list(struct io_ring_ctx *ctx,
1415                                struct io_wq_work_node *node)
1416         __must_hold(&ctx->uring_lock)
1417 {
1418         do {
1419                 struct io_kiocb *req = container_of(node, struct io_kiocb,
1420                                                     comp_list);
1421
1422                 if (unlikely(req->flags & IO_REQ_CLEAN_SLOW_FLAGS)) {
1423                         if (req->flags & REQ_F_REISSUE) {
1424                                 node = req->comp_list.next;
1425                                 req->flags &= ~REQ_F_REISSUE;
1426                                 io_queue_iowq(req);
1427                                 continue;
1428                         }
1429                         if (req->flags & REQ_F_REFCOUNT) {
1430                                 node = req->comp_list.next;
1431                                 if (!req_ref_put_and_test(req))
1432                                         continue;
1433                         }
1434                         if ((req->flags & REQ_F_POLLED) && req->apoll) {
1435                                 struct async_poll *apoll = req->apoll;
1436
1437                                 if (apoll->double_poll)
1438                                         kfree(apoll->double_poll);
1439                                 io_cache_free(&ctx->apoll_cache, apoll);
1440                                 req->flags &= ~REQ_F_POLLED;
1441                         }
1442                         if (req->flags & IO_REQ_LINK_FLAGS)
1443                                 io_queue_next(req);
1444                         if (unlikely(req->flags & IO_REQ_CLEAN_FLAGS))
1445                                 io_clean_op(req);
1446                 }
1447                 io_put_file(req);
1448                 io_req_put_rsrc_nodes(req);
1449                 io_put_task(req);
1450
1451                 node = req->comp_list.next;
1452                 io_req_add_to_cache(req, ctx);
1453         } while (node);
1454 }
1455
1456 void __io_submit_flush_completions(struct io_ring_ctx *ctx)
1457         __must_hold(&ctx->uring_lock)
1458 {
1459         struct io_submit_state *state = &ctx->submit_state;
1460         struct io_wq_work_node *node;
1461
1462         __io_cq_lock(ctx);
1463         __wq_list_for_each(node, &state->compl_reqs) {
1464                 struct io_kiocb *req = container_of(node, struct io_kiocb,
1465                                             comp_list);
1466
1467                 /*
1468                  * Requests marked with REQUEUE should not post a CQE, they
1469                  * will go through the io-wq retry machinery and post one
1470                  * later.
1471                  */
1472                 if (!(req->flags & (REQ_F_CQE_SKIP | REQ_F_REISSUE)) &&
1473                     unlikely(!io_fill_cqe_req(ctx, req))) {
1474                         if (ctx->lockless_cq)
1475                                 io_cqe_overflow(ctx, &req->cqe, &req->big_cqe);
1476                         else
1477                                 io_cqe_overflow_locked(ctx, &req->cqe, &req->big_cqe);
1478                 }
1479         }
1480         __io_cq_unlock_post(ctx);
1481
1482         if (!wq_list_empty(&state->compl_reqs)) {
1483                 io_free_batch_list(ctx, state->compl_reqs.first);
1484                 INIT_WQ_LIST(&state->compl_reqs);
1485         }
1486
1487         if (unlikely(ctx->drain_active))
1488                 io_queue_deferred(ctx);
1489
1490         ctx->submit_state.cq_flush = false;
1491 }
1492
1493 static unsigned io_cqring_events(struct io_ring_ctx *ctx)
1494 {
1495         /* See comment at the top of this file */
1496         smp_rmb();
1497         return __io_cqring_events(ctx);
1498 }
1499
1500 /*
1501  * We can't just wait for polled events to come to us, we have to actively
1502  * find and complete them.
1503  */
1504 static __cold void io_iopoll_try_reap_events(struct io_ring_ctx *ctx)
1505 {
1506         if (!(ctx->flags & IORING_SETUP_IOPOLL))
1507                 return;
1508
1509         mutex_lock(&ctx->uring_lock);
1510         while (!wq_list_empty(&ctx->iopoll_list)) {
1511                 /* let it sleep and repeat later if can't complete a request */
1512                 if (io_do_iopoll(ctx, true) == 0)
1513                         break;
1514                 /*
1515                  * Ensure we allow local-to-the-cpu processing to take place,
1516                  * in this case we need to ensure that we reap all events.
1517                  * Also let task_work, etc. to progress by releasing the mutex
1518                  */
1519                 if (need_resched()) {
1520                         mutex_unlock(&ctx->uring_lock);
1521                         cond_resched();
1522                         mutex_lock(&ctx->uring_lock);
1523                 }
1524         }
1525         mutex_unlock(&ctx->uring_lock);
1526
1527         if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
1528                 io_move_task_work_from_local(ctx);
1529 }
1530
1531 static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned int min_events)
1532 {
1533         unsigned int nr_events = 0;
1534         unsigned long check_cq;
1535
1536         min_events = min(min_events, ctx->cq_entries);
1537
1538         lockdep_assert_held(&ctx->uring_lock);
1539
1540         if (!io_allowed_run_tw(ctx))
1541                 return -EEXIST;
1542
1543         check_cq = READ_ONCE(ctx->check_cq);
1544         if (unlikely(check_cq)) {
1545                 if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT))
1546                         __io_cqring_overflow_flush(ctx, false);
1547                 /*
1548                  * Similarly do not spin if we have not informed the user of any
1549                  * dropped CQE.
1550                  */
1551                 if (check_cq & BIT(IO_CHECK_CQ_DROPPED_BIT))
1552                         return -EBADR;
1553         }
1554         /*
1555          * Don't enter poll loop if we already have events pending.
1556          * If we do, we can potentially be spinning for commands that
1557          * already triggered a CQE (eg in error).
1558          */
1559         if (io_cqring_events(ctx))
1560                 return 0;
1561
1562         do {
1563                 int ret = 0;
1564
1565                 /*
1566                  * If a submit got punted to a workqueue, we can have the
1567                  * application entering polling for a command before it gets
1568                  * issued. That app will hold the uring_lock for the duration
1569                  * of the poll right here, so we need to take a breather every
1570                  * now and then to ensure that the issue has a chance to add
1571                  * the poll to the issued list. Otherwise we can spin here
1572                  * forever, while the workqueue is stuck trying to acquire the
1573                  * very same mutex.
1574                  */
1575                 if (wq_list_empty(&ctx->iopoll_list) ||
1576                     io_task_work_pending(ctx)) {
1577                         u32 tail = ctx->cached_cq_tail;
1578
1579                         (void) io_run_local_work_locked(ctx, min_events);
1580
1581                         if (task_work_pending(current) ||
1582                             wq_list_empty(&ctx->iopoll_list)) {
1583                                 mutex_unlock(&ctx->uring_lock);
1584                                 io_run_task_work();
1585                                 mutex_lock(&ctx->uring_lock);
1586                         }
1587                         /* some requests don't go through iopoll_list */
1588                         if (tail != ctx->cached_cq_tail ||
1589                             wq_list_empty(&ctx->iopoll_list))
1590                                 break;
1591                 }
1592                 ret = io_do_iopoll(ctx, !min_events);
1593                 if (unlikely(ret < 0))
1594                         return ret;
1595
1596                 if (task_sigpending(current))
1597                         return -EINTR;
1598                 if (need_resched())
1599                         break;
1600
1601                 nr_events += ret;
1602         } while (nr_events < min_events);
1603
1604         return 0;
1605 }
1606
1607 void io_req_task_complete(struct io_kiocb *req, io_tw_token_t tw)
1608 {
1609         io_req_complete_defer(req);
1610 }
1611
1612 /*
1613  * After the iocb has been issued, it's safe to be found on the poll list.
1614  * Adding the kiocb to the list AFTER submission ensures that we don't
1615  * find it from a io_do_iopoll() thread before the issuer is done
1616  * accessing the kiocb cookie.
1617  */
1618 static void io_iopoll_req_issued(struct io_kiocb *req, unsigned int issue_flags)
1619 {
1620         struct io_ring_ctx *ctx = req->ctx;
1621         const bool needs_lock = issue_flags & IO_URING_F_UNLOCKED;
1622
1623         /* workqueue context doesn't hold uring_lock, grab it now */
1624         if (unlikely(needs_lock))
1625                 mutex_lock(&ctx->uring_lock);
1626
1627         /*
1628          * Track whether we have multiple files in our lists. This will impact
1629          * how we do polling eventually, not spinning if we're on potentially
1630          * different devices.
1631          */
1632         if (wq_list_empty(&ctx->iopoll_list)) {
1633                 ctx->poll_multi_queue = false;
1634         } else if (!ctx->poll_multi_queue) {
1635                 struct io_kiocb *list_req;
1636
1637                 list_req = container_of(ctx->iopoll_list.first, struct io_kiocb,
1638                                         comp_list);
1639                 if (list_req->file != req->file)
1640                         ctx->poll_multi_queue = true;
1641         }
1642
1643         /*
1644          * For fast devices, IO may have already completed. If it has, add
1645          * it to the front so we find it first.
1646          */
1647         if (READ_ONCE(req->iopoll_completed))
1648                 wq_list_add_head(&req->comp_list, &ctx->iopoll_list);
1649         else
1650                 wq_list_add_tail(&req->comp_list, &ctx->iopoll_list);
1651
1652         if (unlikely(needs_lock)) {
1653                 /*
1654                  * If IORING_SETUP_SQPOLL is enabled, sqes are either handle
1655                  * in sq thread task context or in io worker task context. If
1656                  * current task context is sq thread, we don't need to check
1657                  * whether should wake up sq thread.
1658                  */
1659                 if ((ctx->flags & IORING_SETUP_SQPOLL) &&
1660                     wq_has_sleeper(&ctx->sq_data->wait))
1661                         wake_up(&ctx->sq_data->wait);
1662
1663                 mutex_unlock(&ctx->uring_lock);
1664         }
1665 }
1666
1667 io_req_flags_t io_file_get_flags(struct file *file)
1668 {
1669         struct inode *inode = file_inode(file);
1670         io_req_flags_t res = 0;
1671
1672         BUILD_BUG_ON(REQ_F_ISREG_BIT != REQ_F_SUPPORT_NOWAIT_BIT + 1);
1673
1674         if (S_ISREG(inode->i_mode) && !(inode->i_flags & S_ANON_INODE))
1675                 res |= REQ_F_ISREG;
1676         if ((file->f_flags & O_NONBLOCK) || (file->f_mode & FMODE_NOWAIT))
1677                 res |= REQ_F_SUPPORT_NOWAIT;
1678         return res;
1679 }
1680
1681 static __cold void io_drain_req(struct io_kiocb *req)
1682         __must_hold(&ctx->uring_lock)
1683 {
1684         struct io_ring_ctx *ctx = req->ctx;
1685         bool drain = req->flags & IOSQE_IO_DRAIN;
1686         struct io_defer_entry *de;
1687
1688         de = kmalloc(sizeof(*de), GFP_KERNEL_ACCOUNT);
1689         if (!de) {
1690                 io_req_defer_failed(req, -ENOMEM);
1691                 return;
1692         }
1693
1694         io_prep_async_link(req);
1695         trace_io_uring_defer(req);
1696         de->req = req;
1697
1698         ctx->nr_drained += io_linked_nr(req);
1699         list_add_tail(&de->list, &ctx->defer_list);
1700         io_queue_deferred(ctx);
1701         if (!drain && list_empty(&ctx->defer_list))
1702                 ctx->drain_active = false;
1703 }
1704
1705 static bool io_assign_file(struct io_kiocb *req, const struct io_issue_def *def,
1706                            unsigned int issue_flags)
1707 {
1708         if (req->file || !def->needs_file)
1709                 return true;
1710
1711         if (req->flags & REQ_F_FIXED_FILE)
1712                 req->file = io_file_get_fixed(req, req->cqe.fd, issue_flags);
1713         else
1714                 req->file = io_file_get_normal(req, req->cqe.fd);
1715
1716         return !!req->file;
1717 }
1718
1719 #define REQ_ISSUE_SLOW_FLAGS    (REQ_F_CREDS | REQ_F_ARM_LTIMEOUT)
1720
1721 static inline int __io_issue_sqe(struct io_kiocb *req,
1722                                  unsigned int issue_flags,
1723                                  const struct io_issue_def *def)
1724 {
1725         const struct cred *creds = NULL;
1726         struct io_kiocb *link = NULL;
1727         int ret;
1728
1729         if (unlikely(req->flags & REQ_ISSUE_SLOW_FLAGS)) {
1730                 if ((req->flags & REQ_F_CREDS) && req->creds != current_cred())
1731                         creds = override_creds(req->creds);
1732                 if (req->flags & REQ_F_ARM_LTIMEOUT)
1733                         link = __io_prep_linked_timeout(req);
1734         }
1735
1736         if (!def->audit_skip)
1737                 audit_uring_entry(req->opcode);
1738
1739         ret = def->issue(req, issue_flags);
1740
1741         if (!def->audit_skip)
1742                 audit_uring_exit(!ret, ret);
1743
1744         if (unlikely(creds || link)) {
1745                 if (creds)
1746                         revert_creds(creds);
1747                 if (link)
1748                         io_queue_linked_timeout(link);
1749         }
1750
1751         return ret;
1752 }
1753
1754 static int io_issue_sqe(struct io_kiocb *req, unsigned int issue_flags)
1755 {
1756         const struct io_issue_def *def = &io_issue_defs[req->opcode];
1757         int ret;
1758
1759         if (unlikely(!io_assign_file(req, def, issue_flags)))
1760                 return -EBADF;
1761
1762         ret = __io_issue_sqe(req, issue_flags, def);
1763
1764         if (ret == IOU_COMPLETE) {
1765                 if (issue_flags & IO_URING_F_COMPLETE_DEFER)
1766                         io_req_complete_defer(req);
1767                 else
1768                         io_req_complete_post(req, issue_flags);
1769
1770                 return 0;
1771         }
1772
1773         if (ret == IOU_ISSUE_SKIP_COMPLETE) {
1774                 ret = 0;
1775
1776                 /* If the op doesn't have a file, we're not polling for it */
1777                 if ((req->ctx->flags & IORING_SETUP_IOPOLL) && def->iopoll_queue)
1778                         io_iopoll_req_issued(req, issue_flags);
1779         }
1780         return ret;
1781 }
1782
1783 int io_poll_issue(struct io_kiocb *req, io_tw_token_t tw)
1784 {
1785         const unsigned int issue_flags = IO_URING_F_NONBLOCK |
1786                                          IO_URING_F_MULTISHOT |
1787                                          IO_URING_F_COMPLETE_DEFER;
1788         int ret;
1789
1790         io_tw_lock(req->ctx, tw);
1791
1792         WARN_ON_ONCE(!req->file);
1793         if (WARN_ON_ONCE(req->ctx->flags & IORING_SETUP_IOPOLL))
1794                 return -EFAULT;
1795
1796         ret = __io_issue_sqe(req, issue_flags, &io_issue_defs[req->opcode]);
1797
1798         WARN_ON_ONCE(ret == IOU_ISSUE_SKIP_COMPLETE);
1799         return ret;
1800 }
1801
1802 struct io_wq_work *io_wq_free_work(struct io_wq_work *work)
1803 {
1804         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1805         struct io_kiocb *nxt = NULL;
1806
1807         if (req_ref_put_and_test_atomic(req)) {
1808                 if (req->flags & IO_REQ_LINK_FLAGS)
1809                         nxt = io_req_find_next(req);
1810                 io_free_req(req);
1811         }
1812         return nxt ? &nxt->work : NULL;
1813 }
1814
1815 void io_wq_submit_work(struct io_wq_work *work)
1816 {
1817         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
1818         const struct io_issue_def *def = &io_issue_defs[req->opcode];
1819         unsigned int issue_flags = IO_URING_F_UNLOCKED | IO_URING_F_IOWQ;
1820         bool needs_poll = false;
1821         int ret = 0, err = -ECANCELED;
1822
1823         /* one will be dropped by io_wq_free_work() after returning to io-wq */
1824         if (!(req->flags & REQ_F_REFCOUNT))
1825                 __io_req_set_refcount(req, 2);
1826         else
1827                 req_ref_get(req);
1828
1829         /* either cancelled or io-wq is dying, so don't touch tctx->iowq */
1830         if (atomic_read(&work->flags) & IO_WQ_WORK_CANCEL) {
1831 fail:
1832                 io_req_task_queue_fail(req, err);
1833                 return;
1834         }
1835         if (!io_assign_file(req, def, issue_flags)) {
1836                 err = -EBADF;
1837                 atomic_or(IO_WQ_WORK_CANCEL, &work->flags);
1838                 goto fail;
1839         }
1840
1841         /*
1842          * If DEFER_TASKRUN is set, it's only allowed to post CQEs from the
1843          * submitter task context. Final request completions are handed to the
1844          * right context, however this is not the case of auxiliary CQEs,
1845          * which is the main mean of operation for multishot requests.
1846          * Don't allow any multishot execution from io-wq. It's more restrictive
1847          * than necessary and also cleaner.
1848          */
1849         if (req->flags & (REQ_F_MULTISHOT|REQ_F_APOLL_MULTISHOT)) {
1850                 err = -EBADFD;
1851                 if (!io_file_can_poll(req))
1852                         goto fail;
1853                 if (req->file->f_flags & O_NONBLOCK ||
1854                     req->file->f_mode & FMODE_NOWAIT) {
1855                         err = -ECANCELED;
1856                         if (io_arm_poll_handler(req, issue_flags) != IO_APOLL_OK)
1857                                 goto fail;
1858                         return;
1859                 } else {
1860                         req->flags &= ~(REQ_F_APOLL_MULTISHOT|REQ_F_MULTISHOT);
1861                 }
1862         }
1863
1864         if (req->flags & REQ_F_FORCE_ASYNC) {
1865                 bool opcode_poll = def->pollin || def->pollout;
1866
1867                 if (opcode_poll && io_file_can_poll(req)) {
1868                         needs_poll = true;
1869                         issue_flags |= IO_URING_F_NONBLOCK;
1870                 }
1871         }
1872
1873         do {
1874                 ret = io_issue_sqe(req, issue_flags);
1875                 if (ret != -EAGAIN)
1876                         break;
1877
1878                 /*
1879                  * If REQ_F_NOWAIT is set, then don't wait or retry with
1880                  * poll. -EAGAIN is final for that case.
1881                  */
1882                 if (req->flags & REQ_F_NOWAIT)
1883                         break;
1884
1885                 /*
1886                  * We can get EAGAIN for iopolled IO even though we're
1887                  * forcing a sync submission from here, since we can't
1888                  * wait for request slots on the block side.
1889                  */
1890                 if (!needs_poll) {
1891                         if (!(req->ctx->flags & IORING_SETUP_IOPOLL))
1892                                 break;
1893                         if (io_wq_worker_stopped())
1894                                 break;
1895                         cond_resched();
1896                         continue;
1897                 }
1898
1899                 if (io_arm_poll_handler(req, issue_flags) == IO_APOLL_OK)
1900                         return;
1901                 /* aborted or ready, in either case retry blocking */
1902                 needs_poll = false;
1903                 issue_flags &= ~IO_URING_F_NONBLOCK;
1904         } while (1);
1905
1906         /* avoid locking problems by failing it from a clean context */
1907         if (ret)
1908                 io_req_task_queue_fail(req, ret);
1909 }
1910
1911 inline struct file *io_file_get_fixed(struct io_kiocb *req, int fd,
1912                                       unsigned int issue_flags)
1913 {
1914         struct io_ring_ctx *ctx = req->ctx;
1915         struct io_rsrc_node *node;
1916         struct file *file = NULL;
1917
1918         io_ring_submit_lock(ctx, issue_flags);
1919         node = io_rsrc_node_lookup(&ctx->file_table.data, fd);
1920         if (node) {
1921                 node->refs++;
1922                 req->file_node = node;
1923                 req->flags |= io_slot_flags(node);
1924                 file = io_slot_file(node);
1925         }
1926         io_ring_submit_unlock(ctx, issue_flags);
1927         return file;
1928 }
1929
1930 struct file *io_file_get_normal(struct io_kiocb *req, int fd)
1931 {
1932         struct file *file = fget(fd);
1933
1934         trace_io_uring_file_get(req, fd);
1935
1936         /* we don't allow fixed io_uring files */
1937         if (file && io_is_uring_fops(file))
1938                 io_req_track_inflight(req);
1939         return file;
1940 }
1941
1942 static void io_queue_async(struct io_kiocb *req, int ret)
1943         __must_hold(&req->ctx->uring_lock)
1944 {
1945         if (ret != -EAGAIN || (req->flags & REQ_F_NOWAIT)) {
1946                 io_req_defer_failed(req, ret);
1947                 return;
1948         }
1949
1950         switch (io_arm_poll_handler(req, 0)) {
1951         case IO_APOLL_READY:
1952                 io_kbuf_recycle(req, 0);
1953                 io_req_task_queue(req);
1954                 break;
1955         case IO_APOLL_ABORTED:
1956                 io_kbuf_recycle(req, 0);
1957                 io_queue_iowq(req);
1958                 break;
1959         case IO_APOLL_OK:
1960                 break;
1961         }
1962 }
1963
1964 static inline void io_queue_sqe(struct io_kiocb *req)
1965         __must_hold(&req->ctx->uring_lock)
1966 {
1967         int ret;
1968
1969         ret = io_issue_sqe(req, IO_URING_F_NONBLOCK|IO_URING_F_COMPLETE_DEFER);
1970
1971         /*
1972          * We async punt it if the file wasn't marked NOWAIT, or if the file
1973          * doesn't support non-blocking read/write attempts
1974          */
1975         if (unlikely(ret))
1976                 io_queue_async(req, ret);
1977 }
1978
1979 static void io_queue_sqe_fallback(struct io_kiocb *req)
1980         __must_hold(&req->ctx->uring_lock)
1981 {
1982         if (unlikely(req->flags & REQ_F_FAIL)) {
1983                 /*
1984                  * We don't submit, fail them all, for that replace hardlinks
1985                  * with normal links. Extra REQ_F_LINK is tolerated.
1986                  */
1987                 req->flags &= ~REQ_F_HARDLINK;
1988                 req->flags |= REQ_F_LINK;
1989                 io_req_defer_failed(req, req->cqe.res);
1990         } else {
1991                 if (unlikely(req->ctx->drain_active))
1992                         io_drain_req(req);
1993                 else
1994                         io_queue_iowq(req);
1995         }
1996 }
1997
1998 /*
1999  * Check SQE restrictions (opcode and flags).
2000  *
2001  * Returns 'true' if SQE is allowed, 'false' otherwise.
2002  */
2003 static inline bool io_check_restriction(struct io_ring_ctx *ctx,
2004                                         struct io_kiocb *req,
2005                                         unsigned int sqe_flags)
2006 {
2007         if (!test_bit(req->opcode, ctx->restrictions.sqe_op))
2008                 return false;
2009
2010         if ((sqe_flags & ctx->restrictions.sqe_flags_required) !=
2011             ctx->restrictions.sqe_flags_required)
2012                 return false;
2013
2014         if (sqe_flags & ~(ctx->restrictions.sqe_flags_allowed |
2015                           ctx->restrictions.sqe_flags_required))
2016                 return false;
2017
2018         return true;
2019 }
2020
2021 static void io_init_drain(struct io_ring_ctx *ctx)
2022 {
2023         struct io_kiocb *head = ctx->submit_state.link.head;
2024
2025         ctx->drain_active = true;
2026         if (head) {
2027                 /*
2028                  * If we need to drain a request in the middle of a link, drain
2029                  * the head request and the next request/link after the current
2030                  * link. Considering sequential execution of links,
2031                  * REQ_F_IO_DRAIN will be maintained for every request of our
2032                  * link.
2033                  */
2034                 head->flags |= REQ_F_IO_DRAIN | REQ_F_FORCE_ASYNC;
2035                 ctx->drain_next = true;
2036         }
2037 }
2038
2039 static __cold int io_init_fail_req(struct io_kiocb *req, int err)
2040 {
2041         /* ensure per-opcode data is cleared if we fail before prep */
2042         memset(&req->cmd.data, 0, sizeof(req->cmd.data));
2043         return err;
2044 }
2045
2046 static int io_init_req(struct io_ring_ctx *ctx, struct io_kiocb *req,
2047                        const struct io_uring_sqe *sqe)
2048         __must_hold(&ctx->uring_lock)
2049 {
2050         const struct io_issue_def *def;
2051         unsigned int sqe_flags;
2052         int personality;
2053         u8 opcode;
2054
2055         req->ctx = ctx;
2056         req->opcode = opcode = READ_ONCE(sqe->opcode);
2057         /* same numerical values with corresponding REQ_F_*, safe to copy */
2058         sqe_flags = READ_ONCE(sqe->flags);
2059         req->flags = (__force io_req_flags_t) sqe_flags;
2060         req->cqe.user_data = READ_ONCE(sqe->user_data);
2061         req->file = NULL;
2062         req->tctx = current->io_uring;
2063         req->cancel_seq_set = false;
2064
2065         if (unlikely(opcode >= IORING_OP_LAST)) {
2066                 req->opcode = 0;
2067                 return io_init_fail_req(req, -EINVAL);
2068         }
2069         opcode = array_index_nospec(opcode, IORING_OP_LAST);
2070
2071         def = &io_issue_defs[opcode];
2072         if (unlikely(sqe_flags & ~SQE_COMMON_FLAGS)) {
2073                 /* enforce forwards compatibility on users */
2074                 if (sqe_flags & ~SQE_VALID_FLAGS)
2075                         return io_init_fail_req(req, -EINVAL);
2076                 if (sqe_flags & IOSQE_BUFFER_SELECT) {
2077                         if (!def->buffer_select)
2078                                 return io_init_fail_req(req, -EOPNOTSUPP);
2079                         req->buf_index = READ_ONCE(sqe->buf_group);
2080                 }
2081                 if (sqe_flags & IOSQE_CQE_SKIP_SUCCESS)
2082                         ctx->drain_disabled = true;
2083                 if (sqe_flags & IOSQE_IO_DRAIN) {
2084                         if (ctx->drain_disabled)
2085                                 return io_init_fail_req(req, -EOPNOTSUPP);
2086                         io_init_drain(ctx);
2087                 }
2088         }
2089         if (unlikely(ctx->restricted || ctx->drain_active || ctx->drain_next)) {
2090                 if (ctx->restricted && !io_check_restriction(ctx, req, sqe_flags))
2091                         return io_init_fail_req(req, -EACCES);
2092                 /* knock it to the slow queue path, will be drained there */
2093                 if (ctx->drain_active)
2094                         req->flags |= REQ_F_FORCE_ASYNC;
2095                 /* if there is no link, we're at "next" request and need to drain */
2096                 if (unlikely(ctx->drain_next) && !ctx->submit_state.link.head) {
2097                         ctx->drain_next = false;
2098                         ctx->drain_active = true;
2099                         req->flags |= REQ_F_IO_DRAIN | REQ_F_FORCE_ASYNC;
2100                 }
2101         }
2102
2103         if (!def->ioprio && sqe->ioprio)
2104                 return io_init_fail_req(req, -EINVAL);
2105         if (!def->iopoll && (ctx->flags & IORING_SETUP_IOPOLL))
2106                 return io_init_fail_req(req, -EINVAL);
2107
2108         if (def->needs_file) {
2109                 struct io_submit_state *state = &ctx->submit_state;
2110
2111                 req->cqe.fd = READ_ONCE(sqe->fd);
2112
2113                 /*
2114                  * Plug now if we have more than 2 IO left after this, and the
2115                  * target is potentially a read/write to block based storage.
2116                  */
2117                 if (state->need_plug && def->plug) {
2118                         state->plug_started = true;
2119                         state->need_plug = false;
2120                         blk_start_plug_nr_ios(&state->plug, state->submit_nr);
2121                 }
2122         }
2123
2124         personality = READ_ONCE(sqe->personality);
2125         if (personality) {
2126                 int ret;
2127
2128                 req->creds = xa_load(&ctx->personalities, personality);
2129                 if (!req->creds)
2130                         return io_init_fail_req(req, -EINVAL);
2131                 get_cred(req->creds);
2132                 ret = security_uring_override_creds(req->creds);
2133                 if (ret) {
2134                         put_cred(req->creds);
2135                         return io_init_fail_req(req, ret);
2136                 }
2137                 req->flags |= REQ_F_CREDS;
2138         }
2139
2140         return def->prep(req, sqe);
2141 }
2142
2143 static __cold int io_submit_fail_init(const struct io_uring_sqe *sqe,
2144                                       struct io_kiocb *req, int ret)
2145 {
2146         struct io_ring_ctx *ctx = req->ctx;
2147         struct io_submit_link *link = &ctx->submit_state.link;
2148         struct io_kiocb *head = link->head;
2149
2150         trace_io_uring_req_failed(sqe, req, ret);
2151
2152         /*
2153          * Avoid breaking links in the middle as it renders links with SQPOLL
2154          * unusable. Instead of failing eagerly, continue assembling the link if
2155          * applicable and mark the head with REQ_F_FAIL. The link flushing code
2156          * should find the flag and handle the rest.
2157          */
2158         req_fail_link_node(req, ret);
2159         if (head && !(head->flags & REQ_F_FAIL))
2160                 req_fail_link_node(head, -ECANCELED);
2161
2162         if (!(req->flags & IO_REQ_LINK_FLAGS)) {
2163                 if (head) {
2164                         link->last->link = req;
2165                         link->head = NULL;
2166                         req = head;
2167                 }
2168                 io_queue_sqe_fallback(req);
2169                 return ret;
2170         }
2171
2172         if (head)
2173                 link->last->link = req;
2174         else
2175                 link->head = req;
2176         link->last = req;
2177         return 0;
2178 }
2179
2180 static inline int io_submit_sqe(struct io_ring_ctx *ctx, struct io_kiocb *req,
2181                          const struct io_uring_sqe *sqe)
2182         __must_hold(&ctx->uring_lock)
2183 {
2184         struct io_submit_link *link = &ctx->submit_state.link;
2185         int ret;
2186
2187         ret = io_init_req(ctx, req, sqe);
2188         if (unlikely(ret))
2189                 return io_submit_fail_init(sqe, req, ret);
2190
2191         trace_io_uring_submit_req(req);
2192
2193         /*
2194          * If we already have a head request, queue this one for async
2195          * submittal once the head completes. If we don't have a head but
2196          * IOSQE_IO_LINK is set in the sqe, start a new head. This one will be
2197          * submitted sync once the chain is complete. If none of those
2198          * conditions are true (normal request), then just queue it.
2199          */
2200         if (unlikely(link->head)) {
2201                 trace_io_uring_link(req, link->last);
2202                 link->last->link = req;
2203                 link->last = req;
2204
2205                 if (req->flags & IO_REQ_LINK_FLAGS)
2206                         return 0;
2207                 /* last request of the link, flush it */
2208                 req = link->head;
2209                 link->head = NULL;
2210                 if (req->flags & (REQ_F_FORCE_ASYNC | REQ_F_FAIL))
2211                         goto fallback;
2212
2213         } else if (unlikely(req->flags & (IO_REQ_LINK_FLAGS |
2214                                           REQ_F_FORCE_ASYNC | REQ_F_FAIL))) {
2215                 if (req->flags & IO_REQ_LINK_FLAGS) {
2216                         link->head = req;
2217                         link->last = req;
2218                 } else {
2219 fallback:
2220                         io_queue_sqe_fallback(req);
2221                 }
2222                 return 0;
2223         }
2224
2225         io_queue_sqe(req);
2226         return 0;
2227 }
2228
2229 /*
2230  * Batched submission is done, ensure local IO is flushed out.
2231  */
2232 static void io_submit_state_end(struct io_ring_ctx *ctx)
2233 {
2234         struct io_submit_state *state = &ctx->submit_state;
2235
2236         if (unlikely(state->link.head))
2237                 io_queue_sqe_fallback(state->link.head);
2238         /* flush only after queuing links as they can generate completions */
2239         io_submit_flush_completions(ctx);
2240         if (state->plug_started)
2241                 blk_finish_plug(&state->plug);
2242 }
2243
2244 /*
2245  * Start submission side cache.
2246  */
2247 static void io_submit_state_start(struct io_submit_state *state,
2248                                   unsigned int max_ios)
2249 {
2250         state->plug_started = false;
2251         state->need_plug = max_ios > 2;
2252         state->submit_nr = max_ios;
2253         /* set only head, no need to init link_last in advance */
2254         state->link.head = NULL;
2255 }
2256
2257 static void io_commit_sqring(struct io_ring_ctx *ctx)
2258 {
2259         struct io_rings *rings = ctx->rings;
2260
2261         /*
2262          * Ensure any loads from the SQEs are done at this point,
2263          * since once we write the new head, the application could
2264          * write new data to them.
2265          */
2266         smp_store_release(&rings->sq.head, ctx->cached_sq_head);
2267 }
2268
2269 /*
2270  * Fetch an sqe, if one is available. Note this returns a pointer to memory
2271  * that is mapped by userspace. This means that care needs to be taken to
2272  * ensure that reads are stable, as we cannot rely on userspace always
2273  * being a good citizen. If members of the sqe are validated and then later
2274  * used, it's important that those reads are done through READ_ONCE() to
2275  * prevent a re-load down the line.
2276  */
2277 static bool io_get_sqe(struct io_ring_ctx *ctx, const struct io_uring_sqe **sqe)
2278 {
2279         unsigned mask = ctx->sq_entries - 1;
2280         unsigned head = ctx->cached_sq_head++ & mask;
2281
2282         if (static_branch_unlikely(&io_key_has_sqarray) &&
2283             (!(ctx->flags & IORING_SETUP_NO_SQARRAY))) {
2284                 head = READ_ONCE(ctx->sq_array[head]);
2285                 if (unlikely(head >= ctx->sq_entries)) {
2286                         WRITE_ONCE(ctx->rings->sq_dropped,
2287                                    READ_ONCE(ctx->rings->sq_dropped) + 1);
2288                         return false;
2289                 }
2290                 head = array_index_nospec(head, ctx->sq_entries);
2291         }
2292
2293         /*
2294          * The cached sq head (or cq tail) serves two purposes:
2295          *
2296          * 1) allows us to batch the cost of updating the user visible
2297          *    head updates.
2298          * 2) allows the kernel side to track the head on its own, even
2299          *    though the application is the one updating it.
2300          */
2301
2302         /* double index for 128-byte SQEs, twice as long */
2303         if (ctx->flags & IORING_SETUP_SQE128)
2304                 head <<= 1;
2305         *sqe = &ctx->sq_sqes[head];
2306         return true;
2307 }
2308
2309 int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr)
2310         __must_hold(&ctx->uring_lock)
2311 {
2312         unsigned int entries = io_sqring_entries(ctx);
2313         unsigned int left;
2314         int ret;
2315
2316         if (unlikely(!entries))
2317                 return 0;
2318         /* make sure SQ entry isn't read before tail */
2319         ret = left = min(nr, entries);
2320         io_get_task_refs(left);
2321         io_submit_state_start(&ctx->submit_state, left);
2322
2323         do {
2324                 const struct io_uring_sqe *sqe;
2325                 struct io_kiocb *req;
2326
2327                 if (unlikely(!io_alloc_req(ctx, &req)))
2328                         break;
2329                 if (unlikely(!io_get_sqe(ctx, &sqe))) {
2330                         io_req_add_to_cache(req, ctx);
2331                         break;
2332                 }
2333
2334                 /*
2335                  * Continue submitting even for sqe failure if the
2336                  * ring was setup with IORING_SETUP_SUBMIT_ALL
2337                  */
2338                 if (unlikely(io_submit_sqe(ctx, req, sqe)) &&
2339                     !(ctx->flags & IORING_SETUP_SUBMIT_ALL)) {
2340                         left--;
2341                         break;
2342                 }
2343         } while (--left);
2344
2345         if (unlikely(left)) {
2346                 ret -= left;
2347                 /* try again if it submitted nothing and can't allocate a req */
2348                 if (!ret && io_req_cache_empty(ctx))
2349                         ret = -EAGAIN;
2350                 current->io_uring->cached_refs += left;
2351         }
2352
2353         io_submit_state_end(ctx);
2354          /* Commit SQ ring head once we've consumed and submitted all SQEs */
2355         io_commit_sqring(ctx);
2356         return ret;
2357 }
2358
2359 static int io_wake_function(struct wait_queue_entry *curr, unsigned int mode,
2360                             int wake_flags, void *key)
2361 {
2362         struct io_wait_queue *iowq = container_of(curr, struct io_wait_queue, wq);
2363
2364         /*
2365          * Cannot safely flush overflowed CQEs from here, ensure we wake up
2366          * the task, and the next invocation will do it.
2367          */
2368         if (io_should_wake(iowq) || io_has_work(iowq->ctx))
2369                 return autoremove_wake_function(curr, mode, wake_flags, key);
2370         return -1;
2371 }
2372
2373 int io_run_task_work_sig(struct io_ring_ctx *ctx)
2374 {
2375         if (io_local_work_pending(ctx)) {
2376                 __set_current_state(TASK_RUNNING);
2377                 if (io_run_local_work(ctx, INT_MAX, IO_LOCAL_TW_DEFAULT_MAX) > 0)
2378                         return 0;
2379         }
2380         if (io_run_task_work() > 0)
2381                 return 0;
2382         if (task_sigpending(current))
2383                 return -EINTR;
2384         return 0;
2385 }
2386
2387 static bool current_pending_io(void)
2388 {
2389         struct io_uring_task *tctx = current->io_uring;
2390
2391         if (!tctx)
2392                 return false;
2393         return percpu_counter_read_positive(&tctx->inflight);
2394 }
2395
2396 static enum hrtimer_restart io_cqring_timer_wakeup(struct hrtimer *timer)
2397 {
2398         struct io_wait_queue *iowq = container_of(timer, struct io_wait_queue, t);
2399
2400         WRITE_ONCE(iowq->hit_timeout, 1);
2401         iowq->min_timeout = 0;
2402         wake_up_process(iowq->wq.private);
2403         return HRTIMER_NORESTART;
2404 }
2405
2406 /*
2407  * Doing min_timeout portion. If we saw any timeouts, events, or have work,
2408  * wake up. If not, and we have a normal timeout, switch to that and keep
2409  * sleeping.
2410  */
2411 static enum hrtimer_restart io_cqring_min_timer_wakeup(struct hrtimer *timer)
2412 {
2413         struct io_wait_queue *iowq = container_of(timer, struct io_wait_queue, t);
2414         struct io_ring_ctx *ctx = iowq->ctx;
2415
2416         /* no general timeout, or shorter (or equal), we are done */
2417         if (iowq->timeout == KTIME_MAX ||
2418             ktime_compare(iowq->min_timeout, iowq->timeout) >= 0)
2419                 goto out_wake;
2420         /* work we may need to run, wake function will see if we need to wake */
2421         if (io_has_work(ctx))
2422                 goto out_wake;
2423         /* got events since we started waiting, min timeout is done */
2424         if (iowq->cq_min_tail != READ_ONCE(ctx->rings->cq.tail))
2425                 goto out_wake;
2426         /* if we have any events and min timeout expired, we're done */
2427         if (io_cqring_events(ctx))
2428                 goto out_wake;
2429
2430         /*
2431          * If using deferred task_work running and application is waiting on
2432          * more than one request, ensure we reset it now where we are switching
2433          * to normal sleeps. Any request completion post min_wait should wake
2434          * the task and return.
2435          */
2436         if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
2437                 atomic_set(&ctx->cq_wait_nr, 1);
2438                 smp_mb();
2439                 if (!llist_empty(&ctx->work_llist))
2440                         goto out_wake;
2441         }
2442
2443         hrtimer_update_function(&iowq->t, io_cqring_timer_wakeup);
2444         hrtimer_set_expires(timer, iowq->timeout);
2445         return HRTIMER_RESTART;
2446 out_wake:
2447         return io_cqring_timer_wakeup(timer);
2448 }
2449
2450 static int io_cqring_schedule_timeout(struct io_wait_queue *iowq,
2451                                       clockid_t clock_id, ktime_t start_time)
2452 {
2453         ktime_t timeout;
2454
2455         if (iowq->min_timeout) {
2456                 timeout = ktime_add_ns(iowq->min_timeout, start_time);
2457                 hrtimer_setup_on_stack(&iowq->t, io_cqring_min_timer_wakeup, clock_id,
2458                                        HRTIMER_MODE_ABS);
2459         } else {
2460                 timeout = iowq->timeout;
2461                 hrtimer_setup_on_stack(&iowq->t, io_cqring_timer_wakeup, clock_id,
2462                                        HRTIMER_MODE_ABS);
2463         }
2464
2465         hrtimer_set_expires_range_ns(&iowq->t, timeout, 0);
2466         hrtimer_start_expires(&iowq->t, HRTIMER_MODE_ABS);
2467
2468         if (!READ_ONCE(iowq->hit_timeout))
2469                 schedule();
2470
2471         hrtimer_cancel(&iowq->t);
2472         destroy_hrtimer_on_stack(&iowq->t);
2473         __set_current_state(TASK_RUNNING);
2474
2475         return READ_ONCE(iowq->hit_timeout) ? -ETIME : 0;
2476 }
2477
2478 struct ext_arg {
2479         size_t argsz;
2480         struct timespec64 ts;
2481         const sigset_t __user *sig;
2482         ktime_t min_time;
2483         bool ts_set;
2484         bool iowait;
2485 };
2486
2487 static int __io_cqring_wait_schedule(struct io_ring_ctx *ctx,
2488                                      struct io_wait_queue *iowq,
2489                                      struct ext_arg *ext_arg,
2490                                      ktime_t start_time)
2491 {
2492         int ret = 0;
2493
2494         /*
2495          * Mark us as being in io_wait if we have pending requests, so cpufreq
2496          * can take into account that the task is waiting for IO - turns out
2497          * to be important for low QD IO.
2498          */
2499         if (ext_arg->iowait && current_pending_io())
2500                 current->in_iowait = 1;
2501         if (iowq->timeout != KTIME_MAX || iowq->min_timeout)
2502                 ret = io_cqring_schedule_timeout(iowq, ctx->clockid, start_time);
2503         else
2504                 schedule();
2505         current->in_iowait = 0;
2506         return ret;
2507 }
2508
2509 /* If this returns > 0, the caller should retry */
2510 static inline int io_cqring_wait_schedule(struct io_ring_ctx *ctx,
2511                                           struct io_wait_queue *iowq,
2512                                           struct ext_arg *ext_arg,
2513                                           ktime_t start_time)
2514 {
2515         if (unlikely(READ_ONCE(ctx->check_cq)))
2516                 return 1;
2517         if (unlikely(io_local_work_pending(ctx)))
2518                 return 1;
2519         if (unlikely(task_work_pending(current)))
2520                 return 1;
2521         if (unlikely(task_sigpending(current)))
2522                 return -EINTR;
2523         if (unlikely(io_should_wake(iowq)))
2524                 return 0;
2525
2526         return __io_cqring_wait_schedule(ctx, iowq, ext_arg, start_time);
2527 }
2528
2529 /*
2530  * Wait until events become available, if we don't already have some. The
2531  * application must reap them itself, as they reside on the shared cq ring.
2532  */
2533 static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events, u32 flags,
2534                           struct ext_arg *ext_arg)
2535 {
2536         struct io_wait_queue iowq;
2537         struct io_rings *rings = ctx->rings;
2538         ktime_t start_time;
2539         int ret;
2540
2541         min_events = min_t(int, min_events, ctx->cq_entries);
2542
2543         if (!io_allowed_run_tw(ctx))
2544                 return -EEXIST;
2545         if (io_local_work_pending(ctx))
2546                 io_run_local_work(ctx, min_events,
2547                                   max(IO_LOCAL_TW_DEFAULT_MAX, min_events));
2548         io_run_task_work();
2549
2550         if (unlikely(test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)))
2551                 io_cqring_do_overflow_flush(ctx);
2552         if (__io_cqring_events_user(ctx) >= min_events)
2553                 return 0;
2554
2555         init_waitqueue_func_entry(&iowq.wq, io_wake_function);
2556         iowq.wq.private = current;
2557         INIT_LIST_HEAD(&iowq.wq.entry);
2558         iowq.ctx = ctx;
2559         iowq.cq_tail = READ_ONCE(ctx->rings->cq.head) + min_events;
2560         iowq.cq_min_tail = READ_ONCE(ctx->rings->cq.tail);
2561         iowq.nr_timeouts = atomic_read(&ctx->cq_timeouts);
2562         iowq.hit_timeout = 0;
2563         iowq.min_timeout = ext_arg->min_time;
2564         iowq.timeout = KTIME_MAX;
2565         start_time = io_get_time(ctx);
2566
2567         if (ext_arg->ts_set) {
2568                 iowq.timeout = timespec64_to_ktime(ext_arg->ts);
2569                 if (!(flags & IORING_ENTER_ABS_TIMER))
2570                         iowq.timeout = ktime_add(iowq.timeout, start_time);
2571         }
2572
2573         if (ext_arg->sig) {
2574 #ifdef CONFIG_COMPAT
2575                 if (in_compat_syscall())
2576                         ret = set_compat_user_sigmask((const compat_sigset_t __user *)ext_arg->sig,
2577                                                       ext_arg->argsz);
2578                 else
2579 #endif
2580                         ret = set_user_sigmask(ext_arg->sig, ext_arg->argsz);
2581
2582                 if (ret)
2583                         return ret;
2584         }
2585
2586         io_napi_busy_loop(ctx, &iowq);
2587
2588         trace_io_uring_cqring_wait(ctx, min_events);
2589         do {
2590                 unsigned long check_cq;
2591                 int nr_wait;
2592
2593                 /* if min timeout has been hit, don't reset wait count */
2594                 if (!iowq.hit_timeout)
2595                         nr_wait = (int) iowq.cq_tail -
2596                                         READ_ONCE(ctx->rings->cq.tail);
2597                 else
2598                         nr_wait = 1;
2599
2600                 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
2601                         atomic_set(&ctx->cq_wait_nr, nr_wait);
2602                         set_current_state(TASK_INTERRUPTIBLE);
2603                 } else {
2604                         prepare_to_wait_exclusive(&ctx->cq_wait, &iowq.wq,
2605                                                         TASK_INTERRUPTIBLE);
2606                 }
2607
2608                 ret = io_cqring_wait_schedule(ctx, &iowq, ext_arg, start_time);
2609                 __set_current_state(TASK_RUNNING);
2610                 atomic_set(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT);
2611
2612                 /*
2613                  * Run task_work after scheduling and before io_should_wake().
2614                  * If we got woken because of task_work being processed, run it
2615                  * now rather than let the caller do another wait loop.
2616                  */
2617                 if (io_local_work_pending(ctx))
2618                         io_run_local_work(ctx, nr_wait, nr_wait);
2619                 io_run_task_work();
2620
2621                 /*
2622                  * Non-local task_work will be run on exit to userspace, but
2623                  * if we're using DEFER_TASKRUN, then we could have waited
2624                  * with a timeout for a number of requests. If the timeout
2625                  * hits, we could have some requests ready to process. Ensure
2626                  * this break is _after_ we have run task_work, to avoid
2627                  * deferring running potentially pending requests until the
2628                  * next time we wait for events.
2629                  */
2630                 if (ret < 0)
2631                         break;
2632
2633                 check_cq = READ_ONCE(ctx->check_cq);
2634                 if (unlikely(check_cq)) {
2635                         /* let the caller flush overflows, retry */
2636                         if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT))
2637                                 io_cqring_do_overflow_flush(ctx);
2638                         if (check_cq & BIT(IO_CHECK_CQ_DROPPED_BIT)) {
2639                                 ret = -EBADR;
2640                                 break;
2641                         }
2642                 }
2643
2644                 if (io_should_wake(&iowq)) {
2645                         ret = 0;
2646                         break;
2647                 }
2648                 cond_resched();
2649         } while (1);
2650
2651         if (!(ctx->flags & IORING_SETUP_DEFER_TASKRUN))
2652                 finish_wait(&ctx->cq_wait, &iowq.wq);
2653         restore_saved_sigmask_unless(ret == -EINTR);
2654
2655         return READ_ONCE(rings->cq.head) == READ_ONCE(rings->cq.tail) ? ret : 0;
2656 }
2657
2658 static void io_rings_free(struct io_ring_ctx *ctx)
2659 {
2660         io_free_region(ctx, &ctx->sq_region);
2661         io_free_region(ctx, &ctx->ring_region);
2662         ctx->rings = NULL;
2663         ctx->sq_sqes = NULL;
2664 }
2665
2666 unsigned long rings_size(unsigned int flags, unsigned int sq_entries,
2667                          unsigned int cq_entries, size_t *sq_offset)
2668 {
2669         struct io_rings *rings;
2670         size_t off, sq_array_size;
2671
2672         off = struct_size(rings, cqes, cq_entries);
2673         if (off == SIZE_MAX)
2674                 return SIZE_MAX;
2675         if (flags & IORING_SETUP_CQE32) {
2676                 if (check_shl_overflow(off, 1, &off))
2677                         return SIZE_MAX;
2678         }
2679
2680 #ifdef CONFIG_SMP
2681         off = ALIGN(off, SMP_CACHE_BYTES);
2682         if (off == 0)
2683                 return SIZE_MAX;
2684 #endif
2685
2686         if (flags & IORING_SETUP_NO_SQARRAY) {
2687                 *sq_offset = SIZE_MAX;
2688                 return off;
2689         }
2690
2691         *sq_offset = off;
2692
2693         sq_array_size = array_size(sizeof(u32), sq_entries);
2694         if (sq_array_size == SIZE_MAX)
2695                 return SIZE_MAX;
2696
2697         if (check_add_overflow(off, sq_array_size, &off))
2698                 return SIZE_MAX;
2699
2700         return off;
2701 }
2702
2703 static __cold void __io_req_caches_free(struct io_ring_ctx *ctx)
2704 {
2705         struct io_kiocb *req;
2706         int nr = 0;
2707
2708         while (!io_req_cache_empty(ctx)) {
2709                 req = io_extract_req(ctx);
2710                 kmem_cache_free(req_cachep, req);
2711                 nr++;
2712         }
2713         if (nr) {
2714                 ctx->nr_req_allocated -= nr;
2715                 percpu_ref_put_many(&ctx->refs, nr);
2716         }
2717 }
2718
2719 static __cold void io_req_caches_free(struct io_ring_ctx *ctx)
2720 {
2721         guard(mutex)(&ctx->uring_lock);
2722         __io_req_caches_free(ctx);
2723 }
2724
2725 static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx)
2726 {
2727         io_sq_thread_finish(ctx);
2728
2729         mutex_lock(&ctx->uring_lock);
2730         io_sqe_buffers_unregister(ctx);
2731         io_sqe_files_unregister(ctx);
2732         io_unregister_zcrx_ifqs(ctx);
2733         io_cqring_overflow_kill(ctx);
2734         io_eventfd_unregister(ctx);
2735         io_free_alloc_caches(ctx);
2736         io_destroy_buffers(ctx);
2737         io_free_region(ctx, &ctx->param_region);
2738         mutex_unlock(&ctx->uring_lock);
2739         if (ctx->sq_creds)
2740                 put_cred(ctx->sq_creds);
2741         if (ctx->submitter_task)
2742                 put_task_struct(ctx->submitter_task);
2743
2744         WARN_ON_ONCE(!list_empty(&ctx->ltimeout_list));
2745
2746         if (ctx->mm_account) {
2747                 mmdrop(ctx->mm_account);
2748                 ctx->mm_account = NULL;
2749         }
2750         io_rings_free(ctx);
2751
2752         if (!(ctx->flags & IORING_SETUP_NO_SQARRAY))
2753                 static_branch_dec(&io_key_has_sqarray);
2754
2755         percpu_ref_exit(&ctx->refs);
2756         free_uid(ctx->user);
2757         io_req_caches_free(ctx);
2758
2759         WARN_ON_ONCE(ctx->nr_req_allocated);
2760
2761         if (ctx->hash_map)
2762                 io_wq_put_hash(ctx->hash_map);
2763         io_napi_free(ctx);
2764         kvfree(ctx->cancel_table.hbs);
2765         xa_destroy(&ctx->io_bl_xa);
2766         kfree(ctx);
2767 }
2768
2769 static __cold void io_activate_pollwq_cb(struct callback_head *cb)
2770 {
2771         struct io_ring_ctx *ctx = container_of(cb, struct io_ring_ctx,
2772                                                poll_wq_task_work);
2773
2774         mutex_lock(&ctx->uring_lock);
2775         ctx->poll_activated = true;
2776         mutex_unlock(&ctx->uring_lock);
2777
2778         /*
2779          * Wake ups for some events between start of polling and activation
2780          * might've been lost due to loose synchronisation.
2781          */
2782         wake_up_all(&ctx->poll_wq);
2783         percpu_ref_put(&ctx->refs);
2784 }
2785
2786 __cold void io_activate_pollwq(struct io_ring_ctx *ctx)
2787 {
2788         spin_lock(&ctx->completion_lock);
2789         /* already activated or in progress */
2790         if (ctx->poll_activated || ctx->poll_wq_task_work.func)
2791                 goto out;
2792         if (WARN_ON_ONCE(!ctx->task_complete))
2793                 goto out;
2794         if (!ctx->submitter_task)
2795                 goto out;
2796         /*
2797          * with ->submitter_task only the submitter task completes requests, we
2798          * only need to sync with it, which is done by injecting a tw
2799          */
2800         init_task_work(&ctx->poll_wq_task_work, io_activate_pollwq_cb);
2801         percpu_ref_get(&ctx->refs);
2802         if (task_work_add(ctx->submitter_task, &ctx->poll_wq_task_work, TWA_SIGNAL))
2803                 percpu_ref_put(&ctx->refs);
2804 out:
2805         spin_unlock(&ctx->completion_lock);
2806 }
2807
2808 static __poll_t io_uring_poll(struct file *file, poll_table *wait)
2809 {
2810         struct io_ring_ctx *ctx = file->private_data;
2811         __poll_t mask = 0;
2812
2813         if (unlikely(!ctx->poll_activated))
2814                 io_activate_pollwq(ctx);
2815         /*
2816          * provides mb() which pairs with barrier from wq_has_sleeper
2817          * call in io_commit_cqring
2818          */
2819         poll_wait(file, &ctx->poll_wq, wait);
2820
2821         if (!io_sqring_full(ctx))
2822                 mask |= EPOLLOUT | EPOLLWRNORM;
2823
2824         /*
2825          * Don't flush cqring overflow list here, just do a simple check.
2826          * Otherwise there could possible be ABBA deadlock:
2827          *      CPU0                    CPU1
2828          *      ----                    ----
2829          * lock(&ctx->uring_lock);
2830          *                              lock(&ep->mtx);
2831          *                              lock(&ctx->uring_lock);
2832          * lock(&ep->mtx);
2833          *
2834          * Users may get EPOLLIN meanwhile seeing nothing in cqring, this
2835          * pushes them to do the flush.
2836          */
2837
2838         if (__io_cqring_events_user(ctx) || io_has_work(ctx))
2839                 mask |= EPOLLIN | EPOLLRDNORM;
2840
2841         return mask;
2842 }
2843
2844 struct io_tctx_exit {
2845         struct callback_head            task_work;
2846         struct completion               completion;
2847         struct io_ring_ctx              *ctx;
2848 };
2849
2850 static __cold void io_tctx_exit_cb(struct callback_head *cb)
2851 {
2852         struct io_uring_task *tctx = current->io_uring;
2853         struct io_tctx_exit *work;
2854
2855         work = container_of(cb, struct io_tctx_exit, task_work);
2856         /*
2857          * When @in_cancel, we're in cancellation and it's racy to remove the
2858          * node. It'll be removed by the end of cancellation, just ignore it.
2859          * tctx can be NULL if the queueing of this task_work raced with
2860          * work cancelation off the exec path.
2861          */
2862         if (tctx && !atomic_read(&tctx->in_cancel))
2863                 io_uring_del_tctx_node((unsigned long)work->ctx);
2864         complete(&work->completion);
2865 }
2866
2867 static __cold bool io_cancel_ctx_cb(struct io_wq_work *work, void *data)
2868 {
2869         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
2870
2871         return req->ctx == data;
2872 }
2873
2874 static __cold void io_ring_exit_work(struct work_struct *work)
2875 {
2876         struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx, exit_work);
2877         unsigned long timeout = jiffies + HZ * 60 * 5;
2878         unsigned long interval = HZ / 20;
2879         struct io_tctx_exit exit;
2880         struct io_tctx_node *node;
2881         int ret;
2882
2883         /*
2884          * If we're doing polled IO and end up having requests being
2885          * submitted async (out-of-line), then completions can come in while
2886          * we're waiting for refs to drop. We need to reap these manually,
2887          * as nobody else will be looking for them.
2888          */
2889         do {
2890                 if (test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)) {
2891                         mutex_lock(&ctx->uring_lock);
2892                         io_cqring_overflow_kill(ctx);
2893                         mutex_unlock(&ctx->uring_lock);
2894                 }
2895                 if (!xa_empty(&ctx->zcrx_ctxs)) {
2896                         mutex_lock(&ctx->uring_lock);
2897                         io_shutdown_zcrx_ifqs(ctx);
2898                         mutex_unlock(&ctx->uring_lock);
2899                 }
2900
2901                 if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
2902                         io_move_task_work_from_local(ctx);
2903
2904                 /* The SQPOLL thread never reaches this path */
2905                 while (io_uring_try_cancel_requests(ctx, NULL, true, false))
2906                         cond_resched();
2907
2908                 if (ctx->sq_data) {
2909                         struct io_sq_data *sqd = ctx->sq_data;
2910                         struct task_struct *tsk;
2911
2912                         io_sq_thread_park(sqd);
2913                         tsk = sqpoll_task_locked(sqd);
2914                         if (tsk && tsk->io_uring && tsk->io_uring->io_wq)
2915                                 io_wq_cancel_cb(tsk->io_uring->io_wq,
2916                                                 io_cancel_ctx_cb, ctx, true);
2917                         io_sq_thread_unpark(sqd);
2918                 }
2919
2920                 io_req_caches_free(ctx);
2921
2922                 if (WARN_ON_ONCE(time_after(jiffies, timeout))) {
2923                         /* there is little hope left, don't run it too often */
2924                         interval = HZ * 60;
2925                 }
2926                 /*
2927                  * This is really an uninterruptible wait, as it has to be
2928                  * complete. But it's also run from a kworker, which doesn't
2929                  * take signals, so it's fine to make it interruptible. This
2930                  * avoids scenarios where we knowingly can wait much longer
2931                  * on completions, for example if someone does a SIGSTOP on
2932                  * a task that needs to finish task_work to make this loop
2933                  * complete. That's a synthetic situation that should not
2934                  * cause a stuck task backtrace, and hence a potential panic
2935                  * on stuck tasks if that is enabled.
2936                  */
2937         } while (!wait_for_completion_interruptible_timeout(&ctx->ref_comp, interval));
2938
2939         init_completion(&exit.completion);
2940         init_task_work(&exit.task_work, io_tctx_exit_cb);
2941         exit.ctx = ctx;
2942
2943         mutex_lock(&ctx->uring_lock);
2944         while (!list_empty(&ctx->tctx_list)) {
2945                 WARN_ON_ONCE(time_after(jiffies, timeout));
2946
2947                 node = list_first_entry(&ctx->tctx_list, struct io_tctx_node,
2948                                         ctx_node);
2949                 /* don't spin on a single task if cancellation failed */
2950                 list_rotate_left(&ctx->tctx_list);
2951                 ret = task_work_add(node->task, &exit.task_work, TWA_SIGNAL);
2952                 if (WARN_ON_ONCE(ret))
2953                         continue;
2954
2955                 mutex_unlock(&ctx->uring_lock);
2956                 /*
2957                  * See comment above for
2958                  * wait_for_completion_interruptible_timeout() on why this
2959                  * wait is marked as interruptible.
2960                  */
2961                 wait_for_completion_interruptible(&exit.completion);
2962                 mutex_lock(&ctx->uring_lock);
2963         }
2964         mutex_unlock(&ctx->uring_lock);
2965         spin_lock(&ctx->completion_lock);
2966         spin_unlock(&ctx->completion_lock);
2967
2968         /* pairs with RCU read section in io_req_local_work_add() */
2969         if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
2970                 synchronize_rcu();
2971
2972         io_ring_ctx_free(ctx);
2973 }
2974
2975 static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
2976 {
2977         unsigned long index;
2978         struct creds *creds;
2979
2980         mutex_lock(&ctx->uring_lock);
2981         percpu_ref_kill(&ctx->refs);
2982         xa_for_each(&ctx->personalities, index, creds)
2983                 io_unregister_personality(ctx, index);
2984         mutex_unlock(&ctx->uring_lock);
2985
2986         flush_delayed_work(&ctx->fallback_work);
2987
2988         INIT_WORK(&ctx->exit_work, io_ring_exit_work);
2989         /*
2990          * Use system_unbound_wq to avoid spawning tons of event kworkers
2991          * if we're exiting a ton of rings at the same time. It just adds
2992          * noise and overhead, there's no discernable change in runtime
2993          * over using system_wq.
2994          */
2995         queue_work(iou_wq, &ctx->exit_work);
2996 }
2997
2998 static int io_uring_release(struct inode *inode, struct file *file)
2999 {
3000         struct io_ring_ctx *ctx = file->private_data;
3001
3002         file->private_data = NULL;
3003         io_ring_ctx_wait_and_kill(ctx);
3004         return 0;
3005 }
3006
3007 struct io_task_cancel {
3008         struct io_uring_task *tctx;
3009         bool all;
3010 };
3011
3012 static bool io_cancel_task_cb(struct io_wq_work *work, void *data)
3013 {
3014         struct io_kiocb *req = container_of(work, struct io_kiocb, work);
3015         struct io_task_cancel *cancel = data;
3016
3017         return io_match_task_safe(req, cancel->tctx, cancel->all);
3018 }
3019
3020 static __cold bool io_cancel_defer_files(struct io_ring_ctx *ctx,
3021                                          struct io_uring_task *tctx,
3022                                          bool cancel_all)
3023 {
3024         struct io_defer_entry *de;
3025         LIST_HEAD(list);
3026
3027         list_for_each_entry_reverse(de, &ctx->defer_list, list) {
3028                 if (io_match_task_safe(de->req, tctx, cancel_all)) {
3029                         list_cut_position(&list, &ctx->defer_list, &de->list);
3030                         break;
3031                 }
3032         }
3033         if (list_empty(&list))
3034                 return false;
3035
3036         while (!list_empty(&list)) {
3037                 de = list_first_entry(&list, struct io_defer_entry, list);
3038                 list_del_init(&de->list);
3039                 ctx->nr_drained -= io_linked_nr(de->req);
3040                 io_req_task_queue_fail(de->req, -ECANCELED);
3041                 kfree(de);
3042         }
3043         return true;
3044 }
3045
3046 static __cold bool io_uring_try_cancel_iowq(struct io_ring_ctx *ctx)
3047 {
3048         struct io_tctx_node *node;
3049         enum io_wq_cancel cret;
3050         bool ret = false;
3051
3052         mutex_lock(&ctx->uring_lock);
3053         list_for_each_entry(node, &ctx->tctx_list, ctx_node) {
3054                 struct io_uring_task *tctx = node->task->io_uring;
3055
3056                 /*
3057                  * io_wq will stay alive while we hold uring_lock, because it's
3058                  * killed after ctx nodes, which requires to take the lock.
3059                  */
3060                 if (!tctx || !tctx->io_wq)
3061                         continue;
3062                 cret = io_wq_cancel_cb(tctx->io_wq, io_cancel_ctx_cb, ctx, true);
3063                 ret |= (cret != IO_WQ_CANCEL_NOTFOUND);
3064         }
3065         mutex_unlock(&ctx->uring_lock);
3066
3067         return ret;
3068 }
3069
3070 static __cold bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
3071                                                 struct io_uring_task *tctx,
3072                                                 bool cancel_all,
3073                                                 bool is_sqpoll_thread)
3074 {
3075         struct io_task_cancel cancel = { .tctx = tctx, .all = cancel_all, };
3076         enum io_wq_cancel cret;
3077         bool ret = false;
3078
3079         /* set it so io_req_local_work_add() would wake us up */
3080         if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
3081                 atomic_set(&ctx->cq_wait_nr, 1);
3082                 smp_mb();
3083         }
3084
3085         /* failed during ring init, it couldn't have issued any requests */
3086         if (!ctx->rings)
3087                 return false;
3088
3089         if (!tctx) {
3090                 ret |= io_uring_try_cancel_iowq(ctx);
3091         } else if (tctx->io_wq) {
3092                 /*
3093                  * Cancels requests of all rings, not only @ctx, but
3094                  * it's fine as the task is in exit/exec.
3095                  */
3096                 cret = io_wq_cancel_cb(tctx->io_wq, io_cancel_task_cb,
3097                                        &cancel, true);
3098                 ret |= (cret != IO_WQ_CANCEL_NOTFOUND);
3099         }
3100
3101         /* SQPOLL thread does its own polling */
3102         if ((!(ctx->flags & IORING_SETUP_SQPOLL) && cancel_all) ||
3103             is_sqpoll_thread) {
3104                 while (!wq_list_empty(&ctx->iopoll_list)) {
3105                         io_iopoll_try_reap_events(ctx);
3106                         ret = true;
3107                         cond_resched();
3108                 }
3109         }
3110
3111         if ((ctx->flags & IORING_SETUP_DEFER_TASKRUN) &&
3112             io_allowed_defer_tw_run(ctx))
3113                 ret |= io_run_local_work(ctx, INT_MAX, INT_MAX) > 0;
3114         mutex_lock(&ctx->uring_lock);
3115         ret |= io_cancel_defer_files(ctx, tctx, cancel_all);
3116         ret |= io_poll_remove_all(ctx, tctx, cancel_all);
3117         ret |= io_waitid_remove_all(ctx, tctx, cancel_all);
3118         ret |= io_futex_remove_all(ctx, tctx, cancel_all);
3119         ret |= io_uring_try_cancel_uring_cmd(ctx, tctx, cancel_all);
3120         mutex_unlock(&ctx->uring_lock);
3121         ret |= io_kill_timeouts(ctx, tctx, cancel_all);
3122         if (tctx)
3123                 ret |= io_run_task_work() > 0;
3124         else
3125                 ret |= flush_delayed_work(&ctx->fallback_work);
3126         return ret;
3127 }
3128
3129 static s64 tctx_inflight(struct io_uring_task *tctx, bool tracked)
3130 {
3131         if (tracked)
3132                 return atomic_read(&tctx->inflight_tracked);
3133         return percpu_counter_sum(&tctx->inflight);
3134 }
3135
3136 /*
3137  * Find any io_uring ctx that this task has registered or done IO on, and cancel
3138  * requests. @sqd should be not-null IFF it's an SQPOLL thread cancellation.
3139  */
3140 __cold void io_uring_cancel_generic(bool cancel_all, struct io_sq_data *sqd)
3141 {
3142         struct io_uring_task *tctx = current->io_uring;
3143         struct io_ring_ctx *ctx;
3144         struct io_tctx_node *node;
3145         unsigned long index;
3146         s64 inflight;
3147         DEFINE_WAIT(wait);
3148
3149         WARN_ON_ONCE(sqd && sqpoll_task_locked(sqd) != current);
3150
3151         if (!current->io_uring)
3152                 return;
3153         if (tctx->io_wq)
3154                 io_wq_exit_start(tctx->io_wq);
3155
3156         atomic_inc(&tctx->in_cancel);
3157         do {
3158                 bool loop = false;
3159
3160                 io_uring_drop_tctx_refs(current);
3161                 if (!tctx_inflight(tctx, !cancel_all))
3162                         break;
3163
3164                 /* read completions before cancelations */
3165                 inflight = tctx_inflight(tctx, false);
3166                 if (!inflight)
3167                         break;
3168
3169                 if (!sqd) {
3170                         xa_for_each(&tctx->xa, index, node) {
3171                                 /* sqpoll task will cancel all its requests */
3172                                 if (node->ctx->sq_data)
3173                                         continue;
3174                                 loop |= io_uring_try_cancel_requests(node->ctx,
3175                                                         current->io_uring,
3176                                                         cancel_all,
3177                                                         false);
3178                         }
3179                 } else {
3180                         list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)
3181                                 loop |= io_uring_try_cancel_requests(ctx,
3182                                                                      current->io_uring,
3183                                                                      cancel_all,
3184                                                                      true);
3185                 }
3186
3187                 if (loop) {
3188                         cond_resched();
3189                         continue;
3190                 }
3191
3192                 prepare_to_wait(&tctx->wait, &wait, TASK_INTERRUPTIBLE);
3193                 io_run_task_work();
3194                 io_uring_drop_tctx_refs(current);
3195                 xa_for_each(&tctx->xa, index, node) {
3196                         if (io_local_work_pending(node->ctx)) {
3197                                 WARN_ON_ONCE(node->ctx->submitter_task &&
3198                                              node->ctx->submitter_task != current);
3199                                 goto end_wait;
3200                         }
3201                 }
3202                 /*
3203                  * If we've seen completions, retry without waiting. This
3204                  * avoids a race where a completion comes in before we did
3205                  * prepare_to_wait().
3206                  */
3207                 if (inflight == tctx_inflight(tctx, !cancel_all))
3208                         schedule();
3209 end_wait:
3210                 finish_wait(&tctx->wait, &wait);
3211         } while (1);
3212
3213         io_uring_clean_tctx(tctx);
3214         if (cancel_all) {
3215                 /*
3216                  * We shouldn't run task_works after cancel, so just leave
3217                  * ->in_cancel set for normal exit.
3218                  */
3219                 atomic_dec(&tctx->in_cancel);
3220                 /* for exec all current's requests should be gone, kill tctx */
3221                 __io_uring_free(current);
3222         }
3223 }
3224
3225 void __io_uring_cancel(bool cancel_all)
3226 {
3227         io_uring_unreg_ringfd();
3228         io_uring_cancel_generic(cancel_all, NULL);
3229 }
3230
3231 static struct io_uring_reg_wait *io_get_ext_arg_reg(struct io_ring_ctx *ctx,
3232                         const struct io_uring_getevents_arg __user *uarg)
3233 {
3234         unsigned long size = sizeof(struct io_uring_reg_wait);
3235         unsigned long offset = (uintptr_t)uarg;
3236         unsigned long end;
3237
3238         if (unlikely(offset % sizeof(long)))
3239                 return ERR_PTR(-EFAULT);
3240
3241         /* also protects from NULL ->cq_wait_arg as the size would be 0 */
3242         if (unlikely(check_add_overflow(offset, size, &end) ||
3243                      end > ctx->cq_wait_size))
3244                 return ERR_PTR(-EFAULT);
3245
3246         offset = array_index_nospec(offset, ctx->cq_wait_size - size);
3247         return ctx->cq_wait_arg + offset;
3248 }
3249
3250 static int io_validate_ext_arg(struct io_ring_ctx *ctx, unsigned flags,
3251                                const void __user *argp, size_t argsz)
3252 {
3253         struct io_uring_getevents_arg arg;
3254
3255         if (!(flags & IORING_ENTER_EXT_ARG))
3256                 return 0;
3257         if (flags & IORING_ENTER_EXT_ARG_REG)
3258                 return -EINVAL;
3259         if (argsz != sizeof(arg))
3260                 return -EINVAL;
3261         if (copy_from_user(&arg, argp, sizeof(arg)))
3262                 return -EFAULT;
3263         return 0;
3264 }
3265
3266 static int io_get_ext_arg(struct io_ring_ctx *ctx, unsigned flags,
3267                           const void __user *argp, struct ext_arg *ext_arg)
3268 {
3269         const struct io_uring_getevents_arg __user *uarg = argp;
3270         struct io_uring_getevents_arg arg;
3271
3272         ext_arg->iowait = !(flags & IORING_ENTER_NO_IOWAIT);
3273
3274         /*
3275          * If EXT_ARG isn't set, then we have no timespec and the argp pointer
3276          * is just a pointer to the sigset_t.
3277          */
3278         if (!(flags & IORING_ENTER_EXT_ARG)) {
3279                 ext_arg->sig = (const sigset_t __user *) argp;
3280                 return 0;
3281         }
3282
3283         if (flags & IORING_ENTER_EXT_ARG_REG) {
3284                 struct io_uring_reg_wait *w;
3285
3286                 if (ext_arg->argsz != sizeof(struct io_uring_reg_wait))
3287                         return -EINVAL;
3288                 w = io_get_ext_arg_reg(ctx, argp);
3289                 if (IS_ERR(w))
3290                         return PTR_ERR(w);
3291
3292                 if (w->flags & ~IORING_REG_WAIT_TS)
3293                         return -EINVAL;
3294                 ext_arg->min_time = READ_ONCE(w->min_wait_usec) * NSEC_PER_USEC;
3295                 ext_arg->sig = u64_to_user_ptr(READ_ONCE(w->sigmask));
3296                 ext_arg->argsz = READ_ONCE(w->sigmask_sz);
3297                 if (w->flags & IORING_REG_WAIT_TS) {
3298                         ext_arg->ts.tv_sec = READ_ONCE(w->ts.tv_sec);
3299                         ext_arg->ts.tv_nsec = READ_ONCE(w->ts.tv_nsec);
3300                         ext_arg->ts_set = true;
3301                 }
3302                 return 0;
3303         }
3304
3305         /*
3306          * EXT_ARG is set - ensure we agree on the size of it and copy in our
3307          * timespec and sigset_t pointers if good.
3308          */
3309         if (ext_arg->argsz != sizeof(arg))
3310                 return -EINVAL;
3311 #ifdef CONFIG_64BIT
3312         if (!user_access_begin(uarg, sizeof(*uarg)))
3313                 return -EFAULT;
3314         unsafe_get_user(arg.sigmask, &uarg->sigmask, uaccess_end);
3315         unsafe_get_user(arg.sigmask_sz, &uarg->sigmask_sz, uaccess_end);
3316         unsafe_get_user(arg.min_wait_usec, &uarg->min_wait_usec, uaccess_end);
3317         unsafe_get_user(arg.ts, &uarg->ts, uaccess_end);
3318         user_access_end();
3319 #else
3320         if (copy_from_user(&arg, uarg, sizeof(arg)))
3321                 return -EFAULT;
3322 #endif
3323         ext_arg->min_time = arg.min_wait_usec * NSEC_PER_USEC;
3324         ext_arg->sig = u64_to_user_ptr(arg.sigmask);
3325         ext_arg->argsz = arg.sigmask_sz;
3326         if (arg.ts) {
3327                 if (get_timespec64(&ext_arg->ts, u64_to_user_ptr(arg.ts)))
3328                         return -EFAULT;
3329                 ext_arg->ts_set = true;
3330         }
3331         return 0;
3332 #ifdef CONFIG_64BIT
3333 uaccess_end:
3334         user_access_end();
3335         return -EFAULT;
3336 #endif
3337 }
3338
3339 SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
3340                 u32, min_complete, u32, flags, const void __user *, argp,
3341                 size_t, argsz)
3342 {
3343         struct io_ring_ctx *ctx;
3344         struct file *file;
3345         long ret;
3346
3347         if (unlikely(flags & ~(IORING_ENTER_GETEVENTS | IORING_ENTER_SQ_WAKEUP |
3348                                IORING_ENTER_SQ_WAIT | IORING_ENTER_EXT_ARG |
3349                                IORING_ENTER_REGISTERED_RING |
3350                                IORING_ENTER_ABS_TIMER |
3351                                IORING_ENTER_EXT_ARG_REG |
3352                                IORING_ENTER_NO_IOWAIT)))
3353                 return -EINVAL;
3354
3355         /*
3356          * Ring fd has been registered via IORING_REGISTER_RING_FDS, we
3357          * need only dereference our task private array to find it.
3358          */
3359         if (flags & IORING_ENTER_REGISTERED_RING) {
3360                 struct io_uring_task *tctx = current->io_uring;
3361
3362                 if (unlikely(!tctx || fd >= IO_RINGFD_REG_MAX))
3363                         return -EINVAL;
3364                 fd = array_index_nospec(fd, IO_RINGFD_REG_MAX);
3365                 file = tctx->registered_rings[fd];
3366                 if (unlikely(!file))
3367                         return -EBADF;
3368         } else {
3369                 file = fget(fd);
3370                 if (unlikely(!file))
3371                         return -EBADF;
3372                 ret = -EOPNOTSUPP;
3373                 if (unlikely(!io_is_uring_fops(file)))
3374                         goto out;
3375         }
3376
3377         ctx = file->private_data;
3378         ret = -EBADFD;
3379         if (unlikely(ctx->flags & IORING_SETUP_R_DISABLED))
3380                 goto out;
3381
3382         /*
3383          * For SQ polling, the thread will do all submissions and completions.
3384          * Just return the requested submit count, and wake the thread if
3385          * we were asked to.
3386          */
3387         ret = 0;
3388         if (ctx->flags & IORING_SETUP_SQPOLL) {
3389                 if (unlikely(ctx->sq_data->thread == NULL)) {
3390                         ret = -EOWNERDEAD;
3391                         goto out;
3392                 }
3393                 if (flags & IORING_ENTER_SQ_WAKEUP)
3394                         wake_up(&ctx->sq_data->wait);
3395                 if (flags & IORING_ENTER_SQ_WAIT)
3396                         io_sqpoll_wait_sq(ctx);
3397
3398                 ret = to_submit;
3399         } else if (to_submit) {
3400                 ret = io_uring_add_tctx_node(ctx);
3401                 if (unlikely(ret))
3402                         goto out;
3403
3404                 mutex_lock(&ctx->uring_lock);
3405                 ret = io_submit_sqes(ctx, to_submit);
3406                 if (ret != to_submit) {
3407                         mutex_unlock(&ctx->uring_lock);
3408                         goto out;
3409                 }
3410                 if (flags & IORING_ENTER_GETEVENTS) {
3411                         if (ctx->syscall_iopoll)
3412                                 goto iopoll_locked;
3413                         /*
3414                          * Ignore errors, we'll soon call io_cqring_wait() and
3415                          * it should handle ownership problems if any.
3416                          */
3417                         if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
3418                                 (void)io_run_local_work_locked(ctx, min_complete);
3419                 }
3420                 mutex_unlock(&ctx->uring_lock);
3421         }
3422
3423         if (flags & IORING_ENTER_GETEVENTS) {
3424                 int ret2;
3425
3426                 if (ctx->syscall_iopoll) {
3427                         /*
3428                          * We disallow the app entering submit/complete with
3429                          * polling, but we still need to lock the ring to
3430                          * prevent racing with polled issue that got punted to
3431                          * a workqueue.
3432                          */
3433                         mutex_lock(&ctx->uring_lock);
3434 iopoll_locked:
3435                         ret2 = io_validate_ext_arg(ctx, flags, argp, argsz);
3436                         if (likely(!ret2))
3437                                 ret2 = io_iopoll_check(ctx, min_complete);
3438                         mutex_unlock(&ctx->uring_lock);
3439                 } else {
3440                         struct ext_arg ext_arg = { .argsz = argsz };
3441
3442                         ret2 = io_get_ext_arg(ctx, flags, argp, &ext_arg);
3443                         if (likely(!ret2))
3444                                 ret2 = io_cqring_wait(ctx, min_complete, flags,
3445                                                       &ext_arg);
3446                 }
3447
3448                 if (!ret) {
3449                         ret = ret2;
3450
3451                         /*
3452                          * EBADR indicates that one or more CQE were dropped.
3453                          * Once the user has been informed we can clear the bit
3454                          * as they are obviously ok with those drops.
3455                          */
3456                         if (unlikely(ret2 == -EBADR))
3457                                 clear_bit(IO_CHECK_CQ_DROPPED_BIT,
3458                                           &ctx->check_cq);
3459                 }
3460         }
3461 out:
3462         if (!(flags & IORING_ENTER_REGISTERED_RING))
3463                 fput(file);
3464         return ret;
3465 }
3466
3467 static const struct file_operations io_uring_fops = {
3468         .release        = io_uring_release,
3469         .mmap           = io_uring_mmap,
3470         .get_unmapped_area = io_uring_get_unmapped_area,
3471 #ifndef CONFIG_MMU
3472         .mmap_capabilities = io_uring_nommu_mmap_capabilities,
3473 #endif
3474         .poll           = io_uring_poll,
3475 #ifdef CONFIG_PROC_FS
3476         .show_fdinfo    = io_uring_show_fdinfo,
3477 #endif
3478 };
3479
3480 bool io_is_uring_fops(struct file *file)
3481 {
3482         return file->f_op == &io_uring_fops;
3483 }
3484
3485 static __cold int io_allocate_scq_urings(struct io_ring_ctx *ctx,
3486                                          struct io_uring_params *p)
3487 {
3488         struct io_uring_region_desc rd;
3489         struct io_rings *rings;
3490         size_t size, sq_array_offset;
3491         int ret;
3492
3493         /* make sure these are sane, as we already accounted them */
3494         ctx->sq_entries = p->sq_entries;
3495         ctx->cq_entries = p->cq_entries;
3496
3497         size = rings_size(ctx->flags, p->sq_entries, p->cq_entries,
3498                           &sq_array_offset);
3499         if (size == SIZE_MAX)
3500                 return -EOVERFLOW;
3501
3502         memset(&rd, 0, sizeof(rd));
3503         rd.size = PAGE_ALIGN(size);
3504         if (ctx->flags & IORING_SETUP_NO_MMAP) {
3505                 rd.user_addr = p->cq_off.user_addr;
3506                 rd.flags |= IORING_MEM_REGION_TYPE_USER;
3507         }
3508         ret = io_create_region(ctx, &ctx->ring_region, &rd, IORING_OFF_CQ_RING);
3509         if (ret)
3510                 return ret;
3511         ctx->rings = rings = io_region_get_ptr(&ctx->ring_region);
3512
3513         if (!(ctx->flags & IORING_SETUP_NO_SQARRAY))
3514                 ctx->sq_array = (u32 *)((char *)rings + sq_array_offset);
3515         rings->sq_ring_mask = p->sq_entries - 1;
3516         rings->cq_ring_mask = p->cq_entries - 1;
3517         rings->sq_ring_entries = p->sq_entries;
3518         rings->cq_ring_entries = p->cq_entries;
3519
3520         if (p->flags & IORING_SETUP_SQE128)
3521                 size = array_size(2 * sizeof(struct io_uring_sqe), p->sq_entries);
3522         else
3523                 size = array_size(sizeof(struct io_uring_sqe), p->sq_entries);
3524         if (size == SIZE_MAX) {
3525                 io_rings_free(ctx);
3526                 return -EOVERFLOW;
3527         }
3528
3529         memset(&rd, 0, sizeof(rd));
3530         rd.size = PAGE_ALIGN(size);
3531         if (ctx->flags & IORING_SETUP_NO_MMAP) {
3532                 rd.user_addr = p->sq_off.user_addr;
3533                 rd.flags |= IORING_MEM_REGION_TYPE_USER;
3534         }
3535         ret = io_create_region(ctx, &ctx->sq_region, &rd, IORING_OFF_SQES);
3536         if (ret) {
3537                 io_rings_free(ctx);
3538                 return ret;
3539         }
3540         ctx->sq_sqes = io_region_get_ptr(&ctx->sq_region);
3541         return 0;
3542 }
3543
3544 static int io_uring_install_fd(struct file *file)
3545 {
3546         int fd;
3547
3548         fd = get_unused_fd_flags(O_RDWR | O_CLOEXEC);
3549         if (fd < 0)
3550                 return fd;
3551         fd_install(fd, file);
3552         return fd;
3553 }
3554
3555 /*
3556  * Allocate an anonymous fd, this is what constitutes the application
3557  * visible backing of an io_uring instance. The application mmaps this
3558  * fd to gain access to the SQ/CQ ring details.
3559  */
3560 static struct file *io_uring_get_file(struct io_ring_ctx *ctx)
3561 {
3562         /* Create a new inode so that the LSM can block the creation.  */
3563         return anon_inode_create_getfile("[io_uring]", &io_uring_fops, ctx,
3564                                          O_RDWR | O_CLOEXEC, NULL);
3565 }
3566
3567 static int io_uring_sanitise_params(struct io_uring_params *p)
3568 {
3569         unsigned flags = p->flags;
3570
3571         /* There is no way to mmap rings without a real fd */
3572         if ((flags & IORING_SETUP_REGISTERED_FD_ONLY) &&
3573             !(flags & IORING_SETUP_NO_MMAP))
3574                 return -EINVAL;
3575
3576         if (flags & IORING_SETUP_SQPOLL) {
3577                 /* IPI related flags don't make sense with SQPOLL */
3578                 if (flags & (IORING_SETUP_COOP_TASKRUN |
3579                              IORING_SETUP_TASKRUN_FLAG |
3580                              IORING_SETUP_DEFER_TASKRUN))
3581                         return -EINVAL;
3582         }
3583
3584         if (flags & IORING_SETUP_TASKRUN_FLAG) {
3585                 if (!(flags & (IORING_SETUP_COOP_TASKRUN |
3586                                IORING_SETUP_DEFER_TASKRUN)))
3587                         return -EINVAL;
3588         }
3589
3590         /* HYBRID_IOPOLL only valid with IOPOLL */
3591         if ((flags & IORING_SETUP_HYBRID_IOPOLL) && !(flags & IORING_SETUP_IOPOLL))
3592                 return -EINVAL;
3593
3594         /*
3595          * For DEFER_TASKRUN we require the completion task to be the same as
3596          * the submission task. This implies that there is only one submitter.
3597          */
3598         if ((flags & IORING_SETUP_DEFER_TASKRUN) &&
3599             !(flags & IORING_SETUP_SINGLE_ISSUER))
3600                 return -EINVAL;
3601
3602         return 0;
3603 }
3604
3605 int io_uring_fill_params(unsigned entries, struct io_uring_params *p)
3606 {
3607         if (!entries)
3608                 return -EINVAL;
3609         if (entries > IORING_MAX_ENTRIES) {
3610                 if (!(p->flags & IORING_SETUP_CLAMP))
3611                         return -EINVAL;
3612                 entries = IORING_MAX_ENTRIES;
3613         }
3614
3615         /*
3616          * Use twice as many entries for the CQ ring. It's possible for the
3617          * application to drive a higher depth than the size of the SQ ring,
3618          * since the sqes are only used at submission time. This allows for
3619          * some flexibility in overcommitting a bit. If the application has
3620          * set IORING_SETUP_CQSIZE, it will have passed in the desired number
3621          * of CQ ring entries manually.
3622          */
3623         p->sq_entries = roundup_pow_of_two(entries);
3624         if (p->flags & IORING_SETUP_CQSIZE) {
3625                 /*
3626                  * If IORING_SETUP_CQSIZE is set, we do the same roundup
3627                  * to a power-of-two, if it isn't already. We do NOT impose
3628                  * any cq vs sq ring sizing.
3629                  */
3630                 if (!p->cq_entries)
3631                         return -EINVAL;
3632                 if (p->cq_entries > IORING_MAX_CQ_ENTRIES) {
3633                         if (!(p->flags & IORING_SETUP_CLAMP))
3634                                 return -EINVAL;
3635                         p->cq_entries = IORING_MAX_CQ_ENTRIES;
3636                 }
3637                 p->cq_entries = roundup_pow_of_two(p->cq_entries);
3638                 if (p->cq_entries < p->sq_entries)
3639                         return -EINVAL;
3640         } else {
3641                 p->cq_entries = 2 * p->sq_entries;
3642         }
3643
3644         p->sq_off.head = offsetof(struct io_rings, sq.head);
3645         p->sq_off.tail = offsetof(struct io_rings, sq.tail);
3646         p->sq_off.ring_mask = offsetof(struct io_rings, sq_ring_mask);
3647         p->sq_off.ring_entries = offsetof(struct io_rings, sq_ring_entries);
3648         p->sq_off.flags = offsetof(struct io_rings, sq_flags);
3649         p->sq_off.dropped = offsetof(struct io_rings, sq_dropped);
3650         p->sq_off.resv1 = 0;
3651         if (!(p->flags & IORING_SETUP_NO_MMAP))
3652                 p->sq_off.user_addr = 0;
3653
3654         p->cq_off.head = offsetof(struct io_rings, cq.head);
3655         p->cq_off.tail = offsetof(struct io_rings, cq.tail);
3656         p->cq_off.ring_mask = offsetof(struct io_rings, cq_ring_mask);
3657         p->cq_off.ring_entries = offsetof(struct io_rings, cq_ring_entries);
3658         p->cq_off.overflow = offsetof(struct io_rings, cq_overflow);
3659         p->cq_off.cqes = offsetof(struct io_rings, cqes);
3660         p->cq_off.flags = offsetof(struct io_rings, cq_flags);
3661         p->cq_off.resv1 = 0;
3662         if (!(p->flags & IORING_SETUP_NO_MMAP))
3663                 p->cq_off.user_addr = 0;
3664
3665         return 0;
3666 }
3667
3668 static __cold int io_uring_create(unsigned entries, struct io_uring_params *p,
3669                                   struct io_uring_params __user *params)
3670 {
3671         struct io_ring_ctx *ctx;
3672         struct io_uring_task *tctx;
3673         struct file *file;
3674         int ret;
3675
3676         ret = io_uring_sanitise_params(p);
3677         if (ret)
3678                 return ret;
3679
3680         ret = io_uring_fill_params(entries, p);
3681         if (unlikely(ret))
3682                 return ret;
3683
3684         ctx = io_ring_ctx_alloc(p);
3685         if (!ctx)
3686                 return -ENOMEM;
3687
3688         ctx->clockid = CLOCK_MONOTONIC;
3689         ctx->clock_offset = 0;
3690
3691         if (!(ctx->flags & IORING_SETUP_NO_SQARRAY))
3692                 static_branch_inc(&io_key_has_sqarray);
3693
3694         if ((ctx->flags & IORING_SETUP_DEFER_TASKRUN) &&
3695             !(ctx->flags & IORING_SETUP_IOPOLL) &&
3696             !(ctx->flags & IORING_SETUP_SQPOLL))
3697                 ctx->task_complete = true;
3698
3699         if (ctx->task_complete || (ctx->flags & IORING_SETUP_IOPOLL))
3700                 ctx->lockless_cq = true;
3701
3702         /*
3703          * lazy poll_wq activation relies on ->task_complete for synchronisation
3704          * purposes, see io_activate_pollwq()
3705          */
3706         if (!ctx->task_complete)
3707                 ctx->poll_activated = true;
3708
3709         /*
3710          * When SETUP_IOPOLL and SETUP_SQPOLL are both enabled, user
3711          * space applications don't need to do io completion events
3712          * polling again, they can rely on io_sq_thread to do polling
3713          * work, which can reduce cpu usage and uring_lock contention.
3714          */
3715         if (ctx->flags & IORING_SETUP_IOPOLL &&
3716             !(ctx->flags & IORING_SETUP_SQPOLL))
3717                 ctx->syscall_iopoll = 1;
3718
3719         ctx->compat = in_compat_syscall();
3720         if (!ns_capable_noaudit(&init_user_ns, CAP_IPC_LOCK))
3721                 ctx->user = get_uid(current_user());
3722
3723         /*
3724          * For SQPOLL, we just need a wakeup, always. For !SQPOLL, if
3725          * COOP_TASKRUN is set, then IPIs are never needed by the app.
3726          */
3727         if (ctx->flags & (IORING_SETUP_SQPOLL|IORING_SETUP_COOP_TASKRUN))
3728                 ctx->notify_method = TWA_SIGNAL_NO_IPI;
3729         else
3730                 ctx->notify_method = TWA_SIGNAL;
3731
3732         /*
3733          * This is just grabbed for accounting purposes. When a process exits,
3734          * the mm is exited and dropped before the files, hence we need to hang
3735          * on to this mm purely for the purposes of being able to unaccount
3736          * memory (locked/pinned vm). It's not used for anything else.
3737          */
3738         mmgrab(current->mm);
3739         ctx->mm_account = current->mm;
3740
3741         ret = io_allocate_scq_urings(ctx, p);
3742         if (ret)
3743                 goto err;
3744
3745         if (!(p->flags & IORING_SETUP_NO_SQARRAY))
3746                 p->sq_off.array = (char *)ctx->sq_array - (char *)ctx->rings;
3747
3748         ret = io_sq_offload_create(ctx, p);
3749         if (ret)
3750                 goto err;
3751
3752         p->features = IORING_FEAT_SINGLE_MMAP | IORING_FEAT_NODROP |
3753                         IORING_FEAT_SUBMIT_STABLE | IORING_FEAT_RW_CUR_POS |
3754                         IORING_FEAT_CUR_PERSONALITY | IORING_FEAT_FAST_POLL |
3755                         IORING_FEAT_POLL_32BITS | IORING_FEAT_SQPOLL_NONFIXED |
3756                         IORING_FEAT_EXT_ARG | IORING_FEAT_NATIVE_WORKERS |
3757                         IORING_FEAT_RSRC_TAGS | IORING_FEAT_CQE_SKIP |
3758                         IORING_FEAT_LINKED_FILE | IORING_FEAT_REG_REG_RING |
3759                         IORING_FEAT_RECVSEND_BUNDLE | IORING_FEAT_MIN_TIMEOUT |
3760                         IORING_FEAT_RW_ATTR | IORING_FEAT_NO_IOWAIT;
3761
3762         if (copy_to_user(params, p, sizeof(*p))) {
3763                 ret = -EFAULT;
3764                 goto err;
3765         }
3766
3767         if (ctx->flags & IORING_SETUP_SINGLE_ISSUER
3768             && !(ctx->flags & IORING_SETUP_R_DISABLED))
3769                 WRITE_ONCE(ctx->submitter_task, get_task_struct(current));
3770
3771         file = io_uring_get_file(ctx);
3772         if (IS_ERR(file)) {
3773                 ret = PTR_ERR(file);
3774                 goto err;
3775         }
3776
3777         ret = __io_uring_add_tctx_node(ctx);
3778         if (ret)
3779                 goto err_fput;
3780         tctx = current->io_uring;
3781
3782         /*
3783          * Install ring fd as the very last thing, so we don't risk someone
3784          * having closed it before we finish setup
3785          */
3786         if (p->flags & IORING_SETUP_REGISTERED_FD_ONLY)
3787                 ret = io_ring_add_registered_file(tctx, file, 0, IO_RINGFD_REG_MAX);
3788         else
3789                 ret = io_uring_install_fd(file);
3790         if (ret < 0)
3791                 goto err_fput;
3792
3793         trace_io_uring_create(ret, ctx, p->sq_entries, p->cq_entries, p->flags);
3794         return ret;
3795 err:
3796         io_ring_ctx_wait_and_kill(ctx);
3797         return ret;
3798 err_fput:
3799         fput(file);
3800         return ret;
3801 }
3802
3803 /*
3804  * Sets up an aio uring context, and returns the fd. Applications asks for a
3805  * ring size, we return the actual sq/cq ring sizes (among other things) in the
3806  * params structure passed in.
3807  */
3808 static long io_uring_setup(u32 entries, struct io_uring_params __user *params)
3809 {
3810         struct io_uring_params p;
3811         int i;
3812
3813         if (copy_from_user(&p, params, sizeof(p)))
3814                 return -EFAULT;
3815         for (i = 0; i < ARRAY_SIZE(p.resv); i++) {
3816                 if (p.resv[i])
3817                         return -EINVAL;
3818         }
3819
3820         if (p.flags & ~(IORING_SETUP_IOPOLL | IORING_SETUP_SQPOLL |
3821                         IORING_SETUP_SQ_AFF | IORING_SETUP_CQSIZE |
3822                         IORING_SETUP_CLAMP | IORING_SETUP_ATTACH_WQ |
3823                         IORING_SETUP_R_DISABLED | IORING_SETUP_SUBMIT_ALL |
3824                         IORING_SETUP_COOP_TASKRUN | IORING_SETUP_TASKRUN_FLAG |
3825                         IORING_SETUP_SQE128 | IORING_SETUP_CQE32 |
3826                         IORING_SETUP_SINGLE_ISSUER | IORING_SETUP_DEFER_TASKRUN |
3827                         IORING_SETUP_NO_MMAP | IORING_SETUP_REGISTERED_FD_ONLY |
3828                         IORING_SETUP_NO_SQARRAY | IORING_SETUP_HYBRID_IOPOLL))
3829                 return -EINVAL;
3830
3831         return io_uring_create(entries, &p, params);
3832 }
3833
3834 static inline int io_uring_allowed(void)
3835 {
3836         int disabled = READ_ONCE(sysctl_io_uring_disabled);
3837         kgid_t io_uring_group;
3838
3839         if (disabled == 2)
3840                 return -EPERM;
3841
3842         if (disabled == 0 || capable(CAP_SYS_ADMIN))
3843                 goto allowed_lsm;
3844
3845         io_uring_group = make_kgid(&init_user_ns, sysctl_io_uring_group);
3846         if (!gid_valid(io_uring_group))
3847                 return -EPERM;
3848
3849         if (!in_group_p(io_uring_group))
3850                 return -EPERM;
3851
3852 allowed_lsm:
3853         return security_uring_allowed();
3854 }
3855
3856 SYSCALL_DEFINE2(io_uring_setup, u32, entries,
3857                 struct io_uring_params __user *, params)
3858 {
3859         int ret;
3860
3861         ret = io_uring_allowed();
3862         if (ret)
3863                 return ret;
3864
3865         return io_uring_setup(entries, params);
3866 }
3867
3868 static int __init io_uring_init(void)
3869 {
3870         struct kmem_cache_args kmem_args = {
3871                 .useroffset = offsetof(struct io_kiocb, cmd.data),
3872                 .usersize = sizeof_field(struct io_kiocb, cmd.data),
3873                 .freeptr_offset = offsetof(struct io_kiocb, work),
3874                 .use_freeptr_offset = true,
3875         };
3876
3877 #define __BUILD_BUG_VERIFY_OFFSET_SIZE(stype, eoffset, esize, ename) do { \
3878         BUILD_BUG_ON(offsetof(stype, ename) != eoffset); \
3879         BUILD_BUG_ON(sizeof_field(stype, ename) != esize); \
3880 } while (0)
3881
3882 #define BUILD_BUG_SQE_ELEM(eoffset, etype, ename) \
3883         __BUILD_BUG_VERIFY_OFFSET_SIZE(struct io_uring_sqe, eoffset, sizeof(etype), ename)
3884 #define BUILD_BUG_SQE_ELEM_SIZE(eoffset, esize, ename) \
3885         __BUILD_BUG_VERIFY_OFFSET_SIZE(struct io_uring_sqe, eoffset, esize, ename)
3886         BUILD_BUG_ON(sizeof(struct io_uring_sqe) != 64);
3887         BUILD_BUG_SQE_ELEM(0,  __u8,   opcode);
3888         BUILD_BUG_SQE_ELEM(1,  __u8,   flags);
3889         BUILD_BUG_SQE_ELEM(2,  __u16,  ioprio);
3890         BUILD_BUG_SQE_ELEM(4,  __s32,  fd);
3891         BUILD_BUG_SQE_ELEM(8,  __u64,  off);
3892         BUILD_BUG_SQE_ELEM(8,  __u64,  addr2);
3893         BUILD_BUG_SQE_ELEM(8,  __u32,  cmd_op);
3894         BUILD_BUG_SQE_ELEM(12, __u32, __pad1);
3895         BUILD_BUG_SQE_ELEM(16, __u64,  addr);
3896         BUILD_BUG_SQE_ELEM(16, __u64,  splice_off_in);
3897         BUILD_BUG_SQE_ELEM(24, __u32,  len);
3898         BUILD_BUG_SQE_ELEM(28,     __kernel_rwf_t, rw_flags);
3899         BUILD_BUG_SQE_ELEM(28, /* compat */   int, rw_flags);
3900         BUILD_BUG_SQE_ELEM(28, /* compat */ __u32, rw_flags);
3901         BUILD_BUG_SQE_ELEM(28, __u32,  fsync_flags);
3902         BUILD_BUG_SQE_ELEM(28, /* compat */ __u16,  poll_events);
3903         BUILD_BUG_SQE_ELEM(28, __u32,  poll32_events);
3904         BUILD_BUG_SQE_ELEM(28, __u32,  sync_range_flags);
3905         BUILD_BUG_SQE_ELEM(28, __u32,  msg_flags);
3906         BUILD_BUG_SQE_ELEM(28, __u32,  timeout_flags);
3907         BUILD_BUG_SQE_ELEM(28, __u32,  accept_flags);
3908         BUILD_BUG_SQE_ELEM(28, __u32,  cancel_flags);
3909         BUILD_BUG_SQE_ELEM(28, __u32,  open_flags);
3910         BUILD_BUG_SQE_ELEM(28, __u32,  statx_flags);
3911         BUILD_BUG_SQE_ELEM(28, __u32,  fadvise_advice);
3912         BUILD_BUG_SQE_ELEM(28, __u32,  splice_flags);
3913         BUILD_BUG_SQE_ELEM(28, __u32,  rename_flags);
3914         BUILD_BUG_SQE_ELEM(28, __u32,  unlink_flags);
3915         BUILD_BUG_SQE_ELEM(28, __u32,  hardlink_flags);
3916         BUILD_BUG_SQE_ELEM(28, __u32,  xattr_flags);
3917         BUILD_BUG_SQE_ELEM(28, __u32,  msg_ring_flags);
3918         BUILD_BUG_SQE_ELEM(32, __u64,  user_data);
3919         BUILD_BUG_SQE_ELEM(40, __u16,  buf_index);
3920         BUILD_BUG_SQE_ELEM(40, __u16,  buf_group);
3921         BUILD_BUG_SQE_ELEM(42, __u16,  personality);
3922         BUILD_BUG_SQE_ELEM(44, __s32,  splice_fd_in);
3923         BUILD_BUG_SQE_ELEM(44, __u32,  file_index);
3924         BUILD_BUG_SQE_ELEM(44, __u16,  addr_len);
3925         BUILD_BUG_SQE_ELEM(44, __u8,   write_stream);
3926         BUILD_BUG_SQE_ELEM(45, __u8,   __pad4[0]);
3927         BUILD_BUG_SQE_ELEM(46, __u16,  __pad3[0]);
3928         BUILD_BUG_SQE_ELEM(48, __u64,  addr3);
3929         BUILD_BUG_SQE_ELEM_SIZE(48, 0, cmd);
3930         BUILD_BUG_SQE_ELEM(48, __u64, attr_ptr);
3931         BUILD_BUG_SQE_ELEM(56, __u64, attr_type_mask);
3932         BUILD_BUG_SQE_ELEM(56, __u64,  __pad2);
3933
3934         BUILD_BUG_ON(sizeof(struct io_uring_files_update) !=
3935                      sizeof(struct io_uring_rsrc_update));
3936         BUILD_BUG_ON(sizeof(struct io_uring_rsrc_update) >
3937                      sizeof(struct io_uring_rsrc_update2));
3938
3939         /* ->buf_index is u16 */
3940         BUILD_BUG_ON(offsetof(struct io_uring_buf_ring, bufs) != 0);
3941         BUILD_BUG_ON(offsetof(struct io_uring_buf, resv) !=
3942                      offsetof(struct io_uring_buf_ring, tail));
3943
3944         /* should fit into one byte */
3945         BUILD_BUG_ON(SQE_VALID_FLAGS >= (1 << 8));
3946         BUILD_BUG_ON(SQE_COMMON_FLAGS >= (1 << 8));
3947         BUILD_BUG_ON((SQE_VALID_FLAGS | SQE_COMMON_FLAGS) != SQE_VALID_FLAGS);
3948
3949         BUILD_BUG_ON(__REQ_F_LAST_BIT > 8 * sizeof_field(struct io_kiocb, flags));
3950
3951         BUILD_BUG_ON(sizeof(atomic_t) != sizeof(u32));
3952
3953         /* top 8bits are for internal use */
3954         BUILD_BUG_ON((IORING_URING_CMD_MASK & 0xff000000) != 0);
3955
3956         io_uring_optable_init();
3957
3958         /* imu->dir is u8 */
3959         BUILD_BUG_ON((IO_IMU_DEST | IO_IMU_SOURCE) > U8_MAX);
3960
3961         /*
3962          * Allow user copy in the per-command field, which starts after the
3963          * file in io_kiocb and until the opcode field. The openat2 handling
3964          * requires copying in user memory into the io_kiocb object in that
3965          * range, and HARDENED_USERCOPY will complain if we haven't
3966          * correctly annotated this range.
3967          */
3968         req_cachep = kmem_cache_create("io_kiocb", sizeof(struct io_kiocb), &kmem_args,
3969                                 SLAB_HWCACHE_ALIGN | SLAB_PANIC | SLAB_ACCOUNT |
3970                                 SLAB_TYPESAFE_BY_RCU);
3971
3972         iou_wq = alloc_workqueue("iou_exit", WQ_UNBOUND, 64);
3973         BUG_ON(!iou_wq);
3974
3975 #ifdef CONFIG_SYSCTL
3976         register_sysctl_init("kernel", kernel_io_uring_disabled_table);
3977 #endif
3978
3979         return 0;
3980 };
3981 __initcall(io_uring_init);