mirror/kumomta
1
0
Fork 0
mirror of https://github.com/mailscope/kumomta.git synced 2026-10-27 00:36:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

queue+ready_queue: refactor: remove StdMutex type alias

Browse Source 
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.
This commit is contained in:
Wez Furlong
2025-02-24 14:21:40 -07:00
parent f915a7fbfa
commit 2e4dbd8c5d
2 changed files with 36 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/kumod/src/queue.rs
+21 -21
View File
@@ -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<StdMutex<QueueManager>> =
LazyLock::new(|| StdMutex::new(QueueManager::new()));
static MANAGER: LazyLock<FairMutex<QueueManager>> =
LazyLock::new(|| FairMutex::new(QueueManager::new()));
static SCHEDULED_QUEUE_COUNT: LazyLock<IntGauge> = LazyLock::new(|| {
prometheus::register_int_gauge!(
"scheduled_queue_count",
@@ -90,11 +90,11 @@ static REBIND_MESSAGE_SIG: LazyLock<CallbackSignature<(Message, HashMap<String,
pub static REQUEUE_MESSAGE_SIG: LazyLock<CallbackSignature<(Message, String), ()>> =
LazyLock::new(|| CallbackSignature::new_with_multiple("requeue_message"));
pub static SINGLETON_WHEEL: LazyLock<Arc<StdMutex<TimeQ<WeakMessage>>>> =
LazyLock::new(|| Arc::new(StdMutex::new(TimeQ::new())));
pub static SINGLETON_WHEEL: LazyLock<Arc<FairMutex<TimeQ<WeakMessage>>>> =
LazyLock::new(|| Arc::new(FairMutex::new(TimeQ::new())));
pub static SINGLETON_WHEEL_V2: LazyLock<Arc<StdMutex<TriTimeQ>>> =
LazyLock::new(|| Arc::new(StdMutex::new(TriTimeQ::new(Duration::from_secs(3)))));
pub static SINGLETON_WHEEL_V2: LazyLock<Arc<FairMutex<TriTimeQ>>> =
LazyLock::new(|| Arc::new(FairMutex::new(TriTimeQ::new(Duration::from_secs(3)))));
static TOTAL_DELAY_GAUGE: LazyLock<IntGauge> = LazyLock::new(|| {
prometheus::register_int_gauge!(
@@ -811,22 +811,22 @@ enum QueueInsertResult {
}
enum QueueStructure {
TimerWheel(StdMutex<TimeQ<Message>>),
TimerWheel(FairMutex<TimeQ<Message>>),
SkipList(SkipSet<DelayedEntry>),
SingletonTimerWheel(StdMutex<HashSet<Message>>),
SingletonTimerWheelV2(StdMutex<HashSet<Message>>),
SingletonTimerWheel(FairMutex<HashSet<Message>>),
SingletonTimerWheelV2(FairMutex<HashSet<Message>>),
}
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<String>,
queue: QueueStructure,
notify_maintainer: Arc<Notify>,
last_change: StdMutex<Instant>,
last_change: FairMutex<Instant>,
queue_config: ConfigHandle<QueueConfig>,
metrics: OnceLock<ScheduledMetrics>,
activity: Activity,
rr: ArcSwap<EgressPoolRoundRobin>,
next_config_refresh: StdMutex<Instant>,
next_config_refresh: FairMutex<Instant>,
warned_strategy_change: AtomicBool,
config_epoch: StdMutex<ConfigEpoch>,
config_epoch: FairMutex<ConfigEpoch>,
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<HashSet<Message>>, msg: &Message) -> bool {
fn remove(q: &FairMutex<HashSet<Message>>, 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<HashSet<Message>>, msg: &Message) -> bool {
fn remove(q: &FairMutex<HashSet<Message>>, msg: &Message) -> bool {
q.lock().remove(msg)
}
crates/kumod/src/ready_queue.rs
+15 -15
View File
@@ -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<StdMutex<ReadyQueueManager>> =
LazyLock::new(|| StdMutex::new(ReadyQueueManager::new()));
static MANAGER: LazyLock<FairMutex<ReadyQueueManager>> =
LazyLock::new(|| FairMutex::new(ReadyQueueManager::new()));
static READYQ_RUNTIME: LazyLock<Runtime> =
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<MessageList>,
list: FairMutex<MessageList>,
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<Arc<MailExchanger>>,
notify_maintainer: Arc<Notify>,
notify_dispatcher: Arc<Notify>,
connections: StdMutex<Vec<JoinHandle<()>>>,
connections: FairMutex<Vec<JoinHandle<()>>>,
metrics: DeliveryMetrics,
activity: Activity,
consecutive_connection_failures: Arc<AtomicUsize>,
@@ -607,9 +607,9 @@ pub struct ReadyQueue {
queue_config: ConfigHandle<QueueConfig>,
egress_pool: String,
egress_source: EgressSource,
next_config_refresh: StdMutex<Instant>,
config_epoch: StdMutex<ConfigEpoch>,
pub states: Arc<StdMutex<ReadyQueueStates>>,
next_config_refresh: FairMutex<Instant>,
config_epoch: FairMutex<ConfigEpoch>,
pub states: Arc<FairMutex<ReadyQueueStates>>,
}
impl ReadyQueue {
@@ -1079,7 +1079,7 @@ pub struct Dispatcher {
pub session_id: Uuid,
leases: Vec<LimitLease>,
batch_started: Option<tokio::time::Instant>,
pub states: Arc<StdMutex<ReadyQueueStates>>,
pub states: Arc<FairMutex<ReadyQueueStates>>,
}
impl Drop for Dispatcher {
@@ -1150,7 +1150,7 @@ impl Dispatcher {
egress_source: EgressSource,
egress_pool: String,
leases: Vec<LimitLease>,
states: Arc<StdMutex<ReadyQueueStates>>,
states: Arc<FairMutex<ReadyQueueStates>>,
) -> anyhow::Result<()> {
let activity = Activity::get(format!("ready_queue Dispatcher {name}"))?;