name threads for observability (#247)

lib/collection/src/collection.rs

@@ -401,9 +401,10 @@ impl Drop for Collection {
                 // So the workaround for move out the runtime handler and drop it in the separate thread.
                 // The proper solution is to reconsider the collection to be an owner of the runtime
 
-                let thread_handler = thread::spawn(move || {
-                    drop(handle);
-                });
+                let thread_handler = thread::Builder::new()
+                    .name("collection_drop".to_string())
+                    .spawn(move || drop(handle))
+                    .unwrap();
                 thread_handler.join().unwrap();
             }
         }

lib/collection/src/collection_builder/collection_builder_base.rs

@@ -1,6 +1,7 @@
 use std::cmp::max;
 use std::fs::create_dir_all;
 use std::path::Path;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use num_cpus;
@@ -37,6 +38,11 @@ pub fn construct_collection(
     };
     let optimize_runtime = runtime::Builder::new_multi_thread()
         .worker_threads(2)
+        .thread_name_fn(|| {
+            static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
+            let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
+            format!("optimizer-{}", id)
+        })
         .max_blocking_threads(blocking_threads)
         .build()
         .unwrap();

lib/segment/tests/segment_builder_test.rs

@@ -89,10 +89,13 @@ mod tests {
 
         let stopped_t = stopped.clone();
 
-        std::thread::spawn(move || {
-            std::thread::sleep(Duration::from_millis(stop_timeout_millis));
-            stopped_t.store(true, Ordering::Release);
-        });
+        std::thread::Builder::new()
+            .name("build_estimator_timeout".to_string())
+            .spawn(move || {
+                std::thread::sleep(Duration::from_millis(stop_timeout_millis));
+                stopped_t.store(true, Ordering::Release);
+            })
+            .unwrap();
 
         let res = builder.build(&*stopped);
 

src/common/helpers.rs

@@ -1,3 +1,4 @@
+use std::sync::atomic::{AtomicUsize, Ordering};
 use tokio::runtime;
 use tokio::runtime::Runtime;
 
@@ -11,5 +12,10 @@ pub fn create_search_runtime(max_search_threads: usize) -> std::io::Result<Runti
 
     runtime::Builder::new_multi_thread()
         .worker_threads(search_threads)
+        .thread_name_fn(|| {
+            static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
+            let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
+            format!("search-{}", id)
+        })
         .build()
 }

src/main.rs

@@ -41,14 +41,20 @@ fn main() -> std::io::Result<()> {
     {
         let toc_arc = toc_arc.clone();
         let settings = settings.clone();
-        let handle = thread::spawn(move || actix::init(toc_arc, settings));
+        let handle = thread::Builder::new()
+            .name("web".to_string())
+            .spawn(move || actix::init(toc_arc, settings))
+            .unwrap();
         handles.push(handle);
     }
     #[cfg(feature = "grpc")]
     {
         let toc_arc = toc_arc.clone();
         let settings = settings.clone();
-        let handle = thread::spawn(move || tonic::init(toc_arc, settings));
+        let handle = thread::Builder::new()
+            .name("grpc".to_string())
+            .spawn(move || tonic::init(toc_arc, settings))
+            .unwrap();
         handles.push(handle);
     }
 
@@ -59,26 +65,29 @@ fn main() -> std::io::Result<()> {
 
         const DEADLOCK_CHECK_PERIOD: Duration = Duration::from_secs(10);
 
-        thread::spawn(move || loop {
-            thread::sleep(DEADLOCK_CHECK_PERIOD);
-            let deadlocks = deadlock::check_deadlock();
-            if deadlocks.is_empty() {
-                continue;
-            }
-
-            let mut error = format!("{} deadlocks detected\n", deadlocks.len());
-            for (i, threads) in deadlocks.iter().enumerate() {
-                error.push_str(&format!("Deadlock #{}\n", i));
-                for t in threads {
-                    error.push_str(&format!(
-                        "Thread Id {:#?}\n{:#?}\n",
-                        t.thread_id(),
-                        t.backtrace()
-                    ));
+        thread::Builder::new()
+            .name("deadlock_checker".to_string())
+            .spawn(move || loop {
+                thread::sleep(DEADLOCK_CHECK_PERIOD);
+                let deadlocks = deadlock::check_deadlock();
+                if deadlocks.is_empty() {
+                    continue;
                 }
-            }
-            log::error!("{}", error);
-        });
+
+                let mut error = format!("{} deadlocks detected\n", deadlocks.len());
+                for (i, threads) in deadlocks.iter().enumerate() {
+                    error.push_str(&format!("Deadlock #{}\n", i));
+                    for t in threads {
+                        error.push_str(&format!(
+                            "Thread Id {:#?}\n{:#?}\n",
+                            t.thread_id(),
+                            t.backtrace()
+                        ));
+                    }
+                }
+                log::error!("{}", error);
+            })
+            .unwrap();
     }
 
     for handle in handles.into_iter() {

src/tonic/mod.rs

@@ -10,6 +10,7 @@ use qdrant::points_server::PointsServer;
 use qdrant::qdrant_server::{Qdrant, QdrantServer};
 use qdrant::{HealthCheckReply, HealthCheckRequest};
 use std::net::{IpAddr, SocketAddr};
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use storage::content_manager::toc::TableOfContent;
 use tokio::{runtime, signal};
@@ -41,6 +42,11 @@ pub fn init(toc: Arc<TableOfContent>, settings: Settings) -> std::io::Result<()>
     let tonic_runtime = runtime::Builder::new_multi_thread()
         .enable_io()
         .enable_time()
+        .thread_name_fn(|| {
+            static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
+            let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
+            format!("tonic-{}", id)
+        })
         .build()?;
     tonic_runtime
         .block_on(async {