diff --git a/src/common/base/src/runtime/runtime.rs b/src/common/base/src/runtime/runtime.rs index 9154a36a382f..8b13af7673c3 100644 --- a/src/common/base/src/runtime/runtime.rs +++ b/src/common/base/src/runtime/runtime.rs @@ -151,24 +151,25 @@ impl Runtime { let n = name.clone(); // Block the runtime to shutdown. - let join_handler = Thread::spawn(move || { - let _ = runtime.block_on(recv_stop); - info!( - "Runtime({:?}) received shutdown signal, start to shut down", - n - ); - - match !cfg!(debug_assertions) { - true => false, - false => { - let instant = Instant::now(); - // We wait up to 3 seconds to complete the runtime shutdown. - runtime.shutdown_timeout(Duration::from_secs(3)); - - instant.elapsed() >= Duration::from_secs(3) + let join_handler = + Thread::named_spawn(n.as_ref().map(|n| format!("wait-to-drop-{n}")), move || { + let _ = runtime.block_on(recv_stop); + info!( + "Runtime({:?}) received shutdown signal, start to shut down", + n + ); + + match !cfg!(debug_assertions) { + true => false, + false => { + let instant = Instant::now(); + // We wait up to 3 seconds to complete the runtime shutdown. + runtime.shutdown_timeout(Duration::from_secs(3)); + + instant.elapsed() >= Duration::from_secs(3) + } } - } - }); + }); Ok(Runtime { handle, diff --git a/src/query/service/src/pipelines/builders/builder_compact.rs b/src/query/service/src/pipelines/builders/builder_compact.rs index 03fadf872e31..0a4832272f3a 100644 --- a/src/query/service/src/pipelines/builders/builder_compact.rs +++ b/src/query/service/src/pipelines/builders/builder_compact.rs @@ -73,18 +73,20 @@ impl PipelineBuilder { let column_ids = compact_block.column_ids.clone(); self.main_pipeline.set_on_init(move || { let ctx = query_ctx.clone(); - let partitions = Runtime::with_worker_threads(2, None)?.block_on(async move { - let partitions = BlockCompactMutator::build_compact_tasks( - ctx.clone(), - column_ids.clone(), - cluster_key_id, - thresholds, - lazy_parts, - ) - .await?; + let partitions = + Runtime::with_worker_threads(2, Some("build_compact_tasks".to_string()))? + .block_on(async move { + let partitions = BlockCompactMutator::build_compact_tasks( + ctx.clone(), + column_ids.clone(), + cluster_key_id, + thresholds, + lazy_parts, + ) + .await?; - Result::<_>::Ok(partitions) - })?; + Result::<_>::Ok(partitions) + })?; let partitions = Partitions::create(PartitionsShuffleKind::Mod, partitions); query_ctx.set_partitions(partitions)?; diff --git a/src/query/service/src/pipelines/builders/builder_mutation_source.rs b/src/query/service/src/pipelines/builders/builder_mutation_source.rs index 2bc4e76f4aea..ee06e7362b7d 100644 --- a/src/query/service/src/pipelines/builders/builder_mutation_source.rs +++ b/src/query/service/src/pipelines/builders/builder_mutation_source.rs @@ -68,20 +68,21 @@ impl PipelineBuilder { segment_locations, block_count: None, }; - Runtime::with_worker_threads(2, None)?.block_on(async move { - let (partitions, _) = table_clone - .do_mutation_block_pruning( - ctx_clone, - filters_clone, - projection, - prune_ctx, - true, - true, - ) - .await?; - ctx.set_partitions(partitions)?; - Ok(()) - })?; + Runtime::with_worker_threads(2, Some("do_mutation_block_pruning".to_string()))? + .block_on(async move { + let (partitions, _) = table_clone + .do_mutation_block_pruning( + ctx_clone, + filters_clone, + projection, + prune_ctx, + true, + true, + ) + .await?; + ctx.set_partitions(partitions)?; + Ok(()) + })?; } else { self.ctx .set_partitions(mutation_source.partitions.clone())?; diff --git a/src/query/storages/fuse/src/operations/read_data.rs b/src/query/storages/fuse/src/operations/read_data.rs index 75a9ca4dbf32..5a7cfd7c711b 100644 --- a/src/query/storages/fuse/src/operations/read_data.rs +++ b/src/query/storages/fuse/src/operations/read_data.rs @@ -209,7 +209,8 @@ impl FuseTable { // We cannot use the runtime associated with the query to avoid increasing its lifetime. GlobalIORuntime::instance().spawn(async move { // avoid block global io runtime - let runtime = Runtime::with_worker_threads(2, None)?; + let runtime = + Runtime::with_worker_threads(2, Some("prune-snap-blk".to_string()))?; let handler = runtime.spawn(async move { match table .prune_snapshot_blocks( diff --git a/src/query/storages/fuse/src/operations/read_partitions.rs b/src/query/storages/fuse/src/operations/read_partitions.rs index 08ca97cae625..b116798722aa 100644 --- a/src/query/storages/fuse/src/operations/read_partitions.rs +++ b/src/query/storages/fuse/src/operations/read_partitions.rs @@ -200,7 +200,7 @@ impl FuseTable { // We cannot use the runtime associated with the query to avoid increasing its lifetime. GlobalIORuntime::instance().spawn(async move { // avoid block global io runtime - let runtime = Runtime::with_worker_threads(2, None)?; + let runtime = Runtime::with_worker_threads(2, Some("send-parts".to_string()))?; let join_handler = runtime.spawn(async move { for part in part.partitions { @@ -241,7 +241,7 @@ impl FuseTable { // We cannot use the runtime associated with the query to avoid increasing its lifetime. GlobalIORuntime::instance().spawn(async move { // avoid block global io runtime - let runtime = Runtime::with_worker_threads(2, None)?; + let runtime = Runtime::with_worker_threads(2, Some("prune-seg".to_string()))?; let join_handler = runtime.spawn(async move { let segment_pruned_result = pruner.clone().segment_pruning(lazy_init_segments).await?;