From 4c73d1ba8f46b7dc1d41ecf91cda7571ffef89a0 Mon Sep 17 00:00:00 2001 From: Arnaud Gourlay Date: Mon, 24 Jan 2022 13:39:12 +0100 Subject: [PATCH] name threads for observability (#247) --- lib/collection/src/collection.rs | 7 +-- .../collection_builder_base.rs | 6 +++ lib/segment/tests/segment_builder_test.rs | 11 ++-- src/common/helpers.rs | 6 +++ src/main.rs | 51 +++++++++++-------- src/tonic/mod.rs | 6 +++ 6 files changed, 59 insertions(+), 28 deletions(-) diff --git a/lib/collection/src/collection.rs b/lib/collection/src/collection.rs index 5e16f09a07..9a91d1cb56 100644 --- a/lib/collection/src/collection.rs +++ b/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(); } } diff --git a/lib/collection/src/collection_builder/collection_builder_base.rs b/lib/collection/src/collection_builder/collection_builder_base.rs index 15cc844e29..3c8c3dbd2d 100644 --- a/lib/collection/src/collection_builder/collection_builder_base.rs +++ b/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(); diff --git a/lib/segment/tests/segment_builder_test.rs b/lib/segment/tests/segment_builder_test.rs index f8da120606..96a6975972 100644 --- a/lib/segment/tests/segment_builder_test.rs +++ b/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); diff --git a/src/common/helpers.rs b/src/common/helpers.rs index 21db8f9643..7e9b133f70 100644 --- a/src/common/helpers.rs +++ b/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 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() { diff --git a/src/tonic/mod.rs b/src/tonic/mod.rs index b98d62bcd3..02bbd60692 100644 --- a/src/tonic/mod.rs +++ b/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, 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 {