async 2.2.1 → 2.2.6
raw patch · 7 files changed
Files
- Control/Concurrent/Async.hs +120/−804
- Control/Concurrent/Async/Internal.hs +964/−0
- Control/Concurrent/Async/Warden.hs +95/−0
- Control/Concurrent/Stream.hs +138/−0
- async.cabal +44/−6
- changelog.md +41/−0
- test/test-async.hs +213/−2
Control/Concurrent/Async.hs view
@@ -1,13 +1,3 @@-{-# 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@@ -24,65 +14,123 @@ -- "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.+-- robustness over using 'forkIO' threads and @MVar@ directly. --+-- == High-level API+--+-- @async@'s high-level API spawns /lexically scoped/ threads,+-- ensuring the following key properties that make it safer to use+-- than using plain 'forkIO':+--+-- 1. No exception is swallowed (waiting for results propagates exceptions).+-- 2. No thread is leaked (left running unintentionally).+--+-- (This is done using the 'Control.Exception.bracket' pattern to work in presence+-- of synchronous and asynchronous exceptions.)+--+-- __Most practical/production code should only use the high-level API__.+-- -- 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.+-- exception. An 'Async' is a wrapper around a low-level 'forkIO' thread. --+-- The fundamental function to spawn threads with the high-level API is+-- 'withAsync'.+-- -- 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--- > ...+-- > withAsync (getURL url1) $ \a1 -> do+-- > withAsync (getURL url2) $ \a2 -> do+-- > 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.+-- where 'withAsync' starts the operation in a separate thread, and+-- 'wait' waits for and returns the result. ----- A slight improvement over the previous example is this:+-- * 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. ----- > withAsync (getURL url1) $ \a1 -> do--- > withAsync (getURL url2) $ \a2 -> do--- > page1 <- wait a1--- > page2 <- wait a2--- > ...+-- 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. --+-- 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:+--+-- > (page1, page2) <- concurrently (getURL url1) (getURL url2)+-- > ...+--+-- The section __/High-level utilities/__ covers the most+-- common high-level objectives, including:+--+-- * 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.+--+-- Click here to scroll to that section:+-- "Control.Concurrent.Async#high-level-utilities".+--+-- == Low-level API+--+-- Some use cases require parallelism that is not lexically scoped.+--+-- For those, the low-level function 'async' can be used as a direct+-- equivalent of 'forkIO':+--+-- > -- 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+-- > ...+--+-- In contrast to 'withAsync', this code has a problem.+--+-- 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.+-- -- 'withAsync' is like 'async', except that the 'Async' is -- automatically killed (using 'uninterruptibleCancel') 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.+-- 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. ----- 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:+-- 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". ----- > (page1, page2) <- concurrently (getURL url1) (getURL url2)--- > ...+-- == Miscellaneous -- -- 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)+-- > ghci> withAsync (return 3) (\a -> wait (fmap (+1) a)) -- > 4+--+-- === Resource exhaustion+--+-- 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. ----------------------------------------------------------------------------- @@ -90,785 +138,53 @@ -- * Asynchronous actions Async,- -- ** Spawning- async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask, + -- * High-level API+ -- ** Spawning with automatic 'cancel'ation withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask, withAsyncOnWithUnmask, -- ** Querying 'Async's wait, poll, waitCatch, asyncThreadId,- cancel, uninterruptibleCancel, cancelWith, AsyncCancelled(..),-- -- ** STM operations- waitSTM, pollSTM, waitCatchSTM,-- -- ** 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,-- -- ** Linking- link, link2, ExceptionInLinkedThread(..),+ cancel, cancelMany, uninterruptibleCancel, cancelWith, AsyncCancelled(..), - -- * Convenient utilities+ -- ** #high-level-utilities# High-level utilities race, race_, concurrently, concurrently_, mapConcurrently, forConcurrently, mapConcurrently_, forConcurrently_, replicateConcurrently, replicateConcurrently_, Concurrently(..),+ concurrentlyE,+ ConcurrentlyE(..), compareAsyncs, - ) 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,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---- -------------------------------------------------------------------------------- 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 'Async's that may have different types-compareAsyncs :: Async a -> Async b -> Ordering-compareAsyncs (Async t1 _) (Async t2 _) = compare t1 t2---- | 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, 'uninterruptibleCancel' is called on the @Async@.------ > withAsync action inner = mask $ \restore -> do--- > a <- async (restore action)--- > restore inner `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 :: 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- 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 = 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@.------ > 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---- | 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'.------ 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 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` 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'.------ 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 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` mapM_ cancel 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` (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'.----{-# 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` (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'.----{-# INLINE waitBoth #-}-waitBoth :: Async a -> Async b -> IO (a,b)-waitBoth left right = atomically (waitBothSTM left right)---- | 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- show (ExceptionInLinkedThread (Async t _) e) =- "ExceptionInLinkedThread " ++ show t ++ " " ++ show 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. 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)--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 :: IO a -> IO b -> IO (Either a b)---- | Like 'race', but the result is ignored.----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'.------ > 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 <- 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--concurrently' :: 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--#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"]----mapConcurrently :: 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 :: Traversable t => t a -> (a -> IO b) -> IO (t b)-forConcurrently = flip mapConcurrently---- | `mapConcurrently_` is `mapConcurrently` with the return value discarded,--- just like @mapM_-mapConcurrently_ :: 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,--- just like @forM_-forConcurrently_ :: F.Foldable f => f a -> (a -> IO b) -> IO ()-forConcurrently_ = flip mapConcurrently_---- | 'concurrently', but ignore the result values------ @since 2.1.1-concurrently_ :: IO a -> IO b -> IO ()-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---- | Perform the action in the given number of threads.------ @since 2.1.1-replicateConcurrently :: Int -> IO a -> IO [a]-replicateConcurrently cnt = runConcurrently . sequenceA . replicate cnt . Concurrently---- | Same as 'replicateConcurrently', but ignore the results.------ @since 2.1.1-replicateConcurrently_ :: 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--instance Alternative Concurrently where- empty = Concurrently $ forever (threadDelay maxBound)- Concurrently as <|> Concurrently bs =- Concurrently $ either id id <$> race as bs+ -- ** Specialised operations -#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 (<>)+ -- *** STM operations+ waitSTM, pollSTM, waitCatchSTM, --- | @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+ -- *** 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,5 +1,5 @@ name: async-version: 2.2.1+version: 2.2.6 -- don't forget to update ./changelog.md! synopsis: Run IO operations asynchronously and wait for their results @@ -34,7 +34,28 @@ cabal-version: >=1.10 homepage: https://github.com/simonmar/async bug-reports: https://github.com/simonmar/async/issues-tested-with: GHC==8.2.2, GHC==8.0.2, GHC==7.10.3, GHC==7.8.4, GHC==7.6.3, GHC==7.4.2, GHC==7.2.2, GHC==7.0.4+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@@ -44,20 +65,39 @@ 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.12, hashable >= 1.1.1.0 && < 1.3, 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.12,+ build-depends: base, async, stm, test-framework,@@ -93,5 +133,3 @@ main-is: race.hs build-depends: base, async, stm ghc-options: -O2 -threaded--
changelog.md view
@@ -1,3 +1,44 @@+## 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
test/test-async.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE ScopedTypeVariables,DeriveDataTypeable #-}+{-# LANGUAGE CPP,ScopedTypeVariables,DeriveDataTypeable #-} module Main where import Test.Framework (defaultMain, testGroup)@@ -8,12 +8,16 @@ 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 Data.List (sort)+import Control.Applicative+import Data.List (sort, permutations)+import Data.Foldable (foldMap) import Data.Maybe import Prelude hiding (catch)@@ -34,9 +38,12 @@ , 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@@ -49,6 +56,29 @@ , 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@@ -147,6 +177,16 @@ 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"@@ -162,6 +202,21 @@ 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"@@ -175,6 +230,19 @@ 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"@@ -337,3 +405,146 @@ 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"