diff --git a/Control/Concurrent/AdvSTM.hs b/Control/Concurrent/AdvSTM.hs deleted file mode 100644 --- a/Control/Concurrent/AdvSTM.hs +++ /dev/null @@ -1,298 +0,0 @@ ------------------------------------------------------------------------------ --- | --- 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( onCommit - , unsafeRetryWith - , orElse - , retry - , check - , alwaysSucceeds - , always - , catchSTM - , liftAdv - , newTVar - , readTVarAsync - , writeTVarAsync - , readTVar - , writeTVar - ) - - -- * Monad AdvSTM - , AdvSTM - , 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(..),finally) -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,newMVar,takeMVar,tryTakeMVar,putMVar,tryPutMVar,swapMVar) -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 - 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 - - alwaysSucceeds inv = unlift inv >>= liftAdv . S.alwaysSucceeds - - always inv = unlift inv >>= liftAdv . S.always - - 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 - unsafeIOToAdvSTM $ forkIO $ (do - val <- takeMVar doneMVar - case val of - Nothing -> return () - Just _ -> io - ) `finally` tryPutMVar doneMVar (Just ()) - 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 [] -- IO actions to be run if the transaction commits - retryDoneMVar <- newMVar $ Just () - commitListeners <- STVar.newTVarIO [] - let env = Env { commitTVar = commitVar :: STVar.TVar [IO ()] -- -> IO ()) - , 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' - 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 - 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 <- S.atomically $ STVar.readTVar commitVar - sequence_ [ 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 - --------------------------------------------------------------------------------- - --- | 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 diff --git a/Control/Concurrent/AdvSTM/TArray.hs b/Control/Concurrent/AdvSTM/TArray.hs deleted file mode 100644 --- a/Control/Concurrent/AdvSTM/TArray.hs +++ /dev/null @@ -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 --- 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) - diff --git a/Control/Concurrent/AdvSTM/TChan.hs b/Control/Concurrent/AdvSTM/TChan.hs deleted file mode 100644 --- a/Control/Concurrent/AdvSTM/TChan.hs +++ /dev/null @@ -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 --- 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 - - diff --git a/Control/Concurrent/AdvSTM/TMVar.hs b/Control/Concurrent/AdvSTM/TMVar.hs deleted file mode 100644 --- a/Control/Concurrent/AdvSTM/TMVar.hs +++ /dev/null @@ -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 --- 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 - diff --git a/Control/Concurrent/AdvSTM/TVar.hs b/Control/Concurrent/AdvSTM/TVar.hs deleted file mode 100644 --- a/Control/Concurrent/AdvSTM/TVar.hs +++ /dev/null @@ -1,113 +0,0 @@ ------------------------------------------------------------------------------ --- | --- 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 - , 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 - --------------------------------------------------------------------------------- --} - - diff --git a/Control/Monad/AdvSTM.hs b/Control/Monad/AdvSTM.hs deleted file mode 100644 --- a/Control/Monad/AdvSTM.hs +++ /dev/null @@ -1,31 +0,0 @@ ------------------------------------------------------------------------------ --- | --- 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) - diff --git a/Control/Monad/AdvSTM/Class.hs b/Control/Monad/AdvSTM/Class.hs deleted file mode 100644 --- a/Control/Monad/AdvSTM/Class.hs +++ /dev/null @@ -1,257 +0,0 @@ ------------------------------------------------------------------------------ --- | --- Module : Control.Monad.AdvSTM.Class --- Copyright : Peter Robinson 2008, HaskellWiki 2006-2007 --- 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(..), 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 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'. - -- - onCommit :: IO () -> m () - - -- | 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.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 () - - -- | 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 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 - - unsafeRetryWith = lift . unsafeRetryWith - - 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 - - readTVarAsync = lift . readTVarAsync - - writeTVarAsync tvar = lift . writeTVarAsync 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 - - unsafeRetryWith = lift . unsafeRetryWith - - 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 - - readTVarAsync = lift . readTVarAsync - - writeTVarAsync tvar = lift . writeTVarAsync 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 - - unsafeRetryWith = lift . unsafeRetryWith - - 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 - - - writeTVarAsync tvar = lift . writeTVarAsync tvar - - readTVarAsync = lift . readTVarAsync - - writeTVar tvar = lift . writeTVar tvar - - readTVar = lift . readTVar - - newTVar = lift . newTVar - - --------------------------------------------------------------------------------- diff --git a/Control/Monad/AdvSTM/Def.hs b/Control/Monad/AdvSTM/Def.hs deleted file mode 100644 --- a/Control/Monad/AdvSTM/Def.hs +++ /dev/null @@ -1,47 +0,0 @@ ------------------------------------------------------------------------------ --- | --- 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,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 ()] -- the commit action(s) - , 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) - diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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'. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/Setup.lhs b/Setup.lhs deleted file mode 100644 --- a/Setup.lhs +++ /dev/null @@ -1,3 +0,0 @@ -#!/usr/bin/env runhaskell -> import Distribution.Simple -> main = defaultMain diff --git a/src/Control/Concurrent/AdvSTM.hs b/src/Control/Concurrent/AdvSTM.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/AdvSTM.hs @@ -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 +-- 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 +-} diff --git a/src/Control/Concurrent/AdvSTM/TArray.hs b/src/Control/Concurrent/AdvSTM/TArray.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/AdvSTM/TArray.hs @@ -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 +-- 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) diff --git a/src/Control/Concurrent/AdvSTM/TChan.hs b/src/Control/Concurrent/AdvSTM/TChan.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/AdvSTM/TChan.hs @@ -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 +-- 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 diff --git a/src/Control/Concurrent/AdvSTM/TMVar.hs b/src/Control/Concurrent/AdvSTM/TMVar.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/AdvSTM/TMVar.hs @@ -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 +-- 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 + diff --git a/src/Control/Concurrent/AdvSTM/TVar.hs b/src/Control/Concurrent/AdvSTM/TVar.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/AdvSTM/TVar.hs @@ -0,0 +1,113 @@ +----------------------------------------------------------------------------- +-- | +-- 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 + , 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 + +-------------------------------------------------------------------------------- +-} + + diff --git a/src/Control/Monad/AdvSTM.hs b/src/Control/Monad/AdvSTM.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/AdvSTM.hs @@ -0,0 +1,31 @@ +----------------------------------------------------------------------------- +-- | +-- 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) + diff --git a/src/Control/Monad/AdvSTM/Class.hs b/src/Control/Monad/AdvSTM/Class.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/AdvSTM/Class.hs @@ -0,0 +1,272 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Monad.AdvSTM.Class +-- Copyright : Peter Robinson 2008, HaskellWiki 2006-2007 +-- 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(..), 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 + +-------------------------------------------------------------------------------- diff --git a/src/Control/Monad/AdvSTM/Def.hs b/src/Control/Monad/AdvSTM/Def.hs new file mode 100644 --- /dev/null +++ b/src/Control/Monad/AdvSTM/Def.hs @@ -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 +-- 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) + diff --git a/src/Lib.hs b/src/Lib.hs new file mode 100644 --- /dev/null +++ b/src/Lib.hs @@ -0,0 +1,6 @@ +module Lib + ( someFunc + ) where + +someFunc :: IO () +someFunc = putStrLn "someFunc" diff --git a/stm-io-hooks.cabal b/stm-io-hooks.cabal --- a/stm-io-hooks.cabal +++ b/stm-io-hooks.cabal @@ -1,61 +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 (at least once) if the transaction retries, while commit-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'. This library was inspired by the AdvSTM monad on - the Haskell Wiki (see ). - . - Feedback is welcome! +cabal-version: 1.12 -Category: Concurrency -Stability: experimental -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.6.0 +-- 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 - - build-depends: base >= 4 && < 5 - ,stm >= 2.1.1.2 && < 2.2 - ,array >= 0.2.0.0 && < 0.4 - ,containers >= 0.2.0.0 && < 0.4 - ,mtl >= 1.1.0.2 && < 1.3 - - 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