From 2e4dbd8c5d9da8624a17205ae64983deae6cabfc Mon Sep 17 00:00:00 2001 From: Wez Furlong Date: Mon, 24 Feb 2025 13:53:03 -0700 Subject: [PATCH] queue+ready_queue: refactor: remove StdMutex type alias This came from an iteration when we had both tokio and std mutexes in this file. Since then we've eliminated the former and traded out for parking_lots mutexes. As a result, the StdMutex label for this is quite misleading. Remove the type alias and just use the name of the actual type. This should have zero functional impact as it is just removing a type alias. --- crates/kumod/src/queue.rs | 42 ++++++++++++++++----------------- crates/kumod/src/ready_queue.rs | 30 +++++++++++------------ 2 files changed, 36 insertions(+), 36 deletions(-) diff --git a/crates/kumod/src/queue.rs b/crates/kumod/src/queue.rs index 590f4788..2b4a94fe 100644 --- a/crates/kumod/src/queue.rs +++ b/crates/kumod/src/queue.rs @@ -31,7 +31,7 @@ use message::timeq::TriTimeQ; use message::Message; use mlua::prelude::*; use mlua::UserDataMethods; -use parking_lot::FairMutex as StdMutex; +use parking_lot::FairMutex; use prometheus::{Histogram, IntCounter, IntGauge}; use rfc5321::{EnhancedStatusCode, Response}; use serde::{Deserialize, Serialize}; @@ -46,8 +46,8 @@ use timeq::{PopResult, TimeQ, TimerEntryWithDelay, TimerError}; use tokio::sync::Notify; use tracing::instrument; -static MANAGER: LazyLock> = - LazyLock::new(|| StdMutex::new(QueueManager::new())); +static MANAGER: LazyLock> = + LazyLock::new(|| FairMutex::new(QueueManager::new())); static SCHEDULED_QUEUE_COUNT: LazyLock = LazyLock::new(|| { prometheus::register_int_gauge!( "scheduled_queue_count", @@ -90,11 +90,11 @@ static REBIND_MESSAGE_SIG: LazyLock> = LazyLock::new(|| CallbackSignature::new_with_multiple("requeue_message")); -pub static SINGLETON_WHEEL: LazyLock>>> = - LazyLock::new(|| Arc::new(StdMutex::new(TimeQ::new()))); +pub static SINGLETON_WHEEL: LazyLock>>> = + LazyLock::new(|| Arc::new(FairMutex::new(TimeQ::new()))); -pub static SINGLETON_WHEEL_V2: LazyLock>> = - LazyLock::new(|| Arc::new(StdMutex::new(TriTimeQ::new(Duration::from_secs(3))))); +pub static SINGLETON_WHEEL_V2: LazyLock>> = + LazyLock::new(|| Arc::new(FairMutex::new(TriTimeQ::new(Duration::from_secs(3))))); static TOTAL_DELAY_GAUGE: LazyLock = LazyLock::new(|| { prometheus::register_int_gauge!( @@ -811,22 +811,22 @@ enum QueueInsertResult { } enum QueueStructure { - TimerWheel(StdMutex>), + TimerWheel(FairMutex>), SkipList(SkipSet), - SingletonTimerWheel(StdMutex>), - SingletonTimerWheelV2(StdMutex>), + SingletonTimerWheel(FairMutex>), + SingletonTimerWheelV2(FairMutex>), } impl QueueStructure { fn new(strategy: QueueStrategy) -> Self { match strategy { - QueueStrategy::TimerWheel => Self::TimerWheel(StdMutex::new(TimeQ::new())), + QueueStrategy::TimerWheel => Self::TimerWheel(FairMutex::new(TimeQ::new())), QueueStrategy::SkipList => Self::SkipList(SkipSet::new()), QueueStrategy::SingletonTimerWheel => { - Self::SingletonTimerWheel(StdMutex::new(HashSet::new())) + Self::SingletonTimerWheel(FairMutex::new(HashSet::new())) } QueueStrategy::SingletonTimerWheelV2 => { - Self::SingletonTimerWheelV2(StdMutex::new(HashSet::new())) + Self::SingletonTimerWheelV2(FairMutex::new(HashSet::new())) } } } @@ -1034,14 +1034,14 @@ pub struct Queue { name: Arc, queue: QueueStructure, notify_maintainer: Arc, - last_change: StdMutex, + last_change: FairMutex, queue_config: ConfigHandle, metrics: OnceLock, activity: Activity, rr: ArcSwap, - next_config_refresh: StdMutex, + next_config_refresh: FairMutex, warned_strategy_change: AtomicBool, - config_epoch: StdMutex, + config_epoch: FairMutex, site_name: String, } @@ -1094,7 +1094,7 @@ impl Queue { let activity = Activity::get(format!("Queue {name}"))?; let strategy = queue_config.strategy; - let next_config_refresh = StdMutex::new(Instant::now() + queue_config.refresh_interval); + let next_config_refresh = FairMutex::new(Instant::now() + queue_config.refresh_interval); let queue_config = ConfigHandle::new(queue_config); let site_name = match ReadyQueueManager::compute_queue_name( @@ -1134,7 +1134,7 @@ impl Queue { let handle = Arc::new(Queue { name: name.clone(), queue: QueueStructure::new(strategy), - last_change: StdMutex::new(Instant::now()), + last_change: FairMutex::new(Instant::now()), queue_config, notify_maintainer: Arc::new(Notify::new()), metrics: OnceLock::new(), @@ -1142,7 +1142,7 @@ impl Queue { rr, next_config_refresh, warned_strategy_change: AtomicBool::new(false), - config_epoch: StdMutex::new(epoch), + config_epoch: FairMutex::new(epoch), site_name, }); @@ -2381,7 +2381,7 @@ impl Queue { // Verify that the message is still in the queue match &queue.queue { QueueStructure::SingletonTimerWheelV2(q) => { - fn remove(q: &StdMutex>, msg: &Message) -> bool { + fn remove(q: &FairMutex>, msg: &Message) -> bool { q.lock().remove(msg) } @@ -2474,7 +2474,7 @@ impl Queue { // Verify that the message is still in the queue match &queue.queue { QueueStructure::SingletonTimerWheel(q) => { - fn remove(q: &StdMutex>, msg: &Message) -> bool { + fn remove(q: &FairMutex>, msg: &Message) -> bool { q.lock().remove(msg) } diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index 0375349a..9385b31a 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/crates/kumod/src/ready_queue.rs @@ -30,7 +30,7 @@ use kumo_server_memory::{ use kumo_server_runtime::{get_named_runtime, spawn, Runtime}; use message::message::{MessageList, QueueNameComponents}; use message::Message; -use parking_lot::FairMutex as StdMutex; +use parking_lot::FairMutex; use prometheus::Histogram; use rfc5321::{EnhancedStatusCode, Response}; use serde::Serialize; @@ -47,8 +47,8 @@ use tokio::task::JoinHandle; use tracing::instrument; use uuid::Uuid; -static MANAGER: LazyLock> = - LazyLock::new(|| StdMutex::new(ReadyQueueManager::new())); +static MANAGER: LazyLock> = + LazyLock::new(|| FairMutex::new(ReadyQueueManager::new())); static READYQ_RUNTIME: LazyLock = LazyLock::new(|| Runtime::new("readyq", |cpus| cpus / 2, &READYQ_THREADS).unwrap()); pub static GET_EGRESS_PATH_CONFIG_SIG: LazyLock< @@ -98,7 +98,7 @@ where } pub struct Fifo { - list: StdMutex, + list: FairMutex, count: ReadyCountBundle, capacity: AtomicUsize, } @@ -107,7 +107,7 @@ impl Fifo { pub fn new(capacity: usize, count: ReadyCountBundle) -> Self { Self { count, - list: StdMutex::new(MessageList::new()), + list: FairMutex::new(MessageList::new()), capacity: AtomicUsize::new(capacity), } } @@ -398,7 +398,7 @@ impl ReadyQueueManager { mx, notify_dispatcher, notify_maintainer, - connections: StdMutex::new(vec![]), + connections: FairMutex::new(vec![]), path_config: ConfigHandle::new(path_config), queue_config: queue_config.clone(), egress_source, @@ -406,9 +406,9 @@ impl ReadyQueueManager { activity, consecutive_connection_failures: Arc::new(AtomicUsize::new(0)), egress_pool: egress_pool.to_string(), - next_config_refresh: StdMutex::new(next_config_refresh), - config_epoch: StdMutex::new(config_epoch), - states: Arc::new(StdMutex::new(ReadyQueueStates::default())), + next_config_refresh: FairMutex::new(next_config_refresh), + config_epoch: FairMutex::new(config_epoch), + states: Arc::new(FairMutex::new(ReadyQueueStates::default())), }) }); Ok(handle.clone()) @@ -599,7 +599,7 @@ pub struct ReadyQueue { mx: Option>, notify_maintainer: Arc, notify_dispatcher: Arc, - connections: StdMutex>>, + connections: FairMutex>>, metrics: DeliveryMetrics, activity: Activity, consecutive_connection_failures: Arc, @@ -607,9 +607,9 @@ pub struct ReadyQueue { queue_config: ConfigHandle, egress_pool: String, egress_source: EgressSource, - next_config_refresh: StdMutex, - config_epoch: StdMutex, - pub states: Arc>, + next_config_refresh: FairMutex, + config_epoch: FairMutex, + pub states: Arc>, } impl ReadyQueue { @@ -1079,7 +1079,7 @@ pub struct Dispatcher { pub session_id: Uuid, leases: Vec, batch_started: Option, - pub states: Arc>, + pub states: Arc>, } impl Drop for Dispatcher { @@ -1150,7 +1150,7 @@ impl Dispatcher { egress_source: EgressSource, egress_pool: String, leases: Vec, - states: Arc>, + states: Arc>, ) -> anyhow::Result<()> { let activity = Activity::get(format!("ready_queue Dispatcher {name}"))?;