diff --git a/src/lib.rs b/src/lib.rs index 840c1de..82998c6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,8 +3,6 @@ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ #![deny( - missing_copy_implementations, - missing_debug_implementations, missing_docs, non_camel_case_types, non_snake_case, @@ -43,7 +41,7 @@ mod error; use std::collections::HashMap; use std::fs::File; use std::path::Path; -use std::sync::{RwLock, Mutex}; +use std::sync::{RwLock, RwLockWriteGuard, Mutex}; use std::hash::Hash; use std::borrow::Borrow; @@ -194,6 +192,17 @@ impl Database { } } + /// Checks wether a given key exists in the Database + pub fn contains_key(&self, key: &K) -> bool + where T: Borrow, K: Hash + Eq + { + let map = match self.data.read() { + Ok(guard) => guard, + Err(_) => unimplemented!(), // TODO: Implement recovery? + }; + map.get(key.borrow()).is_some() + } + /// Flushes the Database to disk pub fn flush(&mut self) -> BreakResult<()> { use bincode::serde::serialize; @@ -215,6 +224,100 @@ impl Database { try!(file.flush()); Ok(()) } + + /// Starts a transaction + /// + /// This borrows the Database mutably! Which means that during the Transaction you cannot write + /// to it. This keeps the Database consistent. Be sure to not do anything too costly while it + /// is borrowed. + pub fn transaction<'a>(&'a self) -> Transaction<'a, T> { + let map = match self.data.write() { + Ok(guard) => guard, + Err(_) => unimplemented!(), // TODO: Implement recovery? + }; + Transaction { + cow: Some(map), + lock: &self.data, + data: RwLock::new(HashMap::new()), + } + } +} + +/// A Transaction that is atomic in writes +/// +/// You generate this by calling `transaction` on a `Database` object. +/// The transaction does not get automatically applied when it is dropped, you have to `run` it. +/// This allows for defensive programming where the values are only applied when it is run. +pub struct Transaction<'a, T: Serialize + Deserialize + Eq + Hash + 'a> { + cow: Option>>>, + lock: &'a RwLock>>, + data: RwLock>>, +} + +impl<'a, T: Serialize + Deserialize + Eq + Hash + 'a> Transaction<'a, T> { + /// Insert a given Object into the Database at that key + /// + /// See `Database::insert` for details + pub fn insert(&self, key: &K, obj: S) -> BreakResult<()> + where T: Borrow, K: Hash + PartialEq + ToOwned + { + use bincode::serde::serialize; + use bincode::SizeLimit; + let mut map = match self.data.write() { + Ok(guard) => guard, + Err(_) => unimplemented!(), // TODO: Implement recovery? + }; + map.insert(key.to_owned(), try!(serialize(&obj, SizeLimit::Infinite))); + Ok(()) + } + + /// Retrieves an Object from the Database + /// + /// See `Database::retrieve` for details + pub fn retrieve(&self, key: &K) -> BreakResult + where T: Borrow, K: Hash + Eq + { + use bincode::serde::deserialize; + if self.cow.as_ref().unwrap().contains_key(key) { + match self.cow.as_ref().unwrap().get(key.borrow()) { + Some(t) => Ok(try!(deserialize(t))), + None => Err(BreakError::NotFound), + } + } else { + let map = match self.data.read() { + Ok(guard) => guard, + Err(_) => unimplemented!(), // TODO: Implement recovery? + }; + match map.get(key.borrow()) { + Some(t) => Ok(try!(deserialize(t))), + None => Err(BreakError::NotFound), + } + } + } + + /// Consumes the Transaction and runs it + fn _run(self) -> BreakResult<()> { + if let Some(lock) = self.cow.take() { + drop(lock); + } else { + unreachable!(); + } + + let mut map = match self.data.write() { + Ok(guard) => guard, + Err(_) => unimplemented!(), // TODO: Implement recovery? + }; + + let mut other_map = match self.lock.write() { + Ok(guard) => guard, + Err(_) => unimplemented!(), // TODO: Implement recovery? + }; + + for (k, v) in map.drain() { + other_map.insert(k, v); + } + Ok(()) + } } #[cfg(test)] @@ -243,4 +346,41 @@ mod test { let hello : String = db.retrieve("test").unwrap(); assert_eq!(hello, "Hello World!"); } + + #[test] + fn simple_transaction() { + let tmpf = NamedTempFile::new().unwrap(); + let db = Database::open(tmpf.path()).unwrap(); + assert!(db.retrieve::("test").is_err()); + { + let trans = db.transaction(); + trans.insert("test", "Hello World!").unwrap(); + trans.run(); + } + let hello : String = db.retrieve("test").unwrap(); + assert_eq!(hello, "Hello World!"); + } + + #[test] + fn multithreaded_locking() { + use std::sync::Arc; + let tmpf = NamedTempFile::new().unwrap(); + let db = Arc::new(Database::open(tmpf.path()).unwrap()); + db.insert("value", 0i64).unwrap(); + let mut threads = vec![]; + for _ in 0..10 { + use std::thread; + let a = db.clone(); + threads.push(thread::spawn(move || { + let lock = a.lock(); + let x = lock.retrieve::("value").unwrap(); + lock.insert("value", x + 1).unwrap(); + })); + } + for thr in threads { + thr.join().unwrap(); + } + let x = db.retrieve::("value").unwrap(); + assert_eq!(x, 10); + } }