io_uring/sqpoll.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 /*
   3  * Contains the core associated with submission side polling of the SQ
   4  * ring, offloading submissions from the application to a kernel thread.
   5  */
   6 #include <linux/kernel.h>
   7 #include <linux/errno.h>
   8 #include <linux/file.h>
   9 #include <linux/mm.h>
  10 #include <linux/slab.h>
  11 #include <linux/audit.h>
  12 #include <linux/security.h>
  13 #include <linux/io_uring.h>
  14
  15 #include <uapi/linux/io_uring.h>
  16
  17 #include "io_uring.h"
  18 #include "napi.h"
  19 #include "sqpoll.h"
  20
  21 #define IORING_SQPOLL_CAP_ENTRIES_VALUE 8
  22 #define IORING_TW_CAP_ENTRIES_VALUE     8
  23
  24 enum {
  25         IO_SQ_THREAD_SHOULD_STOP = 0,
  26         IO_SQ_THREAD_SHOULD_PARK,
  27 };
  28
  29 void io_sq_thread_unpark(struct io_sq_data *sqd)
  30         __releases(&sqd->lock)
  31 {
  32         WARN_ON_ONCE(sqd->thread == current);
  33
  34         /*
  35          * Do the dance but not conditional clear_bit() because it'd race with
  36          * other threads incrementing park_pending and setting the bit.
  37          */
  38         clear_bit(IO_SQ_THREAD_SHOULD_PARK, &sqd->state);
  39         if (atomic_dec_return(&sqd->park_pending))
  40                 set_bit(IO_SQ_THREAD_SHOULD_PARK, &sqd->state);
  41         mutex_unlock(&sqd->lock);
  42 }
  43
  44 void io_sq_thread_park(struct io_sq_data *sqd)
  45         __acquires(&sqd->lock)
  46 {
  47         WARN_ON_ONCE(sqd->thread == current);
  48
  49         atomic_inc(&sqd->park_pending);
  50         set_bit(IO_SQ_THREAD_SHOULD_PARK, &sqd->state);
  51         mutex_lock(&sqd->lock);
  52         if (sqd->thread)
  53                 wake_up_process(sqd->thread);
  54 }
  55
  56 void io_sq_thread_stop(struct io_sq_data *sqd)
  57 {
  58         WARN_ON_ONCE(sqd->thread == current);
  59         WARN_ON_ONCE(test_bit(IO_SQ_THREAD_SHOULD_STOP, &sqd->state));
  60
  61         set_bit(IO_SQ_THREAD_SHOULD_STOP, &sqd->state);
  62         mutex_lock(&sqd->lock);
  63         if (sqd->thread)
  64                 wake_up_process(sqd->thread);
  65         mutex_unlock(&sqd->lock);
  66         wait_for_completion(&sqd->exited);
  67 }
  68
  69 void io_put_sq_data(struct io_sq_data *sqd)
  70 {
  71         if (refcount_dec_and_test(&sqd->refs)) {
  72                 WARN_ON_ONCE(atomic_read(&sqd->park_pending));
  73
  74                 io_sq_thread_stop(sqd);
  75                 kfree(sqd);
  76         }
  77 }
  78
  79 static __cold void io_sqd_update_thread_idle(struct io_sq_data *sqd)
  80 {
  81         struct io_ring_ctx *ctx;
  82         unsigned sq_thread_idle = 0;
  83
  84         list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)
  85                 sq_thread_idle = max(sq_thread_idle, ctx->sq_thread_idle);
  86         sqd->sq_thread_idle = sq_thread_idle;
  87 }
  88
  89 void io_sq_thread_finish(struct io_ring_ctx *ctx)
  90 {
  91         struct io_sq_data *sqd = ctx->sq_data;
  92
  93         if (sqd) {
  94                 io_sq_thread_park(sqd);
  95                 list_del_init(&ctx->sqd_list);
  96                 io_sqd_update_thread_idle(sqd);
  97                 io_sq_thread_unpark(sqd);
  98
  99                 io_put_sq_data(sqd);
 100                 ctx->sq_data = NULL;
 101         }
 102 }
 103
 104 static struct io_sq_data *io_attach_sq_data(struct io_uring_params *p)
 105 {
 106         struct io_ring_ctx *ctx_attach;
 107         struct io_sq_data *sqd;
 108         struct fd f;
 109
 110         f = fdget(p->wq_fd);
 111         if (!f.file)
 112                 return ERR_PTR(-ENXIO);
 113         if (!io_is_uring_fops(f.file)) {
 114                 fdput(f);
 115                 return ERR_PTR(-EINVAL);
 116         }
 117
 118         ctx_attach = f.file->private_data;
 119         sqd = ctx_attach->sq_data;
 120         if (!sqd) {
 121                 fdput(f);
 122                 return ERR_PTR(-EINVAL);
 123         }
 124         if (sqd->task_tgid != current->tgid) {
 125                 fdput(f);
 126                 return ERR_PTR(-EPERM);
 127         }
 128
 129         refcount_inc(&sqd->refs);
 130         fdput(f);
 131         return sqd;
 132 }
 133
 134 static struct io_sq_data *io_get_sq_data(struct io_uring_params *p,
 135                                          bool *attached)
 136 {
 137         struct io_sq_data *sqd;
 138
 139         *attached = false;
 140         if (p->flags & IORING_SETUP_ATTACH_WQ) {
 141                 sqd = io_attach_sq_data(p);
 142                 if (!IS_ERR(sqd)) {
 143                         *attached = true;
 144                         return sqd;
 145                 }
 146                 /* fall through for EPERM case, setup new sqd/task */
 147                 if (PTR_ERR(sqd) != -EPERM)
 148                         return sqd;
 149         }
 150
 151         sqd = kzalloc(sizeof(*sqd), GFP_KERNEL);
 152         if (!sqd)
 153                 return ERR_PTR(-ENOMEM);
 154
 155         atomic_set(&sqd->park_pending, 0);
 156         refcount_set(&sqd->refs, 1);
 157         INIT_LIST_HEAD(&sqd->ctx_list);
 158         mutex_init(&sqd->lock);
 159         init_waitqueue_head(&sqd->wait);
 160         init_completion(&sqd->exited);
 161         return sqd;
 162 }
 163
 164 static inline bool io_sqd_events_pending(struct io_sq_data *sqd)
 165 {
 166         return READ_ONCE(sqd->state);
 167 }
 168
 169 static int __io_sq_thread(struct io_ring_ctx *ctx, bool cap_entries)
 170 {
 171         unsigned int to_submit;
 172         int ret = 0;
 173
 174         to_submit = io_sqring_entries(ctx);
 175         /* if we're handling multiple rings, cap submit size for fairness */
 176         if (cap_entries && to_submit > IORING_SQPOLL_CAP_ENTRIES_VALUE)
 177                 to_submit = IORING_SQPOLL_CAP_ENTRIES_VALUE;
 178
 179         if (!wq_list_empty(&ctx->iopoll_list) || to_submit) {
 180                 const struct cred *creds = NULL;
 181
 182                 if (ctx->sq_creds != current_cred())
 183                         creds = override_creds(ctx->sq_creds);
 184
 185                 mutex_lock(&ctx->uring_lock);
 186                 if (!wq_list_empty(&ctx->iopoll_list))
 187                         io_do_iopoll(ctx, true);
 188
 189                 /*
 190                  * Don't submit if refs are dying, good for io_uring_register(),
 191                  * but also it is relied upon by io_ring_exit_work()
 192                  */
 193                 if (to_submit && likely(!percpu_ref_is_dying(&ctx->refs)) &&
 194                     !(ctx->flags & IORING_SETUP_R_DISABLED))
 195                         ret = io_submit_sqes(ctx, to_submit);
 196                 mutex_unlock(&ctx->uring_lock);
 197
 198                 if (io_napi(ctx))
 199                         ret += io_napi_sqpoll_busy_poll(ctx);
 200
 201                 if (to_submit && wq_has_sleeper(&ctx->sqo_sq_wait))
 202                         wake_up(&ctx->sqo_sq_wait);
 203                 if (creds)
 204                         revert_creds(creds);
 205         }
 206
 207         return ret;
 208 }
 209
 210 static bool io_sqd_handle_event(struct io_sq_data *sqd)
 211 {
 212         bool did_sig = false;
 213         struct ksignal ksig;
 214
 215         if (test_bit(IO_SQ_THREAD_SHOULD_PARK, &sqd->state) ||
 216             signal_pending(current)) {
 217                 mutex_unlock(&sqd->lock);
 218                 if (signal_pending(current))
 219                         did_sig = get_signal(&ksig);
 220                 cond_resched();
 221                 mutex_lock(&sqd->lock);
 222                 sqd->sq_cpu = raw_smp_processor_id();
 223         }
 224         return did_sig || test_bit(IO_SQ_THREAD_SHOULD_STOP, &sqd->state);
 225 }
 226
 227 /*
 228  * Run task_work, processing the retry_list first. The retry_list holds
 229  * entries that we passed on in the previous run, if we had more task_work
 230  * than we were asked to process. Newly queued task_work isn't run until the
 231  * retry list has been fully processed.
 232  */
 233 static unsigned int io_sq_tw(struct llist_node **retry_list, int max_entries)
 234 {
 235         struct io_uring_task *tctx = current->io_uring;
 236         unsigned int count = 0;
 237
 238         if (*retry_list) {
 239                 *retry_list = io_handle_tw_list(*retry_list, &count, max_entries);
 240                 if (count >= max_entries)
 241                         return count;
 242                 max_entries -= count;
 243         }
 244
 245         *retry_list = tctx_task_work_run(tctx, max_entries, &count);
 246         return count;
 247 }
 248
 249 static bool io_sq_tw_pending(struct llist_node *retry_list)
 250 {
 251         struct io_uring_task *tctx = current->io_uring;
 252
 253         return retry_list || !llist_empty(&tctx->task_list);
 254 }
 255
 256 static void io_sq_update_worktime(struct io_sq_data *sqd, struct rusage *start)
 257 {
 258         struct rusage end;
 259
 260         getrusage(current, RUSAGE_SELF, &end);
 261         end.ru_stime.tv_sec -= start->ru_stime.tv_sec;
 262         end.ru_stime.tv_usec -= start->ru_stime.tv_usec;
 263
 264         sqd->work_time += end.ru_stime.tv_usec + end.ru_stime.tv_sec * 1000000;
 265 }
 266
 267 static int io_sq_thread(void *data)
 268 {
 269         struct llist_node *retry_list = NULL;
 270         struct io_sq_data *sqd = data;
 271         struct io_ring_ctx *ctx;
 272         struct rusage start;
 273         unsigned long timeout = 0;
 274         char buf[TASK_COMM_LEN];
 275         DEFINE_WAIT(wait);
 276
 277         /* offload context creation failed, just exit */
 278         if (!current->io_uring)
 279                 goto err_out;
 280
 281         snprintf(buf, sizeof(buf), "iou-sqp-%d", sqd->task_pid);
 282         set_task_comm(current, buf);
 283
 284         /* reset to our pid after we've set task_comm, for fdinfo */
 285         sqd->task_pid = current->pid;
 286
 287         if (sqd->sq_cpu != -1) {
 288                 set_cpus_allowed_ptr(current, cpumask_of(sqd->sq_cpu));
 289         } else {
 290                 set_cpus_allowed_ptr(current, cpu_online_mask);
 291                 sqd->sq_cpu = raw_smp_processor_id();
 292         }
 293
 294         /*
 295          * Force audit context to get setup, in case we do prep side async
 296          * operations that would trigger an audit call before any issue side
 297          * audit has been done.
 298          */
 299         audit_uring_entry(IORING_OP_NOP);
 300         audit_uring_exit(true, 0);
 301
 302         mutex_lock(&sqd->lock);
 303         while (1) {
 304                 bool cap_entries, sqt_spin = false;
 305
 306                 if (io_sqd_events_pending(sqd) || signal_pending(current)) {
 307                         if (io_sqd_handle_event(sqd))
 308                                 break;
 309                         timeout = jiffies + sqd->sq_thread_idle;
 310                 }
 311
 312                 cap_entries = !list_is_singular(&sqd->ctx_list);
 313                 getrusage(current, RUSAGE_SELF, &start);
 314                 list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) {
 315                         int ret = __io_sq_thread(ctx, cap_entries);
 316
 317                         if (!sqt_spin && (ret > 0 || !wq_list_empty(&ctx->iopoll_list)))
 318                                 sqt_spin = true;
 319                 }
 320                 if (io_sq_tw(&retry_list, IORING_TW_CAP_ENTRIES_VALUE))
 321                         sqt_spin = true;
 322
 323                 if (sqt_spin || !time_after(jiffies, timeout)) {
 324                         if (sqt_spin) {
 325                                 io_sq_update_worktime(sqd, &start);
 326                                 timeout = jiffies + sqd->sq_thread_idle;
 327                         }
 328                         if (unlikely(need_resched())) {
 329                                 mutex_unlock(&sqd->lock);
 330                                 cond_resched();
 331                                 mutex_lock(&sqd->lock);
 332                                 sqd->sq_cpu = raw_smp_processor_id();
 333                         }
 334                         continue;
 335                 }
 336
 337                 prepare_to_wait(&sqd->wait, &wait, TASK_INTERRUPTIBLE);
 338                 if (!io_sqd_events_pending(sqd) && !io_sq_tw_pending(retry_list)) {
 339                         bool needs_sched = true;
 340
 341                         list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) {
 342                                 atomic_or(IORING_SQ_NEED_WAKEUP,
 343                                                 &ctx->rings->sq_flags);
 344                                 if ((ctx->flags & IORING_SETUP_IOPOLL) &&
 345                                     !wq_list_empty(&ctx->iopoll_list)) {
 346                                         needs_sched = false;
 347                                         break;
 348                                 }
 349
 350                                 /*
 351                                  * Ensure the store of the wakeup flag is not
 352                                  * reordered with the load of the SQ tail
 353                                  */
 354                                 smp_mb__after_atomic();
 355
 356                                 if (io_sqring_entries(ctx)) {
 357                                         needs_sched = false;
 358                                         break;
 359                                 }
 360                         }
 361
 362                         if (needs_sched) {
 363                                 mutex_unlock(&sqd->lock);
 364                                 schedule();
 365                                 mutex_lock(&sqd->lock);
 366                                 sqd->sq_cpu = raw_smp_processor_id();
 367                         }
 368                         list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)
 369                                 atomic_andnot(IORING_SQ_NEED_WAKEUP,
 370                                                 &ctx->rings->sq_flags);
 371                 }
 372
 373                 finish_wait(&sqd->wait, &wait);
 374                 timeout = jiffies + sqd->sq_thread_idle;
 375         }
 376
 377         if (retry_list)
 378                 io_sq_tw(&retry_list, UINT_MAX);
 379
 380         io_uring_cancel_generic(true, sqd);
 381         sqd->thread = NULL;
 382         list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)
 383                 atomic_or(IORING_SQ_NEED_WAKEUP, &ctx->rings->sq_flags);
 384         io_run_task_work();
 385         mutex_unlock(&sqd->lock);
 386 err_out:
 387         complete(&sqd->exited);
 388         do_exit(0);
 389 }
 390
 391 void io_sqpoll_wait_sq(struct io_ring_ctx *ctx)
 392 {
 393         DEFINE_WAIT(wait);
 394
 395         do {
 396                 if (!io_sqring_full(ctx))
 397                         break;
 398                 prepare_to_wait(&ctx->sqo_sq_wait, &wait, TASK_INTERRUPTIBLE);
 399
 400                 if (!io_sqring_full(ctx))
 401                         break;
 402                 schedule();
 403         } while (!signal_pending(current));
 404
 405         finish_wait(&ctx->sqo_sq_wait, &wait);
 406 }
 407
 408 __cold int io_sq_offload_create(struct io_ring_ctx *ctx,
 409                                 struct io_uring_params *p)
 410 {
 411         int ret;
 412
 413         /* Retain compatibility with failing for an invalid attach attempt */
 414         if ((ctx->flags & (IORING_SETUP_ATTACH_WQ | IORING_SETUP_SQPOLL)) ==
 415                                 IORING_SETUP_ATTACH_WQ) {
 416                 struct fd f;
 417
 418                 f = fdget(p->wq_fd);
 419                 if (!f.file)
 420                         return -ENXIO;
 421                 if (!io_is_uring_fops(f.file)) {
 422                         fdput(f);
 423                         return -EINVAL;
 424                 }
 425                 fdput(f);
 426         }
 427         if (ctx->flags & IORING_SETUP_SQPOLL) {
 428                 struct task_struct *tsk;
 429                 struct io_sq_data *sqd;
 430                 bool attached;
 431
 432                 ret = security_uring_sqpoll();
 433                 if (ret)
 434                         return ret;
 435
 436                 sqd = io_get_sq_data(p, &attached);
 437                 if (IS_ERR(sqd)) {
 438                         ret = PTR_ERR(sqd);
 439                         goto err;
 440                 }
 441
 442                 ctx->sq_creds = get_current_cred();
 443                 ctx->sq_data = sqd;
 444                 ctx->sq_thread_idle = msecs_to_jiffies(p->sq_thread_idle);
 445                 if (!ctx->sq_thread_idle)
 446                         ctx->sq_thread_idle = HZ;
 447
 448                 io_sq_thread_park(sqd);
 449                 list_add(&ctx->sqd_list, &sqd->ctx_list);
 450                 io_sqd_update_thread_idle(sqd);
 451                 /* don't attach to a dying SQPOLL thread, would be racy */
 452                 ret = (attached && !sqd->thread) ? -ENXIO : 0;
 453                 io_sq_thread_unpark(sqd);
 454
 455                 if (ret < 0)
 456                         goto err;
 457                 if (attached)
 458                         return 0;
 459
 460                 if (p->flags & IORING_SETUP_SQ_AFF) {
 461                         int cpu = p->sq_thread_cpu;
 462
 463                         ret = -EINVAL;
 464                         if (cpu >= nr_cpu_ids || !cpu_online(cpu))
 465                                 goto err_sqpoll;
 466                         sqd->sq_cpu = cpu;
 467                 } else {
 468                         sqd->sq_cpu = -1;
 469                 }
 470
 471                 sqd->task_pid = current->pid;
 472                 sqd->task_tgid = current->tgid;
 473                 tsk = create_io_thread(io_sq_thread, sqd, NUMA_NO_NODE);
 474                 if (IS_ERR(tsk)) {
 475                         ret = PTR_ERR(tsk);
 476                         goto err_sqpoll;
 477                 }
 478
 479                 sqd->thread = tsk;
 480                 ret = io_uring_alloc_task_context(tsk, ctx);
 481                 wake_up_new_task(tsk);
 482                 if (ret)
 483                         goto err;
 484         } else if (p->flags & IORING_SETUP_SQ_AFF) {
 485                 /* Can't have SQ_AFF without SQPOLL */
 486                 ret = -EINVAL;
 487                 goto err;
 488         }
 489
 490         return 0;
 491 err_sqpoll:
 492         complete(&ctx->sq_data->exited);
 493 err:
 494         io_sq_thread_finish(ctx);
 495         return ret;
 496 }
 497
 498 __cold int io_sqpoll_wq_cpu_affinity(struct io_ring_ctx *ctx,
 499                                      cpumask_var_t mask)
 500 {
 501         struct io_sq_data *sqd = ctx->sq_data;
 502         int ret = -EINVAL;
 503
 504         if (sqd) {
 505                 io_sq_thread_park(sqd);
 506                 /* Don't set affinity for a dying thread */
 507                 if (sqd->thread)
 508                         ret = io_wq_cpu_affinity(sqd->thread->io_uring, mask);
 509                 io_sq_thread_unpark(sqd);
 510         }
 511
 512         return ret;
 513 }