+//! A trivial global lock transaction system.
+//!
+//! At the moment, this is really just a global static mutex, that needs to be
+//! locked, to ensure the atomicity of a transaction.
+
+use std::sync::Mutex;
+use std::cell::RefCell;
+use std::boxed::FnBox;
+
+
+/// The global transaction lock.
+///
+/// TODO: revert this to use a static mutex, as soon as that is stabilized in
+/// the standard library.
+lazy_static! {
+    static ref TRANSACTION_MUTEX: Mutex<()> = Mutex::new(());
+}
+
+/// Registry for callbacks to be executed at the end of a transaction.
+thread_local!(
+    static CURRENT_TRANSACTION: RefCell<Option<Transaction>> =
+        RefCell::new(None)
+);
+
+
+/// A callback.
+type Callback = Box<FnBox() + 'static>;
+
+
+/// A transaction.
+pub struct Transaction {
+    intermediate: Vec<Callback>,
+    finalizers: Vec<Callback>,
+}
+
+impl Transaction {
+    /// Create a new transaction
+    fn new() -> Transaction {
+        Transaction {
+            intermediate: vec![],
+            finalizers: vec![],
+        }
+    }
+
+    /// Add a callback that will be called, when the transaction is done
+    /// except for finalizers.
+    pub fn later<F: FnOnce() + 'static>(&mut self, callback: F) {
+        self.intermediate.push(Box::new(callback));
+    }
+
+    /// Add a finalizing callback. This should not have far reaching
+    /// side-effects, and in particular not commit by itself. Typical operations
+    /// for a finalizer are executing queued state updates.
+    pub fn end<F: FnOnce() + 'static>(&mut self, callback: F) {
+        self.finalizers.push(Box::new(callback));
+    }
+
+    /// Advance transactions by moving out intermediate stage callbacks.
+    fn advance(&mut self) -> Vec<Callback> {
+        use std::mem;
+        let mut intermediate = vec![];
+        mem::swap(&mut intermediate, &mut self.intermediate);
+        intermediate
+    }
+
+    /// Finalize the transaction
+    fn finalize(self) {
+        for finalizer in self.finalizers {
+            finalizer.call_box(());
+        }
+    }
+}
+
+
+/// Commit a transaction.
+///
+/// If the thread is not running any transactions currently, the global lock is
+/// acquired. Otherwise a new transaction begins, since given the interface of
+/// this module it is safely assumed that the lock is already held.
+pub fn commit<A, F: FnOnce() -> A>(body: F) -> A {
+    use std::mem;
+    // Begin a new transaction
+    let mut prev = CURRENT_TRANSACTION.with(|current| {
+        let mut prev = Some(Transaction::new());
+        mem::swap(&mut prev, &mut current.borrow_mut());
+        prev
+    });
+    // Acquire global lock if necessary
+    let _lock = match prev {
+        None => Some(TRANSACTION_MUTEX.lock().ok()
+                .expect("global transaction mutex poisoned")
+        ),
+        Some(_) => None,
+    };
+    // Perform the main body of the transaction
+    let result = body();
+    // Advance the transaction as long as necessary
+    loop {
+        let callbacks = with_current(Transaction::advance);
+        if callbacks.is_empty() { break }
+        for callback in callbacks {
+            callback.call_box(());
+        }
+    }
+    // Call all finalizers and drop the transaction
+    CURRENT_TRANSACTION.with(|current|
+        mem::swap(&mut prev, &mut current.borrow_mut())
+    );
+    prev.unwrap().finalize();
+    // Return
+    result
+}
+
+
+/// Register a callback during a transaction.
+pub fn with_current<A, F: FnOnce(&mut Transaction) -> A>(action: F) -> A {
+    CURRENT_TRANSACTION.with(|current|
+        match &mut *current.borrow_mut() {
+            &mut Some(ref mut trans) => action(trans),
+            _ => panic!("there is no active transaction to register a callback"),
+        }
+    )
+}
+
+pub fn later<F: FnOnce() + 'static>(action: F) {
+    with_current(|c| c.later(action))
+}
+
+pub fn end<F: FnOnce() + 'static>(action: F) {
+    with_current(|c| c.end(action))
+}
+
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn commit_single() {
+        let mut v = 3;
+        commit(|| v += 5);
+        assert_eq!(v, 8);
+    }
+
+    #[test]
+    fn commit_nested() {
+        let mut v = 3;
+        commit(|| {
+            commit(|| v *= 2);
+            v += 4;
+        });
+        assert_eq!(v, 10);
+    }
+
+    #[test]
+    fn commits_parallel() {
+        use std::sync::{Arc, Mutex};
+        use std::thread;
+
+        // Set up a ref-counted value
+        let v = Arc::new(Mutex::new(3));
+        // Spawn a couple of scoped threads performing atomic operations on it
+        let guards: Vec<_> = (0..3)
+            .map(|_| {
+                let v = v.clone();
+                thread::spawn(move || commit(move || {
+                    // Acquire locks independently, s.t. commit atomicity does
+                    // not rely on the local locks here
+                    *v.lock().unwrap() *= 2;
+                    // …and sleep for a bit
+                    thread::sleep_ms(1);
+                    *v.lock().unwrap() -= 1;
+                }))
+            })
+            .collect();
+        // Rejoin with all guards
+        for guard in guards { guard.join().ok().expect("thread failed"); }
+        // Check result
+        assert_eq!(&*v.lock().unwrap(), &17);
+    }
+}
+

+

Keyboard shortcuts

Search tricks