diff --git a/Control/Concurrent/Async.hs b/Control/Concurrent/Async.hs --- a/Control/Concurrent/Async.hs +++ b/Control/Concurrent/Async.hs @@ -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 diff --git a/Control/Concurrent/Async/Internal.hs b/Control/Concurrent/Async/Internal.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Async/Internal.hs @@ -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 +-- 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 diff --git a/Control/Concurrent/Async/Warden.hs b/Control/Concurrent/Async/Warden.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Async/Warden.hs @@ -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 diff --git a/Control/Concurrent/Stream.hs b/Control/Concurrent/Stream.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Stream.hs @@ -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 diff --git a/async.cabal b/async.cabal --- a/async.cabal +++ b/async.cabal @@ -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 - - diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -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 diff --git a/test/test-async.hs b/test/test-async.hs --- a/test/test-async.hs +++ b/test/test-async.hs @@ -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"