niansa-misc/linux
1
0
Fork 0
mirror of synced 2025-03-06 20:59:54 +01:00
Code Issues Projects Releases Packages Wiki Activity

io_uring: patch up IOPOLL overflow_flush sync

Browse source 
IOPOLL skips completion locking but keeps it under uring_lock, thus
io_cqring_overflow_flush() and so io_cqring_events() need additional
locking with uring_lock in some cases for IOPOLL.

Remove __io_cqring_overflow_flush() from io_cqring_events(), introduce a
wrapper around flush doing needed synchronisation and call it by hand.

Cc: stable@vger.kernel.org # 5.5+
Signed-off-by: Pavel Begunkov <asml.silence@gmail.com>
Signed-off-by: Jens Axboe <axboe@kernel.dk>
This commit is contained in:
Pavel Begunkov 2021-01-04 20:36:36 +00:00 • committed by Jens Axboe
parent 81b6d05cca
commit 6c503150ae
1 changed files with 41 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
fs/io_uring.c
View file

@ -1713,7 +1713,7 @@ static void io_cqring_ev_posted(struct io_ring_ctx *ctx)
} }
/* Returns true if there are no backlogged entries after the flush */ /* Returns true if there are no backlogged entries after the flush */
static bool io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force, static bool __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force,
struct task_struct *tsk, struct task_struct *tsk,
struct files_struct *files) struct files_struct *files)
{ {
@ -1768,6 +1768,20 @@ static bool io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force,
return all_flushed; return all_flushed;
} }
static void io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force,
struct task_struct *tsk,
struct files_struct *files)
{
if (test_bit(0, &ctx->cq_check_overflow)) {
/* iopoll syncs against uring_lock, not completion_lock */
if (ctx->flags & IORING_SETUP_IOPOLL)
mutex_lock(&ctx->uring_lock);
__io_cqring_overflow_flush(ctx, force, tsk, files);
if (ctx->flags & IORING_SETUP_IOPOLL)
mutex_unlock(&ctx->uring_lock);
}
}
static void __io_cqring_fill_event(struct io_kiocb *req, long res, long cflags) static void __io_cqring_fill_event(struct io_kiocb *req, long res, long cflags)
{ {
struct io_ring_ctx *ctx = req->ctx; struct io_ring_ctx *ctx = req->ctx;
@ -2314,20 +2328,8 @@ static void io_double_put_req(struct io_kiocb *req)
io_free_req(req); io_free_req(req);
} }
static unsigned io_cqring_events(struct io_ring_ctx *ctx, bool noflush) static unsigned io_cqring_events(struct io_ring_ctx *ctx)
{ {
if (test_bit(0, &ctx->cq_check_overflow)) {
/*
* noflush == true is from the waitqueue handler, just ensure
* we wake up the task, and the next invocation will flush the
* entries. We cannot safely to it from here.
*/
if (noflush)
return -1U;
io_cqring_overflow_flush(ctx, false, NULL, NULL);
}
/* See comment at the top of this file */ /* See comment at the top of this file */
smp_rmb(); smp_rmb();
return __io_cqring_events(ctx); return __io_cqring_events(ctx);
@ -2552,7 +2554,9 @@ static int io_iopoll_check(struct io_ring_ctx *ctx, long min)
* If we do, we can potentially be spinning for commands that * If we do, we can potentially be spinning for commands that
* already triggered a CQE (eg in error). * already triggered a CQE (eg in error).
*/ */
if (io_cqring_events(ctx, false)) if (test_bit(0, &ctx->cq_check_overflow))
__io_cqring_overflow_flush(ctx, false, NULL, NULL);
if (io_cqring_events(ctx))
break; break;
/* /*
@ -6827,7 +6831,7 @@ static int io_submit_sqes(struct io_ring_ctx *ctx, unsigned int nr)
/* if we have a backlog and couldn't flush it all, return BUSY */ /* if we have a backlog and couldn't flush it all, return BUSY */
if (test_bit(0, &ctx->sq_check_overflow)) { if (test_bit(0, &ctx->sq_check_overflow)) {
if (!io_cqring_overflow_flush(ctx, false, NULL, NULL)) if (!__io_cqring_overflow_flush(ctx, false, NULL, NULL))
return -EBUSY; return -EBUSY;
} }
@ -7090,7 +7094,7 @@ struct io_wait_queue {
unsigned nr_timeouts; unsigned nr_timeouts;
}; };
static inline bool io_should_wake(struct io_wait_queue *iowq, bool noflush) static inline bool io_should_wake(struct io_wait_queue *iowq)
{ {
struct io_ring_ctx *ctx = iowq->ctx; struct io_ring_ctx *ctx = iowq->ctx;
@ -7099,7 +7103,7 @@ static inline bool io_should_wake(struct io_wait_queue *iowq, bool noflush)
* started waiting. For timeouts, we always want to return to userspace, * started waiting. For timeouts, we always want to return to userspace,
* regardless of event count. * regardless of event count.
*/ */
return io_cqring_events(ctx, noflush) >= iowq->to_wait || return io_cqring_events(ctx) >= iowq->to_wait ||
atomic_read(&ctx->cq_timeouts) != iowq->nr_timeouts; atomic_read(&ctx->cq_timeouts) != iowq->nr_timeouts;
} }
@ -7109,11 +7113,13 @@ static int io_wake_function(struct wait_queue_entry *curr, unsigned int mode,
struct io_wait_queue *iowq = container_of(curr, struct io_wait_queue, struct io_wait_queue *iowq = container_of(curr, struct io_wait_queue,
wq); wq);
/* use noflush == true, as we can't safely rely on locking context */ /*
if (!io_should_wake(iowq, true)) * Cannot safely flush overflowed CQEs from here, ensure we wake up
return -1; * the task, and the next invocation will do it.
*/
if (io_should_wake(iowq) || test_bit(0, &iowq->ctx->cq_check_overflow))
return autoremove_wake_function(curr, mode, wake_flags, key); return autoremove_wake_function(curr, mode, wake_flags, key);
return -1;
} }
static int io_run_task_work_sig(void) static int io_run_task_work_sig(void)
@ -7150,7 +7156,8 @@ static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
int ret = 0; int ret = 0;
do { do {
if (io_cqring_events(ctx, false) >= min_events) io_cqring_overflow_flush(ctx, false, NULL, NULL);
if (io_cqring_events(ctx) >= min_events)
return 0; return 0;
if (!io_run_task_work()) if (!io_run_task_work())
break; break;
@ -7178,6 +7185,7 @@ static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
iowq.nr_timeouts = atomic_read(&ctx->cq_timeouts); iowq.nr_timeouts = atomic_read(&ctx->cq_timeouts);
trace_io_uring_cqring_wait(ctx, min_events); trace_io_uring_cqring_wait(ctx, min_events);
do { do {
io_cqring_overflow_flush(ctx, false, NULL, NULL);
prepare_to_wait_exclusive(&ctx->wait, &iowq.wq, prepare_to_wait_exclusive(&ctx->wait, &iowq.wq,
TASK_INTERRUPTIBLE); TASK_INTERRUPTIBLE);
/* make sure we run task_work before checking for signals */ /* make sure we run task_work before checking for signals */
@ -7186,8 +7194,10 @@ static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
continue; continue;
else if (ret < 0) else if (ret < 0)
break; break;
if (io_should_wake(&iowq, false)) if (io_should_wake(&iowq))
break; break;
if (test_bit(0, &ctx->cq_check_overflow))
continue;
if (uts) { if (uts) {
timeout = schedule_timeout(timeout); timeout = schedule_timeout(timeout);
if (timeout == 0) { if (timeout == 0) {
@ -8625,7 +8635,8 @@ static __poll_t io_uring_poll(struct file *file, poll_table *wait)
smp_rmb(); smp_rmb();
if (!io_sqring_full(ctx)) if (!io_sqring_full(ctx))
mask |= EPOLLOUT | EPOLLWRNORM; mask |= EPOLLOUT | EPOLLWRNORM;
if (io_cqring_events(ctx, false)) io_cqring_overflow_flush(ctx, false, NULL, NULL);
if (io_cqring_events(ctx))
mask |= EPOLLIN | EPOLLRDNORM; mask |= EPOLLIN | EPOLLRDNORM;
return mask; return mask;
@ -8683,7 +8694,7 @@ static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
/* if force is set, the ring is going away. always drop after that */ /* if force is set, the ring is going away. always drop after that */
ctx->cq_overflow_flushed = 1; ctx->cq_overflow_flushed = 1;
if (ctx->rings) if (ctx->rings)
io_cqring_overflow_flush(ctx, true, NULL, NULL); __io_cqring_overflow_flush(ctx, true, NULL, NULL);
mutex_unlock(&ctx->uring_lock); mutex_unlock(&ctx->uring_lock);
io_kill_timeouts(ctx, NULL, NULL); io_kill_timeouts(ctx, NULL, NULL);
@ -8857,9 +8868,7 @@ static void io_uring_cancel_task_requests(struct io_ring_ctx *ctx,
} }
io_cancel_defer_files(ctx, task, files); io_cancel_defer_files(ctx, task, files);
io_ring_submit_lock(ctx, (ctx->flags & IORING_SETUP_IOPOLL));
io_cqring_overflow_flush(ctx, true, task, files); io_cqring_overflow_flush(ctx, true, task, files);
io_ring_submit_unlock(ctx, (ctx->flags & IORING_SETUP_IOPOLL));
if (!files) if (!files)
__io_uring_cancel_task_requests(ctx, task); __io_uring_cancel_task_requests(ctx, task);
@ -9195,13 +9204,8 @@ SYSCALL_DEFINE6(io_uring_enter, unsigned int, fd, u32, to_submit,
*/ */
ret = 0; ret = 0;
if (ctx->flags & IORING_SETUP_SQPOLL) { if (ctx->flags & IORING_SETUP_SQPOLL) {
if (!list_empty_careful(&ctx->cq_overflow_list)) {
bool needs_lock = ctx->flags & IORING_SETUP_IOPOLL;
io_ring_submit_lock(ctx, needs_lock);
io_cqring_overflow_flush(ctx, false, NULL, NULL); io_cqring_overflow_flush(ctx, false, NULL, NULL);
io_ring_submit_unlock(ctx, needs_lock);
}
if (flags & IORING_ENTER_SQ_WAKEUP) if (flags & IORING_ENTER_SQ_WAKEUP)
wake_up(&ctx->sq_data->wait); wake_up(&ctx->sq_data->wait);
if (flags & IORING_ENTER_SQ_WAIT) if (flags & IORING_ENTER_SQ_WAIT)