Re: Regression in io_uring, leading to data corruption

[Jens Axboe <axboe@kernel.dk>]

On 11/8/23 12:06 PM, Jens Axboe wrote:
>> I have been giving some thought to the CPU pinning of the workers, and
>> one thing that may have been overlooked is that this could potentially
>> force-serialize worker operations.  Did you just have to pin the io
>> workers or did the workqueue also need to be pinned for the corruption
>> to disappear?
> 
> Not sure I follow, the workers ARE the workqueue. For the pinning, I
> just made sure that the workers are on the same CPU. I honestly don't
> remember all the details there outside of what I can read back from the
> emails I sent, it's been a while. My suspicion back then was that it was
> some weird ppc cache aliasing effect with the copy into kernel memory
> happening on cpu X, and then we immediately punt it to cpu Y for
> processing.

You could try something like this, though I still need to verify that we
never end up running it on the wrong CPU. But may be worth a shot, for
debug purposes.

diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h
index d3009d56af0b..3fc9912f6306 100644
--- a/include/linux/io_uring_types.h
+++ b/include/linux/io_uring_types.h
@@ -21,6 +21,7 @@ struct io_wq_work {
 	unsigned flags;
 	/* place it here instead of io_kiocb as it fills padding and saves 4B */
 	int cancel_seq;
+	int cpu;
 };
 
 struct io_fixed_file {
diff --git a/io_uring/io-wq.c b/io_uring/io-wq.c
index 522196dfb0ff..bdaae56c6517 100644
--- a/io_uring/io-wq.c
+++ b/io_uring/io-wq.c
@@ -59,6 +59,7 @@ struct io_worker {
 	unsigned long create_state;
 	struct callback_head create_work;
 	int create_index;
+	int last_cpu;
 
 	union {
 		struct rcu_head rcu;
@@ -286,6 +287,10 @@ static bool io_wq_activate_free_worker(struct io_wq *wq,
 			io_worker_release(worker);
 			continue;
 		}
+		if (worker->last_cpu != raw_smp_processor_id()) {
+			io_worker_release(worker);
+			continue;
+		}
 		/*
 		 * If the worker is already running, it's either already
 		 * starting work or finishing work. In either case, if it does
@@ -581,6 +586,8 @@ static void io_worker_handle_work(struct io_wq_acct *acct,
 		} else {
 			break;
 		}
+		if (work->cpu != worker->last_cpu)
+			printk("work cpu %d, me %d\n", work->cpu, worker->last_cpu);
 		io_assign_current_work(worker, work);
 		__set_current_state(TASK_RUNNING);
 
@@ -639,6 +646,7 @@ static int io_wq_worker(void *data)
 	while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) {
 		long ret;
 
+		worker->last_cpu = raw_smp_processor_id();
 		set_current_state(TASK_INTERRUPTIBLE);
 
 		/*
@@ -664,7 +672,10 @@ static int io_wq_worker(void *data)
 		raw_spin_unlock(&wq->lock);
 		if (io_run_task_work())
 			continue;
+		worker->last_cpu = raw_smp_processor_id();
 		ret = schedule_timeout(WORKER_IDLE_TIMEOUT);
+		if (worker->last_cpu != raw_smp_processor_id())
+			printk("was on %d, now %d\n", worker->last_cpu, raw_smp_processor_id());
 		if (signal_pending(current)) {
 			struct ksignal ksig;
 
@@ -725,9 +736,17 @@ void io_wq_worker_sleeping(struct task_struct *tsk)
 static void io_init_new_worker(struct io_wq *wq, struct io_worker *worker,
 			       struct task_struct *tsk)
 {
+	cpumask_var_t new_mask;
+
 	tsk->worker_private = worker;
 	worker->task = tsk;
-	set_cpus_allowed_ptr(tsk, wq->cpu_mask);
+
+	if (alloc_cpumask_var(&new_mask, GFP_NOIO)) {
+		cpumask_clear(new_mask);
+		cpumask_set_cpu(worker->last_cpu, new_mask);
+		set_cpus_allowed_ptr(tsk, new_mask);
+		free_cpumask_var(new_mask);
+	}
 
 	raw_spin_lock(&wq->lock);
 	hlist_nulls_add_head_rcu(&worker->nulls_node, &wq->free_list);
@@ -835,6 +854,7 @@ static bool create_io_worker(struct io_wq *wq, int index)
 	refcount_set(&worker->ref, 1);
 	worker->wq = wq;
 	raw_spin_lock_init(&worker->lock);
+	worker->last_cpu = raw_smp_processor_id();
 	init_completion(&worker->ref_done);
 
 	if (index == IO_WQ_ACCT_BOUND)
@@ -928,6 +948,8 @@ void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work)
 	unsigned work_flags = work->flags;
 	bool do_create;
 
+	work->cpu = raw_smp_processor_id();
+
 	/*
 	 * If io-wq is exiting for this task, or if the request has explicitly
 	 * been marked as one that should not get executed, cancel it here.

-- 
Jens Axboe