projects/linux·blob master · refs

363 lines · 10.0 KiB

1// SPDX-License-Identifier: GPL-2.0
2/*
3 * Task work handling for io_uring
4 */
5#include <linux/kernel.h>
6#include <linux/errno.h>
7#include <linux/sched/signal.h>
8#include <linux/io_uring.h>
9#include <linux/indirect_call_wrapper.h>
10
11#include "io_uring.h"
12#include "tctx.h"
13#include "poll.h"
14#include "rw.h"
15#include "eventfd.h"
16#include "wait.h"
17#include "mpscq.h"
18
19static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw)
20{
21 if (!ctx)
22 return;
23 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
24 atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
25
26 io_submit_flush_completions(ctx);
27 mutex_unlock(&ctx->uring_lock);
28 percpu_ref_put(&ctx->refs);
29}
30
31void io_tctx_fallback_work(struct work_struct *work)
32{
33 struct io_uring_task *tctx = container_of(work, struct io_uring_task,
34 fallback_work);
35 unsigned int count = 0;
36
37 /*
38 * Run the entries directly. We're in PF_KTHRED context, hence
39 * io_should_terminate_tw() is true and they will be marked as
40 * canceled.
41 */
42 tctx_task_work_run(tctx, UINT_MAX, &count);
43 put_task_struct(tctx->task);
44}
45
46static void io_fallback_tw(struct io_uring_task *tctx)
47{
48 /*
49 * The task ref both keeps ->task valid and, as __io_uring_free() is
50 * only called when the task itself is freed, ensures the tctx (and
51 * the queued work) stay around until the drain has run.
52 */
53 get_task_struct(tctx->task);
54 if (!queue_work(system_dfl_wq, &tctx->fallback_work))
55 put_task_struct(tctx->task);
56}
57
58/*
59 * Run queued task_work, processing no more than max_entries, with the number
60 * of entries processed added to *count. If more entries than max_entries are
61 * available, the remainder simply stay on the queue for the next run.
62 */
63void tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries,
64 unsigned int *count)
65{
66 struct io_ring_ctx *ctx = NULL;
67 struct io_tw_state ts = { };
68
69 while (*count < max_entries) {
70 struct llist_node *node = mpscq_pop(&tctx->task_list,
71 &tctx->task_head);
72 struct io_kiocb *req;
73
74 if (!node) {
75 if (mpscq_empty(&tctx->task_list))
76 break;
77 /*
78 * A producer has published a node but hasn't
79 * linked it into the queue yet (see mpscq_pop()).
80 * Give it a chance to finish rather than spinning,
81 * and don't sit on the ctx lock while doing so.
82 */
83 ctx_flush_and_put(ctx, ts);
84 ctx = NULL;
85 cond_resched();
86 continue;
87 }
88 req = container_of(node, struct io_kiocb, io_task_work.node);
89 if (req->ctx != ctx) {
90 ctx_flush_and_put(ctx, ts);
91 ctx = req->ctx;
92 mutex_lock(&ctx->uring_lock);
93 percpu_ref_get(&ctx->refs);
94 ts.cancel = io_should_terminate_tw(ctx);
95 }
96 INDIRECT_CALL_2(req->io_task_work.func,
97 io_poll_task_func, io_req_rw_complete,
98 (struct io_tw_req){req}, ts);
99 (*count)++;
100 /*
101 * Break if most recent pop emptied the queue. This helps
102 * bound task_work run, and also protects the regular
103 * task_work addition.
104 */
105 if (mpscq_pop_emptied(&tctx->task_list, tctx->task_head))
106 break;
107 if (unlikely(need_resched())) {
108 ctx_flush_and_put(ctx, ts);
109 ctx = NULL;
110 cond_resched();
111 }
112 }
113 ctx_flush_and_put(ctx, ts);
114
115 /*
116 * Relaxed read is enough as only the task itself sets ->in_cancel.
117 * The tctx may also be drained by io_tctx_fallback_work(), in which
118 * case current is a kworker that has no tctx refs to drop.
119 */
120 if (unlikely(atomic_read(&tctx->in_cancel)) &&
121 current->io_uring == tctx)
122 io_uring_drop_tctx_refs(current);
123
124 trace_io_uring_task_work_run(tctx, *count);
125}
126
127void tctx_task_work(struct callback_head *cb)
128{
129 struct io_uring_task *tctx;
130 unsigned int count = 0;
131
132 tctx = container_of(cb, struct io_uring_task, task_work);
133 tctx_task_work_run(tctx, UINT_MAX, &count);
134}
135
136/*
137 * Sets IORING_SQ_TASKRUN in the sq_flags shared with userspace, using the
138 * RCU protected rings pointer to be safe against concurrent ring resizing.
139 */
140static void io_ctx_mark_taskrun(struct io_ring_ctx *ctx)
141{
142 lockdep_assert_in_rcu_read_lock();
143
144 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) {
145 struct io_rings *rings = rcu_dereference(ctx->rings_rcu);
146
147 atomic_or(IORING_SQ_TASKRUN, &rings->sq_flags);
148 }
149}
150
151void io_req_local_work_add(struct io_kiocb *req, unsigned flags)
152{
153 struct io_ring_ctx *ctx = req->ctx;
154 int nr_wait;
155
156 /* pairs with synchronize_rcu() in io_ring_exit_work() */
157 guard(rcu)();
158
159 /*
160 * We don't know how many requests there are in the link and whether
161 * they can even be queued lazily, fall back to non-lazy.
162 */
163 if (req->flags & IO_REQ_LINK_FLAGS)
164 flags &= ~IOU_F_TWQ_LAZY_WAKE;
165
166 /*
167 * The xchg() in mpscq_push() implies a full barrier, which pairs with
168 * the barrier in set_current_state() on the io_cqring_wait() side. This
169 * ensures that either we see the updated ->cq_wait_nr, or waiters going
170 * to sleep will observe the work added to the list, which is similar to
171 * the wait/wake task state sync.
172 */
173 if (mpscq_push(&ctx->work_list, &req->io_task_work.node)) {
174 io_ctx_mark_taskrun(ctx);
175 if (data_race(ctx->int_flags) & IO_RING_F_HAS_EVFD)
176 io_eventfd_signal(ctx, false, flags & IOU_F_TWQ_IN_WAKE);
177 }
178
179 /*
180 * No one is waiting (IO_CQ_WAKE_INIT), or this cycle's wake up has
181 * already been issued (zero or negative, see below).
182 */
183 nr_wait = atomic_read(&ctx->cq_wait_nr);
184 if (nr_wait <= 0)
185 return;
186 if (flags & IOU_F_TWQ_LAZY_WAKE) {
187 /*
188 * ->cq_wait_nr counts down the number of lazy adds, once it
189 * hits zero we're good to wake the waiter. A producer that
190 * gets delayed between pushing its entry and getting here
191 * may count down a later wait cycle. That's OK, it'll be an
192 * early wake, not a lost one.
193 */
194 if (!atomic_dec_and_test(&ctx->cq_wait_nr))
195 return;
196 } else if (atomic_xchg(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT) <= 0) {
197 /*
198 * Potentially raced with lazy add, claim the wake. A value
199 * <= 0 means a lazy add hit zero or another forced add
200 * claimed IO_CQ_WAKE_INIT. Either way, the wake up for this
201 * wait cycle has already been done.
202 */
203 return;
204 }
205 wake_up_state(ctx->submitter_task, TASK_INTERRUPTIBLE);
206}
207
208void io_req_normal_work_add(struct io_kiocb *req)
209{
210 struct io_uring_task *tctx = req->tctx;
211 struct io_ring_ctx *ctx = req->ctx;
212
213 /* SQPOLL can retire the request on push, see io_ring_exit_work() */
214 guard(rcu)();
215
216 /* tw run already pending, nothing else to do */
217 if (!mpscq_push(&tctx->task_list, &req->io_task_work.node))
218 return;
219
220 /*
221 * Doesn't need to use ->rings_rcu, as resizing isn't supported for
222 * !DEFER_TASKRUN.
223 */
224 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
225 atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
226
227 /* SQPOLL doesn't need the task_work added, it'll run it itself */
228 if (ctx->flags & IORING_SETUP_SQPOLL) {
229 __set_notify_signal(tctx->task);
230 return;
231 }
232
233 if (likely(!task_work_add(tctx->task, &tctx->task_work, ctx->notify_method)))
234 return;
235
236 io_fallback_tw(tctx);
237}
238
239void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags)
240{
241 if (WARN_ON_ONCE(!(req->ctx->flags & IORING_SETUP_DEFER_TASKRUN)))
242 return;
243 __io_req_task_work_add(req, flags);
244}
245
246void __cold io_cancel_local_task_work(struct io_ring_ctx *ctx)
247{
248 struct io_tw_state ts = { .cancel = true };
249 struct llist_node *node;
250
251 /*
252 * The work list consumer side is serialized by ->uring_lock, see
253 * __io_run_local_work(). Grab it to guard against racing with normal
254 * task_work running, as the task may be exiting. The ring is going
255 * away, run the entries in cancel mode right here - the callers
256 * provide the same process context the per-ctx fallback work that
257 * they were previously punted to ran in.
258 */
259 guard(mutex)(&ctx->uring_lock);
260
261 while (!mpscq_empty(&ctx->work_list)) {
262 struct io_kiocb *req;
263
264 node = mpscq_pop(&ctx->work_list, &ctx->work_head);
265 if (!node) {
266 /* a producer is mid-push, wait for it to link */
267 cond_resched();
268 continue;
269 }
270 req = container_of(node, struct io_kiocb, io_task_work.node);
271 req->io_task_work.func((struct io_tw_req){req}, ts);
272 }
273 io_submit_flush_completions(ctx);
274}
275
276static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events,
277 int min_events)
278{
279 if (!io_local_work_pending(ctx))
280 return false;
281 if (events < min_events)
282 return true;
283 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
284 atomic_or(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
285 return false;
286}
287
288static int __io_run_local_work_loop(struct io_ring_ctx *ctx,
289 io_tw_token_t tw,
290 int events)
291{
292 int ret = 0;
293
294 while (ret < events) {
295 struct llist_node *node = mpscq_pop(&ctx->work_list, &ctx->work_head);
296 struct io_kiocb *req;
297
298 if (!node)
299 break;
300 req = container_of(node, struct io_kiocb, io_task_work.node);
301 INDIRECT_CALL_2(req->io_task_work.func,
302 io_poll_task_func, io_req_rw_complete,
303 (struct io_tw_req){req}, tw);
304 ret++;
305 }
306
307 return ret;
308}
309
310static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw,
311 int min_events, int max_events)
312{
313 unsigned int loops = 0;
314 int ret = 0;
315
316 if (WARN_ON_ONCE(ctx->submitter_task != current))
317 return -EEXIST;
318 if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
319 atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
320again:
321 /*
322 * If the last loop made no progress while work is still pending,
323 * a producer has published a node but hasn't linked it into the
324 * queue yet (see mpscq_pop()). Give it a chance to finish rather
325 * than spinning on the queue.
326 */
327 if (unlikely(loops && !ret))
328 cond_resched();
329 tw.cancel = io_should_terminate_tw(ctx);
330 min_events -= ret;
331 ret = __io_run_local_work_loop(ctx, tw, max_events);
332 loops++;
333
334 if (io_run_local_work_continue(ctx, ret, min_events))
335 goto again;
336 io_submit_flush_completions(ctx);
337 if (io_run_local_work_continue(ctx, ret, min_events))
338 goto again;
339
340 trace_io_uring_local_work_run(ctx, ret, loops);
341 return ret;
342}
343
344int io_run_local_work_locked(struct io_ring_ctx *ctx, int min_events)
345{
346 struct io_tw_state ts = {};
347
348 if (!io_local_work_pending(ctx))
349 return 0;
350 return __io_run_local_work(ctx, ts, min_events,
351 max(IO_LOCAL_TW_DEFAULT_MAX, min_events));
352}
353
354int io_run_local_work(struct io_ring_ctx *ctx, int min_events, int max_events)
355{
356 struct io_tw_state ts = {};
357 int ret;
358
359 mutex_lock(&ctx->uring_lock);
360 ret = __io_run_local_work(ctx, ts, min_events, max_events);
361 mutex_unlock(&ctx->uring_lock);
362 return ret;
363}