External/bichon
1
0
Fork 0
mirror of https://github.com/rustmailer/bichon.git synced 2026-08-03 07:48:34 +02:00
Code Issues Packages Projects Releases Wiki Activity

perf: reduce tokio worker thread blocking to improve responsiveness on low-core machines

Browse Source 
- Switch memdb durability from Full to Batch(100) with 10s flush worker
  - Offload BlobManager fjall writes to spawn_blocking
  - Wrap Tantivy commit operations in block_in_place
  - Flush memdb WAL on graceful shutdown
This commit is contained in:
rustmailer
2026-05-24 14:34:16 +08:00
parent 0be2670600
commit 0792bb546d
5 changed files with 49 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/core/src/database/manager.rs
+13 -1
View File
@@ -32,12 +32,16 @@ impl DatabaseManager {
let db_path = &DATA_DIR_MANAGER.memdb_dir;
std::fs::create_dir_all(db_path).expect("Failed to create memdb data directory");
let db = MemDb::open_with(db_path, Durability::Full)
let db = MemDb::open_with(db_path, Durability::Batch { max_ops: 100 })
.expect("Failed to open memdb database");
// Start periodic snapshot worker (every 5 minutes)
db.start_snapshot_worker(Duration::from_secs(300));
// Start periodic flush worker (every 10 seconds) so buffered writes
// are flushed regularly and not only at the batch threshold.
db.start_flush_worker(Duration::from_secs(10));
DatabaseManager { db }
}
@@ -45,4 +49,12 @@ impl DatabaseManager {
pub fn db(&self) -> &MemDb {
&self.db
}
/// Flush any buffered WAL entries to disk. Must be called before shutdown
/// to avoid losing writes that haven't hit the batch threshold yet.
pub fn flush(&self) {
if let Err(e) = self.db.flush() {
eprintln!("[memdb] flush error on shutdown: {e}");
}
}
}
crates/core/src/store/blob.rs
+26 -8
View File
@@ -149,9 +149,18 @@ impl BlobManager {
res = receiver.recv() => {
match res {
Some(eml) => {
Self::process_detached_email(eml, &email_ks, &attach_ks);
let mut batch = vec![eml];
while let Ok(next_eml) = receiver.try_recv() {
Self::process_detached_email(next_eml, &email_ks, &attach_ks);
batch.push(next_eml);
}
let email_ks = email_ks.clone();
let attach_ks = attach_ks.clone();
if let Err(e) = tokio::task::spawn_blocking(move || {
for eml in batch {
Self::process_detached_email(eml, &email_ks, &attach_ks);
}
}).await {
tracing::error!("BlobManager: spawn_blocking join error: {:#?}", e);
}
}
None => {
@@ -162,16 +171,25 @@ impl BlobManager {
}
_ = shutdown.recv() => {
receiver.close();
let remaining = receiver.len();
let mut remaining = Vec::new();
while let Some(eml) = receiver.recv().await {
remaining.push(eml);
}
tracing::info!(
"BlobManager: Shutdown signal received. Processing {} remaining tasks...",
remaining
remaining.len()
);
while let Some(eml) = receiver.recv().await {
Self::process_detached_email(eml, &email_ks, &attach_ks);
if !remaining.is_empty() {
let email_ks = email_ks.clone();
let attach_ks = attach_ks.clone();
if let Err(e) = tokio::task::spawn_blocking(move || {
for eml in remaining {
Self::process_detached_email(eml, &email_ks, &attach_ks);
}
}).await {
tracing::error!("BlobManager: shutdown spawn_blocking join error: {:#?}", e);
}
}
tracing::info!("BlobManager: All remaining tasks processed. Closing Fjall.");
break;
}
crates/core/src/store/tantivy/attachment.rs
+4 -4
View File
@@ -154,7 +154,7 @@ impl IndexManager {
"Tantivy: Reached threshold ({} docs), committing...",
pending_count
);
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
pending_count = 0;
commit_interval.reset();
}
@@ -163,7 +163,7 @@ impl IndexManager {
tracing::info!("Tantivy: Receiver closed. Finalizing...");
if pending_count > 0 {
let mut writer = writer.lock().await;
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
}
break;
},
@@ -172,7 +172,7 @@ impl IndexManager {
_ = commit_interval.tick() => {
if pending_count > 0 {
let mut writer = writer.lock().await;
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
pending_count = 0;
tracing::debug!("Tantivy: Periodic commit finished.");
}
@@ -181,7 +181,7 @@ impl IndexManager {
tracing::info!("Tantivy: Shutdown signal received. Performing final commit...");
if pending_count > 0 {
let mut writer = writer.lock().await;
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
}
tracing::info!("Tantivy: Shutdown cleanup complete.");
break;
crates/core/src/store/tantivy/envelope.rs
+4 -4
View File
@@ -169,7 +169,7 @@ impl IndexManager {
"Tantivy: Reached threshold ({} docs), committing...",
pending_count
);
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
pending_count = 0;
commit_interval.reset();
}
@@ -178,7 +178,7 @@ impl IndexManager {
tracing::info!("Tantivy: Receiver closed. Finalizing...");
if pending_count > 0 {
let mut writer = writer.lock().await;
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
}
break;
},
@@ -187,7 +187,7 @@ impl IndexManager {
_ = commit_interval.tick() => {
if pending_count > 0 {
let mut writer = writer.lock().await;
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
pending_count = 0;
tracing::debug!("Tantivy: Periodic commit finished.");
}
@@ -196,7 +196,7 @@ impl IndexManager {
tracing::info!("Tantivy: Shutdown signal received. Performing final commit...");
if pending_count > 0 {
let mut writer = writer.lock().await;
fatal_commit(&mut writer);
tokio::task::block_in_place(|| fatal_commit(&mut writer));
}
tracing::info!("Tantivy: Shutdown cleanup complete.");
break;
crates/server/src/lib.rs
+2
View File
@@ -28,6 +28,7 @@ use bichon_core::{
cache::imap::task::SYNC_TASKS,
common::{rustls::BichonTls, signal::SignalManager},
context::{executors::BichonContext, Initialize},
database::manager::DB_MANAGER,
error::{code::ErrorCode, BichonResult},
logger,
migrate::check_data_status,
@@ -116,6 +117,7 @@ pub async fn run() -> BichonResult<()> {
ENVELOPE_MANAGER.shutdown().await;
ATTACHMENT_MANAGER.shutdown().await;
BLOB_MANAGER.shutdown().await;
DB_MANAGER.flush();
info!("Bichon server stopped.");
Ok(())
}