projects/linux·blob master · refs

564 lines · 12.9 KiB

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