diff --git a/Control/Concurrent/Thread.hs b/Control/Concurrent/Thread.hs --- a/Control/Concurrent/Thread.hs +++ b/Control/Concurrent/Thread.hs @@ -1,6 +1,6 @@ {-# LANGUAGE CPP, NoImplicitPrelude, UnicodeSyntax #-} -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- | -- Module : Control.Concurrent.Thread -- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk @@ -10,63 +10,58 @@ -- -- Standard threads extended with the ability to wait for their termination. -- --- This module exports equivalently named functions from --- @Control.Concurrent@. Avoid ambiguities by importing one or both --- qualified. We suggest importing this module like: +-- This module exports equivalently named functions from @Control.Concurrent@ +-- (and @GHC.Conc@). Avoid ambiguities by importing this module qualified. May +-- we suggest: -- -- @ -- import qualified Control.Concurrent.Thread as Thread ( ... ) -- @ -- -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- module Control.Concurrent.Thread - ( ThreadId - , threadId + ( -- * The result of a thread + Result -- * Forking threads , forkIO , forkOS +#ifdef __GLASGOW_HASKELL__ + , forkOnIO +#endif - -- * Waiting on threads + -- * Waiting for results , wait , wait_ , unsafeWait , unsafeWait_ - -- * Querying thread status + -- * Querying results , status , isRunning - - -- * Convenience functions - , throwTo - , killThread ) where -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- Imports -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- from base: -import qualified Control.Concurrent as C ( ThreadId, forkIO, forkOS, throwTo ) -import Control.Exception ( Exception, SomeException - , AsyncException(ThreadKilled) - , blocked, block, unblock, try - ) -import Control.Monad ( return, (>>=), (>>), fail ) -import Data.Bool ( Bool ) -import Data.Either ( Either, either ) +import qualified Control.Concurrent ( forkIO, forkOS ) +import Control.Concurrent ( ThreadId ) +import Control.Exception ( SomeException , blocked, block, unblock, try ) +import Control.Monad ( return, (>>=), fail ) +import Data.Bool ( Bool(..) ) +import Data.Either ( Either(..), either ) import Data.Function ( ($), const ) import Data.Functor ( fmap ) -import Data.Maybe ( Maybe, isNothing ) +import Data.Maybe ( Maybe(..), isNothing ) import System.IO ( IO ) -#ifdef __HADDOCK__ -import qualified Control.Concurrent as C ( killThread ) -import Data.Bool ( Bool (False, True) ) -import Data.Either ( Either(Left, Right) ) -import Data.Maybe ( Maybe (Nothing, Just) ) +#ifdef __GLASGOW_HASKELL__ +import qualified GHC.Conc ( forkOnIO ) +import Data.Int ( Int ) #endif -- from base-unicode-symbols: @@ -77,20 +72,19 @@ import Control.Concurrent.STM ( atomically ) -- from threads: -import Control.Concurrent.Thread.Internal ( ThreadId(ThreadId) - , result, threadId - ) +import Control.Concurrent.Thread.Result ( Result(Result), unResult ) import Utils ( void, throwInner, tryReadTMVar ) -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- * Forking threads -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- {-| Sparks off a new thread to run the given 'IO' computation and returns the -'ThreadId' of the newly created thread. +'ThreadId' of the newly created thread paired with the 'Result' of the thread +which can be @'wait'ed@ upon. The new thread will be a lightweight thread; if you want to use a foreign library that uses thread-local storage, use 'forkOS' instead. @@ -98,12 +92,13 @@ GHC note: the new thread inherits the blocked state of the parent (see 'Control.Exception.block'). -} -forkIO ∷ IO α → IO (ThreadId α) -forkIO = fork C.forkIO +forkIO ∷ IO α → IO (ThreadId, Result α) +forkIO = fork Control.Concurrent.forkIO {-| Like 'forkIO', this sparks off a new thread to run the given 'IO' computation -and returns the 'ThreadId' of the newly created thread. +and returns the 'ThreadId' of the newly created thread paired with the 'Result' +of the thread which can be @'wait'ed@ upon. 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 @@ -117,54 +112,74 @@ 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 C.forkOS +forkOS ∷ IO α → IO (ThreadId, Result α) +forkOS = fork Control.Concurrent.forkOS +#ifdef __GLASGOW_HASKELL__ {-| -Internally used function which generalises 'forkIO' and 'forkOS'. Parametrised -by the function which does the actual forking. +Like 'forkIO', but lets you specify on which CPU the thread is +created. Unlike a 'forkIO' thread, a thread created by 'forkOnIO' +will stay on the same CPU for its entire lifetime ('forkIO' threads +can migrate between CPUs according to the scheduling policy). +'forkOnIO' is useful for overriding the scheduling policy when you +know in advance how best to distribute the threads. + +The 'Int' argument specifies the CPU number; it is interpreted modulo +'numCapabilities' (note that it actually specifies a capability number +rather than a CPU number, but to a first approximation the two are +equivalent). -} -fork ∷ (IO () → IO C.ThreadId) → IO α → IO (ThreadId α) -fork doFork a = do +forkOnIO ∷ Int → IO α → IO (ThreadId, Result α) +forkOnIO = fork ∘ GHC.Conc.forkOnIO +#endif + +-------------------------------------------------------------------------------- + +-- | Internally used function which generalises 'forkIO', 'forkOS' and +-- 'forkOnIO' by parameterizing the function which does the actual forking. +fork ∷ (IO () → IO ThreadId) → (IO α → IO (ThreadId, Result α)) +fork doFork = \a → do res ← newEmptyTMVarIO - b ← blocked - fmap (ThreadId res) $ block $ doFork $ try (if b then a else unblock a) >>= - atomically ∘ putTMVar res + parentIsBlocked ← blocked + tid ← block $ doFork $ + try (if parentIsBlocked then a else unblock a) >>= + atomically ∘ putTMVar res + return (tid, Result res) -------------------------------------------------------------------------------- --- * Waiting on threads -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- +-- * Waiting for results +-------------------------------------------------------------------------------- {-| -Block until the given thread is terminated. +Block until the thread, to which the given 'Result' belongs, is terminated. * Returns @'Right' x@ if the thread terminated normally and returned @x@. * Returns @'Left' e@ if some exception @e@ was thrown in the thread and wasn't caught. -} -wait ∷ ThreadId α → IO (Either SomeException α) -wait = atomically ∘ readTMVar ∘ result +wait ∷ Result α → IO (Either SomeException α) +wait = atomically ∘ readTMVar ∘ unResult -- | Like 'wait' but will ignore the value returned by the thread. -wait_ ∷ ThreadId α → IO () +wait_ ∷ Result α → IO () wait_ = void ∘ wait -- | Like 'wait' but will either rethrow the exception that was thrown in the -- thread or return the value that was returned by the thread. -unsafeWait ∷ ThreadId α → IO α -unsafeWait tid = wait tid >>= either throwInner return +unsafeWait ∷ Result α → IO α +unsafeWait result = wait result >>= either throwInner return -- | Like 'unsafeWait' in that it will rethrow the exception that was thrown in -- the thread but it will ignore the value returned by the thread. -unsafeWait_ ∷ ThreadId α → IO () -unsafeWait_ tid = wait tid >>= either throwInner (const $ return ()) +unsafeWait_ ∷ Result α → IO () +unsafeWait_ result = wait result >>= either throwInner (const $ return ()) -------------------------------------------------------------------------------- --- * Quering thread status -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- +-- * Quering results +-------------------------------------------------------------------------------- {-| A non-blocking 'wait'. @@ -179,68 +194,18 @@ Notice that this observation is only a snapshot of a thread's state. By the time a program reacts on its result it may already be out of date. -} -status ∷ ThreadId α → IO (Maybe (Either SomeException α)) -status = tryReadTMVar ∘ result +status ∷ Result α → IO (Maybe (Either SomeException α)) +status = atomically ∘ tryReadTMVar ∘ unResult {-| -Returns 'True' if the thread is currently running and 'False' otherwise. +If the thread, to which the given 'Result' belongs, is currently running return +'True' and return 'False' otherwise. Notice that this observation is only a snapshot of a thread's state. By the time a program reacts on its result it may already be out of date. -} -isRunning ∷ ThreadId α → IO Bool +isRunning ∷ Result α → IO Bool isRunning = fmap isNothing ∘ status - - -------------------------------------------------------------------------------- --- * Convenience functions -------------------------------------------------------------------------------- - -{-| -'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\" -(). 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 = C.throwTo ∘ threadId - -{-| -'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'. - -Note that this function is different than @Control.Concurrent.'C.killThread'@ in -that it blocks until the target thread is terminated: - -@killThread tid = 'throwTo' tid 'ThreadKilled' '>>' 'wait_' tid@ --} -killThread ∷ ThreadId α → IO () -killThread tid = throwTo tid ThreadKilled >> wait_ tid -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Thread/Group.hs b/Control/Concurrent/Thread/Group.hs --- a/Control/Concurrent/Thread/Group.hs +++ b/Control/Concurrent/Thread/Group.hs @@ -4,7 +4,7 @@ , UnicodeSyntax #-} -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- | -- Module : Control.Concurrent.Thread.Group -- Copyright : (c) 2010 Bas van Dijk & Roel van Dijk @@ -15,25 +15,28 @@ -- This module extends @Control.Concurrent.Thread@ with the ability to wait for -- a group of threads to terminate. -- --- This module exports equivalently named functions from @Control.Concurrent@ --- and @Control.Concurrent.Thread@. Avoid ambiguities by importing one or both --- qualified. We suggest importing this module like: +-- This module exports equivalently named functions from @Control.Concurrent@, +-- (@GHC.Conc@), and @Control.Concurrent.Thread@. Avoid ambiguities by importing +-- this module qualified. May we suggest: -- -- @ -- import Control.Concurrent.Thread.Group ( ThreadGroup ) -- import qualified Control.Concurrent.Thread.Group as ThreadGroup ( ... ) -- @ -- ---------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- module Control.Concurrent.Thread.Group - ( -- * Creating + ( -- * Groups of threads ThreadGroup , new -- * Forking threads , forkIO , forkOS +#ifdef __GLASGOW_HASKELL__ + , forkOnIO +#endif -- * Waiting & Status , wait @@ -41,89 +44,122 @@ ) where -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- Imports -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- from base: -import Control.Exception ( blocked, block, unblock, try ) -import Control.Monad ( (>>=), (>>), fail, when, liftM2 ) -import Data.Bool ( Bool(..) ) -import Data.Function ( ($) ) -import Data.Functor ( fmap ) -import Data.Typeable ( Typeable ) -import System.IO ( IO ) -import qualified Control.Concurrent as C ( ThreadId, forkIO, forkOS ) -import Prelude ( Integer, fromInteger, (+), (-), ($!) ) +import qualified Control.Concurrent ( forkIO, forkOS ) +import Control.Concurrent ( ThreadId ) +import Control.Exception ( blocked, block, unblock, try ) +import Control.Monad ( return, (>>=), (>>), fail, when ) +import Data.Bool ( Bool(..) ) +import Data.Function ( ($) ) +import Data.Functor ( fmap ) +import Data.Typeable ( Typeable ) +import Prelude ( Integer, fromInteger, succ, pred ) +import System.IO ( IO ) +#ifdef __GLASGOW_HASKELL__ +import qualified GHC.Conc ( forkOnIO ) +import Data.Int ( Int ) +#endif + -- from base-unicode-symbols: -import Data.Eq.Unicode ( (≡) ) -import Data.Function.Unicode ( (∘) ) +import Data.Eq.Unicode ( (≢) ) +import Data.Function.Unicode ( (∘) ) -- from stm: -import Control.Concurrent.STM.TVar ( TVar, newTVar, writeTVar, readTVar ) -import Control.Concurrent.STM.TMVar ( newEmptyTMVarIO, putTMVar ) -import Control.Concurrent.STM ( atomically ) - --- from concurrent-extra: -import Control.Concurrent.STM.Lock ( Lock ) -import qualified Control.Concurrent.STM.Lock as Lock ( new - , acquire, release - , wait, locked - ) +import Control.Concurrent.STM.TVar ( TVar, newTVar, readTVar ) +import Control.Concurrent.STM.TMVar ( newEmptyTMVarIO, putTMVar ) +import Control.Concurrent.STM ( atomically, retry ) -- from threads: -import Control.Concurrent.Thread.Internal ( ThreadId( ThreadId ) ) +import Control.Concurrent.Thread.Result ( Result(Result) ) +import Utils ( modifyTVar ) + #ifdef __HADDOCK__ -import qualified Control.Concurrent.Thread as Thread ( forkIO, forkOS ) +import qualified Control.Concurrent.Thread as Thread ( forkIO + , forkOS +#ifdef __GLASGOW_HASKELL__ + , forkOnIO #endif + ) +#endif -------------------------------------------------------------------------------- --- Thread groups -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- +-- * Thread groups +-------------------------------------------------------------------------------- -data ThreadGroup = ThreadGroup (TVar Integer) Lock deriving Typeable +{-| A @ThreadGroup@ can be understood as a counter which counts the number of +threads that were added to the group minus the ones that have terminated. --- | Create a new empty group. +More formally a @ThreadGroup@ has the following semantics: + +* 'new' initializes the counter to 0. + +* Forking a thread increments the counter. + +* When a forked thread terminates, whether normally or by raising an exception, + the counter is decremented. + +* 'wait' blocks as long as the counter is not 0. +-} +newtype ThreadGroup = ThreadGroup (TVar Integer) deriving Typeable + +-- | Create an empty group of threads. new ∷ IO ThreadGroup -new = atomically $ liftM2 ThreadGroup (newTVar 0) (Lock.new) +new = atomically $ fmap ThreadGroup $ newTVar 0 + +-------------------------------------------------------------------------------- +-- * Forking threads +-------------------------------------------------------------------------------- + -- | Same as @Control.Concurrent.Thread.'Thread.forkIO'@ but additionaly adds -- the thread to the group. -forkIO ∷ ThreadGroup → IO α → IO (ThreadId α) -forkIO = fork C.forkIO +forkIO ∷ ThreadGroup → IO α → IO (ThreadId, Result α) +forkIO = fork Control.Concurrent.forkIO -- | Same as @Control.Concurrent.Thread.'Thread.forkOS'@ but additionaly adds -- the thread to the group. -forkOS ∷ ThreadGroup → IO α → IO (ThreadId α) -forkOS = fork C.forkOS +forkOS ∷ ThreadGroup → IO α → IO (ThreadId, Result α) +forkOS = fork Control.Concurrent.forkOS -fork ∷ (IO () → IO C.ThreadId) → ThreadGroup → IO α → IO (ThreadId α) -fork doFork (ThreadGroup mc l) a = do +#ifdef __GLASGOW_HASKELL__ +-- | Same as @Control.Concurrent.Thread.'Thread.forkOnIO'@ but +-- additionaly adds the thread to the group. (GHC only) +forkOnIO ∷ Int → ThreadGroup → IO α → IO (ThreadId, Result α) +forkOnIO = fork ∘ GHC.Conc.forkOnIO +#endif + +-------------------------------------------------------------------------------- + +-- | Internally used function which generalises 'forkIO', 'forkOS' and +-- 'forkOnIO' by parameterizing the function which does the actual forking. +fork ∷ (IO () → IO ThreadId) → ThreadGroup → IO α → IO (ThreadId, Result α) +fork doFork (ThreadGroup numThreadsTV) a = do res ← newEmptyTMVarIO - b ← blocked - block $ do - atomically increment - fmap (ThreadId res) $ doFork $ do - r ← try (if b then a else unblock a) - atomically $ putTMVar res r >> decrement - where - increment = do numThreads ← readTVar mc - when (numThreads ≡ 0) $ Lock.acquire l - writeTVar mc $! numThreads + 1 + parentIsBlocked ← blocked + tid ← block $ do + atomically $ modifyTVar numThreadsTV succ + doFork $ do + r ← try $ if parentIsBlocked then a else unblock a + atomically $ modifyTVar numThreadsTV pred >> putTMVar res r + return (tid, Result res) - decrement = do numThreads ← readTVar mc - when (numThreads ≡ 1) $ Lock.release l - writeTVar mc $! numThreads - 1 -lock ∷ ThreadGroup → Lock -lock (ThreadGroup _ l) = l +-------------------------------------------------------------------------------- +-- * Waiting & Status +-------------------------------------------------------------------------------- -- | Block until all threads, that were added to the group have terminated. wait ∷ ThreadGroup → IO () -wait = atomically ∘ Lock.wait ∘ lock +wait (ThreadGroup numThreadsTV) = atomically $ do + numThreads ← readTVar numThreadsTV + when (numThreads ≢ 0) retry {-| Returns 'True' if any thread in the group is running and returns 'False' @@ -133,7 +169,8 @@ a program reacts on its result it may already be out of date. -} isAnyRunning ∷ ThreadGroup → IO Bool -isAnyRunning = atomically ∘ Lock.locked ∘ lock +isAnyRunning (ThreadGroup numThreadsTV) = atomically $ + fmap (≢ 0) $ readTVar numThreadsTV -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Thread/Internal.hs b/Control/Concurrent/Thread/Internal.hs deleted file mode 100644 --- a/Control/Concurrent/Thread/Internal.hs +++ /dev/null @@ -1,60 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} - -module Control.Concurrent.Thread.Internal - ( ThreadId( ThreadId ) - , result, threadId - ) where - - -------------------------------------------------------------------------------- --- Imports -------------------------------------------------------------------------------- - --- from base: -import Control.Exception ( SomeException ) -import Data.Eq ( Eq, (==) ) -import Data.Either ( Either ) -import Data.Function ( on ) -import Data.Ord ( Ord, compare ) -import Data.Typeable ( Typeable ) -import Text.Show ( Show, show ) -import qualified Control.Concurrent as C ( ThreadId ) - --- from base-unicode-symbols: -import Data.Function.Unicode ( (∘) ) - --- from stm: -import Control.Concurrent.STM.TMVar ( TMVar ) - - -------------------------------------------------------------------------------- --- ThreadIds -------------------------------------------------------------------------------- - -{-| -A @'ThreadId' α@ is an abstract type representing a handle to a thread -that is executing or has executed a computation of type @'IO' α@. --} -data ThreadId α = ThreadId - { result ∷ TMVar (Either SomeException α) - , threadId ∷ C.ThreadId -- ^ Extract the native - -- @Control.Concurrent.'C.ThreadId'@. - } deriving Typeable - -instance Eq (ThreadId α) where - (==) = (==) `on` threadId - --- | The @Ord@ instance implements an arbitrary total ordering over 'ThreadId's. -instance Ord (ThreadId α) where - compare = compare `on` threadId - -{-| -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. --} -instance Show (ThreadId α) where - show = show ∘ threadId - - --- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Thread/Result.hs b/Control/Concurrent/Thread/Result.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Thread/Result.hs @@ -0,0 +1,22 @@ +{-# LANGUAGE CPP, DeriveDataTypeable, NoImplicitPrelude, UnicodeSyntax #-} + +module Control.Concurrent.Thread.Result ( Result(Result, unResult) ) where + +-- from base: +import Control.Exception ( SomeException ) +import Data.Either ( Either ) +import Data.Typeable ( Typeable ) + +#ifdef __HADDOCK__ +import System.IO ( IO ) +#endif + +-- from stm: +import Control.Concurrent.STM.TMVar ( TMVar ) + +{-| +A @'Result' α@ is an abstract type representing the result of a thread +that is executing or has executed a computation of type @'IO' α@. +-} +newtype Result α = Result { unResult ∷ TMVar (Either SomeException α) } + deriving Typeable diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2009 Bas van Dijk & Roel van Dijk +Copyright (c) 2010 Bas van Dijk & Roel van Dijk All rights reserved. diff --git a/TestUtils.hs b/TestUtils.hs deleted file mode 100644 --- a/TestUtils.hs +++ /dev/null @@ -1,56 +0,0 @@ -{-# 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 --------------------------------------------------------------------- diff --git a/Utils.hs b/Utils.hs --- a/Utils.hs +++ b/Utils.hs @@ -9,39 +9,66 @@ -- from base: import Control.Exception ( SomeException(SomeException), throwIO ) import Control.Monad ( Monad, return, (>>=), (>>), fail ) -import Data.Bool ( Bool ) -import Data.Function ( ($), flip ) +import Data.Bool ( Bool(..) ) +import Data.Eq ( Eq ) +import Data.Function ( flip, id ) import Data.Functor ( Functor, (<$>), (<$) ) -import Data.Maybe ( Maybe(Nothing, Just) ) +import Data.Maybe ( Maybe(Nothing, Just), maybe ) +import Prelude ( ($!) ) import System.IO ( IO ) +-- from base-unicode-symbols: +import Data.Eq.Unicode ( (≡) ) +import Data.Function.Unicode ( (∘) ) + -- from stm: -import Control.Concurrent.STM ( atomically ) +import Control.Concurrent.STM ( STM ) import Control.Concurrent.STM.TMVar ( TMVar, tryTakeTMVar, putTMVar ) +import Control.Concurrent.STM.TVar ( TVar, readTVar, writeTVar ) -------------------------------------------------------------------------------- -- Utility functions -------------------------------------------------------------------------------- +-- | Check if the given value equals 'Just' 'True'. +isJustTrue ∷ Maybe Bool → Bool +isJustTrue = maybe False id + +-- | Check if the given value in the 'Maybe' equals the given reference value. +justEq ∷ Eq α ⇒ α → Maybe α → Bool +justEq = maybe False ∘ (≡) + +-- | A flipped '<$>'. (<$$>) ∷ Functor f ⇒ f α → (α → β) → f β (<$$>) = flip (<$>) +-- | Ignore the value. void ∷ Functor f ⇒ f α → f () void = (() <$) +-- | Monadic @if then else@ ifM ∷ Monad m ⇒ m Bool → m α → m α → m α ifM c t e = c >>= \b → if b then t else e +-- | Throw the exception stored inside the 'SomeException'. throwInner ∷ SomeException → IO α throwInner (SomeException e) = throwIO e -tryReadTMVar ∷ TMVar α → IO (Maybe α) -tryReadTMVar mv = atomically $ do - mx ← tryTakeTMVar mv - case mx of - Nothing → return mx - Just x → putTMVar mv x >> return mx +-- | Non retrying 'takeMVar'. +tryReadTMVar ∷ TMVar α → STM (Maybe α) +tryReadTMVar mv = do mx ← tryTakeTMVar mv + case mx of + Nothing → return mx + Just x → putTMVar mv x >> return mx + +-- | Strictly modify the contents of a 'TVar'. +modifyTVar ∷ TVar α → (α → α) → STM () +modifyTVar tv f = readTVar tv >>= writeTVar tv ∘! f + +-- | Strict function composition +(∘!) ∷ (β → γ) → (α → β) → (α → γ) +f ∘! g = \x → f $! g x -- The End --------------------------------------------------------------------- diff --git a/test.hs b/test.hs --- a/test.hs +++ b/test.hs @@ -1,13 +1,17 @@ -{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax, DeriveDataTypeable #-} +{-# LANGUAGE CPP + , NoImplicitPrelude + , UnicodeSyntax + , DeriveDataTypeable + #-} module Main where -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- Imports -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- from base: -import Control.Concurrent ( threadDelay ) +import Control.Concurrent ( ThreadId, threadDelay, throwTo, killThread ) import Control.Exception ( Exception, fromException , AsyncException(ThreadKilled) , throwIO @@ -17,10 +21,10 @@ import Data.Bool ( Bool(False, True), not ) import Data.Eq ( Eq ) import Data.Either ( either ) -import Data.Function ( ($), id, const ) +import Data.Function ( ($), const ) import Data.Functor ( fmap, (<$>) ) +import Data.Int ( Int ) import Data.IORef ( newIORef, readIORef, writeIORef ) -import Data.Maybe ( maybe ) import Data.Typeable ( Typeable ) import Prelude ( fromInteger ) import System.Timeout ( timeout ) @@ -29,11 +33,10 @@ -- from base-unicode-symbols: import Prelude.Unicode ( (⋅) ) -import Data.Eq.Unicode ( (≡) ) import Data.Function.Unicode ( (∘) ) -- from concurrent-extra: -import qualified Control.Concurrent.Lock as Lock +import qualified Control.Concurrent.Lock as Lock -- from HUnit: import Test.HUnit ( Assertion, assert ) @@ -45,93 +48,142 @@ import Test.Framework.Providers.HUnit ( testCase ) -- from threads: -import qualified Control.Concurrent.Thread as Thread +import Control.Concurrent.Thread ( Result ) +import Control.Concurrent.Thread.Group ( ThreadGroup ) + +import qualified Control.Concurrent.Thread as Thread import qualified Control.Concurrent.Thread.Group as ThreadGroup -import TestUtils ( a_moment ) -import Utils ( (<$$>) ) +import Utils ( isJustTrue, justEq, (<$$>) ) -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- Tests -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- main ∷ IO () main = defaultMain tests tests ∷ [Test] tests = [ testGroup "Thread" $ - [ testCase "wait" $ test_wait Thread.forkIO - , testCase "isRunning" $ test_isRunning Thread.forkIO - , testCase "blockedState" $ test_blockedState Thread.forkIO - , testCase "unblockedState" $ test_unblockedState Thread.forkIO - , testCase "sync exception" $ test_sync_exception Thread.forkIO - , testCase "async exception" $ test_async_exception Thread.forkIO + [ testGroup "forkIO" $ + [ testCase "wait" $ test_wait Thread.forkIO + , testCase "isRunning" $ test_isRunning Thread.forkIO + , testCase "blockedState" $ test_blockedState Thread.forkIO + , testCase "unblockedState" $ test_unblockedState Thread.forkIO + , testCase "sync exception" $ test_sync_exception Thread.forkIO + , testCase "async exception" $ test_async_exception Thread.forkIO + ] + , testGroup "forkOS" $ + [ testCase "wait" $ test_wait Thread.forkOS + , testCase "isRunning" $ test_isRunning Thread.forkOS + , testCase "blockedState" $ test_blockedState Thread.forkOS + , testCase "unblockedState" $ test_unblockedState Thread.forkOS + , testCase "sync exception" $ test_sync_exception Thread.forkOS + , testCase "async exception" $ test_async_exception Thread.forkOS + ] +#ifdef __GLASGOW_HASKELL__ + , testGroup "forkOnIO 0" $ + [ testCase "wait" $ test_wait (Thread.forkOnIO 0) + , testCase "isRunning" $ test_isRunning (Thread.forkOnIO 0) + , testCase "blockedState" $ test_blockedState (Thread.forkOnIO 0) + , testCase "unblockedState" $ test_unblockedState (Thread.forkOnIO 0) + , testCase "sync exception" $ test_sync_exception (Thread.forkOnIO 0) + , testCase "async exception" $ test_async_exception (Thread.forkOnIO 0) + ] +#endif ] , testGroup "ThreadGroup" $ - [ testCase "wait" $ wrap test_wait - , testCase "isRunning" $ wrap test_isRunning - , testCase "blockedState" $ wrap test_blockedState - , testCase "unblockedState" $ wrap test_unblockedState - , testCase "sync exception" $ wrap test_sync_exception - , testCase "async exception" $ wrap test_async_exception + [ testGroup "forkIO" $ + [ testCase "wait" $ wrapIO test_wait + , testCase "isRunning" $ wrapIO test_isRunning + , testCase "blockedState" $ wrapIO test_blockedState + , testCase "unblockedState" $ wrapIO test_unblockedState + , testCase "sync exception" $ wrapIO test_sync_exception + , testCase "async exception" $ wrapIO test_async_exception - , testCase "group single wait" test_group_single_wait - , testCase "group single isRunning" test_group_single_isRunning + , testCase "group single wait" $ test_group_single_wait ThreadGroup.forkIO + , testCase "group single isRunning" $ test_group_single_isRunning ThreadGroup.forkIO + ] + , testGroup "forkOS" $ + [ testCase "wait" $ wrapOS test_wait + , testCase "isRunning" $ wrapOS test_isRunning + , testCase "blockedState" $ wrapOS test_blockedState + , testCase "unblockedState" $ wrapOS test_unblockedState + , testCase "sync exception" $ wrapOS test_sync_exception + , testCase "async exception" $ wrapOS test_async_exception + + , testCase "group single wait" $ test_group_single_wait ThreadGroup.forkOS + , testCase "group single isRunning" $ test_group_single_isRunning ThreadGroup.forkOS + ] +#ifdef __GLASGOW_HASKELL__ + , testGroup "forkOnIO 0" $ + [ testCase "wait" $ wrapOnIO_0 test_wait + , testCase "isRunning" $ wrapOnIO_0 test_isRunning + , testCase "blockedState" $ wrapOnIO_0 test_blockedState + , testCase "unblockedState" $ wrapOnIO_0 test_unblockedState + , testCase "sync exception" $ wrapOnIO_0 test_sync_exception + , testCase "async exception" $ wrapOnIO_0 test_async_exception + + , testCase "group single wait" $ test_group_single_wait (ThreadGroup.forkOnIO 0) + , testCase "group single isRunning" $ test_group_single_isRunning (ThreadGroup.forkOnIO 0) + ] +#endif ] ] +-- Exactly 1 moment. Currently equal to 0.005 seconds. +a_moment ∷ Int +a_moment = 5000 -------------------------------------------------------------------------------- + +-------------------------------------------------------------------------------- -- General properties -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -type Fork α = IO α → IO (Thread.ThreadId α) +type Fork α = IO α → IO (ThreadId, Result α) test_wait ∷ Fork () → Assertion -test_wait fork = assert - $ fmap (maybe False id) - $ timeout (10 ⋅ a_moment) $ do +test_wait fork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do r ← newIORef False - tid ← fork $ do + (_, result) ← fork $ do threadDelay $ 2 ⋅ a_moment writeIORef r True - _ ← Thread.wait tid + _ ← Thread.wait result readIORef r test_isRunning ∷ Fork () → Assertion -test_isRunning fork = assert - $ fmap (maybe False id) - $ timeout (10 ⋅ a_moment) $ do +test_isRunning fork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do l ← Lock.newAcquired - tid ← fork $ Lock.acquire l - r ← Thread.isRunning tid + (_, result) ← fork $ Lock.acquire l + r ← Thread.isRunning result Lock.release l return r test_blockedState ∷ Fork Bool → Assertion -test_blockedState fork = (block $ fork $ blocked) >>= - Thread.unsafeWait >>= assert +test_blockedState fork = do (_, result) ← block $ fork $ blocked + Thread.unsafeWait result >>= assert test_unblockedState ∷ Fork Bool → Assertion -test_unblockedState fork = (unblock $ fork $ not <$> blocked) >>= - Thread.unsafeWait >>= assert +test_unblockedState fork = do (_, result) ← unblock $ fork $ not <$> blocked + Thread.unsafeWait result >>= assert test_sync_exception ∷ Fork () → Assertion -test_sync_exception fork = assert $ - fork (throwIO MyException) >>= waitForException MyException +test_sync_exception fork = assert $ do + (_, result) ← fork $ throwIO MyException + waitForException MyException result -waitForException ∷ (Exception e, Eq e) ⇒ e → Thread.ThreadId α → IO Bool -waitForException e tid = Thread.wait tid <$$> - either (maybe False (≡ e) ∘ fromException) - (const False) +waitForException ∷ (Exception e, Eq e) ⇒ e → Result α → IO Bool +waitForException e result = Thread.wait result <$$> + either (justEq e ∘ fromException) + (const False) test_async_exception ∷ Fork () → Assertion test_async_exception fork = assert $ do l ← Lock.newAcquired - tid ← fork $ Lock.acquire l - Thread.throwTo tid MyException - waitForException MyException tid + (tid, result) ← fork $ Lock.acquire l + throwTo tid MyException + waitForException MyException result data MyException = MyException deriving (Show, Eq, Typeable) instance Exception MyException @@ -139,38 +191,44 @@ test_killThread ∷ Fork () → Assertion test_killThread fork = assert $ do l ← Lock.newAcquired - tid ← fork $ Lock.acquire l - Thread.killThread tid - waitForException ThreadKilled tid + (tid, result) ← fork $ Lock.acquire l + killThread tid + waitForException ThreadKilled result -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -- ThreadGroup -------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- -wrap ∷ (Fork α → IO β) → IO β -wrap test = ThreadGroup.new >>= test ∘ ThreadGroup.forkIO +wrapIO ∷ (Fork α → IO β) → IO β +wrapIO = wrap ThreadGroup.forkIO -test_group_single_wait ∷ Assertion -test_group_single_wait = assert - $ fmap (maybe False id) - $ timeout (10 ⋅ a_moment) $ do +wrapOS ∷ (Fork α → IO β) → IO β +wrapOS = wrap ThreadGroup.forkOS + +#ifdef __GLASGOW_HASKELL__ +wrapOnIO_0 ∷ (Fork α → IO β) → IO β +wrapOnIO_0 = wrap $ ThreadGroup.forkOnIO 0 +#endif + +wrap ∷ (ThreadGroup → Fork α) → (Fork α → IO β) → IO β +wrap doFork test = ThreadGroup.new >>= test ∘ doFork + +test_group_single_wait ∷ (ThreadGroup → Fork ()) → Assertion +test_group_single_wait doFork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do tg ← ThreadGroup.new r ← newIORef False - _ ← ThreadGroup.forkIO tg $ do + _ ← doFork tg $ do threadDelay $ 2 ⋅ a_moment writeIORef r True _ ← ThreadGroup.wait tg readIORef r - -test_group_single_isRunning ∷ Assertion -test_group_single_isRunning = assert - $ fmap (maybe False id) - $ timeout (10 ⋅ a_moment) $ do +test_group_single_isRunning ∷ (ThreadGroup → Fork ()) → Assertion +test_group_single_isRunning doFork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do tg ← ThreadGroup.new l ← Lock.newAcquired - _ ← ThreadGroup.forkIO tg $ Lock.acquire l + _ ← doFork tg $ Lock.acquire l r ← ThreadGroup.isAnyRunning tg Lock.release l return r diff --git a/threads.cabal b/threads.cabal --- a/threads.cabal +++ b/threads.cabal @@ -1,5 +1,5 @@ name: threads -version: 0.1.0.1 +version: 0.2 cabal-version: >= 1.6 build-type: Custom stability: experimental @@ -21,9 +21,8 @@ this packages also provides functions to wait for a group of threads to terminate. . - This package is similar to: - . - The advantages of this package are: + This package is similar to the @threadmanager@ + and @async@ packages. The advantages of this package are: . * Simpler API. . @@ -32,6 +31,8 @@ * No space-leak when forking a large number of threads. . * Correct handling of asynchronous exceptions. + . + * GHC specific functionality like @forkOnIO@. source-repository head Type: darcs @@ -47,33 +48,24 @@ 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.3 , stm >= 2.1 && < 2.2 - , concurrent-extra >= 0.5.1 && < 0.6 exposed-modules: Control.Concurrent.Thread , Control.Concurrent.Thread.Group - other-modules: Control.Concurrent.Thread.Internal + other-modules: Control.Concurrent.Thread.Result , Utils ghc-options: -Wall - if flag(nolib) - buildable: False - ------------------------------------------------------------------------------- executable test-threads main-is: test.hs - other-modules: TestUtils - ghc-options: -Wall + ghc-options: -Wall -threaded if flag(test) build-depends: base >= 3 && < 4.3 @@ -81,10 +73,8 @@ , stm >= 2.1 && < 2.2 , concurrent-extra >= 0.5.1 && < 0.6 , HUnit >= 1.2.2 && < 1.3 - , QuickCheck >= 2.1.0 && < 2.2 , test-framework >= 0.2.4 && < 0.4 , test-framework-hunit >= 0.2.4 && < 0.3 - , test-framework-quickcheck2 >= 0.2.4 && < 0.3 buildable: True else buildable: False