filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Global threadId caching

Browse Source
This commit is contained in:
mdecimus
2024-03-04 18:51:24 +01:00
parent 31bc716a5f
commit ff279b3a39
16 changed files with 208 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
crates/imap/src/core/mod.rs
View File

@@ -24,14 +24,11 @@
use std::{
collections::BTreeMap,
net::IpAddr,
sync::{
atomic::{AtomicU32, AtomicU64},
Arc,
},
sync::{atomic::AtomicU32, Arc},
time::Duration,
};
use ahash::{AHashMap, AHashSet};
use ahash::AHashMap;
use dashmap::DashMap;
use imap_proto::{
protocol::{list::Attribute, ProtocolVersion},
@@ -42,7 +39,7 @@ use jmap::{
auth::{rate_limit::ConcurrencyLimiters, AccessToken},
JMAP,
};
use store::{roaring::RoaringBitmap, write::now};
use store::roaring::RoaringBitmap;
use tokio::{
io::{ReadHalf, WriteHalf},
sync::watch,
@@ -50,6 +47,7 @@ use tokio::{
use utils::{
config::Rate,
listener::{limiter::InFlight, ServerInstance, SessionStream},
CachedItem,
};
pub mod client;
@@ -88,17 +86,8 @@ pub struct IMAP {
pub rate_concurrent: u64,
pub cache_account: DashMap<AccountId, CachedItem<Account>>,
pub cache_account_expiry: u64,
pub cache_mailbox: DashMap<MailboxId, CachedItem<MailboxState>>,
pub cache_mailbox_expiry: u64,
pub cache_threads: DashMap<u32, CachedItem<Threads>>,
pub cache_threads_expiry: u64,
}
#[derive(Clone)]
pub struct CachedItem<T> {
last_access: Arc<AtomicU64>,
item: Arc<tokio::sync::Mutex<T>>,
pub cache_expiry: u64,
}
pub struct Session<T: SessionStream> {
@@ -198,12 +187,6 @@ pub struct MailboxSync {
pub deleted: Vec<String>,
}
#[derive(Debug, Default)]
pub struct Threads {
pub threads: AHashMap<u32, u32>,
pub modseq: Option<u64>,
}
pub enum SavedSearch {
InFlight {
rx: watch::Receiver<Arc<Vec<ImapId>>>,
@@ -284,23 +267,3 @@ impl<T: SessionStream> SessionData<T> {
}
}
}
impl<T> CachedItem<T> {
pub fn new(item: T) -> Self {
Self {
last_access: Arc::new(AtomicU64::new(now())),
item: Arc::new(tokio::sync::Mutex::new(item)),
}
}
pub async fn get(&self) -> tokio::sync::MutexGuard<'_, T> {
let lock = self.item.lock().await;
self.last_access
.store(now(), std::sync::atomic::Ordering::Relaxed);
lock
}
pub fn last_access(&self) -> u64 {
self.last_access.load(std::sync::atomic::Ordering::Relaxed)
}
}

36
crates/imap/src/lib.rs
View File

@@ -21,7 +21,6 @@
* for more details.
*/
use core::mailbox;
use std::{collections::hash_map::RandomState, sync::Arc, time::Duration};
use crate::core::IMAP;
@@ -47,8 +46,6 @@ impl IMAP {
.unwrap_or(32)
.next_power_of_two() as usize;
let todo = "document imap.cache.rate-limit.size and imap.cache.mailbox.size";
Ok(Arc::new(IMAP {
max_request_size: config.property_or_static("imap.request.max-size", "52428800")?,
max_auth_failures: config.property_or_static("imap.auth.max-failures", "3")?,
@@ -70,9 +67,7 @@ impl IMAP {
})
.into_bytes(),
rate_limiter: DashMap::with_capacity_and_hasher_and_shard_amount(
config
.property("imap.cache.rate-limit.size")?
.unwrap_or(2048),
config.property("cache.rate-limit.size")?.unwrap_or(2048),
RandomState::default(),
shard_amount,
),
@@ -81,42 +76,27 @@ impl IMAP {
allow_plain_auth: config.property_or_static("imap.auth.allow-plain-text", "false")?,
enable_uidplus: config.property_or_static("imap.protocol.uidplus", "false")?,
cache_account: DashMap::with_capacity_and_hasher_and_shard_amount(
config.property("imap.cache.account.size")?.unwrap_or(2048),
config.property("cache.messages.size")?.unwrap_or(2048),
RandomState::default(),
shard_amount,
),
cache_mailbox: DashMap::with_capacity_and_hasher_and_shard_amount(
config.property("imap.cache.mailbox.size")?.unwrap_or(2048),
config.property("cache.messages.size")?.unwrap_or(2048),
RandomState::default(),
shard_amount,
),
cache_threads: DashMap::with_capacity_and_hasher_and_shard_amount(
config.property("imap.cache.thread.size")?.unwrap_or(2048),
RandomState::default(),
shard_amount,
),
cache_account_expiry: config
.property_or_static::<Duration>("imap.cache.account.expiry", "1h")?
.as_secs(),
cache_mailbox_expiry: config
.property_or_static::<Duration>("imap.cache.mailbox.expiry", "1h")?
.as_secs(),
cache_threads_expiry: config
.property_or_static::<Duration>("imap.cache.thread.expiry", "1h")?
cache_expiry: config
.property_or_static::<Duration>("cache.messages.ttl", "1h")?
.as_secs(),
}))
}
pub fn purge(&self) {
let account_expiry = now() - self.cache_account_expiry;
let mailbox_expiry = now() - self.cache_mailbox_expiry;
let thread_expiry = now() - self.cache_threads_expiry;
let expiry = now() - self.cache_expiry;
self.cache_account
.retain(|_, item| item.last_access() > account_expiry);
.retain(|_, item| item.last_access() > expiry);
self.cache_mailbox
.retain(|_, item| item.last_access() > mailbox_expiry);
self.cache_threads
.retain(|_, item| item.last_access() > thread_expiry);
.retain(|_, item| item.last_access() > expiry);
}
}

4
crates/imap/src/op/select.rs
View File

@@ -35,9 +35,9 @@ use imap_proto::{
};
use jmap_proto::types::id::Id;
use utils::listener::SessionStream;
use utils::{listener::SessionStream, CachedItem};
use crate::core::{CachedItem, MailboxState, SavedSearch, SelectedMailbox, Session, State};
use crate::core::{MailboxState, SavedSearch, SelectedMailbox, Session, State};
use super::ToModSeq;

67
crates/imap/src/op/thread.rs
View File

@@ -33,11 +33,9 @@ use imap_proto::{
Command, StatusResponse,
};
use jmap_proto::types::{collection::Collection, property::Property};
use store::{write::ValueClass, ValueKey};
use utils::listener::SessionStream;
use crate::core::{CachedItem, SelectedMailbox, Session, SessionData, Threads};
use crate::core::{SelectedMailbox, Session, SessionData};
impl<T: SessionStream> Session<T> {
pub async fn handle_thread(
@@ -87,67 +85,20 @@ impl<T: SessionStream> SessionData<T> {
});
}
// Obtain current state
let modseq = self
// Lock the cache
let thread_ids = self
.jmap
.store
.get_last_change_id(mailbox.id.account_id, Collection::Thread)
.get_cached_thread_ids(mailbox.id.account_id, result_set.results.iter())
.await
.map_err(|err| {
tracing::error!(event = "error",
context = "store",
account_id = mailbox.id.account_id,
collection = ?Collection::Thread,
error = ?err,
"Failed to obtain state");
tracing::error!(
event = "error",
context = "thread_query",
error = ?err,
"Failed to obtain threadId.");
StatusResponse::database_failure()
})?;
// Lock the cache
let thread_cache_ = self
.imap
.cache_threads
.entry(mailbox.id.account_id)
.or_insert_with(|| CachedItem::new(Threads::default()));
let mut thread_cache = thread_cache_.get().await;
// Invalidate cache if the modseq has changed
if thread_cache.modseq != modseq {
thread_cache.threads.clear();
}
// Obtain threadIds for matching messages
let mut thread_ids = Vec::with_capacity(result_set.results.len() as usize);
for document_id in &result_set.results {
if let Some(thread_id) = thread_cache.threads.get(&document_id) {
thread_ids.push((*thread_id).into());
} else if let Some(thread_id) = self
.jmap
.store
.get_value::<u32>(ValueKey {
account_id: mailbox.id.account_id,
collection: Collection::Email.into(),
document_id,
class: ValueClass::Property(Property::ThreadId.into()),
})
.await
.map_err(|err| {
tracing::error!(
event = "error",
context = "thread_query",
error = ?err,
"Failed to obtain threadId.");
StatusResponse::database_failure()
})?
{
thread_ids.push(thread_id.into());
thread_cache.threads.insert(document_id, thread_id);
} else {
thread_ids.push(None);
}
}
thread_cache.modseq = modseq;
// Group messages by thread
let mut threads: AHashMap<u32, Vec<u32>> = AHashMap::new();
let state = mailbox.state.lock();