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}"))?;