diff --git a/Control/Concurrent/AdvSTM.hs b/Control/Concurrent/AdvSTM.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/AdvSTM.hs @@ -0,0 +1,253 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.AdvSTM +-- Copyright : (c) Chris Kuklewicz 2006, Peter Robinson 2009 +-- License : BSD3 +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Extends Control.Concurrent.STM with IO hooks +-- +----------------------------------------------------------------------------- + +module Control.Concurrent.AdvSTM( -- * Class MonadAdvSTM + MonadAdvSTM(..) + -- * Monad AdvSTM + , AdvSTM + , unsafeRetryWith + , atomically + , unsafeIOToAdvSTM + , debugAdvSTM + , debugMode + ) where + +import Prelude hiding (catch) +import Control.Monad.AdvSTM.Def(AdvSTM(..),Env(..),TVarValue(..)) +import Control.Monad.AdvSTM.Class(MonadAdvSTM(..)) + +import Control.Exception(throw,catch,SomeException,fromException,try) +import Control.Monad(mplus,when) +import Control.Monad.Reader(MonadReader(..),ReaderT,runReaderT,lift,asks) +import Control.Concurrent(threadDelay,forkIO,ThreadId,myThreadId) +import Control.Concurrent.Chan(Chan,newChan,readChan,writeChan) +import Control.Concurrent.STM.TMVar(TMVar,putTMVar,takeTMVar) +-- import Control.Concurrent.STM.TChan(TChan,writeTChan) +import Control.Concurrent.MVar(MVar,newEmptyMVar,takeMVar,tryTakeMVar,putMVar) +import qualified Control.Concurrent.STM as S (STM,orElse,retry,catchSTM,atomically,check,always,alwaysSucceeds) +import Control.Concurrent.STM.TVar(TVar,newTVarIO,readTVar,writeTVar) +import GHC.Conc(unsafeIOToSTM) +import Data.IORef(newIORef,readIORef,writeIORef) + +-------------------------------------------------------------------------------- + + +instance MonadAdvSTM AdvSTM where + onCommit ioaction = do + commitVar <- AdvSTM $ asks commitTVar + commitFun <- liftAdv $ readTVar commitVar + liftAdv $ writeTVar commitVar $ commitFun . (ioaction >>) + + unsafeOnRetry ioaction = do + retryVar <- AdvSTM $ asks retryMVar + liftAdv . unsafeIOToSTM $ do + may'retryFun <- tryTakeMVar retryVar + let retryFun = maybe (ioaction >>) (. (ioaction >>)) may'retryFun + putMVar retryVar $! retryFun + + orElse = mplus + + retry = liftAdv S.retry + + check = liftAdv . S.check + + + alwaysSucceeds inv = unlift inv >>= liftAdv . S.alwaysSucceeds + + always inv = unlift inv >>= liftAdv . S.always + + runAtomic = atomically + + catchSTM action handler = do + action' <- unlift action + handler' <- unlift1 handler + let handler'' = \e -> case fromException e of + Nothing -> throw e + Just e' -> handler' e' + liftAdv $ S.catchSTM action' (handler'') + + liftAdv = AdvSTM . lift + +-- | Adds the IO action to the retry queue and then retries the transaction +unsafeRetryWith :: (Monad m, MonadAdvSTM m) => IO () -> m b +unsafeRetryWith io = unsafeOnRetry io >> retry + + + +-------------------------------------------------------------------------------- + + + +atomically :: AdvSTM a -> IO a +atomically (AdvSTM action) = do + let debugging = False -- Switching this to True occasionally causes deadlocks (b/c unsafeIO)! + debugTVar <- newTVarIO debugging + tid <- myThreadId -- ThreadId for controlling TVar access + debug "***************************************************************" 0 debugging + debug (show (tid,"Starting transaction...")) 1000000 debugging + -- Building the Reader monad environment + commitVar <- newTVarIO id -- IO actions to be run if the transaction commits + retryVar <- newEmptyMVar -- full if there's something todo before retrying + commitListeners <- newTVarIO [] + let env = Env { commitTVar = commitVar :: TVar (IO () -> IO ()) + , retryMVar = retryVar :: MVar (IO () -> IO ()) + , transThreadId = tid :: ThreadId + , listeners = commitListeners :: TVar [(TMVar (),TVarValue)] + , debugModeVar = debugTVar + } + -- Setting up communication for the retry-helper thread: + retryChanVar <- newIORef (Nothing :: Maybe (Chan (Maybe (IO ())))) + retryEndVar <- newEmptyMVar -- Termination signal for the retry-helper thread + + let check'retry = do + unsafeIOToSTM $ do + may'todo <- tryTakeMVar retryVar + case may'todo of + Nothing -> return () + Just retryFun -> do + may'chan <- readIORef retryChanVar + chan <- case may'chan of + Nothing -> do + chan <- newChan + writeIORef retryChanVar (Just chan) + spawn'retry'thread (readChan chan) (putMVar retryEndVar ()) + return chan + Just chan -> return chan + writeChan chan $ Just (retryFun (return())) + debugSTM (show (tid,"Calling retry now...")) 0 debugging + debugSTM (show (tid,"*********************")) 1000000 debugging + S.retry + + let wait'retry'finished = do + may'chan <- readIORef retryChanVar + case may'chan of + Nothing -> return () + Just chan -> do + -- Write an "EOF" on the channel and block until the helper thread is done: + writeChan chan Nothing + takeMVar retryEndVar + + let wrappedAction = runReaderT action env `S.orElse` (check'retry) + + result <- S.atomically $ do + debugSTM (show (tid,"wrappedAction: Running S.STM action...")) 0 debugging + result <- wrappedAction + debugSTM (show (tid,"wrappedAction: Finished S.STM action...")) 0 debugging + ls <- readTVar commitListeners + -- Notify the TPVars that we're entering onCommit mode: + debugSTM (show (tid,"wrappedAction: Notifying TVars that we're about to run onCommit...")) 0 debugging + mapM_ (\(l,_) -> + takeTMVar l -- tell the TVar that we're going into onCommit mode + ) ls + return result + + let rollbackOnCommit = do + debug "rollbackOnCommit: rolling back modified TVar values!" 0 debugging + wait'retry'finished + S.atomically $ do + ls <- readTVar commitListeners + mapM_ (\(l,TVarValue (oldValTVar,oldVal)) -> do + writeTVar oldValTVar oldVal -- smthg went wrong, restore the old value + putTMVar l () -- ...and unblock the TVar + ) ls + + -- Now try to run the onCommit IO actions: + commitFun <- S.atomically $ readTVar commitVar + commitFun (return ()) `catch` (\(e::SomeException) -> rollbackOnCommit >> throw e) + + -- Everything's ok, so notify and unblock the TVars: + debug (show (tid,"*********************")) 1000000 debugging + debug (show (tid,"Notifying TPVars that we're done:")) 0 debugging + S.atomically $ do + ls <- readTVar commitListeners + mapM_ (\(l,_) -> do + putTMVar l () + ) ls + wait'retry'finished + debug (show (tid,"Transaction done; retry thread finished")) 1000000 debugging + debug "************************************************************" 0 debugging + return result + + where + -- Helper thread for the retry IO-actions + spawn'retry'thread :: IO (Maybe (IO ())) -> IO () -> IO ThreadId + spawn'retry'thread nextJob atEndAction = forkIO $ loop + where loop = do + may'job <- nextJob + case may'job of + Nothing -> atEndAction + Just job -> do + res <- try job + case res of + Left (e::SomeException) -> throw e + Right _ -> loop + +-------------------------------------------------------------------------------- + +unsafeIOToAdvSTM :: IO a -> AdvSTM a +unsafeIOToAdvSTM = liftAdv . unsafeIOToSTM + +-------------------------------------------------------------------------------- + +-- | Switches the debug mode on or off +-- /WARNING:/ Can lead to deadlocks! +debugMode :: Bool -> AdvSTM () +debugMode switch = do + debVar <- AdvSTM $ asks debugModeVar + liftAdv $ writeTVar debVar switch + +-- | Uses unsafeIOToSTM to output the Thread Id and a message and delays +-- for a given number of ms +-- /WARNING:/ Can lead to deadlocks! +debugAdvSTM :: String -> Int -> AdvSTM () +debugAdvSTM msg delay = do + debVar <- AdvSTM $ asks debugModeVar + debugging <- liftAdv $ readTVar debVar + tid <- AdvSTM $ asks transThreadId + liftAdv $ debugSTM (show (tid,msg)) delay debugging + +-------------------------------------------------------------------------------- +-- The following functions are not exported as they aren't needed by +-- the enduser + +runWith :: Env -> AdvSTM t -> S.STM t +runWith env (AdvSTM action) = runReaderT action env + + +unlifter :: AdvSTM (AdvSTM a -> S.STM a) +unlifter = do + env <- AdvSTM ask + return (runWith env) + +unlift :: AdvSTM a -> AdvSTM (S.STM a) +unlift f = do + u <- unlifter + return (u f) + +unlift1 :: (t -> AdvSTM a) -> AdvSTM (t -> S.STM a) +unlift1 f = do + u <- unlifter + return (\x -> u (f x)) + +-- WARNING: Can lead to deadlocks! +debugSTM :: String -> Int -> Bool -> S.STM () +debugSTM msg delay debugging = + when (debugging) $ unsafeIOToSTM $ putStrLn msg >> threadDelay delay + +-- WARNING: Can lead to deadlocks! +debug :: String -> Int -> Bool -> IO () +debug msg delay debugging = when (debugging) $ putStrLn msg >> threadDelay delay + + + diff --git a/Control/Concurrent/AdvSTM/TArray.hs b/Control/Concurrent/AdvSTM/TArray.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/AdvSTM/TArray.hs @@ -0,0 +1,46 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.AdvSTM.TArray +-- Copyright : (c) Peter Robinson 2009, The University of Glasgow 2004 +-- License : BSD-style (see the file LICENSE) +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Corresponds to "Control.Concurrent.STM.TArray" +-- +----------------------------------------------------------------------------- + + + +module Control.Concurrent.AdvSTM.TArray( TArray ) +where +import Control.Concurrent.AdvSTM +import Control.Concurrent.AdvSTM.TVar + + +import Control.Monad (replicateM) +import Data.Array (Array, bounds) +import Data.Array.Base (listArray, arrEleBottom, unsafeAt, MArray(..), IArray(numElements)) +import Data.Ix (rangeSize) + + +-- |TArray is a transactional array, supporting the usual 'MArray' +-- interface for mutable arrays. + + +newtype TArray i e = TArray (Array i (TVar e)) + +instance MArray TArray e AdvSTM where + getBounds (TArray a) = return (bounds a) + newArray b e = do + a <- replicateM (rangeSize b) (newTVar e) + return $ TArray (listArray b a) + newArray_ b = do + a <- replicateM (rangeSize b) (newTVar arrEleBottom) + return $ TArray (listArray b a) + unsafeRead (TArray a) i = readTVar $ unsafeAt a i + unsafeWrite (TArray a) i e = writeTVar (unsafeAt a i) e + getNumElements (TArray a) = return (numElements a) + diff --git a/Control/Concurrent/AdvSTM/TChan.hs b/Control/Concurrent/AdvSTM/TChan.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/AdvSTM/TChan.hs @@ -0,0 +1,97 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.AdvSTM.TChan +-- Copyright : (c) Peter Robinson 2009, The University of Glasgow 2004 +-- License : BSD-style (see the file LICENSE) +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Corresponds to "Control.Concurrent.STM.TChan" +-- +----------------------------------------------------------------------------- +module Control.Concurrent.AdvSTM.TChan( TChan + , newTChan + , newTChanIO + , readTChan + , writeTChan + , dupTChan + , unGetTChan + , isEmptyTChan + ) +where +import Control.Concurrent.AdvSTM(AdvSTM,retry) +import Control.Concurrent.AdvSTM.TVar(TVar,readTVar,writeTVar,newTVar,newTVarIO) + +-- | 'TChan' is an abstract type representing an unbounded FIFO channel. +data TChan a = TChan (TVar (TVarList a)) (TVar (TVarList a)) + +type TVarList a = TVar (TList a) +data TList a = TNil | TCons a (TVarList a) + +-- |Build and returns a new instance of 'TChan' +newTChan :: AdvSTM (TChan a) +newTChan = do + hole <- newTVar TNil + read <- newTVar hole + write <- newTVar hole + return (TChan read write) + +-- |@IO@ version of 'newTChan'. This is useful for creating top-level +-- 'TChan's using 'System.IO.Unsafe.unsafePerformIO', because using +-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't +-- possible. +newTChanIO :: IO (TChan a) +newTChanIO = do + hole <- newTVarIO TNil + read <- newTVarIO hole + write <- newTVarIO hole + return (TChan read write) + +-- |Write a value to a 'TChan'. +writeTChan :: TChan a -> a -> AdvSTM () +writeTChan (TChan _read write) a = do + listend <- readTVar write -- listend == TVar pointing to TNil + new_listend <- newTVar TNil + writeTVar listend (TCons a new_listend) + writeTVar write new_listend + +-- |Read the next value from the 'TChan'. +readTChan :: TChan a -> AdvSTM a +readTChan (TChan read _write) = do + listhead <- readTVar read + head <- readTVar listhead + case head of + TNil -> retry + TCons a tail -> do + writeTVar read tail + return a + +-- |Duplicate a 'TChan': the duplicate channel begins empty, but data written to +-- either channel from then on will be available from both. Hence this creates +-- a kind of broadcast channel, where data written by anyone is seen by +-- everyone else. +dupTChan :: TChan a -> AdvSTM (TChan a) +dupTChan (TChan read write) = do + hole <- readTVar write + new_read <- newTVar hole + return (TChan new_read write) + +-- |Put a data item back onto a channel, where it will be the next item read. +unGetTChan :: TChan a -> a -> AdvSTM () +unGetTChan (TChan read _write) a = do + listhead <- readTVar read + newhead <- newTVar (TCons a listhead) + writeTVar read newhead + +-- |Returns 'True' if the supplied 'TChan' is empty. +isEmptyTChan :: TChan a -> AdvSTM Bool +isEmptyTChan (TChan read write) = do + listhead <- readTVar read + head <- readTVar listhead + case head of + TNil -> return True + TCons _ _ -> return False + + diff --git a/Control/Concurrent/AdvSTM/TMVar.hs b/Control/Concurrent/AdvSTM/TMVar.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/AdvSTM/TMVar.hs @@ -0,0 +1,143 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.AdvSTM.TMVar +-- Copyright : (c) Peter Robinson 2009, (c) The University of Glasgow 2004 +-- License : BSD3 +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- TMVar: Transactional MVars, for use in the AdvAdvSTM monad +-- +-- Corresponds to "Control.Concurrent.STM.TMVar" +-- +----------------------------------------------------------------------------- + +module Control.Concurrent.AdvSTM.TMVar ( + -- * TVars + TMVar, + newTMVar, + newEmptyTMVar, + newTMVarIO, + newEmptyTMVarIO, + takeTMVar, + putTMVar, + readTMVar, + swapTMVar, + tryTakeTMVar, + tryPutTMVar, + isEmptyTMVar + ) where + +import Control.Concurrent.AdvSTM.TVar(TVar,readTVar,writeTVar,newTVar,newTVarIO) +import Control.Concurrent.AdvSTM(AdvSTM,retry) + +newtype TMVar a = TMVar (TVar (Maybe a)) +{- ^ +A 'TMVar' is a synchronising variable, used +for communication between concurrent threads. It can be thought of +as a box, which may be empty or full. +-} + +-- |Create a 'TMVar' which contains the supplied value. +newTMVar :: a -> AdvSTM (TMVar a) +newTMVar a = do + t <- newTVar (Just a) + return (TMVar t) + +-- |@IO@ version of 'newTMVar'. This is useful for creating top-level +-- 'TMVar's using 'System.IO.Unsafe.unsafePerformIO', because using +-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't +-- possible. +newTMVarIO :: a -> IO (TMVar a) +newTMVarIO a = do + t <- newTVarIO (Just a) + return (TMVar t) + +-- |Create a 'TMVar' which is initially empty. +newEmptyTMVar :: AdvSTM (TMVar a) +newEmptyTMVar = do + t <- newTVar Nothing + return (TMVar t) + +-- |@IO@ version of 'newEmptyTMVar'. This is useful for creating top-level +-- 'TMVar's using 'System.IO.Unsafe.unsafePerformIO', because using +-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't +-- possible. +newEmptyTMVarIO :: IO (TMVar a) +newEmptyTMVarIO = do + t <- newTVarIO Nothing + return (TMVar t) + +-- |Return the contents of the 'TMVar'. If the 'TMVar' is currently +-- empty, the transaction will 'retry'. After a 'takeTMVar', +-- the 'TMVar' is left empty. +takeTMVar :: TMVar a -> AdvSTM a +takeTMVar (TMVar t) = do + m <- readTVar t + case m of + Nothing -> retry + Just a -> do writeTVar t Nothing; return a + +-- | A version of 'takeTMVar' that does not 'retry'. The 'tryTakeTMVar' +-- function returns 'Nothing' if the 'TMVar' was empty, or @'Just' a@ if +-- the 'TMVar' was full with contents @a@. After 'tryTakeTMVar', the +-- 'TMVar' is left empty. +tryTakeTMVar :: TMVar a -> AdvSTM (Maybe a) +tryTakeTMVar (TMVar t) = do + m <- readTVar t + case m of + Nothing -> return Nothing + Just a -> do writeTVar t Nothing; return (Just a) + +-- |Put a value into a 'TMVar'. If the 'TMVar' is currently full, +-- 'putTMVar' will 'retry'. +putTMVar :: TMVar a -> a -> AdvSTM () +putTMVar (TMVar t) a = do + m <- readTVar t + case m of + Nothing -> do writeTVar t (Just a); return () + Just _ -> retry + +-- | A version of 'putTMVar' that does not 'retry'. The 'tryPutTMVar' +-- function attempts to put the value @a@ into the 'TMVar', returning +-- 'True' if it was successful, or 'False' otherwise. +tryPutTMVar :: TMVar a -> a -> AdvSTM Bool +tryPutTMVar (TMVar t) a = do + m <- readTVar t + case m of + Nothing -> do writeTVar t (Just a); return True + Just _ -> return False + +{-| + This is a combination of 'takeTMVar' and 'putTMVar'; ie. it takes the value + from the 'TMVar', puts it back, and also returns it. +-} +readTMVar :: TMVar a -> AdvSTM a +readTMVar (TMVar t) = do + m <- readTVar t + case m of + Nothing -> retry + Just a -> return a + +-- |Swap the contents of a 'TMVar' for a new value. +swapTMVar :: TMVar a -> a -> AdvSTM a +swapTMVar (TMVar t) new = do + m <- readTVar t + case m of + Nothing -> retry + Just old -> do writeTVar t (Just new); return old + +-- |Check whether a given 'TMVar' is empty. +-- +-- Notice that the boolean value returned is just a snapshot of +-- the state of the 'TMVar'. By the time you get to react on its result, +-- the 'TMVar' may have been filled (or emptied) - so be extremely +-- careful when using this operation. Use 'tryTakeTMVar' instead if possible. +isEmptyTMVar :: TMVar a -> AdvSTM Bool +isEmptyTMVar (TMVar t) = do + m <- readTVar t + case m of + Nothing -> return True + Just _ -> return False diff --git a/Control/Concurrent/AdvSTM/TVar.hs b/Control/Concurrent/AdvSTM/TVar.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/AdvSTM/TVar.hs @@ -0,0 +1,112 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.AdvSTM.TVar +-- Copyright : (c) Peter Robinson 2009 +-- License : BSD-style (see the file LICENSE) +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- +-- +----------------------------------------------------------------------------- + +module Control.Concurrent.AdvSTM.TVar( -- * TVars + TVar + , newTVar + , newTVarIO + , readTVar + , writeTVar + ) +where +import qualified Control.Concurrent.STM.TVar as OldTVar +import qualified Control.Concurrent.STM as STM +import qualified Control.Concurrent.STM.TMVar as OldTMVar +import qualified Control.Concurrent.STM.TChan as OldTChan +import Control.Exception(throw,Deadlock(Deadlock)) +import Control.Concurrent.AdvSTM(liftAdv,orElse,retry) +import Control.Monad.AdvSTM.Def(AdvSTM(AdvSTM),transThreadId,listeners,TVarValue(TVarValue)) +import Control.Concurrent(ThreadId) +import Control.Monad(liftM,ap,unless) +import Control.Monad.Reader(asks) +import Data.Maybe(isJust,Maybe,fromJust) +import qualified Data.Set as S + +-------------------------------------------------------------------------------- + +data TVar a = TVar + { valueTVar :: OldTVar.TVar a + , onCommitLock :: OldTMVar.TMVar () + , currentTid :: OldTVar.TVar (Maybe ThreadId) + } + +-------------------------------------------------------------------------------- + +-- | See 'OldTVar.newTVar' +newTVar :: a -> AdvSTM (TVar a) +newTVar a = TVar `liftM` (liftAdv $ OldTVar.newTVar a) + `ap` (liftAdv $ OldTMVar.newTMVar ()) + `ap` (liftAdv $ OldTVar.newTVar Nothing) + +-- | See 'OldTVar.newTVarIO' +newTVarIO :: a -> IO (TVar a) +newTVarIO a = TVar `liftM` OldTVar.newTVarIO a + `ap` OldTMVar.newTMVarIO () + `ap` OldTVar.newTVarIO Nothing + +-------------------------------------------------------------------------------- + +-- | Writes a value to a TVar. Blocks until the onCommit IO-action(s) are +-- complete. See 'onCommit' for details. +writeTVar :: TVar a -> a -> AdvSTM () +writeTVar tvar a = do + commitLock <- liftAdv $ OldTMVar.tryTakeTMVar (onCommitLock tvar) + -- Get ThreadID of current transaction: + curTid <- AdvSTM $ asks transThreadId + storedTid <- liftAdv $ OldTVar.readTVar (currentTid tvar) + case commitLock of + Nothing -> do + if isJust storedTid && fromJust storedTid == curTid + then throw Deadlock -- No transaction during onCommit-phase! + else retry + Just _ -> do + unless (isJust storedTid && (fromJust storedTid == curTid)) $ do + -- First write access, update the ThreadId: + liftAdv $ OldTVar.writeTVar (currentTid tvar) $ Just curTid + -- Add this TVar to the onCommit-listener list: + lsTVar <- AdvSTM $ asks listeners + ls <- liftAdv $ OldTVar.readTVar lsTVar + -- Remember the old value for rollback: + oldval <- liftAdv $ OldTVar.readTVar (valueTVar tvar) + liftAdv $ OldTVar.writeTVar lsTVar $ + (onCommitLock tvar,TVarValue (valueTVar tvar,oldval)) : ls + + liftAdv $ OldTVar.writeTVar (valueTVar tvar) a + liftAdv $ OldTMVar.putTMVar (onCommitLock tvar) () + + +-------------------------------------------------------------------------------- + +-- | Reads a value from a TVar. Blocks until the IO onCommit action(s) of +-- the corresponding transaction are complete. +-- See 'onCommit' for a more detailed description of this behaviour. +readTVar :: TVar a -> AdvSTM a +readTVar tvar = do + commitLock <- liftAdv $ OldTMVar.tryTakeTMVar (onCommitLock tvar) + case commitLock of + Nothing -> do + storedTid <- liftAdv $ OldTVar.readTVar (currentTid tvar) + curTid <- AdvSTM $ asks transThreadId + if isJust storedTid && fromJust storedTid == curTid + then throw Deadlock + else retry + Just _ -> do + result <- liftAdv $ OldTVar.readTVar $ valueTVar tvar + liftAdv $ OldTMVar.putTMVar (onCommitLock tvar) () + return result + +-------------------------------------------------------------------------------- + + + diff --git a/Control/Monad/AdvSTM.hs b/Control/Monad/AdvSTM.hs new file mode 100644 --- /dev/null +++ b/Control/Monad/AdvSTM.hs @@ -0,0 +1,82 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Monad.AdvSTM +-- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008 +-- License : BSD3 +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Provides the type class MonadAdvSTM and the AdvSTM monad. +-- Parts of this implementation were taken from the HaskellWiki Page of +-- MonadAdvSTM (see package description). +----------------------------------------------------------------------------- + + +module Control.Monad.AdvSTM( MonadAdvSTM(..) + , AdvSTM + ) +where +import Control.Monad.AdvSTM.Def +import Control.Monad.AdvSTM.Class + +import Control.Exception(Exception) +import qualified Control.Concurrent.STM as S +import Control.Concurrent.STM.TVar(TVar) +import Control.Concurrent.STM.TMVar(TMVar) +import Control.Concurrent(MVar,ThreadId) +import Control.Monad(Monad,MonadPlus) +import Control.Monad.Reader(MonadReader,ReaderT) + +{- +-- | A type class for extended-STM monads. See 'AdvSTM' for a concrete +-- instantiation +class Monad m => MonadAdvSTM m where + -- | Takes an IO action that will be executed if the STM transaction commits. + -- /Beware:/ If the IO action throws an exception, the effects of the STM + -- action will still be visible. + -- + -- /Atomicity and Race Conditions:/ + -- + -- * 'TVar', 'TChan', 'TArray' or 'TMVar': There is a race hole between the TM transaction + -- and the execution of the IO action, i.e., other + -- transactions might commit before the IO action is run. + -- + -- * 'TTVar': When a 'TTVar' was modified in the TM action, it cannot be accessed by any + -- other thread before the onCommit IO action was run. Forking + -- another thread inside of the onCommit IO action breaks this atomicity + -- guarantee. + onCommit :: IO a -> m () + + -- | Adds the IO action to the retry-queue. If the transaction retries, + -- a new thread is forked that runs the retry actions. When this thread is + -- done, the transaction retries. + onRetry :: IO a -> m () + + -- | See 'S.orElse' + orElse :: m a -> m a -> m a + + -- | See 'S.retry' + retry :: m a + + -- | Runs a transaction atomically in the 'IO' monad. + runAtomic :: m a -> IO a + + -- | See 'S.catchSTM' + catchSTM :: Exception e => m a -> (e -> m a) -> m a + + -- | Lifts STM actions to 'MonadAdvSTM'. + liftAdv :: S.STM a -> m a +-} +-------------------------------------------------------------------------------- +{- +-- | Drop-in replacement for the STM monad +newtype AdvSTM a = AdvSTM (ReaderT Env S.STM a) + deriving ( Functor + , Monad + , MonadPlus +-- , MonadReader Env + ) + +-} diff --git a/Control/Monad/AdvSTM/Class.hs b/Control/Monad/AdvSTM/Class.hs new file mode 100644 --- /dev/null +++ b/Control/Monad/AdvSTM/Class.hs @@ -0,0 +1,85 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Monad.AdvSTM.Class +-- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008 +-- License : BSD3 +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Provides the type class MonadAdvSTM +-- Parts of this implementation were taken from the HaskellWiki Page of +-- MonadAdvSTM (see package description). +----------------------------------------------------------------------------- + +module Control.Monad.AdvSTM.Class( MonadAdvSTM(..)) +where + +import Control.Exception(Exception,Deadlock) +import qualified Control.Concurrent.STM as S +import Control.Monad(Monad) +import Control.Monad.AdvSTM.Def(AdvSTM) + + + +-- | A type class for extended-STM monads. For a concrete instantiation see +-- 'AdvSTM' +class Monad m => MonadAdvSTM m where + + -- | Takes an IO action that will be executed iff the transaction commits. + -- + -- * When a TVar was modified in a transaction and this transaction commits, + -- this update remains invisible to other threads until the corresponding + -- onCommit action was run. + -- + -- * If the onCommit action throws an exception, the original value of + -- the TVars will be restored. + -- + -- * Accessing a modified TVar within the onCommit action will cause a + -- 'Deadlock' exception to be thrown. + -- + -- As a general rule, 'onCommit' should + -- only be used for \"real\" (i.e. without atomic blocks) IO actions and is certainly + -- not the right place to fiddle with TVars. For example, if you wanted to + -- write a TVar value to a file on commit, you could write: + -- + -- > tvar <- newTVarIO "bla" + -- > atomically $ do + -- > x <- readTVar tvar + -- > onCommit (writeFile "myfile" x) + -- + -- + onCommit :: IO () -> m () + + -- | Adds an IO action to the retry job-queue. If the transaction retries, + -- a new helper thread is forked that runs the retry actions, and, after the helper + -- thread is done, the transaction retries. + -- /Warning:/ Uses unsafeIOToSTM to fork a helper thread that runs the retry + -- actions! + unsafeOnRetry :: IO () -> m () + + -- | See 'S.orElse' + orElse :: m a -> m a -> m a + + -- | See 'S.retry'. Skips any IO actions added by unsafeOnRetry. + retry :: m a + + -- | See 'S.check' + check :: Bool -> m () + + -- | See 'S.alwaysSucceeds' + alwaysSucceeds :: m a -> m () + + -- | See 'S.always' + always :: m Bool -> m () + + -- | Runs a transaction atomically in the 'IO' monad. + runAtomic :: m a -> IO a + + -- | See 'S.catchSTM' + catchSTM :: Exception e => m a -> (e -> m a) -> m a + + -- | Lifts STM actions to 'MonadAdvSTM'. + liftAdv :: S.STM a -> m a + diff --git a/Control/Monad/AdvSTM/Def.hs b/Control/Monad/AdvSTM/Def.hs new file mode 100644 --- /dev/null +++ b/Control/Monad/AdvSTM/Def.hs @@ -0,0 +1,47 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Monad.AdvSTM.Def +-- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008 +-- License : BSD3 +-- +-- Maintainer : Peter Robinson +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- +-- This is an internal module. +----------------------------------------------------------------------------- + +module Control.Monad.AdvSTM.Def(AdvSTM(..),Env(..),TVarValue(..)) +where + +import qualified Control.Concurrent.STM as S +import Control.Concurrent.STM.TVar(TVar) +import Control.Concurrent.STM.TMVar(TMVar) +import Control.Concurrent(MVar,ThreadId) +import Control.Monad(Monad,MonadPlus) +import Control.Monad.Reader(ReaderT) + +-- | Drop-in replacement for the STM monad +newtype AdvSTM a = AdvSTM (ReaderT Env S.STM a) + deriving ( Functor + , Monad + , MonadPlus + ) + + + +-- | The environment used for the Reader Monad +data Env = Env { commitTVar :: TVar (IO () -> IO ()) -- the commit action(s) + , retryMVar :: MVar (IO () -> IO ()) -- the retry action(s) + , transThreadId :: ThreadId -- the current ThreadId + , listeners :: TVar [(TMVar (),TVarValue)] -- ,TVar (Maybe ThreadId),TChan (Maybe ThreadId))] + -- Contains communication facilities for modified TVars: + -- [(tVar-Lock,Old-tVar-Value,Maybe (current ThreadId) + -- ,death-notice channel)] + , debugModeVar :: TVar Bool +-- , isInRetryMode :: TVar Bool + } + +data TVarValue = forall a. TVarValue ((TVar a),a) + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,90 @@ +This library is derived from code from several sources: + + * Code from the STM library which is distributable under + the Glasgow Haskell Compiler license (see below) + + * Code from the Haskell Wiki which is licensed under the + simple permissive license (see below) + +The full text of these licenses is reproduced below. All of the +licenses are BSD-style or compatible. + +-------------------------------------------------------------------------------- + +Copyright (c) 2009, Peter Robinson +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. + * The names of its contributors + * may not 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. + +-------------------------------------------------------------------------------- + +The Glasgow Haskell Compiler License + +Copyright 2004, The University Court of the University of Glasgow. +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 name of the University nor the names of its contributors may be +used to endorse or promote products derived from this software without +specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE UNIVERSITY COURT OF THE UNIVERSITY OF +GLASGOW AND THE 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 +UNIVERSITY COURT OF THE UNIVERSITY OF GLASGOW OR THE 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. + + +-------------------------------------------------------------------------------- + +Permission is hereby granted, free of charge, to any person obtaining this work +(the "Work"), to deal in the Work without restriction, including without +limitation the rights to use, copy, modify, merge, publish, distribute, +sublicense, and/or sell copies of the Work, and to permit persons to whom the +Work is furnished to do so. + +THE WORK IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +CONNECTION WITH THE WORK OR THE USE OR OTHER DEALINGS IN THE WORK. + + diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/stm-io-hooks.cabal b/stm-io-hooks.cabal new file mode 100644 --- /dev/null +++ b/stm-io-hooks.cabal @@ -0,0 +1,49 @@ +Name: stm-io-hooks +Synopsis: An STM monad with IO hooks +Description: + This library provides an STM monad with commit and retry IO hooks. + A retry-action is run (once) if the transaction retries, while commit-actions are + executed iff the transaction commits. The library also gives some atomicity + guarantees for commit-actions, see documentation for details. + . + Note: Part of this library uses code from + the Haskell Wiki (see ). +Category: Concurrency +Author: Peter Robinson 2009, Chris Kuklewicz 2006 +Maintainer: Peter Robinson +License: BSD3 +License-file: LICENSE +Homepage: http://darcs.monoid.at/stm-io-hooks +Version: 0.0.1 + +Build-type: Simple +Cabal-Version: >= 1.2.3 + +library + Exposed-Modules: Control.Monad.AdvSTM + ,Control.Concurrent.AdvSTM + ,Control.Concurrent.AdvSTM.TMVar + ,Control.Concurrent.AdvSTM.TVar + ,Control.Concurrent.AdvSTM.TArray + ,Control.Concurrent.AdvSTM.TChan + + Other-Modules: Control.Monad.AdvSTM.Def + ,Control.Monad.AdvSTM.Class + -- TODO: add Sample module! + + build-depends: base >= 4.0.0.0 + ,stm >= 2.1.1.2 + ,array + ,containers >= 0.2.0.0 + ,mtl >= 1.1.0.2 + + extensions: MultiParamTypeClasses + ,FunctionalDependencies + ,FlexibleInstances + ,GeneralizedNewtypeDeriving + ,ScopedTypeVariables + ,DeriveDataTypeable + ,RankNTypes + ,ExistentialQuantification + +