perf: fix store mem leak, batch DB flushes, and remove double-clones
This commit is contained in:
1 parent
2c5953e4f5
commit
d5a05f8416
2 files changed
+24
-15
No files matched your search
+20
-11
@@ -5,21 +5,32 @@ use std::sync::{Arc, RwLock};
|
||||
pub const STORE_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("store");
|
||||
|
||||
pub struct Store<T> {
|
||||
pub cache: RwLock<T>,
|
||||
tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
|
||||
pub cache: Arc<RwLock<T>>,
|
||||
tx: tokio::sync::mpsc::Sender<()>,
|
||||
}
|
||||
|
||||
impl<T: DeserializeOwned + Default + Serialize + Clone + Send + 'static> Store<T> {
|
||||
impl<T: DeserializeOwned + Default + Serialize + Clone + Send + Sync + 'static> Store<T> {
|
||||
pub fn new(key: &str, db: Arc<Database>) -> Self {
|
||||
let initial_data = Self::load_from_db(key, &db);
|
||||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
|
||||
let cache = Arc::new(RwLock::new(initial_data));
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(1);
|
||||
|
||||
let db_clone = db.clone();
|
||||
let key_clone = key.to_string();
|
||||
let cache_clone = cache.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
while let Some(json_data) = rx.recv().await {
|
||||
while rx.recv().await.is_some() {
|
||||
// Drain any other pending notifications so we batch writes
|
||||
while let Ok(_) = rx.try_recv() {}
|
||||
|
||||
let db_inner = db_clone.clone();
|
||||
let key_inner = key_clone.clone();
|
||||
let json_data = {
|
||||
let lock = cache_clone.read().unwrap();
|
||||
serde_json::to_vec(&*lock).unwrap()
|
||||
};
|
||||
|
||||
let _ = tokio::task::spawn_blocking(move || {
|
||||
let write_txn = db_inner.begin_write().unwrap();
|
||||
{
|
||||
@@ -32,7 +43,7 @@ impl<T: DeserializeOwned + Default + Serialize + Clone + Send + 'static> Store<T
|
||||
});
|
||||
|
||||
Self {
|
||||
cache: RwLock::new(initial_data),
|
||||
cache,
|
||||
tx,
|
||||
}
|
||||
}
|
||||
@@ -61,13 +72,11 @@ impl<T: DeserializeOwned + Default + Serialize + Clone + Send + 'static> Store<T
|
||||
}
|
||||
|
||||
pub fn modify<F: FnOnce(&mut T)>(&self, f: F) {
|
||||
let json_data = {
|
||||
{
|
||||
let mut lock = self.cache.write().unwrap();
|
||||
f(&mut lock);
|
||||
// Serialize while holding lock to avoid expensive deep clone of T
|
||||
serde_json::to_vec(&*lock).unwrap()
|
||||
};
|
||||
let _ = self.tx.send(json_data);
|
||||
}
|
||||
let _ = self.tx.try_send(());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user