async 2.1.0 → 2.1.1

raw patch · 4 files changed

+264/−54 lines, 4 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Control/Concurrent/Async.hs +130/−53
• async.cabal +1/−1
• changelog.md +12/−0
• test/test-async.hs +121/−0

@@ -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  -- ----------------------------------------------------------------------------- 

@@ -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 

@@ -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

@@ -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