On Wed, Jul 07, 2021 at 11:19:14AM +0800, Ming Lei wrote:

On Tue, Jul 06, 2021 at 09:55:36AM -0400, Dan Schatzberg wrote:

quoted

On Mon, Jul 05, 2021 at 06:26:07PM +0800, Ming Lei wrote:

quoted

 	}
+
+	spin_lock(lock);
 	list_add_tail(&cmd->list_entry, cmd_list);
+	spin_unlock(lock);
 	queue_work(lo->workqueue, work);
-	spin_unlock(&lo->lo_work_lock);
 }
 
 static void loop_update_rotational(struct loop_device *lo)

@@ -1131,20 +1159,18 @@ static void loop_set_timer(struct loop_device *lo)
 
 static void __loop_free_idle_workers(struct loop_device *lo, bool force)
 {
-	struct loop_worker *pos, *worker;
+	struct loop_worker *worker;
+	unsigned long id;
 
 	spin_lock(&lo->lo_work_lock);
-	list_for_each_entry_safe(worker, pos, &lo->idle_worker_list,
-				idle_list) {
+	xa_for_each(&lo->workers, id, worker) {
 		if (!force && time_is_after_jiffies(worker->last_ran_at +
 						LOOP_IDLE_WORKER_TIMEOUT))
 			break;
-		list_del(&worker->idle_list);
-		xa_erase(&lo->workers, worker->blkcg_css->id);
-		css_put(worker->blkcg_css);
-		kfree(worker);
+		if (refcount_dec_and_test(&worker->refcnt))
+			loop_release_worker(worker);

This one is puzzling to me. Can't you hit this refcount decrement
superfluously each time the loop timer fires?

Not sure I get your point.

As I mentioned above, this one is the counter pair of INIT reference,
but one new lo_cmd may just grab it when queueing rq before erasing the
worker from xarray, so we can't release worker here until the command is
completed.

Suppose at this point there's still an outstanding loop_cmd to be
serviced for this worker. The refcount_dec_and_test should decrement
the refcount and then fail the conditional, not calling
loop_release_worker. What happens if __loop_free_idle_workers fires
again before the loop_cmd is processed? Won't you decrement the
refcount again, and then end up calling loop_release_worker before the
loop_cmd is processed?