diff --git a/src/AggregateFunctions/UniqExactSet.h b/src/AggregateFunctions/UniqExactSet.h index 2ae8c3a8386b..3c76cb9493b6 100644 --- a/src/AggregateFunctions/UniqExactSet.h +++ b/src/AggregateFunctions/UniqExactSet.h @@ -65,14 +65,7 @@ class UniqExactSet auto data_vec_atomic_index = std::make_shared(0); auto thread_func = [data_vec, data_vec_atomic_index, &is_cancelled, thread_group = CurrentThread::getGroup()]() { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - - setThreadName("UniqExaConvert"); + ThreadGroupSwitcher switcher(thread_group, "UniqExaConvert"); while (true) { @@ -128,13 +121,7 @@ class UniqExactSet auto thread_func = [&lhs, &rhs, next_bucket_to_merge, is_cancelled, thread_group = CurrentThread::getGroup()]() { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - setThreadName("UniqExactMerger"); + ThreadGroupSwitcher switcher(thread_group, "UniqExactMerger"); while (true) { diff --git a/src/Common/CurrentMetrics.cpp b/src/Common/CurrentMetrics.cpp index adc074738786..74ac9fab28e7 100644 --- a/src/Common/CurrentMetrics.cpp +++ b/src/Common/CurrentMetrics.cpp @@ -228,6 +228,9 @@ M(OutdatedPartsLoadingThreads, "Number of threads in the threadpool for loading Outdated data parts.") \ M(OutdatedPartsLoadingThreadsActive, "Number of active threads in the threadpool for loading Outdated data parts.") \ M(OutdatedPartsLoadingThreadsScheduled, "Number of queued or active jobs in the threadpool for loading Outdated data parts.") \ + M(PolygonDictionaryThreads, "Number of threads in the threadpool for polygon dictionaries.") \ + M(PolygonDictionaryThreadsActive, "Number of active threads in the threadpool for polygon dictionaries.") \ + M(PolygonDictionaryThreadsScheduled, "Number of queued or active jobs in the threadpool for polygon dictionaries.") \ M(DistributedBytesToInsert, "Number of pending bytes to process for asynchronous insertion into Distributed tables. Number of bytes for every shard is summed.") \ M(BrokenDistributedBytesToInsert, "Number of bytes for asynchronous insertion into Distributed tables that has been marked as broken. Number of bytes for every shard is summed.") \ M(DistributedFilesToInsert, "Number of pending files to process for asynchronous insertion into Distributed tables. Number of files for every shard is summed.") \ diff --git a/src/Common/ThreadPool.cpp b/src/Common/ThreadPool.cpp index 5dfefcbfccf1..d54d0be6f1bd 100644 --- a/src/Common/ThreadPool.cpp +++ b/src/Common/ThreadPool.cpp @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -769,6 +770,11 @@ void ThreadPoolImpl::ThreadFromThreadPool::worker() CurrentMetrics::Increment metric_active_pool_threads(parent_pool.metric_active_threads); +#ifdef DEBUG_OR_SANITIZER_BUILD + DB::ThreadStatus * initial_thread = DB::current_thread; + DB::ThreadGroupPtr initial_thread_group = DB::CurrentThread::getGroup(); +#endif + if constexpr (!std::is_same_v) { Stopwatch watch; @@ -785,6 +791,13 @@ void ThreadPoolImpl::ThreadFromThreadPool::worker() job_data->job(); } +#ifdef DEBUG_OR_SANITIZER_BUILD + DB::ThreadStatus * final_thread = DB::current_thread; + DB::ThreadGroupPtr final_thread_group = DB::CurrentThread::getGroup(); + if (final_thread != initial_thread || final_thread_group != initial_thread_group) + throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Thread pool job changed current ThreadStatus pointer ({} -> {}) or ThreadGroup ({} -> {}).", initial_thread ? "non-nullptr" : "nullptr", final_thread ? "non-nullptr" : "nullptr", initial_thread_group ? "master_thread_id " + std::to_string(initial_thread_group->master_thread_id) : "nullptr", final_thread_group ? "master_thread_id " + std::to_string(final_thread_group->master_thread_id) : "nullptr"); +#endif + if (thread_trace_context.root_span.isTraceEnabled()) { diff --git a/src/Common/ThreadStatus.h b/src/Common/ThreadStatus.h index 0c02ab8fdb0c..337d2499ac20 100644 --- a/src/Common/ThreadStatus.h +++ b/src/Common/ThreadStatus.h @@ -133,17 +133,31 @@ class ThreadGroup }; /** - * Since merge is executed with multiple threads, this class - * switches the parent MemoryTracker as part of the thread group to account all the memory used. + * RAII wrapper around CurrentThread::attachToGroup/detachFromGroupIfNotDetached. + * + * Typically used for inheriting thread group when scheduling tasks on a thread pool: + * pool->scheduleOrThrow([thread_group = CurrentThread::getGroup()]() + * { + * ThreadGroupSwitcher switcher(thread_group, "MyThread"); + * ... + * }); */ class ThreadGroupSwitcher : private boost::noncopyable { public: - explicit ThreadGroupSwitcher(ThreadGroupPtr thread_group); + /// If thread_group_ is nullptr or equal to current thread group, does nothing. + /// allow_existing_group: + /// * If false, asserts that the thread is not already attached to a different group. + /// Use this when running a task in a thread pool. + /// * If true, remembers the current group and restores it in destructor. + /// If thread_name is not empty, calls setThreadName along the way; should be at most 15 bytes long. + explicit ThreadGroupSwitcher(ThreadGroupPtr thread_group_, const char * thread_name, bool allow_existing_group = false) noexcept; ~ThreadGroupSwitcher(); private: + ThreadStatus * prev_thread = nullptr; ThreadGroupPtr prev_thread_group; + ThreadGroupPtr thread_group; }; diff --git a/src/Common/threadPoolCallbackRunner.h b/src/Common/threadPoolCallbackRunner.h index afbdcf2df19f..1f7afb73b337 100644 --- a/src/Common/threadPoolCallbackRunner.h +++ b/src/Common/threadPoolCallbackRunner.h @@ -31,24 +31,15 @@ ThreadPoolCallbackRunnerUnsafe threadPoolCallbackRunnerUnsafe( { auto task = std::make_shared>([thread_group, thread_name, my_callback = std::move(callback)]() mutable -> Result { - if (thread_group) - CurrentThread::attachToGroup(thread_group); + ThreadGroupSwitcher switcher(thread_group, thread_name.c_str()); SCOPE_EXIT_SAFE( { - { - /// Release all captured resources before detaching thread group - /// Releasing has to use proper memory tracker which has been set here before callback - - [[maybe_unused]] auto tmp = std::move(my_callback); - } - - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); + /// Release all captured resources before detaching thread group + /// Releasing has to use proper memory tracker which has been set here before callback + [[maybe_unused]] auto tmp = std::move(my_callback); }); - setThreadName(thread_name.data()); - return my_callback(); }); @@ -75,6 +66,8 @@ std::future scheduleFromThreadPoolUnsafe(T && task, ThreadPool & pool, c template > class ThreadPoolCallbackRunnerLocal { + static_assert(!std::is_same_v, "Scheduling tasks directly on GlobalThreadPool is not allowed because it doesn't set up CurrentThread. Create a new ThreadPool (local or in SharedThreadPools.h) or use ThreadFromGlobalPool."); + PoolT & pool; std::string thread_name; @@ -124,6 +117,8 @@ class ThreadPoolCallbackRunnerLocal auto task_func = std::make_shared>( [task, thread_group = CurrentThread::getGroup(), my_thread_name = thread_name, my_callback = std::move(callback)]() mutable -> Result { + ThreadGroupSwitcher switcher(thread_group, my_thread_name.c_str()); + TaskState expected = SCHEDULED; if (!task->state.compare_exchange_strong(expected, RUNNING)) { @@ -139,9 +134,6 @@ class ThreadPoolCallbackRunnerLocal throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected state {} when finishing a task in {}", expected, my_thread_name); }); - if (thread_group) - CurrentThread::attachToGroup(thread_group); - SCOPE_EXIT_SAFE( { { @@ -150,13 +142,8 @@ class ThreadPoolCallbackRunnerLocal [[maybe_unused]] auto tmp = std::move(my_callback); } - - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); }); - setThreadName(my_thread_name.data()); - return my_callback(); }); diff --git a/src/Dictionaries/HashedDictionary.h b/src/Dictionaries/HashedDictionary.h index 7e935fe4855b..4b8582663692 100644 --- a/src/Dictionaries/HashedDictionary.h +++ b/src/Dictionaries/HashedDictionary.h @@ -336,18 +336,11 @@ HashedDictionary::~HashedDictionary() pool.trySchedule([&container, thread_group = CurrentThread::getGroup()] { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); + ThreadGroupSwitcher switcher(thread_group, "HashedDictDtor"); /// Do not account memory that was occupied by the dictionaries for the query/user context. MemoryTrackerBlockerInThread memory_blocker; - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - setThreadName("HashedDictDtor"); - clearContainer(container); }); diff --git a/src/Dictionaries/HashedDictionaryParallelLoader.h b/src/Dictionaries/HashedDictionaryParallelLoader.h index ef5e6976c17a..55eaa550448f 100644 --- a/src/Dictionaries/HashedDictionaryParallelLoader.h +++ b/src/Dictionaries/HashedDictionaryParallelLoader.h @@ -68,22 +68,17 @@ class HashedDictionaryParallelLoader : public boost::noncopyable { pool.scheduleOrThrowOnError([this, shard, thread_group = CurrentThread::getGroup()] { + ThreadGroupSwitcher switcher(thread_group, "HashedDictLoad"); + WorkerStatistic statistic; SCOPE_EXIT_SAFE( LOG_TRACE(dictionary.log, "Finished worker for dictionary {} shard {}, processed {} blocks, {} rows, total time {}ms", dictionary_name, shard, statistic.total_blocks, statistic.total_rows, statistic.total_elapsed_ms); - - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); ); /// Do not account memory that was occupied by the dictionaries for the query/user context. MemoryTrackerBlockerInThread memory_blocker; - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - setThreadName("HashedDictLoad"); - LOG_TRACE(dictionary.log, "Starting worker for dictionary {}, shard {}", dictionary_name, shard); threadWorker(shard, statistic); diff --git a/src/Dictionaries/PolygonDictionaryUtils.h b/src/Dictionaries/PolygonDictionaryUtils.h index 9fba467a3630..3f6f382bbc76 100644 --- a/src/Dictionaries/PolygonDictionaryUtils.h +++ b/src/Dictionaries/PolygonDictionaryUtils.h @@ -15,6 +15,12 @@ #include +namespace CurrentMetrics +{ + extern const Metric PolygonDictionaryThreads; + extern const Metric PolygonDictionaryThreadsActive; + extern const Metric PolygonDictionaryThreadsScheduled; +} namespace DB { @@ -252,7 +258,8 @@ class GridRoot : public ICell std::vector>> children; children.resize(DividedCell::kSplit * DividedCell::kSplit); - ThreadPoolCallbackRunnerLocal runner(GlobalThreadPool::instance(), "PolygonDict"); + ThreadPool pool(CurrentMetrics::PolygonDictionaryThreads, CurrentMetrics::PolygonDictionaryThreadsActive, CurrentMetrics::PolygonDictionaryThreadsScheduled, 128); + ThreadPoolCallbackRunnerLocal runner(pool, "PolygonDict"); for (size_t i = 0; i < DividedCell::kSplit; current_min_x += x_shift, ++i) { auto handle_row = [this, &children, &y_shift, &x_shift, &possible_ids, &depth, i, x = current_min_x, y = current_min_y]() mutable diff --git a/src/Interpreters/Aggregator.cpp b/src/Interpreters/Aggregator.cpp index d1aa8a0fff0e..3b335d2f63ec 100644 --- a/src/Interpreters/Aggregator.cpp +++ b/src/Interpreters/Aggregator.cpp @@ -2209,15 +2209,8 @@ BlocksList Aggregator::prepareBlocksAndFillTwoLevelImpl( std::atomic next_bucket_to_merge = 0; - auto converter = [&](size_t thread_id, ThreadGroupPtr thread_group) + auto converter = [&](size_t thread_id) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - BlocksList blocks; while (true) { @@ -2245,10 +2238,14 @@ BlocksList Aggregator::prepareBlocksAndFillTwoLevelImpl( for (size_t thread_id = 0; thread_id < max_threads; ++thread_id) { tasks[thread_id] = std::packaged_task( - [group = CurrentThread::getGroup(), thread_id, &converter] { return converter(thread_id, group); }); + [thread_id, &converter] { return converter(thread_id); }); if (thread_pool) - thread_pool->scheduleOrThrowOnError([thread_id, &tasks] { tasks[thread_id](); }); + thread_pool->scheduleOrThrowOnError([thread_id, &tasks, thread_group = CurrentThread::getGroup()] + { + ThreadGroupSwitcher switcher(thread_group, ""); + tasks[thread_id](); + }); else tasks[thread_id](); } @@ -2970,15 +2967,8 @@ void Aggregator::mergeBlocks(BucketToBlocks bucket_to_blocks, AggregatedDataVari LOG_TRACE(log, "Merging partially aggregated two-level data."); - auto merge_bucket = [&bucket_to_blocks, &result, this](Int32 bucket, Arena * aggregates_pool, ThreadGroupPtr thread_group) + auto merge_bucket = [&bucket_to_blocks, &result, this](Int32 bucket, Arena * aggregates_pool) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - for (Block & block : bucket_to_blocks[bucket]) { /// Copy to avoid race. @@ -3013,15 +3003,14 @@ void Aggregator::mergeBlocks(BucketToBlocks bucket_to_blocks, AggregatedDataVari result.aggregates_pools.push_back(std::make_shared()); Arena * aggregates_pool = result.aggregates_pools.back().get(); - /// if we don't use thread pool we don't need to attach and definitely don't want to detach from the thread group - /// because this thread is already attached - auto task = [group = thread_pool != nullptr ? CurrentThread::getGroup() : nullptr, bucket, &merge_bucket, aggregates_pool] - { merge_bucket(bucket, aggregates_pool, group); }; - if (thread_pool) - thread_pool->scheduleOrThrowOnError(task); + thread_pool->scheduleOrThrowOnError([bucket, &merge_bucket, aggregates_pool, thread_group = CurrentThread::getGroup()] + { + ThreadGroupSwitcher switcher(thread_group, ""); + merge_bucket(bucket, aggregates_pool); + }); else - task(); + merge_bucket(bucket, aggregates_pool); } if (thread_pool) diff --git a/src/Interpreters/ConcurrentHashJoin.cpp b/src/Interpreters/ConcurrentHashJoin.cpp index ac940c62a1ab..df46c9e3f7a8 100644 --- a/src/Interpreters/ConcurrentHashJoin.cpp +++ b/src/Interpreters/ConcurrentHashJoin.cpp @@ -97,14 +97,7 @@ ConcurrentHashJoin::ConcurrentHashJoin( pool->scheduleOrThrow( [&, idx = i, thread_group = CurrentThread::getGroup()]() { - SCOPE_EXIT_SAFE({ - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - }); - - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - setThreadName("ConcurrentJoin"); + ThreadGroupSwitcher switcher(thread_group, "ConcurrentJoin"); size_t reserve_size = 0; if (auto hint = getSizeHint(stats_collecting_params, slots)) @@ -144,14 +137,7 @@ ConcurrentHashJoin::~ConcurrentHashJoin() pool->scheduleOrThrow( [join = std::move(hash_joins[i]), thread_group = CurrentThread::getGroup()]() { - SCOPE_EXIT_SAFE({ - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - }); - - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - setThreadName("ConcurrentJoin"); + ThreadGroupSwitcher switcher(thread_group, "ConcurrentJoin"); }); } pool->wait(); diff --git a/src/Interpreters/ExternalLoader.cpp b/src/Interpreters/ExternalLoader.cpp index 511300be2e05..d000b4aec16c 100644 --- a/src/Interpreters/ExternalLoader.cpp +++ b/src/Interpreters/ExternalLoader.cpp @@ -970,13 +970,7 @@ class ExternalLoader::LoadingDispatcher : private boost::noncopyable /// Does the loading, possibly in the separate thread. void doLoading(const String & name, size_t loading_id, bool forced_to_reload, size_t min_id_to_finish_loading_dependencies_, bool async, ThreadGroupPtr thread_group = {}) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - - if (thread_group) - CurrentThread::attachToGroup(thread_group); + ThreadGroupSwitcher switcher(thread_group, "ExternalLoader"); /// Do not account memory that was occupied by the dictionaries for the query/user context. MemoryTrackerBlockerInThread memory_blocker; diff --git a/src/Interpreters/ProfileEventsExt.cpp b/src/Interpreters/ProfileEventsExt.cpp index 82840d275b82..be9976809acb 100644 --- a/src/Interpreters/ProfileEventsExt.cpp +++ b/src/Interpreters/ProfileEventsExt.cpp @@ -12,6 +12,11 @@ #include #include +namespace DB::ErrorCodes +{ + extern const int LOGICAL_ERROR; +} + namespace ProfileEvents { @@ -105,6 +110,9 @@ void getProfileEvents( ThreadIdToCountersSnapshot & last_sent_snapshots) { using namespace DB; + if (!CurrentThread::isInitialized()) + throw Exception(ErrorCodes::LOGICAL_ERROR, "CurrentThread is not initialized"); + static const NamesAndTypesList column_names_and_types = { {"host_name", std::make_shared()}, {"current_time", std::make_shared()}, @@ -121,6 +129,9 @@ void getProfileEvents( block = std::move(temp_columns); MutableColumns columns = block.mutateColumns(); auto thread_group = CurrentThread::getGroup(); + if (!thread_group) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Current thread is not attached to any thread group"); + ThreadIdToCountersSnapshot new_snapshots; ProfileEventsSnapshot group_snapshot; diff --git a/src/Interpreters/ThreadStatusExt.cpp b/src/Interpreters/ThreadStatusExt.cpp index 1f7c6b1fe682..b22bd1252da3 100644 --- a/src/Interpreters/ThreadStatusExt.cpp +++ b/src/Interpreters/ThreadStatusExt.cpp @@ -146,22 +146,74 @@ void ThreadGroup::attachInternalProfileEventsQueue(const InternalProfileEventsQu shared_data.profile_queue_ptr = profile_queue; } -ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group) +ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group_, const char * thread_name, bool allow_existing_group) noexcept + : thread_group(std::move(thread_group_)) { - chassert(thread_group); + try + { + if (!thread_group) + return; + + prev_thread = current_thread; + prev_thread_group = CurrentThread::getGroup(); + if (prev_thread_group) + { + if (prev_thread_group == thread_group) + { + thread_group = nullptr; + prev_thread_group = nullptr; + return; + } + else if (!allow_existing_group) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Thread ({}) is already attached to a group (master_thread_id {})", thread_name, prev_thread_group->master_thread_id); + else + CurrentThread::detachFromGroupIfNotDetached(); + } + + if (!prev_thread) + throw Exception(ErrorCodes::LOGICAL_ERROR, "Tried to attach thread ({}) to a group, but the ThreadStatus is not initialized", thread_name); - /// might be nullptr - prev_thread_group = CurrentThread::getGroup(); + LockMemoryExceptionInThread lock_memory_tracker(VariableContext::Global); - CurrentThread::detachFromGroupIfNotDetached(); - CurrentThread::attachToGroup(thread_group); + CurrentThread::attachToGroup(thread_group); + if (thread_name[0] != '\0') + setThreadName(thread_name); + } + catch (...) + { + tryLogCurrentException(__PRETTY_FUNCTION__); + thread_group = nullptr; + prev_thread_group = nullptr; + } } ThreadGroupSwitcher::~ThreadGroupSwitcher() { - CurrentThread::detachFromGroupIfNotDetached(); - if (prev_thread_group) - CurrentThread::attachToGroup(prev_thread_group); + if (!thread_group) + return; + + try + { + ThreadStatus * cur_thread = current_thread; + ThreadGroupPtr cur_thread_group = CurrentThread::getGroup(); + if (cur_thread != prev_thread) + throw Exception(ErrorCodes::LOGICAL_ERROR, "ThreadGroupSwitcher-s are not properly nested: current thread changed between scope start ({}) and end ({})", prev_thread ? std::to_string(prev_thread->thread_id) : "nullptr", cur_thread ? std::to_string(cur_thread->thread_id) : "nullptr"); + if (cur_thread_group != thread_group) + throw Exception(ErrorCodes::LOGICAL_ERROR, "ThreadGroupSwitcher-s are not properly nested: current thread group changed between scope start (master_thread_id {}) and end ({})", thread_group->master_thread_id, cur_thread_group ? "master_thread_id " + std::to_string(cur_thread_group->master_thread_id) : "nullptr"); + thread_group.reset(); + + CurrentThread::detachFromGroupIfNotDetached(); + + if (prev_thread_group) + { + LockMemoryExceptionInThread lock_memory_tracker(VariableContext::Global); + CurrentThread::attachToGroup(prev_thread_group); + } + } + catch (...) + { + tryLogCurrentException(__PRETTY_FUNCTION__); + } } void ThreadStatus::attachInternalProfileEventsQueue(const InternalProfileEventsQueuePtr & profile_queue) diff --git a/src/Processors/Executors/CompletedPipelineExecutor.cpp b/src/Processors/Executors/CompletedPipelineExecutor.cpp index 888835c9beb4..125f3028fbfb 100644 --- a/src/Processors/Executors/CompletedPipelineExecutor.cpp +++ b/src/Processors/Executors/CompletedPipelineExecutor.cpp @@ -35,16 +35,9 @@ struct CompletedPipelineExecutor::Data static void threadFunction( CompletedPipelineExecutor::Data & data, ThreadGroupPtr thread_group, size_t num_threads, bool concurrency_control) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - setThreadName("QueryCompPipeEx"); - try { - if (thread_group) - CurrentThread::attachToGroup(thread_group); + ThreadGroupSwitcher switcher(thread_group, "QueryCompPipeEx"); data.executor->execute(num_threads, concurrency_control); } diff --git a/src/Processors/Executors/PipelineExecutor.cpp b/src/Processors/Executors/PipelineExecutor.cpp index 82cad471a29d..8b4ec27dcca1 100644 --- a/src/Processors/Executors/PipelineExecutor.cpp +++ b/src/Processors/Executors/PipelineExecutor.cpp @@ -359,14 +359,7 @@ void PipelineExecutor::spawnThreads() /// Start new thread pool->scheduleOrThrowOnError([this, thread_num, thread_group = CurrentThread::getGroup(), my_slot = std::move(slot)] { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - setThreadName("QueryPipelineEx"); - - if (thread_group) - CurrentThread::attachToGroup(thread_group); + ThreadGroupSwitcher switcher(thread_group, "QueryPipelineEx"); try { diff --git a/src/Processors/Executors/PullingAsyncPipelineExecutor.cpp b/src/Processors/Executors/PullingAsyncPipelineExecutor.cpp index d9fab88fe1f0..b5a01d3642a2 100644 --- a/src/Processors/Executors/PullingAsyncPipelineExecutor.cpp +++ b/src/Processors/Executors/PullingAsyncPipelineExecutor.cpp @@ -69,16 +69,9 @@ const Block & PullingAsyncPipelineExecutor::getHeader() const static void threadFunction( PullingAsyncPipelineExecutor::Data & data, ThreadGroupPtr thread_group, size_t num_threads, bool concurrency_control) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - setThreadName("QueryPullPipeEx"); - try { - if (thread_group) - CurrentThread::attachToGroup(thread_group); + ThreadGroupSwitcher switcher(thread_group, "QueryPullPipeEx"); data.executor->execute(num_threads, concurrency_control); } diff --git a/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp b/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp index 830a96533edf..6cb4241e1163 100644 --- a/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp +++ b/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp @@ -98,16 +98,9 @@ struct PushingAsyncPipelineExecutor::Data static void threadFunction( PushingAsyncPipelineExecutor::Data & data, ThreadGroupPtr thread_group, size_t num_threads, bool concurrency_control) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - setThreadName("QueryPushPipeEx"); - try { - if (thread_group) - CurrentThread::attachToGroup(thread_group); + ThreadGroupSwitcher switcher(thread_group, "QueryPushPipeEx"); data.executor->execute(num_threads, concurrency_control); } diff --git a/src/Processors/Formats/Impl/DWARFBlockInputFormat.cpp b/src/Processors/Formats/Impl/DWARFBlockInputFormat.cpp index 00a0d426dcfe..1ea288121e21 100644 --- a/src/Processors/Formats/Impl/DWARFBlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/DWARFBlockInputFormat.cpp @@ -244,12 +244,9 @@ void DWARFBlockInputFormat::initializeIfNeeded() pool->scheduleOrThrowOnError( [this, thread_group = CurrentThread::getGroup()]() { - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - SCOPE_EXIT_SAFE(if (thread_group) CurrentThread::detachFromGroupIfNotDetached();); try { - setThreadName("DWARFDecoder"); + ThreadGroupSwitcher switcher(thread_group, "DWARFDecoder"); std::unique_lock lock(mutex); while (!units_queue.empty() && !is_stopped) diff --git a/src/Processors/Formats/Impl/ParallelFormattingOutputFormat.cpp b/src/Processors/Formats/Impl/ParallelFormattingOutputFormat.cpp index 5d404d493a6b..6d218bb07363 100644 --- a/src/Processors/Formats/Impl/ParallelFormattingOutputFormat.cpp +++ b/src/Processors/Formats/Impl/ParallelFormattingOutputFormat.cpp @@ -125,13 +125,7 @@ namespace DB void ParallelFormattingOutputFormat::collectorThreadFunction(const ThreadGroupPtr & thread_group) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - setThreadName("Collector"); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); + ThreadGroupSwitcher switcher(thread_group, "Collector"); try { @@ -196,13 +190,7 @@ namespace DB void ParallelFormattingOutputFormat::formatterThreadFunction(size_t current_unit_number, size_t first_row_num, const ThreadGroupPtr & thread_group) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - setThreadName("Formatter"); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); + ThreadGroupSwitcher switcher(thread_group, "Formatter"); try { diff --git a/src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp b/src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp index 447adb1ed48f..208e0fbdceae 100644 --- a/src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp @@ -10,14 +10,7 @@ namespace DB void ParallelParsingInputFormat::segmentatorThreadFunction(ThreadGroupPtr thread_group) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroup(thread_group); - - setThreadName("Segmentator"); + ThreadGroupSwitcher switcher(thread_group, "Segmentator"); try { @@ -67,20 +60,13 @@ void ParallelParsingInputFormat::segmentatorThreadFunction(ThreadGroupPtr thread void ParallelParsingInputFormat::parserThreadFunction(ThreadGroupPtr thread_group, size_t current_ticket_number) { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); + ThreadGroupSwitcher switcher(thread_group, "ChunkParser"); const auto parser_unit_number = current_ticket_number % processing_units.size(); auto & unit = processing_units[parser_unit_number]; try { - setThreadName("ChunkParser"); - /* * This is kind of suspicious -- the input_process_creator contract with * respect to multithreaded use is not clear, but we hope that it is diff --git a/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp index d6ce2d81b750..9f461bfc9532 100644 --- a/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp @@ -833,13 +833,9 @@ void ParquetBlockInputFormat::scheduleRowGroup(size_t row_group_batch_idx) pool->scheduleOrThrowOnError( [this, row_group_batch_idx, thread_group = CurrentThread::getGroup()]() { - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - SCOPE_EXIT_SAFE(if (thread_group) CurrentThread::detachFromGroupIfNotDetached();); - try { - setThreadName("ParquetDecoder"); + ThreadGroupSwitcher switcher(thread_group, "ParquetDecoder"); threadFunction(row_group_batch_idx); } diff --git a/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp b/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp index 96d47b2ffb96..84b4e71e2fe0 100644 --- a/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp @@ -473,13 +473,9 @@ void ParquetBlockOutputFormat::startMoreThreadsIfNeeded(const std::unique_lockaggregator) { - // variant is moved here and will be destroyed in the destructor of the lambda function. pool->trySchedule( - [my_variant = std::move(variant), thread_group = CurrentThread::getGroup()]() + [my_variant = std::move(variant), thread_group = CurrentThread::getGroup()]() mutable { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - - setThreadName("AggregDestruct"); + ThreadGroupSwitcher switcher(thread_group, "AggregDestruct"); + my_variant.reset(); }); } } diff --git a/src/Storages/Distributed/DistributedSink.cpp b/src/Storages/Distributed/DistributedSink.cpp index ba6348c67235..db94976f1325 100644 --- a/src/Storages/Distributed/DistributedSink.cpp +++ b/src/Storages/Distributed/DistributedSink.cpp @@ -303,15 +303,9 @@ DistributedSink::runWritingJob(JobReplica & job, const Block & current_block, si if (isCancelled()) throw Exception(ErrorCodes::ABORTED, "Writing job was cancelled"); - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - OpenTelemetry::SpanHolder span(__PRETTY_FUNCTION__); + ThreadGroupSwitcher switcher(thread_group, "DistrOutStrProc"); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - setThreadName("DistrOutStrProc"); + OpenTelemetry::SpanHolder span(__PRETTY_FUNCTION__); ++job.blocks_started; @@ -571,12 +565,7 @@ void DistributedSink::onFinish() { pool->scheduleOrThrowOnError([&job, thread_group = CurrentThread::getGroup()]() { - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); + ThreadGroupSwitcher switcher(thread_group, ""); job.executor->finish(); }); diff --git a/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp b/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp index be44177847cc..ab2d0415cfb7 100644 --- a/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp +++ b/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp @@ -39,7 +39,7 @@ bool MergePlainMergeTreeTask::executeStep() std::optional switcher; if (merge_list_entry) { - switcher.emplace((*merge_list_entry)->thread_group); + switcher.emplace((*merge_list_entry)->thread_group, "", true); } switch (state) diff --git a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp index d3b6fe80b66e..704add54b4db 100644 --- a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp +++ b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp @@ -751,14 +751,7 @@ RangesInDataParts MergeTreeDataSelectExecutor::filterPartsByPrimaryKeyAndSkipInd { pool.scheduleOrThrow([&, part_index, thread_group = CurrentThread::getGroup()] { - setThreadName("MergeTreeIndex"); - - SCOPE_EXIT_SAFE( - if (thread_group) - CurrentThread::detachFromGroupIfNotDetached(); - ); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); + ThreadGroupSwitcher switcher(thread_group, "MergeTreeIndex"); process_part(part_index); }, Priority{}, context->getSettingsRef().lock_acquire_timeout.totalMicroseconds()); diff --git a/src/Storages/MergeTree/MutatePlainMergeTreeTask.cpp b/src/Storages/MergeTree/MutatePlainMergeTreeTask.cpp index 10461eb59421..6d628481286b 100644 --- a/src/Storages/MergeTree/MutatePlainMergeTreeTask.cpp +++ b/src/Storages/MergeTree/MutatePlainMergeTreeTask.cpp @@ -74,7 +74,7 @@ bool MutatePlainMergeTreeTask::executeStep() /// Make out memory tracker a parent of current thread memory tracker std::optional switcher; if (merge_list_entry) - switcher.emplace((*merge_list_entry)->thread_group); + switcher.emplace((*merge_list_entry)->thread_group, "", true); switch (state) { diff --git a/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp b/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp index 24929365b72e..e42d4aff7d9a 100644 --- a/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp +++ b/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp @@ -139,7 +139,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl() { std::optional switcher; if (merge_mutate_entry) - switcher.emplace((*merge_mutate_entry)->thread_group); + switcher.emplace((*merge_mutate_entry)->thread_group, "", true); auto remove_processed_entry = [&] () -> bool { diff --git a/src/Storages/System/StorageSystemReplicas.cpp b/src/Storages/System/StorageSystemReplicas.cpp index c197172174e7..b9f5b541e376 100644 --- a/src/Storages/System/StorageSystemReplicas.cpp +++ b/src/Storages/System/StorageSystemReplicas.cpp @@ -138,11 +138,7 @@ class StatusRequestsPool auto get_status_task = [this, storage, with_zk_fields, promise, thread_group = CurrentThread::getGroup()]() mutable { - SCOPE_EXIT_SAFE(if (thread_group) CurrentThread::detachFromGroupIfNotDetached();); - if (thread_group) - CurrentThread::attachToGroupIfDetached(thread_group); - - setThreadName("SystemReplicas"); + ThreadGroupSwitcher switcher(thread_group, "SystemReplicas"); try {