refactor: set name for all runtime worker threads. (#17084)

databendlabs · Dec 24, 2024 · ff6cb3e · ff6cb3e
1 parent b0bb940
commit ff6cb3e
Show file tree

Hide file tree

Showing 5 changed files with 50 additions and 45 deletions.
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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?;