fix data race on downscaling

serxa · serxa · commit 8335bbc65e3e · 2025-09-17T14:46:00.000Z
diff --git a/src/Processors/Executors/ExecutorTasks.cpp b/src/Processors/Executors/ExecutorTasks.cpp
@@ -290,19 +290,18 @@ void ExecutorTasks::preempt(size_t slot_id)
     std::unique_lock lock(mutex);
     --total_slots;
 
-    // We should make sure that preempted thread has no local task inside context.
-    // It is allowed to have tasks in `task_queue` or `fast_task_queue` because they can be stealed by other threads.
+    /// We should make sure that preempted thread has no local task inside context.
+    /// It is allowed to have tasks in `task_queue` or `fast_task_queue` because they can be stealed by other threads.
     auto & context = executor_contexts[slot_id];
     if (auto * task = context->popTask())
     {
         task_queue.push(task, slot_id);
         /// Wake up at least one thread to avoid deadlocks (all other threads maybe idle)
-        tryWakeUpAnyOtherThreadWithTasks(*context, lock);
+        tryWakeUpAnyOtherThreadWithTasks(*context, lock); // this releases the lock if it wakes up a thread
     }
-
-    // Finish pipeline if preempted thread was the last non-idle thread executed the last task of the whole pipeline
-    if (task_queue.empty() && fast_task_queue.empty() && async_task_queue.empty() && threads_queue.size() == total_slots)
+    else if (task_queue.empty() && fast_task_queue.empty() && async_task_queue.empty() && threads_queue.size() == total_slots)
     {
+        /// Finish pipeline if preempted thread was the last non-idle thread executed the last task of the whole pipeline
         lock.unlock();
         finish();
     }

Original file line number	Diff line number	Diff line change
`@@ -290,19 +290,18 @@ void ExecutorTasks::preempt(size_t slot_id)`
`290`	`290`	`std::unique_lock lock(mutex);`
`291`	`291`	`--total_slots;`
`292`	`292`
`293`		`- // We should make sure that preempted thread has no local task inside context.`
`294`		- // It is allowed to have tasks in `task_queue` or `fast_task_queue` because they can be stealed by other threads.
	`293`	`+ /// We should make sure that preempted thread has no local task inside context.`
	`294`	+ /// It is allowed to have tasks in `task_queue` or `fast_task_queue` because they can be stealed by other threads.
`295`	`295`	`auto & context = executor_contexts[slot_id];`
`296`	`296`	`if (auto * task = context->popTask())`
`297`	`297`	`{`
`298`	`298`	`task_queue.push(task, slot_id);`
`299`	`299`	`/// Wake up at least one thread to avoid deadlocks (all other threads maybe idle)`
`300`		`- tryWakeUpAnyOtherThreadWithTasks(*context, lock);`
	`300`	`+ tryWakeUpAnyOtherThreadWithTasks(*context, lock); // this releases the lock if it wakes up a thread`
`301`	`301`	`}`
`302`		`-`
`303`		`- // Finish pipeline if preempted thread was the last non-idle thread executed the last task of the whole pipeline`
`304`		`- if (task_queue.empty() && fast_task_queue.empty() && async_task_queue.empty() && threads_queue.size() == total_slots)`
	`302`	`+ else if (task_queue.empty() && fast_task_queue.empty() && async_task_queue.empty() && threads_queue.size() == total_slots)`
`305`	`303`	`{`
	`304`	`+ /// Finish pipeline if preempted thread was the last non-idle thread executed the last task of the whole pipeline`
`306`	`305`	`lock.unlock();`
`307`	`306`	`finish();`
`308`	`307`	`}`