diff --git a/lightning-persister/Cargo.toml b/lightning-persister/Cargo.toml index ec34fa8a88d..031a690476e 100644 --- a/lightning-persister/Cargo.toml +++ b/lightning-persister/Cargo.toml @@ -17,6 +17,9 @@ rustdoc-args = ["--cfg", "docsrs"] bitcoin = "0.32.2" lightning = { version = "0.2.0", path = "../lightning" } +# TODO: Make conditional? +tokio = { version = "1.35", features = [ "macros", "rt", "rt-multi-thread", "sync", "time", "fs", "io-util" ] } + [target.'cfg(windows)'.dependencies] windows-sys = { version = "0.48.0", default-features = false, features = ["Win32_Storage_FileSystem", "Win32_Foundation"] } diff --git a/lightning-persister/src/fs_store.rs b/lightning-persister/src/fs_store.rs index 9f490eb6fb2..332fec34f42 100644 --- a/lightning-persister/src/fs_store.rs +++ b/lightning-persister/src/fs_store.rs @@ -30,15 +30,14 @@ fn path_to_windows_str>(path: &T) -> Vec { path.as_ref().encode_wide().chain(Some(0)).collect() } -// The number of read/write/remove/list operations after which we clean up our `locks` HashMap. -const GC_LOCK_INTERVAL: usize = 25; - /// A [`KVStoreSync`] implementation that writes to and reads from the file system. pub struct FilesystemStore { data_dir: PathBuf, tmp_file_counter: AtomicUsize, - gc_counter: AtomicUsize, - locks: Mutex>>>, + + // Per path lock that ensures that we don't have concurrent writes to the same file. The lock also encapsulates the + // latest written version per key. + locks: Mutex>>>, } impl FilesystemStore { @@ -46,8 +45,7 @@ impl FilesystemStore { pub fn new(data_dir: PathBuf) -> Self { let locks = Mutex::new(HashMap::new()); let tmp_file_counter = AtomicUsize::new(0); - let gc_counter = AtomicUsize::new(1); - Self { data_dir, tmp_file_counter, gc_counter, locks } + Self { data_dir, tmp_file_counter, locks } } /// Returns the data directory. @@ -55,18 +53,6 @@ impl FilesystemStore { self.data_dir.clone() } - fn garbage_collect_locks(&self) { - let gc_counter = self.gc_counter.fetch_add(1, Ordering::AcqRel); - - if gc_counter % GC_LOCK_INTERVAL == 0 { - // Take outer lock for the cleanup. - let mut outer_lock = self.locks.lock().unwrap(); - - // Garbage collect all lock entries that are not referenced anymore. - outer_lock.retain(|_, v| Arc::strong_count(&v) > 1); - } - } - fn get_dest_dir_path( &self, primary_namespace: &str, secondary_namespace: &str, ) -> std::io::Result { @@ -90,36 +76,12 @@ impl FilesystemStore { Ok(dest_dir_path) } -} - -impl KVStoreSync for FilesystemStore { - fn read( - &self, primary_namespace: &str, secondary_namespace: &str, key: &str, - ) -> lightning::io::Result> { - check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "read")?; - - let mut dest_file_path = self.get_dest_dir_path(primary_namespace, secondary_namespace)?; - dest_file_path.push(key); - - let mut buf = Vec::new(); - { - let inner_lock_ref = { - let mut outer_lock = self.locks.lock().unwrap(); - Arc::clone(&outer_lock.entry(dest_file_path.clone()).or_default()) - }; - let _guard = inner_lock_ref.read().unwrap(); - - let mut f = fs::File::open(dest_file_path)?; - f.read_to_end(&mut buf)?; - } - - self.garbage_collect_locks(); - - Ok(buf) - } - fn write( + /// Writes a specific version of a key to the filesystem. If a newer version has been written already, this function + /// returns early without writing. + pub(crate) fn write_version( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8], + version: Option, ) -> lightning::io::Result<()> { check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "write")?; @@ -153,7 +115,18 @@ impl KVStoreSync for FilesystemStore { let mut outer_lock = self.locks.lock().unwrap(); Arc::clone(&outer_lock.entry(dest_file_path.clone()).or_default()) }; - let _guard = inner_lock_ref.write().unwrap(); + let mut last_written_version = inner_lock_ref.write().unwrap(); + + // If a version is provided, we check if we already have a newer version written. This is used in async + // contexts to realize eventual consistency. + if let Some(version) = version { + if version <= *last_written_version { + // If the version is not greater, we don't write the file. + return Ok(()); + } + + *last_written_version = version; + } #[cfg(not(target_os = "windows"))] { @@ -200,10 +173,39 @@ impl KVStoreSync for FilesystemStore { } }; - self.garbage_collect_locks(); - res } +} + +impl KVStoreSync for FilesystemStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> lightning::io::Result> { + check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "read")?; + + let mut dest_file_path = self.get_dest_dir_path(primary_namespace, secondary_namespace)?; + dest_file_path.push(key); + + let mut buf = Vec::new(); + { + let inner_lock_ref = { + let mut outer_lock = self.locks.lock().unwrap(); + Arc::clone(&outer_lock.entry(dest_file_path.clone()).or_default()) + }; + let _guard = inner_lock_ref.read().unwrap(); + + let mut f = fs::File::open(dest_file_path)?; + f.read_to_end(&mut buf)?; + } + + Ok(buf) + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8], + ) -> lightning::io::Result<()> { + self.write_version(primary_namespace, secondary_namespace, key, buf, None) + } fn remove( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, @@ -295,8 +297,6 @@ impl KVStoreSync for FilesystemStore { } } - self.garbage_collect_locks(); - Ok(()) } @@ -325,8 +325,6 @@ impl KVStoreSync for FilesystemStore { keys.push(key); } - self.garbage_collect_locks(); - Ok(keys) } } diff --git a/lightning-persister/src/fs_store_async.rs b/lightning-persister/src/fs_store_async.rs new file mode 100644 index 00000000000..082dfca6886 --- /dev/null +++ b/lightning-persister/src/fs_store_async.rs @@ -0,0 +1,155 @@ +//! Objects related to [`FilesystemStoreAsync`] live here. + +use std::sync::{ + atomic::{AtomicU64, Ordering}, + Arc, +}; + +use crate::fs_store::FilesystemStore; +use core::future::Future; +use core::pin::Pin; +use lightning::util::persist::{KVStore, KVStoreSync}; + +/// An asynchronous extension of FilesystemStore, implementing the `KVStore` trait for async operations. It is shaped as +/// a wrapper around an existing [`FilesystemStore`] so that the same locks are used. This allows both the sync and +/// async interface to be used simultaneously. +pub struct FilesystemStoreAsync { + inner: Arc, + + // Version counter to ensure that writes are applied in the correct order. It is assumed that read, list and remove + // operations aren't sensitive to the order of execution. + version_counter: AtomicU64, +} + +impl FilesystemStoreAsync { + /// Creates a new instance of [`FilesystemStoreAsync`]. + pub fn new(inner: Arc) -> Self { + Self { inner, version_counter: AtomicU64::new(0) } + } +} + +impl KVStore for FilesystemStoreAsync { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> Pin, lightning::io::Error>> + 'static + Send>> { + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + let this = Arc::clone(&self.inner); + + Box::pin(async move { + tokio::task::spawn_blocking(move || { + this.read(&primary_namespace, &secondary_namespace, &key) + }) + .await + .unwrap_or_else(|e| Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))) + }) + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8], + ) -> Pin> + 'static + Send>> { + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + let buf = buf.to_vec(); + let this = Arc::clone(&self.inner); + + // Obtain a version number to retain the call sequence. + let version = self.version_counter.fetch_add(1, Ordering::SeqCst); + + Box::pin(async move { + tokio::task::spawn_blocking(move || { + this.write_version( + &primary_namespace, + &secondary_namespace, + &key, + &buf, + Some(version), + ) + }) + .await + .unwrap_or_else(|e| Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))) + }) + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> Pin> + 'static + Send>> { + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + let this = Arc::clone(&self.inner); + + Box::pin(async move { + tokio::task::spawn_blocking(move || { + this.remove(&primary_namespace, &secondary_namespace, &key, lazy) + }) + .await + .unwrap_or_else(|e| Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))) + }) + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> Pin, lightning::io::Error>> + 'static + Send>> { + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let this = Arc::clone(&self.inner); + + Box::pin(async move { + tokio::task::spawn_blocking(move || this.list(&primary_namespace, &secondary_namespace)) + .await + .unwrap_or_else(|e| { + Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e)) + }) + }) + } +} + +mod test { + use crate::{fs_store::FilesystemStore, fs_store_async::FilesystemStoreAsync}; + use lightning::util::persist::KVStore; + use std::sync::Arc; + + #[tokio::test] + async fn read_write_remove_list_persist() { + let mut temp_path = std::env::temp_dir(); + temp_path.push("test_read_write_remove_list_persist"); + let fs_store = Arc::new(FilesystemStore::new(temp_path)); + let fs_store_async = FilesystemStoreAsync::new(Arc::clone(&fs_store)); + + let data1 = [42u8; 32]; + let data2 = [43u8; 32]; + + let primary_namespace = "testspace"; + let secondary_namespace = "testsubspace"; + let key = "testkey"; + + // Test writing the same key twice with different data. Execute the asynchronous part out of order to ensure + // that eventual consistency works. + let fut1 = fs_store_async.write(primary_namespace, secondary_namespace, key, &data1); + let fut2 = fs_store_async.write(primary_namespace, secondary_namespace, key, &data2); + + fut2.await.unwrap(); + fut1.await.unwrap(); + + // Test list. + let listed_keys = + fs_store_async.list(primary_namespace, secondary_namespace).await.unwrap(); + assert_eq!(listed_keys.len(), 1); + assert_eq!(listed_keys[0], key); + + // Test read. We expect to read data2, as the write call was initiated later. + let read_data = + fs_store_async.read(primary_namespace, secondary_namespace, key).await.unwrap(); + assert_eq!(data2, &*read_data); + + // Test remove. + fs_store_async.remove(primary_namespace, secondary_namespace, key, false).await.unwrap(); + + let listed_keys = + fs_store_async.list(primary_namespace, secondary_namespace).await.unwrap(); + assert_eq!(listed_keys.len(), 0); + } +} diff --git a/lightning-persister/src/lib.rs b/lightning-persister/src/lib.rs index 9d4df264d24..23b085ec426 100644 --- a/lightning-persister/src/lib.rs +++ b/lightning-persister/src/lib.rs @@ -9,6 +9,7 @@ extern crate criterion; pub mod fs_store; +pub mod fs_store_async; mod utils;