diff --git a/Control/Concurrent/Async.hs b/Control/Concurrent/Async.hs --- a/Control/Concurrent/Async.hs +++ b/Control/Concurrent/Async.hs @@ -53,15 +53,15 @@ -- > ... -- -- 'withAsync' is like 'async', except that the 'Async' is --- automatically killed (using 'cancel') if the enclosing IO operation --- returns before it has completed. Consider the case when the first --- 'wait' throws an exception; then the second 'Async' will be --- automatically killed rather than being left to run in the --- background, possibly indefinitely. This is the second way that the --- library provides additional safety: using 'withAsync' means we can --- avoid accidentally leaving threads running. Furthermore, --- 'withAsync' allows a tree of threads to be built, such that --- children are automatically killed if their parents die for any +-- automatically killed (using 'uninterruptibleCancel') if the +-- enclosing IO opercation returns before it has completed. Consider +-- the case when the first 'wait' throws an exception; then the second +-- 'Async' will be automatically killed rather than being left to run +-- in the background, possibly indefinitely. This is the second way +-- that the library provides additional safety: using 'withAsync' +-- means we can avoid accidentally leaving threads running. +-- Furthermore, 'withAsync' allows a tree of threads to be built, such +-- that children are automatically killed if their parents die for any -- reason. -- -- The pattern of performing two IO actions concurrently and waiting @@ -90,10 +90,11 @@ async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask, -- ** Spawning with automatic 'cancel'ation - withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask, withAsyncOnWithUnmask, + withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask, + withAsyncOnWithUnmask, -- ** Querying 'Async's - wait, poll, waitCatch, cancel, cancelWith, + wait, poll, waitCatch, cancel, uninterruptibleCancel, cancelWith, asyncThreadId, -- ** STM operations @@ -115,7 +116,11 @@ link, link2, -- * Convenient utilities - race, race_, concurrently, mapConcurrently, forConcurrently, + race, race_, + concurrently, concurrently_, + mapConcurrently, forConcurrently, + mapConcurrently_, forConcurrently_, + replicateConcurrently, replicateConcurrently_, Concurrently(..), ) where @@ -123,6 +128,7 @@ 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 @@ -136,6 +142,7 @@ import Data.Semigroup (Semigroup((<>))) #endif +import Data.IORef import GHC.Exts import GHC.IO hiding (finally, onException) @@ -150,9 +157,12 @@ -- 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) } +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 @@ -175,15 +185,18 @@ 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. +-- | 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. +-- | 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) +asyncOnWithUnmask cpu actionWith = + asyncUsing (rawForkOn cpu) (actionWith unsafeUnmask) asyncUsing :: (IO () -> IO ThreadId) -> IO a -> IO (Async a) @@ -197,16 +210,13 @@ -- | Spawn an asynchronous action in a separate thread, and pass its -- @Async@ handle to the supplied function. When the function returns --- or throws an exception, 'cancel' is called on the @Async@. +-- or throws an exception, 'uninterruptibleCancel' is called on the @Async@. -- --- > withAsync action inner = bracket (async action) cancel inner +-- > withAsync action inner = bracket (async action) uninterruptibleCancel inner -- -- This is a useful variant of 'async' that ensures an @Async@ is -- never left running unintentionally. -- --- Since 'cancel' may block, 'withAsync' may also block; see 'cancel' --- for details. --- withAsync :: IO a -> (Async a -> IO b) -> IO b withAsync = inline withAsyncUsing rawForkIO @@ -218,15 +228,21 @@ 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 '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) +-- | 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 @@ -237,8 +253,10 @@ mask $ \restore -> do t <- doFork $ try (restore action) >>= atomically . putTMVar var let a = Async t (readTMVar var) - r <- restore (inner a) `catchAll` \e -> do cancel a; throwIO e - cancel a + 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 @@ -295,23 +313,32 @@ pollSTM (Async _ w) = (Just <$> w) `orElse` return Nothing -- | Cancel an asynchronous action by throwing the @ThreadKilled@ --- exception to it. Has no effect if the 'Async' has already --- completed. +-- 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) ThreadKilled +-- > cancel a = throwTo (asyncThreadId a) ThreadKilled <* waitCatch w -- --- Note that 'cancel' is synchronous in the same sense as 'throwTo'. --- It does not return until the exception has been thrown in the --- target thread, or the target thread has completed. In particular, --- if the target thread is making a foreign call, the exception will --- not be thrown until the foreign call returns, and in this case --- 'cancel' may block indefinitely. An asynchronous 'cancel' can --- of course be obtained by wrapping 'cancel' itself in 'async'. +-- 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 (Async t _) = throwTo t ThreadKilled +cancel a@(Async t _) = throwTo t ThreadKilled <* waitCatch a +-- | 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. -- @@ -541,7 +568,7 @@ race left right = concurrently' left right collect where collect m = do - e <- takeMVar m + e <- m case e of Left ex -> throwIO ex Right r -> return r @@ -555,13 +582,13 @@ collect [Left a, Right b] _ = return (a,b) collect [Right b, Left a] _ = return (a,b) collect xs m = do - e <- takeMVar m + e <- m case e of Left ex -> throwIO ex Right r -> collect (r:xs) m concurrently' :: IO a -> IO b - -> (MVar (Either SomeException (Either a b)) -> IO r) + -> (IO (Either SomeException (Either a b)) -> IO r) -> IO r concurrently' left right collect = do done <- newEmptyMVar @@ -570,10 +597,25 @@ `catchAll` (putMVar done . Left) rid <- forkIO $ restore (right >>= putMVar done . Right . Right) `catchAll` (putMVar done . Left) - let stop = killThread rid >> killThread lid - -- kill right before left, to match the semantics of - -- the version using withAsync. (#27) - r <- restore (collect done) `onException` stop + + count <- newIORef (2 :: Int) + let takeDone = do + -- 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. + modifyIORef count (subtract 1) + + takeMVar done + + let stop = do + -- kill right before left, to match the semantics of + -- the version using withAsync. (#27) + uninterruptibleMask_ $ do + killThread rid >> killThread lid + -- ensure the children are really dead + count' <- readIORef count + replicateM_ count' (takeMVar done) + r <- restore (collect takeDone) `onException` stop stop return r @@ -598,6 +640,41 @@ -- @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 -- ----------------------------------------------------------------------------- diff --git a/async.cabal b/async.cabal --- a/async.cabal +++ b/async.cabal @@ -1,5 +1,5 @@ name: async -version: 2.1.0 +version: 2.1.1 -- don't forget to update ./changelog.md! synopsis: Run IO operations asynchronously and wait for their results diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,15 @@ +## Changes in 2.1.1: + + - Add `concurrently_` + - Add `replicateConcurrently` + - Add `replicateConcurrently_` + - Fix incorrect argument order in `forConcurrently_` + - Generalize `mapConcurrently_` and `forConcurrently_` to `Foldable` + - `withAsync` now reliably kills the thread, by using an + uninterruptible cancel + - Make `cancel` wait for the thread to finish, and adjust + 'concurrently' to match + ## Changes in 2.1.0: - Bump base dependency to allow 4.10 diff --git a/test/test-async.hs b/test/test-async.hs --- a/test/test-async.hs +++ b/test/test-async.hs @@ -8,9 +8,11 @@ import Control.Concurrent.Async import Control.Exception +import Data.IORef import Data.Typeable import Control.Concurrent import Control.Monad +import Data.List (sort) import Data.Maybe import Prelude hiding (catch) @@ -30,6 +32,17 @@ , testCase "async_poll" async_poll , testCase "async_poll2" async_poll2 , testCase "withasync_waitCatch_blocked" withasync_waitCatch_blocked + , testGroup "children surviving too long" + [ testCase "concurrently+success" concurrently_success + , testCase "concurrently+failure" concurrently_failure + , testCase "race+success" race_success + , testCase "race+failure" race_failure + , testCase "cancel" cancel_survive + , testCase "withAsync" withasync_survive + ] + , testCase "concurrently_" case_concurrently_ + , testCase "replicateConcurrently_" case_replicateConcurrently + , testCase "replicateConcurrently" case_replicateConcurrently_ ] value = 42 :: Int @@ -115,3 +128,111 @@ Just BlockedIndefinitelyOnMVar -> return () Nothing -> assertFailure $ show e Right () -> assertFailure "" + +concurrently_success :: Assertion +concurrently_success = do + finalRes <- newIORef "never filled" + baton <- newEmptyMVar + let quick = return () + slow = threadDelay 10000 `finally` do + threadDelay 10000 + writeIORef finalRes "slow" + putMVar baton () + _ <- concurrently quick slow + writeIORef finalRes "parent" + takeMVar baton + res <- readIORef finalRes + res @?= "parent" + +concurrently_failure :: Assertion +concurrently_failure = do + finalRes <- newIORef "never filled" + let quick = error "a quick death" + slow = threadDelay 10000 `finally` do + threadDelay 10000 + writeIORef finalRes "slow" + _ :: Either SomeException ((), ()) <- try (concurrently quick slow) + writeIORef finalRes "parent" + threadDelay 1000000 -- not using the baton, can lead to deadlock detection + res <- readIORef finalRes + res @?= "parent" + +race_success :: Assertion +race_success = do + finalRes <- newIORef "never filled" + let quick = return () + slow = threadDelay 10000 `finally` do + threadDelay 10000 + writeIORef finalRes "slow" + race_ quick slow + writeIORef finalRes "parent" + threadDelay 1000000 -- not using the baton, can lead to deadlock detection + res <- readIORef finalRes + res @?= "parent" + +race_failure :: Assertion +race_failure = do + finalRes <- newIORef "never filled" + baton <- newEmptyMVar + let quick = error "a quick death" + slow restore = restore (threadDelay 10000) `finally` do + threadDelay 10000 + writeIORef finalRes "slow" + putMVar baton () + _ :: Either SomeException () <- + try $ mask $ \restore -> + race_ quick (slow restore) + writeIORef finalRes "parent" + takeMVar baton + res <- readIORef finalRes + res @?= "parent" + +cancel_survive :: Assertion +cancel_survive = do + finalRes <- newIORef "never filled" + a <- async $ threadDelay 10000 `finally` do + threadDelay 10000 + writeIORef finalRes "child" + cancel a + writeIORef finalRes "parent" + threadDelay 1000000 -- not using the baton, can lead to deadlock detection + res <- readIORef finalRes + res @?= "parent" + +withasync_survive :: Assertion +withasync_survive = do + finalRes <- newIORef "never filled" + let child = threadDelay 10000 `finally` do + threadDelay 10000 + writeIORef finalRes "child" + withAsync child (\_ -> return ()) + writeIORef finalRes "parent" + threadDelay 1000000 -- not using the baton, can lead to deadlock detection + res <- readIORef finalRes + res @?= "parent" + +case_concurrently_ :: Assertion +case_concurrently_ = do + ref <- newIORef 0 + () <- concurrently_ + (atomicModifyIORef ref (\x -> (x + 1, True))) + (atomicModifyIORef ref (\x -> (x + 2, 'x'))) + res <- readIORef ref + res @?= 3 + +case_replicateConcurrently :: Assertion +case_replicateConcurrently = do + ref <- newIORef 0 + let action = atomicModifyIORef ref (\x -> (x + 1, x + 1)) + resList <- replicateConcurrently 100 action + resVal <- readIORef ref + resVal @?= 100 + sort resList @?= [1..100] + +case_replicateConcurrently_ :: Assertion +case_replicateConcurrently_ = do + ref <- newIORef 0 + let action = atomicModifyIORef ref (\x -> (x + 1, x + 1)) + () <- replicateConcurrently_ 100 action + resVal <- readIORef ref + resVal @?= 100