io_uring/io_uring.h

   1 #ifndef IOU_CORE_H
   2 #define IOU_CORE_H
   3
   4 #include <linux/errno.h>
   5 #include <linux/lockdep.h>
   6 #include <linux/resume_user_mode.h>
   7 #include <linux/kasan.h>
   8 #include <linux/poll.h>
   9 #include <linux/io_uring_types.h>
  10 #include <uapi/linux/eventpoll.h>
  11 #include "io-wq.h"
  12 #include "slist.h"
  13 #include "filetable.h"
  14
  15 #ifndef CREATE_TRACE_POINTS
  16 #include <trace/events/io_uring.h>
  17 #endif
  18
  19 enum {
  20         IOU_OK                  = 0,
  21         IOU_ISSUE_SKIP_COMPLETE = -EIOCBQUEUED,
  22
  23         /*
  24          * Requeue the task_work to restart operations on this request. The
  25          * actual value isn't important, should just be not an otherwise
  26          * valid error code, yet less than -MAX_ERRNO and valid internally.
  27          */
  28         IOU_REQUEUE             = -3072,
  29
  30         /*
  31          * Intended only when both IO_URING_F_MULTISHOT is passed
  32          * to indicate to the poll runner that multishot should be
  33          * removed and the result is set on req->cqe.res.
  34          */
  35         IOU_STOP_MULTISHOT      = -ECANCELED,
  36 };
  37
  38 struct io_wait_queue {
  39         struct wait_queue_entry wq;
  40         struct io_ring_ctx *ctx;
  41         unsigned cq_tail;
  42         unsigned cq_min_tail;
  43         unsigned nr_timeouts;
  44         int hit_timeout;
  45         ktime_t min_timeout;
  46         ktime_t timeout;
  47         struct hrtimer t;
  48
  49 #ifdef CONFIG_NET_RX_BUSY_POLL
  50         ktime_t napi_busy_poll_dt;
  51         bool napi_prefer_busy_poll;
  52 #endif
  53 };
  54
  55 static inline bool io_should_wake(struct io_wait_queue *iowq)
  56 {
  57         struct io_ring_ctx *ctx = iowq->ctx;
  58         int dist = READ_ONCE(ctx->rings->cq.tail) - (int) iowq->cq_tail;
  59
  60         /*
  61          * Wake up if we have enough events, or if a timeout occurred since we
  62          * started waiting. For timeouts, we always want to return to userspace,
  63          * regardless of event count.
  64          */
  65         return dist >= 0 || atomic_read(&ctx->cq_timeouts) != iowq->nr_timeouts;
  66 }
  67
  68 #define IORING_MAX_ENTRIES      32768
  69 #define IORING_MAX_CQ_ENTRIES   (2 * IORING_MAX_ENTRIES)
  70
  71 unsigned long rings_size(unsigned int flags, unsigned int sq_entries,
  72                          unsigned int cq_entries, size_t *sq_offset);
  73 int io_uring_fill_params(unsigned entries, struct io_uring_params *p);
  74 bool io_cqe_cache_refill(struct io_ring_ctx *ctx, bool overflow);
  75 int io_run_task_work_sig(struct io_ring_ctx *ctx);
  76 void io_req_defer_failed(struct io_kiocb *req, s32 res);
  77 bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags);
  78 void io_add_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags);
  79 bool io_req_post_cqe(struct io_kiocb *req, s32 res, u32 cflags);
  80 void __io_commit_cqring_flush(struct io_ring_ctx *ctx);
  81
  82 struct file *io_file_get_normal(struct io_kiocb *req, int fd);
  83 struct file *io_file_get_fixed(struct io_kiocb *req, int fd,
  84                                unsigned issue_flags);
  85
  86 void __io_req_task_work_add(struct io_kiocb *req, unsigned flags);
  87 void io_req_task_work_add_remote(struct io_kiocb *req, struct io_ring_ctx *ctx,
  88                                  unsigned flags);
  89 bool io_alloc_async_data(struct io_kiocb *req);
  90 void io_req_task_queue(struct io_kiocb *req);
  91 void io_req_task_complete(struct io_kiocb *req, struct io_tw_state *ts);
  92 void io_req_task_queue_fail(struct io_kiocb *req, int ret);
  93 void io_req_task_submit(struct io_kiocb *req, struct io_tw_state *ts);
  94 struct llist_node *io_handle_tw_list(struct llist_node *node, unsigned int *count, unsigned int max_entries);
  95 struct llist_node *tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, unsigned int *count);
  96 void tctx_task_work(struct callback_head *cb);
  97 __cold void io_uring_cancel_generic(bool cancel_all, struct io_sq_data *sqd);
  98 int io_uring_alloc_task_context(struct task_struct *task,
  99                                 struct io_ring_ctx *ctx);
 100
 101 int io_ring_add_registered_file(struct io_uring_task *tctx, struct file *file,
 102                                      int start, int end);
 103 void io_req_queue_iowq(struct io_kiocb *req);
 104
 105 int io_poll_issue(struct io_kiocb *req, struct io_tw_state *ts);
 106 int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr);
 107 int io_do_iopoll(struct io_ring_ctx *ctx, bool force_nonspin);
 108 void __io_submit_flush_completions(struct io_ring_ctx *ctx);
 109
 110 struct io_wq_work *io_wq_free_work(struct io_wq_work *work);
 111 void io_wq_submit_work(struct io_wq_work *work);
 112
 113 void io_free_req(struct io_kiocb *req);
 114 void io_queue_next(struct io_kiocb *req);
 115 void io_task_refs_refill(struct io_uring_task *tctx);
 116 bool __io_alloc_req_refill(struct io_ring_ctx *ctx);
 117
 118 bool io_match_task_safe(struct io_kiocb *head, struct io_uring_task *tctx,
 119                         bool cancel_all);
 120
 121 void io_activate_pollwq(struct io_ring_ctx *ctx);
 122
 123 static inline void io_lockdep_assert_cq_locked(struct io_ring_ctx *ctx)
 124 {
 125 #if defined(CONFIG_PROVE_LOCKING)
 126         lockdep_assert(in_task());
 127
 128         if (ctx->flags & IORING_SETUP_IOPOLL) {
 129                 lockdep_assert_held(&ctx->uring_lock);
 130         } else if (!ctx->task_complete) {
 131                 lockdep_assert_held(&ctx->completion_lock);
 132         } else if (ctx->submitter_task) {
 133                 /*
 134                  * ->submitter_task may be NULL and we can still post a CQE,
 135                  * if the ring has been setup with IORING_SETUP_R_DISABLED.
 136                  * Not from an SQE, as those cannot be submitted, but via
 137                  * updating tagged resources.
 138                  */
 139                 if (percpu_ref_is_dying(&ctx->refs))
 140                         lockdep_assert(current_work());
 141                 else
 142                         lockdep_assert(current == ctx->submitter_task);
 143         }
 144 #endif
 145 }
 146
 147 static inline void io_req_task_work_add(struct io_kiocb *req)
 148 {
 149         __io_req_task_work_add(req, 0);
 150 }
 151
 152 static inline void io_submit_flush_completions(struct io_ring_ctx *ctx)
 153 {
 154         if (!wq_list_empty(&ctx->submit_state.compl_reqs) ||
 155             ctx->submit_state.cq_flush)
 156                 __io_submit_flush_completions(ctx);
 157 }
 158
 159 #define io_for_each_link(pos, head) \
 160         for (pos = (head); pos; pos = pos->link)
 161
 162 static inline bool io_get_cqe_overflow(struct io_ring_ctx *ctx,
 163                                         struct io_uring_cqe **ret,
 164                                         bool overflow)
 165 {
 166         io_lockdep_assert_cq_locked(ctx);
 167
 168         if (unlikely(ctx->cqe_cached >= ctx->cqe_sentinel)) {
 169                 if (unlikely(!io_cqe_cache_refill(ctx, overflow)))
 170                         return false;
 171         }
 172         *ret = ctx->cqe_cached;
 173         ctx->cached_cq_tail++;
 174         ctx->cqe_cached++;
 175         if (ctx->flags & IORING_SETUP_CQE32)
 176                 ctx->cqe_cached++;
 177         return true;
 178 }
 179
 180 static inline bool io_get_cqe(struct io_ring_ctx *ctx, struct io_uring_cqe **ret)
 181 {
 182         return io_get_cqe_overflow(ctx, ret, false);
 183 }
 184
 185 static __always_inline bool io_fill_cqe_req(struct io_ring_ctx *ctx,
 186                                             struct io_kiocb *req)
 187 {
 188         struct io_uring_cqe *cqe;
 189
 190         /*
 191          * If we can't get a cq entry, userspace overflowed the
 192          * submission (by quite a lot). Increment the overflow count in
 193          * the ring.
 194          */
 195         if (unlikely(!io_get_cqe(ctx, &cqe)))
 196                 return false;
 197
 198
 199         memcpy(cqe, &req->cqe, sizeof(*cqe));
 200         if (ctx->flags & IORING_SETUP_CQE32) {
 201                 memcpy(cqe->big_cqe, &req->big_cqe, sizeof(*cqe));
 202                 memset(&req->big_cqe, 0, sizeof(req->big_cqe));
 203         }
 204
 205         if (trace_io_uring_complete_enabled())
 206                 trace_io_uring_complete(req->ctx, req, cqe);
 207         return true;
 208 }
 209
 210 static inline void req_set_fail(struct io_kiocb *req)
 211 {
 212         req->flags |= REQ_F_FAIL;
 213         if (req->flags & REQ_F_CQE_SKIP) {
 214                 req->flags &= ~REQ_F_CQE_SKIP;
 215                 req->flags |= REQ_F_SKIP_LINK_CQES;
 216         }
 217 }
 218
 219 static inline void io_req_set_res(struct io_kiocb *req, s32 res, u32 cflags)
 220 {
 221         req->cqe.res = res;
 222         req->cqe.flags = cflags;
 223 }
 224
 225 static inline bool req_has_async_data(struct io_kiocb *req)
 226 {
 227         return req->flags & REQ_F_ASYNC_DATA;
 228 }
 229
 230 static inline void io_put_file(struct io_kiocb *req)
 231 {
 232         if (!(req->flags & REQ_F_FIXED_FILE) && req->file)
 233                 fput(req->file);
 234 }
 235
 236 static inline void io_ring_submit_unlock(struct io_ring_ctx *ctx,
 237                                          unsigned issue_flags)
 238 {
 239         lockdep_assert_held(&ctx->uring_lock);
 240         if (unlikely(issue_flags & IO_URING_F_UNLOCKED))
 241                 mutex_unlock(&ctx->uring_lock);
 242 }
 243
 244 static inline void io_ring_submit_lock(struct io_ring_ctx *ctx,
 245                                        unsigned issue_flags)
 246 {
 247         /*
 248          * "Normal" inline submissions always hold the uring_lock, since we
 249          * grab it from the system call. Same is true for the SQPOLL offload.
 250          * The only exception is when we've detached the request and issue it
 251          * from an async worker thread, grab the lock for that case.
 252          */
 253         if (unlikely(issue_flags & IO_URING_F_UNLOCKED))
 254                 mutex_lock(&ctx->uring_lock);
 255         lockdep_assert_held(&ctx->uring_lock);
 256 }
 257
 258 static inline void io_commit_cqring(struct io_ring_ctx *ctx)
 259 {
 260         /* order cqe stores with ring update */
 261         smp_store_release(&ctx->rings->cq.tail, ctx->cached_cq_tail);
 262 }
 263
 264 static inline void io_poll_wq_wake(struct io_ring_ctx *ctx)
 265 {
 266         if (wq_has_sleeper(&ctx->poll_wq))
 267                 __wake_up(&ctx->poll_wq, TASK_NORMAL, 0,
 268                                 poll_to_key(EPOLL_URING_WAKE | EPOLLIN));
 269 }
 270
 271 static inline void io_cqring_wake(struct io_ring_ctx *ctx)
 272 {
 273         /*
 274          * Trigger waitqueue handler on all waiters on our waitqueue. This
 275          * won't necessarily wake up all the tasks, io_should_wake() will make
 276          * that decision.
 277          *
 278          * Pass in EPOLLIN|EPOLL_URING_WAKE as the poll wakeup key. The latter
 279          * set in the mask so that if we recurse back into our own poll
 280          * waitqueue handlers, we know we have a dependency between eventfd or
 281          * epoll and should terminate multishot poll at that point.
 282          */
 283         if (wq_has_sleeper(&ctx->cq_wait))
 284                 __wake_up(&ctx->cq_wait, TASK_NORMAL, 0,
 285                                 poll_to_key(EPOLL_URING_WAKE | EPOLLIN));
 286 }
 287
 288 static inline bool io_sqring_full(struct io_ring_ctx *ctx)
 289 {
 290         struct io_rings *r = ctx->rings;
 291
 292         /*
 293          * SQPOLL must use the actual sqring head, as using the cached_sq_head
 294          * is race prone if the SQPOLL thread has grabbed entries but not yet
 295          * committed them to the ring. For !SQPOLL, this doesn't matter, but
 296          * since this helper is just used for SQPOLL sqring waits (or POLLOUT),
 297          * just read the actual sqring head unconditionally.
 298          */
 299         return READ_ONCE(r->sq.tail) - READ_ONCE(r->sq.head) == ctx->sq_entries;
 300 }
 301
 302 static inline unsigned int io_sqring_entries(struct io_ring_ctx *ctx)
 303 {
 304         struct io_rings *rings = ctx->rings;
 305         unsigned int entries;
 306
 307         /* make sure SQ entry isn't read before tail */
 308         entries = smp_load_acquire(&rings->sq.tail) - ctx->cached_sq_head;
 309         return min(entries, ctx->sq_entries);
 310 }
 311
 312 static inline int io_run_task_work(void)
 313 {
 314         bool ret = false;
 315
 316         /*
 317          * Always check-and-clear the task_work notification signal. With how
 318          * signaling works for task_work, we can find it set with nothing to
 319          * run. We need to clear it for that case, like get_signal() does.
 320          */
 321         if (test_thread_flag(TIF_NOTIFY_SIGNAL))
 322                 clear_notify_signal();
 323         /*
 324          * PF_IO_WORKER never returns to userspace, so check here if we have
 325          * notify work that needs processing.
 326          */
 327         if (current->flags & PF_IO_WORKER) {
 328                 if (test_thread_flag(TIF_NOTIFY_RESUME)) {
 329                         __set_current_state(TASK_RUNNING);
 330                         resume_user_mode_work(NULL);
 331                 }
 332                 if (current->io_uring) {
 333                         unsigned int count = 0;
 334
 335                         __set_current_state(TASK_RUNNING);
 336                         tctx_task_work_run(current->io_uring, UINT_MAX, &count);
 337                         if (count)
 338                                 ret = true;
 339                 }
 340         }
 341         if (task_work_pending(current)) {
 342                 __set_current_state(TASK_RUNNING);
 343                 task_work_run();
 344                 ret = true;
 345         }
 346
 347         return ret;
 348 }
 349
 350 static inline bool io_local_work_pending(struct io_ring_ctx *ctx)
 351 {
 352         return !llist_empty(&ctx->work_llist) || !llist_empty(&ctx->retry_llist);
 353 }
 354
 355 static inline bool io_task_work_pending(struct io_ring_ctx *ctx)
 356 {
 357         return task_work_pending(current) || io_local_work_pending(ctx);
 358 }
 359
 360 static inline void io_tw_lock(struct io_ring_ctx *ctx, struct io_tw_state *ts)
 361 {
 362         lockdep_assert_held(&ctx->uring_lock);
 363 }
 364
 365 /*
 366  * Don't complete immediately but use deferred completion infrastructure.
 367  * Protected by ->uring_lock and can only be used either with
 368  * IO_URING_F_COMPLETE_DEFER or inside a tw handler holding the mutex.
 369  */
 370 static inline void io_req_complete_defer(struct io_kiocb *req)
 371         __must_hold(&req->ctx->uring_lock)
 372 {
 373         struct io_submit_state *state = &req->ctx->submit_state;
 374
 375         lockdep_assert_held(&req->ctx->uring_lock);
 376
 377         wq_list_add_tail(&req->comp_list, &state->compl_reqs);
 378 }
 379
 380 static inline void io_commit_cqring_flush(struct io_ring_ctx *ctx)
 381 {
 382         if (unlikely(ctx->off_timeout_used || ctx->drain_active ||
 383                      ctx->has_evfd || ctx->poll_activated))
 384                 __io_commit_cqring_flush(ctx);
 385 }
 386
 387 static inline void io_get_task_refs(int nr)
 388 {
 389         struct io_uring_task *tctx = current->io_uring;
 390
 391         tctx->cached_refs -= nr;
 392         if (unlikely(tctx->cached_refs < 0))
 393                 io_task_refs_refill(tctx);
 394 }
 395
 396 static inline bool io_req_cache_empty(struct io_ring_ctx *ctx)
 397 {
 398         return !ctx->submit_state.free_list.next;
 399 }
 400
 401 extern struct kmem_cache *req_cachep;
 402 extern struct kmem_cache *io_buf_cachep;
 403
 404 static inline struct io_kiocb *io_extract_req(struct io_ring_ctx *ctx)
 405 {
 406         struct io_kiocb *req;
 407
 408         req = container_of(ctx->submit_state.free_list.next, struct io_kiocb, comp_list);
 409         wq_stack_extract(&ctx->submit_state.free_list);
 410         return req;
 411 }
 412
 413 static inline bool io_alloc_req(struct io_ring_ctx *ctx, struct io_kiocb **req)
 414 {
 415         if (unlikely(io_req_cache_empty(ctx))) {
 416                 if (!__io_alloc_req_refill(ctx))
 417                         return false;
 418         }
 419         *req = io_extract_req(ctx);
 420         return true;
 421 }
 422
 423 static inline bool io_allowed_defer_tw_run(struct io_ring_ctx *ctx)
 424 {
 425         return likely(ctx->submitter_task == current);
 426 }
 427
 428 static inline bool io_allowed_run_tw(struct io_ring_ctx *ctx)
 429 {
 430         return likely(!(ctx->flags & IORING_SETUP_DEFER_TASKRUN) ||
 431                       ctx->submitter_task == current);
 432 }
 433
 434 /*
 435  * Terminate the request if either of these conditions are true:
 436  *
 437  * 1) It's being executed by the original task, but that task is marked
 438  *    with PF_EXITING as it's exiting.
 439  * 2) PF_KTHREAD is set, in which case the invoker of the task_work is
 440  *    our fallback task_work.
 441  */
 442 static inline bool io_should_terminate_tw(void)
 443 {
 444         return current->flags & (PF_KTHREAD | PF_EXITING);
 445 }
 446
 447 static inline void io_req_queue_tw_complete(struct io_kiocb *req, s32 res)
 448 {
 449         io_req_set_res(req, res, 0);
 450         req->io_task_work.func = io_req_task_complete;
 451         io_req_task_work_add(req);
 452 }
 453
 454 /*
 455  * IORING_SETUP_SQE128 contexts allocate twice the normal SQE size for each
 456  * slot.
 457  */
 458 static inline size_t uring_sqe_size(struct io_ring_ctx *ctx)
 459 {
 460         if (ctx->flags & IORING_SETUP_SQE128)
 461                 return 2 * sizeof(struct io_uring_sqe);
 462         return sizeof(struct io_uring_sqe);
 463 }
 464
 465 static inline bool io_file_can_poll(struct io_kiocb *req)
 466 {
 467         if (req->flags & REQ_F_CAN_POLL)
 468                 return true;
 469         if (req->file && file_can_poll(req->file)) {
 470                 req->flags |= REQ_F_CAN_POLL;
 471                 return true;
 472         }
 473         return false;
 474 }
 475
 476 static inline ktime_t io_get_time(struct io_ring_ctx *ctx)
 477 {
 478         if (ctx->clockid == CLOCK_MONOTONIC)
 479                 return ktime_get();
 480
 481         return ktime_get_with_offset(ctx->clock_offset);
 482 }
 483
 484 enum {
 485         IO_CHECK_CQ_OVERFLOW_BIT,
 486         IO_CHECK_CQ_DROPPED_BIT,
 487 };
 488
 489 static inline bool io_has_work(struct io_ring_ctx *ctx)
 490 {
 491         return test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq) ||
 492                io_local_work_pending(ctx);
 493 }
 494 #endif