concurrent-extra 0.1.0.1 → 0.2
raw patch · 26 files changed
+1802/−469 lines, 26 filessetup-changedPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
- Control.Concurrent.Event: Cleared :: State
- Control.Concurrent.Event: Set :: State
- Control.Concurrent.Event: data State
- Control.Concurrent.Event: instance Enum State
- Control.Concurrent.Event: instance Eq State
- Control.Concurrent.Event: instance Ord State
- Control.Concurrent.Event: instance Read State
- Control.Concurrent.Event: instance Show State
- Control.Concurrent.Event: instance Typeable State
- Control.Concurrent.Event: state :: Event -> IO State
- Control.Concurrent.RLock: recursionLevel :: RLock -> IO Integer
- Control.Concurrent.STM.Event: Cleared :: State
- Control.Concurrent.STM.Event: Set :: State
- Control.Concurrent.STM.Event: data State
- Control.Concurrent.STM.Event: state :: Event -> STM State
+ Control.Concurrent.Broadcast: clear :: Broadcast α -> IO ()
+ Control.Concurrent.Broadcast: data Broadcast α
+ Control.Concurrent.Broadcast: instance Eq (Broadcast α)
+ Control.Concurrent.Broadcast: instance Typeable1 Broadcast
+ Control.Concurrent.Broadcast: new :: IO (Broadcast α)
+ Control.Concurrent.Broadcast: newWritten :: α -> IO (Broadcast α)
+ Control.Concurrent.Broadcast: read :: Broadcast α -> IO α
+ Control.Concurrent.Broadcast: readTimeout :: Broadcast α -> Integer -> IO (Maybe α)
+ Control.Concurrent.Broadcast: tryRead :: Broadcast α -> IO (Maybe α)
+ Control.Concurrent.Broadcast: write :: Broadcast α -> α -> IO ()
+ Control.Concurrent.Event: isSet :: Event -> IO Bool
+ Control.Concurrent.Event: newSet :: IO Event
+ Control.Concurrent.Lock: wait :: Lock -> IO ()
+ Control.Concurrent.RLock: state :: RLock -> IO State
+ Control.Concurrent.RLock: type State = Maybe (ThreadId, Integer)
+ Control.Concurrent.RLock: wait :: RLock -> IO ()
+ Control.Concurrent.ReadWriteLock: newAcquiredRead :: IO RWLock
+ Control.Concurrent.ReadWriteLock: newAcquiredWrite :: IO RWLock
+ Control.Concurrent.STM.Broadcast: clear :: Broadcast α -> STM ()
+ Control.Concurrent.STM.Broadcast: data Broadcast α
+ Control.Concurrent.STM.Broadcast: instance Eq (Broadcast α)
+ Control.Concurrent.STM.Broadcast: instance Typeable1 Broadcast
+ Control.Concurrent.STM.Broadcast: new :: STM (Broadcast α)
+ Control.Concurrent.STM.Broadcast: newWritten :: α -> STM (Broadcast α)
+ Control.Concurrent.STM.Broadcast: read :: Broadcast α -> STM α
+ Control.Concurrent.STM.Broadcast: tryRead :: Broadcast α -> STM (Maybe α)
+ Control.Concurrent.STM.Broadcast: write :: Broadcast α -> α -> STM ()
+ Control.Concurrent.STM.Event: isSet :: Event -> STM Bool
+ Control.Concurrent.STM.Event: newSet :: STM Event
+ Control.Concurrent.Thread: data ThreadId
+ Control.Concurrent.Thread: forkIO :: IO () -> IO ThreadId
+ Control.Concurrent.Thread: forkOS :: IO () -> IO ThreadId
+ Control.Concurrent.Thread: instance Eq ThreadId
+ Control.Concurrent.Thread: instance Ord ThreadId
+ Control.Concurrent.Thread: instance Show ThreadId
+ Control.Concurrent.Thread: instance Typeable ThreadId
+ Control.Concurrent.Thread: isRunning :: ThreadId -> IO Bool
+ Control.Concurrent.Thread: killThread :: ThreadId -> IO ()
+ Control.Concurrent.Thread: killThreadTimeout :: ThreadId -> Integer -> IO Bool
+ Control.Concurrent.Thread: threadId :: ThreadId -> ThreadId
+ Control.Concurrent.Thread: throwTo :: (Exception e) => ThreadId -> e -> IO ()
+ Control.Concurrent.Thread: wait :: ThreadId -> IO (Maybe SomeException)
+ Control.Concurrent.Thread: waitTimeout :: ThreadId -> Integer -> IO (Maybe (Maybe SomeException))
+ Control.Concurrent.Thread.Delay: delay :: Integer -> IO ()
+ Control.Concurrent.Timeout: instance Eq Timeout
+ Control.Concurrent.Timeout: instance Exception Timeout
+ Control.Concurrent.Timeout: instance Show Timeout
+ Control.Concurrent.Timeout: instance Typeable Timeout
+ Control.Concurrent.Timeout: timeout :: Integer -> IO α -> IO (Maybe α)
- Control.Concurrent.Event: waitTimeout :: Event -> Int -> IO Bool
+ Control.Concurrent.Event: waitTimeout :: Event -> Integer -> IO Bool
Files
- Control/Concurrent/Broadcast.hs +160/−0
- Control/Concurrent/Broadcast/Test.hs +40/−0
- Control/Concurrent/Event.hs +60/−88
- Control/Concurrent/Event/Test.hs +109/−0
- Control/Concurrent/Lock.hs +60/−34
- Control/Concurrent/Lock/Test.hs +86/−0
- Control/Concurrent/RLock.hs +131/−30
- Control/Concurrent/RLock/Test.hs +81/−0
- Control/Concurrent/ReadWriteLock.hs +48/−27
- Control/Concurrent/ReadWriteLock/Test.hs +40/−0
- Control/Concurrent/ReadWriteVar.hs +12/−24
- Control/Concurrent/ReadWriteVar/Test.hs +40/−0
- Control/Concurrent/STM/Broadcast.hs +101/−0
- Control/Concurrent/STM/Broadcast/Test.hs +40/−0
- Control/Concurrent/STM/Event.hs +37/−37
- Control/Concurrent/STM/Event/Test.hs +40/−0
- Control/Concurrent/Thread.hs +248/−0
- Control/Concurrent/Thread/Delay.hs +54/−0
- Control/Concurrent/Thread/Test.hs +80/−0
- Control/Concurrent/Timeout.hs +115/−0
- Control/Concurrent/Timeout/Test.hs +40/−0
- Setup.hs +8/−0
- TestUtils.hs +56/−0
- Utils.hs +43/−0
- concurrent-extra.cabal +45/−12
- test.hs +28/−217
+ Control/Concurrent/Broadcast.hs view
@@ -0,0 +1,160 @@+{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-}++-------------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.Broadcast+-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk+-- License : BSD3 (see the file LICENSE)+-- Maintainer : Bas van Dijk <v.dijk.bas@gmail.com>+-- , Roel van Dijk <vandijk.roel@gmail.com>+--+-- 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.+--+-- All functions are /exception safe/. Throwing asynchronous exceptions will not+-- compromise the internal state of a 'Broadcast' variable.+--+-- This module is designed to be imported qualified. We suggest importing it+-- like:+--+-- @+-- import Control.Concurrent.Broadcast ( Broadcast )+-- import qualified Control.Concurrent.Broadcast as Broadcast ( ... )+-- @+-------------------------------------------------------------------------------++module Control.Concurrent.Broadcast+ ( Broadcast+ , new+ , newWritten+ , read+ , tryRead+ , readTimeout+ , write+ , clear+ ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Applicative ( (<$>) )+import Control.Arrow ( first )+import Control.Monad ( (>>=), (>>), return, fmap, forM_, fail )+import Control.Concurrent.MVar ( MVar, newMVar, newEmptyMVar+ , takeMVar, putMVar, readMVar, modifyMVar_+ )+import Control.Exception ( block, unblock )+import Data.Eq ( Eq )+import Data.Function ( ($), const )+import Data.List ( delete, length )+import Data.Maybe ( Maybe(Nothing, Just) )+import Data.Ord ( Ord, max )+import Data.Tuple ( fst )+import Data.Typeable ( Typeable )+import Prelude ( Integer, fromInteger, seq )+import System.IO ( IO )++-- from base-unicode-symbols:+import Data.Function.Unicode ( (∘) )++-- from concurrent-extra:+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 α])}+ deriving (Eq, Typeable)++-- | Create a new empty 'Broadcast' variable.+new ∷ IO (Broadcast α)+new = Broadcast <$> newMVar (Nothing, [])++-- | Create a new 'Broadcast' variable containing an initial value.+newWritten ∷ α → IO (Broadcast α)+newWritten x = Broadcast <$> newMVar (Just x, [])++{-| Read the value of a 'Broadcast' variable.++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.+-}+read ∷ Broadcast α → IO α+read (Broadcast mv) = block $ do+ t@(mx, ls) ← takeMVar mv+ case mx of+ Nothing → do l ← newEmptyMVar+ putMVar mv (mx, l:ls)+ takeMVar l+ Just x → do putMVar mv t+ return x++{-| Try to read the value of a 'Broadcast' variable; non blocking.++Like 'read' but doesn't block. Returns 'Just' the contents of the 'Broadcast' if+it wasn't empty, 'Nothing' otherwise.+-}+tryRead ∷ Broadcast α → IO (Maybe α)+tryRead = fmap fst ∘ readMVar ∘ unBroadcast++{-| Read the value of a 'Broadcast' variable 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.++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.+-}+readTimeout ∷ Broadcast α → Integer → IO (Maybe α)+readTimeout (Broadcast mv) time = block $ do+ t@(mx, ls) ← 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++{-| 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 α → α → IO ()+write (Broadcast mv) x =+ modifyMVar_ mv $ \(_, ls) → do+ forM_ ls $ \l → putMVar l x+ return (Just x, [])++-- | Clear the contents of a 'Broadcast' variable.+clear ∷ Broadcast α → IO ()+clear (Broadcast mv) = purelyModifyMVar mv $ first $ const Nothing+++-------------------------------------------------------------------------------++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')+++-- The End ---------------------------------------------------------------------+
+ Control/Concurrent/Broadcast/Test.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ #-}++module Control.Concurrent.Broadcast.Test ( tests ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( )++-- from base-unicode-symbols:+import Prelude.Unicode ( )++-- from concurrent-extra:+import qualified Control.Concurrent.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 Broadcast+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = []+++-- The End ---------------------------------------------------------------------
Control/Concurrent/Event.hs view
@@ -11,9 +11,9 @@ -- 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+-- 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+-- '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,126 +32,98 @@ module Control.Concurrent.Event ( Event- , State(..) , new+ , newSet , wait , waitTimeout , set , clear- , state+ , isSet ) 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 Control.Monad ( fmap )+import Data.Bool ( Bool(..) ) 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.Maybe ( isJust ) import Data.Typeable ( Typeable )-import Prelude ( Enum, fromInteger )+import Prelude ( Integer ) 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- )+import Control.Concurrent.Broadcast ( Broadcast )+import qualified Control.Concurrent.Broadcast as Broadcast+ ( new, newWritten, read, tryRead, readTimeout, write, clear ) ------------------------------------------------------------------------------- -- 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)+-- | An event is in one of two possible states: \"Set\" or \"Cleared\".+newtype Event = Event {evBroadcast ∷ Broadcast ()} deriving (Eq, Typeable) --- | Create an event. The initial state is 'Cleared'.+-- | Create an event in the \"Cleared\" state. new ∷ IO Event-new = Event <$> newMVar (Cleared, [])+new = Event <$> Broadcast.new --- | 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.+-- | Create an event in the \"Set\" state.+newSet ∷ IO Event+newSet = Event <$> Broadcast.newWritten ()++{-| 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+wait = Broadcast.read ∘ evBroadcast --- | 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+{-| Block until the event is 'set' or until a timer expires. --- | 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.+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.+-}+waitTimeout ∷ Event → Integer → IO Bool+waitTimeout ev time = isJust <$> Broadcast.readTimeout (evBroadcast ev) time++{-| 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, [])+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 block until the state is changed to '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)+clear = Broadcast.clear ∘ evBroadcast --- | 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+{-| 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.tryRead ∘ evBroadcast+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/Event/Test.hs view
@@ -0,0 +1,109 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ , ScopedTypeVariables+ #-}++module Control.Concurrent.Event.Test ( tests ) where++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Exception ( catch, throwTo, ErrorCall(..) )+import Control.Concurrent ( forkIO )+import Control.Monad ( return, (>>=), fail, (>>)+ , mapM_, replicateM, replicateM_+ )+import Data.Function ( ($) )+import Data.Int ( Int )+import Prelude ( fromInteger, toInteger )++-- from base-unicode-symbols:+import Prelude.Unicode ( (⋅) )++-- from concurrent-extra:+import qualified Control.Concurrent.Event as Event+import TestUtils++-- from HUnit:+import Test.HUnit ( Assertion, assert )++-- from test-framework:+import Test.Framework ( Test )++-- from test-framework-hunit:+import Test.Framework.Providers.HUnit ( testCase )+++-------------------------------------------------------------------------------+-- Tests for Event+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = [ 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 "wait blocks" $ test_event_6+ ]++-- 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 $ toInteger a_moment++test_event_6 ∷ Assertion+test_event_6 = assert $ notWithin (10 ⋅ a_moment) $ do+ e ← Event.new+ Event.wait e+++-- The End ---------------------------------------------------------------------
Control/Concurrent/Lock.hs view
@@ -8,7 +8,7 @@ -- Maintainer : Bas van Dijk <v.dijk.bas@gmail.com> -- , Roel van Dijk <vandijk.roel@gmail.com> ----- This module provides the 'Lock' synchronization mechanism. It was inspired by+-- This module provides the 'Lock' synchronisation mechanism. It was inspired by -- the Python and Java @Lock@ objects and should behave in a similar way. See: -- -- <http://docs.python.org/3.1/library/threading.html#lock-objects>@@ -31,21 +31,17 @@ module Control.Concurrent.Lock ( Lock- -- * Creating locks , new , newAcquired- -- * Locking and unlocking , acquire , tryAcquire- , release- -- * Convenience functions , with , tryWith-+ , wait -- * Querying locks , locked ) where@@ -55,7 +51,7 @@ -- Imports -------------------------------------------------------------------------------- --- from base+-- from base: import Control.Applicative ( (<$>), liftA2 ) import Control.Concurrent.MVar ( MVar, newMVar, newEmptyMVar , takeMVar, tryTakeMVar@@ -63,7 +59,7 @@ , isEmptyMVar ) import Control.Exception ( block, bracket_, finally )-import Control.Monad ( Monad, return, (>>=), fail, when, fmap )+import Control.Monad ( Monad, return, (>>=), (>>), fail, when, fmap ) import Data.Bool ( Bool, not ) import Data.Eq ( Eq ) import Data.Function ( ($) )@@ -72,7 +68,7 @@ import Prelude ( error ) import System.IO ( IO ) --- from base-unicode-symbols+-- from base-unicode-symbols: import Data.Function.Unicode ( (∘) ) @@ -80,47 +76,58 @@ -- Locks -------------------------------------------------------------------------------- -{-| A lock is in one of two states, \"locked\" or \"unlocked\". -}+-- | A lock is in one of two states: \"locked\" or \"unlocked\". newtype Lock = Lock {un ∷ MVar ()} deriving (Eq, Typeable) --- | Create an unlocked lock.+-- | Create a lock in the \"unlocked\" state. new ∷ IO Lock new = Lock <$> newMVar () --- | Create a locked lock.+-- | Create a lock in the \"locked\" state. 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.+{-| +Acquires the 'Lock'. Blocks if another thread has acquired the 'Lock'. +@acquire@ behaves as follows:++* 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+to \"locked\".+ 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+@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)+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'.+{-| +A non-blocking 'acquire'.++* When the state is \"unlocked\" @tryAcquire@ changes the state to \"locked\"+and returns 'True'.++* When the state is \"locked\" @tryAcquire@ leaves the state unchanged and+returns 'False'. -} tryAcquire ∷ Lock → IO Bool tryAcquire = fmap isJust ∘ tryTakeMVar ∘ un -{-| @release@ changes the state to unlocked and returns immediately.+{-| +@release@ changes the state to \"unlocked\" and returns immediately. -Note that it is an error to release an unlocked lock!+Note that it is an error to release a lock in the \"unlocked\" state! If there are any threads blocked on 'acquire' the thread that first called @acquire@ will be woken up.@@ -130,17 +137,19 @@ 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.+{-| +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+{-| +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.@@ -149,13 +158,30 @@ tryWith l a = block $ do acquired ← tryAcquire l if acquired- then fmap Just $ a `finally` release l+ then 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.+{-|+* When the state is \"locked\", @wait@ /blocks/ until a call to 'release' in+another thread changes it to \"unlocked\".++* When the state is \"unlocked\" @wait@ returns immediately.++@wait@ does not alter the state of the lock.++Note that @wait@ is just a convenience function defined as:++@wait l = 'block' '$' 'acquire' l '>>' 'release' l@+-}+wait ∷ Lock → IO ()+wait l = block $ acquire l >> release l++{-| +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+on its result it may already be out of date.+-} locked ∷ Lock → IO Bool locked = isEmptyMVar ∘ un
+ Control/Concurrent/Lock/Test.hs view
@@ -0,0 +1,86 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ , ScopedTypeVariables+ #-}++module Control.Concurrent.Lock.Test ( tests ) where++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( forkIO )+import Control.Monad ( (>>=), fail, (>>), fmap )+import Data.Bool ( not )+import Data.Function ( ($) )+import Prelude ( fromInteger )++-- from base-unicode-symbols:+import Data.Function.Unicode ( (∘) )+import Prelude.Unicode ( (⋅) )++-- from concurrent-extra:+import qualified Control.Concurrent.Lock as Lock+import TestUtils++-- from HUnit:+import Test.HUnit ( Assertion, assert )++-- from test-framework:+import Test.Framework ( Test )++-- from test-framework-hunit:+import Test.Framework.Providers.HUnit ( testCase )+++-------------------------------------------------------------------------------+-- Tests for Lock+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = [ 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+ ]++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 "" $ Lock.new >>= Lock.release++test_lock_4 ∷ Assertion+test_lock_4 = assert $ Lock.new >>= fmap not ∘ Lock.locked++test_lock_5 ∷ Assertion+test_lock_5 = assert $ Lock.newAcquired >>= Lock.locked++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+++-- The End ---------------------------------------------------------------------
Control/Concurrent/RLock.hs view
@@ -1,4 +1,8 @@-{-# LANGUAGE BangPatterns, DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-}+{-# LANGUAGE BangPatterns+ , DeriveDataTypeable+ , NoImplicitPrelude+ , UnicodeSyntax+ #-} -------------------------------------------------------------------------------- -- |@@ -8,7 +12,7 @@ -- Maintainer : Bas van Dijk <v.dijk.bas@gmail.com> -- , Roel van Dijk <vandijk.roel@gmail.com> ----- This module provides the 'RLock' synchronization mechanism. It was inspired+-- This module provides the 'RLock' synchronisation mechanism. It was inspired -- by the Python @RLock@ and Java @ReentrantLock@ objects and should behave in a -- similar way. See: --@@ -19,7 +23,7 @@ -- <http://java.sun.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantLock.html> -- -- All functions are /exception safe/. Throwing asynchronous exceptions will not--- compromise the internal state of a 'RLock'.+-- compromise the internal state of an 'RLock'. -- -- This module is intended to be imported qualified. We suggest importing it like: --@@ -32,23 +36,20 @@ module Control.Concurrent.RLock ( RLock- -- * Creating reentrant locks , new , newAcquired- -- * Locking and unlocking , acquire , tryAcquire- , release- -- * Convenience functions , with , tryWith-+ , wait -- * Querying reentrant locks- , recursionLevel+ , State+ , state ) where @@ -65,55 +66,112 @@ import Data.Bool ( Bool(False, True), otherwise ) import Data.Eq ( Eq ) import Data.Function ( ($) )-import Data.Maybe ( Maybe(Nothing, Just), maybe )-import Data.Tuple ( fst, snd )+import Data.Maybe ( Maybe(Nothing, Just) )+import Data.Tuple ( fst ) import Data.Typeable ( Typeable ) import Prelude ( Integer, fromInteger, succ, pred, error ) import System.IO ( IO ) --- from base-unicode-symbols+-- from base-unicode-symbols: import Data.Eq.Unicode ( (≡) ) import Data.Function.Unicode ( (∘) ) import Data.Monoid.Unicode ( (⊕) ) --- from ourselves:+-- from concurrent-extra: import Control.Concurrent.Lock ( Lock ) import qualified Control.Concurrent.Lock as Lock- ( new, newAcquired, acquire, release )+ ( new, newAcquired, acquire, release, wait ) -------------------------------------------------------------------------------- -- Reentrant locks -------------------------------------------------------------------------------- -newtype RLock = RLock {un ∷ MVar (Maybe (ThreadId, Integer), Lock)}+{-| A reentrant lock is in one of two states: \"locked\" or \"unlocked\". When+the lock is in the \"locked\" state it has two additional properties:++* Its /owner/: the thread that acquired the lock.++* Its /acquired count/: how many times its owner acquired the lock.+-}+newtype RLock = RLock {un ∷ MVar (State, Lock)} deriving (Eq, Typeable) +{-| The state of an 'RLock'.++* 'Nothing' indicates an \"unlocked\" state.++* @'Just' (tid, n)@ indicates a \"locked\" state where the thread identified by @tid@+acquired the lock @n@ times.+-}+type State = Maybe (ThreadId, Integer)++-- | 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).+-} newAcquired ∷ IO RLock newAcquired = do myTID ← myThreadId lock ← Lock.newAcquired RLock <$> newMVar (Just (myTID, 1), lock) +{-| +Acquires the 'RLock'. Blocks if another thread has acquired the 'RLock'.++@acquire@ behaves as follows:++* When the state is \"unlocked\", @acquire@ changes the state to \"locked\"+with the current thread as owner and an acquired count of 1.++* When the state is \"locked\" and the current thread owns the lock @acquire@+only increments the acquired count.++* When the state is \"locked\" and the current thread does not own the lock+@acquire@ /blocks/ until the owner releases the lock. If the thread that called+@acquire@ is woken upon release of the lock it will take ownership and change+the state to \"locked\" with an acquired count of 1.++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 ∷ RLock → IO () acquire (RLock mv) = do myTID ← myThreadId- block $ let go = do t@(mb, lock) ← takeMVar mv- case mb of- Nothing → do Lock.acquire lock- putMVar mv (Just (myTID, 1), lock)- Just (tid, n)- | myTID ≡ tid → let !sn = succ n- in putMVar mv (Just (tid, sn), lock)- | otherwise → do putMVar mv t- Lock.acquire lock- Lock.release lock- go- in go+ block $ let acq = do t@(mb, lock) ← takeMVar mv+ case mb of+ Nothing → do Lock.acquire lock+ putMVar mv (Just (myTID, 1), lock)+ Just (tid, n)+ | myTID ≡ tid → let !sn = succ n+ in putMVar mv (Just (tid, sn), lock)+ | otherwise → do putMVar mv t+ Lock.wait lock+ acq+ in acq +{-| +A non-blocking 'acquire'.++* When the state is \"unlocked\" @tryAcquire@ changes the state to \"locked\"+(with the current thread as owner and an acquired count of 1) and returns+'True'.++* When the state is \"locked\" @tryAcquire@ leaves the state unchanged and+returns 'False'.+-} tryAcquire ∷ RLock → IO Bool tryAcquire (RLock mv) = do myTID ← myThreadId@@ -131,6 +189,15 @@ | otherwise → do putMVar mv t return False +{-| @release@ decrements the acquired count. When a lock is released with an+acquired count of 1 its state is changed to \"unlocked\".++Note that it is both an error to release a lock in the \"unlocked\" state and to+release a lock that is not owned by the current thread.++If there are any threads blocked on 'acquire' the thread that first called+@acquire@ will be woken up.+-} release ∷ RLock → IO () release (RLock mv) = do myTID ← myThreadId@@ -148,18 +215,52 @@ in putMVar mv (Just (tid, pn), lock) | otherwise → err "Calling thread does not own the RLock!" +{-| 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 ∷ 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+by raising an exception, the lock is released and 'Just' the result of the+computation is returned.+-} tryWith ∷ RLock → IO α → IO (Maybe α) tryWith l a = block $ do acquired ← tryAcquire l if acquired- then fmap Just $ a `finally` release l+ then Just <$> a `finally` release l else return Nothing -recursionLevel ∷ RLock → IO Integer-recursionLevel = fmap (maybe 0 snd ∘ fst) ∘ readMVar ∘ un+{-|+* When the state is \"locked\" @wait@ /blocks/ until a call to 'release' in+another thread changes it to \"unlocked\".++* When the state is \"unlocked\" @wait@ returns immediately.++@wait@ does not alter the state of the lock.++Note that @wait@ is just a convenience function defined as:++@wait l = 'block' '$' 'acquire' l '>>' 'release' l@+-}+wait ∷ RLock → IO ()+wait l = block $ acquire l >> release l++{-| +Determine the state of the reentrant lock.++Note 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.+-}+state ∷ RLock → IO State+state = fmap fst ∘ readMVar ∘ un -- The End ---------------------------------------------------------------------
+ Control/Concurrent/RLock/Test.hs view
@@ -0,0 +1,81 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ , ScopedTypeVariables + #-}++module Control.Concurrent.RLock.Test ( tests ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( forkIO, threadDelay )+import Control.Monad ( (>>=), fail, (>>), replicateM_ )+import Data.Function ( ($) )+import Data.Int ( Int )+import Prelude ( fromInteger )++-- from base-unicode-symbols:+import Data.Function.Unicode ( (∘) )+import Prelude.Unicode ( (⋅) )++-- from concurrent-extra:+import qualified Control.Concurrent.Event as Event ( new, set, wait )+import qualified Control.Concurrent.RLock as RLock+import TestUtils++-- from HUnit:+import Test.HUnit ( Assertion, assert )++-- from test-framework:+import Test.Framework ( Test )++-- from test-framework-hunit:+import Test.Framework.Providers.HUnit ( testCase )+++-------------------------------------------------------------------------------+-- Tests for RLock+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = [ testCase "recursive acquire" $ test_rlock_1 5+ , testCase "conc acquire" $ test_rlock_2+ ]++test_rlock_1 ∷ Int → Assertion+test_rlock_1 n = assert ∘ within (10 ⋅ a_moment) $ do+ l ← RLock.new+ replicateM_ n $ RLock.acquire l+ replicateM_ n $ RLock.release l++-- Tests for bug found by Felipe Lessa.+test_rlock_2 ∷ Assertion+test_rlock_2 = assert ∘ within (20 ⋅ a_moment) $ do+ rl ← RLock.new+ t1_has_rlock ← Event.new+ t1_done ← Event.new+ t2_done ← Event.new++ -- Thread 1+ _ ← forkIO $ do+ RLock.acquire rl+ Event.set t1_has_rlock+ threadDelay $ 10 ⋅ a_moment+ RLock.release rl+ Event.set t1_done++ -- Thread 2+ _ ← forkIO $ do+ Event.wait t1_has_rlock+ RLock.acquire rl+ RLock.release rl+ Event.set t2_done++ Event.wait t1_done+ Event.wait t2_done+++-- The End ---------------------------------------------------------------------
Control/Concurrent/ReadWriteLock.hs view
@@ -1,4 +1,5 @@-{-# LANGUAGE DeriveDataTypeable+{-# LANGUAGE BangPatterns+ , DeriveDataTypeable , NamedFieldPuns , NoImplicitPrelude , TupleSections@@ -37,6 +38,8 @@ ( RWLock -- *Creating Read-Write Locks , new+ , newAcquiredRead+ , newAcquiredWrite -- *Read access -- **Blocking , acquireRead@@ -67,18 +70,16 @@ ) import Control.Exception ( block, bracket_, finally ) import Control.Monad ( return, (>>=), return, fail, (>>)- , when, liftM3, Functor+ , when, liftM3 ) import Data.Bool ( Bool(False, True) ) import Data.Char ( String ) import Data.Eq ( Eq, (==) )-import Data.Function ( ($), const )+import Data.Function ( ($), const, on ) 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,7 +90,9 @@ import Control.Concurrent.Lock ( Lock ) import qualified Control.Concurrent.Lock as Lock +import Utils ( void ) + ------------------------------------------------------------------------------- -- Read Write Lock -------------------------------------------------------------------------------@@ -106,21 +109,42 @@ data RWLock = RWLock { state ∷ MVar State , readLock ∷ Lock , writeLock ∷ Lock- } deriving (Eq, Typeable)+ } deriving Typeable +instance Eq RWLock where+ (==) = (==) `on` state+ -- | 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.+{-| 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) Lock.new Lock.new+new = liftM3 RWLock (newMVar Free)+ Lock.new+ Lock.new +{-| Create a new 'RWLock' in the \"Read\" state; only read can be acquired+without blocking.+-}+newAcquiredRead ∷ IO RWLock+newAcquiredRead = liftM3 RWLock (newMVar $ Read 1)+ Lock.newAcquired+ Lock.new++{-| Create a new 'RWLock' in the \"Write\" state; either acquiring read or write+will block.+-}+newAcquiredWrite ∷ IO RWLock+newAcquiredWrite = liftM3 RWLock (newMVar Write)+ Lock.new+ Lock.newAcquired+ {-| 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.@@ -131,7 +155,8 @@ case st of Free → do Lock.acquire readLock putMVar state (Read 1)- Read n → putMVar state (Read ∘ succ $! n)+ Read n → let !sn = succ n+ in putMVar state (Read sn) Write → do putMVar state st Lock.acquire writeLock modifyMVar_ state ∘ const $ do@@ -142,7 +167,7 @@ {-| 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@@ -151,7 +176,8 @@ Free → do Lock.acquire readLock putMVar state (Read 1) return True- Read n → do putMVar state (Read ∘ succ $! n)+ Read n → do let !sn = succ n+ putMVar state (Read sn) return True Write → do putMVar state st return False@@ -159,10 +185,10 @@ {-| 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@@ -171,7 +197,8 @@ Free → putMVar state st >> err Read 1 → do Lock.release readLock putMVar state Free- Read n → putMVar state (Read ∘ pred $! n)+ Read n → let !pn = pred n+ in putMVar state (Read pn) Write → putMVar state st >> err where err = error $ moduleName ⊕ ".releaseRead: already released"@@ -199,7 +226,7 @@ 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@@ -219,7 +246,7 @@ {-| 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@@ -238,10 +265,10 @@ {-| 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@@ -273,14 +300,8 @@ then Just <$> a `finally` releaseWrite l else return Nothing ---------------------------------------------------------------------------------- moduleName ∷ String moduleName = "Control.Concurrent.ReadWriteLock"--void ∷ Functor f ⇒ f α → f ()-void = (const () <$>) -- The End --------------------------------------------------------------------
+ Control/Concurrent/ReadWriteLock/Test.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ #-}++module Control.Concurrent.ReadWriteLock.Test ( tests ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( )++-- from base-unicode-symbols:+import Prelude.Unicode ( )++-- from concurrent-extra:+import qualified Control.Concurrent.ReadWriteLock as RWLock ( )+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 ReadWriteLock+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = []+++-- The End ---------------------------------------------------------------------
Control/Concurrent/ReadWriteVar.hs view
@@ -47,12 +47,9 @@ module Control.Concurrent.ReadWriteVar ( RWVar- -- *Creating Read-Write Variables , new- -- *Reading , with , tryWith- -- *Writing , modify_ , modify , tryModify_@@ -65,14 +62,12 @@ ------------------------------------------------------------------------------- -- from base-import Control.Monad ( return, (>>=), return, fail, (>>)- , fmap, liftM2- )+import Control.Monad ( (>>=), fmap, liftM2 ) import Data.Bool ( Bool(..) )-import Data.Eq ( Eq )-import Data.Function ( ($) )+import Data.Eq ( Eq, (==) )+import Data.Function ( ($), on ) import Data.Maybe ( Maybe(..), isJust )-import Data.IORef ( IORef, newIORef, readIORef, writeIORef )+import Data.IORef ( IORef, newIORef, readIORef ) import Data.Typeable ( Typeable ) import System.IO ( IO ) #ifdef __HADDOCK__@@ -87,14 +82,21 @@ import Control.Concurrent.ReadWriteLock ( RWLock ) import qualified Control.Concurrent.ReadWriteLock as RWLock +import Utils ( modifyIORefM, modifyIORefM_ ) + ------------------------------------------------------------------------------- -- Read-Write Variables: concurrent read, sequential write ------------------------------------------------------------------------------- -- | Concurrently readable and sequentially writable variable.-data RWVar α = RWVar RWLock (IORef α) deriving (Eq, Typeable)+data RWVar α = RWVar RWLock (IORef α) deriving Typeable +instance Eq (RWVar α) where+ (==) = (==) `on` rwlock+ where+ rwlock (RWVar rwl _) = rwl+ -- | Create a new 'RWVar'. new ∷ α → IO (RWVar α) new = liftM2 RWVar RWLock.new ∘ newIORef@@ -148,20 +150,6 @@ -} 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 --------------------------------------------------------------------
+ Control/Concurrent/ReadWriteVar/Test.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ #-}++module Control.Concurrent.ReadWriteVar.Test ( tests ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( )++-- from base-unicode-symbols:+import Prelude.Unicode ( )++-- from concurrent-extra:+import qualified Control.Concurrent.ReadWriteVar as RWVar ( )+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 ReadWriteVar+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = []+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/STM/Broadcast.hs view
@@ -0,0 +1,101 @@+{-# 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 <v.dijk.bas@gmail.com>+-- , Roel van Dijk <vandijk.roel@gmail.com>+--+-- 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 ---------------------------------------------------------------------
+ Control/Concurrent/STM/Broadcast/Test.hs view
@@ -0,0 +1,40 @@+{-# 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 ---------------------------------------------------------------------
Control/Concurrent/STM/Event.hs view
@@ -11,9 +11,9 @@ -- 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+-- 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+-- '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,72 +32,72 @@ module Control.Concurrent.STM.Event ( Event- , State(..) , new+ , newSet , wait , set , clear- , state+ , isSet ) 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 )+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.Eq.Unicode ( (≡) )+import Data.Function.Unicode ( (∘) ) -- from concurrent-extra-import Control.Concurrent.Event ( State(Cleared, Set) )+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 (TVar State) deriving (Eq, Typeable)+-- | An event is in one of two possible states: \"Set\" or \"Cleared\".+newtype Event = Event {evBroadcast ∷ Broadcast ()} deriving (Eq, Typeable) --- | Create an event. The initial state is 'Cleared'.+-- | Create an event in the \"Cleared\" state. new ∷ STM Event-new = Event <$> newTVar Cleared+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+-- 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+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'+-- | 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+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'.+-- | 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+clear = Broadcast.clear ∘ evBroadcast --- | The current state of the event.-state ∷ Event → STM State-state (Event tv) = readTVar tv+isSet ∷ Event → STM Bool+isSet = fmap isJust ∘ Broadcast.tryRead ∘ evBroadcast+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/STM/Event/Test.hs view
@@ -0,0 +1,40 @@+{-# 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 ---------------------------------------------------------------------
+ Control/Concurrent/Thread.hs view
@@ -0,0 +1,248 @@+{-# LANGUAGE DeriveDataTypeable+ , NoImplicitPrelude+ , UnicodeSyntax+ #-}++-------------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.Thread+-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk+-- License : BSD3 (see the file LICENSE)+-- Maintainer : Bas van Dijk <v.dijk.bas@gmail.com>+-- , Roel van Dijk <vandijk.roel@gmail.com>+--+-- Standard threads extended with the ability to wait for their termination.+--+-- Inspired by: <http://hackage.haskell.org/package/threadmanager>+--+-- This module re-implements several functions from @Control.Concurrent@. Avoid+-- ambiguities by importing one or both qualified. We suggest importing this+-- module like:+--+-- @+-- import qualified Control.Concurrent.Thread as Thread ( ... )+-- @+--+-------------------------------------------------------------------------------++module Control.Concurrent.Thread+ ( ThreadId+ , threadId+ , forkIO+ , forkOS+ , wait+ , waitTimeout+ , isRunning+ , killThread+ , killThreadTimeout+ , throwTo+ ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Applicative ( (<$>) )+import Control.Exception ( Exception, SomeException+ , AsyncException(ThreadKilled)+ , catch, block, unblock+ )+import Control.Monad ( return, (>>=), fail, (>>), fmap )+import Data.Bool ( Bool )+import Data.Eq ( Eq, (==) )+import Data.Function ( ($), on )+import Data.Maybe ( Maybe(Nothing, Just), 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 )++-- from base-unicode-symbols:+import Data.Function.Unicode ( (∘) )++-- from concurrent-extra:+import Control.Concurrent.Broadcast ( Broadcast )+import qualified Control.Concurrent.Broadcast as Broadcast+ ( new, write, read, tryRead, readTimeout )++import Utils ( void )+++-------------------------------------------------------------------------------+-- Threads+-------------------------------------------------------------------------------++{-|+A 'ThreadId' is an abstract type representing a handle to a thread. 'ThreadId'+is an instance of 'Eq', 'Ord' and 'Show', where the 'Ord' instance implements an+arbitrary total ordering over 'ThreadId's. The 'Show' instance lets you convert+an arbitrary-valued 'ThreadId' to string form; showing a 'ThreadId' value is+occasionally useful when debugging or diagnosing the behaviour of a concurrent+program.++Note: This is a wrapper around 'Conc.ThreadId'.+-}+data ThreadId = ThreadId+ { stopped ∷ Broadcast (Maybe SomeException)+ -- | Extract the underlying 'Conc.ThreadId'.+ , threadId ∷ Conc.ThreadId+ } deriving Typeable++instance Eq ThreadId where+ (==) = (==) `on` threadId++instance Ord ThreadId where+ compare = compare `on` threadId++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+ tid ← block $ doFork+ $ let writeStop = Broadcast.write stop+ in catch (unblock a >> writeStop Nothing)+ (writeStop ∘ Just)+ return $ ThreadId stop tid++{-|+Sparks off a new thread to run the given 'IO' computation and returns the+'ThreadId' of the newly created thread.++The new thread will be a lightweight thread; if you want to use a foreign+library that uses thread-local storage, use 'forkOS' instead.++GHC note: the new thread inherits the blocked state of the parent (see+'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@).+-}+forkIO ∷ IO () → IO ThreadId+forkIO = fork Conc.forkIO++{-|+Like 'forkIO', this sparks off a new thread to run the given 'IO' computation+and returns the 'ThreadId' of the newly created thread.++Unlike 'forkIO', 'forkOS' creates a /bound/ thread, which is necessary if you+need to call foreign (non-Haskell) libraries that make use of thread-local+state, such as OpenGL (see 'Control.Concurrent').++Using 'forkOS' instead of 'forkIO' makes no difference at all to the scheduling+behaviour of the Haskell runtime system. It is a common misconception that you+need to use 'forkOS' instead of 'forkIO' to avoid blocking all the Haskell+threads when making a foreign call; this isn't the case. To allow foreign calls+to be made without blocking all the Haskell threads (with GHC), it is only+necessary to use the @-threaded@ option when linking your program, and to make+sure the foreign import is not marked @unsafe@.+-}+forkOS ∷ IO () → IO ThreadId+forkOS = fork Conc.forkOS++{-|+Block until the given thread is terminated.++Returns 'Nothing' if the thread terminated normally or 'Just' @e@ if the+exception @e@ was thrown in the thread and wasn't caught.+-}+wait ∷ ThreadId → IO (Maybe SomeException)+wait = Broadcast.read ∘ stopped++{-|+Block until the given thread is terminated or until a timer expires.++Returns 'Nothing' if a timeout occurred or 'Just' the result 'wait' would return+when the thread finished within the specified time.++The timeout is specified in microseconds.+-}+waitTimeout ∷ ThreadId → Integer → IO (Maybe (Maybe SomeException))+waitTimeout = Broadcast.readTimeout ∘ stopped++{-|+Returns 'True' if the given thread is currently running.++Notice that this observation is only a snapshot of thread's state. By the time a+program reacts on its result it may already be out of date.+-}+isRunning ∷ ThreadId → IO Bool+isRunning = fmap isNothing ∘ Broadcast.tryRead ∘ stopped++-------------------------------------------------------------------------------+-- Convenience functions+-------------------------------------------------------------------------------++{-|+'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 t = throwTo t ThreadKilled >> void (wait t)++{-|+Like 'killThread' but with a timeout. Returns 'True' if the target thread was+terminated without 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++{-|+'throwTo' raises an arbitrary exception in the target thread (GHC only).++'throwTo' does not return until the exception has been raised in the target+thread. The calling thread can thus be certain that the target thread has+received the exception. This is a useful property to know when dealing with race+conditions: eg. if there are two threads that can kill each other, it is+guaranteed that only one of the threads will get to kill the other.++If the target thread is currently making a foreign call, then the exception will+not be raised (and hence 'throwTo' will not return) until the call has+completed. This is the case regardless of whether the call is inside a 'block'+or not.++Important note: the behaviour of 'throwTo' differs from that described in the+paper \"Asynchronous exceptions in Haskell\"+(<http://research.microsoft.com/~simonpj/Papers/asynch-exns.htm>). In the paper,+'throwTo' is non-blocking; but the library implementation adopts a more+synchronous design in which 'throwTo' does not return until the exception is+received by the target thread. The trade-off is discussed in Section 9 of the+paper. Like any blocking operation, 'throwTo' is therefore interruptible (see+Section 5.3 of the paper).++There is currently no guarantee that the exception delivered by 'throwTo' will+be delivered at the first possible opportunity. In particular, a thread may+'unblock' and then re-'block' exceptions without receiving a pending+'throwTo'. This is arguably undesirable behaviour.+-}+throwTo ∷ Exception e ⇒ ThreadId → e → IO ()+throwTo = Conc.throwTo ∘ threadId+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/Thread/Delay.hs view
@@ -0,0 +1,54 @@+{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax #-}++-------------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.Thread.Delay+-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk+-- License : BSD3 (see the file LICENSE)+-- Maintainer : Bas van Dijk <v.dijk.bas@gmail.com>+-- , Roel van Dijk <vandijk.roel@gmail.com>+--+-- Arbitrarily long thread delays.+-------------------------------------------------------------------------------++module Control.Concurrent.Thread.Delay ( delay ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( threadDelay )+import Control.Monad ( (>>), when )+import Data.Function ( ($) )+import Data.Int ( Int )+import Data.Ord ( min )+import Prelude ( Integer, toInteger, fromInteger, maxBound, (-) )+import System.IO ( IO )++-- from base-unicode-symbols:+import Data.Eq.Unicode ( (≢) )+++-------------------------------------------------------------------------------+-- Delay+-------------------------------------------------------------------------------++{-|+Like 'threadDelay', but not bounded by an 'Int'.++Suspends the current thread for a given number of microseconds (GHC only).++There is no guarantee that the thread will be rescheduled promptly when the+delay has expired, but the thread will never continue to run earlier than+specified.+-}+delay ∷ Integer → IO ()+delay time = do+ let maxWait = min time $ toInteger (maxBound ∷ Int)+ threadDelay $ fromInteger maxWait+ when (maxWait ≢ time) $ delay (time - maxWait)+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/Thread/Test.hs view
@@ -0,0 +1,80 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ #-}++module Control.Concurrent.Thread.Test ( tests ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( threadDelay )+import Control.Monad ( return, (>>=), fail, (>>), fmap )+import Data.Bool ( Bool(False, True) )+import Data.Function ( ($), id )+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 ( (⋅) )++-- from concurrent-extra:+import qualified Control.Concurrent.Lock as Lock+import qualified Control.Concurrent.Thread as Thread+import TestUtils ( a_moment, within )++-- from HUnit:+import Test.HUnit ( Assertion, assert )++-- from test-framework:+import Test.Framework ( Test )++-- from test-framework-hunit:+import Test.Framework.Providers.HUnit ( testCase )+++-------------------------------------------------------------------------------+-- Tests for Thread+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = [ testCase "wait" test_wait+ , testCase "waitTimeout" test_waitTimeout+ , testCase "isRunning" test_isRunning+ ]++test_wait ∷ Assertion+test_wait = assert $ fmap (maybe False id)+ $ timeout (10 ⋅ a_moment)+ $ do+ r ← newIORef False+ tid ← Thread.forkIO $ do+ threadDelay $ 2 ⋅ a_moment+ writeIORef r True+ _ ← Thread.wait tid+ readIORef r++test_waitTimeout ∷ Assertion+test_waitTimeout = assert $ within (10 ⋅ a_moment) $ do+ l ← Lock.newAcquired+ tid ← Thread.forkIO $ do+ 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+ l ← Lock.newAcquired+ tid ← Thread.forkIO $ Lock.acquire l+ r ← Thread.isRunning tid+ Lock.release l+ return r+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/Timeout.hs view
@@ -0,0 +1,115 @@+{-# LANGUAGE CPP+ , DeriveDataTypeable+ , NoImplicitPrelude+ , UnicodeSyntax+ #-}++-------------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.Timeout+-- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk+-- License : BSD3 (see the file LICENSE)+-- Maintainer : Bas van Dijk <v.dijk.bas@gmail.com>+-- , Roel van Dijk <vandijk.roel@gmail.com>+--+-- Wait arbitrarily long for an IO computation to finish.+-------------------------------------------------------------------------------++module Control.Concurrent.Timeout ( timeout ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( forkIO, myThreadId, throwTo, killThread )+import Control.Exception ( Exception, bracket, handleJust )+import Control.Monad ( return, (>>=), fail, (>>), fmap )+import Data.Bool ( otherwise )+import Data.Eq ( Eq )+import Data.Maybe ( Maybe(Nothing, Just) )+import Data.Ord ( (<) )+import Data.Typeable ( Typeable )+import Data.Unique ( Unique, newUnique )+import Prelude ( Integer, fromInteger+ )+import System.IO ( IO )+import Text.Show ( Show, show )++#ifdef __HADDOCK__+import Data.Int ( Int )+import System.IO ( hGetBuf, hPutBuf, hWaitForInput )+import qualified System.Timeout ( timeout )+#endif++-- from base-unicode-symbols:+import Data.Eq.Unicode ( (≡) )++-- from concurrent-extra:+import Control.Concurrent.Thread.Delay ( delay )+++-------------------------------------------------------------------------------+-- Long delays and timeouts+-------------------------------------------------------------------------------++{-+The following code was mostly copied from the module System.Timeout in the+package base-4.2.0.0.++(c) The University of Glasgow 2007+-}++data Timeout = Timeout Unique deriving (Eq, Typeable)++instance Show Timeout where+ show _ = "<<timeout>>"++instance Exception Timeout++{-|+Like 'System.Timeout.timeout', but not bounded by an 'Int'.++Wrap an 'IO' computation to time out and return 'Nothing' in case no result is+available within @n@ microseconds (@1\/10^6@ seconds). In case a result is+available before the timeout expires, 'Just' @a@ is returned. A negative timeout+interval means \"wait indefinitely\".++The design of this combinator was guided by the objective that @timeout n f@+should behave exactly the same as @f@ as long as @f@ doesn't time out. This+means that @f@ has the same 'myThreadId' it would have without the timeout+wrapper. Any exceptions @f@ might throw cancel the timeout and propagate further+up. It also possible for @f@ to receive exceptions thrown to it by another+thread.++A tricky implementation detail is the question of how to abort an 'IO'+computation. This combinator relies on asynchronous exceptions internally. The+technique works very well for computations executing inside of the Haskell+runtime system, but it doesn't work at all for non-Haskell code. Foreign+function calls, for example, cannot be timed out with this combinator simply+because an arbitrary C function cannot receive asynchronous exceptions. When+@timeout@ is used to wrap an FFI call that blocks, no timeout event can be+delivered until the FFI call returns, which pretty much negates the purpose of+the combinator. In practice, however, this limitation is less severe than it may+sound. Standard I\/O functions like 'System.IO.hGetBuf', 'System.IO.hPutBuf',+Network.Socket.accept, or 'System.IO.hWaitForInput' appear to be blocking, but+they really don't because the runtime system uses scheduling mechanisms like+@select(2)@ to perform asynchronous I\/O, so it is possible to interrupt+standard socket I\/O or file I\/O using this combinator.+-}+timeout ∷ Integer → IO α → IO (Maybe α)+timeout n f+ | n < 0 = fmap Just f+ | n ≡ 0 = return Nothing+ | otherwise = do+ pid ← myThreadId+ ex ← fmap Timeout newUnique+ handleJust (\e → if e ≡ ex then Just () else Nothing)+ (\_ → return Nothing)+ (bracket (forkIO (delay n >> throwTo pid ex))+ (killThread)+ (\_ → fmap Just f))+++-- The End ---------------------------------------------------------------------
+ Control/Concurrent/Timeout/Test.hs view
@@ -0,0 +1,40 @@+{-# LANGUAGE NoImplicitPrelude+ , UnicodeSyntax+ #-}++module Control.Concurrent.Timeout.Test ( tests ) where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Concurrent ( )++-- from base-unicode-symbols:+import Prelude.Unicode ( )++-- from concurrent-extra:+import Control.Concurrent.Timeout ( )+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 timeout+-------------------------------------------------------------------------------++tests ∷ [Test]+tests = []+++-- The End ---------------------------------------------------------------------
Setup.hs view
@@ -4,6 +4,7 @@ module Main (main) where + ------------------------------------------------------------------------------- -- Imports -------------------------------------------------------------------------------@@ -27,7 +28,11 @@ import Distribution.Simple.Setup ( HaddockFlags ) import Distribution.PackageDescription ( PackageDescription(..) ) + -------------------------------------------------------------------------------+-- Cabal setup program with support for 'cabal test' and+-- which sets the CPP define '__HADDOCK __' when haddock is run.+------------------------------------------------------------------------------- main ∷ IO () main = defaultMainWithHooks hooks@@ -52,3 +57,6 @@ haddockHook simpleUserHooks pkg (lbi { withPrograms = p }) where p = userSpecifyArgs "haddock" ["--optghc=-D__HADDOCK__"] (withPrograms lbi)+++-- The End ---------------------------------------------------------------------
+ TestUtils.hs view
@@ -0,0 +1,56 @@+{-# LANGUAGE NoImplicitPrelude+ , ScopedTypeVariables+ , UnicodeSyntax + #-}++module TestUtils where+++-------------------------------------------------------------------------------+-- Imports+-------------------------------------------------------------------------------++-- from base:+import Control.Applicative ( (<$>) )+import Control.Concurrent ( threadDelay )+import Control.Exception ( try, SomeException )+import Control.Monad ( (>>=), return, fail )+import Data.Bool ( Bool, not )+import Data.Char ( String )+import Data.Either ( Either(Left, Right) )+import Data.Int ( Int )+import Data.Maybe ( isJust )+import Prelude ( fromInteger )+import System.IO ( IO )+import System.Timeout ( timeout )++-- from HUnit:+import Test.HUnit ( Assertion, assertFailure )+++-------------------------------------------------------------------------------+-- Utilities for testing+-------------------------------------------------------------------------------++-- 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+++-- The End ---------------------------------------------------------------------
+ Utils.hs view
@@ -0,0 +1,43 @@+{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax #-}++module Utils where++--------------------------------------------------------------------------------+-- Imports+--------------------------------------------------------------------------------++-- 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 Data.IORef ( IORef, readIORef, writeIORef )+import System.IO ( IO )++-- from base-unicode-symbols:+import Data.Function.Unicode ( (∘) )+++--------------------------------------------------------------------------------+-- Utility functions+--------------------------------------------------------------------------------++void ∷ Functor f ⇒ f α → f ()+void = fmap $ const ()++purelyModifyMVar ∷ MVar α → (α → α) → IO ()+purelyModifyMVar mv f = block $ takeMVar mv >>= putMVar mv ∘ f++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 = readIORef r >>= f >>= writeIORef r+++-- The End ---------------------------------------------------------------------
concurrent-extra.cabal view
@@ -1,5 +1,5 @@ name: concurrent-extra-version: 0.1.0.1+version: 0.2 cabal-version: >= 1.6 build-type: Custom stability: experimental@@ -13,30 +13,45 @@ category: Concurrency synopsis: Extra concurrency primitives description:- Offers a selection of synchronization primitives:+ The @concurrent-extra@ package offers among other things the+ following selection of synchronisation primitives: .- * Lock: Enforce exclusive access to a resource. Also known as a- binary semaphore.+ * @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+ * @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.+ * @Broadcast@: Wake multiple threads by broadcasting a value. .- * ReadWriteLock: Multiple-reader, single-writer locks. Used to+ * @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.+ * @ReadWriteVar@: Concurrent read, sequential write variables. .+ The package also provides @STM@ versions of @Broadcast@ and @Event@.+ .+ Besides these synchronisation primitives the package provides:+ .+ * @Thread@: Threads extended with the ability to wait for their+ termination.+ .+ * @delay@: Arbitrarily long thread delays.+ .+ * @timeout@: Wait arbitrarily long for an IO computation to finish.+ . Please consult the API documentation of the individual modules for more detailed information. .- Inspired by the concurrency libraries of Java and Python.+ This package was inspired by the concurrency libraries of Java and+ Python. ---source-repository head--- Type: darcs--- Location: http://code.haskell.org/concurrent-extra+source-repository head+ Type: darcs+ Location: http://code.haskell.org/concurrent-extra ------------------------------------------------------------------------------- @@ -61,9 +76,15 @@ exposed-modules: Control.Concurrent.Lock , Control.Concurrent.RLock , Control.Concurrent.Event+ , Control.Concurrent.Broadcast , Control.Concurrent.ReadWriteLock , 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 if flag(nolib)@@ -73,6 +94,18 @@ executable test-concurrent-extra main-is: test.hs+ other-modules: Control.Concurrent.Event.Test+ , Control.Concurrent.Lock.Test+ , Control.Concurrent.RLock.Test+ , Control.Concurrent.Broadcast.Test+ , 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+ ghc-options: -Wall if flag(test)
test.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax, ScopedTypeVariables #-}+{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax #-} module Main where @@ -6,42 +6,25 @@ -- 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.RLock as RLock-import qualified Control.Concurrent.Event as Event+-- from base:+import System.IO ( IO ) --- from HUnit-import Test.HUnit hiding ( Test )+-- 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 ) --- from test-framework+-- from test-framework: import Test.Framework ( Test, defaultMain, testGroup ) --- from test-framework-hunit-import Test.Framework.Providers.HUnit ( testCase ) - ------------------------------------------------------------------------------- -- Tests -------------------------------------------------------------------------------@@ -50,193 +33,21 @@ 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 "wait blocks" $ 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+tests = [ testGroup "Pessimistic locking"+ [ testGroup "Event" Event.tests+ , testGroup "Lock" Lock.tests+ , testGroup "RLock" RLock.tests+ , testGroup "Broadcast" Broadcast.tests+ , testGroup "ReadWriteLock" RWLock.tests+ , testGroup "ReadWriteVar" RWVar.tests+ , testGroup "Thread" Thread.tests ]- , testGroup "RLock"- [ testCase "recursive acquire" $ test_rlock_1 5- , testCase "conc acquire" $ test_rlock_2+ , testGroup "STM"+ [ testGroup "Event" STM_Event.tests+ , testGroup "Broadcast" STM_Broadcast.tests ]+ , testGroup "Timeout" Timeout.tests ] ------------------------------------------------------------------------------------ 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-----------------------------------------------------------------------------------test_rlock_1 ∷ Int → Assertion-test_rlock_1 n = assert ∘ within (10 ⋅ a_moment) $ do- l ← RLock.new- replicateM_ n $ RLock.acquire l- replicateM_ n $ RLock.release l---- Tests for bug found by Felipe Lessa.-test_rlock_2 ∷ Assertion-test_rlock_2 = assert ∘ within (20 ⋅ a_moment) $ do- rl ← RLock.new- t1_has_rlock ← Event.new- t1_done ← Event.new- t2_done ← Event.new-- -- Thread 1- _ ← forkIO $ do- RLock.acquire rl- Event.set t1_has_rlock- threadDelay $ 10 ⋅ a_moment- RLock.release rl- Event.set t1_done-- -- Thread 2- _ ← forkIO $ do- Event.wait t1_has_rlock- RLock.acquire rl- RLock.release rl- Event.set t2_done-- Event.wait t1_done- Event.wait t2_done------------------------------------------------------------------------------------- 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+-- The End ---------------------------------------------------------------------