async 2.0.2 → 2.2.6
raw patch · 9 files changed
Files
- Control/Concurrent/Async.hs +128/−548
- Control/Concurrent/Async/Internal.hs +964/−0
- Control/Concurrent/Async/Warden.hs +95/−0
- Control/Concurrent/Stream.hs +138/−0
- async.cabal +83/−44
- bench/concasync.hs +13/−0
- bench/conccancel.hs +11/−0
- changelog.md +124/−0
- test/test-async.hs +435/−2
Control/Concurrent/Async.hs view
@@ -1,9 +1,3 @@-{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes #-}-#if __GLASGOW_HASKELL__ >= 701-{-# LANGUAGE Trustworthy #-}-#endif-{-# OPTIONS -Wall #-}- ----------------------------------------------------------------------------- -- | -- Module : Control.Concurrent.Async@@ -20,591 +14,177 @@ -- "Control.Concurrent". The main additional functionality it -- provides is the ability to wait for the return value of a thread, -- but the interface also provides some additional safety and--- robustness over using threads and @MVar@ directly.------ The basic type is @'Async' a@, which represents an asynchronous--- @IO@ action that will return a value of type @a@, or die with an--- exception. An @Async@ corresponds to a thread, and its 'ThreadId'--- can be obtained with 'asyncThreadId', although that should rarely--- be necessary.------ For example, to fetch two web pages at the same time, we could do--- this (assuming a suitable @getURL@ function):------ > do a1 <- async (getURL url1)--- > a2 <- async (getURL url2)--- > page1 <- wait a1--- > page2 <- wait a2--- > ...------ where 'async' starts the operation in a separate thread, and--- 'wait' waits for and returns the result. If the operation--- throws an exception, then that exception is re-thrown by--- 'wait'. This is one of the ways in which this library--- provides some additional safety: it is harder to accidentally--- forget about exceptions thrown in child threads.------ A slight improvement over the previous example is this:------ > withAsync (getURL url1) $ \a1 -> do--- > withAsync (getURL url2) $ \a2 -> do--- > page1 <- wait a1--- > page2 <- wait a2--- > ...------ 'withAsync' is like 'async', except that the 'Async' is--- automatically killed (using 'cancel') if the enclosing IO operation--- returns before it has completed. Consider the case when the first--- 'wait' throws an exception; then the second 'Async' will be--- automatically killed rather than being left to run in the--- background, possibly indefinitely. This is the second way that the--- library provides additional safety: using 'withAsync' means we can--- avoid accidentally leaving threads running. Furthermore,--- 'withAsync' allows a tree of threads to be built, such that--- children are automatically killed if their parents die for any--- reason.------ The pattern of performing two IO actions concurrently and waiting--- for their results is packaged up in a combinator 'concurrently', so--- we can further shorten the above example to:------ > (page1, page2) <- concurrently (getURL url1) (getURL url2)--- > ...------ The 'Functor' instance can be used to change the result of an--- 'Async'. For example:------ > ghci> a <- async (return 3)--- > ghci> wait a--- > 3--- > ghci> wait (fmap (+1) a)--- > 4---------------------------------------------------------------------------------module Control.Concurrent.Async (-- -- * Asynchronous actions- Async,- -- ** Spawning- async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask,-- -- ** Spawning with automatic 'cancel'ation- withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask, withAsyncOnWithUnmask,-- -- ** Querying 'Async's- wait, poll, waitCatch, cancel, cancelWith,- asyncThreadId,-- -- ** STM operations- waitSTM, pollSTM, waitCatchSTM,-- -- ** Waiting for multiple 'Async's- waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel,- waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel,- waitEither_,- waitBoth,-- -- ** Linking- link, link2,-- -- * Convenient utilities- race, race_, concurrently, mapConcurrently,- Concurrently(..),-- ) where--import Control.Concurrent.STM-import Control.Exception-import Control.Concurrent-#if !MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif-import Control.Monad-import Control.Applicative-import Data.Traversable--import GHC.Exts-import GHC.IO hiding (finally, onException)-import GHC.Conc---- -------------------------------------------------------------------------------- STM Async API----- | An asynchronous action spawned by 'async' or 'withAsync'.--- Asynchronous actions are executed in a separate thread, and--- operations are provided for waiting for asynchronous actions to--- complete and obtaining their results (see e.g. 'wait').----data Async a = Async { asyncThreadId :: {-# UNPACK #-} !ThreadId- -- ^ Returns the 'ThreadId' of the thread running the given 'Async'.- , _asyncWait :: STM (Either SomeException a) }--instance Eq (Async a) where- Async a _ == Async b _ = a == b--instance Ord (Async a) where- Async a _ `compare` Async b _ = a `compare` b--instance Functor Async where- fmap f (Async a w) = Async a (fmap (fmap f) w)---- | Spawn an asynchronous action in a separate thread.-async :: IO a -> IO (Async a)-async = inline asyncUsing rawForkIO---- | Like 'async' but using 'forkOS' internally.-asyncBound :: IO a -> IO (Async a)-asyncBound = asyncUsing forkOS---- | Like 'async' but using 'forkOn' internally.-asyncOn :: Int -> IO a -> IO (Async a)-asyncOn = asyncUsing . rawForkOn---- | Like 'async' but using 'forkIOWithUnmask' internally.--- The child thread is passed a function that can be used to unmask asynchronous exceptions.-asyncWithUnmask :: ((forall b . IO b -> IO b) -> IO a) -> IO (Async a)-asyncWithUnmask actionWith = asyncUsing rawForkIO (actionWith unsafeUnmask)---- | Like 'asyncOn' but using 'forkOnWithUnmask' internally.--- The child thread is passed a function that can be used to unmask asynchronous exceptions.-asyncOnWithUnmask :: Int -> ((forall b . IO b -> IO b) -> IO a) -> IO (Async a)-asyncOnWithUnmask cpu actionWith = asyncUsing (rawForkOn cpu) (actionWith unsafeUnmask)--asyncUsing :: (IO () -> IO ThreadId)- -> IO a -> IO (Async a)-asyncUsing doFork = \action -> do- var <- newEmptyTMVarIO- -- t <- forkFinally action (\r -> atomically $ putTMVar var r)- -- slightly faster:- t <- mask $ \restore ->- doFork $ try (restore action) >>= atomically . putTMVar var- return (Async t (readTMVar var))---- | Spawn an asynchronous action in a separate thread, and pass its--- @Async@ handle to the supplied function. When the function returns--- or throws an exception, 'cancel' is called on the @Async@.------ > withAsync action inner = bracket (async action) cancel inner------ This is a useful variant of 'async' that ensures an @Async@ is--- never left running unintentionally.------ Since 'cancel' may block, 'withAsync' may also block; see 'cancel'--- for details.----withAsync :: IO a -> (Async a -> IO b) -> IO b-withAsync = inline withAsyncUsing rawForkIO---- | Like 'withAsync' but uses 'forkOS' internally.-withAsyncBound :: IO a -> (Async a -> IO b) -> IO b-withAsyncBound = withAsyncUsing forkOS---- | Like 'withAsync' but uses 'forkOn' internally.-withAsyncOn :: Int -> IO a -> (Async a -> IO b) -> IO b-withAsyncOn = withAsyncUsing . rawForkOn---- | Like 'withAsync' but uses 'forkIOWithUnmask' internally.--- The child thread is passed a function that can be used to unmask asynchronous exceptions.-withAsyncWithUnmask :: ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b-withAsyncWithUnmask actionWith = withAsyncUsing rawForkIO (actionWith unsafeUnmask)---- | Like 'withAsyncOn' but uses 'forkOnWithUnmask' internally.--- The child thread is passed a function that can be used to unmask asynchronous exceptions-withAsyncOnWithUnmask :: Int -> ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b-withAsyncOnWithUnmask cpu actionWith = withAsyncUsing (rawForkOn cpu) (actionWith unsafeUnmask)--withAsyncUsing :: (IO () -> IO ThreadId)- -> IO a -> (Async a -> IO b) -> IO b--- The bracket version works, but is slow. We can do better by--- hand-coding it:-withAsyncUsing doFork = \action inner -> do- var <- newEmptyTMVarIO- mask $ \restore -> do- t <- doFork $ try (restore action) >>= atomically . putTMVar var- let a = Async t (readTMVar var)- r <- restore (inner a) `catchAll` \e -> do cancel a; throwIO e- cancel a- return r---- | Wait for an asynchronous action to complete, and return its--- value. If the asynchronous action threw an exception, then the--- exception is re-thrown by 'wait'.------ > wait = atomically . waitSTM+-- robustness over using 'forkIO' threads and @MVar@ directly. ---{-# INLINE wait #-}-wait :: Async a -> IO a-wait = atomically . waitSTM---- | Wait for an asynchronous action to complete, and return either--- @Left e@ if the action raised an exception @e@, or @Right a@ if it--- returned a value @a@.+-- == High-level API ----- > waitCatch = atomically . waitCatchSTM+-- @async@'s high-level API spawns /lexically scoped/ threads,+-- ensuring the following key properties that make it safer to use+-- than using plain 'forkIO': ---{-# INLINE waitCatch #-}-waitCatch :: Async a -> IO (Either SomeException a)-waitCatch = tryAgain . atomically . waitCatchSTM- where- -- See: https://github.com/simonmar/async/issues/14- tryAgain f = f `catch` \BlockedIndefinitelyOnSTM -> f---- | Check whether an 'Async' has completed yet. If it has not--- completed yet, then the result is @Nothing@, otherwise the result--- is @Just e@ where @e@ is @Left x@ if the @Async@ raised an--- exception @x@, or @Right a@ if it returned a value @a@.+-- 1. No exception is swallowed (waiting for results propagates exceptions).+-- 2. No thread is leaked (left running unintentionally). ----- > poll = atomically . pollSTM+-- (This is done using the 'Control.Exception.bracket' pattern to work in presence+-- of synchronous and asynchronous exceptions.) ---{-# INLINE poll #-}-poll :: Async a -> IO (Maybe (Either SomeException a))-poll = atomically . pollSTM---- | A version of 'wait' that can be used inside an STM transaction.+-- __Most practical/production code should only use the high-level API__. ---waitSTM :: Async a -> STM a-waitSTM a = do- r <- waitCatchSTM a- either throwSTM return r---- | A version of 'waitCatch' that can be used inside an STM transaction.+-- The basic type is @'Async' a@, which represents an asynchronous+-- @IO@ action that will return a value of type @a@, or die with an+-- exception. An 'Async' is a wrapper around a low-level 'forkIO' thread. ---{-# INLINE waitCatchSTM #-}-waitCatchSTM :: Async a -> STM (Either SomeException a)-waitCatchSTM (Async _ w) = w---- | A version of 'poll' that can be used inside an STM transaction.+-- The fundamental function to spawn threads with the high-level API is+-- 'withAsync'. ---{-# INLINE pollSTM #-}-pollSTM :: Async a -> STM (Maybe (Either SomeException a))-pollSTM (Async _ w) = (Just <$> w) `orElse` return Nothing---- | Cancel an asynchronous action by throwing the @ThreadKilled@--- exception to it. Has no effect if the 'Async' has already--- completed.+-- For example, to fetch two web pages at the same time, we could do+-- this (assuming a suitable @getURL@ function): ----- > cancel a = throwTo (asyncThreadId a) ThreadKilled+-- > withAsync (getURL url1) $ \a1 -> do+-- > withAsync (getURL url2) $ \a2 -> do+-- > page1 <- wait a1+-- > page2 <- wait a2+-- > ... ----- Note that 'cancel' is synchronous in the same sense as 'throwTo'.--- It does not return until the exception has been thrown in the--- target thread, or the target thread has completed. In particular,--- if the target thread is making a foreign call, the exception will--- not be thrown until the foreign call returns, and in this case--- 'cancel' may block indefinitely. An asynchronous 'cancel' can--- of course be obtained by wrapping 'cancel' itself in 'async'.+-- where 'withAsync' starts the operation in a separate thread, and+-- 'wait' waits for and returns the result. ---{-# INLINE cancel #-}-cancel :: Async a -> IO ()-cancel (Async t _) = throwTo t ThreadKilled---- | Cancel an asynchronous action by throwing the supplied exception--- to it.+-- * If the operation throws an exception, then that exception is re-thrown+-- by 'wait'. This ensures property (1): No exception is swallowed.+-- * If an exception bubbles up through a 'withAsync', then the 'Async'+-- it spawned is 'cancel'ed. This ensures property (2): No thread is leaked. ----- > cancelWith a x = throwTo (asyncThreadId a) x+-- Often we do not care to work manually with 'Async' handles like+-- @a1@ and @a2@. Instead, we want to express high-level objectives like+-- performing two or more tasks concurrently, and waiting for one or all+-- of them to finish. ----- The notes about the synchronous nature of 'cancel' also apply to--- 'cancelWith'.-cancelWith :: Exception e => Async a -> e -> IO ()-cancelWith (Async t _) e = throwTo t e---- | Wait for any of the supplied asynchronous operations to complete.--- The value returned is a pair of the 'Async' that completed, and the--- result that would be returned by 'wait' on that 'Async'.+-- For example, the pattern of performing two IO actions concurrently and+-- waiting for both their results is packaged up in a combinator 'concurrently',+-- so we can further shorten the above example to: ----- If multiple 'Async's complete or have completed, then the value--- returned corresponds to the first completed 'Async' in the list.+-- > (page1, page2) <- concurrently (getURL url1) (getURL url2)+-- > ... ---waitAnyCatch :: [Async a] -> IO (Async a, Either SomeException a)-waitAnyCatch asyncs =- atomically $- foldr orElse retry $- map (\a -> do r <- waitCatchSTM a; return (a, r)) asyncs---- | Like 'waitAnyCatch', but also cancels the other asynchronous--- operations as soon as one has completed.+-- The section __/High-level utilities/__ covers the most+-- common high-level objectives, including: ---waitAnyCatchCancel :: [Async a] -> IO (Async a, Either SomeException a)-waitAnyCatchCancel asyncs =- waitAnyCatch asyncs `finally` mapM_ cancel asyncs---- | Wait for any of the supplied @Async@s to complete. If the first--- to complete throws an exception, then that exception is re-thrown--- by 'waitAny'.+-- * Waiting for 2 results ('concurrently').+-- * Waiting for many results ('mapConcurrently' / 'forConcurrently').+-- * Waiting for the first of 2 results ('race').+-- * Waiting for arbitrary nestings of "all of /N/" and "the first of /N/"+-- results with the 'Concurrently' newtype and its 'Applicative' and+-- 'Alternative' instances. ----- If multiple 'Async's complete or have completed, then the value--- returned corresponds to the first completed 'Async' in the list.+-- Click here to scroll to that section:+-- "Control.Concurrent.Async#high-level-utilities". ---waitAny :: [Async a] -> IO (Async a, a)-waitAny asyncs =- atomically $- foldr orElse retry $- map (\a -> do r <- waitSTM a; return (a, r)) asyncs---- | Like 'waitAny', but also cancels the other asynchronous--- operations as soon as one has completed.+-- == Low-level API ---waitAnyCancel :: [Async a] -> IO (Async a, a)-waitAnyCancel asyncs =- waitAny asyncs `finally` mapM_ cancel asyncs---- | Wait for the first of two @Async@s to finish.-waitEitherCatch :: Async a -> Async b- -> IO (Either (Either SomeException a)- (Either SomeException b))-waitEitherCatch left right =- atomically $- (Left <$> waitCatchSTM left)- `orElse`- (Right <$> waitCatchSTM right)---- | Like 'waitEitherCatch', but also 'cancel's both @Async@s before--- returning.+-- Some use cases require parallelism that is not lexically scoped. ---waitEitherCatchCancel :: Async a -> Async b- -> IO (Either (Either SomeException a)- (Either SomeException b))-waitEitherCatchCancel left right =- waitEitherCatch left right `finally` (cancel left >> cancel right)---- | Wait for the first of two @Async@s to finish. If the @Async@--- that finished first raised an exception, then the exception is--- re-thrown by 'waitEither'.+-- For those, the low-level function 'async' can be used as a direct+-- equivalent of 'forkIO': ---waitEither :: Async a -> Async b -> IO (Either a b)-waitEither left right =- atomically $- (Left <$> waitSTM left)- `orElse`- (Right <$> waitSTM right)---- | Like 'waitEither', but the result is ignored.+-- > -- Do NOT use this code in production, it has a flaw (explained below).+-- > do+-- > a1 <- async (getURL url1)+-- > a2 <- async (getURL url2)+-- > page1 <- wait a1+-- > page2 <- wait a2+-- > ... ---waitEither_ :: Async a -> Async b -> IO ()-waitEither_ left right =- atomically $- (void $ waitSTM left)- `orElse`- (void $ waitSTM right)---- | Like 'waitEither', but also 'cancel's both @Async@s before--- returning.+-- In contrast to 'withAsync', this code has a problem. ---waitEitherCancel :: Async a -> Async b -> IO (Either a b)-waitEitherCancel left right =- waitEither left right `finally` (cancel left >> cancel right)---- | Waits for both @Async@s to finish, but if either of them throws--- an exception before they have both finished, then the exception is--- re-thrown by 'waitBoth'.+-- It still fulfills property (1) in that an exception arising from+-- @getUrl@ will be re-thrown by 'wait', but it does not fulfill+-- property (2).+-- Consider the case when the first 'wait' throws an exception; then the+-- second 'wait' will not happen, and the second 'async' may be left+-- running in the background, possibly indefinitely. ---waitBoth :: Async a -> Async b -> IO (a,b)-waitBoth left right =- atomically $ do- a <- waitSTM left- `orElse`- (waitSTM right >> retry)- b <- waitSTM right- return (a,b)----- | Link the given @Async@ to the current thread, such that if the--- @Async@ raises an exception, that exception will be re-thrown in--- the current thread.+-- 'withAsync' is like 'async', except that the 'Async' is+-- automatically killed (using 'uninterruptibleCancel') if the+-- enclosing IO operation returns before it has completed.+-- Furthermore, 'withAsync' allows a tree of threads to be built, such+-- that children are automatically killed if their parents die for any+-- reason. ---link :: Async a -> IO ()-link (Async _ w) = do- me <- myThreadId- void $ forkRepeat $ do- r <- atomically $ w- case r of- Left e -> throwTo me e- _ -> return ()---- | Link two @Async@s together, such that if either raises an--- exception, the same exception is re-thrown in the other @Async@.+-- If you need to use the low-level API, ensure that you guarantee+-- property (2) by other means, such as 'link'ing asyncs that need+-- to die together, and protecting against asynchronous exceptions+-- using 'Control.Exception.bracket', 'Control.Exception.mask',+-- or other functions from "Control.Exception". ---link2 :: Async a -> Async b -> IO ()-link2 left@(Async tl _) right@(Async tr _) =- void $ forkRepeat $ do- r <- waitEitherCatch left right- case r of- Left (Left e) -> throwTo tr e- Right (Left e) -> throwTo tl e- _ -> return ()----- --------------------------------------------------------------------------------- | Run two @IO@ actions concurrently, and return the first to--- finish. The loser of the race is 'cancel'led.+-- == Miscellaneous ----- > race left right =--- > withAsync left $ \a ->--- > withAsync right $ \b ->--- > waitEither a b+-- The 'Functor' instance can be used to change the result of an+-- 'Async'. For example: ---race :: IO a -> IO b -> IO (Either a b)---- | Like 'race', but the result is ignored.+-- > ghci> withAsync (return 3) (\a -> wait (fmap (+1) a))+-- > 4 ---race_ :: IO a -> IO b -> IO ()---- | Run two @IO@ actions concurrently, and return both results. If--- either action throws an exception at any time, then the other--- action is 'cancel'led, and the exception is re-thrown by--- 'concurrently'.+-- === Resource exhaustion ----- > concurrently left right =--- > withAsync left $ \a ->--- > withAsync right $ \b ->--- > waitBoth a b-concurrently :: IO a -> IO b -> IO (a,b)--#define USE_ASYNC_VERSIONS 0--#if USE_ASYNC_VERSIONS--race left right =- withAsync left $ \a ->- withAsync right $ \b ->- waitEither a b--race_ left right =- withAsync left $ \a ->- withAsync right $ \b ->- waitEither_ a b--concurrently left right =- withAsync left $ \a ->- withAsync right $ \b ->- waitBoth a b--#else---- MVar versions of race/concurrently--- More ugly than the Async versions, but quite a bit faster.---- race :: IO a -> IO b -> IO (Either a b)-race left right = concurrently' left right collect- where- collect m = do- e <- takeMVar m- case e of- Left ex -> throwIO ex- Right r -> return r---- race_ :: IO a -> IO b -> IO ()-race_ left right = void $ race left right+-- As with all concurrent programming, keep in mind that while+-- Haskell's cooperative ("green") multithreading carries low overhead,+-- spawning too many of them at the same time may lead to resource exhaustion+-- (of memory, file descriptors, or other limited resources), given that the+-- actions running in the threads consume these resources. --- concurrently :: IO a -> IO b -> IO (a,b)-concurrently left right = concurrently' left right (collect [])- where- collect [Left a, Right b] _ = return (a,b)- collect [Right b, Left a] _ = return (a,b)- collect xs m = do- e <- takeMVar m- case e of- Left ex -> throwIO ex- Right r -> collect (r:xs) m+----------------------------------------------------------------------------- -concurrently' :: IO a -> IO b- -> (MVar (Either SomeException (Either a b)) -> IO r)- -> IO r-concurrently' left right collect = do- done <- newEmptyMVar- mask $ \restore -> do- lid <- forkIO $ restore (left >>= putMVar done . Right . Left)- `catchAll` (putMVar done . Left)- rid <- forkIO $ restore (right >>= putMVar done . Right . Right)- `catchAll` (putMVar done . Left)- let stop = killThread lid >> killThread rid- r <- restore (collect done) `onException` stop- stop- return r+module Control.Concurrent.Async ( -#endif+ -- * Asynchronous actions+ Async, --- | maps an @IO@-performing function over any @Traversable@ data--- type, performing all the @IO@ actions concurrently, and returning--- the original data structure with the arguments replaced by the--- results.------ For example, @mapConcurrently@ works with lists:------ > pages <- mapConcurrently getURL ["url1", "url2", "url3"]----mapConcurrently :: Traversable t => (a -> IO b) -> t a -> IO (t b)-mapConcurrently f = runConcurrently . traverse (Concurrently . f)+ -- * High-level API --- -----------------------------------------------------------------------------+ -- ** Spawning with automatic 'cancel'ation+ withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask,+ withAsyncOnWithUnmask, --- | A value of type @Concurrently a@ is an @IO@ operation that can be--- composed with other @Concurrently@ values, using the @Applicative@--- and @Alternative@ instances.------ Calling @runConcurrently@ on a value of type @Concurrently a@ will--- execute the @IO@ operations it contains concurrently, before--- delivering the result of type @a@.------ For example------ > (page1, page2, page3)--- > <- runConcurrently $ (,,)--- > <$> Concurrently (getURL "url1")--- > <*> Concurrently (getURL "url2")--- > <*> Concurrently (getURL "url3")----newtype Concurrently a = Concurrently { runConcurrently :: IO a }+ -- ** Querying 'Async's+ wait, poll, waitCatch, asyncThreadId,+ cancel, cancelMany, uninterruptibleCancel, cancelWith, AsyncCancelled(..), -instance Functor Concurrently where- fmap f (Concurrently a) = Concurrently $ f <$> a+ -- ** #high-level-utilities# High-level utilities+ race, race_,+ concurrently, concurrently_,+ mapConcurrently, forConcurrently,+ mapConcurrently_, forConcurrently_,+ replicateConcurrently, replicateConcurrently_,+ Concurrently(..),+ concurrentlyE,+ ConcurrentlyE(..),+ compareAsyncs, -instance Applicative Concurrently where- pure = Concurrently . return- Concurrently fs <*> Concurrently as =- Concurrently $ (\(f, a) -> f a) <$> concurrently fs as+ -- ** Specialised operations -instance Alternative Concurrently where- empty = Concurrently $ forever (threadDelay maxBound)- Concurrently as <|> Concurrently bs =- Concurrently $ either id id <$> race as bs+ -- *** STM operations+ waitSTM, pollSTM, waitCatchSTM, -instance Monad Concurrently where- return = pure- Concurrently a >>= f =- Concurrently $ a >>= runConcurrently . f+ -- *** Waiting for multiple 'Async's+ waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel,+ waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel,+ waitEither_,+ waitBoth, --- ----------------------------------------------------------------------------+ -- *** Waiting for multiple 'Async's in STM+ waitAnySTM, waitAnyCatchSTM,+ waitEitherSTM, waitEitherCatchSTM,+ waitEitherSTM_,+ waitBothSTM, --- | Fork a thread that runs the supplied action, and if it raises an--- exception, re-runs the action. The thread terminates only when the--- action runs to completion without raising an exception.-forkRepeat :: IO a -> IO ThreadId-forkRepeat action =- mask $ \restore ->- let go = do r <- tryAll (restore action)- case r of- Left _ -> go- _ -> return ()- in forkIO go+ -- * Low-level API -catchAll :: IO a -> (SomeException -> IO a) -> IO a-catchAll = catch+ -- ** Spawning (low-level API)+ async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask, -tryAll :: IO a -> IO (Either SomeException a)-tryAll = try+ -- ** Linking+ link, linkOnly, link2, link2Only, ExceptionInLinkedThread(..), --- A version of forkIO that does not include the outer exception--- handler: saves a bit of time when we will be installing our own--- exception handler.-{-# INLINE rawForkIO #-}-rawForkIO :: IO () -> IO ThreadId-rawForkIO action = IO $ \ s ->- case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)+ ) where -{-# INLINE rawForkOn #-}-rawForkOn :: Int -> IO () -> IO ThreadId-rawForkOn (I# cpu) action = IO $ \ s ->- case (forkOn# cpu action s) of (# s1, tid #) -> (# s1, ThreadId tid #)+import Control.Concurrent.Async.Internal
+ Control/Concurrent/Async/Internal.hs view
@@ -0,0 +1,964 @@+{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes,+ ExistentialQuantification #-}+#if __GLASGOW_HASKELL__ >= 701+{-# LANGUAGE Trustworthy #-}+#endif+#if __GLASGOW_HASKELL__ < 710+{-# LANGUAGE DeriveDataTypeable #-}+#endif+{-# OPTIONS -Wall #-}++-----------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.Async.Internal+-- Copyright : (c) Simon Marlow 2012+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Simon Marlow <marlowsd@gmail.com>+-- Stability : provisional+-- Portability : non-portable (requires concurrency)+--+-- This module is an internal module. The public API is provided in+-- "Control.Concurrent.Async". Breaking changes to this module will not be+-- reflected in a major bump, and using this module may break your code+-- unless you are extremely careful.+--+-----------------------------------------------------------------------------++module Control.Concurrent.Async.Internal where++import Control.Concurrent.STM+import Control.Exception+import Control.Concurrent+import qualified Data.Foldable as F+#if !MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif+import Control.Monad+import Control.Applicative+#if !MIN_VERSION_base(4,8,0)+import Data.Monoid (Monoid(mempty,mappend))+import Data.Traversable+#endif+#if __GLASGOW_HASKELL__ < 710+import Data.Typeable+#endif+#if MIN_VERSION_base(4,8,0)+import Data.Bifunctor+#endif+#if MIN_VERSION_base(4,9,0)+import Data.Semigroup (Semigroup((<>)))+#endif+import Data.Hashable (Hashable(hashWithSalt))++import Data.IORef++import GHC.Exts+import GHC.IO hiding (finally, onException)+import GHC.Conc (ThreadId(..))++#if defined(__MHS__)+import Data.Traversable+#endif++#ifdef DEBUG_AUTO_LABEL+import qualified GHC.Stack+#endif++#ifdef DEBUG_AUTO_LABEL+#define CALLSTACK GHC.Stack.HasCallStack =>+#else+#define CALLSTACK+#endif++-- -----------------------------------------------------------------------------+-- STM Async API+++-- | An asynchronous action spawned by 'async' or 'withAsync'.+-- Asynchronous actions are executed in a separate thread, and+-- operations are provided for waiting for asynchronous actions to+-- complete and obtaining their results (see e.g. 'wait').+--+data Async a = Async+ { asyncThreadId :: {-# UNPACK #-} !ThreadId+ -- ^ Returns the 'ThreadId' of the thread running+ -- the given 'Async'.+ , _asyncWait :: STM (Either SomeException a)+ }++instance Eq (Async a) where+ Async a _ == Async b _ = a == b++instance Ord (Async a) where+ Async a _ `compare` Async b _ = a `compare` b++instance Hashable (Async a) where+ hashWithSalt salt (Async a _) = hashWithSalt salt a++instance Functor Async where+ fmap f (Async a w) = Async a (fmap (fmap f) w)++-- | Compare two Asyncs that may have different types by their 'ThreadId'.+compareAsyncs :: Async a -> Async b -> Ordering+compareAsyncs (Async t1 _) (Async t2 _) = compare t1 t2++-- | Spawn an asynchronous action in a separate thread.+--+-- Like for 'forkIO', the action may be left running unintentionally+-- (see module-level documentation for details).+--+-- __Use 'withAsync' style functions wherever you can instead!__+async ::+ CALLSTACK+ IO a -> IO (Async a)+async = inline asyncUsing rawForkIO++-- | Like 'async' but using 'forkOS' internally.+asyncBound ::+ CALLSTACK+ IO a -> IO (Async a)+asyncBound = asyncUsing forkOS++-- | Like 'async' but using 'forkOn' internally.+asyncOn ::+ CALLSTACK+ Int -> IO a -> IO (Async a)+asyncOn = asyncUsing . rawForkOn++-- | Like 'async' but using 'forkIOWithUnmask' internally. The child+-- thread is passed a function that can be used to unmask asynchronous+-- exceptions.+asyncWithUnmask ::+ CALLSTACK+ ((forall b . IO b -> IO b) -> IO a) -> IO (Async a)+asyncWithUnmask actionWith = asyncUsing rawForkIO (actionWith unsafeUnmask)++-- | Like 'asyncOn' but using 'forkOnWithUnmask' internally. The+-- child thread is passed a function that can be used to unmask+-- asynchronous exceptions.+asyncOnWithUnmask ::+ CALLSTACK+ Int -> ((forall b . IO b -> IO b) -> IO a) -> IO (Async a)+asyncOnWithUnmask cpu actionWith =+ asyncUsing (rawForkOn cpu) (actionWith unsafeUnmask)++asyncUsing ::+ CALLSTACK+ (IO () -> IO ThreadId) -> IO a -> IO (Async a)+asyncUsing doFork action = do+ var <- newEmptyTMVarIO+ let action_plus = debugLabelMe >> action+ -- t <- forkFinally action (\r -> atomically $ putTMVar var r)+ -- slightly faster:+ t <- mask $ \restore ->+ doFork $ try (restore action_plus) >>= atomically . putTMVar var+ return (Async t (readTMVar var))+++-- | Spawn an asynchronous action in a separate thread, and pass its+-- @Async@ handle to the supplied function. When the function returns+-- or throws an exception, 'uninterruptibleCancel' is called on the @Async@.+--+-- > withAsync action inner = mask $ \restore -> do+-- > a <- async (restore action)+-- > restore (inner a) `finally` uninterruptibleCancel a+--+-- This is a useful variant of 'async' that ensures an @Async@ is+-- never left running unintentionally.+--+-- Note: a reference to the child thread is kept alive until the call+-- to `withAsync` returns, so nesting many `withAsync` calls requires+-- linear memory.+--+withAsync ::+ CALLSTACK+ IO a -> (Async a -> IO b) -> IO b+withAsync = inline withAsyncUsing rawForkIO++-- | Like 'withAsync' but uses 'forkOS' internally.+withAsyncBound ::+ CALLSTACK+ IO a -> (Async a -> IO b) -> IO b+withAsyncBound = withAsyncUsing forkOS++-- | Like 'withAsync' but uses 'forkOn' internally.+withAsyncOn ::+ CALLSTACK+ Int -> IO a -> (Async a -> IO b) -> IO b+withAsyncOn = withAsyncUsing . rawForkOn++-- | Like 'withAsync' but uses 'forkIOWithUnmask' internally. The+-- child thread is passed a function that can be used to unmask+-- asynchronous exceptions.+withAsyncWithUnmask ::+ CALLSTACK+ ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b+withAsyncWithUnmask actionWith =+ withAsyncUsing rawForkIO (actionWith unsafeUnmask)++-- | Like 'withAsyncOn' but uses 'forkOnWithUnmask' internally. The+-- child thread is passed a function that can be used to unmask+-- asynchronous exceptions+withAsyncOnWithUnmask ::+ CALLSTACK+ Int -> ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b+withAsyncOnWithUnmask cpu actionWith =+ withAsyncUsing (rawForkOn cpu) (actionWith unsafeUnmask)++withAsyncUsing ::+ CALLSTACK+ (IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b+-- The bracket version works, but is slow. We can do better by+-- hand-coding it:+withAsyncUsing doFork action inner = do+ var <- newEmptyTMVarIO+ mask $ \restore -> do+ let action_plus = debugLabelMe >> action+ t <- doFork $ try (restore action_plus) >>= atomically . putTMVar var+ let a = Async t (readTMVar var)+ r <- restore (inner a) `catchAll` \e -> do+ uninterruptibleCancel a+ throwIO e+ uninterruptibleCancel a+ return r++-- | Wait for an asynchronous action to complete, and return its+-- value. If the asynchronous action threw an exception, then the+-- exception is re-thrown by 'wait'.+--+-- > wait = atomically . waitSTM+--+{-# INLINE wait #-}+wait :: Async a -> IO a+wait = tryAgain . atomically . waitSTM+ where+ -- See: https://github.com/simonmar/async/issues/14+ tryAgain f = f `catch` \BlockedIndefinitelyOnSTM -> f++-- | Wait for an asynchronous action to complete, and return either+-- @Left e@ if the action raised an exception @e@, or @Right a@ if it+-- returned a value @a@.+--+-- > waitCatch = atomically . waitCatchSTM+--+{-# INLINE waitCatch #-}+waitCatch :: Async a -> IO (Either SomeException a)+waitCatch = tryAgain . atomically . waitCatchSTM+ where+ -- See: https://github.com/simonmar/async/issues/14+ tryAgain f = f `catch` \BlockedIndefinitelyOnSTM -> f++-- | Check whether an 'Async' has completed yet. If it has not+-- completed yet, then the result is @Nothing@, otherwise the result+-- is @Just e@ where @e@ is @Left x@ if the @Async@ raised an+-- exception @x@, or @Right a@ if it returned a value @a@.+--+-- > poll = atomically . pollSTM+--+{-# INLINE poll #-}+poll :: Async a -> IO (Maybe (Either SomeException a))+poll = atomically . pollSTM++-- | A version of 'wait' that can be used inside an STM transaction.+--+waitSTM :: Async a -> STM a+waitSTM a = do+ r <- waitCatchSTM a+ either throwSTM return r++-- | A version of 'waitCatch' that can be used inside an STM transaction.+--+{-# INLINE waitCatchSTM #-}+waitCatchSTM :: Async a -> STM (Either SomeException a)+waitCatchSTM (Async _ w) = w++-- | A version of 'poll' that can be used inside an STM transaction.+--+{-# INLINE pollSTM #-}+pollSTM :: Async a -> STM (Maybe (Either SomeException a))+pollSTM (Async _ w) = (Just <$> w) `orElse` return Nothing++-- | Cancel an asynchronous action by throwing the @AsyncCancelled@+-- exception to it, and waiting for the `Async` thread to quit.+-- Has no effect if the 'Async' has already completed.+--+-- > cancel a = throwTo (asyncThreadId a) AsyncCancelled <* waitCatch a+--+-- Note that 'cancel' will not terminate until the thread the 'Async'+-- refers to has terminated. This means that 'cancel' will block for+-- as long said thread blocks when receiving an asynchronous exception.+--+-- For example, it could block if:+--+-- * It's executing a foreign call, and thus cannot receive the asynchronous+-- exception;+-- * It's executing some cleanup handler after having received the exception,+-- and the handler is blocking.+{-# INLINE cancel #-}+cancel :: Async a -> IO ()+cancel a@(Async t _) = throwTo t AsyncCancelled <* waitCatch a++-- | Cancel multiple asynchronous actions by throwing the @AsyncCancelled@+-- exception to each of them in turn, then waiting for all the `Async` threads+-- to complete.+--+-- @since 2.2.5+cancelMany :: [Async a] -> IO ()+cancelMany as = do+ mapM_ (\(Async t _) -> throwTo t AsyncCancelled) as+ mapM_ waitCatch as++-- | The exception thrown by `cancel` to terminate a thread.+data AsyncCancelled = AsyncCancelled+ deriving (Show, Eq+#if __GLASGOW_HASKELL__ < 710+ ,Typeable+#endif+ )++instance Exception AsyncCancelled where+#if __GLASGOW_HASKELL__ >= 708+ fromException = asyncExceptionFromException+ toException = asyncExceptionToException+#endif++-- | Cancel an asynchronous action+--+-- This is a variant of `cancel`, but it is not interruptible.+{-# INLINE uninterruptibleCancel #-}+uninterruptibleCancel :: Async a -> IO ()+uninterruptibleCancel = uninterruptibleMask_ . cancel++-- | Cancel an asynchronous action by throwing the supplied exception+-- to it.+--+-- > cancelWith a x = throwTo (asyncThreadId a) x+--+-- The notes about the synchronous nature of 'cancel' also apply to+-- 'cancelWith'.+cancelWith :: Exception e => Async a -> e -> IO ()+cancelWith a@(Async t _) e = throwTo t e <* waitCatch a++-- | Wait for any of the supplied asynchronous operations to complete.+-- The value returned is a pair of the 'Async' that completed, and the+-- result that would be returned by 'wait' on that 'Async'.+-- The input list must be non-empty.+--+-- If multiple 'Async's complete or have completed, then the value+-- returned corresponds to the first completed 'Async' in the list.+--+{-# INLINE waitAnyCatch #-}+waitAnyCatch :: [Async a] -> IO (Async a, Either SomeException a)+waitAnyCatch = atomically . waitAnyCatchSTM++-- | A version of 'waitAnyCatch' that can be used inside an STM transaction.+--+-- @since 2.1.0+waitAnyCatchSTM :: [Async a] -> STM (Async a, Either SomeException a)+waitAnyCatchSTM [] =+ throwSTM $ ErrorCall+ "waitAnyCatchSTM: invalid argument: input list must be non-empty"+waitAnyCatchSTM asyncs =+ foldr orElse retry $+ map (\a -> do r <- waitCatchSTM a; return (a, r)) asyncs++-- | Like 'waitAnyCatch', but also cancels the other asynchronous+-- operations as soon as one has completed.+--+waitAnyCatchCancel :: [Async a] -> IO (Async a, Either SomeException a)+waitAnyCatchCancel asyncs =+ waitAnyCatch asyncs `finally` cancelMany asyncs++-- | Wait for any of the supplied @Async@s to complete. If the first+-- to complete throws an exception, then that exception is re-thrown+-- by 'waitAny'.+-- The input list must be non-empty.+--+-- If multiple 'Async's complete or have completed, then the value+-- returned corresponds to the first completed 'Async' in the list.+--+{-# INLINE waitAny #-}+waitAny :: [Async a] -> IO (Async a, a)+waitAny = atomically . waitAnySTM++-- | A version of 'waitAny' that can be used inside an STM transaction.+--+-- @since 2.1.0+waitAnySTM :: [Async a] -> STM (Async a, a)+waitAnySTM [] =+ throwSTM $ ErrorCall+ "waitAnySTM: invalid argument: input list must be non-empty"+waitAnySTM asyncs =+ foldr orElse retry $+ map (\a -> do r <- waitSTM a; return (a, r)) asyncs++-- | Like 'waitAny', but also cancels the other asynchronous+-- operations as soon as one has completed.+--+waitAnyCancel :: [Async a] -> IO (Async a, a)+waitAnyCancel asyncs =+ waitAny asyncs `finally` cancelMany asyncs++-- | Wait for the first of two @Async@s to finish.+{-# INLINE waitEitherCatch #-}+waitEitherCatch :: Async a -> Async b+ -> IO (Either (Either SomeException a)+ (Either SomeException b))+waitEitherCatch left right =+ tryAgain $ atomically (waitEitherCatchSTM left right)+ where+ -- See: https://github.com/simonmar/async/issues/14+ tryAgain f = f `catch` \BlockedIndefinitelyOnSTM -> f++-- | A version of 'waitEitherCatch' that can be used inside an STM transaction.+--+-- @since 2.1.0+waitEitherCatchSTM :: Async a -> Async b+ -> STM (Either (Either SomeException a)+ (Either SomeException b))+waitEitherCatchSTM left right =+ (Left <$> waitCatchSTM left)+ `orElse`+ (Right <$> waitCatchSTM right)++-- | Like 'waitEitherCatch', but also 'cancel's both @Async@s before+-- returning.+--+waitEitherCatchCancel :: Async a -> Async b+ -> IO (Either (Either SomeException a)+ (Either SomeException b))+waitEitherCatchCancel left right =+ waitEitherCatch left right `finally` cancelMany [() <$ left, () <$ right]++-- | Wait for the first of two @Async@s to finish. If the @Async@+-- that finished first raised an exception, then the exception is+-- re-thrown by 'waitEither'.+--+{-# INLINE waitEither #-}+waitEither :: Async a -> Async b -> IO (Either a b)+waitEither left right = atomically (waitEitherSTM left right)++-- | A version of 'waitEither' that can be used inside an STM transaction.+--+-- @since 2.1.0+waitEitherSTM :: Async a -> Async b -> STM (Either a b)+waitEitherSTM left right =+ (Left <$> waitSTM left)+ `orElse`+ (Right <$> waitSTM right)++-- | Like 'waitEither', but the result is ignored.+--+{-# INLINE waitEither_ #-}+waitEither_ :: Async a -> Async b -> IO ()+waitEither_ left right = atomically (waitEitherSTM_ left right)++-- | A version of 'waitEither_' that can be used inside an STM transaction.+--+-- @since 2.1.0+waitEitherSTM_:: Async a -> Async b -> STM ()+waitEitherSTM_ left right =+ (void $ waitSTM left)+ `orElse`+ (void $ waitSTM right)++-- | Like 'waitEither', but also 'cancel's both @Async@s before+-- returning.+--+waitEitherCancel :: Async a -> Async b -> IO (Either a b)+waitEitherCancel left right =+ waitEither left right `finally` cancelMany [() <$ left, () <$ right]++-- | Waits for both @Async@s to finish, but if either of them throws+-- an exception before they have both finished, then the exception is+-- re-thrown by 'waitBoth'.+--+{-# INLINE waitBoth #-}+waitBoth :: Async a -> Async b -> IO (a,b)+waitBoth left right = tryAgain $ atomically (waitBothSTM left right)+ where+ -- See: https://github.com/simonmar/async/issues/14+ tryAgain f = f `catch` \BlockedIndefinitelyOnSTM -> f++-- | A version of 'waitBoth' that can be used inside an STM transaction.+--+-- @since 2.1.0+waitBothSTM :: Async a -> Async b -> STM (a,b)+waitBothSTM left right = do+ a <- waitSTM left+ `orElse`+ (waitSTM right >> retry)+ b <- waitSTM right+ return (a,b)+++-- -----------------------------------------------------------------------------+-- Linking threads++data ExceptionInLinkedThread =+ forall a . ExceptionInLinkedThread (Async a) SomeException+#if __GLASGOW_HASKELL__ < 710+ deriving Typeable+#endif++instance Show ExceptionInLinkedThread where+ showsPrec p (ExceptionInLinkedThread (Async t _) e) =+ showParen (p >= 11) $+ showString "ExceptionInLinkedThread " .+ showsPrec 11 t .+ showString " " .+ showsPrec 11 e++instance Exception ExceptionInLinkedThread where+#if __GLASGOW_HASKELL__ >= 708+ fromException = asyncExceptionFromException+ toException = asyncExceptionToException+#endif++-- | Link the given @Async@ to the current thread, such that if the+-- @Async@ raises an exception, that exception will be re-thrown in+-- the current thread, wrapped in 'ExceptionInLinkedThread'.+--+-- 'link' ignores 'AsyncCancelled' exceptions thrown in the other thread,+-- so that it's safe to 'cancel' a thread you're linked to. If you want+-- different behaviour, use 'linkOnly'.+--+link :: Async a -> IO ()+link = linkOnly (not . isCancel)++-- | Link the given @Async@ to the current thread, such that if the+-- @Async@ raises an exception, that exception will be re-thrown in+-- the current thread, wrapped in 'ExceptionInLinkedThread'.+--+-- The supplied predicate determines which exceptions in the target+-- thread should be propagated to the source thread.+--+linkOnly+ :: (SomeException -> Bool) -- ^ return 'True' if the exception+ -- should be propagated, 'False'+ -- otherwise.+ -> Async a+ -> IO ()+linkOnly shouldThrow a = do+ me <- myThreadId+ void $ forkRepeat $ do+ r <- waitCatch a+ case r of+ Left e | shouldThrow e -> throwTo me (ExceptionInLinkedThread a e)+ _otherwise -> return ()++-- | Link two @Async@s together, such that if either raises an+-- exception, the same exception is re-thrown in the other @Async@,+-- wrapped in 'ExceptionInLinkedThread'.+--+-- 'link2' ignores 'AsyncCancelled' exceptions, so that it's possible+-- to 'cancel' either thread without cancelling the other. If you+-- want different behaviour, use 'link2Only'.+--+link2 :: Async a -> Async b -> IO ()+link2 = link2Only (not . isCancel)++-- | Link two @Async@s together, such that if either raises an+-- exception, the same exception is re-thrown in the other @Async@,+-- wrapped in 'ExceptionInLinkedThread'.+--+-- The supplied predicate determines which exceptions in the target+-- thread should be propagated to the source thread.+--+link2Only :: (SomeException -> Bool) -> Async a -> Async b -> IO ()+link2Only shouldThrow left@(Async tl _) right@(Async tr _) =+ void $ forkRepeat $ do+ r <- waitEitherCatch left right+ case r of+ Left (Left e) | shouldThrow e ->+ throwTo tr (ExceptionInLinkedThread left e)+ Right (Left e) | shouldThrow e ->+ throwTo tl (ExceptionInLinkedThread right e)+ _ -> return ()++isCancel :: SomeException -> Bool+isCancel e+ | Just AsyncCancelled <- fromException e = True+ | otherwise = False+++-- -----------------------------------------------------------------------------++-- | Run two @IO@ actions concurrently, and return the first to+-- finish. The loser of the race is 'cancel'led.+--+-- > race left right =+-- > withAsync left $ \a ->+-- > withAsync right $ \b ->+-- > waitEither a b+--+race ::+ CALLSTACK+ IO a -> IO b -> IO (Either a b)++-- | Like 'race', but the result is ignored.+--+race_ ::+ CALLSTACK+ IO a -> IO b -> IO ()+++-- | Run two @IO@ actions concurrently, and return both results. If+-- either action throws an exception at any time, then the other+-- action is 'cancel'led, and the exception is re-thrown by+-- 'concurrently'.+--+-- > concurrently left right =+-- > withAsync left $ \a ->+-- > withAsync right $ \b ->+-- > waitBoth a b+--+-- To run more than two actions concurrently, see 'mapConcurrently'.+concurrently ::+ CALLSTACK+ IO a -> IO b -> IO (a,b)+++-- | Run two @IO@ actions concurrently. If both of them end with @Right@,+-- return both results. If one of then ends with @Left@, interrupt the other+-- action and return the @Left@.+--+concurrentlyE ::+ CALLSTACK+ IO (Either e a) -> IO (Either e b) -> IO (Either e (a, b))++-- | 'concurrently', but ignore the result values+--+-- @since 2.1.1+concurrently_ ::+ CALLSTACK+ IO a -> IO b -> IO ()++#define USE_ASYNC_VERSIONS 0++#if USE_ASYNC_VERSIONS++race left right =+ withAsync left $ \a ->+ withAsync right $ \b ->+ waitEither a b++race_ left right = void $ race left right++concurrently left right =+ withAsync left $ \a ->+ withAsync right $ \b ->+ waitBoth a b++concurrently_ left right = void $ concurrently left right++#else++-- MVar versions of race/concurrently+-- More ugly than the Async versions, but quite a bit faster.++-- race :: IO a -> IO b -> IO (Either a b)+race left right = concurrently' left right collect+ where+ collect m = do+ e <- m+ case e of+ Left ex -> throwIO ex+ Right r -> return r++-- race_ :: IO a -> IO b -> IO ()+race_ left right = void $ race left right++-- concurrently :: IO a -> IO b -> IO (a,b)+concurrently left right = concurrently' left right (collect [])+ where+ collect [Left a, Right b] _ = return (a,b)+ collect [Right b, Left a] _ = return (a,b)+ collect xs m = do+ e <- m+ case e of+ Left ex -> throwIO ex+ Right r -> collect (r:xs) m++-- concurrentlyE :: IO (Either e a) -> IO (Either e b) -> IO (Either e (a, b))+concurrentlyE left right = concurrently' left right (collect [])+ where+ collect [Left (Right a), Right (Right b)] _ = return $ Right (a,b)+ collect [Right (Right b), Left (Right a)] _ = return $ Right (a,b)+ collect (Left (Left ea):_) _ = return $ Left ea+ collect (Right (Left eb):_) _ = return $ Left eb+ collect xs m = do+ e <- m+ case e of+ Left ex -> throwIO ex+ Right r -> collect (r:xs) m++concurrently' ::+ CALLSTACK+ IO a -> IO b+ -> (IO (Either SomeException (Either a b)) -> IO r)+ -> IO r+concurrently' left right collect = do+ done <- newEmptyMVar+ mask $ \restore -> do+ -- Note: uninterruptibleMask here is because we must not allow+ -- the putMVar in the exception handler to be interrupted,+ -- otherwise the parent thread will deadlock when it waits for+ -- the thread to terminate.+ lid <- forkIO $ uninterruptibleMask_ $+ restore (left >>= putMVar done . Right . Left)+ `catchAll` (putMVar done . Left)+ rid <- forkIO $ uninterruptibleMask_ $+ restore (right >>= putMVar done . Right . Right)+ `catchAll` (putMVar done . Left)++ count <- newIORef (2 :: Int)+ let takeDone = do+ r <- takeMVar done -- interruptible+ -- Decrement the counter so we know how many takes are left.+ -- Since only the parent thread is calling this, we can+ -- use non-atomic modifications.+ -- NB. do this *after* takeMVar, because takeMVar might be+ -- interrupted.+ modifyIORef count (subtract 1)+ return r++ let tryAgain f = f `catch` \BlockedIndefinitelyOnMVar -> f++ stop = do+ -- kill right before left, to match the semantics of+ -- the version using withAsync. (#27)+ uninterruptibleMask_ $ do+ count' <- readIORef count+ -- we only need to use killThread if there are still+ -- children alive. Note: forkIO here is because the+ -- child thread could be in an uninterruptible+ -- putMVar.+ when (count' > 0) $+ void $ forkIO $ do+ throwTo rid AsyncCancelled+ throwTo lid AsyncCancelled+ -- ensure the children are really dead+ replicateM_ count' (tryAgain $ takeMVar done)++ r <- collect (tryAgain takeDone) `onException` stop+ stop+ return r++concurrently_ left right = concurrently' left right (collect 0)+ where+ collect 2 _ = return ()+ collect i m = do+ e <- m+ case e of+ Left ex -> throwIO ex+ Right _ -> collect (i + 1 :: Int) m+++#endif++-- | Maps an 'IO'-performing function over any 'Traversable' data+-- type, performing all the @IO@ actions concurrently, and returning+-- the original data structure with the arguments replaced by the+-- results.+--+-- If any of the actions throw an exception, then all other actions are+-- cancelled and the exception is re-thrown.+--+-- For example, @mapConcurrently@ works with lists:+--+-- > pages <- mapConcurrently getURL ["url1", "url2", "url3"]+--+-- If you just have a list of actions, run them concurrently with+--+-- > results <- mapConcurrently id [act1, act2, act3]+--+-- NOTE: @mapConcurrently@ will immediately spawn a thread for each+-- element of the @Traversable@, so running this on large inputs can+-- lead to resource exhaustion (of memory, file descriptors, or other+-- limited resources). To avoid unbounded resource usage, see+-- "Control.Concurrent.Stream".+mapConcurrently ::+ CALLSTACK+ Traversable t => (a -> IO b) -> t a -> IO (t b)+mapConcurrently f = runConcurrently . traverse (Concurrently . f)++-- | `forConcurrently` is `mapConcurrently` with its arguments flipped+--+-- > pages <- forConcurrently ["url1", "url2", "url3"] $ \url -> getURL url+--+-- @since 2.1.0+forConcurrently ::+ CALLSTACK+ Traversable t => t a -> (a -> IO b) -> IO (t b)+forConcurrently = flip mapConcurrently++-- | `mapConcurrently_` is `mapConcurrently` with the return value discarded;+-- a concurrent equivalent of 'mapM_'.+mapConcurrently_ ::+ CALLSTACK+ F.Foldable f => (a -> IO b) -> f a -> IO ()+mapConcurrently_ f = runConcurrently . F.foldMap (Concurrently . void . f)++-- | `forConcurrently_` is `forConcurrently` with the return value discarded;+-- a concurrent equivalent of 'forM_'.+forConcurrently_ ::+ CALLSTACK+ F.Foldable f => f a -> (a -> IO b) -> IO ()+forConcurrently_ = flip mapConcurrently_++-- | Perform the action in the given number of threads.+--+-- @since 2.1.1+replicateConcurrently ::+ CALLSTACK+ Int -> IO a -> IO [a]+replicateConcurrently cnt = runConcurrently . replicateM cnt . Concurrently++-- | Same as 'replicateConcurrently', but ignore the results.+--+-- @since 2.1.1+replicateConcurrently_ ::+ CALLSTACK+ Int -> IO a -> IO ()+replicateConcurrently_ cnt = runConcurrently . F.fold . replicate cnt . Concurrently . void++-- -----------------------------------------------------------------------------++-- | A value of type @Concurrently a@ is an @IO@ operation that can be+-- composed with other @Concurrently@ values, using the @Applicative@+-- and @Alternative@ instances.+--+-- Calling @runConcurrently@ on a value of type @Concurrently a@ will+-- execute the @IO@ operations it contains concurrently, before+-- delivering the result of type @a@.+--+-- For example+--+-- > (page1, page2, page3)+-- > <- runConcurrently $ (,,)+-- > <$> Concurrently (getURL "url1")+-- > <*> Concurrently (getURL "url2")+-- > <*> Concurrently (getURL "url3")+--+newtype Concurrently a = Concurrently { runConcurrently :: IO a }++instance Functor Concurrently where+ fmap f (Concurrently a) = Concurrently $ f <$> a++instance Applicative Concurrently where+ pure = Concurrently . return+ Concurrently fs <*> Concurrently as =+ Concurrently $ (\(f, a) -> f a) <$> concurrently fs as++-- | 'Control.Alternative.empty' waits forever. 'Control.Alternative.<|>' returns the first to finish and 'cancel's the other.+instance Alternative Concurrently where+ empty = Concurrently $ forever (threadDelay maxBound)+ Concurrently as <|> Concurrently bs =+ Concurrently $ either id id <$> race as bs++#if MIN_VERSION_base(4,9,0)+-- | Only defined by @async@ for @base >= 4.9@+--+-- @since 2.1.0+instance Semigroup a => Semigroup (Concurrently a) where+ (<>) = liftA2 (<>)++-- | @since 2.1.0+instance (Semigroup a, Monoid a) => Monoid (Concurrently a) where+ mempty = pure mempty+ mappend = (<>)+#else+-- | @since 2.1.0+instance Monoid a => Monoid (Concurrently a) where+ mempty = pure mempty+ mappend = liftA2 mappend+#endif++-- | A value of type @ConcurrentlyE e a@ is an @IO@ operation that can be+-- composed with other @ConcurrentlyE@ values, using the @Applicative@ instance.+--+-- Calling @runConcurrentlyE@ on a value of type @ConcurrentlyE e a@ will+-- execute the @IO@ operations it contains concurrently, before delivering+-- either the result of type @a@, or an error of type @e@ if one of the actions+-- returns @Left@.+--+-- | @since 2.2.5+newtype ConcurrentlyE e a = ConcurrentlyE { runConcurrentlyE :: IO (Either e a) }++instance Functor (ConcurrentlyE e) where+ fmap f (ConcurrentlyE ea) = ConcurrentlyE $ fmap (fmap f) ea++#if MIN_VERSION_base(4,8,0)+instance Bifunctor ConcurrentlyE where+ bimap f g (ConcurrentlyE ea) = ConcurrentlyE $ fmap (bimap f g) ea+#endif++instance Applicative (ConcurrentlyE e) where+ pure = ConcurrentlyE . return . return+ ConcurrentlyE fs <*> ConcurrentlyE eas =+ ConcurrentlyE $ fmap (\(f, a) -> f a) <$> concurrentlyE fs eas++#if MIN_VERSION_base(4,9,0)+-- | Either the combination of the successful results, or the first failure.+instance Semigroup a => Semigroup (ConcurrentlyE e a) where+ (<>) = liftA2 (<>)++instance (Semigroup a, Monoid a) => Monoid (ConcurrentlyE e a) where+ mempty = pure mempty+ mappend = (<>)+#endif++-- ----------------------------------------------------------------------------++-- | Fork a thread that runs the supplied action, and if it raises an+-- exception, re-runs the action. The thread terminates only when the+-- action runs to completion without raising an exception.+forkRepeat ::+ CALLSTACK+ IO a -> IO ThreadId+forkRepeat action =+ mask $ \restore ->+ let go = do r <- tryAll (restore action)+ case r of+ Left _ -> go+ _ -> return ()+ in forkIO (debugLabelMe >> go)++catchAll :: IO a -> (SomeException -> IO a) -> IO a+catchAll = catch++tryAll :: IO a -> IO (Either SomeException a)+tryAll = try++-- A version of forkIO that does not include the outer exception+-- handler: saves a bit of time when we will be installing our own+-- exception handler.+{-# INLINE rawForkIO #-}+rawForkIO ::+ CALLSTACK+ IO () -> IO ThreadId+rawForkIO action = IO $ \ s ->+ case fork# action_plus s of (# s1, tid #) -> (# s1, ThreadId tid #)+ where+ (IO action_plus) = debugLabelMe >> action++{-# INLINE rawForkOn #-}+rawForkOn ::+ CALLSTACK+ Int -> IO () -> IO ThreadId+rawForkOn (I# cpu) action = IO $ \ s ->+ case forkOn# cpu action_plus s of (# s1, tid #) -> (# s1, ThreadId tid #)+ where+ (IO action_plus) = debugLabelMe >> action++debugLabelMe ::+ CALLSTACK+ IO ()+debugLabelMe =+#ifdef DEBUG_AUTO_LABEL+ myThreadId >>= flip labelThread (GHC.Stack.prettyCallStack callStack)+#else+ pure ()+#endif
+ Control/Concurrent/Async/Warden.hs view
@@ -0,0 +1,95 @@+{-+ Copyright (c) Meta Platforms, Inc. and affiliates.+ All rights reserved.++ This source code is licensed under the BSD-style license found in the+ LICENSE file in the root directory of this source tree.+-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | A more flexible way to create 'Async's and have them automatically+-- cancelled when the 'Warden' is shut down.+module Control.Concurrent.Async.Warden+ ( Warden+ , withWarden+ , create+ , shutdown+ , spawn+ , spawn_+ , spawnMask+ , WardenException(..)+ ) where++import Control.Concurrent (forkIO)+import Control.Concurrent.Async (Async)+import qualified Control.Concurrent.Async as Async+import Control.Concurrent.MVar+import Control.Exception+import Data.HashSet (HashSet)+import qualified Data.HashSet as HashSet+import System.IO (fixIO)++#if defined(__MHS__)+import Prelude hiding(mapM_)+import Control.Monad hiding(mapM_)+import Data.Foldable(mapM_)+#else+import Control.Monad+#endif++-- | A 'Warden' is an owner of 'Async's which cancels them on 'shutdown'.+--+-- 'Nothing' in the MVar means the 'Warden' has been shut down.+newtype Warden = Warden (MVar (Maybe (HashSet (Async ()))))++-- | Run the action with a new 'Warden', and call 'shutdown' when the action+-- exits.+withWarden :: (Warden -> IO a) -> IO a+withWarden = bracket create shutdown++-- | Create a new 'Warden'.+create :: IO Warden+create = Warden <$> newMVar (Just mempty)++-- | Shutdown a 'Warden', calling 'cancel' on all owned threads. Subsequent+-- calls to 'spawn' and 'shutdown' will be no-ops. +-- +-- Note that any exceptions thrown by the threads will be ignored. If you want+-- exceptions to be propagated, either call `wait` explicitly on the 'Async', +-- or use 'link'.+shutdown :: Warden -> IO ()+shutdown (Warden v) = do+ r <- swapMVar v Nothing+ mapM_ (Async.mapConcurrently_ Async.cancel) r++forget :: Warden -> Async a -> IO ()+forget (Warden v) async = modifyMVar_ v $ \x -> case x of+ Just xs -> return $! Just $! HashSet.delete (void async) xs+ Nothing -> return Nothing++-- | Spawn a thread with masked exceptions and pass an unmask function to the+-- action.+spawnMask :: Warden -> ((forall b. IO b -> IO b) -> IO a) -> IO (Async a)+spawnMask (Warden v) action = modifyMVar v $ \r -> case r of+ Just asyncs -> do+ -- Create a new thread which removes itself from the 'HashSet' when it+ -- exits.+ this <- fixIO $ \this -> mask_ $ Async.asyncWithUnmask $ \unmask ->+ action unmask `finally` forget (Warden v) this+ return (Just $ HashSet.insert (void this) asyncs, this)+ Nothing -> throwIO $ WardenException "Warden has been shut down"++newtype WardenException = WardenException String+ deriving (Show)++instance Exception WardenException++-- | Spawn a new thread owned by the 'Warden'.+spawn :: Warden -> IO a -> IO (Async a)+spawn warden action = spawnMask warden $ \unmask -> unmask action++-- | Spawn a new thread owned by the 'Warden'.+spawn_ :: Warden -> IO () -> IO ()+spawn_ w = void . spawn w
+ Control/Concurrent/Stream.hs view
@@ -0,0 +1,138 @@+{-+ Copyright (c) Meta Platforms, Inc. and affiliates.+ All rights reserved.++ This source code is licensed under the BSD-style license found in the+ LICENSE file in the root directory of this source tree.+-}++-- | Processing streams with a fixed number of worker threads+module Control.Concurrent.Stream+ ( stream+ , streamBound+ , streamWithInput+ , streamWithOutput+ , streamWithInputOutput+ , mapConcurrentlyBounded+ , forConcurrentlyBounded+ ) where++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception+import Control.Monad+import Data.Maybe+import Data.IORef++data ShouldBindThreads = BoundThreads | UnboundThreads++-- | Maps a fixed number of workers concurrently over a stream of values+-- produced by a producer function. The producer is passed a function to+-- call for each work item. If a worker throws a synchronous exception, it+-- will be propagated to the caller.+stream+ :: Int -- ^ Maximum Concurrency+ -> ((a -> IO ()) -> IO ()) -- ^ Producer+ -> (a -> IO ()) -- ^ Worker+ -> IO ()+stream maxConcurrency producer worker =+ streamWithInput producer (replicate maxConcurrency ()) $ const worker++-- | Like stream, but uses bound threads for the workers. See+-- 'Control.Concurrent.forkOS' for details on bound threads.+streamBound+ :: Int -- ^ Maximum Concurrency+ -> ((a -> IO ()) -> IO ()) -- ^ Producer+ -> (a -> IO ()) -- ^ Worker+ -> IO ()+streamBound maxConcurrency producer worker =+ stream_ BoundThreads producer (replicate maxConcurrency ()) $ const worker++-- | Like stream, but each worker is passed an element of an input list.+streamWithInput+ :: ((a -> IO ()) -> IO ()) -- ^ Producer+ -> [b] -- ^ Worker state+ -> (b -> a -> IO ()) -- ^ Worker+ -> IO ()+streamWithInput = stream_ UnboundThreads++-- | Like 'stream', but collects the results of each worker+streamWithOutput+ :: Int + -> ((a -> IO ()) -> IO ()) -- ^ Producer+ -> (a -> IO c) -- ^ Worker+ -> IO [c]+streamWithOutput maxConcurrency producer worker =+ streamWithInputOutput producer (replicate maxConcurrency ()) $ + const worker++-- | Like 'streamWithInput', but collects the results of each worker+streamWithInputOutput+ :: ((a -> IO ()) -> IO ()) -- ^ Producer+ -> [b] -- ^ Worker input+ -> (b -> a -> IO c) -- ^ Worker+ -> IO [c]+streamWithInputOutput producer workerInput worker = do+ results <- newIORef []+ let prod write = producer $ \a -> do+ res <- newIORef Nothing+ modifyIORef results (res :)+ write (a, res)+ stream_ UnboundThreads prod workerInput $ \s (a,ref) -> do+ worker s a >>= writeIORef ref . Just+ readIORef results >>= mapM readIORef >>= return . catMaybes . reverse+ +stream_+ :: ShouldBindThreads -- use bound threads?+ -> ((a -> IO ()) -> IO ()) -- ^ Producer+ -> [b] -- Worker input+ -> (b -> a -> IO ()) -- ^ Worker+ -> IO ()+stream_ useBoundThreads producer workerInput worker = do+ let maxConcurrency = length workerInput+ q <- atomically $ newTBQueue (fromIntegral maxConcurrency)+ let write x = atomically $ writeTBQueue q (Just x)+ mask $ \unmask ->+ concurrently_ (runWorkers unmask q) $ unmask $ do+ -- run the producer+ producer write+ -- write end-markers for all workers+ replicateM_ maxConcurrency $+ atomically $ writeTBQueue q Nothing+ where+ runWorkers unmask q = case useBoundThreads of+ BoundThreads ->+ foldr1 concurrentlyBound $+ map (runWorker unmask q) workerInput+ UnboundThreads ->+ mapConcurrently_ (runWorker unmask q) workerInput++ concurrentlyBound l r =+ withAsyncBound l $ \a ->+ withAsyncBound r $ \b ->+ void $ waitBoth a b++ runWorker unmask q s = do+ v <- atomically $ readTBQueue q+ case v of+ Nothing -> return ()+ Just t -> do+ unmask (worker s t)+ runWorker unmask q s++-- | Concurrent map over a list of values, using a bounded number of threads.+mapConcurrentlyBounded+ :: Int -- ^ Maximum concurrency+ -> (a -> IO b) -- ^ Function to map over the input values+ -> [a] -- ^ List of input values+ -> IO [b] -- ^ List of output values+mapConcurrentlyBounded maxConcurrency f input =+ streamWithOutput maxConcurrency (forM_ input) f+ +-- | 'mapConcurrentlyBounded' but with its arguments reversed+forConcurrentlyBounded+ :: Int -- ^ Maximum concurrency+ -> [a] -- ^ List of input values+ -> (a -> IO b) -- ^ Function to map over the input values+ -> IO [b] -- ^ List of output values+forConcurrentlyBounded = flip . mapConcurrentlyBounded
async.cabal view
@@ -1,4 +1,6 @@ name: async+version: 2.2.6+-- don't forget to update ./changelog.md! synopsis: Run IO operations asynchronously and wait for their results description:@@ -21,47 +23,7 @@ * The API makes it possible to build a tree of threads that are automatically killed when their parent dies (see 'withAsync').- .- Changes in 2.0.2:- .- * Add a Monad instance for Concurrently- * Bump base dependency to allow 4.9- .- Changes in 2.0.1.6:- .- * Add workaround to waitCatch for #14- .- Changes in 2.0.1.5:- .- * Bump @base@ dependencies for GHC 7.8- .- Changes in 2.0.1.4:- .- * Bump @base@ dependency of test suite- .- Changes in 2.0.1.3:- .- * Bump @base@ dependency to allow 4.6- .- Changes in 2.0.1.2:- .- * Bump @stm@ dependency to 2.4- .- Changes in 2.0.1.1:- .- * Safe Haskell support: @Control.Concurrent.Async@ is now @Trustworthy@- .- Changes in 2.0.1.0:- .- * Added a @Functor@ instance for @Async@- .- * Added @asyncBound@, @asyncOn@, @asyncWithUnmask@, @asyncOnWithUnmask@, @withAsyncBound@, @withAsyncOn@, @withAsyncWithUnmask@, @withAsyncOnWithUnmask@.- .- * Added @mapConcurrently@- .- * Added @Concurrently@ (with @Applicative@ and @Alternative@ instances) -version: 2.0.2 license: BSD3 license-file: LICENSE author: Simon Marlow@@ -69,28 +31,105 @@ copyright: (c) Simon Marlow 2012 category: Concurrency build-type: Simple-cabal-version: >=1.8+cabal-version: >=1.10 homepage: https://github.com/simonmar/async bug-reports: https://github.com/simonmar/async/issues-tested-with: GHC==7.0.3, GHC==7.2.2, GHC==7.4.1+tested-with:+ GHC == 9.14.1+ GHC == 9.12.2+ GHC == 9.10.3+ GHC == 9.8.4+ GHC == 9.6.7+ GHC == 9.4.8+ GHC == 9.2.8+ GHC == 9.0.2+ GHC == 8.10.7+ GHC == 8.8.4+ GHC == 8.6.5+ GHC == 8.4.4+ GHC == 8.2.2+ GHC == 8.0.2+ -- CI does not support GHC 7+ -- GHC == 7.10.3+ -- GHC == 7.8.4+ -- GHC == 7.6.3+ -- GHC == 7.4.2+ -- GHC == 7.2.2+ -- GHC == 7.0.4 extra-source-files:+ changelog.md bench/race.hs source-repository head type: git location: https://github.com/simonmar/async.git +flag debug-auto-label+ description:+ Strictly for debugging as it might have a non-negligible overhead.++ Enabling this flag will auto-label the threads spawned by @async@. Use it to+ find where are unlabelled threads spawned in your program (be it your code or+ dependency code).+ default: False+ manual: True+ library+ default-language: Haskell2010+ other-extensions: CPP, MagicHash, RankNTypes, UnboxedTuples+ if impl(ghc>=7.1)+ other-extensions: Trustworthy exposed-modules: Control.Concurrent.Async- build-depends: base >= 4.3 && < 4.9, stm >= 2.2 && < 2.5+ Control.Concurrent.Async.Internal+ Control.Concurrent.Async.Warden+ Control.Concurrent.Stream+ build-depends: base >= 4.3 && < 4.23,+ hashable >= 1.1.2.0 && < 1.6,+ stm >= 2.2 && < 2.6,+ unordered-containers >= 0.2 && < 0.3+ if flag(debug-auto-label)+ cpp-options: -DDEBUG_AUTO_LABEL test-suite test-async+ default-language: Haskell2010+ ghc-options: -threaded type: exitcode-stdio-1.0 hs-source-dirs: test main-is: test-async.hs- build-depends: base >= 4.3 && < 4.9,+ build-depends: base, async,+ stm, test-framework, test-framework-hunit, HUnit++flag bench+ default: False++executable concasync+ if !flag(bench)+ buildable: False+ default-language: Haskell2010+ hs-source-dirs: bench+ main-is: concasync.hs+ build-depends: base, async, stm+ ghc-options: -O2++executable conccancel+ if !flag(bench)+ buildable: False+ default-language: Haskell2010+ hs-source-dirs: bench+ main-is: conccancel.hs+ build-depends: base, async, stm+ ghc-options: -O2 -threaded++executable race+ if !flag(bench)+ buildable: False+ default-language: Haskell2010+ hs-source-dirs: bench+ main-is: race.hs+ build-depends: base, async, stm+ ghc-options: -O2 -threaded
+ bench/concasync.hs view
@@ -0,0 +1,13 @@+import Control.Concurrent.Async+import System.Environment+import Control.Monad+import Control.Concurrent++main = runInUnboundThread $ do+ [n] <- fmap (fmap read) getArgs+ replicateM_ n $ concurrently (return 1) (return 2)++concurrently' left right =+ withAsync left $ \a ->+ withAsync right $ \b ->+ waitBoth a b
+ bench/conccancel.hs view
@@ -0,0 +1,11 @@+import Control.Exception+import Control.Concurrent.Async+import System.Environment+import Control.Monad+import Control.Concurrent++main = runInUnboundThread $ do+ [n] <- fmap (fmap read) getArgs+ runConcurrently $ traverse Concurrently $+ replicate n (threadDelay 1000000) ++ [throwIO (ErrorCall "oops")]+
+ changelog.md view
@@ -0,0 +1,124 @@+## Changes in 2.2.6++ - Added Control.Concurrent.Stream for processing streams with a fixed+ number of workers. Includes a bounded version of mapConcurrently:+ mapConcurrentlyBounded.+ - Added Control.Concurrent.Async.Warden for a way to create Asyncs that+ is more flexible than 'withAsync' but retains the guarantee of cancelling+ orphaned threads, unlike 'async'.+ - support GHC 9.12, GHC 9.14, and MicroHs+ - cabal flag debug-auto-label: label threads automatically (#167)++## Changes in 2.2.5++ - #117: Document that empty for Concurrently waits forever+ - #120: Add ConcurrentlyE.+ - #123: Fix failing concurrentlyE tests in older GHCs.+ - #124: Allow hashable 1.4+ - #126: Semigroup and Monoid instances for ConcurrentlyE+ - #120: Add ConcurrentlyE+ - #138: expose internals as Control.Concurrent.Async.Internal+ - #131: Fix typos in docs+ - #132: waitAny(Catch): clarify non-empty input list requirement+ - #142: Add cancelMany+ - #135, #145, #150: Support for GHC 9.4, 9.6, 9.8+ - Document that empty for Concurrently waits forever.++## Changes in 2.2.4:++ - Support for GHC 9.2++## Changes in 2.2.3:++ - Documentation fixes++## Changes in 2.2.2:++ - Builds with GHC 8.6.x+ - linkOnly and link2Only are now exported+ - wait now has the same behaviour with BlockedIndefinitelyOnSTM as waitCatch+ - Documentation fixes++## Changes in 2.2.1:++ - Add a Hashable instance for Async+ - Bump upper bounds+ - Documentation updates++## Changes in 2.2:+ - cancel now throws AsyncCancelled instead of ThreadKilled+ - link and link2 now wrap exceptions in ExceptionInLinkedThread when+ throwing to the linked thread. ExceptionInLinkedThread is a child+ of AsyncException in the exception hierarchy, so this maintains the+ invariant that exceptions thrown asynchronously should be+ AsyncExceptions.+ - link and link2 do not propagate AsyncCancelled, so it's now+ possible to cancel a linked thread without cancelling yourself.+ - Added linkOnly and link2Only to specify which exceptions should be+ propagated,if you want something other than the default behaviour+ of ignoring AsyncCancelled.+ - new utility function compareAsyncs for comparing Asyncs of+ different types.+ - Add a `Hashable` instance for `Async a`++## Changes in 2.1.1.1:+ - Make 'cancelWith' wait for the cancelled thread to terminate, like 'cancel'+ - Updates to dependency bounds for GHC 8.2++## Changes in 2.1.1:++ - Add `concurrently_`+ - Add `replicateConcurrently`+ - Add `replicateConcurrently_`+ - Fix incorrect argument order in `forConcurrently_`+ - Generalize `mapConcurrently_` and `forConcurrently_` to `Foldable`+ - `withAsync` now reliably kills the thread, by using an+ uninterruptible cancel+ - Make `cancel` wait for the thread to finish, and adjust+ 'concurrently' to match++## Changes in 2.1.0:++ - Bump base dependency to allow 4.10+ - Remove invalid Monad instance for `Concurrently`+ - Add `Monoid` and `Semigroup` instances for `Concurrently`+ - Add `forConcurrently` (flipped version of `mapConcurrently`)+ - Add STM version of all applicable IO functions:+ `waitAnySTM`, `waitAnyCatchSTM`, `waitEitherSTM`,+ `waitEitherCatchSTM`, `waitEitherSTM_`, and `waitBothSTM`.++## Changes in 2.0.2:++ - Add a Monad instance for `Concurrently`+ - Bump base dependency to allow 4.9++## Changes in 2.0.1.6:++ - Add workaround to waitCatch for #14++## Changes in 2.0.1.5:++ - Bump `base` dependencies for GHC 7.8++## Changes in 2.0.1.4:++ - Bump `base` dependency of test suite++## Changes in 2.0.1.3:++ - Bump `base` dependency to allow 4.6++## Changes in 2.0.1.2:++ - Bump `stm` dependency to 2.4++## Changes in 2.0.1.1:++ - Safe Haskell support: `Control.Concurrent.Async` is now `Trustworthy`++## Changes in 2.0.1.0:++ - Added a `Functor` instance for `Async`+ - Added `asyncBound`, `asyncOn`, `asyncWithUnmask`, `asyncOnWithUnmask`, `withAsyncBound`, `withAsyncOn`, `withAsyncWithUnmask`, `withAsyncOnWithUnmask`.+ - Added `mapConcurrently`+ - Added `Concurrently` (with `Applicative` and `Alternative` instances)
test/test-async.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE ScopedTypeVariables,DeriveDataTypeable #-}+{-# LANGUAGE CPP,ScopedTypeVariables,DeriveDataTypeable #-} module Main where import Test.Framework (defaultMain, testGroup)@@ -6,11 +6,18 @@ import Test.HUnit +import Control.Concurrent.STM import Control.Concurrent.Async+import Control.Concurrent.Async.Warden+import Control.Concurrent.Stream import Control.Exception+import Data.IORef import Data.Typeable import Control.Concurrent import Control.Monad+import Control.Applicative+import Data.List (sort, permutations)+import Data.Foldable (foldMap) import Data.Maybe import Prelude hiding (catch)@@ -27,9 +34,51 @@ , testGroup "async_cancel_rep" $ replicate 1000 $ testCase "async_cancel" async_cancel+ , testCase "async_cancelmany" async_cancelmany , testCase "async_poll" async_poll , testCase "async_poll2" async_poll2 , testCase "withasync_waitCatch_blocked" withasync_waitCatch_blocked+ , testCase "withasync_wait_blocked" withasync_wait_blocked+ , testGroup "children surviving too long"+ [ testCase "concurrently+success" concurrently_success+ , testCase "concurrently+failure" concurrently_failure+ , testCase "concurrentlyE+success" concurrentlyE_success+ , testCase "concurrentlyE+failure" concurrentlyE_failure+ , testCase "race+success" race_success+ , testCase "race+failure" race_failure+ , testCase "cancel" cancel_survive+ , testCase "withAsync" withasync_survive+ ]+ , testCase "concurrently_" case_concurrently_+ , testCase "replicateConcurrently_" case_replicateConcurrently+ , testCase "replicateConcurrently" case_replicateConcurrently_+ , testCase "link1" case_link1+ , testCase "link2" case_link2+ , testCase "link1_cancel" case_link1cancel+ , testCase "concurrently_deadlock" case_concurrently_deadlock+ , testCase "concurrentlyE_deadlock" case_concurrentlyE_deadlock+ , testGroup "concurrentlyE" [+ testCase "concurrentlyE_right" concurrentlyE_right+ , testCase "concurrentlyE_left1" concurrentlyE_left1+ , testCase "concurrentlyE_left2" concurrentlyE_left2+ , testCase "concurrentlyE_earlyException" concurrentlyE_earlyException+ , testCase "concurrentlyE_lateException" concurrentlyE_lateException+#if MIN_VERSION_base(4,9,0)+ , testCase "concurrentlyE_Monoid" concurrentlyE_Monoid+ , testCase "concurrentlyE_Monoid_fail" concurrentlyE_Monoid_fail+#endif+ , testCase "stream" $ case_stream False+ , testCase "streamBound" $ case_stream True+ , testCase "stream_exception" $ case_stream_exception False+ , testCase "streamBound_exception" $ case_stream_exception True+ , testCase "streamWithInput" case_streamInput+ , testCase "streamWithInput_exception" case_streamInput_exception+ , testCase "mapConcurrentlyBounded" case_mapConcurrentlyBounded+ , testCase "mapConcurrentlyBounded_exception" + case_mapConcurrentlyBounded_exception+ , testCase "Warden" case_Warden+ , testCase "Warden_spawn_after_shutdown" case_Warden_spawn_after_shutdown+ ] ] value = 42 :: Int@@ -77,7 +126,7 @@ a <- withAsync (threadDelay 1000000) $ return r <- waitCatch a case r of- Left e -> fromException e @?= Just ThreadKilled+ Left e -> fromException e @?= Just AsyncCancelled Right _ -> assertFailure "" async_cancel :: Assertion@@ -89,6 +138,18 @@ Left e -> fromException e @?= Just TestException Right r -> r @?= value +async_cancelmany :: Assertion -- issue 59+async_cancelmany = do+ r <- newIORef []+ a <- async $ forConcurrently_ ['a'..'z'] $ \c ->+ delay 2 `finally` atomicModifyIORef r (\i -> (c:i,()))+ delay 1+ cancel a+ v <- readIORef r+ assertEqual "cancelmany" 26 (length v)+ where+ delay sec = threadDelay (sec * 1000000)+ async_poll :: Assertion async_poll = do a <- async (threadDelay 1000000)@@ -115,3 +176,375 @@ Just BlockedIndefinitelyOnMVar -> return () Nothing -> assertFailure $ show e Right () -> assertFailure ""++withasync_wait_blocked :: Assertion+withasync_wait_blocked = do+ r <- try $ withAsync (newEmptyMVar >>= takeMVar) wait+ case r of+ Left e ->+ case fromException e of+ Just BlockedIndefinitelyOnMVar -> return ()+ Nothing -> assertFailure $ show e+ Right () -> assertFailure ""++concurrently_success :: Assertion+concurrently_success = do+ finalRes <- newIORef "never filled"+ baton <- newEmptyMVar+ let quick = return ()+ slow = threadDelay 10000 `finally` do+ threadDelay 10000+ writeIORef finalRes "slow"+ putMVar baton ()+ _ <- concurrently quick slow+ writeIORef finalRes "parent"+ takeMVar baton+ res <- readIORef finalRes+ res @?= "parent"++concurrentlyE_success :: Assertion+concurrentlyE_success = do+ finalRes <- newIORef "never filled"+ baton <- newEmptyMVar+ let quick = return (Right ())+ slow = threadDelay 10000 *> return (Right ()) `finally` do+ threadDelay 10000+ writeIORef finalRes "slow"+ putMVar baton ()+ _ <- concurrentlyE quick slow+ writeIORef finalRes "parent"+ takeMVar baton+ res <- readIORef finalRes+ res @?= "parent"++concurrently_failure :: Assertion+concurrently_failure = do+ finalRes <- newIORef "never filled"+ let quick = error "a quick death"+ slow = threadDelay 10000 `finally` do+ threadDelay 10000+ writeIORef finalRes "slow"+ _ :: Either SomeException ((), ()) <- try (concurrently quick slow)+ writeIORef finalRes "parent"+ threadDelay 1000000 -- not using the baton, can lead to deadlock detection+ res <- readIORef finalRes+ res @?= "parent"++concurrentlyE_failure :: Assertion+concurrentlyE_failure = do+ finalRes <- newIORef "never filled"+ let quick = error "a quick death"+ slow = threadDelay 10000 *> return (Right ()) `finally` do+ threadDelay 10000+ writeIORef finalRes "slow"+ _ :: Either SomeException (Either () ((), ())) <- try (concurrentlyE quick slow)+ writeIORef finalRes "parent"+ threadDelay 1000000 -- not using the baton, can lead to deadlock detection+ res <- readIORef finalRes+ res @?= "parent"++race_success :: Assertion+race_success = do+ finalRes <- newIORef "never filled"+ let quick = return ()+ slow = threadDelay 10000 `finally` do+ threadDelay 10000+ writeIORef finalRes "slow"+ race_ quick slow+ writeIORef finalRes "parent"+ threadDelay 1000000 -- not using the baton, can lead to deadlock detection+ res <- readIORef finalRes+ res @?= "parent"++race_failure :: Assertion+race_failure = do+ finalRes <- newIORef "never filled"+ baton <- newEmptyMVar+ let quick = error "a quick death"+ slow restore = restore (threadDelay 10000) `finally` do+ threadDelay 10000+ writeIORef finalRes "slow"+ putMVar baton ()+ _ :: Either SomeException () <-+ try $ mask $ \restore ->+ race_ quick (slow restore)+ writeIORef finalRes "parent"+ takeMVar baton+ res <- readIORef finalRes+ res @?= "parent"++cancel_survive :: Assertion+cancel_survive = do+ finalRes <- newIORef "never filled"+ a <- async $ threadDelay 10000 `finally` do+ threadDelay 10000+ writeIORef finalRes "child"+ cancel a+ writeIORef finalRes "parent"+ threadDelay 1000000 -- not using the baton, can lead to deadlock detection+ res <- readIORef finalRes+ res @?= "parent"++withasync_survive :: Assertion+withasync_survive = do+ finalRes <- newIORef "never filled"+ let child = threadDelay 10000 `finally` do+ threadDelay 10000+ writeIORef finalRes "child"+ withAsync child (\_ -> return ())+ writeIORef finalRes "parent"+ threadDelay 1000000 -- not using the baton, can lead to deadlock detection+ res <- readIORef finalRes+ res @?= "parent"++case_concurrently_ :: Assertion+case_concurrently_ = do+ ref <- newIORef 0+ () <- concurrently_+ (atomicModifyIORef ref (\x -> (x + 1, True)))+ (atomicModifyIORef ref (\x -> (x + 2, 'x')))+ res <- readIORef ref+ res @?= 3++case_replicateConcurrently :: Assertion+case_replicateConcurrently = do+ ref <- newIORef 0+ let action = atomicModifyIORef ref (\x -> (x + 1, x + 1))+ resList <- replicateConcurrently 100 action+ resVal <- readIORef ref+ resVal @?= 100+ sort resList @?= [1..100]++case_replicateConcurrently_ :: Assertion+case_replicateConcurrently_ = do+ ref <- newIORef 0+ let action = atomicModifyIORef ref (\x -> (x + 1, x + 1))+ () <- replicateConcurrently_ 100 action+ resVal <- readIORef ref+ resVal @?= 100++case_link1 :: Assertion+case_link1 = do+ m1 <- newEmptyMVar+ m2 <- newEmptyMVar+ let ex = ErrorCall "oops"+ a <- async $ do takeMVar m1; throwIO ex; putMVar m2 ()+ link a+ e <- try $ (do+ putMVar m1 ()+ takeMVar m2)+ assertBool "link1" $+ case e of+ Left (ExceptionInLinkedThread a' e') ->+ compareAsyncs a' a == EQ &&+ case fromException e' of+ Just (ErrorCall s) -> s == "oops"+ _otherwise -> False+ _other -> False++case_link2 :: Assertion+case_link2 = do+ let+ setup = do+ m1 <- newEmptyMVar+ m2 <- newEmptyMVar+ let ex1 = ErrorCall "oops1"; ex2 = ErrorCall "oops2"+ a <- async $ do takeMVar m1; throwIO ex1+ b <- async $ do takeMVar m2; throwIO ex2+ link2 a b+ return (m1,m2,a,b)++ (m1,m2,a,b) <- setup+ e <- try $ do+ putMVar m1 ()+ wait b+ putMVar m2 () -- ensure the other thread is not deadlocked+ assertBool "link2a" $+ case e of+ Left (ExceptionInLinkedThread a' e') ->+ compareAsyncs a' a == EQ &&+ case fromException e' of+ Just (ErrorCall s) -> s == "oops1"+ _otherwise -> False+ _other -> False++ (m1,m2,a,b) <- setup+ e <- try $ do+ putMVar m2 ()+ wait a+ putMVar m1 () -- ensure the other thread is not deadlocked+ assertBool "link2b" $+ case e of+ Left (ExceptionInLinkedThread a' e') ->+ compareAsyncs a' b == EQ &&+ case fromException e' of+ Just (ErrorCall s) -> s == "oops2"+ _otherwise -> False+ _other -> False++case_link1cancel :: Assertion+case_link1cancel = do+ m1 <- newEmptyMVar+ let ex = ErrorCall "oops"+ a <- async $ do takeMVar m1+ link a+ e <- try $ do cancel a; wait a+ putMVar m1 ()+ assertBool "link1cancel" $+ case e of+ Left AsyncCancelled -> True -- should not be ExceptionInLinkedThread+ _other -> False++-- See Issue #62+case_concurrently_deadlock :: Assertion+case_concurrently_deadlock = do+ tvar <- newTVarIO False :: IO (TVar Bool)+ e <- try $ void $ join (concurrently) (atomically $ readTVar tvar >>= check)+ -- should throw BlockedIndefinitelyOnSTM not BlockedIndefinitelyOnMVar+ assertBool "concurrently_deadlock" $+ case e of+ Left BlockedIndefinitelyOnSTM{} -> True+ _other -> False++-- See Issue #62+case_concurrentlyE_deadlock :: Assertion+case_concurrentlyE_deadlock = do+ tvar <- newTVarIO False :: IO (TVar Bool)+ e <- try $ void $ join (concurrentlyE) (fmap Right $ atomically $ readTVar tvar >>= check)+ -- should throw BlockedIndefinitelyOnSTM not BlockedIndefinitelyOnMVar+ assertBool "concurrentlyE_deadlock" $+ case e of+ Left BlockedIndefinitelyOnSTM{} -> True+ _other -> False++concurrentlyE_right :: Assertion+concurrentlyE_right = do+ r :: Either () (Bool,Bool) <- concurrentlyE (Right . const False <$> threadDelay 10000) (Right . const True <$> threadDelay 10000)+ assertEqual "should be Right" (Right (False,True)) r++concurrentlyE_left1 :: Assertion+concurrentlyE_left1 = do+ r :: Either () ((),()) <- concurrentlyE (Left <$> threadDelay 10000) (Right <$> forever (threadDelay 10000))+ assertEqual "should be Left" (Left ()) r++concurrentlyE_left2 :: Assertion+concurrentlyE_left2 = do+ r :: Either () ((),()) <- concurrentlyE (Right <$> forever (threadDelay 10000)) (Left <$> threadDelay 10000) + assertEqual "should be Left" (Left ()) r++concurrentlyE_earlyException :: Assertion+concurrentlyE_earlyException = do+ ref <- newIORef "never filled"+ r :: Either TestException (Either () (Bool,Bool)) <- try $ + concurrentlyE + ((Right . const False <$> forever (threadDelay 10000)) `onException` writeIORef ref "finalized")+ (threadDelay 1000 *> throwIO TestException)+ refVal <- readIORef ref+ assertEqual "should be Exception" (Left TestException, "finalized") (r, refVal)++concurrentlyE_lateException :: Assertion+concurrentlyE_lateException = do+ ref <- newIORef "never filled"+ r :: Either TestException (Either () (Bool,Bool)) <- try $ + concurrentlyE + ((Right . const False <$> threadDelay 100) `onException` writeIORef ref "finalized")+ (threadDelay 100000 *> throwIO TestException)+ refVal <- readIORef ref+ assertEqual "should be Exception" (Left TestException, "never filled") (r, refVal)++#if MIN_VERSION_base(4,9,0)+concurrentlyE_Monoid :: Assertion+concurrentlyE_Monoid = do+ let delays :: [Int]+ delays = [1000, 10000, 100000]+ actions = zipWith (*>) (threadDelay <$> delays) (pure . Right . (:[]) <$> ['a'..])+ r :: Either () String <- runConcurrentlyE $ foldMap ConcurrentlyE $ actions+ assertEqual "Combined result in order" (Right "abc") r++concurrentlyE_Monoid_fail :: Assertion+concurrentlyE_Monoid_fail = do+ let delays :: [Int]+ delays = [1000, 200000]+ actions = zipWith (*>) (threadDelay <$> delays) (pure . Right . (:[]) <$> ['a'..])+ failDelays = [10000, 100000]+ failActions = zipWith (*>) (threadDelay <$> delays) (pure . Left <$> ['u'..])+ forM_ (permutations (actions ++ failActions)) $ \current -> do+ r :: Either Char [Char] <- runConcurrentlyE $ foldMap ConcurrentlyE $ current+ assertEqual "The earliest failure" (Left 'u') r+#endif++case_stream :: Bool -> Assertion+case_stream bound = do+ ref <- newIORef []+ let inp = [1..100]+ let producer write = forM_ inp $ \x -> write (show x)+ (if bound then streamBound else stream) 4 producer $ \s -> atomicModifyIORef ref (\l -> (s:l, ()))+ res <- readIORef ref+ sort res @?= sort (map show inp)++case_stream_exception :: Bool -> Assertion+case_stream_exception bound = do+ let inp = [1..100]+ let producer write = forM_ inp $ \x -> write (show x)+ r <- try $ (if bound then streamBound else stream) 4 producer $ \s -> + when (s == "3") $ throwIO (ErrorCall s)+ r @?= Left (ErrorCall "3" :: ErrorCall)++case_streamInput :: Assertion+case_streamInput = do+ ref <- newIORef []+ let inp = [1..100]; workers = [1..4] :: [Int]+ let producer write = forM_ inp $ \x -> write (show x)+ streamWithInput producer workers $ \s t -> atomicModifyIORef ref (\l -> ((s,t):l, ()))+ res <- readIORef ref+ sort (map snd res) @?= sort (map show inp)+ all ((`elem` workers) . fst) res @?= True++case_streamInput_exception :: Assertion+case_streamInput_exception = do+ let inp = [1..100]; workers = [1..4] :: [Int]+ let producer write = forM_ inp $ \x -> write (show x)+ r <- try $ streamWithInput producer workers $ \s t -> + when (t == "3") $ throwIO (ErrorCall t)+ r @?= Left (ErrorCall "3" :: ErrorCall)++case_mapConcurrentlyBounded :: Assertion+case_mapConcurrentlyBounded = do+ let inp = [1..100]+ let f x = threadDelay 1000 >> return (x * 2)+ res <- mapConcurrentlyBounded 4 f inp+ res @?= map (*2) inp++case_mapConcurrentlyBounded_exception :: Assertion+case_mapConcurrentlyBounded_exception = do+ let inp = [1..100]+ let f x | x == 3 = throwIO $ ErrorCall "3"+ | otherwise = threadDelay 1000 >> return (x * 2)+ res <- try $ mapConcurrentlyBounded 4 f inp+ res @?= Left (ErrorCall "3" :: ErrorCall)++case_Warden :: Assertion+case_Warden = do+ a3 <- withWarden $ \warden -> do+ a1 <- spawn warden $ return 1+ a2 <- spawnMask warden $ \unmask -> unmask (return 2)+ a3 <- spawn warden $ threadDelay 10000000+ spawn_ warden $ throwIO (ErrorCall "a4") -- ignored+ r1 <- wait a1+ r1 @?= 1+ r2 <- wait a2+ r2 @?= 2+ return a3+ r3 <- waitCatch a3+ case r3 of+ Right _ -> assertFailure "Expected AsyncCancelled"+ Left e -> fromException e @?= Just AsyncCancelled++case_Warden_spawn_after_shutdown :: Assertion+case_Warden_spawn_after_shutdown = do+ warden <- create+ shutdown warden+ r <- try $ spawn warden $ return ()+ case r of+ Left (WardenException{}) -> return () -- expected+ Right _ -> assertFailure "Expected WardenException"