threads 0.2 → 0.3
raw patch · 6 files changed
+165/−297 lines, 6 files
Files
- Control/Concurrent/Thread.hs +53/−99
- Control/Concurrent/Thread/Group.hs +38/−34
- Control/Concurrent/Thread/Result.hs +0/−22
- Utils.hs +0/−74
- test.hs +72/−63
- threads.cabal +2/−5
Control/Concurrent/Thread.hs view
@@ -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 ---------------------------------------------------------------------
Control/Concurrent/Thread/Group.hs view
@@ -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 ---------------------------------------------------------------------
− Control/Concurrent/Thread/Result.hs
@@ -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
− Utils.hs
@@ -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 ---------------------------------------------------------------------
test.hs view
@@ -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 ---------------------------------------------------------------------
threads.cabal view
@@ -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 -------------------------------------------------------------------------------