diff --git a/src/commons/storage/backends/disk.rs b/src/commons/storage/backends/disk.rs index d4c35604..345c96f3 100644 --- a/src/commons/storage/backends/disk.rs +++ b/src/commons/storage/backends/disk.rs @@ -194,9 +194,18 @@ impl Store { where F: for<'a> Fn(&mut SuperTransaction<'a>) -> Result { - let mut file_lock = FileLock::create(self.scope_lock_path(scope))?; - let _write_lock = file_lock.write()?; - op(&mut SuperTransaction::from(self)) + if scope.is_none() { + let mut file_lock = FileLock::create(self.scope_lock_path(scope))?; + let _write_lock = file_lock.write()?; + op(&mut SuperTransaction::from(self)) + } + else { + let mut root_lock = FileLock::create(self.scope_lock_path(None))?; + let _root_lock = root_lock.read()?; + let mut file_lock = FileLock::create(self.scope_lock_path(scope))?; + let _write_lock = file_lock.write()?; + op(&mut SuperTransaction::from(self)) + } } /// Returns the path for the given key. @@ -646,6 +655,12 @@ impl FileLock { Ok(FileLock { lock: fd_lock::RwLock::new(lock_file) }) } + fn read(&mut self) -> Result, Error> { + self.lock + .read() + .map_err(|e| Error::other(format!("Cannot get file lock: {e}"))) + } + fn write(&mut self) -> Result, Error> { self.lock .write() diff --git a/src/commons/storage/backends/memory.rs b/src/commons/storage/backends/memory.rs index 6b290383..4742cf5e 100644 --- a/src/commons/storage/backends/memory.rs +++ b/src/commons/storage/backends/memory.rs @@ -1,11 +1,10 @@ //! In-memory storage. -use std::{error, fmt, mem, thread}; -use std::collections::{HashMap, HashSet}; +use std::{error, fmt, mem}; +use std::collections::HashMap; use std::ops::DerefMut; use std::str::FromStr; -use std::sync::{Arc, Mutex, MutexGuard}; -use std::time::Duration; +use std::sync::{Arc, Mutex, MutexGuard, RwLock}; use serde::de::DeserializeOwned; use serde::ser::Serialize; use serde_json::Value; @@ -68,14 +67,10 @@ impl Location { return Err(Error::MissingSourceNamespace(src_ns.into())) }; - if !src.locks().is_empty() { - return Err(Error::PendingLocks); - } + src.try_clear_locks()?; let dst = namespaces.entry(dst_ns.into()).or_default().clone(); - if !dst.locks().is_empty() { - return Err(Error::PendingLocks); - } + dst.try_clear_locks()?; let mut dst_scopes = dst.scopes(); if !dst_scopes.is_empty() { @@ -110,25 +105,34 @@ impl Store { where F: for<'a> Fn(&mut SuperTransaction<'a>) -> Result { - let wait = Duration::from_millis(10); - let tries = 1000; - - for i in 0..tries { - if self.namespace.locks().insert(scope.map(Into::into)) { - // The scope was not yet present. We’ve won and can go on. - break - } - else if i >= tries { - return Err(Error::ScopeLocked(scope.map(Into::into)).into()) - } - thread::sleep(wait); + match scope { + Some(scope) => self.execute_scoped(scope.into(), op), + None => self.execute_global(op), } + } - let res = op(&mut SuperTransaction::from(self)); + fn execute_global( + &self, op: F + ) -> Result + where + F: for<'a> Fn(&mut SuperTransaction<'a>) -> Result + { + let _lock = self.namespace.get_lock(None); + let _lock = _lock.write().expect("poisoned lock"); + op(&mut SuperTransaction::from(self)) + } - self.namespace.locks().remove(&scope.map(Into::into)); - - res + fn execute_scoped( + &self, scope: Box, op: F + ) -> Result + where + F: for<'a> Fn(&mut SuperTransaction<'a>) -> Result + { + let _root_lock = self.namespace.get_lock(None); + let _root_lock = _root_lock.read().expect("poisoned lock"); + let _lock = self.namespace.get_lock(Some(scope)); + let _lock = _lock.write().expect("poisoned lock"); + op(&mut SuperTransaction::from(self)) } } @@ -408,7 +412,7 @@ impl MemoryScopes { #[derive(Debug, Default)] struct MemoryNamespace { scopes: Mutex, - locks: Mutex>>>, + locks: Mutex>, Arc>>>, } impl MemoryNamespace { @@ -416,9 +420,27 @@ impl MemoryNamespace { self.scopes.lock().expect("poisoned lock") } - fn locks(&self) -> MutexGuard<'_, HashSet>>> { + fn get_lock(&self, scope: Option>) -> Arc> { + self.locks().entry(scope).or_default().clone() + } + + fn locks( + &self + ) -> MutexGuard<'_, HashMap>, Arc>>> { self.locks.lock().expect("poisoned lock") } + + fn try_clear_locks(&self) -> Result<(), Error> { + // Try to get a write lock on every present lock. If that succeeds, + // clear the hash map. + + let mut locks = self.locks(); + for lock in locks.values() { + drop(lock.try_write().map_err(|_| Error::PendingLocks)?); + } + locks.clear(); + Ok(()) + } } diff --git a/src/commons/storage/test.rs b/src/commons/storage/test.rs index 067947ea..a0123de2 100644 --- a/src/commons/storage/test.rs +++ b/src/commons/storage/test.rs @@ -127,14 +127,21 @@ testfns! { fn drop_global_key(harness: impl Harness) { let store = harness.store(NAMESPACE); + eprintln!("1"); assert!(store.drop_key(None, KEY).is_err()); + eprintln!("2"); + store.store(None, KEY, &CONTENT).unwrap(); + eprintln!("3"); assert!(store.has(None, KEY).unwrap()); + eprintln!("4"); store.drop_key(None, KEY).unwrap(); + eprintln!("5"); assert!(!store.has(None, KEY).unwrap()); + eprintln!("6"); } fn drop_scoped_key(harness: impl Harness) {