diff --git a/mstate.cabal b/mstate.cabal --- a/mstate.cabal +++ b/mstate.cabal @@ -3,13 +3,13 @@ Description: MState offers a State monad which can be used in concurrent applications. It also manages new threads and waits until the whole state monad has been evaluated/executed before it returns - the state values. + the state values (if desired). Author: Nils Schweinsberg Maintainer: -Version: 0.2 -Category: Concurrent, Monads +Version: 0.2.1 +Category: Concurrency, Monads License: BSD3 License-File: LICENSE Cabal-Version: >= 1.6 diff --git a/src/Control/Concurrent/MState.hs b/src/Control/Concurrent/MState.hs --- a/src/Control/Concurrent/MState.hs +++ b/src/Control/Concurrent/MState.hs @@ -1,4 +1,5 @@ -{-# LANGUAGE FlexibleInstances, MultiParamTypeClasses, UndecidableInstances #-} +{-# LANGUAGE FlexibleInstances, MultiParamTypeClasses, UndecidableInstances, + ScopedTypeVariables #-} --------------------------------------------------------------------------- -- | @@ -18,21 +19,24 @@ ( -- * The MState Monad MState + , module Control.Monad.State.Class , runMState , evalMState , execMState , mapMState - , withMState + -- , withMState , modifyM -- * Concurrency , Forkable (..) , forkM + , killMState -- * Example -- $example ) where +import Prelude hiding (catch) import Control.Monad.State.Class import Control.Monad.Cont @@ -45,76 +49,79 @@ import Control.Monad.IO.Peel import Control.Exception.Peel +import Control.Monad.Trans.Peel --- | The MState is an abstract data definition for a State monad which can be --- used in concurrent applications. Use `forkM` to start a new thread with the --- same state. -newtype MState t m a = MState { runMState' :: (TVar t, TVar [TMVar ()]) -> m a } - +-- | The MState monad is a state monad for concurrent applications. To create a +-- new thread sharing the same (modifiable) state use the `forkM` function. +newtype MState t m a = MState { runMState' :: (TVar t, TVar [(ThreadId, TMVar ())]) -> m a } --- | Typeclass for forkable monads, for instance: --- --- > instance Forkable IO where --- > fork = forkIO --- --- This is only the basic information about how to fork a new thread in the --- current monad. To start a new thread in a `MState` application you should --- always use `forkM`. +-- | Typeclass for forkable monads. This is the basic information about how to +-- fork a new thread in the current monad. To start a new thread in a `MState` +-- application you should always use `forkM`. class (MonadPeelIO m) => Forkable m where fork :: m () -> m ThreadId -instance Forkable IO where - fork = forkIO - -instance Forkable (ReaderT s IO) where - fork newT = ask >>= liftIO . forkIO . runReaderT newT - - --- | Wait for all `TMVars` to get filled by their processes +-- | Wait for all `TMVars` to get filled by their processes. waitForTermination :: MonadIO m - => TVar [TMVar ()] + => TVar [(ThreadId, TMVar ())] -> m () -waitForTermination = liftIO . atomically . (mapM_ takeTMVar <=< readTVar) +waitForTermination = liftIO . atomically . (mapM_ (takeTMVar . snd) <=< readTVar) --- | Run a `MState` application, returning both, the function value and the --- final state +-- | Run a `MState` application, returning both, the function value and the +-- final state. Note that this function has to wait for all threads to finish +-- before it can return the final state. runMState :: Forkable m - => MState t m a -- ^ Action to run - -> t -- ^ Initial state value - -> m (a,t) + => MState t m a -- ^ Action to run + -> t -- ^ Initial state value + -> m (a,t) runMState m t = do - - ref <- liftIO $ newTVarIO t - c <- liftIO $ newTVarIO [] - mv <- liftIO newEmptyMVar - - _ <- runMState' (forkM $ m >>= liftIO . putMVar mv) (ref, c) + (a, Just t') <- runAndWaitMaybe True m t + return (a, t') - waitForTermination c - a <- liftIO $ takeMVar mv - t' <- liftIO . atomically $ readTVar ref - return (a,t') +runAndWaitMaybe :: Forkable m + => Bool + -> MState t m a + -> t + -> m (a, Maybe t) +runAndWaitMaybe b m t = do + myI <- liftIO myThreadId + myM <- liftIO newEmptyTMVarIO + ref <- liftIO $ newTVarIO t + c <- liftIO $ newTVarIO [(myI, myM)] + a <- runMState' m (ref, c) `finally` liftIO (atomically $ putTMVar myM ()) + if b then do + -- wait before getting the final state + waitForTermination c + t' <- liftIO . atomically $ readTVar ref + return (a, Just t') + else + -- don't wait for other threads + return (a, Nothing) --- | Run a `MState` application, ignoring the final state +-- | Run a `MState` application, ignoring the final state. If the first +-- argument is `True` this function will wait for all threads to finish before +-- returning the final result, otherwise it will return the function value as +-- soon as its acquired. evalMState :: Forkable m - => MState t m a -- ^ Action to evaluate + => Bool -- ^ Wait for all threads to finish? + -> MState t m a -- ^ Action to evaluate -> t -- ^ Initial state value -> m a -evalMState m t = runMState m t >>= return . fst - +evalMState b m t = runAndWaitMaybe b m t >>= return . fst --- | Run a `MState` application, ignoring the function value +-- | Run a `MState` application, ignoring the function value. This function +-- will wait for all threads to finish before returning the final state. execMState :: Forkable m => MState t m a -- ^ Action to execute -> t -- ^ Initial state value -> m t execMState m t = runMState m t >>= return . snd - -- | Map a stateful computation from one @(return value, state)@ pair to --- another. See "Control.Monad.State.Lazy" for more information. +-- another. See "Control.Monad.State.Lazy" for more information. Be aware that +-- both MStates still share the same state. mapMState :: (MonadIO m, MonadIO n) => (m (a,t) -> n (b,t)) -> MState t m a @@ -127,6 +134,7 @@ liftIO . atomically $ writeTVar r v' return b +{- TODO: What's the point of this function? Does it make sense for MStates? -- | Apply a function to the state before running the `MState` withMState :: (MonadIO m) @@ -139,35 +147,45 @@ writeTVar r (f v) runMState' m s - +-} -- | Modify the MState, block all other threads from accessing the state in the --- meantime. +-- meantime (using `atomically` from the "Control.Concurrent.STM" library). modifyM :: (MonadIO m) => (t -> t) -> MState t m () modifyM f = MState $ \(t,_) -> liftIO . atomically $ do v <- readTVar t writeTVar t (f v) - -- | Start a new thread, using the `fork` function from the `Forkable` type -- class. When using this function, the main process will wait for all child --- processes to finish. +-- processes to finish (if desired). forkM :: Forkable m => MState t m () -- ^ State action to be forked -> MState t m ThreadId forkM m = MState $ \s@(_,c) -> do - -- Add new thread MVar to our waiting channel w <- liftIO newEmptyTMVarIO + + tid <- fork $ + -- Use `finally` to make sure our TMVar gets filled + runMState' m s `finally` liftIO (atomically $ putTMVar w ()) + + -- Add the new thread to our waiting TVar liftIO . atomically $ do r <- readTVar c - writeTVar c (w:r) + writeTVar c ((tid,w):r) - -- Use `finally` to make sure our TMVar gets filled - fork $ - runMState' m s `finally` liftIO (atomically $ putTMVar w ()) + return tid +-- | Kill all threads in the current `MState` application. +killMState :: Forkable m => MState t m () +killMState = MState $ \(_,tv) -> do + tms <- liftIO . atomically $ readTVar tv + -- run this in a new thread so it doesn't kill itself + _ <- liftIO . forkIO $ + mapM_ (killThread . fst) tms + return () -------------------------------------------------------------------------------- -- Monad instances @@ -196,7 +214,6 @@ instance (MonadFix m) => MonadFix (MState t m) where mfix f = MState $ \s -> mfix $ \a -> runMState' (f a) s - -------------------------------------------------------------------------------- -- mtl instances -------------------------------------------------------------------------------- @@ -226,7 +243,31 @@ listen m = MState $ listen . runMState' m pass m = MState $ pass . runMState' m +-------------------------------------------------------------------------------- +-- MonadPeel instances +-------------------------------------------------------------------------------- +instance MonadTransPeel (MState t) where + peel = MState $ \t -> return $ \m -> do + a <- runMState' m t + return $ return a + +instance MonadPeelIO m => MonadPeelIO (MState t m) where + peelIO = liftPeel peelIO + +-------------------------------------------------------------------------------- +-- Forkable instances +-------------------------------------------------------------------------------- + +instance Forkable IO where + fork = forkIO + +instance Forkable m => Forkable (ReaderT s m) where + fork newT = ask >>= lift . fork . runReaderT newT + +instance (Error e, Forkable m) => Forkable (ErrorT e m) where + fork newT = lift . fork $ runErrorT newT >> return () + {- $example Example usage: @@ -243,20 +284,22 @@ > > incTwice :: MyState () > incTwice = do -> -> -- First increase in the current thread +> -- increase in the current thread > inc > -- This thread should get killed before it can "inc" our state: -> kill =<< forkM incDelayed -> -- Second increase with a small delay in a forked thread -> forkM incDelayed -> +> t_id <- forkM $ do +> delay 2 +> inc +> -- Second increase with a small delay in a forked thread, killing the +> -- thread above +> forkM $ do +> delay 1 +> inc +> kill t_id > return () -> > where -> inc = modifyM (+1) -> kill = liftIO . killThread -> incDelayed = do liftIO $ threadDelay 2000000 -> inc +> inc = modifyM (+1) +> kill = liftIO . killThread +> delay = liftIO . threadDelay . (*1000000) -- in seconds -}