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