adjust active thread count adaptively

src/metrics.rs

@@ -27,6 +27,17 @@ lazy_static! {
     )
     .unwrap();
 
+    /// The count of active workers.
+    pub static ref ACTIVE_WORKERS_COUNT: HistogramVec = HistogramVec::new(
+        histogram_opts!(
+            "yatp_active_workers_count",
+            "the count of backup workers",
+            vec![1.0, 2.0, 4.0, 6.0, 8.0, 12.0, 16.0, 24.0, 32.0, 48.0, 64.0, 96.0, 128.0]
+        ),
+        &["name"]
+    )
+    .unwrap();
+
     static ref NAMESPACE: Mutex<Option<String>> = Mutex::new(None);
 }
 

src/pool/builder.rs

@@ -1,5 +1,6 @@
 // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
 
+use crate::metrics::*;
 use crate::pool::spawn::QueueCore;
 use crate::pool::worker::WorkerThread;
 use crate::pool::{CloneRunnerBuilder, Local, Remote, Runner, RunnerBuilder, ThreadPool};
@@ -76,14 +77,22 @@ where
         F::Runner: Runner<TaskCell = T> + Send + 'static,
     {
         let mut threads = Vec::with_capacity(self.builder.sched_config.max_thread_count);
+        let active_count_histogram = ACTIVE_WORKERS_COUNT
+            .get_metric_with_label_values(&[&self.builder.name_prefix])
+            .unwrap();
         for (i, local_queue) in self.local_queues.into_iter().enumerate() {
             let runner = factory.build();
             let name = format!("{}-{}", self.builder.name_prefix, i);
             let mut builder = thread::Builder::new().name(name);
             if let Some(size) = self.builder.stack_size {
                 builder = builder.stack_size(size)
             }
-            let local = Local::new(i + 1, local_queue, self.core.clone());
+            let local = Local::new(
+                i + 1,
+                local_queue,
+                self.core.clone(),
+                active_count_histogram.clone(),
+            );
             let thd = WorkerThread::new(local, runner);
             threads.push(
                 builder