Destroy ThreadGroupSwitcher before set value in std::future

azat · azat · commit 362026fc37d5 · 2025-06-30T11:40:07.000+02:00
Otherwise it may lead to data-race: WARNING: ThreadSanitizer: data race (pid=26127) Write of size 6 at 0x7298000099e8 by main thread: 0 __tsan_memcpy <null> (clickhouse-82444-tsan+0x8be71be) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 1 DB::SettingsTraits::Data::operator=(DB::SettingsTraits::Data const&) ci/tmp/build/./src/Core/Settings.cpp:6978:1 (clickhouse-82444-tsan+0x192ae4f9) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 2 DB::BaseSettings<DB::SettingsTraits>::operator=(DB::BaseSettings<DB::SettingsTraits> const&) ci/tmp/build/./src/Core/BaseSettings.h:109:60 (clickhouse-82444-tsan+0x190f2bcc) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 3 DB::SettingsImpl::operator=(DB::SettingsImpl const&) ci/tmp/build/./src/Core/Settings.cpp:6984:8 (clickhouse-82444-tsan+0x190f2bcc) 4 DB::Settings::operator=(DB::Settings const&) ci/tmp/build/./src/Core/Settings.cpp:7191:11 (clickhouse-82444-tsan+0x190f2bcc) 5 DB::Context::setSettings(DB::Settings const&) ci/tmp/build/./src/Interpreters/Context.cpp:2635:15 (clickhouse-82444-tsan+0x1aa94c0a) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 6 DB::ClientBase::processParsedSingleQuery(std::__1::basic_string_view<char, std::__1::char_traits<char>>, std::__1::shared_ptr<DB::IAST>, bool&, unsigned long)::$_0::operator()() const ci/tmp/build/./src/Client/ClientBase.cpp:2184:9 (clickhouse-82444-tsan+0x1df0f4d7) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 7 BasicScopeGuard<DB::ClientBase::processParsedSingleQuery(std::__1::basic_string_view<char, std::__1::char_traits<char>>, std::__1::shared_ptr<DB::IAST>, bool&, unsigned long)::$_0>::invoke() ci/tmp/build/./base/base/../base/scope_guard.h:101:9 (clickhouse-82444-tsan+0x1df0f4d7) 8 BasicScopeGuard<DB::ClientBase::processParsedSingleQuery(std::__1::basic_string_view<char, std::__1::char_traits<char>>, std::__1::shared_ptr<DB::IAST>, bool&, unsigned long)::$_0>::~BasicScopeGuard() ci/tmp/build/./base/base/../base/scope_guard.h:50:26 (clickhouse-82444-tsan+0x1df0f4d7) 9 DB::ClientBase::processParsedSingleQuery(std::__1::basic_string_view<char, std::__1::char_traits<char>>, std::__1::shared_ptr<DB::IAST>, bool&, unsigned long) ci/tmp/build/./src/Client/ClientBase.cpp:2249:5 (clickhouse-82444-tsan+0x1df021ca) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) Previous read of size 1 at 0x7298000099e8 by thread T4: #0 DB::SettingFieldNumber<bool>::operator bool() const ci/tmp/build/./src/Core/SettingsFields.h:40:36 (clickhouse-82444-tsan+0x1b114e5f) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) #1 DB::ThreadStatus::finalizePerformanceCounters() ci/tmp/build/./src/Interpreters/ThreadStatusExt.cpp:547:17 (clickhouse-82444-tsan+0x1b114e5f) #2 DB::ThreadStatus::detachFromGroup() ci/tmp/build/./src/Interpreters/ThreadStatusExt.cpp:375:5 (clickhouse-82444-tsan+0x1b113aec) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) #3 DB::CurrentThread::detachFromGroupIfNotDetached() ci/tmp/build/./src/Interpreters/ThreadStatusExt.cpp:716:21 (clickhouse-82444-tsan+0x1b1119b3) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) #4 DB::ThreadGroupSwitcher::~ThreadGroupSwitcher() ci/tmp/build/./src/Interpreters/ThreadStatusExt.cpp:261:9 (clickhouse-82444-tsan+0x1b1119b3) #5 DB::ThreadPoolCallbackRunnerLocal<void, ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>, std::__1::function<void ()>>::operator()(std::__1::function<void ()>&&, Priority, std::__1::optional<unsigned long>)::'lambda'()::operator()() ci/tmp/build/./src/Common/threadPoolCallbackRunner.h:179:9 (clickhouse-82444-tsan+0x14beafde) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) Thread T4 'ThreadPool' (tid=26133, running) created by thread T3 at: ... 12 DB::ThreadPoolCallbackRunnerLocal<void, ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>, std::__1::function<void ()>>::operator()(std::__1::function<void ()>&&, Priority, std::__1::optional<unsigned long>) ci/tmp/build/./src/Common/threadPoolCallbackRunner.h:188:22 (clickhouse-82444-tsan+0x14be522d) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 13 DB::ParallelParsingInputFormat::scheduleParserThreadForUnitWithNumber(unsigned long) ci/tmp/build/./src/Processors/Formats/Impl/ParallelParsingInputFormat.h:287:9 (clickhouse-82444-tsan+0x1e495627) (BuildId: e5eeeb8fc2a0ab65c098bb1f3db89853f2439448) 14 DB::ParallelParsingInputFormat::segmentatorThreadFunction(std::__1::shared_ptr<DB::ThreadGroup>) ci/tmp/build/./src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp:45:13 (clickhouse-82444-tsan+0x1e495627) Refs: https://pastila.nl/?0001fdef/9a58e9d59c32d45100a481de26dccf68#T4iNrMFUnu4F2hWST5wLdQ==
diff --git a/src/Common/threadPoolCallbackRunner.h b/src/Common/threadPoolCallbackRunner.h
@@ -101,14 +101,18 @@ class ThreadPoolCallbackRunnerLocal final
 
     /// Set promise result for non-void callbacks
     template <typename Function, typename FunctionResult>
