diff --git a/Control/Concurrent/Thread.hs b/Control/Concurrent/Thread.hs --- a/Control/Concurrent/Thread.hs +++ b/Control/Concurrent/Thread.hs @@ -18,28 +18,32 @@ -- import qualified Control.Concurrent.Thread as Thread ( ... ) -- @ -- +-- The following is an example how to use this module: +-- +-- @ +-- +-- import qualified Control.Concurrent.Thread as Thread ( 'forkIO', 'unsafeResult' ) +-- +-- main = do (tid, wait) <- Thread.'forkIO' $ do x <- someExpensiveComputation +-- return x +-- doSomethingElse +-- x <- Thread.'unsafeResult' =<< 'wait' +-- doSomethingWithResult x +-- @ +-- -------------------------------------------------------------------------------- module Control.Concurrent.Thread - ( -- * The result of a thread - Result - - -- * Forking threads - , forkIO + ( -- * Forking threads + forkIO , forkOS #ifdef __GLASGOW_HASKELL__ , forkOnIO #endif - -- * Waiting for results - , wait - , wait_ - , unsafeWait - , unsafeWait_ - - -- * Querying results - , status - , isRunning + -- * Results + , Result + , unsafeResult ) where @@ -49,32 +53,23 @@ -- from base: 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 System.IO ( IO ) +import Control.Concurrent ( ThreadId ) +import Control.Concurrent.MVar ( newEmptyMVar, putMVar, readMVar ) +import Control.Exception ( SomeException(SomeException) + , blocked, block, unblock, try, throwIO + ) +import Control.Monad ( return, (>>=), fail ) +import Data.Either ( Either(..), either ) +import Data.Function ( ($) ) +import System.IO ( IO ) #ifdef __GLASGOW_HASKELL__ -import qualified GHC.Conc ( forkOnIO ) -import Data.Int ( Int ) +import qualified GHC.Conc ( forkOnIO ) +import Data.Int ( Int ) #endif -- from base-unicode-symbols: -import Data.Function.Unicode ( (∘) ) - --- from stm: -import Control.Concurrent.STM.TMVar ( newEmptyTMVarIO, putTMVar, readTMVar ) -import Control.Concurrent.STM ( atomically ) - --- from threads: -import Control.Concurrent.Thread.Result ( Result(Result), unResult ) - -import Utils ( void, throwInner, tryReadTMVar ) +import Data.Function.Unicode ( (∘) ) -------------------------------------------------------------------------------- @@ -83,8 +78,8 @@ {-| Sparks off a new thread to run the given 'IO' computation and returns the -'ThreadId' of the newly created thread paired with the 'Result' of the thread -which can be @'wait'ed@ upon. +'ThreadId' of the newly created thread paired with an IO computation that waits +for the result of the 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. @@ -92,13 +87,13 @@ GHC note: the new thread inherits the blocked state of the parent (see 'Control.Exception.block'). -} -forkIO ∷ IO α → IO (ThreadId, Result α) +forkIO ∷ IO α → IO (ThreadId, IO (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 paired with the 'Result' -of the thread which can be @'wait'ed@ upon. +and returns the 'ThreadId' of the newly created thread paired with an IO +computation that waits for the result of the 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 @@ -112,7 +107,7 @@ 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, Result α) +forkOS ∷ IO α → IO (ThreadId, IO (Result α)) forkOS = fork Control.Concurrent.forkOS #ifdef __GLASGOW_HASKELL__ @@ -129,7 +124,7 @@ rather than a CPU number, but to a first approximation the two are equivalent). -} -forkOnIO ∷ Int → IO α → IO (ThreadId, Result α) +forkOnIO ∷ Int → IO α → IO (ThreadId, IO (Result α)) forkOnIO = fork ∘ GHC.Conc.forkOnIO #endif @@ -137,75 +132,34 @@ -- | 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 ∷ (IO () → IO ThreadId) → (IO α → IO (ThreadId, IO (Result α))) fork doFork = \a → do - res ← newEmptyTMVarIO + res ← newEmptyMVar parentIsBlocked ← blocked tid ← block $ doFork $ - try (if parentIsBlocked then a else unblock a) >>= - atomically ∘ putTMVar res - return (tid, Result res) - - --------------------------------------------------------------------------------- --- * Waiting for results --------------------------------------------------------------------------------- - -{-| -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 ∷ Result α → IO (Either SomeException α) -wait = atomically ∘ readTMVar ∘ unResult - --- | Like 'wait' but will ignore the value returned by the thread. -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 ∷ 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_ ∷ Result α → IO () -unsafeWait_ result = wait result >>= either throwInner (const $ return ()) + try (if parentIsBlocked then a else unblock a) >>= putMVar res + return (tid, readMVar res) -------------------------------------------------------------------------------- --- * Quering results +-- Results -------------------------------------------------------------------------------- -{-| -A non-blocking 'wait'. - -* Returns 'Nothing' if the thread is still running. - -* Returns @'Just' ('Right' x)@ if the thread terminated normally and returned @x@. +-- | A result of a thread is either some exception that was thrown in the thread +-- and wasn't catched or the actual value that was returned by the thread. +type Result α = Either SomeException α -* Returns @'Just' ('Left' e)@ if some exception @e@ was thrown in the thread and -wasn't caught. +{-| Unsafely retrieve the actual value from the result. -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. +When the result is 'SomeException' the exception stored inside of the +'SomeException' is rethrown in the current thread. -} -status ∷ Result α → IO (Maybe (Either SomeException α)) -status = atomically ∘ tryReadTMVar ∘ unResult - -{-| -If the thread, to which the given 'Result' belongs, is currently running return -'True' and return 'False' otherwise. +unsafeResult ∷ Result α → IO α +unsafeResult = either throwInner return -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 ∷ Result α → IO Bool -isRunning = fmap isNothing ∘ status +-- | Throw the exception stored inside the 'SomeException'. +throwInner ∷ SomeException → IO α +throwInner (SomeException e) = throwIO e -- 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 @@ -30,6 +30,7 @@ ( -- * Groups of threads ThreadGroup , new + , nrOfRunning -- * Forking threads , forkIO @@ -38,9 +39,8 @@ , forkOnIO #endif - -- * Waiting & Status + -- * Waiting , wait - , isAnyRunning ) where @@ -51,13 +51,13 @@ -- from base: import qualified Control.Concurrent ( forkIO, forkOS ) import Control.Concurrent ( ThreadId ) +import Control.Concurrent.MVar ( newEmptyMVar, putMVar, readMVar ) 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 Prelude ( ($!), Integer, fromInteger, succ, pred ) import System.IO ( IO ) #ifdef __GLASGOW_HASKELL__ @@ -70,14 +70,11 @@ import Data.Function.Unicode ( (∘) ) -- from stm: -import Control.Concurrent.STM.TVar ( TVar, newTVar, readTVar ) -import Control.Concurrent.STM.TMVar ( newEmptyTMVarIO, putTMVar ) -import Control.Concurrent.STM ( atomically, retry ) +import Control.Concurrent.STM.TVar ( TVar, newTVar, readTVar, writeTVar ) +import Control.Concurrent.STM ( STM, atomically, retry ) -- from threads: -import Control.Concurrent.Thread.Result ( Result(Result) ) - -import Utils ( modifyTVar ) +import Control.Concurrent.Thread ( Result ) #ifdef __HADDOCK__ import qualified Control.Concurrent.Thread as Thread ( forkIO @@ -105,6 +102,8 @@ * When a forked thread terminates, whether normally or by raising an exception, the counter is decremented. +* 'nrOfRunning' yields a transaction that returns the counter. + * 'wait' blocks as long as the counter is not 0. -} newtype ThreadGroup = ThreadGroup (TVar Integer) deriving Typeable @@ -113,25 +112,34 @@ new ∷ IO ThreadGroup new = atomically $ fmap ThreadGroup $ newTVar 0 +{-| Yield a transaction that returns the number of running threads in the +group. +Note that because this function yields a 'STM' computation, the returned number +is guaranteed to be consistent inside the transaction. +-} +nrOfRunning ∷ ThreadGroup → STM Integer +nrOfRunning (ThreadGroup numThreadsTV) = readTVar numThreadsTV + + -------------------------------------------------------------------------------- -- * Forking threads -------------------------------------------------------------------------------- -- | Same as @Control.Concurrent.Thread.'Thread.forkIO'@ but additionaly adds -- the thread to the group. -forkIO ∷ ThreadGroup → IO α → IO (ThreadId, Result α) +forkIO ∷ ThreadGroup → IO α → IO (ThreadId, IO (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, Result α) +forkOS ∷ ThreadGroup → IO α → IO (ThreadId, IO (Result α)) forkOS = fork Control.Concurrent.forkOS #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 ∷ Int → ThreadGroup → IO α → IO (ThreadId, IO (Result α)) forkOnIO = fork ∘ GHC.Conc.forkOnIO #endif @@ -139,38 +147,34 @@ -- | 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 ∷ (IO () → IO ThreadId) → ThreadGroup → IO α → IO (ThreadId, IO (Result α)) fork doFork (ThreadGroup numThreadsTV) a = do - res ← newEmptyTMVarIO + res ← newEmptyMVar 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) + try (if parentIsBlocked then a else unblock a) >>= putMVar res + atomically $ modifyTVar numThreadsTV pred + return (tid, readMVar res) +-- | 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 + + -------------------------------------------------------------------------------- --- * Waiting & Status +-- * Waiting -------------------------------------------------------------------------------- --- | Block until all threads, that were added to the group have terminated. +-- | Convenience function which blocks until all threads, that were added to the +-- group have terminated. wait ∷ ThreadGroup → IO () -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' -otherwise. - -Notice that this observation is only a snapshot of a group's state. By the time -a program reacts on its result it may already be out of date. --} -isAnyRunning ∷ ThreadGroup → IO Bool -isAnyRunning (ThreadGroup numThreadsTV) = atomically $ - fmap (≢ 0) $ readTVar numThreadsTV +wait tg = atomically $ nrOfRunning tg >>= \n → when (n ≢ 0) retry -- The End --------------------------------------------------------------------- diff --git a/Control/Concurrent/Thread/Result.hs b/Control/Concurrent/Thread/Result.hs deleted file mode 100644 --- a/Control/Concurrent/Thread/Result.hs +++ /dev/null @@ -1,22 +0,0 @@ -{-# 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/Utils.hs b/Utils.hs deleted file mode 100644 --- a/Utils.hs +++ /dev/null @@ -1,74 +0,0 @@ -{-# LANGUAGE NoImplicitPrelude, UnicodeSyntax #-} - -module Utils where - --------------------------------------------------------------------------------- --- Imports --------------------------------------------------------------------------------- - --- from base: -import Control.Exception ( SomeException(SomeException), throwIO ) -import Control.Monad ( Monad, return, (>>=), (>>), fail ) -import Data.Bool ( Bool(..) ) -import Data.Eq ( Eq ) -import Data.Function ( flip, id ) -import Data.Functor ( Functor, (<$>), (<$) ) -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 ( 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 - --- | 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 @@ -17,13 +17,14 @@ , throwIO , unblock, block, blocked ) -import Control.Monad ( return, (>>=), fail, (>>) ) +import Control.Monad ( return, (>>=), fail, (>>), replicateM_ ) import Data.Bool ( Bool(False, True), not ) import Data.Eq ( Eq ) import Data.Either ( either ) -import Data.Function ( ($), const ) -import Data.Functor ( fmap, (<$>) ) +import Data.Function ( ($), id, const, flip ) +import Data.Functor ( Functor(fmap), (<$>) ) import Data.Int ( Int ) +import Data.Maybe ( Maybe, maybe ) import Data.IORef ( newIORef, readIORef, writeIORef ) import Data.Typeable ( Typeable ) import Prelude ( fromInteger ) @@ -32,12 +33,16 @@ import Text.Show ( Show ) -- from base-unicode-symbols: +import Data.Eq.Unicode ( (≡) ) import Prelude.Unicode ( (⋅) ) import Data.Function.Unicode ( (∘) ) -- from concurrent-extra: import qualified Control.Concurrent.Lock as Lock +-- from stm: +import Control.Concurrent.STM ( atomically ) + -- from HUnit: import Test.HUnit ( Assertion, assert ) @@ -48,15 +53,13 @@ import Test.Framework.Providers.HUnit ( testCase ) -- from threads: -import Control.Concurrent.Thread ( Result ) +import Control.Concurrent.Thread ( Result, unsafeResult ) import Control.Concurrent.Thread.Group ( ThreadGroup ) import qualified Control.Concurrent.Thread as Thread import qualified Control.Concurrent.Thread.Group as ThreadGroup -import Utils ( isJustTrue, justEq, (<$$>) ) - -------------------------------------------------------------------------------- -- Tests -------------------------------------------------------------------------------- @@ -68,7 +71,6 @@ tests = [ testGroup "Thread" $ [ 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 @@ -76,7 +78,6 @@ ] , 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 @@ -85,7 +86,6 @@ #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) @@ -95,38 +95,35 @@ ] , testGroup "ThreadGroup" $ [ 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 "wait" $ wrapIO test_wait + , 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 ThreadGroup.forkIO - , testCase "group single isRunning" $ test_group_single_isRunning ThreadGroup.forkIO + , testCase "group single wait" $ test_group_single_wait ThreadGroup.forkIO + , testCase "group nrOfRunning" $ test_group_nrOfRunning 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 "wait" $ wrapOS test_wait + , 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 + , testCase "group single wait" $ test_group_single_wait ThreadGroup.forkOS + , testCase "group nrOfRunning" $ test_group_nrOfRunning 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 "wait" $ wrapOnIO_0 test_wait + , 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) + , testCase "group single wait" $ test_group_single_wait (ThreadGroup.forkOnIO 0) + , testCase "group nrOfRunning" $ test_group_nrOfRunning (ThreadGroup.forkOnIO 0) ] #endif ] @@ -141,49 +138,40 @@ -- General properties -------------------------------------------------------------------------------- -type Fork α = IO α → IO (ThreadId, Result α) +type Fork α = IO α → IO (ThreadId, IO (Result α)) test_wait ∷ Fork () → Assertion test_wait fork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do r ← newIORef False - (_, result) ← fork $ do + (_, wait) ← fork $ do threadDelay $ 2 ⋅ a_moment writeIORef r True - _ ← Thread.wait result + _ ← wait readIORef r -test_isRunning ∷ Fork () → Assertion -test_isRunning fork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do - l ← Lock.newAcquired - (_, result) ← fork $ Lock.acquire l - r ← Thread.isRunning result - Lock.release l - return r - test_blockedState ∷ Fork Bool → Assertion -test_blockedState fork = do (_, result) ← block $ fork $ blocked - Thread.unsafeWait result >>= assert +test_blockedState fork = do (_, wait) ← block $ fork $ blocked + wait >>= unsafeResult >>= assert test_unblockedState ∷ Fork Bool → Assertion -test_unblockedState fork = do (_, result) ← unblock $ fork $ not <$> blocked - Thread.unsafeWait result >>= assert +test_unblockedState fork = do (_, wait) ← unblock $ fork $ not <$> blocked + wait >>= unsafeResult >>= assert test_sync_exception ∷ Fork () → Assertion test_sync_exception fork = assert $ do - (_, result) ← fork $ throwIO MyException - waitForException MyException result + (_, wait) ← fork $ throwIO MyException + waitForException MyException wait -waitForException ∷ (Exception e, Eq e) ⇒ e → Result α → IO Bool -waitForException e result = Thread.wait result <$$> - either (justEq e ∘ fromException) - (const False) +waitForException ∷ (Exception e, Eq e) ⇒ e → IO (Result α) → IO Bool +waitForException e wait = wait <$$> either (justEq e ∘ fromException) + (const False) test_async_exception ∷ Fork () → Assertion test_async_exception fork = assert $ do l ← Lock.newAcquired - (tid, result) ← fork $ Lock.acquire l + (tid, wait) ← fork $ Lock.acquire l throwTo tid MyException - waitForException MyException result + waitForException MyException wait data MyException = MyException deriving (Show, Eq, Typeable) instance Exception MyException @@ -191,9 +179,9 @@ test_killThread ∷ Fork () → Assertion test_killThread fork = assert $ do l ← Lock.newAcquired - (tid, result) ← fork $ Lock.acquire l + (tid, wait) ← fork $ Lock.acquire l killThread tid - waitForException ThreadKilled result + waitForException ThreadKilled wait -------------------------------------------------------------------------------- @@ -224,14 +212,35 @@ _ ← ThreadGroup.wait tg readIORef r -test_group_single_isRunning ∷ (ThreadGroup → Fork ()) → Assertion -test_group_single_isRunning doFork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do +test_group_nrOfRunning ∷ (ThreadGroup → Fork ()) → Assertion +test_group_nrOfRunning doFork = assert $ fmap isJustTrue $ timeout (10 ⋅ a_moment) $ do tg ← ThreadGroup.new l ← Lock.newAcquired - _ ← doFork tg $ Lock.acquire l - r ← ThreadGroup.isAnyRunning tg + replicateM_ (fromInteger n) $ doFork tg $ Lock.acquire l + true ← fmap (≡ n) $ atomically $ ThreadGroup.nrOfRunning tg Lock.release l - return r + return true + where + -- Don't set this number too big otherwise forkOS might throw an exception + -- indicating that too many OS threads have been created: + n = 100 + + +-------------------------------------------------------------------------------- +-- Utils +-------------------------------------------------------------------------------- + +-- | 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 (<$>) -- The End --------------------------------------------------------------------- diff --git a/threads.cabal b/threads.cabal --- a/threads.cabal +++ b/threads.cabal @@ -1,5 +1,5 @@ name: threads -version: 0.2 +version: 0.3 cabal-version: >= 1.6 build-type: Custom stability: experimental @@ -13,8 +13,7 @@ category: Concurrency synopsis: Fork threads and wait for their result description: This package provides functions to fork threads and - wait for their termination. The result of a thread can - also be retrieved, whether it's an exception or a + wait for their result, whether it's an exception or a normal value. . Besides waiting for the termination of a single thread @@ -56,8 +55,6 @@ , stm >= 2.1 && < 2.2 exposed-modules: Control.Concurrent.Thread , Control.Concurrent.Thread.Group - other-modules: Control.Concurrent.Thread.Result - , Utils ghc-options: -Wall -------------------------------------------------------------------------------