From be620ef36b6293b42166afb0d7d48f26f88dfcaf Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 14:29:41 +0000 Subject: [PATCH 1/8] add burst monitoring for global task injector Signed-off-by: zyguan --- src/metrics.rs | 152 ++++++++++++++++++++++++++++++++++++++++++ src/pool/builder.rs | 34 +++++++++- src/pool/spawn.rs | 88 ++++++++++++++++++++++-- src/pool/tests.rs | 36 ++++++++++ src/queue/priority.rs | 1 + 5 files changed, 306 insertions(+), 5 deletions(-) diff --git a/src/metrics.rs b/src/metrics.rs index 7152c3b..ccce24c 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -3,7 +3,9 @@ //! Metrics of the thread pool. use lazy_static::lazy_static; +use prometheus::core::{Collector, Desc, Metric, MetricVec, MetricVecBuilder}; use prometheus::*; + use std::sync::Mutex; lazy_static! { @@ -71,6 +73,16 @@ lazy_static! { ) .unwrap(); + /// Max enqueue throughput of the global task injector (QueueCore) since last scrape. + pub static ref QUEUE_CORE_BURST_THROUGHPUT: MaxGaugeVec = MaxGaugeVec::new( + new_opts( + "yatp_queue_core_burst_throughput", + "max enqueue throughput (tasks/sec) of the global task injector since last scrape" + ), + &["name"] + ) + .unwrap(); + static ref NAMESPACE: Mutex> = Mutex::new(None); } @@ -98,3 +110,143 @@ fn new_histogram_opts(name: &str, help: &str, buckets: Vec) -> HistogramOpt opts } + +/// A gauge that tracks the maximum value since the last scrape. +#[derive(Clone, Debug)] +pub struct MaxGauge { + gauge: Gauge, +} + +impl MaxGauge { + /// Observe a value, keeping the maximum since the last scrape. Note that it's not a accurate maximum if there are + /// concurrent calls to `observe`, but it should be good enough for most use cases. + pub fn observe(&self, v: f64) { + let current = self.gauge.get(); + if v > current { + self.gauge.set(v); + } + } +} + +impl Collector for MaxGauge { + fn desc(&self) -> Vec<&Desc> { + self.gauge.desc() + } + + fn collect(&self) -> Vec { + let mfs = self.gauge.collect(); + self.gauge.set(0.0); + mfs + } +} + +impl Metric for MaxGauge { + fn metric(&self) -> proto::Metric { + let m = self.gauge.metric(); + self.gauge.set(0.0); + m + } +} + +/// Builder for `MaxGaugeVec`. +#[derive(Clone, Debug)] +pub struct MaxGaugeVecBuilder; + +impl MaxGaugeVecBuilder { + /// Create a new `MaxGaugeVecBuilder`. + pub fn new() -> Self { + Self + } +} + +impl MetricVecBuilder for MaxGaugeVecBuilder { + type M = MaxGauge; + type P = Opts; + + fn build(&self, opts: &Opts, vals: &[&str]) -> Result { + let mut opts = opts.clone(); + for (name, val) in opts.variable_labels.iter().zip(vals.iter()) { + opts.const_labels.insert(name.clone(), (*val).to_owned()); + } + opts.variable_labels.clear(); + + let gauge = Gauge::with_opts(opts)?; + Ok(MaxGauge { gauge }) + } +} + +/// A `MetricVec` for `MaxGauge` values, partitioned by label values. +#[derive(Clone, Debug)] +pub struct MaxGaugeVec { + inner: MetricVec, +} + +impl MaxGaugeVec { + /// Create a new `MaxGaugeVec` with the given label names. + pub fn new(opts: Opts, label_names: &[&str]) -> Result { + let variable_names = label_names.iter().map(|s| (*s).to_owned()).collect(); + let opts = opts.variable_labels(variable_names); + let inner = MetricVec::create(proto::MetricType::GAUGE, MaxGaugeVecBuilder::new(), opts)?; + + Ok(Self { inner }) + } +} + +impl std::ops::Deref for MaxGaugeVec { + type Target = MetricVec; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl Collector for MaxGaugeVec { + fn desc(&self) -> Vec<&Desc> { + self.inner.desc() + } + + fn collect(&self) -> Vec { + self.inner.collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_max_gauge_resets_on_metric() { + let gauge = Gauge::with_opts(Opts::new("test_max_gauge", "test max gauge")).unwrap(); + let max = MaxGauge { gauge }; + + max.observe(1.0); + max.observe(3.0); + max.observe(2.0); + assert_eq!(max.gauge.get(), 3.0); + + let _ = max.metric(); + assert_eq!(max.gauge.get(), 0.0); + } + + #[test] + fn test_max_gauge_vec_resets_on_collect() { + let vec = MaxGaugeVec::new( + Opts::new("test_max_gauge_vec", "test max gauge vec"), + &["l1"], + ) + .unwrap(); + let m = vec.with_label_values(&["v1"]); + m.observe(5.0); + + let mfs = vec.collect(); + assert_eq!(mfs.len(), 1); + let mf = &mfs[0]; + let metric = mf.get_metric().get(0).unwrap(); + assert_eq!(metric.get_label().len(), 1); + assert_eq!(metric.get_label()[0].get_name(), "l1"); + assert_eq!(metric.get_label()[0].get_value(), "v1"); + assert_eq!(metric.get_gauge().get_value(), 5.0); + + assert_eq!(m.gauge.get(), 0.0); + } +} diff --git a/src/pool/builder.rs b/src/pool/builder.rs index 29e3be3..eba9fd2 100644 --- a/src/pool/builder.rs +++ b/src/pool/builder.rs @@ -38,6 +38,12 @@ pub struct SchedConfig { pub alloc_slot_backoff: Duration, } +#[derive(Clone, Copy)] +pub(crate) struct BurstMonitorConfig { + pub(crate) per_worker_multiplier: usize, + pub(crate) min_sample_size: usize, +} + impl Default for SchedConfig { fn default() -> SchedConfig { SchedConfig { @@ -129,6 +135,7 @@ pub struct Builder { name_prefix: String, stack_size: Option, sched_config: SchedConfig, + burst_monitoring: Option, } impl Builder { @@ -138,6 +145,7 @@ impl Builder { name_prefix: name_prefix.into(), stack_size: None, sched_config: SchedConfig::default(), + burst_monitoring: None, } } @@ -212,6 +220,23 @@ impl Builder { self } + /// Enables monitoring of global injector enqueue burst throughput. + /// + /// The sample size is computed as `max(min_sample_size, per_worker_multiplier * active_workers)`. + pub fn enable_burst_monitoring( + &mut self, + per_worker_multiplier: usize, + min_sample_size: usize, + ) -> &mut Self { + if per_worker_multiplier > 0 && min_sample_size > 0 { + self.burst_monitoring = Some(BurstMonitorConfig { + per_worker_multiplier, + min_sample_size, + }); + } + self + } + /// Freezes the configurations and returns the task scheduler and /// a builder to for lazy spawning threads. /// @@ -252,7 +277,14 @@ impl Builder { .store(self.sched_config.min_thread_count, Ordering::SeqCst); } let (injector, local_queues) = queue::build(queue_type, self.sched_config.max_thread_count); - let core = Arc::new(QueueCore::new(injector, self.sched_config.clone())); + let burst_monitor = self + .burst_monitoring + .map(|cfg| super::spawn::BurstMonitor::new(&self.name_prefix, cfg)); + let core = Arc::new(QueueCore::new( + injector, + self.sched_config.clone(), + burst_monitor, + )); ( Remote::new(core.clone()), diff --git a/src/pool/spawn.rs b/src/pool/spawn.rs index f66f91a..94e85c0 100644 --- a/src/pool/spawn.rs +++ b/src/pool/spawn.rs @@ -4,14 +4,17 @@ //! woken up when new tasks arrived and go to sleep when there are no //! tasks waiting to be handled. +use super::builder::BurstMonitorConfig; +use crate::metrics::{MaxGauge, QUEUE_CORE_BURST_THROUGHPUT}; use crate::pool::SchedConfig; use crate::queue::{Extras, LocalQueue, Pop, TaskCell, TaskInjector, WithExtras}; use fail::fail_point; use parking_lot_core::{FilterOp, ParkResult, ParkToken, UnparkToken}; use std::sync::{ - atomic::{AtomicUsize, Ordering}, - Arc, Weak, + atomic::{AtomicU64, AtomicUsize, Ordering}, + Arc, OnceLock, Weak, }; +use std::time::Instant; /// An usize is used to trace the threads that are working actively. /// To save additional memory and atomic operation, the number and @@ -26,6 +29,74 @@ const SHUTDOWN_BIT: usize = 1; const WORKER_COUNT_SHIFT: usize = 1; const WORKER_COUNT_BASE: usize = 2; +fn now_ns() -> u64 { + static START: OnceLock = OnceLock::new(); + START.get_or_init(Instant::now).elapsed().as_nanos() as u64 +} + +pub(crate) struct BurstMonitor { + per_worker_multiplier: usize, + min_sample_size: usize, + count: AtomicUsize, + sample_target: AtomicUsize, + start_ns: AtomicU64, + metric: MaxGauge, +} + +impl BurstMonitor { + pub(crate) fn new(name: &str, config: BurstMonitorConfig) -> BurstMonitor { + let metric = QUEUE_CORE_BURST_THROUGHPUT + .get_metric_with_label_values(&[name]) + .unwrap(); + BurstMonitor { + per_worker_multiplier: config.per_worker_multiplier, + min_sample_size: config.min_sample_size, + count: AtomicUsize::new(0), + sample_target: AtomicUsize::new(0), + start_ns: AtomicU64::new(0), + metric, + } + } + + fn on_enqueue(&self, active_workers: &AtomicUsize) { + let new_count = self.count.fetch_add(1, Ordering::Relaxed) + 1; + if new_count == 1 { + let workers = active_workers.load(Ordering::Relaxed) >> WORKER_COUNT_SHIFT; + let target = self + .per_worker_multiplier + .saturating_mul(workers) + .max(self.min_sample_size); + self.sample_target.store(target, Ordering::Relaxed); + self.start_ns.store(now_ns(), Ordering::Relaxed); + } + + let target = self.sample_target.load(Ordering::Relaxed); + if target == 0 || new_count < target { + return; + } + + let start_ns = self.start_ns.load(Ordering::Relaxed); + if let Ok(count) = self + .count + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + if c >= target { + Some(0) + } else { + None + } + }) + { + // `count` can exceed `target` under contention; it reflects actual enqueues + // observed before the reset, so the measured window size is variable. + let elapsed_ns = now_ns().saturating_sub(start_ns); + if elapsed_ns > 0 { + let throughput = (count as f64) / (elapsed_ns as f64 / 1_000_000_000.0); + self.metric.observe(throughput); + } + } + } +} + /// Checks if shutdown bit is set. pub fn is_shutdown(cnt: usize) -> bool { cnt & SHUTDOWN_BIT == SHUTDOWN_BIT @@ -39,14 +110,20 @@ pub(crate) struct QueueCore { global_queue: TaskInjector, active_workers: AtomicUsize, config: SchedConfig, + burst_monitor: Option, } impl QueueCore { - pub fn new(global_queue: TaskInjector, config: SchedConfig) -> QueueCore { + pub fn new( + global_queue: TaskInjector, + config: SchedConfig, + burst_monitor: Option, + ) -> QueueCore { QueueCore { global_queue, active_workers: AtomicUsize::new(config.max_thread_count << WORKER_COUNT_SHIFT), config, + burst_monitor, } } @@ -159,6 +236,9 @@ impl QueueCore { /// /// `source` is used to trace who triggers the action. fn push(&self, source: usize, task: T) { + if let Some(monitor) = self.burst_monitor.as_ref() { + monitor.on_enqueue(&self.active_workers); + } self.global_queue.push(task); self.ensure_workers(source); } @@ -348,7 +428,7 @@ where { let queue_type = queue_type.into(); let (global, locals) = crate::queue::build(queue_type, config.max_thread_count); - let core = Arc::new(QueueCore::new(global, config)); + let core = Arc::new(QueueCore::new(global, config, None)); let l = locals .into_iter() .enumerate() diff --git a/src/pool/tests.rs b/src/pool/tests.rs index 9b96737..ca24940 100644 --- a/src/pool/tests.rs +++ b/src/pool/tests.rs @@ -1,8 +1,10 @@ // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0. +use crate::metrics::QUEUE_CORE_BURST_THROUGHPUT; use crate::pool::*; use crate::task::callback::Handle; use futures_timer::Delay; +use prometheus::core::Metric as _; use rand::seq::SliceRandom; use std::sync::mpsc; use std::thread; @@ -294,3 +296,37 @@ fn test_scale_down_workers() { pool.shutdown(); } + +#[test] +fn test_burst_monitoring() { + let name = "test_burst_monitoring"; + let pool = Builder::new(name) + .max_thread_count(1) + .enable_burst_monitoring(1, 200) + .build_callback_pool(); + + let metric = QUEUE_CORE_BURST_THROUGHPUT + .get_metric_with_label_values(&[name]) + .unwrap(); + let spawn_n = |n| { + for _ in 0..n { + pool.spawn(|_: &mut Handle<'_>| {}); + } + }; + + // spawn at throughput of 500 tasks/sec + spawn_n(100); + thread::sleep(Duration::from_millis(400)); + spawn_n(100); + let value = metric.metric().get_gauge().get_value(); + assert!((value - 500.0).abs() < 10.0); + + // spawn at throughput of 1000 tasks/sec + spawn_n(100); + thread::sleep(Duration::from_millis(200)); + spawn_n(100); + let value = metric.metric().get_gauge().get_value(); + assert!((value - 1000.0).abs() < 10.0); + + pool.shutdown(); +} diff --git a/src/queue/priority.rs b/src/queue/priority.rs index 017728d..240ccec 100644 --- a/src/queue/priority.rs +++ b/src/queue/priority.rs @@ -341,6 +341,7 @@ mod tests { let core = Arc::new(crate::pool::spawn::QueueCore::new( injecter, crate::pool::SchedConfig::default(), + None, )); let mut locals: Vec<_> = locals .into_iter() From 00ad45f9a1b34018d12cee45d9aceb7a0d9edbcd Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 14:58:30 +0000 Subject: [PATCH 2/8] fix lint issue Signed-off-by: zyguan --- src/pool/spawn.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/pool/spawn.rs b/src/pool/spawn.rs index 94e85c0..eadd48e 100644 --- a/src/pool/spawn.rs +++ b/src/pool/spawn.rs @@ -288,8 +288,10 @@ impl Clone for Remote { /// Note that implements of Runner assumes `Remote` is `Sync` and `Send`. /// So we need to use assert trait to ensure the constraint at compile time /// to avoid future breaks. +#[allow(dead_code)] trait AssertSync: Sync {} impl AssertSync for Remote {} +#[allow(dead_code)] trait AssertSend: Send {} impl AssertSend for Remote {} From f175ad0901469e1ca498bc7d3eb91a9888cf95c4 Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 16:18:16 +0000 Subject: [PATCH 3/8] address the comments Signed-off-by: zyguan --- src/metrics.rs | 74 ++++++++++++++++++++++++++++++++++++----------- src/pool/tests.rs | 20 +++++++------ 2 files changed, 68 insertions(+), 26 deletions(-) diff --git a/src/metrics.rs b/src/metrics.rs index ccce24c..d2cb0d7 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -6,7 +6,8 @@ use lazy_static::lazy_static; use prometheus::core::{Collector, Desc, Metric, MetricVec, MetricVecBuilder}; use prometheus::*; -use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; lazy_static! { /// Elapsed time of each level in the multilevel task queue. @@ -115,16 +116,54 @@ fn new_histogram_opts(name: &str, help: &str, buckets: Vec) -> HistogramOpt #[derive(Clone, Debug)] pub struct MaxGauge { gauge: Gauge, + max_val: Arc, } impl MaxGauge { - /// Observe a value, keeping the maximum since the last scrape. Note that it's not a accurate maximum if there are - /// concurrent calls to `observe`, but it should be good enough for most use cases. + /// Wraps a `Gauge` to create a `MaxGauge`. The `Gauge` should not be used directly after being wrapped, otherwise + /// the maximum tracking will be broken. + pub fn wrap(gauge: Gauge) -> Self { + let val = gauge.get().to_bits(); + Self { + gauge, + max_val: Arc::new(AtomicU64::new(val)), + } + } + + /// Observe a value, keeping the maximum since the last scrape. pub fn observe(&self, v: f64) { - let current = self.gauge.get(); - if v > current { - self.gauge.set(v); + if !v.is_finite() { + return; } + let mut current = self.max_val.load(Ordering::Relaxed); + loop { + if v <= f64::from_bits(current) { + break; + } + match self.max_val.compare_exchange_weak( + current, + v.to_bits(), + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => { + self.gauge.set(v); + break; + } + Err(actual) => current = actual, + } + } + } + + /// Get the current maximum value without resetting it. + pub fn get(&self) -> f64 { + let val = self.max_val.load(Ordering::Relaxed); + f64::from_bits(val) + } + + fn take(&self) -> f64 { + let val = self.max_val.swap(0f64.to_bits(), Ordering::Relaxed); + f64::from_bits(val) } } @@ -134,17 +173,17 @@ impl Collector for MaxGauge { } fn collect(&self) -> Vec { - let mfs = self.gauge.collect(); - self.gauge.set(0.0); - mfs + let val = self.take(); + self.gauge.set(val); + self.gauge.collect() } } impl Metric for MaxGauge { fn metric(&self) -> proto::Metric { - let m = self.gauge.metric(); - self.gauge.set(0.0); - m + let val = self.take(); + self.gauge.set(val); + self.gauge.metric() } } @@ -171,7 +210,7 @@ impl MetricVecBuilder for MaxGaugeVecBuilder { opts.variable_labels.clear(); let gauge = Gauge::with_opts(opts)?; - Ok(MaxGauge { gauge }) + Ok(MaxGauge::wrap(gauge)) } } @@ -217,15 +256,16 @@ mod tests { #[test] fn test_max_gauge_resets_on_metric() { let gauge = Gauge::with_opts(Opts::new("test_max_gauge", "test max gauge")).unwrap(); - let max = MaxGauge { gauge }; + let max = MaxGauge::wrap(gauge); max.observe(1.0); max.observe(3.0); max.observe(2.0); - assert_eq!(max.gauge.get(), 3.0); + assert_eq!(max.get(), 3.0); let _ = max.metric(); - assert_eq!(max.gauge.get(), 0.0); + assert_eq!(max.get(), 0.0); + assert_eq!(max.gauge.get(), 3.0); } #[test] @@ -247,6 +287,6 @@ mod tests { assert_eq!(metric.get_label()[0].get_value(), "v1"); assert_eq!(metric.get_gauge().get_value(), 5.0); - assert_eq!(m.gauge.get(), 0.0); + assert_eq!(m.get(), 0.0); } } diff --git a/src/pool/tests.rs b/src/pool/tests.rs index ca24940..fbc9539 100644 --- a/src/pool/tests.rs +++ b/src/pool/tests.rs @@ -302,7 +302,7 @@ fn test_burst_monitoring() { let name = "test_burst_monitoring"; let pool = Builder::new(name) .max_thread_count(1) - .enable_burst_monitoring(1, 200) + .enable_burst_monitoring(1, 400) .build_callback_pool(); let metric = QUEUE_CORE_BURST_THROUGHPUT @@ -315,18 +315,20 @@ fn test_burst_monitoring() { }; // spawn at throughput of 500 tasks/sec - spawn_n(100); - thread::sleep(Duration::from_millis(400)); - spawn_n(100); + let target = 500.0; + spawn_n(200); + thread::sleep(Duration::from_millis(800)); + spawn_n(200); let value = metric.metric().get_gauge().get_value(); - assert!((value - 500.0).abs() < 10.0); + assert!((value - target).abs() / target < 0.1); // spawn at throughput of 1000 tasks/sec - spawn_n(100); - thread::sleep(Duration::from_millis(200)); - spawn_n(100); + let target = 1000.0; + spawn_n(200); + thread::sleep(Duration::from_millis(400)); + spawn_n(200); let value = metric.metric().get_gauge().get_value(); - assert!((value - 1000.0).abs() < 10.0); + assert!((value - target).abs() / target < 0.1); pool.shutdown(); } From 7ae8e403e00762e5150166e99de7aede96f82455 Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 16:22:26 +0000 Subject: [PATCH 4/8] make test_burst_monitoring stable Signed-off-by: zyguan --- src/pool/tests.rs | 8 -------- 1 file changed, 8 deletions(-) diff --git a/src/pool/tests.rs b/src/pool/tests.rs index fbc9539..32cb1a6 100644 --- a/src/pool/tests.rs +++ b/src/pool/tests.rs @@ -322,13 +322,5 @@ fn test_burst_monitoring() { let value = metric.metric().get_gauge().get_value(); assert!((value - target).abs() / target < 0.1); - // spawn at throughput of 1000 tasks/sec - let target = 1000.0; - spawn_n(200); - thread::sleep(Duration::from_millis(400)); - spawn_n(200); - let value = metric.metric().get_gauge().get_value(); - assert!((value - target).abs() / target < 0.1); - pool.shutdown(); } From af34a0caaf670b6381f08a64a35dcf1476ca1ec7 Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 16:30:15 +0000 Subject: [PATCH 5/8] make test_burst_monitoring more stable Signed-off-by: zyguan --- src/pool/tests.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/pool/tests.rs b/src/pool/tests.rs index 32cb1a6..243bba6 100644 --- a/src/pool/tests.rs +++ b/src/pool/tests.rs @@ -302,7 +302,7 @@ fn test_burst_monitoring() { let name = "test_burst_monitoring"; let pool = Builder::new(name) .max_thread_count(1) - .enable_burst_monitoring(1, 400) + .enable_burst_monitoring(1, 1000) .build_callback_pool(); let metric = QUEUE_CORE_BURST_THROUGHPUT @@ -314,11 +314,11 @@ fn test_burst_monitoring() { } }; - // spawn at throughput of 500 tasks/sec - let target = 500.0; - spawn_n(200); - thread::sleep(Duration::from_millis(800)); - spawn_n(200); + // spawn at throughput of 1000 tasks/sec + let target = 1000.0; + spawn_n(500); + thread::sleep(Duration::from_secs(1)); + spawn_n(500); let value = metric.metric().get_gauge().get_value(); assert!((value - target).abs() / target < 0.1); From 5b47610f3a5b6e14d203c57a5e082ea0fa63a0f9 Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 16:38:24 +0000 Subject: [PATCH 6/8] make clippy happy Signed-off-by: zyguan --- src/metrics.rs | 6 ++++++ src/queue.rs | 8 ++------ src/task/future.rs | 4 ++-- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/src/metrics.rs b/src/metrics.rs index d2cb0d7..9a03bcb 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -198,6 +198,12 @@ impl MaxGaugeVecBuilder { } } +impl Default for MaxGaugeVecBuilder { + fn default() -> Self { + Self::new() + } +} + impl MetricVecBuilder for MaxGaugeVecBuilder { type M = MaxGauge; type P = Opts; diff --git a/src/queue.rs b/src/queue.rs index c8d7fb4..f5e5448 100644 --- a/src/queue.rs +++ b/src/queue.rs @@ -128,8 +128,10 @@ impl LocalQueue { } /// Supported available queues. +#[derive(Default)] pub enum QueueType { /// A single level work stealing queue. + #[default] SingleLevel, /// A multilevel feedback queue. /// @@ -139,12 +141,6 @@ pub enum QueueType { Priority(priority::Builder), } -impl Default for QueueType { - fn default() -> QueueType { - QueueType::SingleLevel - } -} - impl From for QueueType { fn from(b: multilevel::Builder) -> QueueType { QueueType::Multilevel(b) diff --git a/src/task/future.rs b/src/task/future.rs index 9b9b846..61b774d 100644 --- a/src/task/future.rs +++ b/src/task/future.rs @@ -299,7 +299,7 @@ unsafe fn clone_task(task: *const ()) -> TaskCell { thread_local! { /// Local queue reference that is set before polling and unset after polled. - static LOCAL: Cell<*mut Local> = Cell::new(std::ptr::null_mut()); + static LOCAL: Cell<*mut Local> = const { Cell::new(std::ptr::null_mut()) }; } unsafe fn wake_task(task: Cow<'_, TaskCell>, reschedule: bool) { @@ -370,7 +370,7 @@ impl Runner { } thread_local! { - static NEED_RESCHEDULE: Cell = Cell::new(false); + static NEED_RESCHEDULE: Cell = const { Cell::new(false) }; } impl crate::pool::Runner for Runner { From 30b88f1eac1135e976c159d702b5624c64aeaf28 Mon Sep 17 00:00:00 2001 From: zyguan Date: Tue, 10 Feb 2026 16:51:38 +0000 Subject: [PATCH 7/8] simplify test_burst_monitoring Signed-off-by: zyguan --- src/pool/tests.rs | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/src/pool/tests.rs b/src/pool/tests.rs index 243bba6..bb3497d 100644 --- a/src/pool/tests.rs +++ b/src/pool/tests.rs @@ -302,25 +302,17 @@ fn test_burst_monitoring() { let name = "test_burst_monitoring"; let pool = Builder::new(name) .max_thread_count(1) - .enable_burst_monitoring(1, 1000) + .enable_burst_monitoring(1, 100) .build_callback_pool(); let metric = QUEUE_CORE_BURST_THROUGHPUT .get_metric_with_label_values(&[name]) .unwrap(); - let spawn_n = |n| { - for _ in 0..n { - pool.spawn(|_: &mut Handle<'_>| {}); - } - }; - // spawn at throughput of 1000 tasks/sec - let target = 1000.0; - spawn_n(500); - thread::sleep(Duration::from_secs(1)); - spawn_n(500); + for _ in 0..100 { + pool.spawn(|_: &mut Handle<'_>| {}); + } let value = metric.metric().get_gauge().get_value(); - assert!((value - target).abs() / target < 0.1); - + assert!(value > 100.0); // the above loop should be executed within 1s pool.shutdown(); } From d01ed0d1141383ceb5131fb2c2dfa7eff2cd04a8 Mon Sep 17 00:00:00 2001 From: zyguan Date: Wed, 11 Feb 2026 02:01:18 +0000 Subject: [PATCH 8/8] address some comments from copilot Signed-off-by: zyguan --- src/metrics.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/metrics.rs b/src/metrics.rs index 9a03bcb..1eb50a2 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -112,7 +112,8 @@ fn new_histogram_opts(name: &str, help: &str, buckets: Vec) -> HistogramOpt opts } -/// A gauge that tracks the maximum value since the last scrape. +/// A gauge that tracks the maximum value since the last scrape. Note that it aims to track positive values, negative +/// values are ignored since `max_val` is reset to 0 after each scrape. #[derive(Clone, Debug)] pub struct MaxGauge { gauge: Gauge, @@ -123,10 +124,10 @@ impl MaxGauge { /// Wraps a `Gauge` to create a `MaxGauge`. The `Gauge` should not be used directly after being wrapped, otherwise /// the maximum tracking will be broken. pub fn wrap(gauge: Gauge) -> Self { - let val = gauge.get().to_bits(); + let val = gauge.get().max(0f64); Self { gauge, - max_val: Arc::new(AtomicU64::new(val)), + max_val: Arc::new(AtomicU64::new(val.to_bits())), } } @@ -146,10 +147,7 @@ impl MaxGauge { Ordering::Relaxed, Ordering::Relaxed, ) { - Ok(_) => { - self.gauge.set(v); - break; - } + Ok(_) => break, Err(actual) => current = actual, } }