Skip to content

Commit

Permalink
io_uring: tie async worker side to the task context
Browse files Browse the repository at this point in the history 
Move it outside of the io_ring_ctx, and tie it to the io_uring task
context.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
  • Loading branch information
Jens Axboe committed Feb 22, 2021
1 parent 3b094e7 commit 5aa75ed
Show file tree
Hide file tree
Showing 2 changed files with 35 additions and 50 deletions.
84 changes: 34 additions & 50 deletions fs/io_uring.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -366,9 +366,6 @@ struct io_ring_ctx {

struct io_rings *rings;

/* IO offload */
struct io_wq *io_wq;

/*
* For SQPOLL usage - we hold a reference to the parent task, so we
* have access to the ->files
Expand Down Expand Up @@ -1634,10 +1631,11 @@ static struct io_kiocb *__io_queue_async_work(struct io_kiocb *req)
{
struct io_ring_ctx *ctx = req->ctx;
struct io_kiocb *link = io_prep_linked_timeout(req);
struct io_uring_task *tctx = req->task->io_uring;

trace_io_uring_queue_async_work(ctx, io_wq_is_hashed(&req->work), req,
&req->work, req->flags);
io_wq_enqueue(ctx->io_wq, &req->work);
io_wq_enqueue(tctx->io_wq, &req->work);
return link;
}

Expand Down Expand Up @@ -5960,12 +5958,15 @@ static bool io_cancel_cb(struct io_wq_work *work, void *data)
return req->user_data == (unsigned long) data;
}

static int io_async_cancel_one(struct io_ring_ctx *ctx, void *sqe_addr)
static int io_async_cancel_one(struct io_uring_task *tctx, void *sqe_addr)
{
enum io_wq_cancel cancel_ret;
int ret = 0;

cancel_ret = io_wq_cancel_cb(ctx->io_wq, io_cancel_cb, sqe_addr, false);
if (!tctx->io_wq)
return -ENOENT;

cancel_ret = io_wq_cancel_cb(tctx->io_wq, io_cancel_cb, sqe_addr, false);
switch (cancel_ret) {
case IO_WQ_CANCEL_OK:
ret = 0;
Expand All @@ -5988,7 +5989,8 @@ static void io_async_find_and_cancel(struct io_ring_ctx *ctx,
unsigned long flags;
int ret;

ret = io_async_cancel_one(ctx, (void *) (unsigned long) sqe_addr);
ret = io_async_cancel_one(req->task->io_uring,
(void *) (unsigned long) sqe_addr);
if (ret != -ENOENT) {
spin_lock_irqsave(&ctx->completion_lock, flags);
goto done;
Expand Down Expand Up @@ -7537,16 +7539,6 @@ static void io_sq_thread_stop(struct io_ring_ctx *ctx)
}
}

static void io_finish_async(struct io_ring_ctx *ctx)
{
io_sq_thread_stop(ctx);

if (ctx->io_wq) {
io_wq_destroy(ctx->io_wq);
ctx->io_wq = NULL;
}
}

#if defined(CONFIG_UNIX)
/*
* Ensure the UNIX gc is aware of our file set, so we are certain that
Expand Down Expand Up @@ -8105,11 +8097,10 @@ static struct io_wq_work *io_free_work(struct io_wq_work *work)
return req ? &req->work : NULL;
}

static int io_init_wq_offload(struct io_ring_ctx *ctx)
static struct io_wq *io_init_wq_offload(struct io_ring_ctx *ctx)
{
struct io_wq_data data;
unsigned int concurrency;
int ret = 0;

data.user = ctx->user;
data.free_work = io_free_work;
Expand All @@ -8118,16 +8109,11 @@ static int io_init_wq_offload(struct io_ring_ctx *ctx)
/* Do QD, or 4 * CPUS, whatever is smallest */
concurrency = min(ctx->sq_entries, 4 * num_online_cpus());

ctx->io_wq = io_wq_create(concurrency, &data);
if (IS_ERR(ctx->io_wq)) {
ret = PTR_ERR(ctx->io_wq);
ctx->io_wq = NULL;
}

return ret;
return io_wq_create(concurrency, &data);
}

static int io_uring_alloc_task_context(struct task_struct *task)
static int io_uring_alloc_task_context(struct task_struct *task,
struct io_ring_ctx *ctx)
{
struct io_uring_task *tctx;
int ret;
Expand All @@ -8142,6 +8128,14 @@ static int io_uring_alloc_task_context(struct task_struct *task)
return ret;
}

