Commit 3a4b89a2 authored by Jens Axboe's avatar Jens Axboe
Browse files

io_uring: serialize ctx->rings->sq_flags with atomic_or/and 



Rather than require ctx->completion_lock for ensuring that we don't
clobber the flags, use the atomic bitop helpers instead. This removes
the need to grab the completion_lock, in preparation for needing to set
or clear sq_flags when we don't know the status of this lock.

Reviewed-by: default avatarPavel Begunkov <asml.silence@gmail.com>
Link: https://lore.kernel.org/r/20220426014904.60384-3-axboe@kernel.dk


Signed-off-by: default avatarJens Axboe <axboe@kernel.dk>
parent e788be95

fs/io_uring.c

+10 −26
Original line number Diff line number Diff line
@@ -169,7 +169,7 @@ struct io_rings { 
	 * The application needs a full memory barrier before checking

	 * for IORING_SQ_NEED_WAKEUP after updating the sq tail.

	 */

	u32			sq_flags;

	atomic_t		sq_flags;

	/*

	 * Runtime CQ flags

	 *
@@ -2030,8 +2030,7 @@ static bool __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force) 
	all_flushed = list_empty(&ctx->cq_overflow_list);

	if (all_flushed) {

		clear_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);

		WRITE_ONCE(ctx->rings->sq_flags,

			   ctx->rings->sq_flags & ~IORING_SQ_CQ_OVERFLOW);

		atomic_andnot(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);

	}



	io_commit_cqring(ctx);
@@ -2125,8 +2124,7 @@ static bool io_cqring_event_overflow(struct io_ring_ctx *ctx, u64 user_data, 
	}

	if (list_empty(&ctx->cq_overflow_list)) {

		set_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);

		WRITE_ONCE(ctx->rings->sq_flags,

			   ctx->rings->sq_flags | IORING_SQ_CQ_OVERFLOW);

		atomic_or(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);



	}

	ocqe->cqe.user_data = user_data;
@@ -8108,23 +8106,6 @@ static inline bool io_sqd_events_pending(struct io_sq_data *sqd) 
	return READ_ONCE(sqd->state);

}



static inline void io_ring_set_wakeup_flag(struct io_ring_ctx *ctx)

{

	/* Tell userspace we may need a wakeup call */

	spin_lock(&ctx->completion_lock);

	WRITE_ONCE(ctx->rings->sq_flags,

		   ctx->rings->sq_flags | IORING_SQ_NEED_WAKEUP);

	spin_unlock(&ctx->completion_lock);

}



static inline void io_ring_clear_wakeup_flag(struct io_ring_ctx *ctx)

{

	spin_lock(&ctx->completion_lock);

	WRITE_ONCE(ctx->rings->sq_flags,

		   ctx->rings->sq_flags & ~IORING_SQ_NEED_WAKEUP);

	spin_unlock(&ctx->completion_lock);

}



static int __io_sq_thread(struct io_ring_ctx *ctx, bool cap_entries)

{

	unsigned int to_submit;
@@ -8240,8 +8221,8 @@ static int io_sq_thread(void *data) 
			bool needs_sched = true;



			list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) {

				io_ring_set_wakeup_flag(ctx);



				atomic_or(IORING_SQ_NEED_WAKEUP,

						&ctx->rings->sq_flags);

				if ((ctx->flags & IORING_SETUP_IOPOLL) &&

				    !wq_list_empty(&ctx->iopoll_list)) {

					needs_sched = false;
@@ -8266,7 +8247,8 @@ static int io_sq_thread(void *data) 
				mutex_lock(&sqd->lock);

			}

			list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)

				io_ring_clear_wakeup_flag(ctx);

				atomic_andnot(IORING_SQ_NEED_WAKEUP,

						&ctx->rings->sq_flags);

		}



		finish_wait(&sqd->wait, &wait);
@@ -8276,7 +8258,7 @@ static int io_sq_thread(void *data) 
	io_uring_cancel_generic(true, sqd);

	sqd->thread = NULL;

	list_for_each_entry(ctx, &sqd->ctx_list, sqd_list)

		io_ring_set_wakeup_flag(ctx);

		atomic_or(IORING_SQ_NEED_WAKEUP, &ctx->rings->sq_flags);

	io_run_task_work();

	mutex_unlock(&sqd->lock);
@@ -12029,6 +12011,8 @@ static int __init io_uring_init(void) 
	BUILD_BUG_ON(ARRAY_SIZE(io_op_defs) != IORING_OP_LAST);

	BUILD_BUG_ON(__REQ_F_LAST_BIT > 8 * sizeof(int));



	BUILD_BUG_ON(sizeof(atomic_t) != sizeof(u32));



	req_cachep = KMEM_CACHE(io_kiocb, SLAB_HWCACHE_ALIGN | SLAB_PANIC |

				SLAB_ACCOUNT);

	return 0;