Re: [PATCH 2/4] io_uring: move locking inside overflow posting

[Pavel Begunkov <asml.silence@gmail.com>]

On 5/14/25 22:52, Jens Axboe wrote:
> Since sometimes it's easier to talk in code that in English, something
> like the below (just applied on top, and utterly untested) I think is
> much cleaner. Didn't spend a lot of time on it, I'm sure it could get
> condensed down some more with a helper or something. But it keeps the
> locking in the caller, while still allowing GFP_KERNEL alloc for
> lockless_cq.
> 
> Somewhat unrelated, but also fills in an io_cqe and passes in for the
> non-cqe32 parts, just for readability's sake.
> 
> 
> diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c
> index da1075b66a87..9b6d09633a29 100644
> --- a/io_uring/io_uring.c
> +++ b/io_uring/io_uring.c
> @@ -744,44 +744,12 @@ static bool __io_cqring_event_overflow(struct io_ring_ctx *ctx,
>   	return true;
>   }
>   
> -static bool io_cqring_event_overflow(struct io_ring_ctx *ctx, bool locked,
> -				     u64 user_data, s32 res, u32 cflags,
> -				     u64 extra1, u64 extra2)
> +static void io_cqring_event_overflow(struct io_ring_ctx *ctx,
> +				     struct io_overflow_cqe *ocqe)
>   {
> -	struct io_overflow_cqe *ocqe;
> -	size_t ocq_size = sizeof(struct io_overflow_cqe);
> -	bool is_cqe32 = (ctx->flags & IORING_SETUP_CQE32);
> -	gfp_t gfp = GFP_KERNEL_ACCOUNT;
> -	bool queued;
> -
> -	io_lockdep_assert_cq_locked(ctx);
> -
> -	if (is_cqe32)
> -		ocq_size += sizeof(struct io_uring_cqe);
> -	if (locked)
> -		gfp = GFP_ATOMIC | __GFP_ACCOUNT;
> -
> -	ocqe = kmalloc(ocq_size, gfp);
> -	trace_io_uring_cqe_overflow(ctx, user_data, res, cflags, ocqe);
> -
> -	if (ocqe) {
> -		ocqe->cqe.user_data = user_data;
> -		ocqe->cqe.res = res;
> -		ocqe->cqe.flags = cflags;
> -		if (is_cqe32) {
> -			ocqe->cqe.big_cqe[0] = extra1;
> -			ocqe->cqe.big_cqe[1] = extra2;
> -		}
> -	}
> -
> -	if (locked) {
> -		queued = __io_cqring_event_overflow(ctx, ocqe);
> -	} else {
> -		spin_lock(&ctx->completion_lock);
> -		queued = __io_cqring_event_overflow(ctx, ocqe);
> -		spin_unlock(&ctx->completion_lock);
> -	}
> -	return queued;
> +	spin_lock(&ctx->completion_lock);
> +	__io_cqring_event_overflow(ctx, ocqe);
> +	spin_unlock(&ctx->completion_lock);
>   }
>   
>   /*
> @@ -842,15 +810,49 @@ static bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res,
>   	return false;
>   }
>   
> +static struct io_overflow_cqe *io_get_ocqe(struct io_ring_ctx *ctx,
> +					   struct io_cqe *cqe, u64 extra1,
> +					   u64 extra2, gfp_t gfp)
> +{
> +	size_t ocq_size = sizeof(struct io_overflow_cqe);
> +	bool is_cqe32 = ctx->flags & IORING_SETUP_CQE32;
> +	struct io_overflow_cqe *ocqe;
> +
> +	if (is_cqe32)
> +		ocq_size += sizeof(struct io_uring_cqe);
> +
> +	ocqe = kmalloc(ocq_size, gfp);
> +	trace_io_uring_cqe_overflow(ctx, cqe->user_data, cqe->res, cqe->flags, ocqe);
> +	if (ocqe) {
> +		ocqe->cqe.user_data = cqe->user_data;
> +		ocqe->cqe.res = cqe->res;
> +		ocqe->cqe.flags = cqe->flags;
> +		if (is_cqe32) {
> +			ocqe->cqe.big_cqe[0] = extra1;
> +			ocqe->cqe.big_cqe[1] = extra2;
> +		}
> +	}
> +
> +	return ocqe;
> +}
> +
>   bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
>   {
>   	bool filled;
>   
>   	io_cq_lock(ctx);
>   	filled = io_fill_cqe_aux(ctx, user_data, res, cflags);
> -	if (!filled)
> -		filled = io_cqring_event_overflow(ctx, true,
> -						  user_data, res, cflags, 0, 0);
> +	if (unlikely(!filled)) {
> +		struct io_cqe cqe = {
> +			.user_data	= user_data,
> +			.res		= res,
> +			.flags		= cflags
> +		};
> +		struct io_overflow_cqe *ocqe;
> +
> +		ocqe = io_get_ocqe(ctx, &cqe, 0, 0, GFP_ATOMIC);
> +		filled = __io_cqring_event_overflow(ctx, ocqe);
> +	}
>   	io_cq_unlock_post(ctx);
>   	return filled;
>   }
> @@ -864,8 +866,17 @@ void io_add_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
>   	lockdep_assert_held(&ctx->uring_lock);
>   	lockdep_assert(ctx->lockless_cq);
>   
> -	if (!io_fill_cqe_aux(ctx, user_data, res, cflags))
> -		io_cqring_event_overflow(ctx, false, user_data, res, cflags, 0, 0);
> +	if (!io_fill_cqe_aux(ctx, user_data, res, cflags)) {
> +		struct io_cqe cqe = {
> +			.user_data	= user_data,
> +			.res		= res,
> +			.flags		= cflags
> +		};
> +		struct io_overflow_cqe *ocqe;
> +
> +		ocqe = io_get_ocqe(ctx, &cqe, 0, 0, GFP_KERNEL);
> +		io_cqring_event_overflow(ctx, ocqe);
> +	}
>   
>   	ctx->submit_state.cq_flush = true;
>   }
> @@ -1437,6 +1448,20 @@ static void io_free_batch_list(struct io_ring_ctx *ctx,
>   	} while (node);
>   }
>   
> +static __cold void io_cqe_overflow_fill(struct io_ring_ctx *ctx,
> +					struct io_kiocb *req)
> +{
> +	gfp_t gfp = ctx->lockless_cq ? GFP_KERNEL : GFP_ATOMIC;
> +	struct io_overflow_cqe *ocqe;
> +
> +	ocqe = io_get_ocqe(ctx, &req->cqe, req->big_cqe.extra1, req->big_cqe.extra2, gfp);
> +	if (ctx->lockless_cq)
> +		io_cqring_event_overflow(ctx, ocqe);
> +	else
> +		__io_cqring_event_overflow(ctx, ocqe);
> +	memset(&req->big_cqe, 0, sizeof(req->big_cqe));
> +}

Implicitly passing the locking state b/w functions is much worse. And
instead being contained in a single helper all callers need to care
about some allocator function, which doesn't help in making it cleaner
and also contributes to bloating. Let's just leave How it's already
in the tree.

-- 
Pavel Begunkov