[PATCH 8/9] io_uring/msg_ring: add basic wakeup batch support

[Jens Axboe <axboe@kernel.dk>]

Factor in the number of overflow entries waiting, on both the msg ring
and local task_work add side.

Signed-off-by: Jens Axboe <axboe@kernel.dk>
---
 io_uring/msg_ring.c | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/io_uring/msg_ring.c b/io_uring/msg_ring.c
index 9a7c63f38c46..eeca1563ceed 100644
--- a/io_uring/msg_ring.c
+++ b/io_uring/msg_ring.c
@@ -109,6 +109,8 @@ static void io_msg_add_overflow(struct io_msg *msg,
 				u32 flags)
 	__releases(&target_ctx->completion_lock)
 {
+	unsigned nr_prev, nr_wait;
+
 	if (list_empty(&target_ctx->cq_overflow_list)) {
 		set_bit(IO_CHECK_CQ_OVERFLOW_BIT, &target_ctx->check_cq);
 		atomic_or(IORING_SQ_TASKRUN, &target_ctx->rings->sq_flags);
@@ -117,10 +119,14 @@ static void io_msg_add_overflow(struct io_msg *msg,
 	ocqe->cqe.user_data = msg->user_data;
 	ocqe->cqe.res = ret;
 	ocqe->cqe.flags = flags;
-	target_ctx->nr_overflow++;
+	nr_prev = target_ctx->nr_overflow++;
 	list_add_tail(&ocqe->list, &target_ctx->cq_overflow_list);
 	spin_unlock(&target_ctx->completion_lock);
-	wake_up_state(target_ctx->submitter_task, TASK_INTERRUPTIBLE);
+	rcu_read_lock();
+	io_defer_tw_count(target_ctx, &nr_wait);
+	nr_prev += nr_wait;
+	io_defer_wake(target_ctx, nr_prev + 1, nr_prev);
+	rcu_read_unlock();
 }
 
 static int io_msg_fill_remote(struct io_msg *msg, unsigned int issue_flags,
-- 
2.43.0