stm-io-hooks 0.4.2 → 0.5.4
raw patch · 5 files changed
+67/−159 lines, 5 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
- Control.Concurrent.AdvSTM: onRetry :: (MonadAdvSTM m) => IO () -> m ()
- Control.Concurrent.AdvSTM: retryWith :: (Monad m, MonadAdvSTM m) => IO () -> m b
- Control.Monad.AdvSTM: onRetry :: (MonadAdvSTM m) => IO () -> m ()
+ Control.Concurrent.AdvSTM: unsafeRetryWith :: (MonadAdvSTM m) => IO () -> m b
+ Control.Monad.AdvSTM: unsafeRetryWith :: (MonadAdvSTM m) => IO () -> m b
Files
- Control/Concurrent/AdvSTM.hs +33/−76
- Control/Concurrent/AdvSTM/TVar.hs-boot +0/−43
- Control/Monad/AdvSTM/Class.hs +9/−17
- Control/Monad/AdvSTM/Def.hs +1/−1
- stm-io-hooks.cabal +24/−22
Control/Concurrent/AdvSTM.hs view
@@ -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'
− Control/Concurrent/AdvSTM/TVar.hs-boot
@@ -1,43 +0,0 @@--------------------------------------------------------------------------------- |--- Module : Control.Concurrent.AdvSTM.TVar--- Copyright : (c) Peter Robinson 2009--- License : BSD-style (see the file LICENSE)--- --- Maintainer : Peter Robinson <robinson@ecs.tuwien.ac.at>--- Stability : experimental--- Portability : non-portable (requires STM)------ -----------------------------------------------------------------------------------module Control.Concurrent.AdvSTM.TVar( TVar ) {- ( -- * TVars- TVar - , newTVar- , newTVarIO- , readTVar- , writeTVar- ) -}-where-import qualified Control.Concurrent.STM.TVar as OldTVar-import qualified Control.Concurrent.STM as STM-import qualified Control.Concurrent.STM.TMVar as OldTMVar-import qualified Control.Concurrent.STM.TChan as OldTChan-import Control.Exception(throw,Deadlock(Deadlock))--- import Control.Concurrent.AdvSTM(liftAdv,orElse,retry)--- import Control.Monad.AdvSTM.Def(AdvSTM(AdvSTM),transThreadId,listeners,TVarValue(TVarValue))-import Control.Concurrent(ThreadId)-import Control.Monad(liftM,ap,unless)-import Control.Monad.Reader(asks)-import Data.Maybe(isJust,Maybe,fromJust)-import qualified Data.Set as S------------------------------------------------------------------------------------data TVar a = TVar - { valueTVar :: OldTVar.TVar a - , onCommitLock :: OldTMVar.TMVar () - , currentTid :: OldTVar.TVar (Maybe ThreadId)- }-
Control/Monad/AdvSTM/Class.hs view
@@ -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
Control/Monad/AdvSTM/Def.hs view
@@ -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:
stm-io-hooks.cabal view
@@ -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 <http://haskell.org/haskellwiki/?title=New_monads/MonadAdvSTM>).+ .+ Feedback is welcome! Category: Concurrency+Stability: experimental Author: Peter Robinson 2009, Chris Kuklewicz 2006-Maintainer: Peter Robinson <robinson@ecs.tuwien.ac.at>+Maintainer: Peter Robinson <thaldyron@gmail.com> 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