-    static void executeCallback(std::promise<FunctionResult> & promise, Function && callback)
+    static void executeCallback(std::promise<FunctionResult> & promise, Function && callback, ThreadGroupPtr thread_group, const std::string & thread_name)
     {
         /// Release callback before setting value to the promise to avoid
         /// destruction of captured resources after waitForAllToFinish returns.
         try
         {
-            FunctionResult res = callback();
-            callback = {};
+            FunctionResult res;
+            {
+                ThreadGroupSwitcher switcher(thread_group, thread_name.c_str());
+                res = callback();
+                callback = {};
+            }
             promise.set_value(std::move(res));
         }
         catch (...)
@@ -120,14 +124,17 @@ class ThreadPoolCallbackRunnerLocal final
 
     /// Set promise result for void callbacks
     template <typename Function>
-    static void executeCallback(std::promise<void> & promise, Function && callback)
+    static void executeCallback(std::promise<void> & promise, Function && callback, ThreadGroupPtr thread_group, const std::string & thread_name)
     {
         /// Release callback before setting value to the promise to avoid
         /// destruction of captured resources after waitForAllToFinish returns.
         try
         {
-            callback();
-            callback = {};
+            {
+                ThreadGroupSwitcher switcher(thread_group, thread_name.c_str());
+                callback();
+                callback = {};
+            }
             promise.set_value();
         }
         catch (...)
@@ -156,26 +163,24 @@ class ThreadPoolCallbackRunnerLocal final
         auto & task = tasks.emplace_back(std::make_shared<Task>());
         task->future = promise->get_future();
 
-        auto task_func = [task, thread_group = CurrentThread::getGroup(), my_thread_name = thread_name, my_callback = std::move(callback), promise]() mutable -> void
+        auto task_func = [this, task, thread_group = CurrentThread::getGroup(), my_callback = std::move(callback), promise]() mutable -> void
         {
-            ThreadGroupSwitcher switcher(thread_group, my_thread_name.c_str());
-
             TaskState expected = SCHEDULED;
             if (!task->state.compare_exchange_strong(expected, RUNNING))
             {
                 if (expected == CANCELLED)
                     return;
-                throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected state {} when running a task in {}", expected, my_thread_name);
+                throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected state {} when running a task in {}", expected, thread_name);
             }
 
             SCOPE_EXIT_SAFE(
             {
                 expected = RUNNING;
                 if (!task->state.compare_exchange_strong(expected, FINISHED))
-                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected state {} when finishing a task in {}", expected, my_thread_name);
+                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected state {} when finishing a task in {}", expected, thread_name);
             });
 
-            executeCallback(*promise, std::move(my_callback));
+            executeCallback(*promise, std::move(my_callback), std::move(thread_group), thread_name);
         };
 
         try
