blocking-transactions (empty) → 0.1.0.4
raw patch · 5 files changed
+631/−0 lines, 5 filesdep +arraydep +basedep +containerssetup-changed
Dependencies added: array, base, containers, parallel, random, stm
Files
- BlockingTransactions/BlockingTransactions.hs +409/−0
- Examples/RandomAccounts.hs +146/−0
- LICENSE +30/−0
- Setup.hs +3/−0
- blocking-transactions.cabal +43/−0
+ BlockingTransactions/BlockingTransactions.hs view
@@ -0,0 +1,409 @@+{-# LANGUAGE Arrows, GADTs, ExistentialQuantification, Rank2Types #-}+{-# OPTIONS_GHC -Wall #-}++-- | Composable blocking transactions, based on the+-- blog post:+-- <http://blog.downstairspeople.org/2010/06/06/composable-blocking-transactions/>+module BlockingTransactions.BlockingTransactions+ (+ -- * Transactional Variables+ BVar,+ newBVar,+ peekBVar,+ pokeBVar,+ modifyBVar,+ -- * Blocking Transaction Monad+ runBTM,+ BTM,+ Value,+ when, unless,+ readBVar,+ writeBVar,+ retry,+ -- * Blocking Transaction Arrow+ runBTA,+ BTA,+ fetchBVar,+ storeBVar,+ retryWhen,+ retryUnless) where++import Prelude hiding ((.),id)+import Control.Category+import Control.Arrow+import Control.Applicative hiding (empty)+import System.IO.Unsafe+import Control.Monad hiding (when,unless)+import Control.Concurrent+import Control.Parallel+import Data.IORef+import qualified Data.Set as Set++-- | A transactional variable with a blocking implementation.+data BVar a = BVar {+ bvar_index :: Integer,+ -- ^ The unique value for ordering purposes.+ bvar_lock :: (MVar ()),+ -- ^ The mutex.+ bvar_data :: (IORef a),+ -- ^ User data.+ bvar_retry_list :: (IORef [MVar ()])+ -- ^ Single-use modification listeners. Whenever a variable is modified,+ -- these will all be written to and then discarded.+ }++-- | An untyped BVar.+data AnonymousVar where+ Anon :: BVar a -> AnonymousVar++instance Eq (BVar a) where+ a == b = bvar_index a == bvar_index b++instance Ord (BVar a) where+ BVar a _ _ _ `compare` BVar b _ _ _ = a `compare` b+ BVar a _ _ _ > BVar b _ _ _ = a > b+ BVar a _ _ _ >= BVar b _ _ _ = a >= b+ BVar a _ _ _ < BVar b _ _ _ = a > b+ BVar a _ _ _ <= BVar b _ _ _ = a <= b++instance Eq AnonymousVar where+ Anon (BVar a _ _ _) == Anon (BVar b _ _ _) = a == b++instance Ord AnonymousVar where+ Anon (BVar a _ _ _) `compare` Anon (BVar b _ _ _) = a `compare` b+ Anon (BVar a _ _ _) > Anon (BVar b _ _ _) = a > b+ Anon (BVar a _ _ _) >= Anon (BVar b _ _ _) = a >= b+ Anon (BVar a _ _ _) < Anon (BVar b _ _ _) = a < b+ Anon (BVar a _ _ _) <= Anon (BVar b _ _ _) = a <= b++{-# NOINLINE unique_source #-}+unique_source :: IORef Integer+unique_source = unsafePerformIO $ newIORef 0++-- | Construct a new transactional variable.+newBVar :: a -> IO (BVar a)+newBVar a =+ BVar <$>+ atomicModifyIORef unique_source (\n -> (succ n,n)) <*>+ newMVar () <*>+ newIORef a <*>+ newIORef []++-- | Observe the contents of a transactional variable.+peekBVar :: BVar a -> IO a+peekBVar bvar =+ do takeMVar $ bvar_lock bvar+ result <- readIORef $ bvar_data bvar+ putMVar (bvar_lock bvar) ()+ return result++-- | One-off write to a transactional variable.+pokeBVar :: BVar a -> a -> IO ()+pokeBVar bvar value =+ do takeMVar $ bvar_lock bvar+ writeIORef (bvar_data bvar) value+ signal_list <- readIORef (bvar_retry_list bvar)+ writeIORef (bvar_retry_list bvar) []+ putMVar (bvar_lock bvar) ()+ forM_ signal_list $ \v -> tryPutMVar v ()++-- | Perform a transaction using only a single variable.+modifyBVar :: BVar a -> (a -> (a,b)) -> IO b+modifyBVar bvar f =+ do takeMVar $ bvar_lock bvar+ a <- readIORef (bvar_data bvar)+ let ab = f a+ writeIORef (bvar_data bvar) $ fst ab+ signal_list <- readIORef (bvar_retry_list bvar)+ putMVar (bvar_lock bvar) ()+ forM_ signal_list $ \v -> tryPutMVar v ()+ return $ snd ab++{------------------------------------------------------------------------------}+-- Monadic Interface+{------------------------------------------------------------------------------}++-- | State during the progress of a black transaction.+data BlockingStatus = BlockingStatus {+ should_retry :: IORef Bool,+ -- ^ Set to true if the user decides to retry a transaction.+ -- Initially this value is false.+ is_active_branch :: IORef Bool+ -- ^ Set to indicate conditional sections. When false, all+ -- writes are supressed.+ }++-- | An opaque value. It can be modified and combined with other opaque values,+-- but not observed.+--+-- The type variable @e@ binds the value to the monadic context in which it+-- occurs, (this is identical to the @runST@ existential type trick).+data Value e a = Value { fromValue :: a }++-- | The blocking transaction monad.+data BTM e a =+ BTReturn a+-- Consists of: a static return value, a constructor for the working set of+-- transactional variables, and the transactional behavior.+ | BTM a ([AnonymousVar] -> [AnonymousVar]) (BlockingStatus -> IO a)++-- | Unsafe: get the result of a transaction without running it.+-- (Doesn't preserve type system properties of BTM, and any values+-- returned will be internall undefined.)+staticResult :: BTM e a -> a+staticResult (BTReturn a) = a+staticResult (BTM a _ _) = a++-- | Unsafe: get the transactional behavior.+-- (Doesn't preserve type system properties of BTM.)+operation :: BTM e a -> BlockingStatus -> IO a+operation (BTReturn a) _ = return a+operation (BTM _ _ op) x = op x++-- | Construct the working set of a transaction.+workingSet :: BTM e a -> [AnonymousVar] -> [AnonymousVar]+workingSet (BTReturn _) = id+workingSet (BTM _ working_set _) = working_set++instance Functor (BTM e) where+ fmap f = liftM f++instance Functor (Value e) where+ fmap f (Value a) = Value $ f a++instance Applicative (BTM e) where+ pure = return+ a <*> b =+ do a' <- a+ b' <- b+ return $ a' b'++instance Applicative (Value e) where+ pure x = Value x+ Value a <*> Value b = Value $ a b++instance Monad (BTM e) where+ return = BTReturn+ (BTReturn k) >>= m = m k+ -- The key here is that we bind the static result of the previous+ -- operation to the static result of the subsequent operation, and+ -- the monadic IO result to the subsequent monadic IO result.+ --+ -- For example, if we read a variable that contains the number 23,+ -- the static result will be $Value undefined$ while the monadic+ -- result will be $Value 23$.+ --+ -- The monadic result is unobservable without running the transaction+ -- but the static result, wrapped in a 'Value', is simply unobservable.+ --+ -- However, whenever we return a constant value, which is unwrapped,+ -- the value is observable.+ k >>= m =+ case m $ staticResult k of+ BTReturn j ->+ BTM j+ (workingSet k) $+ \x ->+ do k' <- operation k x+ operation (m k') x+ BTM j ws _ ->+ BTM j+ (ws . workingSet k) $+ \x ->+ do k' <- operation k x+ operation (m k') x++instance Monad (Value e) where+ return = Value+ (Value k) >>= m = m k++invalid_value :: Value e a+invalid_value = Value $ error "BlockingTransaction (Value): Inaccessable value."++-- | Flow control. Skip the critical section if the predicate is false.+when :: Value e Bool ->+ BTM e (Value e ()) ->+ BTM e (Value e ())+when b a = switch b a >> return (Value ())++-- | Flow control. Skip the critical section if the predicate is true.+unless :: Value e Bool ->+ BTM e (Value e ()) ->+ BTM e (Value e ())+unless v = when (fmap not v)++-- | Flow control for the BTM monad. If the predicate is false,+-- then all writes during the critical section are suppressed.+-- Since reads always happen, the return value is available+-- even in a suppressed branch.+switch :: Value e Bool ->+ BTM e (Value e a) ->+ BTM e (Value e a)+switch _ (BTReturn a) = return a+switch v action = BTM (staticResult action) (workingSet action) $+ \x ->+ do is_active <- readIORef $ is_active_branch x+ let active_section = if is_active then fromValue v else False+ writeIORef (is_active_branch x) $ active_section+ a <- operation action x+ writeIORef (is_active_branch x) is_active+ return a++-- | Write to a variable.+writeBVar :: BVar a -> Value e a -> BTM e (Value e ())+writeBVar bv@(BVar _ _ ref _) (Value i) = bv `seq` BTM+ (Value ()) (Anon bv:) $+ \x ->+ do is_active <- readIORef $ is_active_branch x+ modifyIORef ref $ if is_active then const i else id+ return $ Value ()++-- | Read from a variable.+{-# NOINLINE readBVar #-}+readBVar :: BVar a -> BTM e (Value e a)+readBVar bv@(BVar _ _ ref _) = bv `seq` BTM+ invalid_value (Anon bv:) $+ \_ -> liftM Value $ readIORef ref++-- | Electively retry. This will restore all variables to their+-- state before the transaction began, and listen for a change+-- to any variable in the working set before trying the+-- transaction again.+retry :: BTM e (Value e ())+retry = BTM (Value ()) id $+ \x ->+ do is_active <- readIORef $ is_active_branch x+ modifyIORef (should_retry x) $ (||) is_active+ return $ Value ()++-- | Commit a blocking transaction.+runBTM :: (forall e. BTM e (Value e a)) -> IO a+runBTM action = runBlockingTransaction action++{------------------------------------------------------------------------------}+-- Evaluation --+{------------------------------------------------------------------------------}++-- | Run a blocking transaction without type system tricks.+runBlockingTransaction :: forall a. BTM () (Value () a) -> IO a+runBlockingTransaction bm =+ do -- construct the working set+ let working_set = Set.toAscList $ Set.fromList $ workingSet bm []+ -- create state variables+ should_retry_var <- newIORef False+ is_active_var <- newIORef True+ -- acquire locks on the working set+ restore <- openVariables working_set+ -- perform the transaction+ (Value pre_result) <- operation bm (BlockingStatus should_retry_var is_active_var)+ -- in the event of a retry, store a signal in the working set,+ -- so that we can wait for modification.+ retry_signal <- newEmptyMVar+ b <- readIORef should_retry_var+ restore retry_signal b+ -- release the locks on the working set+ closeVariables b working_set+ -- this is the first place where we actually force a result of+ -- the transaction: conditionally wait on the retry signal and re-run the+ -- the transaction.+ result <- if b then+ do takeMVar retry_signal+ runBlockingTransaction bm+ else return pre_result+ return result++-- | Acquire locks on the working set: returns an operation to restore the+-- working set to its initial state. The restoration function+-- requires a retry signal, and a value indicating whether+-- or not the user elected to retry. (If that value is false,+-- the entire restoration function is suppressed.)+openVariables :: [AnonymousVar] -> IO (MVar () -> Bool -> IO ())+openVariables s =+ liftM (\as retry_flag b -> sequence_ $ map (\f -> f retry_flag b) as) $+ forM s $ \(Anon (BVar _ m ref retry_signal_list)) ->+ do takeMVar m+ a <- readIORef ref+ return $ \retry_flag b ->+ do modifyIORef ref $ if b then (const a) else id+ modifyIORef retry_signal_list $+ if b then (retry_flag:) else id++-- | Release locks on the working set.+closeVariables :: Bool -> [AnonymousVar] -> IO ()+closeVariables b s =+ do retry_signals <- liftM concat $ forM s $+ \(Anon bv) ->+ do result <- readIORef (bvar_retry_list bv)+ modifyIORef (bvar_retry_list bv) $ if b then id else (const [])+ a <- readIORef (bvar_data bv)+ a `par` return ()+ _ <- tryPutMVar (bvar_lock bv) ()+ return result+ if (not b) then forM_ retry_signals $ \m ->+ tryPutMVar m ()+ else return ()++{------------------------------------------------------------------------------}+-- Arrow Interface --+{------------------------------------------------------------------------------}++-- | The blocking transaction arrow. The semantics are+-- identical to the equivalent operations on the+-- monadic interface.+newtype BTA a b = BTA ((Value () a) -> BTM () (Value () b))++runBTA :: BTA a b -> a -> IO b+runBTA (BTA action) a = runBlockingTransaction (action (Value a))++instance Functor (BTA a) where+ fmap f = (<<<) (arr f)++instance Applicative (BTA a) where+ pure a = arr (const a)+ a <*> b = proc i ->+ do a' <- a -< i+ b' <- b -< i+ returnA -< a' b'++instance Category BTA where+ (BTA a) . (BTA b) = BTA $ \i ->+ do x <- b i+ a x+ id = BTA return++instance Arrow BTA where+ arr f = BTA $ return . fmap f+ first (BTA action) = BTA $ \ab ->+ do a <- action $ fmap fst ab+ return $+ do ab' <- ab+ a' <- a+ return $ first (const a') ab'+ second (BTA action) = BTA $ \ab ->+ do b <- action $ fmap snd ab+ return $+ do ab' <- ab+ b' <- b+ return $ second (const b') ab'++instance ArrowChoice BTA where+ left (BTA a) = BTA $ \i ->+ do result <- switch (fmap (either (const True) (const False)) i) $+ a $ fmap (either id (const $ fromValue invalid_value)) i+ return $ fmap (either (Left . const (fromValue result)) Right) i++-- | As 'readBVar'.+storeBVar :: BVar a -> BTA a ()+storeBVar bv = BTA $ writeBVar bv++-- | As 'writeBVar'.+fetchBVar :: BVar a -> BTA () a+fetchBVar bv = BTA $ const $ readBVar bv++-- | As 'retry'.+retryWhen :: BTA Bool ()+retryWhen = BTA $ \b -> when b retry++retryUnless :: BTA Bool ()+retryUnless = BTA $ \b -> unless b retry+
+ Examples/RandomAccounts.hs view
@@ -0,0 +1,146 @@+{-# LANGUAGE Arrows #-}+module Main+ (main)+ where++import System.IO+import Data.Array+import Control.Concurrent+import Control.Concurrent.STM+import Control.Monad+import Control.Applicative+import Control.Arrow+import qualified BlockingTransactions.BlockingTransactions as BT+import System.Environment+import System.Random++-- A benchmark that creates a number (param_accounts) of bank accounts+-- and then performs a large number (param_transactions) of transactions on a+-- large number (param_threads) of threads, running in parallel. Transactions+-- always move one dollar from a larger account to a smaller account.++-- This benchmark is written (hopefull identically) using STM (conventional+-- software transactional memory), BTM (blocking transactions monad),+-- and BTA (blocking transactions arrow).++-- Only the STM implementation is documented, since all three implementations+-- should have the same form.++param_threads :: Int+param_threads = 100++param_accounts :: Int+param_accounts = 100++param_transactions :: Int+param_transactions = 1000++main :: IO ()+main =+ do args <- getArgs+ case args of+ ["stm"] -> benchmarkSTM+ ["btm"] -> benchmarkBTM+ ["bta"] -> benchmarkBTA++benchmarkSTM :: IO ()+benchmarkSTM =+ do -- a flag indicating when we can start+ start <- newTVarIO False+ -- variable indicating when we are finished+ threads_remaining <- newTVarIO param_threads+ -- populate the accounts+ accounts <- liftM (listArray (1,param_accounts)) $+ forM [1..param_accounts] $ \_ -> (newTVarIO =<<) $+ getStdRandom $ randomR (1,param_transactions)+ -- launch worker threads+ forM_ [1..param_threads] $ \_ -> (>> return ()) $ forkIO $+ do -- wait until we can start+ atomically $+ do can_start <- readTVar start+ when (not can_start) retry+ -- perform 1-dollar transactions between random accounts+ forM_ [1..param_transactions] $ \_ ->+ do ac1 <- getStdRandom $ randomR (1,param_accounts)+ ac2 <- getStdRandom $ randomR (1,param_accounts)+ atomically $+ do v1 <- readTVar (accounts ! ac1)+ v2 <- readTVar (accounts ! ac2)+ when (v1 > v2) $+ do writeTVar (accounts ! ac1) $ pred v1+ writeTVar (accounts ! ac2) $ succ v2+ -- indicate that we are finished+ atomically $ writeTVar threads_remaining . pred =<<+ readTVar threads_remaining+ return ()+ -- indicate that we can start (all threads are live)+ atomically $ writeTVar start True+ -- wait until we are finished+ atomically $+ do x <- readTVar threads_remaining+ when (x /= 0) retry+ -- print the answer+ print =<< mapM (atomically . readTVar) (elems accounts)+ return ()++benchmarkBTM :: IO ()+benchmarkBTM =+ do start <- BT.newBVar False+ threads_remaining <- BT.newBVar param_threads+ accounts <- liftM (listArray (1,param_accounts)) $+ forM [1..param_accounts] $ \_ -> (BT.newBVar =<<) $+ getStdRandom $ randomR (1,param_transactions)+ forM_ [1..param_threads] $ \_ -> (>> return ()) $ forkIO $+ do BT.runBTM $+ do can_start <- BT.readBVar start+ BT.unless can_start BT.retry+ forM_ [1..param_transactions] $ \_ ->+ do ac1 <- getStdRandom $ randomR (1,param_accounts)+ ac2 <- getStdRandom $ randomR (1,param_accounts)+ BT.runBTM $+ do v1 <- BT.readBVar (accounts ! ac1)+ v2 <- BT.readBVar (accounts ! ac2)+ BT.when ((>) <$> v1 <*> v2) $+ do BT.writeBVar (accounts ! ac1) $ fmap pred v1+ BT.writeBVar (accounts ! ac2) $ fmap succ v2+ BT.runBTM $ BT.writeBVar threads_remaining . fmap pred =<<+ BT.readBVar threads_remaining+ return ()+ BT.pokeBVar start True+ BT.runBTM $+ do x <- BT.readBVar threads_remaining+ BT.when (fmap (/= 0) x) BT.retry+ print =<< mapM (\v -> BT.runBTM $ BT.readBVar v) (elems accounts)+ return ()++benchmarkBTA :: IO ()+benchmarkBTA =+ do start <- BT.newBVar False+ threads_remaining <- BT.newBVar param_threads+ accounts <- liftM (listArray (1,param_accounts)) $+ forM [1..param_accounts] $ \_ -> (BT.newBVar =<<) $+ getStdRandom $ randomR (1,param_transactions)+ forM_ [1..param_threads] $ \_ -> (>> return ()) $ forkIO $+ do flip BT.runBTA () $ proc () ->+ do can_start <- BT.fetchBVar start -< ()+ BT.retryUnless -< can_start+ forM_ [1..param_transactions] $ \_ ->+ do ac1 <- getStdRandom $ randomR (1,param_accounts)+ ac2 <- getStdRandom $ randomR (1,param_accounts)+ flip BT.runBTA () $ proc () ->+ do v1 <- BT.fetchBVar (accounts ! ac1) -< ()+ v2 <- BT.fetchBVar (accounts ! ac2) -< ()+ case () of+ () | v1 > v2 ->+ do BT.storeBVar (accounts ! ac1) -< pred v1+ BT.storeBVar (accounts ! ac2) -< succ v2+ () | otherwise -> returnA -< ()+ flip BT.runBTA () $ BT.storeBVar threads_remaining <<< arr pred+ <<< BT.fetchBVar threads_remaining+ return ()+ BT.pokeBVar start True+ flip BT.runBTA () $+ BT.retryWhen <<< arr (/= 0) <<< BT.fetchBVar threads_remaining+ print =<< mapM (flip BT.runBTA () . BT.fetchBVar) (elems accounts)+ return ()+
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Christopher Lane Hinson 2010++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Christopher Lane Hinson nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+import Distribution.Simple+main = defaultMain
+ blocking-transactions.cabal view
@@ -0,0 +1,43 @@+Name: blocking-transactions++-- The package version. See the Haskell package versioning policy+-- (http://www.haskell.org/haskellwiki/Package_versioning_policy) for+-- standards guiding when and how versions should be incremented.+Version: 0.1.0.4++Synopsis: Composable, blocking transactions.++Description: A variable type (BVar), monad (BTM), and arrow+ (BTA), which provide fast, atomic, composable, blocking+ inter-thread communication. Blocking transactions are+ strictly less expressive than software transactional+ memory, but may be more efficient under very high+ contention.++Homepage: http://www.downstairspeople.org/git/blocking-transactions.git+License: BSD3+License-file: LICENSE+Author: Christopher Lane Hinson+Maintainer: lane@downstairspeople.org+Stability: Experimental+Category: Concurrency+Build-type: Simple+Cabal-version: >=1.2++Flag Tests+ Default: False++Library+ Exposed-modules: BlockingTransactions.BlockingTransactions++ Build-depends: base >=4&&<5, containers, parallel+ ghc-options: -O2++Executable _RandomAccounts+ main-is: Examples/RandomAccounts.hs+ Build-depends: base >=4&&<5, random, array, stm+ ghc-options: -threaded -O2+ ghc-prof-options: -prof -auto-all+ if( !flag(Tests) )+ buildable: False+