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