stm-io-hooks 0.4.2 → 1.1.2
raw patch · 23 files changed
Files
- Control/Concurrent/AdvSTM.hs +0/−335
- Control/Concurrent/AdvSTM/TArray.hs +0/−46
- Control/Concurrent/AdvSTM/TChan.hs +0/−97
- Control/Concurrent/AdvSTM/TMVar.hs +0/−144
- Control/Concurrent/AdvSTM/TVar.hs +0/−110
- Control/Concurrent/AdvSTM/TVar.hs-boot +0/−43
- Control/Monad/AdvSTM.hs +0/−31
- Control/Monad/AdvSTM/Class.hs +0/−240
- Control/Monad/AdvSTM/Def.hs +0/−47
- LICENSE +21/−81
- README.md +11/−0
- Setup.hs +2/−0
- Setup.lhs +0/−3
- src/Control/Concurrent/AdvSTM.hs +302/−0
- src/Control/Concurrent/AdvSTM/TArray.hs +47/−0
- src/Control/Concurrent/AdvSTM/TChan.hs +95/−0
- src/Control/Concurrent/AdvSTM/TMVar.hs +144/−0
- src/Control/Concurrent/AdvSTM/TVar.hs +113/−0
- src/Control/Monad/AdvSTM.hs +31/−0
- src/Control/Monad/AdvSTM/Class.hs +272/−0
- src/Control/Monad/AdvSTM/Def.hs +61/−0
- src/Lib.hs +6/−0
- stm-io-hooks.cabal +39/−55
− Control/Concurrent/AdvSTM.hs
@@ -1,335 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Concurrent.AdvSTM--- Copyright : (c) Chris Kuklewicz 2006, Peter Robinson 2009--- License : BSD3--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- Stability : experimental--- Portability : non-portable (requires STM)------ Extends Control.Concurrent.STM with IO hooks--- ------------------------------------------------------------------------------- -module Control.Concurrent.AdvSTM( -- * Class MonadAdvSTM- MonadAdvSTM( onCommit- , onRetry- , orElse- , retry- , check- , alwaysSucceeds- , always--- , runAtomic- , catchSTM- , liftAdv- , newTVar- , readTVar- , writeTVar- )- - -- * Monad AdvSTM- , AdvSTM- , retryWith- , atomically- , unsafeIOToAdvSTM- , handleSTM- , debugAdvSTM- , debugMode- ) where- -import Prelude hiding (catch)-import Control.Monad.AdvSTM.Def(AdvSTM(..),Env(..),TVarValue(..))-import Control.Monad.AdvSTM.Class( - MonadAdvSTM(..), handleSTM, TVar( TVar ), onCommitLock, currentTid, valueTVar )--- import Control.Monad.Reader(ReaderT(ReaderT),mapReaderT,runReaderT)--import Control.Exception(Exception,throw,catch,SomeException,fromException,try,block,Deadlock(..)) -import Control.Monad(mplus,when,liftM,ap,unless)-import Control.Monad.Error(MonadError(..))-import Control.Concurrent(threadDelay,forkIO,ThreadId,myThreadId,throwTo)-import Control.Concurrent.Chan(Chan,newChan,readChan,writeChan)-import Control.Concurrent.STM.TMVar(TMVar,putTMVar,takeTMVar,newTMVar,tryTakeTMVar)--- 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 qualified Control.Concurrent.STM.TVar as STVar -- (TVar,newTVarIO,readTVar,writeTVar)-import GHC.Conc(unsafeIOToSTM)-import Data.IORef(newIORef,readIORef,writeIORef)-import Data.Maybe(isJust,Maybe,fromJust)-import Control.Monad.Reader(MonadReader(..),ReaderT(ReaderT),runReaderT,lift,asks)- ------------------------------------------------------------------------------------instance MonadAdvSTM AdvSTM where- onCommit ioaction = do- commitVar <- AdvSTM $ asks commitTVar- commitFun <- liftAdv $ STVar.readTVar commitVar- liftAdv $ STVar.writeTVar commitVar $ commitFun . (ioaction >>)-- onRetry 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 --- -- | See 'STVar.newTVar'- newTVar a = TVar `liftM` (liftAdv $ STVar.newTVar a) - `ap` (liftAdv $ newTMVar ()) - `ap` (liftAdv $ STVar.newTVar Nothing)--- -- | Writes a value to a TVar. Blocks until the onCommit IO-action(s) are- -- complete. See 'onCommit' for details.- writeTVar tvar a = do - commitLock <- liftAdv $ tryTakeTMVar (onCommitLock tvar)- -- Get ThreadID of current transaction:- curTid <- AdvSTM $ asks transThreadId - storedTid <- liftAdv $ STVar.readTVar (currentTid tvar) - case commitLock of- Nothing -> - if isJust storedTid && fromJust storedTid == curTid- then throw Deadlock -- Can't write the TVar in the onCommit phase- else retry- Just _ -> do- unless (isJust storedTid && (fromJust storedTid == curTid)) $ do- -- First write access, update the ThreadId:- liftAdv $ STVar.writeTVar (currentTid tvar) $ Just curTid- -- Add this TVar to the onCommit-listener list:- lsTVar <- AdvSTM $ asks listeners- ls <- liftAdv $ STVar.readTVar lsTVar- -- Remember the old value for rollback:- oldval <- liftAdv $ STVar.readTVar (valueTVar tvar)- liftAdv $ STVar.writeTVar lsTVar $- (onCommitLock tvar,TVarValue (valueTVar tvar,oldval)) : ls- - liftAdv $ STVar.writeTVar (valueTVar tvar) a - liftAdv $ 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 = do - commitLock <- liftAdv $ tryTakeTMVar (onCommitLock tvar)- case commitLock of- Nothing -> do- storedTid <- liftAdv $ STVar.readTVar (currentTid tvar) - curTid <- AdvSTM $ asks transThreadId - if isJust storedTid && fromJust storedTid == curTid- then throw Deadlock- else retry- Just _ -> do- result <- liftAdv $ STVar.readTVar $ valueTVar tvar- liftAdv $ putTMVar (onCommitLock tvar) ()- return result-- ------------------------------------------------------------------------------------- | Adds the IO action to the retry queue and then retries the transaction-retryWith :: (Monad m, MonadAdvSTM m) => IO () -> m b-retryWith io = onRetry io >> retry- -- -------------------------------------------------------------------------------------- | See 'S.atomically'-atomically :: AdvSTM a -> IO a-atomically (AdvSTM action) = do- let debugging = False -- Switching this to True occasionally causes deadlocks (b/c unsafeIO)!- debugTVar <- STVar.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 <- STVar.newTVarIO id -- IO actions to be run if the transaction commits- retryVar <- newEmptyMVar -- full if there's something todo before retrying- commitListeners <- STVar.newTVarIO [] - let env = Env { commitTVar = commitVar :: STVar.TVar (IO () -> IO ())- , retryMVar = retryVar :: MVar (IO () -> IO ())- , transThreadId = tid :: ThreadId- , listeners = commitListeners :: STVar.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 ()) tid- 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-- -- Block interruptions from other threads for the rest of 'atomically'- block $ do - 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 <- STVar.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 <- STVar.readTVar commitListeners- mapM_ (\(l,TVarValue (oldValTVar,oldVal)) -> do - STVar.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 $ STVar.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 <- STVar.readTVar commitListeners - mapM_ (\(l,_) -> - 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 () -> ThreadId -> IO ThreadId- spawn'retry'thread nextJob atEndAction mainTId = 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) -> throwTo mainTId e- Right _ -> loop- - -- ------------------------------------------------------------------------------------- | See 'unsafeIOToSTM'-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 $ STVar.writeTVar debVar switch---- | Uses unsafeIOToSTM to output the Thread Id and a message and delays --- for a given number of time.--- /WARNING:/ Can lead to deadlocks! -debugAdvSTM :: String -> Int -> AdvSTM ()-debugAdvSTM msg delay = do - debVar <- AdvSTM $ asks debugModeVar- debugging <- liftAdv $ STVar.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 (u . f)- --- 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--instance (Exception e) => MonadError e AdvSTM where- throwError = throw- catchError = catchSTM
− Control/Concurrent/AdvSTM/TArray.hs
@@ -1,46 +0,0 @@--------------------------------------------------------------------------------- |--- 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 <thaldyron@gmail.com>--- Stability : experimental--- Portability : non-portable (requires STM)------ Corresponds to "Control.Concurrent.STM.TArray" -------------------------------------------------------------------------------------module Control.Concurrent.AdvSTM.TArray( TArray )-where-import Control.Monad.AdvSTM( MonadAdvSTM )-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 MonadAdvSTM m => MArray TArray e m 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)-
− Control/Concurrent/AdvSTM/TChan.hs
@@ -1,97 +0,0 @@--------------------------------------------------------------------------------- |--- 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 <robinson@ecs.tuwien.ac.at>--- 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.Monad.AdvSTM(MonadAdvSTM,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 :: MonadAdvSTM m => m (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 :: MonadAdvSTM m => TChan a -> a -> m ()-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 :: MonadAdvSTM m => TChan a -> m 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 :: MonadAdvSTM m => TChan a -> m (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 :: MonadAdvSTM m => TChan a -> a -> m ()-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 :: MonadAdvSTM m => TChan a -> m Bool-isEmptyTChan (TChan read write) = do- listhead <- readTVar read- head <- readTVar listhead- case head of- TNil -> return True- TCons _ _ -> return False--
− Control/Concurrent/AdvSTM/TMVar.hs
@@ -1,144 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Concurrent.AdvSTM.TMVar--- Copyright : (c) Peter Robinson 2009, (c) The University of Glasgow 2004--- License : BSD3--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- 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.Monad.AdvSTM(MonadAdvSTM,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 :: MonadAdvSTM m => a -> m (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 :: MonadAdvSTM m => m (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 :: MonadAdvSTM m => TMVar a -> m 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 :: MonadAdvSTM m => TMVar a -> m (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 :: MonadAdvSTM m => TMVar a -> a -> m ()-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 :: MonadAdvSTM m => TMVar a -> a -> m 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 :: MonadAdvSTM m => TMVar a -> m 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 :: MonadAdvSTM m => TMVar a -> a -> m 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 :: MonadAdvSTM m => TMVar a -> m Bool-isEmptyTMVar (TMVar t) = do- m <- readTVar t- case m of- Nothing -> return True- Just _ -> return False-
− Control/Concurrent/AdvSTM/TVar.hs
@@ -1,110 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Concurrent.AdvSTM.TVar--- Copyright : (c) Peter Robinson 2009--- License : BSD-style (see the file LICENSE)--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- Stability : experimental--- Portability : non-portable (requires STM)------ -----------------------------------------------------------------------------------module Control.Concurrent.AdvSTM.TVar( -- * TVars- TVar - , newTVar- , newTVarIO- , readTVar- , writeTVar- )-where-import Control.Monad.AdvSTM.Class(readTVar,writeTVar,newTVar,TVar(TVar))-import qualified Control.Concurrent.STM.TVar as OldTVar-import qualified Control.Concurrent.STM.TMVar as OldTMVar-import Control.Monad(liftM,ap)-{--import qualified Control.Concurrent.STM as STM-import qualified Control.Concurrent.STM.TChan as OldTChan-import Control.Exception(throw,Deadlock(Deadlock))-import Control.Concurrent.AdvSTM(liftAdv,orElse,retry,readTVar,writeTVar,newTVar)-import Control.Monad.AdvSTM.Def(AdvSTM(AdvSTM),transThreadId,listeners,TVarValue(TVarValue))-import Control.Concurrent(ThreadId)-import Control.Monad.Reader(asks)-import Data.Maybe(isJust,Maybe,fromJust)-import qualified Data.Set as S--}------------------------------------------------------------------------------------- | See 'OldTVar.newTVarIO'----newTVarIO :: a -> IO (TVar a)-newTVarIO a = TVar `liftM` OldTVar.newTVarIO a- `ap` OldTMVar.newTMVarIO ()- `ap` OldTVar.newTVarIO Nothing-----------------------------------------------------------------------------------{---- | 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)--}-{-------------------------------------------------------------------------------------- | 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------------------------------------------------------------------------------------}--
− Control/Concurrent/AdvSTM/TVar.hs-boot
@@ -1,43 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Concurrent.AdvSTM.TVar--- Copyright : (c) Peter Robinson 2009--- License : BSD-style (see the file LICENSE)--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- Stability : experimental--- Portability : non-portable (requires STM)------ -----------------------------------------------------------------------------------module Control.Concurrent.AdvSTM.TVar( 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)- }-
− Control/Monad/AdvSTM.hs
@@ -1,31 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Monad.AdvSTM--- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008--- License : BSD3--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- 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)-
− Control/Monad/AdvSTM/Class.hs
@@ -1,240 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Monad.AdvSTM.Class--- Copyright : Peter Robinson 2008, HaskellWiki 2006-2007--- License : BSD3--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- 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(..), handleSTM, TVar(TVar), valueTVar, onCommitLock, currentTid)-where--import Control.Exception(Exception,throw)-import qualified Control.Concurrent.STM as S-import qualified Control.Concurrent.STM.TVar as OldTVar-import qualified Control.Concurrent.STM.TMVar as OldTMVar-import Control.Monad(Monad,liftM,ap)-import Control.Monad.Trans(lift)-import Control.Monad.State(StateT(StateT),mapStateT,runStateT,evalStateT)-import Control.Monad.Writer(WriterT(WriterT),mapWriterT,runWriterT,execWriterT)-import Control.Monad.Reader(ReaderT(ReaderT),mapReaderT,runReaderT)--- import Control.Monad.AdvSTM.Def(AdvSTM)-import Control.Concurrent( ThreadId )---import GHC.Conc( unsafeIOToSTM )--- import {-# SOURCE #-} Control.Concurrent.AdvSTM.TVar-import Data.Monoid--data TVar a = TVar - { valueTVar :: OldTVar.TVar a - , onCommitLock :: OldTMVar.TMVar () - , currentTid :: OldTVar.TVar (Maybe ThreadId)- }---- | 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.- -- - -- /Note:/ When the transaction is retried, 'unsafeIOToSTM' is used to fork a - -- helper thread that runs the retry actions (if any). It is your- -- responsibility to ensure that your retry IO-actions are ``safe''. Any- -- exceptions occurring in the retry-thread will be thrown to the- -- thread where the transaction is running and immediately cause the transaction to be- -- aborted, since 'catchSTM' does not catch asynchronous exceptions.- onRetry :: IO () -- ^ IO action that will be run if the transaction is (explicitly) retried.- -> m ()-- -- | See 'S.orElse'- orElse :: m a -> m a -> m a-- -- | Runs any IO actions added by 'onRetry' and then retries the- -- transaction.- retry :: m a-- -- | See 'S.check'- check :: Bool -> m ()-- -- | See 'S.alwaysSucceeds'- alwaysSucceeds :: m a -> m ()-- -- | See 'S.always'- always :: m Bool -> m ()-- -- | See 'S.catchSTM'- catchSTM :: Exception e => m a -> (e -> m a) -> m a--- -- | Lifts STM actions to 'MonadAdvSTM'.- liftAdv :: S.STM a -> m a- - -- | 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 -> m a-- -- | Writes a value to a TVar. Blocks until the onCommit IO-action(s) are- -- complete. See 'onCommit' for details.- writeTVar :: TVar a -> a -> m ()-- -- | See 'OldTVar.newTVar'- newTVar :: a -> m (TVar a)---- -- See 'S.atomically'--- runAtomic :: m a -> IO a--- newTVarIO :: a -> IO (TVar a)- --- | A version of 'catchSTM' with the arguments swapped around.-handleSTM :: (MonadAdvSTM m, Exception e) => (e -> m a) -> m a -> m a-handleSTM = flip catchSTM---------------------------------------------------------------------------------------mapStateT2 :: (m (a, s) -> n (b, s) -> o (c,s)) - -> StateT s m a -> StateT s n b -> StateT s o c-mapStateT2 f m1 m2 = StateT $ \s -> f (runStateT m1 s) (runStateT m2 s)--liftAndSkipStateT f m = StateT $ \s -> let a = evalStateT m s- in do r <- f a- return (r,s)--instance MonadAdvSTM m => MonadAdvSTM (StateT s m) where- onCommit = lift . onCommit -- onRetry = lift . onRetry -- orElse = mapStateT2 orElse-- retry = lift retry-- check = lift . check -- -- Note: The state modifications of the invariant action- -- are thrown away!- alwaysSucceeds = liftAndSkipStateT alwaysSucceeds - always = liftAndSkipStateT always- - catchSTM m h = StateT (\r -> catchSTM (runStateT m r) (\e -> runStateT (h e) r))-- liftAdv = lift . liftAdv - - readTVar = lift . readTVar-- writeTVar tvar = lift . writeTVar tvar-- newTVar = lift . newTVar ------------------------------------------------------------------------------------mapWriterT2 :: (m (a, w) -> n (b, w) -> o (c,w)) - -> WriterT w m a -> WriterT w n b -> WriterT w o c-mapWriterT2 f m1 m2 = WriterT $ f (runWriterT m1) (runWriterT m2)--- mapWriterT2 f m1 m2 = liftM f m1 `ap` m2--evalWriterT :: Monad m => WriterT w m a -> m a-evalWriterT m = do- (a,_) <- runWriterT m- return a--liftAndSkipWriterT :: (Monad m,Monoid w)- => (m a -> m b)- -> WriterT w m a -> WriterT w m b-liftAndSkipWriterT f m = WriterT $ - let a = evalWriterT m- in do r <- f a- return (r,mempty)--instance (MonadAdvSTM m, Monoid w) => MonadAdvSTM (WriterT w m) where- onCommit = lift . onCommit -- onRetry = lift . onRetry -- orElse = mapWriterT2 orElse-- retry = lift retry-- check = lift . check -- -- Note: The writer-log modifications of the invariant action- -- are thrown away!- alwaysSucceeds = liftAndSkipWriterT alwaysSucceeds - always = liftAndSkipWriterT always- - catchSTM m h = WriterT (catchSTM (runWriterT m) (\e -> runWriterT (h e)))-- liftAdv = lift . liftAdv - - readTVar = lift . readTVar-- writeTVar tvar = lift . writeTVar tvar-- newTVar = lift . newTVar ------------------------------------------------------------------------------------mapReaderT2 :: (m a -> n b -> o c) -> ReaderT r m a -> ReaderT r n b -> ReaderT r o c-mapReaderT2 f m1 m2 = ReaderT $ \r -> f (runReaderT m1 r) (runReaderT m2 r) --instance MonadAdvSTM m => MonadAdvSTM (ReaderT r m) where- onCommit = lift . onCommit -- onRetry = lift . onRetry -- orElse = mapReaderT2 orElse-- retry = lift retry-- check = lift . check -- alwaysSucceeds = mapReaderT alwaysSucceeds- always = mapReaderT always-- catchSTM m h = ReaderT (\r -> catchSTM (runReaderT m r) (\e -> runReaderT (h e) r))-- liftAdv = lift . liftAdv - - readTVar = lift . readTVar-- writeTVar tvar = lift . writeTVar tvar-- newTVar = lift . newTVar -----------------------------------------------------------------------------------
− Control/Monad/AdvSTM/Def.hs
@@ -1,47 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Monad.AdvSTM.Def--- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008--- License : BSD3--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- 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,mapReaderT,runReaderT)---- | 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)-
LICENSE view
@@ -1,90 +1,30 @@-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 Peter Robinson (c) 2019 -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.+ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer. -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.+ * 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 Peter Robinson 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.
+ README.md view
@@ -0,0 +1,11 @@+This library provides a Software Transactional Memory (STM) monad with commit and retry IO hooks. A retry-action is run (at least once) if the transaction retries, while comm it-actions are executed iff the transaction commits. The AdvSTM monad also gives some atomicity guarantees for commit-actions:++* When a TVar is modified in a transaction and this transaction commits,+the update remains invisible to other threads until the corresponding+onCommit action is run.++* If the onCommit action throws an exception, the original values of+the modified TVars are restored.++Note: The package can be used as a drop-in replacement for+'Control.Concurrent.STM'.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
− Setup.lhs
@@ -1,3 +0,0 @@-#!/usr/bin/env runhaskell-> import Distribution.Simple-> main = defaultMain
+ src/Control/Concurrent/AdvSTM.hs view
@@ -0,0 +1,302 @@+{-# LANGUAGE ScopedTypeVariables, FlexibleInstances, MultiParamTypeClasses, UndecidableInstances #-}+-----------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.AdvSTM+-- Copyright : (c) Chris Kuklewicz 2006, Peter Robinson 2009+-- License : BSD3+--+-- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- Extends Control.Concurrent.STM with IO hooks+--+-----------------------------------------------------------------------------++module Control.Concurrent.AdvSTM( -- * Class MonadAdvSTM+ MonadAdvSTM( onCommitWith+ , onCommit+ , unsafeRetryWith+ , orElse+ , retry+ , check+ , catchSTM+ , liftAdv+ , newTVar+ , readTVarAsync+ , writeTVarAsync+ , readTVar+ , writeTVar+ )++ -- * Monad AdvSTM+ , AdvSTM+ , atomically+ , unsafeIOToSTM+ , handleSTM+ , debugAdvSTM+ , debugMode+ ) where++import Prelude hiding (catch)+import Control.Monad.AdvSTM.Def(AdvSTM(..),Env(..),TVarValue(..))+import Control.Monad.AdvSTM.Class(+ MonadAdvSTM(..), handleSTM, TVar( TVar ), onCommitLock, currentTid, valueTVar )+-- import Control.Monad.Reader(ReaderT(ReaderT),mapReaderT,runReaderT)++import Control.Exception(Exception,throw,catch,SomeException,fromException,try,mask_,Deadlock(..),finally)+import Control.Monad(mplus,when,liftM,ap,unless)+import Control.Monad.Except+import Control.Concurrent(threadDelay,forkIO,ThreadId,myThreadId,throwTo)+import Control.Concurrent.Chan(Chan,newChan,readChan,writeChan)+import Control.Concurrent.STM.TMVar(TMVar,putTMVar,takeTMVar,newTMVar,tryTakeTMVar)+-- import Control.Concurrent.STM.TChan(TChan,writeTChan)+import Control.Concurrent.MVar(MVar,newMVar,takeMVar,tryTakeMVar,putMVar,tryPutMVar,swapMVar)+import qualified Control.Concurrent.STM as S (STM,orElse,retry,catchSTM,atomically,check)+import qualified Control.Concurrent.STM.TVar as STVar -- (TVar,newTVarIO,readTVar,writeTVar)+import qualified GHC.Conc as G -- unsafeIOToSTM+import Data.IORef(newIORef,readIORef,writeIORef)+import Data.Maybe(isJust,Maybe,fromJust)+import Control.Monad.Reader(MonadReader(..),ReaderT(ReaderT),runReaderT,lift,asks)++--------------------------------------------------------------------------------+++instance MonadAdvSTM (AdvSTM) where++ onCommitWith ioclosure = do+ commitCl <- AdvSTM $ asks commitClosure+ liftAdv $ STVar.writeTVar commitCl ioclosure++ onCommit ioaction = do+ commitVar <- AdvSTM $ asks commitTVar+ commitActions <- liftAdv $ STVar.readTVar commitVar+ liftAdv $ STVar.writeTVar commitVar $ ioaction : commitActions++-- 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++ 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+++ -- | See 'STVar.newTVar'+ newTVar a = TVar `liftM` liftAdv (STVar.newTVar a)+ `ap` liftAdv (newTMVar ())+ `ap` liftAdv (STVar.newTVar Nothing)+++ -- | Writes a value to a TVar. Blocks until the onCommit IO-action(s) are+ -- complete. See 'onCommit' for details.+ writeTVar tvar a = do+ commitLock <- liftAdv $ tryTakeTMVar (onCommitLock tvar)+ -- Get ThreadID of current transaction:+ curTid <- AdvSTM $ asks transThreadId+ storedTid <- liftAdv $ STVar.readTVar (currentTid tvar)+ case commitLock of+ Nothing ->+ if isJust storedTid && fromJust storedTid == curTid+ then throw Deadlock -- Can't write the TVar in the onCommit phase+ else retry+ Just _ -> do+ unless (isJust storedTid && (fromJust storedTid == curTid)) $ do+ -- First write access, update the ThreadId:+ liftAdv $ STVar.writeTVar (currentTid tvar) $ Just curTid+ -- Add this TVar to the onCommit-listener list:+ lsTVar <- AdvSTM $ asks listeners+ ls <- liftAdv $ STVar.readTVar lsTVar+ -- Remember the old value for rollback:+ oldval <- liftAdv $ STVar.readTVar (valueTVar tvar)+ liftAdv $ STVar.writeTVar lsTVar $+ (onCommitLock tvar,TVarValue (valueTVar tvar,oldval)) : ls++ liftAdv $ STVar.writeTVar (valueTVar tvar) a+ liftAdv $ putTMVar (onCommitLock tvar) ()+++ writeTVarAsync tvar = liftAdv . STVar.writeTVar (valueTVar 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 = do+ commitLock <- liftAdv $ tryTakeTMVar (onCommitLock tvar)+ case commitLock of+ Nothing -> do+ storedTid <- liftAdv $ STVar.readTVar (currentTid tvar)+ curTid <- AdvSTM $ asks transThreadId+ if isJust storedTid && fromJust storedTid == curTid+ then throw Deadlock+ else retry+ Just _ -> do+ result <- liftAdv $ STVar.readTVar $ valueTVar tvar+ liftAdv $ putTMVar (onCommitLock tvar) ()+ return result+++ readTVarAsync = liftAdv . STVar.readTVar . valueTVar+++ -- | Forks a separate thread to run the IO action and then retries the transaction.+ unsafeRetryWith io = do -- unsafeOnRetry io >> retry+ doneMVar <- AdvSTM $ asks retryDoneMVar+ unsafeIOToSTM $ forkIO $ (do+ val <- takeMVar doneMVar+ case val of+ Nothing -> return ()+ Just _ -> io+ ) `finally` tryPutMVar doneMVar (Just ())+ retry++ unsafeIOToSTM = liftAdv . G.unsafeIOToSTM++--------------------------------------------------------------------------------+++-- | See 'S.atomically'+atomically :: AdvSTM a -> IO a+atomically (AdvSTM action) = do+ let debugging = False -- Switching this to True occasionally causes deadlocks (b/c unsafeIO)!+ debugTVar <- STVar.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 <- STVar.newTVarIO [] -- IO actions to be run if the transaction commits+ commitCl <- STVar.newTVarIO (sequence_)+ -- IO action that \"contains\" the commit IO actions.+ retryDoneMVar <- newMVar $ Just ()+ commitListeners <- STVar.newTVarIO []+ let env = Env { commitTVar = commitVar :: STVar.TVar [IO ()] -- -> IO ())+ , commitClosure = commitCl+ , retryDoneMVar = retryDoneMVar :: MVar (Maybe ()) -- (IO () -> IO ())+ , transThreadId = tid :: ThreadId+ , listeners = commitListeners :: STVar.TVar [(TMVar (),TVarValue)]+ , debugModeVar = debugTVar+ }++ let stopRetryWith = swapMVar retryDoneMVar Nothing++ let wrappedAction = runReaderT action env -- `S.orElse` check'retry++ -- Block exceptions from other threads for the rest of 'atomically'+ mask_ $ do+ 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 <- STVar.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+ S.atomically $ do+ ls <- STVar.readTVar commitListeners+ mapM_ (\(l,TVarValue (oldValTVar,oldVal)) -> do+ STVar.writeTVar oldValTVar oldVal -- smthg went wrong, restore the old value+ putTMVar l () -- ...and unblock the TVar+ ) ls++ -- Wait for the retryWith thread(s) to be done before running the onCommit+ -- actions.+ stopRetryWith++ -- Now try to run the onCommit IO actions:+ commitAcs <- liftM reverse $ S.atomically $ STVar.readTVar commitVar+ commitClAc <- S.atomically $ STVar.readTVar commitCl+ commitClAc [ ac | ac <- commitAcs ]+ `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 <- STVar.readTVar commitListeners+ mapM_ (\(l,_) ->+ putTMVar l ()+ ) ls+ debug (show (tid,"Transaction done; retry thread finished")) 1000000 debugging+ debug "************************************************************" 0 debugging++ return result++--------------------------------------------------------------------------------++-- | Switches the debug mode on or off.+-- /WARNING:/ Can lead to deadlocks!+debugMode :: Bool -> AdvSTM ()+debugMode switch = do+ debVar <- AdvSTM $ asks debugModeVar+ liftAdv $ STVar.writeTVar debVar switch++-- | Uses unsafeIOToSTM to output the Thread Id and a message and delays+-- for a given number of time.+-- /WARNING:/ Can lead to deadlocks!+debugAdvSTM :: String -> Int -> AdvSTM ()+debugAdvSTM msg delay = do+ debVar <- AdvSTM $ asks debugModeVar+ debugging <- liftAdv $ STVar.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 (u . f)++-- WARNING: Can lead to deadlocks!+debugSTM :: String -> Int -> Bool -> S.STM ()+debugSTM msg delay debugging =+ when debugging $ G.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++{-+instance forall a. (Exception e, a) => MonadError e (AdvSTM a) where+ throwError = throw+ catchError = catchSTM+-}
+ src/Control/Concurrent/AdvSTM/TArray.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE FlexibleInstances, MultiParamTypeClasses #-}++-----------------------------------------------------------------------------+-- |+-- 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 <thaldyron@gmail.com>+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- Corresponds to "Control.Concurrent.STM.TArray"+--+-----------------------------------------------------------------------------++++module Control.Concurrent.AdvSTM.TArray( TArray )+where+import Control.Monad.AdvSTM( MonadAdvSTM )+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 MonadAdvSTM m => MArray TArray e m 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)
+ src/Control/Concurrent/AdvSTM/TChan.hs view
@@ -0,0 +1,95 @@+-----------------------------------------------------------------------------+-- |+-- 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 <robinson@ecs.tuwien.ac.at>+-- 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.Monad.AdvSTM(MonadAdvSTM,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 :: MonadAdvSTM m => m (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 :: MonadAdvSTM m => TChan a -> a -> m ()+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 :: MonadAdvSTM m => TChan a -> m 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 :: MonadAdvSTM m => TChan a -> m (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 :: MonadAdvSTM m => TChan a -> a -> m ()+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 :: MonadAdvSTM m => TChan a -> m Bool+isEmptyTChan (TChan read write) = do+ listhead <- readTVar read+ head <- readTVar listhead+ case head of+ TNil -> return True+ TCons _ _ -> return False
+ src/Control/Concurrent/AdvSTM/TMVar.hs view
@@ -0,0 +1,144 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.AdvSTM.TMVar+-- Copyright : (c) Peter Robinson 2009, (c) The University of Glasgow 2004+-- License : BSD3+-- +-- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>+-- 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.Monad.AdvSTM(MonadAdvSTM,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 :: MonadAdvSTM m => a -> m (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 :: MonadAdvSTM m => m (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 :: MonadAdvSTM m => TMVar a -> m 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 :: MonadAdvSTM m => TMVar a -> m (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 :: MonadAdvSTM m => TMVar a -> a -> m ()+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 :: MonadAdvSTM m => TMVar a -> a -> m 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 :: MonadAdvSTM m => TMVar a -> m 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 :: MonadAdvSTM m => TMVar a -> a -> m 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 :: MonadAdvSTM m => TMVar a -> m Bool+isEmptyTMVar (TMVar t) = do+ m <- readTVar t+ case m of+ Nothing -> return True+ Just _ -> return False+
+ src/Control/Concurrent/AdvSTM/TVar.hs view
@@ -0,0 +1,113 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.AdvSTM.TVar+-- Copyright : (c) Peter Robinson 2009+-- License : BSD-style (see the file LICENSE)+-- +-- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- +--+-----------------------------------------------------------------------------++module Control.Concurrent.AdvSTM.TVar( -- * TVars+ TVar + , newTVar+ , newTVarIO+ , readTVar+ , writeTVar+ , readTVarAsync+ , writeTVarAsync+ )+where+import Control.Monad.AdvSTM.Class+ (readTVar,writeTVar,newTVar,TVar(TVar),valueTVar,readTVarAsync,writeTVarAsync)+import qualified Control.Concurrent.STM.TVar as OldTVar+import qualified Control.Concurrent.STM.TMVar as OldTMVar+import Control.Monad(liftM,ap)+{-+import qualified Control.Concurrent.STM as STM+import qualified Control.Concurrent.STM.TChan as OldTChan+import Control.Exception(throw,Deadlock(Deadlock))+import Control.Concurrent.AdvSTM(liftAdv,orElse,retry,readTVar,writeTVar,newTVar)+import Control.Monad.AdvSTM.Def(AdvSTM(AdvSTM),transThreadId,listeners,TVarValue(TVarValue))+import Control.Concurrent(ThreadId)+import Control.Monad.Reader(asks)+import Data.Maybe(isJust,Maybe,fromJust)+import qualified Data.Set as S+-}+--------------------------------------------------------------------------------++-- | See 'OldTVar.newTVarIO'+--+newTVarIO :: a -> IO (TVar a)+newTVarIO a = TVar `liftM` OldTVar.newTVarIO a+ `ap` OldTMVar.newTMVarIO ()+ `ap` OldTVar.newTVarIO Nothing++--------------------------------------------------------------------------------+{-+-- | 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)+-}+{-+--------------------------------------------------------------------------------++-- | 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++--------------------------------------------------------------------------------+-}++
+ src/Control/Monad/AdvSTM.hs view
@@ -0,0 +1,31 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Monad.AdvSTM+-- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008+-- License : BSD3+-- +-- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>+-- 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)+
+ src/Control/Monad/AdvSTM/Class.hs view
@@ -0,0 +1,272 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Monad.AdvSTM.Class+-- Copyright : Peter Robinson 2008, HaskellWiki 2006-2007+-- License : BSD3+--+-- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>+-- 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(..), handleSTM, TVar(TVar), valueTVar, onCommitLock, currentTid)+where++import Control.Exception(Exception,throw)+import qualified Control.Concurrent.STM as S+import qualified Control.Concurrent.STM.TVar as OldTVar+import qualified Control.Concurrent.STM.TMVar as OldTMVar+import Control.Monad(Monad,liftM,ap)+import Control.Monad.Trans(lift)+import Control.Monad.State(StateT(StateT),mapStateT,runStateT,evalStateT)+import Control.Monad.Writer(WriterT(WriterT),mapWriterT,runWriterT,execWriterT)+import Control.Monad.Reader(ReaderT(ReaderT),mapReaderT,runReaderT)+-- import Control.Monad.AdvSTM.Def(AdvSTM)+import Control.Concurrent( ThreadId )+--import GHC.Conc( unsafeIOToSTM )+-- import {-# SOURCE #-} Control.Concurrent.AdvSTM.TVar+import Data.Monoid++data TVar a = TVar+ { valueTVar :: OldTVar.TVar a+ , onCommitLock :: OldTMVar.TMVar ()+ , currentTid :: OldTVar.TVar (Maybe ThreadId)+ }++-- | A type class for extended-STM monads. For a concrete instantiation see+-- 'AdvSTM'+class Monad m => MonadAdvSTM m where++ -- | Takes a closure IO action and a commit IO action.+ -- The commit IO action will be executed /iff/ the transaction commits.+ -- Commit actions are sequenced (within the same transaction), i.e.,+ --+ -- > onCommitWith id (putStr "hello")+ -- > onCommitWith id (putStr " world")+ --+ -- will print \"hello world\".+ --+ -- The closure action is useful for encapsulating the commit actions,+ -- e.g., within a database transaction.+ -- The last call of onCommitWith in the transaction+ -- is applied to the sequence of commit actions, i.e.:+ --+ -- > onCommitWith id (putStr "hello")+ -- > onCommitWith (\s -> do { putStrLn "start"; s; putStrLn "\nend"}) (putStr " world")+ --+ -- * When a TVar was modified in a transaction and the transaction tries to commit,+ -- this update remains invisible to other threads until the corresponding+ -- onCommit action is dispatched.+ --+ -- * 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)+ --+ -- Note: If you /really/ need to access the 'TVar' within an onCommit action+ -- (e.g. to recover from an IO exception), you can use 'writeTVarAsync'.+ onCommitWith :: ([IO ()] -> IO ()) -- ^ closure action+ -> m ()+ -- | Works like 'onCommitWith' without closure action:+ -- 'onCommit = onCommitWith id'+ onCommit :: IO () -> m ()+-- onCommit = onCommitWith sequence_++ -- | Retries the transaction and uses 'unsafeIOToSTM' to fork off a+ -- thread that runs the given IO action. Since a transaction might be rerun+ -- several times by the runtime system, it is your responsibility to+ -- ensure that the IO-action is idempotent and releases all acquired locks.+ unsafeRetryWith :: IO () -> m b++ -- | See 'S.orElse'+ orElse :: m a -> m a -> m a++ -- | See 'S.retry'+ retry :: m a++ -- | See 'S.check'+ check :: Bool -> m ()++ -- | See 'S.catchSTM'+ catchSTM :: Exception e => m a -> (e -> m a) -> m a+++ -- | Lifts STM actions to 'MonadAdvSTM'.+ liftAdv :: S.STM a -> m a++ -- | Reads a value from a TVar. Blocks until the IO onCommit aidction(s) of+ -- the corresponding transaction are complete.is not the last function+ -- See 'onCommit' for a more detailed description of this behaviour.+ readTVar :: TVar a -> m a++ -- | Writes a value to a TVar. Blocks until the onCommit IO-action(s) are+ -- complete. See 'onCommit' for details.+ writeTVar :: TVar a -> a -> m ()++ -- | Reads a value directly from the TVar. Does not block when the+ -- onCommit actions aren't done yet. NOTE: Only use this function when+ -- you know what you're doing.+ readTVarAsync :: TVar a -> m a++ -- | Writes a value directly to the TVar. Does not block when+ -- onCommit actions aren't done yet. This function comes in handy for+ -- error recovery of exceptions that occur in onCommit.+ writeTVarAsync :: TVar a -> a -> m ()++ -- | See 'OldTVar.newTVar'+ newTVar :: a -> m (TVar a)++-- -- See 'S.atomically'+-- runAtomic :: m a -> IO aid+-- newTVarIO :: a -> IO (TVar a)+ unsafeIOToSTM :: IO a -> m a++-- | A version of 'catchSTM' with the arguments swapped around.+handleSTM :: (MonadAdvSTM m, Exception e) => (e -> m a) -> m a -> m a+handleSTM = flip catchSTM++++--------------------------------------------------------------------------------+++mapStateT2 :: (m (a, s) -> n (b, s) -> o (c,s))+ -> StateT s m a -> StateT s n b -> StateT s o c+mapStateT2 f m1 m2 = StateT $ \s -> f (runStateT m1 s) (runStateT m2 s)++liftAndSkipStateT f m = StateT $ \s -> let a = evalStateT m s+ in do r <- f a+ return (r,s)++instance MonadAdvSTM m => MonadAdvSTM (StateT s m) where++ onCommit ca = lift (onCommit ca)++ onCommitWith cc = lift (onCommitWith cc)++ unsafeRetryWith = lift . unsafeRetryWith++ orElse = mapStateT2 orElse++ retry = lift retry++ check = lift . check++ catchSTM m h = StateT (\r -> catchSTM (runStateT m r) (\e -> runStateT (h e) r))++ liftAdv = lift . liftAdv++ readTVar = lift . readTVar++ writeTVar tvar = lift . writeTVar tvar++ readTVarAsync = lift . readTVarAsync++ writeTVarAsync tvar = lift . writeTVarAsync tvar++ newTVar = lift . newTVar++ unsafeIOToSTM = lift . unsafeIOToSTM++--------------------------------------------------------------------------------++mapWriterT2 :: (m (a, w) -> n (b, w) -> o (c,w))+ -> WriterT w m a -> WriterT w n b -> WriterT w o c+mapWriterT2 f m1 m2 = WriterT $ f (runWriterT m1) (runWriterT m2)+-- mapWriterT2 f m1 m2 = liftM f m1 `ap` m2++evalWriterT :: Monad m => WriterT w m a -> m a+evalWriterT m = do+ (a,_) <- runWriterT m+ return a++liftAndSkipWriterT :: (Monad m,Monoid w)+ => (m a -> m b)+ -> WriterT w m a -> WriterT w m b+liftAndSkipWriterT f m = WriterT $+ let a = evalWriterT m+ in do r <- f a+ return (r,mempty)++instance (MonadAdvSTM m, Monoid w) => MonadAdvSTM (WriterT w m) where+ onCommit ca = lift (onCommit ca)++ onCommitWith cc = lift (onCommitWith cc)++ unsafeRetryWith = lift . unsafeRetryWith++ orElse = mapWriterT2 orElse++ retry = lift retry++ check = lift . check++ -- Note: The writer-log modifications of the invariant action+ -- are thrown away!++ catchSTM m h = WriterT (catchSTM (runWriterT m) (\e -> runWriterT (h e)))++ liftAdv = lift . liftAdv++ readTVar = lift . readTVar++ writeTVar tvar = lift . writeTVar tvar++ readTVarAsync = lift . readTVarAsync++ writeTVarAsync tvar = lift . writeTVarAsync tvar++ newTVar = lift . newTVar++ unsafeIOToSTM = lift . unsafeIOToSTM+--------------------------------------------------------------------------------++mapReaderT2 :: (m a -> n b -> o c) -> ReaderT r m a -> ReaderT r n b -> ReaderT r o c+mapReaderT2 f m1 m2 = ReaderT $ \r -> f (runReaderT m1 r) (runReaderT m2 r)++instance MonadAdvSTM m => MonadAdvSTM (ReaderT r m) where+ onCommit ca = lift (onCommit ca)++ onCommitWith cc = lift (onCommitWith cc)++ unsafeRetryWith = lift . unsafeRetryWith++ orElse = mapReaderT2 orElse++ retry = lift retry++ check = lift . check++ catchSTM m h = ReaderT (\r -> catchSTM (runReaderT m r) (\e -> runReaderT (h e) r))++ liftAdv = lift . liftAdv+++ writeTVarAsync tvar = lift . writeTVarAsync tvar++ readTVarAsync = lift . readTVarAsync++ writeTVar tvar = lift . writeTVar tvar++ readTVar = lift . readTVar++ newTVar = lift . newTVar++ unsafeIOToSTM = lift . unsafeIOToSTM++--------------------------------------------------------------------------------
+ src/Control/Monad/AdvSTM/Def.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE ExistentialQuantification, GeneralizedNewtypeDeriving #-}++-----------------------------------------------------------------------------+-- |+-- Module : Control.Monad.AdvSTM.Def+-- Copyright : (c) HaskellWiki 2006-2007, Peter Robinson 2008+-- License : BSD3+-- +-- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>+-- 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.Applicative(Alternative,(<|>),(<*>),empty)+import Control.Monad(Monad,MonadPlus,ap,liftM,mplus,mzero)+import Control.Monad.Reader(ReaderT,mapReaderT,runReaderT)++instance Functor AdvSTM where+ fmap = liftM++instance Applicative AdvSTM where+ pure = return+ (<*>) = ap ++instance Alternative AdvSTM where+ (<|>) = mplus+ empty = mzero++-- | Drop-in replacement for the STM monad+newtype AdvSTM a = AdvSTM (ReaderT Env S.STM a)+ deriving ( Monad+ , MonadPlus+ )++++-- | The environment used for the Reader Monad+data Env = Env { commitTVar :: TVar [IO ()] -- the commit action(s)+ , commitClosure :: TVar ([IO ()] -> IO ())+ , retryDoneMVar :: MVar (Maybe ()) -- (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)+
+ src/Lib.hs view
@@ -0,0 +1,6 @@+module Lib+ ( someFunc+ ) where++someFunc :: IO ()+someFunc = putStrLn "someFunc"
stm-io-hooks.cabal view
@@ -1,59 +1,43 @@-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:- .- * When a TVar is modified in a transaction and this transaction commits,- the update remains invisible to other threads until the corresponding - onCommit action is run. - .- * If the onCommit action throws an exception, the original values of - the modified TVars are restored.- .- Note: The package can be used as a drop-in replacement for - 'Control.Concurrent.STM'. Part of this library uses code from- the Haskell Wiki (see <http://haskell.org/haskellwiki/?title=New_monads/MonadAdvSTM>).+cabal-version: 1.12 -Category: Concurrency-Author: Peter Robinson 2009, Chris Kuklewicz 2006-Maintainer: Peter Robinson <robinson@ecs.tuwien.ac.at>-License: BSD3-License-file: LICENSE-Homepage: http://darcs.monoid.at/stm-io-hooks-Version: 0.4.2+-- This file has been generated from package.yaml by hpack version 0.31.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: 13e68e1ad3386562bc42e2bf7af55114fc1b80e348f2b942e6f24db6186620df -Build-type: Simple -Cabal-Version: >= 1.2.3+name: stm-io-hooks+version: 1.1.2+synopsis: Launch your IO-actions from within the STM monad+description: Please see the README+category: Concurrency+author: Peter Robinson+maintainer: thaldyron@gmail.com+copyright: Peter Robinson 2009-2019, Chris Kuklewicz 2006+license: BSD3+license-file: LICENSE+build-type: Simple+extra-source-files:+ README.md 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 && < 5- ,stm >= 2.1.1.2 && < 2.2- ,array >= 0.2.0.0 && < 0.3- ,containers >= 0.2.0.0 && < 0.3- ,mtl >= 1.1.0.2 && < 1.2-- extensions: MultiParamTypeClasses- ,FunctionalDependencies- ,FlexibleInstances- ,GeneralizedNewtypeDeriving- ,ScopedTypeVariables- ,DeriveDataTypeable- ,RankNTypes- ,ExistentialQuantification- ,UndecidableInstances- -+ exposed-modules:+ Control.Concurrent.AdvSTM+ Control.Concurrent.AdvSTM.TArray+ Control.Concurrent.AdvSTM.TChan+ Control.Concurrent.AdvSTM.TMVar+ Control.Concurrent.AdvSTM.TVar+ Control.Monad.AdvSTM+ Control.Monad.AdvSTM.Class+ Control.Monad.AdvSTM.Def+ Lib+ other-modules:+ Paths_stm_io_hooks+ hs-source-dirs:+ src+ build-depends:+ array >=0.5.3.0 && <1+ , base >=4.7 && <5+ , mtl >=2.2.2 && <3+ , stm >=2.2 && <3+ default-language: Haskell2010