From 0792bb546d3cf648ae1e08dc09233eb6f9e0e77b Mon Sep 17 00:00:00 2001 From: rustmailer Date: Sun, 24 May 2026 14:34:16 +0800 Subject: [PATCH] perf: reduce tokio worker thread blocking to improve responsiveness on low-core machines - 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 --- crates/core/src/database/manager.rs | 14 ++++++++- crates/core/src/store/blob.rs | 34 ++++++++++++++++----- crates/core/src/store/tantivy/attachment.rs | 8 ++--- crates/core/src/store/tantivy/envelope.rs | 8 ++--- crates/server/src/lib.rs | 2 ++ 5 files changed, 49 insertions(+), 17 deletions(-) diff --git a/crates/core/src/database/manager.rs b/crates/core/src/database/manager.rs index b6b00bd..afd0b98 100644 --- a/crates/core/src/database/manager.rs +++ b/crates/core/src/database/manager.rs @@ -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}"); + } + } } diff --git a/crates/core/src/store/blob.rs b/crates/core/src/store/blob.rs index ede48a8..34245de 100644 --- a/crates/core/src/store/blob.rs +++ b/crates/core/src/store/blob.rs @@ -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; } diff --git a/crates/core/src/store/tantivy/attachment.rs b/crates/core/src/store/tantivy/attachment.rs index 108ec98..e72c93b 100644 --- a/crates/core/src/store/tantivy/attachment.rs +++ b/crates/core/src/store/tantivy/attachment.rs @@ -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; diff --git a/crates/core/src/store/tantivy/envelope.rs b/crates/core/src/store/tantivy/envelope.rs index b73f967..f53b699 100644 --- a/crates/core/src/store/tantivy/envelope.rs +++ b/crates/core/src/store/tantivy/envelope.rs @@ -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; diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index b450714..f89e6de 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -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(()) }