diff --git a/Control/Concurrent/Broadcast.hs b/Control/Concurrent/Broadcast.hs --- a/Control/Concurrent/Broadcast.hs +++ b/Control/Concurrent/Broadcast.hs @@ -8,13 +8,13 @@ -- Maintainer : Bas van Dijk -- , Roel van Dijk -- --- A Broadcast variable is a mechanism for communication between --- threads. Multiple reader threads can wait until a broadcaster thread writes a --- signal. The readers block until the signal is received. When the broadcaster --- sends the signal all readers are woken. +-- A 'Broadcast' is a mechanism for communication between threads. Multiple +-- @'listen'ers@ wait until a broadcaster @'broadcast's@ a value. The listeners +-- block until the value is received. When the broadcaster broadcasts a value +-- all listeners are woken. -- -- All functions are /exception safe/. Throwing asynchronous exceptions will not --- compromise the internal state of a 'Broadcast' variable. +-- compromise the internal state of a broadcast. -- -- This module is designed to be imported qualified. We suggest importing it -- like: @@ -27,13 +27,20 @@ module Control.Concurrent.Broadcast ( Broadcast + + -- * Creating broadcasts , new - , newWritten - , read - , tryRead - , readTimeout - , write - , clear + , newBroadcasting + + -- * Listening to broadcasts + , listen + , tryListen + , listenTimeout + + -- * Broadcasting + , broadcast + , signal + , silence ) where @@ -42,19 +49,18 @@ ------------------------------------------------------------------------------- -- from base: -import Control.Applicative ( (<$>) ) -import Control.Arrow ( first ) -import Control.Monad ( (>>=), (>>), return, fmap, forM_, fail ) +import Control.Monad ( (>>=), (>>), return, forM_, fail, when ) import Control.Concurrent.MVar ( MVar, newMVar, newEmptyMVar , takeMVar, putMVar, readMVar, modifyMVar_ ) -import Control.Exception ( block, unblock ) +import Control.Exception ( block, onException ) import Data.Eq ( Eq ) +import Data.Either ( Either(Left ,Right), either ) import Data.Function ( ($), const ) +import Data.Functor ( fmap, (<$>) ) import Data.List ( delete, length ) -import Data.Maybe ( Maybe(Nothing, Just) ) +import Data.Maybe ( Maybe(Nothing, Just), isNothing ) import Data.Ord ( Ord, max ) -import Data.Tuple ( fst ) import Data.Typeable ( Typeable ) import Prelude ( Integer, fromInteger, seq ) import System.IO ( IO ) @@ -66,95 +72,129 @@ import Utils ( purelyModifyMVar ) import Control.Concurrent.Timeout ( timeout ) + ------------------------------------------------------------------------------- -- Broadcast ------------------------------------------------------------------------------- --- | A broadcast variable. It can be thought of as a box, which may be empty of --- full. -newtype Broadcast α = Broadcast {unBroadcast ∷ MVar (Maybe α, [MVar α])} +{-| +A broadcast is in one of two possible states: + +* \"Silent\": @'listen'ing@ to the broadcast will block until a value is +@'broadcast'ed@. + +* \"Broadcasting @x@\": @'listen'ing@ to the broadcast will return the value @x@ +without blocking. +-} +newtype Broadcast α = Broadcast {unBroadcast ∷ MVar (Either [MVar α] α)} deriving (Eq, Typeable) --- | Create a new empty 'Broadcast' variable. +-- | @new@ Creates a broadcast in the \"silent\" state. new ∷ IO (Broadcast α) -new = Broadcast <$> newMVar (Nothing, []) +new = Broadcast <$> newMVar (Left []) --- | Create a new 'Broadcast' variable containing an initial value. -newWritten ∷ α → IO (Broadcast α) -newWritten x = Broadcast <$> newMVar (Just x, []) +-- | @newBroadcasting x@ Creates a broadcast in the \"broadcasting @x@\" state. +newBroadcasting ∷ α → IO (Broadcast α) +newBroadcasting x = Broadcast <$> newMVar (Right x) -{-| Read the value of a 'Broadcast' variable. +{-| +Listen to a broadcast. -If the 'Broadcast' variable contains a value it will be returned immediately, -otherwise it will block until another thread 'write's a value to the 'Broadcast' -variable. +* If the broadcast is \"broadcasting @x@\", @listen@ will return @x@ +immediately. + +* If the broadcast is \"silent\", @listen@ will block until another thread +@'broadcast's@ a value to the broadcast. -} -read ∷ Broadcast α → IO α -read (Broadcast mv) = block $ do - t@(mx, ls) ← takeMVar mv +listen ∷ Broadcast α → IO α +listen (Broadcast mv) = block $ do + mx ← takeMVar mv case mx of - Nothing → do l ← newEmptyMVar - putMVar mv (mx, l:ls) + Left ls → do l ← newEmptyMVar + putMVar mv $ Left $ l:ls takeMVar l - Just x → do putMVar mv t + Right x → do putMVar mv mx return x -{-| Try to read the value of a 'Broadcast' variable; non blocking. +{-| +Try to listen to a broadcast; non blocking. -Like 'read' but doesn't block. Returns 'Just' the contents of the 'Broadcast' if -it wasn't empty, 'Nothing' otherwise. +* If the broadcast is \"broadcasting @x@\", @tryListen@ will return 'Just' @x@ +immediately. + +* If the broadcast is \"silent\", @tryListen@ returns 'Nothing' immediately. -} -tryRead ∷ Broadcast α → IO (Maybe α) -tryRead = fmap fst ∘ readMVar ∘ unBroadcast +tryListen ∷ Broadcast α → IO (Maybe α) +tryListen = fmap (either (const Nothing) Just) ∘ readMVar ∘ unBroadcast -{-| Read the value of a 'Broadcast' variable if it is available within a given -amount of time. +{-| +Listen to a broadcast if it is available within a given amount of time. -Like 'read', but with a timeout. A return value of 'Nothing' indicates a timeout -occurred. +Like 'listen', but with a timeout. A return value of 'Nothing' indicates a +timeout occurred. -The timeout is specified in microseconds. A timeout of 0 μs will cause -the function to return 'Nothing' without blocking in case the 'Broadcast' was -empty. Negative timeouts are treated the same as a timeout of 0 μs. +The timeout is specified in microseconds. + +If the broadcast is \"silent\" and a timeout of 0 μs is specified the +function returns 'Nothing' without blocking. + +Negative timeouts are treated the same as a timeout of 0 μs. -} -readTimeout ∷ Broadcast α → Integer → IO (Maybe α) -readTimeout (Broadcast mv) time = block $ do - t@(mx, ls) ← takeMVar mv +listenTimeout ∷ Broadcast α → Integer → IO (Maybe α) +listenTimeout (Broadcast mv) time = block $ do + mx ← takeMVar mv case mx of - Nothing → do l ← newEmptyMVar - putMVar mv (mx, l:ls) - my ← unblock $ timeout (max time 0) (takeMVar l) - case my of - Nothing → do deleteReader l mv - return my - Just _ → return my - Just _ → do putMVar mv t - return mx + Left ls → do l ← newEmptyMVar + putMVar mv $ Left $ l:ls + my ← timeout (max time 0) (takeMVar l) + `onException` deleteReader l + when (isNothing my) (deleteReader l) + return my + Right x → do putMVar mv mx + return $ Just x + where + deleteReader l = do mx ← takeMVar mv + case mx of + Left ls → let ls' = delete l ls + in length ls' `seq` putMVar mv (Left ls') + Right _ → putMVar mv mx -{-| Write a new value into a 'Broadcast' variable. +{-| +Broadcast a value. -If the variable is empty any threads that are reading from the variable will be -woken. If the variable is full its contents will simply be overwritten. +@broadcast b x@ changes the state of the broadcast @b@ to \"broadcasting @x@\". + +If the broadcast was \"silent\" all threads that are @'listen'ing@ to the +broadcast will be woken. -} -write ∷ Broadcast α → α → IO () -write (Broadcast mv) x = - modifyMVar_ mv $ \(_, ls) → do - forM_ ls $ \l → putMVar l x - return (Just x, []) +broadcast ∷ Broadcast α → α → IO () +broadcast (Broadcast mv) x = modifyMVar_ mv $ \mx → do + case mx of + Left ls → do forM_ ls (`putMVar` x) + return $ Right x + Right _ → return $ Right x +{-| +Broadcast a value before becoming \"silent\". --- | Clear the contents of a 'Broadcast' variable. -clear ∷ Broadcast α → IO () -clear (Broadcast mv) = purelyModifyMVar mv $ first $ const Nothing +The state of the broadcast is changed to \"silent\" after all threads that are +@'listen'ing@ to the broadcast are woken and resume with the signalled value. +The semantics of signal are equivalent to the following definition: -------------------------------------------------------------------------------- +@ + signal b x = 'block' $ 'broadcast' b x >> 'silence' b +@ +-} +signal ∷ Broadcast α → α → IO () +signal (Broadcast mv) x = modifyMVar_ mv $ \mx → do + case mx of + Left ls → do forM_ ls (`putMVar` x) + return $ Left [] + Right _ → return $ Left [] -deleteReader ∷ MVar α → MVar (Maybe α, [MVar α]) → IO () -deleteReader l mv = do - (mx, ls) ← takeMVar mv - let ls' = delete l ls - length ls' `seq` putMVar mv (mx, ls') +-- | Set a broadcast to the \"silent\" state. +silence ∷ Broadcast α → IO () +silence (Broadcast mv) = purelyModifyMVar mv $ either Left $ const $ Left [] -- The End --------------------------------------------------------------------- - diff --git a/Control/Concurrent/Event.hs b/Control/Concurrent/Event.hs --- a/Control/Concurrent/Event.hs +++ b/Control/Concurrent/Event.hs @@ -1,4 +1,8 @@ -{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} +{-# LANGUAGE CPP + , DeriveDataTypeable + , NoImplicitPrelude + , UnicodeSyntax + #-} ------------------------------------------------------------------------------- -- | @@ -11,10 +15,10 @@ -- An Event is a simple mechanism for communication between threads: one thread -- signals an event and other threads wait for it. -- --- Each event has an internal state which is either \"Set\" or \"Cleared\". This --- state can be changed with the corresponding functions 'set' and 'clear'. The --- 'wait' function blocks until the state is \"Set\". An important property of --- setting an event is that /all/ threads waiting for it are woken. +-- An event has a state which is either \"set\" or \"cleared\". This state can +-- be changed with the corresponding functions 'set' and 'clear'. The 'wait' +-- function blocks until the state is \"set\". An important property of setting +-- an event is that /all/ threads waiting for it are woken. -- -- It was inspired by the Python @Event@ object. See: -- @@ -32,13 +36,20 @@ module Control.Concurrent.Event ( Event + + -- * Creating events , new , newSet + + -- * Waiting for events , wait , waitTimeout + , isSet + + -- * Setting events , set + , signal , clear - , isSet ) where @@ -47,12 +58,14 @@ ------------------------------------------------------------------------------- -- from base -import Control.Applicative ( (<$>) ) -import Control.Monad ( fmap ) import Data.Bool ( Bool(..) ) import Data.Eq ( Eq ) +import Data.Functor ( fmap, (<$>) ) import Data.Maybe ( isJust ) import Data.Typeable ( Typeable ) +#ifdef __HADDOCK__ +import Control.Exception ( block ) +#endif import Prelude ( Integer ) import System.IO ( IO ) @@ -62,68 +75,107 @@ -- from concurrent-extra import Control.Concurrent.Broadcast ( Broadcast ) import qualified Control.Concurrent.Broadcast as Broadcast - ( new, newWritten, read, tryRead, readTimeout, write, clear ) + ( new, newBroadcasting + , listen, tryListen, listenTimeout + , broadcast, signal, silence + ) ------------------------------------------------------------------------------- -- Events ------------------------------------------------------------------------------- --- | An event is in one of two possible states: \"Set\" or \"Cleared\". +-- | An event is in one of two possible states: \"set\" or \"cleared\". newtype Event = Event {evBroadcast ∷ Broadcast ()} deriving (Eq, Typeable) --- | Create an event in the \"Cleared\" state. + +------------------------------------------------------------------------------- +-- Creating events +------------------------------------------------------------------------------- + +-- | Create an event in the \"cleared\" state. new ∷ IO Event new = Event <$> Broadcast.new --- | Create an event in the \"Set\" state. +-- | Create an event in the \"set\" state. newSet ∷ IO Event -newSet = Event <$> Broadcast.newWritten () +newSet = Event <$> Broadcast.newBroadcasting () -{-| Block until the event is 'set'. -If the state of the event is already \"Set\" this function will return +------------------------------------------------------------------------------- +-- Waiting for events +------------------------------------------------------------------------------- + +{-| +Block until the event is 'set'. + +If the state of the event is already \"set\" this function will return immediately. Otherwise it will block until another thread calls 'set'. -You can also stop a thread that is waiting for an event by throwing an -asynchronous exception. +(You can also resume a thread that is waiting for an event by throwing an +asynchronous exception.) -} wait ∷ Event → IO () -wait = Broadcast.read ∘ evBroadcast +wait = Broadcast.listen ∘ evBroadcast -{-| Block until the event is 'set' or until a timer expires. +{-| +Block until the event is 'set' or until a timer expires. Like 'wait', but with a timeout. A return value of 'False' indicates a timeout occurred. -The timeout is specified in microseconds. A timeout of 0 μs will cause the -function to return 'False' without blocking in case the event state is -\"Cleared\". Negative timeouts are treated the same as a timeout of 0 μs. +The timeout is specified in microseconds. + +If the event is \"cleared\" and a timeout of 0 μs is specified the +function returns 'False' without blocking. + +Negative timeouts are treated the same as a timeout of 0 μs. -} waitTimeout ∷ Event → Integer → IO Bool -waitTimeout ev time = isJust <$> Broadcast.readTimeout (evBroadcast ev) time +waitTimeout ev time = isJust <$> Broadcast.listenTimeout (evBroadcast ev) time -{-| Changes the state of the event to \"Set\". All threads that where waiting +{-| +Returns 'True' if the state of the event is \"set\" and 'False' if the state +is \"cleared\". + +Notice that this is only a snapshot of the state. By the time a program reacts +on its result it may already be out of date. +-} +isSet ∷ Event → IO Bool +isSet = fmap isJust ∘ Broadcast.tryListen ∘ evBroadcast + + +------------------------------------------------------------------------------- +-- Setting events +------------------------------------------------------------------------------- + +{-| +Changes the state of the event to \"set\". All threads that where waiting for this event are woken. Threads that 'wait' after the state is changed to -\"Set\" will not block at all. +\"set\" will not block at all. -} set ∷ Event → IO () -set ev = Broadcast.write (evBroadcast ev) () +set ev = Broadcast.broadcast (evBroadcast ev) () -{-| Changes the state of the event to \"Cleared\". Threads that 'wait' after the -state is changed to \"Cleared\" will block until the state is changed to \"Set\". --} -clear ∷ Event → IO () -clear = Broadcast.clear ∘ evBroadcast +{-| +Changes the state to \"cleared\" after all threads that where waiting for this +event are woken. Threads that 'wait' after a @signal@ will block until the event +is 'set' again. -{-| Returns 'True' if the state of the event is \"Set\" and 'False' if the state -is \"Cleared\". +The semantics of signal are equivalent to the following definition: -Notice that this is only a snapshot of the state. By the time a program reacts -on its result it may already be out of date. +@ + signal e = 'block' $ 'set' e >> 'clear' e +@-} +signal ∷ Event → IO () +signal ev = Broadcast.signal (evBroadcast ev) () + +{-| +Changes the state of the event to \"cleared\". Threads that 'wait' after the +state is changed to \"cleared\" will block until the state is changed to \"set\". -} -isSet ∷ Event → IO Bool -isSet = fmap isJust ∘ Broadcast.tryRead ∘ evBroadcast +clear ∷ Event → IO () +clear = Broadcast.silence ∘ evBroadcast -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Lock.hs b/Control/Concurrent/Lock.hs --- a/Control/Concurrent/Lock.hs +++ b/Control/Concurrent/Lock.hs @@ -31,17 +31,21 @@ module Control.Concurrent.Lock ( Lock + -- * Creating locks , new , newAcquired + -- * Locking and unlocking , acquire , tryAcquire , release + -- * Convenience functions , with , tryWith , wait + -- * Querying locks , locked ) where @@ -52,17 +56,18 @@ -------------------------------------------------------------------------------- -- from base: -import Control.Applicative ( (<$>), liftA2 ) +import Control.Applicative ( liftA2 ) import Control.Concurrent.MVar ( MVar, newMVar, newEmptyMVar , takeMVar, tryTakeMVar , tryPutMVar , isEmptyMVar ) import Control.Exception ( block, bracket_, finally ) -import Control.Monad ( Monad, return, (>>=), (>>), fail, when, fmap ) +import Control.Monad ( Monad, return, (>>=), (>>), fail, when ) import Data.Bool ( Bool, not ) import Data.Eq ( Eq ) import Data.Function ( ($) ) +import Data.Functor ( fmap, (<$>) ) import Data.Maybe ( Maybe(Nothing, Just), isJust ) import Data.Typeable ( Typeable ) import Prelude ( error ) @@ -79,6 +84,11 @@ -- | A lock is in one of two states: \"locked\" or \"unlocked\". newtype Lock = Lock {un ∷ MVar ()} deriving (Eq, Typeable) + +-------------------------------------------------------------------------------- +-- Creating locks +-------------------------------------------------------------------------------- + -- | Create a lock in the \"unlocked\" state. new ∷ IO Lock new = Lock <$> newMVar () @@ -87,7 +97,12 @@ newAcquired ∷ IO Lock newAcquired = Lock <$> newEmptyMVar -{-| + +-------------------------------------------------------------------------------- +-- Locking and unlocking +-------------------------------------------------------------------------------- + +{-| Acquires the 'Lock'. Blocks if another thread has acquired the 'Lock'. @acquire@ behaves as follows: @@ -95,7 +110,7 @@ * When the state is \"unlocked\" @acquire@ changes the state to \"locked\". * When the state is \"locked\" @acquire@ /blocks/ until a call to 'release' in -another thread wakes the calling thread. Upon awakening it will change to state +another thread wakes the calling thread. Upon awakening it will change the state to \"locked\". There are two further important properties of @acquire@: @@ -112,7 +127,7 @@ acquire ∷ Lock → IO () acquire = takeMVar ∘ un -{-| +{-| A non-blocking 'acquire'. * When the state is \"unlocked\" @tryAcquire@ changes the state to \"locked\" @@ -124,7 +139,7 @@ tryAcquire ∷ Lock → IO Bool tryAcquire = fmap isJust ∘ tryTakeMVar ∘ un -{-| +{-| @release@ changes the state to \"unlocked\" and returns immediately. Note that it is an error to release a lock in the \"unlocked\" state! @@ -137,7 +152,12 @@ b ← tryPutMVar mv () when (not b) $ error "Control.Concurrent.Lock.release: Can't release unlocked Lock!" -{-| + +-------------------------------------------------------------------------------- +-- Convenience functions +-------------------------------------------------------------------------------- + +{-| A convenience function which first acquires the lock and then performs the computation. When the computation terminates, whether normally or by raising an exception, the lock is released. @@ -147,7 +167,7 @@ with ∷ Lock → IO a → IO a with = liftA2 bracket_ acquire release -{-| +{-| A non-blocking 'with'. @tryWith@ is a convenience function which first tries to acquire the lock. If that fails, 'Nothing' is returned. If it succeeds, the computation is performed. When the computation terminates, whether normally or @@ -176,7 +196,12 @@ wait ∷ Lock → IO () wait l = block $ acquire l >> release l -{-| + +-------------------------------------------------------------------------------- +-- Querying locks +-------------------------------------------------------------------------------- + +{-| Determines if the lock is in the \"locked\" state. Note that this is only a snapshot of the state. By the time a program reacts diff --git a/Control/Concurrent/Lock/Test.hs b/Control/Concurrent/Lock/Test.hs --- a/Control/Concurrent/Lock/Test.hs +++ b/Control/Concurrent/Lock/Test.hs @@ -11,9 +11,10 @@ -- from base: import Control.Concurrent ( forkIO ) -import Control.Monad ( (>>=), fail, (>>), fmap ) +import Control.Monad ( (>>=), fail, (>>) ) import Data.Bool ( not ) import Data.Function ( ($) ) +import Data.Functor ( fmap ) import Prelude ( fromInteger ) -- from base-unicode-symbols: diff --git a/Control/Concurrent/RLock.hs b/Control/Concurrent/RLock.hs --- a/Control/Concurrent/RLock.hs +++ b/Control/Concurrent/RLock.hs @@ -36,17 +36,21 @@ module Control.Concurrent.RLock ( RLock + -- * Creating reentrant locks , new , newAcquired + -- * Locking and unlocking , acquire , tryAcquire , release + -- * Convenience functions , with , tryWith , wait + -- * Querying reentrant locks , State , state @@ -58,14 +62,15 @@ -------------------------------------------------------------------------------- -- from base: -import Control.Applicative ( (<$>), liftA2 ) +import Control.Applicative ( liftA2 ) import Control.Concurrent ( ThreadId, myThreadId ) import Control.Concurrent.MVar ( MVar, newMVar, takeMVar, readMVar, putMVar ) import Control.Exception ( block, bracket_, finally ) -import Control.Monad ( Monad, return, (>>=), fail, (>>), fmap ) +import Control.Monad ( Monad, return, (>>=), fail, (>>) ) import Data.Bool ( Bool(False, True), otherwise ) import Data.Eq ( Eq ) import Data.Function ( ($) ) +import Data.Functor ( fmap, (<$>) ) import Data.Maybe ( Maybe(Nothing, Just) ) import Data.Tuple ( fst ) import Data.Typeable ( Typeable ) @@ -101,17 +106,22 @@ * 'Nothing' indicates an \"unlocked\" state. -* @'Just' (tid, n)@ indicates a \"locked\" state where the thread identified by @tid@ -acquired the lock @n@ times. +* @'Just' (tid, n)@ indicates a \"locked\" state where the thread identified by +@tid@ acquired the lock @n@ times. -} type State = Maybe (ThreadId, Integer) + +-------------------------------------------------------------------------------- +-- * Creating reentrant locks +-------------------------------------------------------------------------------- + -- | Create a reentrant lock in the \"unlocked\" state. new ∷ IO RLock new = do lock ← Lock.new RLock <$> newMVar (Nothing, lock) -{-| +{-| Create a reentrant lock in the \"locked\" state (with the current thread as owner and an acquired count of 1). -} @@ -120,7 +130,12 @@ lock ← Lock.newAcquired RLock <$> newMVar (Just (myTID, 1), lock) -{-| + +-------------------------------------------------------------------------------- +-- * Locking and unlocking +-------------------------------------------------------------------------------- + +{-| Acquires the 'RLock'. Blocks if another thread has acquired the 'RLock'. @acquire@ behaves as follows: @@ -162,7 +177,7 @@ acq in acq -{-| +{-| A non-blocking 'acquire'. * When the state is \"unlocked\" @tryAcquire@ changes the state to \"locked\" @@ -215,6 +230,11 @@ in putMVar mv (Just (tid, pn), lock) | otherwise → err "Calling thread does not own the RLock!" + +-------------------------------------------------------------------------------- +-- * Convenience functions +-------------------------------------------------------------------------------- + {-| A convenience function which first acquires the lock and then performs the computation. When the computation terminates, whether normally or by raising an exception, the lock is released. @@ -224,7 +244,7 @@ with ∷ RLock → IO α → IO α with = liftA2 bracket_ acquire release -{-| +{-| A non-blocking 'with'. @tryWith@ is a convenience function which first tries to acquire the lock. If that fails, 'Nothing' is returned. If it succeeds, the computation is performed. When the computation terminates, whether normally or @@ -253,7 +273,12 @@ wait ∷ RLock → IO () wait l = block $ acquire l >> release l -{-| + +-------------------------------------------------------------------------------- +-- * Querying reentrant locks +-------------------------------------------------------------------------------- + +{-| Determine the state of the reentrant lock. Note that this is only a snapshot of the state. By the time a program reacts on diff --git a/Control/Concurrent/ReadWriteLock.hs b/Control/Concurrent/ReadWriteLock.hs --- a/Control/Concurrent/ReadWriteLock.hs +++ b/Control/Concurrent/ReadWriteLock.hs @@ -1,8 +1,6 @@ -{-# LANGUAGE BangPatterns - , DeriveDataTypeable +{-# LANGUAGE DeriveDataTypeable , NamedFieldPuns , NoImplicitPrelude - , TupleSections , UnicodeSyntax #-} @@ -36,23 +34,28 @@ module Control.Concurrent.ReadWriteLock ( RWLock + -- *Creating Read-Write Locks , new , newAcquiredRead , newAcquiredWrite + -- *Read access -- **Blocking , acquireRead , releaseRead , withRead + , waitRead -- **Non-blocking , tryAcquireRead , tryWithRead + -- *Write access -- **Blocking , acquireWrite , releaseWrite , withWrite + , waitWrite -- **Non-blocking , tryAcquireWrite , tryWithWrite @@ -64,22 +67,19 @@ ------------------------------------------------------------------------------- -- from base -import Control.Applicative ( (<$>), liftA2 ) -import Control.Concurrent.MVar ( MVar, newMVar, takeMVar, putMVar - , modifyMVar_, swapMVar - ) +import Control.Applicative ( liftA2, liftA3 ) +import Control.Concurrent.MVar ( MVar, newMVar, takeMVar, putMVar, swapMVar ) import Control.Exception ( block, bracket_, finally ) -import Control.Monad ( return, (>>=), return, fail, (>>) - , when, liftM3 - ) +import Control.Monad ( return, (>>=), return, fail, (>>), when ) import Data.Bool ( Bool(False, True) ) import Data.Char ( String ) import Data.Eq ( Eq, (==) ) -import Data.Function ( ($), const, on ) +import Data.Function ( ($), on ) +import Data.Functor ( (<$>) ) import Data.Int ( Int ) import Data.Maybe ( Maybe(Nothing, Just) ) import Data.Typeable ( Typeable ) -import Prelude ( fromInteger, succ, pred, error ) +import Prelude ( ($!), fromInteger, succ, pred, error ) import System.IO ( IO ) -- from base-unicode-symbols @@ -89,6 +89,7 @@ -- from concurrent-extra import Control.Concurrent.Lock ( Lock ) import qualified Control.Concurrent.Lock as Lock + ( new, newAcquired, acquire, tryAcquire, release, wait ) import Utils ( void ) @@ -97,7 +98,8 @@ -- Read Write Lock ------------------------------------------------------------------------------- -{-| Multiple-reader, single-writer lock. Is in one of three states: +{-| +Multiple-reader, single-writer lock. Is in one of three states: * \"Free\": Read or write access can be acquired without blocking. @@ -117,100 +119,105 @@ -- | Internal state of the 'RWLock'. data State = Free | Read Int | Write -{-| Create a new 'RWLock' in the \"Free\" state; either read or write access can -be acquired without blocking. --} + +------------------------------------------------------------------------------- +-- * Creating Read-Write Locks +------------------------------------------------------------------------------- + +-- | Create a new 'RWLock' in the \"free\" state; either read or write access +-- can be acquired without blocking. new ∷ IO RWLock -new = liftM3 RWLock (newMVar Free) +new = liftA3 RWLock (newMVar Free) Lock.new Lock.new -{-| Create a new 'RWLock' in the \"Read\" state; only read can be acquired -without blocking. --} +-- | Create a new 'RWLock' in the \"read\" state; only read can be acquired +-- without blocking. newAcquiredRead ∷ IO RWLock -newAcquiredRead = liftM3 RWLock (newMVar $ Read 1) +newAcquiredRead = liftA3 RWLock (newMVar $ Read 1) Lock.newAcquired Lock.new -{-| Create a new 'RWLock' in the \"Write\" state; either acquiring read or write -will block. --} +-- | Create a new 'RWLock' in the \"write\" state; either acquiring read or +-- write will block. newAcquiredWrite ∷ IO RWLock -newAcquiredWrite = liftM3 RWLock (newMVar Write) +newAcquiredWrite = liftA3 RWLock (newMVar Write) Lock.new Lock.newAcquired -{-| Acquire the read lock. +------------------------------------------------------------------------------- +-- * Read access +------------------------------------------------------------------------------- + +{-| +Acquire the read lock. + Blocks if another thread has acquired write access. If @acquireRead@ terminates -without throwing an exception the state of the 'RWLock' will be \"Read\". +without throwing an exception the state of the 'RWLock' will be \"read\". Implementation note: Throws an exception when more than (maxBound :: Int) simultaneous threads acquire the read lock. But that is unlikely. -} acquireRead ∷ RWLock → IO () -acquireRead (RWLock {state, readLock, writeLock}) = block $ do - st ← takeMVar state - case st of - Free → do Lock.acquire readLock - putMVar state (Read 1) - Read n → let !sn = succ n - in putMVar state (Read sn) - Write → do putMVar state st - Lock.acquire writeLock - modifyMVar_ state ∘ const $ do - Lock.acquire readLock - return $ Read 1 - +acquireRead (RWLock {state, readLock, writeLock}) = block acqRead + where + acqRead = do st ← takeMVar state + case st of + Free → do Lock.acquire readLock + putMVar state $ Read 1 + Read n → putMVar state ∘ Read $! succ n + Write → do putMVar state st + Lock.wait writeLock + acqRead -{-| Try to acquire the read lock; non blocking. +{-| +Try to acquire the read lock; non blocking. Like 'acquireRead', but doesn't block. Returns 'True' if the resulting state is -\"Read\", 'False' otherwise. +\"read\", 'False' otherwise. -} tryAcquireRead ∷ RWLock → IO Bool tryAcquireRead (RWLock {state, readLock}) = block $ do st ← takeMVar state case st of Free → do Lock.acquire readLock - putMVar state (Read 1) + putMVar state $ Read 1 return True - Read n → do let !sn = succ n - putMVar state (Read sn) + Read n → do putMVar state ∘ Read $! succ n return True Write → do putMVar state st return False -{-| Release the read lock. +{-| +Release the read lock. If the calling thread was the last one to relinquish read access the state will -revert to \"Free\". +revert to \"free\". It is an error to release read access to an 'RWLock' which is not in the -\"Read\" state. +\"read\" state. -} releaseRead ∷ RWLock → IO () releaseRead (RWLock {state, readLock}) = block $ do st ← takeMVar state case st of - Free → putMVar state st >> err Read 1 → do Lock.release readLock putMVar state Free - Read n → let !pn = pred n - in putMVar state (Read pn) - Write → putMVar state st >> err - where - err = error $ moduleName ⊕ ".releaseRead: already released" + Read n → putMVar state ∘ Read $! pred n + _ → do putMVar state st + error $ moduleName ⊕ ".releaseRead: already released" -{-| A convenience function wich first acquires read access and then performs the +{-| +A convenience function wich first acquires read access and then performs the computation. When the computation terminates, whether normally or by raising an exception, the read lock is released. -} withRead ∷ RWLock → IO α → IO α withRead = liftA2 bracket_ acquireRead releaseRead -{-| A non-blocking 'withRead'. First tries to acquire the lock. If that fails, +{-| +A non-blocking 'withRead'. First tries to acquire the lock. If that fails, 'Nothing' is returned. If it succeeds, the computation is performed. When the computation terminates, whether normally or by raising an exception, the lock is released and 'Just' the result of the computation is returned. @@ -222,31 +229,52 @@ then Just <$> a `finally` releaseRead l else return Nothing -{-| Acquire the write lock. +{-| +* When the state is \"write\", @waitRead@ /blocks/ until a call to +'releaseWrite' in another thread changes the state to \"free\". +* When the state is \"free\" or \"read\" @waitRead@ returns immediately. + +@waitRead@ does not alter the state of the lock. + +Note that @waitRead@ is just a convenience function defined as: + +@waitRead l = 'block' '$' 'acquireRead' l '>>' 'releaseRead' l@ +-} +waitRead ∷ RWLock → IO () +waitRead l = block $ acquireRead l >> releaseRead l + + +------------------------------------------------------------------------------- +-- *Write access +------------------------------------------------------------------------------- + +{-| +Acquire the write lock. + Blocks if another thread has acquired either read or write access. If @acquireWrite@ terminates without throwing an exception the state of the -'RWLock' will be \"Write\". +'RWLock' will be \"write\". -} acquireWrite ∷ RWLock → IO () -acquireWrite (RWLock {state, readLock, writeLock}) = block $ do - st ← takeMVar state - case st of - Free → do Lock.acquire writeLock - putMVar state Write - Read _ → do putMVar state st - Lock.acquire readLock - modifyMVar_ state ∘ const $ do - Lock.acquire writeLock - return Write - Write → do putMVar state st - Lock.acquire writeLock - void $ swapMVar state Write +acquireWrite (RWLock {state, readLock, writeLock}) = block acqWrite + where + acqWrite = do st ← takeMVar state + case st of + Free → do Lock.acquire writeLock + putMVar state Write + Read _ → do putMVar state st + Lock.wait readLock + acqWrite + Write → do putMVar state st + Lock.acquire writeLock + void $ swapMVar state Write -{-| Try to acquire the write lock; non blocking. +{-| +Try to acquire the write lock; non blocking. Like 'acquireWrite', but doesn't block. Returns 'True' if the resulting state is -\"Write\", 'False' otherwise. +\"write\", 'False' otherwise. -} tryAcquireWrite ∷ RWLock → IO Bool tryAcquireWrite (RWLock {state, writeLock}) = block $ do @@ -259,36 +287,37 @@ return False Write → do putMVar state st b ← Lock.tryAcquire writeLock - when b ∘ void $ swapMVar state Write + when b $ void $ swapMVar state Write return b -{-| Release the write lock. +{-| +Release the write lock. If @releaseWrite@ terminates without throwing an exception the state will be -\"Free\". +\"free\". It is an error to release write access to an 'RWLock' which is not in the -\"Write\" state. +\"write\" state. -} releaseWrite ∷ RWLock → IO () releaseWrite (RWLock {state, writeLock}) = block $ do st ← takeMVar state case st of - Free → putMVar state st >> err - Read _ → putMVar state st >> err - Write → do Lock.release writeLock - putMVar state Free - where - err = error $ moduleName ⊕ ".releaseWrite: already released" + Write → do Lock.release writeLock + putMVar state Free + _ → do putMVar state st + error $ moduleName ⊕ ".releaseWrite: already released" -{-| A convenience function wich first acquires write access and then performs +{-| +A convenience function wich first acquires write access and then performs the computation. When the computation terminates, whether normally or by raising an exception, the write lock is released. -} withWrite ∷ RWLock → IO α → IO α withWrite = liftA2 bracket_ acquireWrite releaseWrite -{-| A non-blocking 'withWrite'. First tries to acquire the lock. If that fails, +{-| +A non-blocking 'withWrite'. First tries to acquire the lock. If that fails, 'Nothing' is returned. If it succeeds, the computation is performed. When the computation terminates, whether normally or by raising an exception, the lock is released and 'Just' the result of the computation is returned. @@ -299,6 +328,21 @@ if acquired then Just <$> a `finally` releaseWrite l else return Nothing + +{-| +* When the state is \"write\" or \"read\" @waitWrite@ /blocks/ until a call to +'releaseWrite' or 'releaseRead' in another thread changes the state to \"free\". + +* When the state is \"free\" @waitWrite@ returns immediately. + +@waitWrite@ does not alter the state of the lock. + +Note that @waitWrite@ is just a convenience function defined as: + +@waitWrite l = 'block' '$' 'acquireWrite' l '>>' 'releaseWrite' l@ +-} +waitWrite ∷ RWLock → IO () +waitWrite l = block $ acquireWrite l >> releaseWrite l moduleName ∷ String moduleName = "Control.Concurrent.ReadWriteLock" diff --git a/Control/Concurrent/ReadWriteLock/Test.hs b/Control/Concurrent/ReadWriteLock/Test.hs --- a/Control/Concurrent/ReadWriteLock/Test.hs +++ b/Control/Concurrent/ReadWriteLock/Test.hs @@ -1,6 +1,4 @@ -{-# LANGUAGE NoImplicitPrelude - , UnicodeSyntax - #-} +{-# LANGUAGE NoImplicitPrelude , UnicodeSyntax #-} module Control.Concurrent.ReadWriteLock.Test ( tests ) where @@ -10,23 +8,30 @@ ------------------------------------------------------------------------------- -- from base: -import Control.Concurrent ( ) +import Control.Monad ( (>>=), (>>), fail ) +import Control.Concurrent ( forkIO, threadDelay ) +import Data.Function ( ($) ) +import Prelude ( fromInteger ) -- from base-unicode-symbols: -import Prelude.Unicode ( ) +import Prelude.Unicode ( (⋅) ) -- from concurrent-extra: -import qualified Control.Concurrent.ReadWriteLock as RWLock ( ) -import TestUtils ( ) +import qualified Control.Concurrent.ReadWriteLock as RWLock + ( new, acquireWrite, acquireRead, releaseWrite, releaseRead ) +import TestUtils ( within, a_moment ) + +import Utils ( void ) + -- from HUnit: -import Test.HUnit ( ) +import Test.HUnit ( Assertion, assert ) -- from test-framework: import Test.Framework ( Test ) -- from test-framework-hunit: -import Test.Framework.Providers.HUnit ( ) +import Test.Framework.Providers.HUnit ( testCase ) ------------------------------------------------------------------------------- @@ -34,7 +39,53 @@ ------------------------------------------------------------------------------- tests ∷ [Test] -tests = [] +tests = [ testCase "test1" test1 + , testCase "test2" test2 + ] + +test1 ∷ Assertion +test1 = assert $ within (10 ⋅ a_moment) $ do + -- Create a new read-write-lock (in the "Free" state): + rwl ← RWLock.new + + -- Put the read-write-lock in the "Write" state: + RWLock.acquireWrite rwl + + -- Fork a thread that releases the write-lock after a moment: + void $ forkIO $ threadDelay a_moment >> RWLock.releaseWrite rwl + + -- This blocks until the write-lock is released in the above thread. + RWLock.acquireRead rwl + + -- Release the read-lock so that the read-write-lock can either be + -- acquired again by 'acquireRead' or 'acquireWrite': + RWLock.releaseRead rwl + + -- The read-write-lock should now be in the "Free" state so the + -- following shouldn't deadlock: + RWLock.acquireWrite rwl + +test2 ∷ Assertion +test2 = assert $ within (10 ⋅ a_moment) $ do + -- Create a new read-write-lock (in the "Free" state): + rwl ← RWLock.new + + -- Put the read-write-lock in the "Read" state: + RWLock.acquireRead rwl + + -- Fork a thread that releases the read-lock after a moment: + void $ forkIO $ threadDelay a_moment >> RWLock.releaseRead rwl + + -- This blocks until the read-lock is released in the above thread. + RWLock.acquireWrite rwl + + -- Release the write-lock so that the read-write-lock can either be + -- acquired again by 'acquireRead' or 'acquireWrite': + RWLock.releaseWrite rwl + + -- The read-write-lock should now be in the "Free" state so the + -- following shouldn't deadlock: + RWLock.acquireRead rwl -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/ReadWriteVar.hs b/Control/Concurrent/ReadWriteVar.hs --- a/Control/Concurrent/ReadWriteVar.hs +++ b/Control/Concurrent/ReadWriteVar.hs @@ -62,10 +62,12 @@ ------------------------------------------------------------------------------- -- from base -import Control.Monad ( (>>=), fmap, liftM2 ) +import Control.Applicative ( liftA2 ) +import Control.Monad ( (>>=) ) import Data.Bool ( Bool(..) ) import Data.Eq ( Eq, (==) ) import Data.Function ( ($), on ) +import Data.Functor ( fmap ) import Data.Maybe ( Maybe(..), isJust ) import Data.IORef ( IORef, newIORef, readIORef ) import Data.Typeable ( Typeable ) @@ -99,7 +101,7 @@ -- | Create a new 'RWVar'. new ∷ α → IO (RWVar α) -new = liftM2 RWVar RWLock.new ∘ newIORef +new = liftA2 RWVar RWLock.new ∘ newIORef {-| Execute an action that operates on the contents of the 'RWVar'. diff --git a/Control/Concurrent/STM/Broadcast.hs b/Control/Concurrent/STM/Broadcast.hs deleted file mode 100644 --- a/Control/Concurrent/STM/Broadcast.hs +++ /dev/null @@ -1,101 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} - -------------------------------------------------------------------------------- --- | --- Module : Control.Concurrent.STM.Broadcast --- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk --- License : BSD3 (see the file LICENSE) --- Maintainer : Bas van Dijk --- , Roel van Dijk --- --- A Broadcast variable is a mechanism for communication between --- threads. Multiple reader threads can wait until a broadcaster thread writes a --- signal. The readers retry until the signal is received. When the broadcaster --- sends the signal all readers are woken. --- --- This module is designed to be imported qualified. We suggest importing it --- like: --- --- @ --- import Control.Concurrent.STM.Broadcast ( Broadcast ) --- import qualified Control.Concurrent.STM.Broadcast as Broadcast ( ... ) --- @ --- -------------------------------------------------------------------------------- - -module Control.Concurrent.STM.Broadcast - ( Broadcast - , new - , newWritten - , read - , tryRead - , write - , clear - ) where - - -------------------------------------------------------------------------------- --- Imports -------------------------------------------------------------------------------- - --- from base -import Control.Applicative ( (<$>) ) -import Control.Monad ( return, (=<<), fmap ) -import Control.Concurrent.STM ( STM, retry ) -import Control.Concurrent.STM.TVar ( TVar, newTVar, readTVar, writeTVar ) -import Data.Eq ( Eq ) -import Data.Maybe ( Maybe(Nothing, Just), maybe ) -import Data.Typeable ( Typeable ) - --- from base-unicode-symbols -import Data.Function.Unicode ( (∘) ) - - -------------------------------------------------------------------------------- --- Broadcast -------------------------------------------------------------------------------- - --- | A broadcast variable. It can be thought of as a box, which may be empty of --- full. -newtype Broadcast α = Broadcast {unBroadcast ∷ TVar (Maybe α)} - deriving (Eq, Typeable) - --- | Create a new empty 'Broadcast' variable. -new ∷ STM (Broadcast α) -new = Broadcast <$> newTVar Nothing - --- | Create a new 'Broadcast' variable containing an initial value. -newWritten ∷ α → STM (Broadcast α) -newWritten = fmap Broadcast ∘ newTVar ∘ Just - -{-| Read the value of a 'Broadcast' variable. - -If the 'Broadcast' variable contains a value it will be returned immediately, -otherwise it will retry until another thread 'write's a value to the 'Broadcast' -variable. --} -read ∷ Broadcast α → STM α -read = (maybe retry return =<<) ∘ readTVar ∘ unBroadcast - -{-| Try to read the value of a 'Broadcast' variable; non blocking. - -Like 'read' but doesn't retry. Returns 'Just' the contents of the 'Broadcast' if -it wasn't empty, 'Nothing' otherwise. --} -tryRead ∷ Broadcast α → STM (Maybe α) -tryRead = readTVar ∘ unBroadcast - -{-| Write a new value into a 'Broadcast' variable. - -If the variable is empty any threads that are reading from the variable will be -woken. If the variable is full its contents will simply be overwritten. --} -write ∷ Broadcast α → α → STM () -write b = writeTVar (unBroadcast b) ∘ Just - --- | Clear the contents of a 'Broadcast' variable. -clear ∷ Broadcast α → STM () -clear b = writeTVar (unBroadcast b) Nothing - - --- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/STM/Broadcast/Test.hs b/Control/Concurrent/STM/Broadcast/Test.hs deleted file mode 100644 --- a/Control/Concurrent/STM/Broadcast/Test.hs +++ /dev/null @@ -1,40 +0,0 @@ -{-# LANGUAGE NoImplicitPrelude - , UnicodeSyntax - #-} - -module Control.Concurrent.STM.Broadcast.Test ( tests ) where - - -------------------------------------------------------------------------------- --- Imports -------------------------------------------------------------------------------- - --- from base: -import Control.Concurrent ( ) - --- from base-unicode-symbols: -import Prelude.Unicode ( ) - --- from concurrent-extra: -import qualified Control.Concurrent.STM.Broadcast as Broadcast ( ) -import TestUtils ( ) - --- from HUnit: -import Test.HUnit ( ) - --- from test-framework: -import Test.Framework ( Test ) - --- from test-framework-hunit: -import Test.Framework.Providers.HUnit ( ) - - -------------------------------------------------------------------------------- --- Tests for STM Broadcast -------------------------------------------------------------------------------- - -tests ∷ [Test] -tests = [] - - --- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/STM/Event.hs b/Control/Concurrent/STM/Event.hs deleted file mode 100644 --- a/Control/Concurrent/STM/Event.hs +++ /dev/null @@ -1,103 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} - -------------------------------------------------------------------------------- --- | --- Module : Control.Concurrent.STM.Event --- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk --- License : BSD3 (see the file LICENSE) --- Maintainer : Bas van Dijk --- , Roel van Dijk --- --- An Event is a simple mechanism for communication between threads: one thread --- signals an event and other threads wait for it. --- --- Each event has an internal state which is either \"Set\" or \"Cleared\". This --- state can be changed with the corresponding functions 'set' and 'clear'. The --- 'wait' function blocks until the state is \"Set\". An important property of --- setting an event is that /all/ threads waiting for it are woken. --- --- It was inspired by the Python @Event@ object. See: --- --- --- --- This module is designed to be imported qualified. We suggest importing it --- like: --- --- @ --- import Control.Concurrent.STM.Event ( Event ) --- import qualified Control.Concurrent.STM.Event as Event ( ... ) --- @ --- -------------------------------------------------------------------------------- - -module Control.Concurrent.STM.Event - ( Event - , new - , newSet - , wait - , set - , clear - , isSet - ) where - - -------------------------------------------------------------------------------- --- Imports -------------------------------------------------------------------------------- - --- from base -import Control.Applicative ( (<$>) ) -import Control.Monad ( fmap ) -import Control.Concurrent.STM ( STM ) -import Data.Bool ( Bool ) -import Data.Eq ( Eq ) -import Data.Maybe ( isJust ) -import Data.Typeable ( Typeable ) - --- from base-unicode-symbols -import Data.Function.Unicode ( (∘) ) - --- from concurrent-extra -import Control.Concurrent.STM.Broadcast ( Broadcast ) -import qualified Control.Concurrent.STM.Broadcast as Broadcast - ( new, newWritten, read, tryRead, write, clear ) - - -------------------------------------------------------------------------------- --- Events -------------------------------------------------------------------------------- - --- | An event is in one of two possible states: \"Set\" or \"Cleared\". -newtype Event = Event {evBroadcast ∷ Broadcast ()} deriving (Eq, Typeable) - --- | Create an event in the \"Cleared\" state. -new ∷ STM Event -new = Event <$> Broadcast.new - --- | Create an event in the \"Set\" state. -newSet ∷ STM Event -newSet = Event <$> Broadcast.newWritten () - --- | Retry until the event is 'set'. --- --- If the state of the event is already \"Set\" this function will return --- immediately. Otherwise it will retry until another thread calls 'set'. -wait ∷ Event → STM () -wait = Broadcast.read ∘ evBroadcast - --- | Changes the state of the event to \"Set\". All threads that where waiting for --- this event are woken. Threads that 'wait' after the state is changed to \"Set\" --- will not retry. -set ∷ Event → STM () -set ev = Broadcast.write (evBroadcast ev) () - --- | Changes the state of the event to \"Cleared\". Threads that 'wait' after the --- state is changed to \"Cleared\" will retry until the state is changed to \"Set\". -clear ∷ Event → STM () -clear = Broadcast.clear ∘ evBroadcast - -isSet ∷ Event → STM Bool -isSet = fmap isJust ∘ Broadcast.tryRead ∘ evBroadcast - - --- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/STM/Event/Test.hs b/Control/Concurrent/STM/Event/Test.hs deleted file mode 100644 --- a/Control/Concurrent/STM/Event/Test.hs +++ /dev/null @@ -1,40 +0,0 @@ -{-# LANGUAGE NoImplicitPrelude - , UnicodeSyntax - #-} - -module Control.Concurrent.STM.Event.Test ( tests ) where - - -------------------------------------------------------------------------------- --- Imports -------------------------------------------------------------------------------- - --- from base: -import Control.Concurrent ( ) - --- from base-unicode-symbols: -import Prelude.Unicode ( ) - --- from concurrent-extra: -import qualified Control.Concurrent.STM.Event as Event ( ) -import TestUtils ( ) - --- from HUnit: -import Test.HUnit ( ) - --- from test-framework: -import Test.Framework ( Test ) - --- from test-framework-hunit: -import Test.Framework.Providers.HUnit ( ) - - -------------------------------------------------------------------------------- --- Tests for STM Event -------------------------------------------------------------------------------- - -tests ∷ [Test] -tests = [] - - --- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Thread.hs b/Control/Concurrent/Thread.hs --- a/Control/Concurrent/Thread.hs +++ b/Control/Concurrent/Thread.hs @@ -1,7 +1,4 @@ -{-# LANGUAGE DeriveDataTypeable - , NoImplicitPrelude - , UnicodeSyntax - #-} +{-# LANGUAGE CPP, DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} ------------------------------------------------------------------------------- -- | @@ -28,17 +25,30 @@ module Control.Concurrent.Thread ( ThreadId , threadId + + -- * Forking threads , forkIO , forkOS + + -- * Waiting on threads , wait + , wait_ + , unsafeWait + , unsafeWait_ + + -- ** Waiting with a timeout , waitTimeout + , waitTimeout_ + , unsafeWaitTimeout + , unsafeWaitTimeout_ + + -- * Quering thread status , isRunning -- * Convenience functions + , throwTo , killThread , killThreadTimeout - , throwTo - , unsafeWait ) where @@ -47,23 +57,25 @@ ------------------------------------------------------------------------------- -- from base: -import Control.Applicative ( (<$>) ) -import Control.Exception ( Exception, SomeException(SomeException) - , AsyncException(ThreadKilled) - , try, blocked, block, unblock, throwIO - ) -import Control.Monad ( return, (>>=), fail, (>>), fmap ) -import Data.Bool ( Bool(..) ) -import Data.Eq ( Eq, (==) ) -import Data.Either ( Either, either ) -import Data.Function ( ($), on ) -import Data.Maybe ( Maybe(..), isNothing, isJust ) -import Data.Ord ( Ord, compare ) -import Data.Typeable ( Typeable ) -import Prelude ( Integer ) -import System.IO ( IO ) - -import Text.Show ( Show, show ) +import Control.Exception ( Exception, SomeException + , AsyncException(ThreadKilled) + , try, blocked, block, unblock + ) +#ifdef __HADDOCK__ +import Control.Exception ( BlockedIndefinitelyOnMVar, BlockedIndefinitelyOnSTM ) +#endif +import Control.Monad ( return, (>>=), fail, (>>) ) +import Data.Bool ( Bool(..) ) +import Data.Eq ( Eq, (==) ) +import Data.Either ( Either(..), either ) +import Data.Function ( ($), on ) +import Data.Functor ( fmap, (<$>) ) +import Data.Maybe ( Maybe(..), maybe, isNothing, isJust ) +import Data.Ord ( Ord, compare ) +import Data.Typeable ( Typeable ) +import Prelude ( Integer ) +import System.IO ( IO ) +import Text.Show ( Show, show ) import qualified Control.Concurrent as Conc ( ThreadId, forkIO, forkOS, throwTo ) @@ -74,9 +86,9 @@ -- from concurrent-extra: import Control.Concurrent.Broadcast ( Broadcast ) import qualified Control.Concurrent.Broadcast as Broadcast - ( new, write, read, tryRead, readTimeout ) + ( new, broadcast, listen, tryListen, listenTimeout ) -import Utils ( void ) +import Utils ( void, ifM, throwInner ) ------------------------------------------------------------------------------- @@ -94,10 +106,9 @@ behaviour of a concurrent program. -} data ThreadId α = ThreadId - { stopped ∷ Broadcast (Either SomeException α) - -- | Extract the underlying 'Conc.ThreadId' - -- (@Conctrol.Concurrent.ThreadId@). - , threadId ∷ Conc.ThreadId + { stopped ∷ Broadcast (Either SomeException α) + , threadId ∷ Conc.ThreadId -- ^ Extract the underlying 'Conc.ThreadId' + -- (@Conctrol.Concurrent.ThreadId@). } deriving Typeable instance Eq (ThreadId α) where @@ -109,18 +120,11 @@ instance Show (ThreadId α) where show = show ∘ threadId -{-| -Internally used function which generalises 'forkIO' and 'forkOS'. Parametrised -by the function which does the actual forking. --} -fork ∷ (IO () → IO Conc.ThreadId) → IO α → IO (ThreadId α) -fork doFork a = do - stop ← Broadcast.new - b ← blocked - tid ← block $ doFork $ try (if b then a else unblock a) >>= - Broadcast.write stop - return $ ThreadId stop tid +------------------------------------------------------------------------------- +-- * Forking threads +------------------------------------------------------------------------------- + {-| Sparks off a new thread to run the given 'IO' computation and returns the 'ThreadId' of the newly created thread. @@ -132,9 +136,9 @@ 'Control.Exception.block'). The newly created thread has an exception handler that discards the exceptions -@BlockedOnDeadMVar@, @BlockedIndefinitely@, and @ThreadKilled@, and passes all -other exceptions to the uncaught exception handler (see -@setUncaughtExceptionHandler@). +'BlockedIndefinitelyOnMVar', 'BlockedIndefinitelyOnSTM', and 'ThreadKilled'. All +other exceptions are recorded in the 'ThreadId' and can be retrieved using +'wait'. -} forkIO ∷ IO α → IO (ThreadId α) forkIO = fork Conc.forkIO @@ -159,6 +163,23 @@ forkOS = fork Conc.forkOS {-| +Internally used function which generalises 'forkIO' and 'forkOS'. Parametrised +by the function which does the actual forking. +-} +fork ∷ (IO () → IO Conc.ThreadId) → IO α → IO (ThreadId α) +fork doFork a = do + stop ← Broadcast.new + let broadcastToStop = Broadcast.broadcast stop + tid ← ifM blocked ( doFork $ try a >>= broadcastToStop) + (block $ doFork $ try (unblock a) >>= broadcastToStop) + return $ ThreadId stop tid + + +------------------------------------------------------------------------------- +-- * Waiting on threads +------------------------------------------------------------------------------- + +{-| Block until the given thread is terminated. * Returns @'Right' x@ if the thread terminated normally and returned @x@. @@ -167,8 +188,25 @@ caught. -} wait ∷ ThreadId α → IO (Either SomeException α) -wait = Broadcast.read ∘ stopped +wait = Broadcast.listen ∘ stopped +-- | Like 'wait' but will ignore the value returned by the thread. +wait_ ∷ ThreadId α → IO () +wait_ = void ∘ wait + +-- | Like 'wait' but will either rethrow the exception that was thrown in the +-- thread or return the value that was returned by the thread. +unsafeWait ∷ ThreadId α → IO α +unsafeWait tid = wait tid >>= either throwInner return + +-- | Like 'unsafeWait' in that it will rethrow the exception that was thrown in +-- the thread but it will ignore the value returned by the thread. +unsafeWait_ ∷ ThreadId α → IO () +unsafeWait_ = void ∘ unsafeWait + + +-- ** Waiting with a timeout + {-| Block until the given thread is terminated or until a timer expires. @@ -180,47 +218,47 @@ The timeout is specified in microseconds. -} waitTimeout ∷ ThreadId α → Integer → IO (Maybe (Either SomeException α)) -waitTimeout = Broadcast.readTimeout ∘ stopped +waitTimeout = Broadcast.listenTimeout ∘ stopped -{-| -Returns 'True' if the given thread is currently running. +-- | Like 'waitTimeout' but will ignore the value returned by the thread. +-- Returns 'False' when a timeout occurred and 'True' otherwise. +waitTimeout_ ∷ ThreadId α → Integer → IO Bool +waitTimeout_ tid t = isJust <$> waitTimeout tid t -Notice that this observation is only a snapshot of a thread's state. By the time -a program reacts on its result it may already be out of date. +{-| +Like 'waitTimeout' but will rethrow the exception that was thrown in the +thread. Returns 'Nothing' if a timeout occured or 'Just' the value returned from +the target thread. -} -isRunning ∷ ThreadId α → IO Bool -isRunning = fmap isNothing ∘ Broadcast.tryRead ∘ stopped +unsafeWaitTimeout ∷ ThreadId α → Integer → IO (Maybe α) +unsafeWaitTimeout tid t = waitTimeout tid t >>= maybe (return Nothing) + (either throwInner + (return ∘ Just)) +-- | Like 'unsafeWaitTimeout' in that it will rethrow the exception that was +-- thrown in the thread but it will ignore the value returned by the thread. +-- Returns 'False' when a timeout occurred and 'True' otherwise. +unsafeWaitTimeout_ ∷ ThreadId α → Integer → IO Bool +unsafeWaitTimeout_ tid t = isJust <$> unsafeWaitTimeout tid t + ------------------------------------------------------------------------------- --- Convenience functions +-- * Quering thread status ------------------------------------------------------------------------------- {-| -'killThread' terminates the given thread (GHC only). Any work already done by -the thread isn't lost: the computation is suspended until required by another -thread. The memory used by the thread will be garbage collected if it isn't -referenced from anywhere. The 'killThread' function is defined in terms of -'throwTo'. +Returns 'True' if the given thread is currently running. -This function blocks until the target thread is terminated. It is a no-op if the -target thread has already completed. +Notice that this observation is only a snapshot of a thread's state. By the time +a program reacts on its result it may already be out of date. -} -killThread ∷ ThreadId α → IO () -killThread t = throwTo t ThreadKilled >> void (wait t) +isRunning ∷ ThreadId α → IO Bool +isRunning = fmap isNothing ∘ Broadcast.tryListen ∘ stopped -{-| -Like 'killThread' but with a timeout. Returns 'True' if the target thread was -terminated within the given amount of time, 'False' otherwise. -The timeout is specified in microseconds. - -Note that even when a timeout occurs, the target thread can still terminate at a -later time as a direct result of calling this function. --} -killThreadTimeout ∷ ThreadId α → Integer → IO Bool -killThreadTimeout t time = do throwTo t ThreadKilled - isJust <$> waitTimeout t time +------------------------------------------------------------------------------- +-- * Convenience functions +------------------------------------------------------------------------------- {-| 'throwTo' raises an arbitrary exception in the target thread (GHC only). @@ -253,9 +291,30 @@ throwTo ∷ Exception e ⇒ ThreadId α → e → IO () throwTo = Conc.throwTo ∘ threadId --- |Like 'wait' but will rethrow the exception that was thrown in target thread. -unsafeWait ∷ ThreadId α → IO α -unsafeWait tid = wait tid >>= either (\(SomeException e) → throwIO e) return +{-| +'killThread' terminates the given thread (GHC only). Any work already done by +the thread isn't lost: the computation is suspended until required by another +thread. The memory used by the thread will be garbage collected if it isn't +referenced from anywhere. The 'killThread' function is defined in terms of +'throwTo'. + +This function blocks until the target thread is terminated. It is a no-op if the +target thread has already completed. +-} +killThread ∷ ThreadId α → IO () +killThread tid = throwTo tid ThreadKilled >> wait_ tid + +{-| +Like 'killThread' but with a timeout. Returns 'True' if the target thread was +terminated within the given amount of time, 'False' otherwise. + +The timeout is specified in microseconds. + +Note that even when a timeout occurs, the target thread can still terminate at a +later time as a direct result of calling this function. +-} +killThreadTimeout ∷ ThreadId α → Integer → IO Bool +killThreadTimeout tid time = throwTo tid ThreadKilled >> waitTimeout_ tid time -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Thread/Test.hs b/Control/Concurrent/Thread/Test.hs --- a/Control/Concurrent/Thread/Test.hs +++ b/Control/Concurrent/Thread/Test.hs @@ -11,16 +11,18 @@ -- from base: import Control.Concurrent ( threadDelay ) -import Control.Monad ( return, (>>=), fail, (>>), fmap ) -import Data.Bool ( Bool(False, True) ) +import Control.Exception ( unblock, block, blocked ) +import Control.Monad ( return, (>>=), fail, (>>) ) +import Data.Bool ( Bool(False, True), not ) import Data.Function ( ($), id ) +import Data.Functor ( fmap ) import Data.IORef ( newIORef, readIORef, writeIORef ) import Data.Maybe ( maybe ) import Prelude ( fromInteger, toInteger ) import System.Timeout ( timeout ) -- from base-unicode-symbols: -import Prelude.Unicode ( (⋅) ) +import Prelude.Unicode ( (⋅) ) -- from concurrent-extra: import qualified Control.Concurrent.Lock as Lock @@ -42,15 +44,15 @@ ------------------------------------------------------------------------------- tests ∷ [Test] -tests = [ testCase "wait" test_wait - , testCase "waitTimeout" test_waitTimeout - , testCase "isRunning" test_isRunning +tests = [ testCase "wait" test_wait + , testCase "waitTimeout" test_waitTimeout + , testCase "isRunning" test_isRunning + , testCase "blockedState" test_blockedState + , testCase "unblockedState" test_unblockedState ] test_wait ∷ Assertion -test_wait = assert $ fmap (maybe False id) - $ timeout (10 ⋅ a_moment) - $ do +test_wait = assert $ fmap (maybe False id) $ timeout (10 ⋅ a_moment) $ do r ← newIORef False tid ← Thread.forkIO $ do threadDelay $ 2 ⋅ a_moment @@ -61,20 +63,25 @@ test_waitTimeout ∷ Assertion test_waitTimeout = assert $ within (10 ⋅ a_moment) $ do l ← Lock.newAcquired - tid ← Thread.forkIO $ do - Lock.acquire l + tid ← Thread.forkIO $ Lock.acquire l _ ← Thread.waitTimeout tid (toInteger $ 5 ⋅ a_moment) Lock.release l test_isRunning ∷ Assertion -test_isRunning = assert $ fmap (maybe False id) - $ timeout (10 ⋅ a_moment) - $ do +test_isRunning = assert $ fmap (maybe False id) $ timeout (10 ⋅ a_moment) $ do l ← Lock.newAcquired tid ← Thread.forkIO $ Lock.acquire l r ← Thread.isRunning tid Lock.release l return r + +test_blockedState ∷ Assertion +test_blockedState = (block $ Thread.forkIO $ blocked) >>= + Thread.unsafeWait >>= assert + +test_unblockedState ∷ Assertion +test_unblockedState = (unblock $ Thread.forkIO $ fmap not $ blocked) >>= + Thread.unsafeWait >>= assert -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Timeout.hs b/Control/Concurrent/Timeout.hs --- a/Control/Concurrent/Timeout.hs +++ b/Control/Concurrent/Timeout.hs @@ -25,9 +25,10 @@ -- from base: import Control.Concurrent ( forkIO, myThreadId, throwTo, killThread ) import Control.Exception ( Exception, bracket, handleJust ) -import Control.Monad ( return, (>>=), fail, (>>), fmap ) +import Control.Monad ( return, (>>=), fail, (>>) ) import Data.Bool ( otherwise ) import Data.Eq ( Eq ) +import Data.Functor ( fmap ) import Data.Maybe ( Maybe(Nothing, Just) ) import Data.Ord ( (<) ) import Data.Typeable ( Typeable ) @@ -109,7 +110,8 @@ (\_ → return Nothing) (bracket (forkIO (delay n >> throwTo pid ex)) (killThread) - (\_ → fmap Just f)) + (\_ → fmap Just f) + ) -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Timeout/Test.hs b/Control/Concurrent/Timeout/Test.hs --- a/Control/Concurrent/Timeout/Test.hs +++ b/Control/Concurrent/Timeout/Test.hs @@ -11,22 +11,28 @@ -- from base: import Control.Concurrent ( ) +import Control.Exception ( block ) +import Control.Monad ( (>>=), fail, (>>) ) +import Data.Function ( ($) ) +import Prelude ( fromInteger, toInteger ) -- from base-unicode-symbols: -import Prelude.Unicode ( ) +import Prelude.Unicode ( (⋅) ) -- from concurrent-extra: -import Control.Concurrent.Timeout ( ) -import TestUtils ( ) +import qualified Control.Concurrent.Lock as Lock +import qualified Control.Concurrent.Thread as Thread +import Control.Concurrent.Timeout ( timeout ) +import TestUtils ( a_moment, within ) -- from HUnit: -import Test.HUnit ( ) +import Test.HUnit ( Assertion, assert ) -- from test-framework: import Test.Framework ( Test ) -- from test-framework-hunit: -import Test.Framework.Providers.HUnit ( ) +import Test.Framework.Providers.HUnit ( testCase ) ------------------------------------------------------------------------------- @@ -34,7 +40,16 @@ ------------------------------------------------------------------------------- tests ∷ [Test] -tests = [] +tests = [ testCase "timeout exception" test_timeout_exception ] + +test_timeout_exception ∷ Assertion +test_timeout_exception = assert $ within (10 ⋅ a_moment) $ do + l ← Lock.newAcquired + tid ← block $ Thread.forkIO + $ timeout (toInteger $ 1000 ⋅ a_moment) + $ Lock.acquire l + Thread.killThread tid + Lock.release l -- The End --------------------------------------------------------------------- diff --git a/Utils.hs b/Utils.hs --- a/Utils.hs +++ b/Utils.hs @@ -8,11 +8,11 @@ -- from base: import Control.Concurrent.MVar ( MVar, takeMVar, putMVar ) -import Control.Exception ( block ) -import Control.Monad ( return, (>>=), (>>), fail - , Functor, fmap - ) -import Data.Function ( ($), const ) +import Control.Exception ( SomeException(SomeException), block, throwIO ) +import Control.Monad ( Monad, return, (>>=), (>>), fail ) +import Data.Bool ( Bool ) +import Data.Function ( ($) ) +import Data.Functor ( Functor, (<$) ) import Data.IORef ( IORef, readIORef, writeIORef ) import System.IO ( IO ) @@ -25,7 +25,13 @@ -------------------------------------------------------------------------------- void ∷ Functor f ⇒ f α → f () -void = fmap $ const () +void = (() <$) + +ifM ∷ Monad m ⇒ m Bool → m α → m α → m α +ifM c t e = c >>= \b → if b then t else e + +throwInner ∷ SomeException → IO α +throwInner (SomeException e) = throwIO e purelyModifyMVar ∷ MVar α → (α → α) → IO () purelyModifyMVar mv f = block $ takeMVar mv >>= putMVar mv ∘ f diff --git a/concurrent-extra.cabal b/concurrent-extra.cabal --- a/concurrent-extra.cabal +++ b/concurrent-extra.cabal @@ -1,5 +1,5 @@ name: concurrent-extra -version: 0.3.1 +version: 0.4 cabal-version: >= 1.6 build-type: Custom stability: experimental @@ -16,32 +16,30 @@ The @concurrent-extra@ package offers among other things the following selection of synchronisation primitives: . + * @Broadcast@: Wake multiple threads by broadcasting a value. + . + * @Event@: Wake multiple threads by signalling an event. + . * @Lock@: Enforce exclusive access to a resource. Also known as a binary semaphore or mutex. . * @RLock@: A lock which can be acquired multiple times by the same thread. Also known as a reentrant mutex. . - * @Broadcast@: Wake multiple threads by broadcasting a value. - . - * @Event@: Wake multiple threads by signalling an event. - . * @ReadWriteLock@: Multiple-reader, single-writer locks. Used to protect shared resources which may be concurrently read, but only sequentially written. . * @ReadWriteVar@: Concurrent read, sequential write variables. . - The package also provides @STM@ versions of @Broadcast@ and @Event@. - . - Besides these synchronisation primitives the package provides: + Besides these synchronisation primitives the package also provides: . * @Thread@: Threads extended with the ability to wait for their termination. . - * @delay@: Arbitrarily long thread delays. + * @Thread.Delay@: Arbitrarily long thread delays. . - * @timeout@: Wait arbitrarily long for an IO computation to finish. + * @Timeout@: Wait arbitrarily long for an IO computation to finish. . Please consult the API documentation of the individual modules for more detailed information. @@ -72,7 +70,6 @@ library build-depends: base >= 3 && < 4.3 , base-unicode-symbols >= 0.1.1 && < 0.2 - , stm >= 2.1.1 && < 2.2 exposed-modules: Control.Concurrent.Lock , Control.Concurrent.RLock , Control.Concurrent.Event @@ -81,8 +78,6 @@ , Control.Concurrent.ReadWriteVar , Control.Concurrent.Thread , Control.Concurrent.Thread.Delay - , Control.Concurrent.STM.Event - , Control.Concurrent.STM.Broadcast , Control.Concurrent.Timeout other-modules: Utils ghc-options: -Wall @@ -101,8 +96,6 @@ , Control.Concurrent.ReadWriteLock.Test , Control.Concurrent.ReadWriteVar.Test , Control.Concurrent.Thread.Test - , Control.Concurrent.STM.Event.Test - , Control.Concurrent.STM.Broadcast.Test , Control.Concurrent.Timeout.Test , TestUtils diff --git a/test.hs b/test.hs --- a/test.hs +++ b/test.hs @@ -10,19 +10,17 @@ import System.IO ( IO ) -- from concurrent-extra: -import qualified Control.Concurrent.Event.Test as Event ( tests ) -import qualified Control.Concurrent.Lock.Test as Lock ( tests ) -import qualified Control.Concurrent.RLock.Test as RLock ( tests ) -import qualified Control.Concurrent.Broadcast.Test as Broadcast ( tests ) -import qualified Control.Concurrent.ReadWriteLock.Test as RWLock ( tests ) -import qualified Control.Concurrent.ReadWriteVar.Test as RWVar ( tests ) -import qualified Control.Concurrent.Thread.Test as Thread ( tests ) -import qualified Control.Concurrent.STM.Event.Test as STM_Event ( tests ) -import qualified Control.Concurrent.STM.Broadcast.Test as STM_Broadcast ( tests ) -import qualified Control.Concurrent.Timeout.Test as Timeout ( tests ) +import qualified Control.Concurrent.Event.Test as Event ( tests ) +import qualified Control.Concurrent.Lock.Test as Lock ( tests ) +import qualified Control.Concurrent.RLock.Test as RLock ( tests ) +import qualified Control.Concurrent.Broadcast.Test as Broadcast ( tests ) +import qualified Control.Concurrent.ReadWriteLock.Test as RWLock ( tests ) +import qualified Control.Concurrent.ReadWriteVar.Test as RWVar ( tests ) +import qualified Control.Concurrent.Thread.Test as Thread ( tests ) +import qualified Control.Concurrent.Timeout.Test as Timeout ( tests ) -- from test-framework: -import Test.Framework ( Test, defaultMain, testGroup ) +import Test.Framework ( Test, defaultMain, testGroup ) ------------------------------------------------------------------------------- @@ -41,10 +39,6 @@ , testGroup "ReadWriteLock" RWLock.tests , testGroup "ReadWriteVar" RWVar.tests , testGroup "Thread" Thread.tests - ] - , testGroup "STM" - [ testGroup "Event" STM_Event.tests - , testGroup "Broadcast" STM_Broadcast.tests ] , testGroup "Timeout" Timeout.tests ]