diff --git a/Control/Concurrent/AdvSTM.hs b/Control/Concurrent/AdvSTM.hs --- a/Control/Concurrent/AdvSTM.hs +++ b/Control/Concurrent/AdvSTM.hs @@ -14,13 +14,12 @@ module Control.Concurrent.AdvSTM( -- * Class MonadAdvSTM MonadAdvSTM( onCommit - , onRetry + , unsafeRetryWith , orElse , retry , check , alwaysSucceeds , always --- , runAtomic , catchSTM , liftAdv , newTVar @@ -30,7 +29,6 @@ -- * Monad AdvSTM , AdvSTM - , retryWith , atomically , unsafeIOToAdvSTM , handleSTM @@ -44,14 +42,14 @@ 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.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,newEmptyMVar,takeMVar,tryTakeMVar,putMVar) +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) @@ -68,12 +66,12 @@ 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 +-- 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 @@ -85,8 +83,6 @@ always inv = unlift inv >>= liftAdv . S.always --- runAtomic = atomically - catchSTM action handler = do action' <- unlift action handler' <- unlift1 handler @@ -99,9 +95,9 @@ -- | See 'STVar.newTVar' - newTVar a = TVar `liftM` (liftAdv $ STVar.newTVar a) - `ap` (liftAdv $ newTMVar ()) - `ap` (liftAdv $ STVar.newTVar Nothing) + 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 @@ -151,14 +147,18 @@ 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 - - + -- | 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 -------------------------------------------------------------------------------- @@ -173,49 +173,20 @@ 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 + retryDoneMVar <- newMVar $ Just () commitListeners <- STVar.newTVarIO [] let env = Env { commitTVar = commitVar :: STVar.TVar (IO () -> IO ()) - , retryMVar = retryVar :: MVar (IO () -> IO ()) + , retryDoneMVar = retryDoneMVar :: MVar (Maybe ()) -- (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 stopRetryWith = swapMVar retryDoneMVar Nothing - let wrappedAction = runReaderT action env `S.orElse` check'retry + let wrappedAction = runReaderT action env -- `S.orElse` check'retry - -- Block interruptions from other threads for the rest of 'atomically' + -- 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 @@ -225,13 +196,12 @@ -- 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 + 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 @@ -239,6 +209,10 @@ 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: commitFun <- S.atomically $ STVar.readTVar commitVar commitFun (return ()) `catch` (\(e::SomeException) -> rollbackOnCommit >> throw e) @@ -251,28 +225,11 @@ 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 - - + return result - -------------------------------------------------------------------------------- -- | See 'unsafeIOToSTM' diff --git a/Control/Concurrent/AdvSTM/TVar.hs-boot b/Control/Concurrent/AdvSTM/TVar.hs-boot deleted file mode 100644 --- a/Control/Concurrent/AdvSTM/TVar.hs-boot +++ /dev/null @@ -1,43 +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( 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) - } - diff --git a/Control/Monad/AdvSTM/Class.hs b/Control/Monad/AdvSTM/Class.hs --- a/Control/Monad/AdvSTM/Class.hs +++ b/Control/Monad/AdvSTM/Class.hs @@ -66,24 +66,16 @@ -- 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 () + -- | 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 - -- | Runs any IO actions added by 'onRetry' and then retries the - -- transaction. + -- | See 'S.retry' retry :: m a -- | See 'S.check' @@ -138,7 +130,7 @@ instance MonadAdvSTM m => MonadAdvSTM (StateT s m) where onCommit = lift . onCommit - onRetry = lift . onRetry + unsafeRetryWith = lift . unsafeRetryWith orElse = mapStateT2 orElse @@ -184,7 +176,7 @@ instance (MonadAdvSTM m, Monoid w) => MonadAdvSTM (WriterT w m) where onCommit = lift . onCommit - onRetry = lift . onRetry + unsafeRetryWith = lift . unsafeRetryWith orElse = mapWriterT2 orElse @@ -215,7 +207,7 @@ instance MonadAdvSTM m => MonadAdvSTM (ReaderT r m) where onCommit = lift . onCommit - onRetry = lift . onRetry + unsafeRetryWith = lift . unsafeRetryWith orElse = mapReaderT2 orElse diff --git a/Control/Monad/AdvSTM/Def.hs b/Control/Monad/AdvSTM/Def.hs --- a/Control/Monad/AdvSTM/Def.hs +++ b/Control/Monad/AdvSTM/Def.hs @@ -33,7 +33,7 @@ -- | 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) + , 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: diff --git a/stm-io-hooks.cabal b/stm-io-hooks.cabal --- a/stm-io-hooks.cabal +++ b/stm-io-hooks.cabal @@ -2,8 +2,8 @@ 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 + 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, @@ -16,29 +16,31 @@ 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 ). + . + Feedback is welcome! Category: Concurrency +Stability: experimental Author: Peter Robinson 2009, Chris Kuklewicz 2006 -Maintainer: Peter Robinson +Maintainer: Peter Robinson License: BSD3 License-file: LICENSE Homepage: http://darcs.monoid.at/stm-io-hooks -Version: 0.4.2 +Version: 0.5.4 Build-type: Simple Cabal-Version: >= 1.2.3 library - Exposed-Modules: Control.Monad.AdvSTM - ,Control.Concurrent.AdvSTM - ,Control.Concurrent.AdvSTM.TMVar - ,Control.Concurrent.AdvSTM.TVar - ,Control.Concurrent.AdvSTM.TArray - ,Control.Concurrent.AdvSTM.TChan + 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! + Other-Modules: Control.Monad.AdvSTM.Def + Control.Monad.AdvSTM.Class build-depends: base >= 4 && < 5 ,stm >= 2.1.1.2 && < 2.2 @@ -46,14 +48,14 @@ ,containers >= 0.2.0.0 && < 0.3 ,mtl >= 1.1.0.2 && < 1.2 - extensions: MultiParamTypeClasses - ,FunctionalDependencies - ,FlexibleInstances - ,GeneralizedNewtypeDeriving - ,ScopedTypeVariables - ,DeriveDataTypeable - ,RankNTypes - ,ExistentialQuantification - ,UndecidableInstances + extensions: MultiParamTypeClasses + FunctionalDependencies + FlexibleInstances + GeneralizedNewtypeDeriving + ScopedTypeVariables + DeriveDataTypeable + RankNTypes + ExistentialQuantification + UndecidableInstances