diff --git a/Control/Concurrent/Event.hs b/Control/Concurrent/Event.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Event.hs @@ -0,0 +1,157 @@ +{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} + +------------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.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.Event ( Event ) +-- import qualified Control.Concurrent.Event as Event ( ... ) +-- @ +-- +------------------------------------------------------------------------------- + +module Control.Concurrent.Event + ( Event + , State(..) + , new + , wait + , waitTimeout + , set + , clear + , state + ) where + +------------------------------------------------------------------------------- +-- Imports +------------------------------------------------------------------------------- + +-- from base +import Control.Applicative ( (<$>) ) +import Control.Arrow ( first, second ) +import Control.Monad ( (>>=), (>>), return, fmap, forM_, fail ) +import Control.Concurrent.MVar ( MVar, newMVar + , takeMVar, putMVar, readMVar, modifyMVar_ + ) +import Control.Exception ( block, unblock ) +import Data.Bool ( Bool(False, True) ) +import Data.Eq ( Eq ) +import Data.Function ( ($), const ) +import Data.Int ( Int ) +import Data.List ( delete ) +import Data.Maybe ( Maybe(Nothing, Just) ) +import Data.Ord ( Ord, max ) +import Data.Tuple ( fst ) +import Data.Typeable ( Typeable ) +import Prelude ( Enum, fromInteger ) +import System.IO ( IO ) +import System.Timeout ( timeout ) +import Text.Read ( Read ) +import Text.Show ( Show ) + +-- from base-unicode-symbols +import Data.Function.Unicode ( (∘) ) + +-- from concurrent-extra +import Control.Concurrent.Lock ( Lock ) +import qualified Control.Concurrent.Lock as Lock ( newAcquired + , acquire, release + ) + + +------------------------------------------------------------------------------- +-- Events +------------------------------------------------------------------------------- + +-- | An event is in one of two possible states: 'Set' or 'Cleared'. +newtype Event = Event {unEvent ∷ (MVar (State, [Lock]))} + deriving (Eq, Typeable) + +-- | The internal state of an 'Event'. Only interesting when you use +-- the 'state' function. +data State = Cleared | Set deriving (Enum, Eq, Ord, Show, Read, Typeable) + +-- | Create an event. The initial state is 'Cleared'. +new ∷ IO Event +new = Event <$> newMVar (Cleared, []) + +-- | 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. +wait ∷ Event → IO () +wait (Event mv) = block $ do + t@(st, _) ← takeMVar mv + case st of + Set → putMVar mv t + Cleared → do l ← Lock.newAcquired + putMVar mv $ second (l:) t + Lock.acquire l + +-- | 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 maximum timeout is constrained by the range of the 'Int' +-- type. The Haskell standard guarantees an upper bound of at least @2^29-1@ +-- giving a maximum timeout of at least @(2^29-1) / 10^6@ = ~536 seconds. +waitTimeout ∷ Event → Int → IO Bool +waitTimeout (Event mv) time = block $ do + t@(st, _) ← takeMVar mv + case st of + Set → do putMVar mv t + return True + Cleared → do l ← Lock.newAcquired + putMVar mv $ second (l:) t + r ← unblock $ timeout (max time 0) (Lock.acquire l) + case r of + Just () → return True + Nothing → do modifyMVar_ mv $ return ∘ second (delete l) + return False + +-- | 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 ∷ Event → IO () +set (Event mv) = modifyMVar_ mv $ \(_, ls) → do + forM_ ls Lock.release + return (Set, []) + +-- | 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 (Event mv) = modifyMVar_ mv $ return ∘ first (const Cleared) + +-- | Determines the current state of the event. +-- +-- 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. This can be avoided by +-- synchronizing access to the event between threads. +state ∷ Event → IO State +state = fmap fst ∘ readMVar ∘ unEvent diff --git a/Control/Concurrent/Lock.hs b/Control/Concurrent/Lock.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Lock.hs @@ -0,0 +1,163 @@ +{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} + +-------------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.Lock +-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk +-- License : BSD3 (see the file LICENSE) +-- Maintainer : Bas van Dijk +-- , Roel van Dijk +-- +-- This module provides the 'Lock' synchronization mechanism. It was inspired by +-- the Python and Java @Lock@ objects and should behave in a similar way. See: +-- +-- +-- +-- and: +-- +-- +-- +-- All functions are /exception safe/. Throwing asynchronous exceptions will not +-- compromise the internal state of a 'Lock'. +-- +-- This module is intended to be imported qualified. We suggest importing it like: +-- +-- @ +-- import Control.Concurrent.Lock ( Lock ) +-- import qualified Control.Concurrent.Lock as Lock ( ... ) +-- @ +-- +-------------------------------------------------------------------------------- + +module Control.Concurrent.Lock + ( Lock + + -- * Creating locks + , new + , newAcquired + + -- * Locking and unlocking + , acquire + , tryAcquire + + , release + + -- * Convenience functions + , with + , tryWith + + -- * Querying locks + , locked + ) where + + +-------------------------------------------------------------------------------- +-- Imports +-------------------------------------------------------------------------------- + +-- from base +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 Data.Bool ( Bool, not ) +import Data.Eq ( Eq ) +import Data.Function ( ($) ) +import Data.Maybe ( Maybe(Nothing, Just), isJust ) +import Data.Typeable ( Typeable ) +import Prelude ( error ) +import System.IO ( IO ) + +-- from base-unicode-symbols +import Data.Function.Unicode ( (∘) ) + + +-------------------------------------------------------------------------------- +-- Locks +-------------------------------------------------------------------------------- + +{-| A lock is in one of two states, \"locked\" or \"unlocked\". -} +newtype Lock = Lock {un ∷ MVar ()} deriving (Eq, Typeable) + +-- | Create an unlocked lock. +new ∷ IO Lock +new = Lock <$> newMVar () + +-- | Create a locked lock. +newAcquired ∷ IO Lock +newAcquired = Lock <$> newEmptyMVar + +{-| When the state is unlocked, @acquire@ changes the state to locked and +returns immediately. When the state is locked, @acquire@ blocks until a call to +'release' in another thread changes it to unlocked, then the @acquire@ call +resets it to locked and returns. + +There are two further important properties of @acquire@: + +* @acquire@ is single-wakeup. That is, if there are multiple threads blocked on +@acquire@, and the lock is released, only one thread will be woken up. The +runtime guarantees that the woken thread completes its @acquire@ operation. + +* When multiple threads are blocked on @acquire@, they are woken up in FIFO +order. This is useful for providing fairness properties of abstractions built +using locks. (Note that this differs from the Python implementation where the +wake-up order is undefined) +-} +acquire ∷ Lock → IO () +acquire = takeMVar ∘ un + +{-| A non-blocking 'acquire'. When the state is unlocked, @tryAcquire@ changes +the state to locked and returns immediately with 'True'. When the state is +locked, @tryAcquire@ leaves the state unchanged and returns immediately with +'False'. +-} +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 an unlocked lock! + +If there are any threads blocked on 'acquire' the thread that first called +@acquire@ will be woken up. +-} +release ∷ Lock → IO () +release (Lock mv) = do + b ← tryPutMVar mv () + when (not b) $ error "Control.Concurrent.Lock.release: Can't release unlocked Lock!" + +{-| 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. + +Note that: @with = 'liftA2' 'bracket_' 'acquire' 'release'@. +-} +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 +by raising an exception, the lock is released and 'Just' the result of the +computation is returned. +-} +tryWith ∷ Lock → IO α → IO (Maybe α) +tryWith l a = block $ do + acquired ← tryAcquire l + if acquired + then fmap Just $ a `finally` release l + else return Nothing + +-- | Determines if the lock is in the locked state. +-- +-- 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. +locked ∷ Lock → IO Bool +locked = isEmptyMVar ∘ un + + +-- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/RLock.hs b/Control/Concurrent/RLock.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/RLock.hs @@ -0,0 +1,162 @@ +{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} + +-------------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.RLock +-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk +-- License : BSD3 (see the file LICENSE) +-- Maintainer : Bas van Dijk +-- , Roel van Dijk +-- +-- This module provides the 'RLock' synchronization mechanism. It was inspired +-- by the Python @RLock@ and Java @ReentrantLock@ objects and should behave in a +-- similar way. See: +-- +-- +-- +-- and: +-- +-- +-- +-- All functions are /exception safe/. Throwing asynchronous exceptions will not +-- compromise the internal state of a 'RLock'. +-- +-- This module is intended to be imported qualified. We suggest importing it like: +-- +-- @ +-- import Control.Concurrent.RLock ( RLock ) +-- import qualified Control.Concurrent.RLock as RLock ( ... ) +-- @ +-- +-------------------------------------------------------------------------------- + +module Control.Concurrent.RLock + ( RLock + + -- * Creating reentrant locks + , new + , newAcquired + + -- * Locking and unlocking + , acquire + , tryAcquire + + , release + + -- * Convenience functions + , with + , tryWith + + -- * Querying reentrant locks + , recursionLevel + ) where + + +-------------------------------------------------------------------------------- +-- Imports +-------------------------------------------------------------------------------- + +-- from base: +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 Data.Bool ( Bool(False, True), otherwise ) +import Data.Eq ( Eq ) +import Data.Function ( ($) ) +import Data.Maybe ( Maybe(Nothing, Just), maybe ) +import Data.List ( (++) ) +import Data.Typeable ( Typeable ) +import Prelude ( Integer, fromInteger, succ, pred, error, seq ) +import System.IO ( IO ) + +-- from base-unicode-symbols +import Data.Eq.Unicode ( (≡) ) +import Data.Function.Unicode ( (∘) ) + +-- from ourselves: +import Control.Concurrent.Lock ( Lock ) +import qualified Control.Concurrent.Lock as Lock + ( newAcquired, acquire, release ) + + +-------------------------------------------------------------------------------- +-- Reentrant locks +-------------------------------------------------------------------------------- + +newtype RLock = RLock {un ∷ MVar (Maybe (ThreadId, Integer, Lock))} + deriving (Eq, Typeable) + +new ∷ IO RLock +new = RLock <$> newMVar Nothing + +newAcquired ∷ IO RLock +newAcquired = do myTID ← myThreadId + lock ← Lock.newAcquired + RLock <$> newMVar (Just (myTID, 1, lock)) + +acquire ∷ RLock → IO () +acquire (RLock mv) = do + myTID ← myThreadId + block $ do + mb ← takeMVar mv + case mb of + Nothing → do lock ← Lock.newAcquired + putMVar mv $ Just (myTID, 1, lock) + Just (tid, n, lock) + | myTID ≡ tid → do let sn = succ n + sn `seq` putMVar mv $ Just (tid, sn, lock) + + | otherwise → do putMVar mv mb + Lock.acquire lock + +tryAcquire ∷ RLock → IO Bool +tryAcquire (RLock mv) = do + myTID ← myThreadId + block $ do + mb ← takeMVar mv + case mb of + Nothing → do lock ← Lock.newAcquired + putMVar mv $ Just (myTID, 1, lock) + return True + Just (tid, n, lock) + | myTID ≡ tid → do let sn = succ n + sn `seq` putMVar mv $ Just (tid, sn, lock) + return True + + | otherwise → do putMVar mv mb + return False + +release ∷ RLock → IO () +release (RLock mv) = do + myTID ← myThreadId + block $ do + mb ← takeMVar mv + let myError str = do putMVar mv mb + error $ "Control.Concurrent.RLock.release: " ++ str + case mb of + Nothing → myError "Can't release an unacquired RLock!" + Just (tid, n, lock) + | myTID ≡ tid → if n ≡ 1 + then do Lock.release lock + putMVar mv Nothing + else do let pn = pred n + pn `seq` putMVar mv $ Just (tid, pn, lock) + | otherwise → myError "Calling thread does not own the RLock!" + +with ∷ RLock → IO α → IO α +with = liftA2 bracket_ acquire release + +tryWith ∷ RLock → IO α → IO (Maybe α) +tryWith l a = block $ do + acquired ← tryAcquire l + if acquired + then fmap Just $ a `finally` release l + else return Nothing + +recursionLevel ∷ RLock → IO Integer +recursionLevel = fmap (maybe 0 (\(_, n, _) → n)) ∘ readMVar ∘ un + + +-- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/ReadWriteLock.hs b/Control/Concurrent/ReadWriteLock.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ReadWriteLock.hs @@ -0,0 +1,286 @@ +{-# LANGUAGE DeriveDataTypeable + , NamedFieldPuns + , NoImplicitPrelude + , TupleSections + , UnicodeSyntax + #-} + +------------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.ReadWriteLock +-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk +-- License : BSD3 (see the file LICENSE) +-- Maintainer : Bas van Dijk +-- , Roel van Dijk +-- +-- Multiple-reader, single-writer locks. Used to protect shared resources which +-- may be concurrently read, but only sequentially written. +-- +-- All functions are /exception safe/. Throwing asynchronous exceptions will not +-- compromise the internal state of an 'RWLock'. This means it is perfectly safe +-- to kill a thread that is blocking on, for example, 'acquireRead'. +-- +-- See also Java's version: +-- +-- +-- This module is designed to be imported qualified. We suggest importing it +-- like: +-- +-- @ +-- import Control.Concurrent.ReadWriteLock ( RWLock ) +-- import qualified Control.Concurrent.ReadWriteLock as RWL ( ... ) +-- @ +-- +------------------------------------------------------------------------------- + +module Control.Concurrent.ReadWriteLock + ( RWLock + -- *Creating Read-Write Locks + , new + -- *Read access + -- **Blocking + , acquireRead + , releaseRead + , withRead + -- **Non-blocking + , tryAcquireRead + , tryWithRead + -- *Write access + -- **Blocking + , acquireWrite + , releaseWrite + , withWrite + -- **Non-blocking + , tryAcquireWrite + , tryWithWrite + ) where + + +------------------------------------------------------------------------------- +-- Imports +------------------------------------------------------------------------------- + +-- from base +import Control.Applicative ( (<$>), liftA2 ) +import Control.Concurrent.MVar ( MVar, newMVar, takeMVar, putMVar + , modifyMVar_, swapMVar + ) +import Control.Exception ( block, bracket_, finally ) +import Control.Monad ( return, (>>=), return, fail, (>>) + , when, liftM3, Functor + ) +import Data.Bool ( Bool(False, True) ) +import Data.Char ( String ) +import Data.Eq ( Eq, (==) ) +import Data.Function ( ($), const ) +import Data.Int ( Int ) +import Data.Maybe ( Maybe(Nothing, Just) ) +import Data.Typeable ( Typeable ) +import Prelude ( fromInteger, succ, pred + , ($!), error + ) +import System.IO ( IO ) + +-- from base-unicode-symbols +import Data.Function.Unicode ( (∘) ) +import Data.Monoid.Unicode ( (⊕) ) + +-- from concurrent-extra +import Control.Concurrent.Lock ( Lock ) +import qualified Control.Concurrent.Lock as Lock + + +------------------------------------------------------------------------------- +-- Read Write Lock +------------------------------------------------------------------------------- + +{-| Multiple-reader, single-writer lock. Is in one of three states: + +* \"Free\": Read or write access can be acquired without blocking. + +* \"Read\": One or more threads have acquired read access. Blocks write access. + +* \"Write\": A single thread has acquired write access. Blocks other threads +from acquiring both read and write access. +-} +data RWLock = RWLock { state ∷ MVar State + , readLock ∷ Lock + , writeLock ∷ Lock + } deriving (Eq, Typeable) + +-- | Internal state of the 'RWLock'. +data State = Free | Read Int | Write + +{-| Create a new 'RWLock'. The initial state is \"free\"; either read or write +access can be acquired without blocking. +-} +new ∷ IO RWLock +new = liftM3 RWLock (newMVar Free) Lock.new Lock.new + +{-| 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\". + +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 → putMVar state (Read ∘ succ $! n) + Write → do putMVar state st + Lock.acquire writeLock + modifyMVar_ state ∘ const $ do + Lock.acquire readLock + return $ Read 1 + + +{-| Try to acquire the read lock; non blocking. + +Like 'acquireRead', but doesn't block. Returns 'True' if the resulting state is +\"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) + return True + Read n → do putMVar state (Read ∘ succ $! n) + return True + Write → do putMVar state st + return False + +{-| Release the read lock. + +If the calling thread was the last one to relinquish read access the state will +revert to \"free\". + +It is an error to release read access to an 'RWLock' which is not in the +\"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 → putMVar state (Read ∘ pred $! n) + Write → putMVar state st >> err + where + err = error $ moduleName ⊕ ".releaseRead: already released" + +{-| 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, +'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. +-} +tryWithRead ∷ RWLock → IO α → IO (Maybe α) +tryWithRead l a = block $ do + acquired ← tryAcquireRead l + if acquired + then Just <$> a `finally` releaseRead l + else return Nothing + +{-| 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\". +-} +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 + +{-| Try to acquire the write lock; non blocking. + +Like 'acquireWrite', but doesn't block. Returns 'True' if the resulting state is +\"write\", 'False' otherwise. +-} +tryAcquireWrite ∷ RWLock → IO Bool +tryAcquireWrite (RWLock {state, writeLock}) = block $ do + st ← takeMVar state + case st of + Free → do Lock.acquire writeLock + putMVar state Write + return True + Read _ → do putMVar state st + return False + Write → do putMVar state st + b ← Lock.tryAcquire writeLock + when b ∘ void $ swapMVar state Write + return b + +{-| Release the write lock. + +If @releaseWrite@ terminates without throwing an exception the state will be +\"free\". + +It is an error to release write access to an 'RWLock' which is not in the +\"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" + +{-| 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, +'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. +-} +tryWithWrite ∷ RWLock → IO α → IO (Maybe α) +tryWithWrite l a = block $ do + acquired ← tryAcquireWrite l + if acquired + then Just <$> a `finally` releaseWrite l + else return Nothing + + +------------------------------------------------------------------------------- + +moduleName ∷ String +moduleName = "Control.Concurrent.ReadWriteLock" + +void ∷ Functor f ⇒ f α → f () +void = (const () <$>) + + +-- The End -------------------------------------------------------------------- diff --git a/Control/Concurrent/ReadWriteVar.hs b/Control/Concurrent/ReadWriteVar.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/ReadWriteVar.hs @@ -0,0 +1,169 @@ +{-# LANGUAGE CPP + , DeriveDataTypeable + , NoImplicitPrelude + , TupleSections + , UnicodeSyntax + #-} + +------------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.ReadWriteVar +-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk +-- License : BSD3 (see the file LICENSE) +-- Maintainer : Bas van Dijk +-- , Roel van Dijk +-- +-- Concurrent read, sequential write variables. Comparable to an 'IORef' with +-- more advanced synchronization mechanisms. The value stored inside the 'RWVar' +-- can be read and used by multiple threads at the same time. Concurrent +-- computations inside a 'with' \"block\" observe the same value. +-- +-- Observing and changing the contents of an 'RWVar' are mutually exclusive. The +-- 'with' function will block if 'modify' is active and vice-versa. Furthermore +-- 'with' is fully sequential and will also block on concurrent calls of 'with'. +-- +-- The following are guaranteed deadlocks: +-- +-- * @'modify_' v '$' 'const' '$' 'with' v '$' 'const' 'undefined'@ +-- +-- * @'with' v '$' 'const' '$' 'modify_' v '$' 'const' 'undefined'@ +-- +-- * @'modify_' v '$' 'const' '$' 'modify_' v '$' 'const' 'undefined'@ +-- +-- All functions are /exception safe/. Throwing asynchronous exceptions will not +-- compromise the internal state of an 'RWVar'. This also means that threads +-- blocking on 'with' or 'modify' and friends can still be unblocked by throwing +-- an asynchronous exception. +-- +-- This module is designed to be imported qualified. We suggest importing it +-- like: +-- +-- @ +-- import Control.Concurrent.ReadWriteVar ( RWVar ) +-- import qualified Control.Concurrent.ReadWriteVar as RWV ( ... ) +-- @ +-- +------------------------------------------------------------------------------- + +module Control.Concurrent.ReadWriteVar + ( RWVar + -- *Creating Read-Write Variables + , new + -- *Reading + , with + , tryWith + -- *Writing + , modify_ + , modify + , tryModify_ + , tryModify + ) where + + +------------------------------------------------------------------------------- +-- Imports +------------------------------------------------------------------------------- + +-- from base +import Control.Monad ( return, (>>=), return, fail, (>>) + , fmap, liftM2 + ) +import Data.Bool ( Bool(..) ) +import Data.Eq ( Eq ) +import Data.Function ( ($) ) +import Data.Maybe ( Maybe(..), isJust ) +import Data.IORef ( IORef, newIORef, readIORef, writeIORef ) +import Data.Typeable ( Typeable ) +import System.IO ( IO ) +#ifdef __HADDOCK__ +import Data.Function ( const ) +import Prelude ( undefined ) +#endif + +-- from base-unicode-symbols +import Data.Function.Unicode ( (∘) ) + +-- from concurrent-extra +import Control.Concurrent.ReadWriteLock ( RWLock ) +import qualified Control.Concurrent.ReadWriteLock as RWLock + + +------------------------------------------------------------------------------- +-- Read-Write Variables: concurrent read, sequential write +------------------------------------------------------------------------------- + +-- | Concurrently readable and sequentially writable variable. +data RWVar α = RWVar RWLock (IORef α) deriving (Eq, Typeable) + +-- | Create a new 'RWVar'. +new ∷ α → IO (RWVar α) +new = liftM2 RWVar RWLock.new ∘ newIORef + +{-| Execute an action that operates on the contents of the 'RWVar'. + +The action is guaranteed to have a consistent view of the stored value. Any +function that attempts to 'modify' the contents will block until the action is +completed. + +If another thread is modifying the contents of the 'RWVar' this function will +block until the other thread finishes its action. +-} +with ∷ RWVar α → (α → IO β) → IO β +with (RWVar l r) f = RWLock.withRead l $ readIORef r >>= f + +{-| Like 'with' but doesn't block. Returns 'Just' the result if read access +could be acquired without blocking, 'Nothing' otherwise. +-} +tryWith ∷ RWVar α → (α → IO β) → IO (Maybe β) +tryWith (RWVar l r) f = RWLock.tryWithRead l $ readIORef r >>= f + +{-| Modify the contents of an 'RWVar'. + +This function needs exclusive write access to the 'RWVar'. Only one thread can +modify an 'RWVar' at the same time. All others will block. +-} +modify_ ∷ RWVar α → (α → IO α) → IO () +modify_ (RWVar l r) = RWLock.withWrite l ∘ modifyIORefM_ r + +{-| Modify the contents of an 'RWVar' and return an additional value. + +Like 'modify_', but allows a value to be returned (β) in addition to the +modified value of the 'RWVar'. +-} +modify ∷ RWVar α → (α → IO (α, β)) → IO β +modify (RWVar l r) = RWLock.withWrite l ∘ modifyIORefM r + +{-| Attempt to modify the contents of an 'RWVar'. + +Like 'modify_', but doesn't block. Returns 'True' if the contents could be +replaced, 'False' otherwise. +-} +tryModify_ ∷ RWVar α → (α → IO α) → IO Bool +tryModify_ (RWVar l r) = fmap isJust ∘ RWLock.tryWithWrite l ∘ modifyIORefM_ r + +{-| Attempt to modify the contents of an 'RWVar' and return an additional value. + +Like 'modify', but doesn't block. Returns 'Just' the additional value if the +contents could be replaced, 'Nothing' otherwise. +-} +tryModify ∷ RWVar α → (α → IO (α, β)) → IO (Maybe β) +tryModify (RWVar l r) = RWLock.tryWithWrite l ∘ modifyIORefM r + + +------------------------------------------------------------------------------- + +modifyIORefM ∷ IORef α → (α → IO (α, β)) → IO β +modifyIORefM r f = do x ← readIORef r + (y, z) ← f x + writeIORef r y + return z + +modifyIORefM_ ∷ IORef α → (α → IO α) → IO () +modifyIORefM_ r f = do x ← readIORef r + y ← f x + writeIORef r y + + +-- The End -------------------------------------------------------------------- + + diff --git a/Control/Concurrent/STM/Event.hs b/Control/Concurrent/STM/Event.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/STM/Event.hs @@ -0,0 +1,103 @@ +{-# 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 + , State(..) + , new + , wait + , set + , clear + , state + ) where + +------------------------------------------------------------------------------- +-- Imports +------------------------------------------------------------------------------- + +-- from base +import Control.Applicative ( (<$>) ) +import Control.Monad ( (>>=), fail, when ) +import Control.Concurrent.STM ( STM, retry ) +import Control.Concurrent.STM.TVar ( TVar, newTVar, readTVar, writeTVar ) +import Data.Eq ( Eq ) +import Data.Function ( ($) ) +import Data.Typeable ( Typeable ) + +-- from base-unicode-symbols +import Data.Eq.Unicode ( (≡) ) + +-- from concurrent-extra +import Control.Concurrent.Event ( State(Cleared, Set) ) + + +------------------------------------------------------------------------------- +-- Events +------------------------------------------------------------------------------- + +-- | An event is in one of two possible states: 'Set' or 'Cleared'. +newtype Event = Event (TVar State) deriving (Eq, Typeable) + +-- | Create an event. The initial state is 'Cleared'. +new ∷ STM Event +new = Event <$> newTVar Cleared + +-- | 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'. +-- +-- You can also stop a thread that is waiting for an event by throwing an +-- asynchronous exception. +wait ∷ Event → STM () +wait (Event tv) = do + st ← readTVar tv + when (st ≡ Cleared) retry + +-- | 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 (Event tv) = do + st ← readTVar tv + when (st ≡ Cleared) $ writeTVar tv Set + +-- | 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 (Event tv) = do + st ← readTVar tv + when (st ≡ Set) $ writeTVar tv Cleared + +-- | The current state of the event. +state ∷ Event → STM State +state (Event tv) = readTVar tv diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,32 @@ +Copyright (c) 2009 Bas van Dijk & Roel van Dijk + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * The names of Bas van Dijk, Roel van Dijk and the names of + contributors may NOT be used to endorse or promote products + derived from this software without specific prior written + permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,54 @@ +#! /usr/bin/env runhaskell + +{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax #-} + +module Main (main) where + +------------------------------------------------------------------------------- +-- Imports +------------------------------------------------------------------------------- + +-- from base +import Control.Monad ( (>>), return ) +import Data.Bool ( Bool ) +import System.Cmd ( system ) +import System.FilePath ( () ) +import System.IO ( IO ) + +-- from cabal +import Distribution.Simple ( defaultMainWithHooks + , simpleUserHooks + , UserHooks(runTests, haddockHook) + , Args + ) + +import Distribution.Simple.LocalBuildInfo ( LocalBuildInfo(..) ) +import Distribution.Simple.Program ( userSpecifyArgs ) +import Distribution.Simple.Setup ( HaddockFlags ) +import Distribution.PackageDescription ( PackageDescription(..) ) + +------------------------------------------------------------------------------- + +main ∷ IO () +main = defaultMainWithHooks hooks + where + hooks = simpleUserHooks + { runTests = runTests' + , haddockHook = haddockHook' + } + +-- Run a 'test' binary that gets built when configured with '-ftest'. +runTests' ∷ Args → Bool → PackageDescription → LocalBuildInfo → IO () +runTests' _ _ _ _ = system testcmd >> return () + where testcmd = "." + "dist" + "build" + "test-concurrent-extra" + "test-concurrent-extra" + +-- Define __HADDOCK__ for CPP when running haddock. +haddockHook' ∷ PackageDescription → LocalBuildInfo → UserHooks → HaddockFlags → IO () +haddockHook' pkg lbi = + haddockHook simpleUserHooks pkg (lbi { withPrograms = p }) + where + p = userSpecifyArgs "haddock" ["--optghc=-D__HADDOCK__"] (withPrograms lbi) diff --git a/concurrent-extra.cabal b/concurrent-extra.cabal new file mode 100644 --- /dev/null +++ b/concurrent-extra.cabal @@ -0,0 +1,89 @@ +name: concurrent-extra +version: 0.1 +cabal-version: >= 1.6 +build-type: Custom +stability: experimental +author: Bas van Dijk + Roel van Dijk +maintainer: Bas van Dijk + Roel van Dijk +copyright: (c) 2010 Bas van Dijk & Roel van Dijk +license: BSD3 +license-file: LICENSE +category: Concurrency +synopsis: Extra concurrency primitives +description: + Offers a selection of synchronization primitives: + . + * Lock: Enforce exclusive access to a resource. Also known as a + binary semaphore. + . + * RLock: A lock which can be acquired multiple times by the same + thread. Also known as a reentrant mutex. + . + * Event: Wake multiple threads by signaling 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. + . + Please consult the API documentation of the individual modules for + more detailed information. + . + Inspired by the concurrency libraries of Java and Python. + +--source-repository head +-- Type: darcs +-- Location: http://code.haskell.org/concurrent-extra + +------------------------------------------------------------------------------- + +flag test + description: Build the testing suite + default: False + +flag hpc + description: Enable program coverage on test executable + default: False + +flag nolib + description: Don't build the library + default: False + +------------------------------------------------------------------------------- + +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 + , Control.Concurrent.ReadWriteLock + , Control.Concurrent.ReadWriteVar + , Control.Concurrent.STM.Event + ghc-options: -Wall + + if flag(nolib) + buildable: False + +------------------------------------------------------------------------------- + +executable test-concurrent-extra + main-is: test.hs + ghc-options: -Wall + + if flag(test) + build-depends: HUnit >= 1.2.2 && < 1.3 + , QuickCheck >= 2.1.0 && < 2.2 + , test-framework >= 0.2.4 && < 0.3 + , test-framework-hunit >= 0.2.4 && < 0.3 + , test-framework-quickcheck2 >= 0.2.4 && < 0.3 + buildable: True + else + buildable: False + + if flag(hpc) + ghc-options: -fhpc diff --git a/test.hs b/test.hs new file mode 100644 --- /dev/null +++ b/test.hs @@ -0,0 +1,210 @@ +{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax, ScopedTypeVariables #-} + +module Main where + +------------------------------------------------------------------------------- +-- Imports +------------------------------------------------------------------------------- + +-- from base +import Control.Applicative ( (<$>) ) +import Control.Concurrent ( forkIO, threadDelay ) +import Control.Exception ( catch, try, throwTo, ErrorCall(..), SomeException ) +import Control.Monad ( (>>=), (>>), return, fail, fmap + , mapM_, replicateM, replicateM_ + ) +import Data.Bool ( Bool, not ) +import Data.Char ( String ) +import Data.Either ( Either(Left, Right) ) +import Data.Function ( ($) ) +import Data.Int ( Int ) +import Data.Maybe ( isJust ) +import Prelude ( fromInteger ) +import System.IO ( IO ) +import System.Timeout ( timeout ) + +-- from base-unicode-symbols +import Data.Function.Unicode ( (∘) ) +import Prelude.Unicode ( (⋅) ) + +-- from concurrent-extra +import qualified Control.Concurrent.Lock as Lock +import qualified Control.Concurrent.Event as Event + +-- from HUnit +import Test.HUnit hiding ( Test ) + +-- from test-framework +import Test.Framework ( Test, defaultMain, testGroup ) + +-- from test-framework-hunit +import Test.Framework.Providers.HUnit ( testCase ) + + + +------------------------------------------------------------------------------- +-- Tests +------------------------------------------------------------------------------- + +main ∷ IO () +main = defaultMain tests + +tests ∷ [Test] +tests = [ testGroup "Events" + [ testCase "set wait a" $ test_event_1 1 1 + , testCase "set wait b" $ test_event_1 5 1 + , testCase "set wait c" $ test_event_1 1 5 + , testCase "set wait d" $ test_event_1 5 5 + , testCase "conc set wait" $ test_event_2 + , testCase "multi wake" $ test_event_3 10 + , testCase "exception" $ test_event_4 + , testCase "wait timeout" $ test_event_5 + , testCase "foo!" $ test_event_6 + ] + , testGroup "Lock" + [ testCase "acquire release" test_lock_1 + , testCase "acquire acquire" test_lock_2 + , testCase "new release" test_lock_3 + , testCase "new unlocked" test_lock_4 + , testCase "newAcquired locked" test_lock_5 + , testCase "acq rel unlocked" test_lock_6 + , testCase "conc release" test_lock_7 + ] + , testGroup "RLock" + [ + ] + ] + + +-------------------------------------------------------------------------------- +-- Events +-------------------------------------------------------------------------------- + +-- Set an event 's' times then wait for it 'w' times. This should +-- terminate within a few moments. +test_event_1 ∷ Int → Int → Assertion +test_event_1 s w = assert $ within (10 ⋅ a_moment) $ do + e ← Event.new + replicateM_ s $ Event.set e + replicateM_ w $ Event.wait e + +test_event_2 ∷ Assertion +test_event_2 = assert $ within (10 ⋅ a_moment) $ do + e1 ← Event.new + e2 ← Event.new + _ ← forkIO $ do + Event.wait e1 + Event.set e2 + wait_a_moment + Event.set e1 + Event.wait e2 + +-- Waking multiple threads with a single Event. +test_event_3 ∷ Int → Assertion +test_event_3 n = assert $ within (10 ⋅ a_moment) $ do + e1 ← Event.new + es ← replicateM n $ do + e2 ← Event.new + _ ← forkIO $ do + Event.wait e1 + Event.set e2 + return e2 + wait_a_moment + Event.set e1 + mapM_ Event.wait es + +-- Exception handling while waiting for an Event. +test_event_4 ∷ Assertion +test_event_4 = assert $ within (10 ⋅ a_moment) $ do + e1 ← Event.new + e2 ← Event.new + helperId ← forkIO $ catch (Event.wait e1) $ \(_ ∷ ErrorCall) → Event.set e2 + wait_a_moment + throwTo helperId $ ErrorCall "Boo!" + Event.wait e2 + +test_event_5 ∷ Assertion +test_event_5 = assert $ within (10 ⋅ a_moment) $ do + e ← Event.new + Event.waitTimeout e a_moment + +test_event_6 ∷ Assertion +test_event_6 = assert $ notWithin (10 ⋅ a_moment) $ do + e ← Event.new + Event.wait e + + +-------------------------------------------------------------------------------- +-- Locks +-------------------------------------------------------------------------------- + +test_lock_1 ∷ Assertion +test_lock_1 = assert $ within a_moment $ do + l ← Lock.new + Lock.acquire l + Lock.release l + +test_lock_2 ∷ Assertion +test_lock_2 = assert $ notWithin (10 ⋅ a_moment) $ do + l ← Lock.new + Lock.acquire l + Lock.acquire l + +test_lock_3 ∷ Assertion +test_lock_3 = assertException "" $ do + l ← Lock.new + Lock.release l + +test_lock_4 ∷ Assertion +test_lock_4 = assert $ do + l ← Lock.new + fmap not $ Lock.locked l + +test_lock_5 ∷ Assertion +test_lock_5 = assert $ do + l ← Lock.newAcquired + Lock.locked l + +test_lock_6 ∷ Assertion +test_lock_6 = assert $ do + l ← Lock.new + Lock.acquire l + Lock.release l + fmap not $ Lock.locked l + +test_lock_7 ∷ Assertion +test_lock_7 = assert ∘ within (10 ⋅ a_moment) $ do + l ← Lock.newAcquired + _ ← forkIO $ wait_a_moment >> Lock.release l + Lock.acquire l + + +-------------------------------------------------------------------------------- +-- RLocks +-------------------------------------------------------------------------------- + +-- TODO + +------------------------------------------------------------------------------- +-- Misc +------------------------------------------------------------------------------- + +-- Exactly 1 moment. Currently equal to 0.005 seconds. +a_moment ∷ Int +a_moment = 5000 + +wait_a_moment ∷ IO () +wait_a_moment = threadDelay a_moment + +-- True if the action 'a' evaluates within 't' μs. +within ∷ Int → IO α → IO Bool +within t a = isJust <$> timeout t a + +notWithin ∷ Int → IO α → IO Bool +notWithin t a = not <$> within t a + +assertException ∷ String → IO α → Assertion +assertException errMsg a = do e ← try a + case e of + Left (_ ∷ SomeException ) → return () + Right _ → assertFailure errMsg