add exclusive tx run method, #44

src/file.rs

@@ -484,11 +484,10 @@ impl File {
                 let tx_handle = self.tx_handle.take().unwrap();
                 let mut end_pos = 0;
 
-                tx_handle.run(|| {
+                tx_handle.run_all_exclusive(|| {
                     end_pos = wtr.finish()?;
                     Ok(())
                 })?;
-                tx_handle.commit()?;
 
                 // set position
                 self.pos = SeekFrom::Start(end_pos as u64);
@@ -562,7 +561,7 @@ impl File {
 
         let txmgr = self.handle.txmgr.upgrade().ok_or(Error::RepoClosed)?;
         let tx_handle = TxMgr::begin_trans(&txmgr)?;
-        tx_handle.run_all(|| {
+        tx_handle.run_all_exclusive(|| {
             Fnode::set_len(self.handle.clone(), len, tx_handle.txid)
         })?;
 

src/fs/fs.rs

@@ -315,7 +315,7 @@ impl Fs {
 
         let mut fnode = FnodeRef::default();
         let tx_handle = TxMgr::begin_trans(&self.txmgr)?;
-        tx_handle.run_all(|| {
+        tx_handle.run_all_exclusive(|| {
             fnode = Fnode::new_under(
                 &parent,
                 &name,
@@ -500,7 +500,7 @@ impl Fs {
 
         // begin and run transaction
         let tx_handle = TxMgr::begin_trans(&self.txmgr)?;
-        tx_handle.run_all(move || {
+        tx_handle.run_all_exclusive(move || {
             Fnode::remove_from_parent(&fnode_ref, &self.txmgr)?;
             let mut fnode = fnode_ref.write().unwrap();
             fnode
@@ -612,7 +612,7 @@ impl Fs {
         let (tgt_parent, name) = self.resolve_parent(to)?;
 
         // begin and run transaction
-        TxMgr::begin_trans(&self.txmgr)?.run_all(|| {
+        TxMgr::begin_trans(&self.txmgr)?.run_all_exclusive(|| {
             // remove from source
             Fnode::remove_from_parent(&src, &self.txmgr)?;
 

src/trans/txmgr.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 use std::fmt::{self, Debug};
-use std::sync::{Arc, RwLock, Weak};
+use std::sync::{Arc, Mutex, RwLock, Weak};
 
 use linked_hash_map::LinkedHashMap;
 
@@ -189,6 +189,11 @@ impl IntoRef for TxMgr {}
 pub type TxMgrRef = Arc<RwLock<TxMgr>>;
 pub type TxMgrWeakRef = Weak<RwLock<TxMgr>>;
 
+// lock for running exclusive transactions
+lazy_static! {
+    static ref EXCL_TX_LOCK: Arc<Mutex<usize>> = { Arc::new(Mutex::new(0)) };
+}
+
 // Transaction handle
 #[derive(Debug, Default, Clone)]
 pub struct TxHandle {
@@ -221,6 +226,16 @@ impl TxHandle {
         }
     }
 
+    /// Run operations in transaction exclusively and commit
+    #[inline]
+    pub fn run_all_exclusive<F>(&self, oper: F) -> Result<()>
+    where
+        F: FnOnce() -> Result<()>,
+    {
+        let _lock = EXCL_TX_LOCK.lock().unwrap();
+        self.run_all(oper)
+    }
+
     /// Commit a transaction
     #[inline]
     pub fn commit(&self) -> Result<()> {

tests/common/fuzzer.rs

@@ -704,7 +704,7 @@ impl Fuzzer {
     // run the fuzz test
     pub fn run(
         fuzzer: Arc<RwLock<Fuzzer>>,
-        tester: Arc<RwLock<Testable>>,
+        tester: Arc<RwLock<dyn Testable>>,
         rounds: usize,
         worker_cnt: usize,
     ) {
@@ -799,7 +799,7 @@ impl Fuzzer {
     }
 
     // load fuzz test and re-run it
-    pub fn rerun(batch: &str, tester: Box<Testable>) {
+    pub fn rerun(batch: &str, tester: Box<dyn Testable>) {
         // load fuzzer
         let mut fuzzer = Fuzzer::load(batch);
 

tests/file.rs

@@ -156,7 +156,7 @@ fn file_read_write_st() {
         OpenOptions::new()
             .create(true)
             .open(&mut repo, "/file/new")
-            .is_err();
+            .unwrap_err();
 
         // write data to the file should okay
         let mut f = OpenOptions::new()
@@ -400,7 +400,8 @@ fn file_read_write_mt() {
                     .create(true)
                     .open(&mut env.repo, &path)
                     .unwrap();
-                f.write_once(&buf[..]).unwrap();
+                f.write_all(&buf[..]).unwrap();
+                f.finish().unwrap();
             }
         }));
     }

tests/repo.rs

@@ -110,7 +110,7 @@ fn repo_oper() {
         assert_eq!(info.ops_limit(), OpsLimit::Moderate);
         assert_eq!(info.mem_limit(), MemLimit::Interactive);
     }
-    RepoOpener::new().open(&path, &pwd).is_err();
+    RepoOpener::new().open(&path, &pwd).unwrap_err();
     let repo = RepoOpener::new().open(&path, &new_pwd).unwrap();
     let info = repo.info().unwrap();
     assert_eq!(info.ops_limit(), OpsLimit::Moderate);