tctx->io_wq = io_init_wq_offload(ctx);
if (IS_ERR(tctx->io_wq)) {
ret = PTR_ERR(tctx->io_wq);
percpu_counter_destroy(&tctx->inflight);
kfree(tctx);
return ret;
}

xa_init(&tctx->xa);
init_waitqueue_head(&tctx->wait);
tctx->last = NULL;
Expand Down Expand Up @@ -8214,7 +8208,7 @@ static int io_sq_offload_create(struct io_ring_ctx *ctx,
ctx->sq_thread_idle = HZ;

if (sqd->thread)
goto done;
return 0;

if (p->flags & IORING_SETUP_SQ_AFF) {
int cpu = p->sq_thread_cpu;
Expand All @@ -8236,7 +8230,7 @@ static int io_sq_offload_create(struct io_ring_ctx *ctx,
sqd->thread = NULL;
goto err;
}
ret = io_uring_alloc_task_context(sqd->thread);
ret = io_uring_alloc_task_context(sqd->thread, ctx);
if (ret)
goto err;
} else if (p->flags & IORING_SETUP_SQ_AFF) {
Expand All @@ -8245,14 +8239,9 @@ static int io_sq_offload_create(struct io_ring_ctx *ctx,
goto err;
}

done:
ret = io_init_wq_offload(ctx);
if (ret)
goto err;

return 0;
err:
io_finish_async(ctx);
io_sq_thread_stop(ctx);
return ret;
}

Expand Down Expand Up @@ -8727,7 +8716,7 @@ static void io_ring_ctx_free(struct io_ring_ctx *ctx)
mutex_lock(&ctx->uring_lock);
mutex_unlock(&ctx->uring_lock);

io_finish_async(ctx);
io_sq_thread_stop(ctx);
io_sqe_buffers_unregister(ctx);

if (ctx->sqo_task) {
Expand Down Expand Up @@ -8870,13 +8859,6 @@ static void io_ring_exit_work(struct work_struct *work)
io_ring_ctx_free(ctx);
}

static bool io_cancel_ctx_cb(struct io_wq_work *work, void *data)
{
struct io_kiocb *req = container_of(work, struct io_kiocb, work);

return req->ctx == data;
}

static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
{
mutex_lock(&ctx->uring_lock);
Expand All @@ -8895,9 +8877,6 @@ static void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
io_kill_timeouts(ctx, NULL, NULL);
io_poll_remove_all(ctx, NULL, NULL);

if (ctx->io_wq)
io_wq_cancel_cb(ctx->io_wq, io_cancel_ctx_cb, ctx, true);

/* if we failed setting up the ctx, we might not have any rings */
io_iopoll_try_reap_events(ctx);

Expand Down Expand Up @@ -8976,13 +8955,14 @@ static void io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
struct files_struct *files)
{
struct io_task_cancel cancel = { .task = task, .files = files, };
struct io_uring_task *tctx = current->io_uring;

while (1) {
enum io_wq_cancel cret;
bool ret = false;

if (ctx->io_wq) {
cret = io_wq_cancel_cb(ctx->io_wq, io_cancel_task_cb,
if (tctx && tctx->io_wq) {
cret = io_wq_cancel_cb(tctx->io_wq, io_cancel_task_cb,
&cancel, true);
ret |= (cret != IO_WQ_CANCEL_NOTFOUND);
}
Expand Down Expand Up @@ -9094,7 +9074,7 @@ static int io_uring_add_task_file(struct io_ring_ctx *ctx, struct file *file)
int ret;

if (unlikely(!tctx)) {
ret = io_uring_alloc_task_context(current);
ret = io_uring_alloc_task_context(current, ctx);
if (unlikely(ret))
return ret;
tctx = current->io_uring;
Expand Down Expand Up @@ -9164,8 +9144,12 @@ void __io_uring_files_cancel(struct files_struct *files)
io_uring_cancel_task_requests(file->private_data, files);
atomic_dec(&tctx->in_idle);

if (files)
if (files) {
io_uring_remove_task_files(tctx);
} else if (tctx->io_wq && current->flags & PF_EXITING) {
io_wq_destroy(tctx->io_wq);
tctx->io_wq = NULL;
}
}

static s64 tctx_inflight(struct io_uring_task *tctx)
Expand Down
1 change: 1 addition & 0 deletions include/linux/io_uring.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ struct io_uring_task {
struct xarray xa;
struct wait_queue_head wait;
struct file *last;
void *io_wq;
struct percpu_counter inflight;
struct io_identity __identity;
struct io_identity *identity;
Expand Down

0 comments on commit 5aa75ed

Please sign in to comment.