Re: Regression in io_uring, leading to data corruption

[Jens Axboe <axboe@kernel.dk>]

On 11/8/23 4:58 PM, Jens Axboe wrote:
> On 11/8/23 3:18 PM, Jens Axboe wrote:
>> OK, let me actually test this thing and see if I can make it solid
>> first...
> 
> Here's a suitable hack - it just creates a new io worker for each item,
> ensuring that that worker is run on the same CPU.

Turns out I didn't send you the current one, this one was older and
untested. Please try this one instead.

diff --git a/io_uring/io-wq.c b/io_uring/io-wq.c
index 522196dfb0ff..41b4e281db8c 100644
--- a/io_uring/io-wq.c
+++ b/io_uring/io-wq.c
@@ -178,6 +178,8 @@ bool io_wq_worker_stopped(void)
 {
 	struct io_worker *worker = current->worker_private;
 
+	if (current->worker_private == current)
+		return false;
 	if (WARN_ON_ONCE(!io_wq_current_is_worker()))
 		return true;
 
@@ -693,7 +695,7 @@ void io_wq_worker_running(struct task_struct *tsk)
 {
 	struct io_worker *worker = tsk->worker_private;
 
-	if (!worker)
+	if (!worker || tsk->worker_private == tsk)
 		return;
 	if (!(worker->flags & IO_WORKER_F_UP))
 		return;
@@ -711,7 +713,7 @@ void io_wq_worker_sleeping(struct task_struct *tsk)
 {
 	struct io_worker *worker = tsk->worker_private;
 
-	if (!worker)
+	if (!worker || tsk->worker_private == tsk)
 		return;
 	if (!(worker->flags & IO_WORKER_F_UP))
 		return;
@@ -921,6 +923,67 @@ static bool io_wq_work_match_item(struct io_wq_work *work, void *data)
 	return work == data;
 }
 
+struct work_data {
+	struct io_wq_work *work;
+	struct io_wq *wq;
+	int cpu;
+};
+
+static int io_wq_single_worker(void *data)
+{
+	struct work_data *wd = data;
+	struct io_wq_work *work = wd->work;
+	struct io_wq *wq = wd->wq;
+
+	WARN_ON_ONCE(wd->cpu != raw_smp_processor_id());
+	kfree(wd);
+	do {
+		if (test_bit(IO_WQ_BIT_EXIT, &wq->state))
+			work->flags |= IO_WQ_WORK_CANCEL;
+		wq->do_work(work);
+		work = wq->free_work(work);
+	} while (work);
+
+	do_exit(0);
+	return 0;
+}
+
+static int io_wq_create_single(struct io_wq *wq, struct io_wq_work *work)
+{
+	struct task_struct *tsk;
+	struct work_data *wd;
+	cpumask_var_t new_mask;
+
+	wd = kmalloc(sizeof(*wd), GFP_NOIO);
+	if (!wd)
+		return false;
+
+	if (!alloc_cpumask_var(&new_mask, GFP_NOIO)) {
+		kfree(wd);
+		return false;
+	}
+
+	wd->work = work;
+	wd->cpu = raw_smp_processor_id();
+	wd->wq = wq,
+
+	cpumask_clear(new_mask);
+	cpumask_set_cpu(wd->cpu, new_mask);
+
+	tsk = create_io_thread(io_wq_single_worker, wd, NUMA_NO_NODE);
+	if (!IS_ERR(tsk)) {
+		tsk->worker_private = tsk;
+		set_cpus_allowed_ptr(tsk, new_mask);
+		free_cpumask_var(new_mask);
+		wake_up_new_task(tsk);
+		return true;
+	}
+
+	free_cpumask_var(new_mask);
+	kfree(wd);
+	return false;
+}
+
 void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work)
 {
 	struct io_wq_acct *acct = io_work_get_acct(wq, work);
@@ -938,6 +1001,10 @@ void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work)
 		return;
 	}
 
+	if (io_wq_create_single(wq, work))
+		return;
+
+	WARN_ON_ONCE(1);
 	raw_spin_lock(&acct->lock);
 	io_wq_insert_work(wq, work);
 	clear_bit(IO_ACCT_STALLED_BIT, &acct->flags);

-- 
Jens Axboe