async 2.1.1.1 → 2.2.1

raw patch · 6 files changed

+328/−35 lines, 6 filesdep +hashabledep ~basedep ~stmnew-component:exe:concasyncnew-component:exe:conccancelnew-component:exe:racePVP ok

version bump matches the API change (PVP)

Dependencies added: hashable

Dependency ranges changed: base, stm

API changes (from Hackage documentation)

Files

• Control/Concurrent/Async.hs +143/−30
• async.cabal +37/−4
• bench/concasync.hs +13/−0
• bench/conccancel.hs +11/−0
• changelog.md +22/−0
• test/test-async.hs +102/−1

@@ -1,7 +1,11 @@-{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes #-}+{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes,+    ExistentialQuantification #-} #if __GLASGOW_HASKELL__ >= 701 {-# LANGUAGE Trustworthy #-} #endif+#if __GLASGOW_HASKELL__ < 710+{-# LANGUAGE DeriveDataTypeable #-}+#endif {-# OPTIONS -Wall #-}  -----------------------------------------------------------------------------@@ -94,8 +98,8 @@     withAsyncOnWithUnmask,      -- ** Querying 'Async's-    wait, poll, waitCatch, cancel, uninterruptibleCancel, cancelWith,-    asyncThreadId,+    wait, poll, waitCatch, asyncThreadId,+    cancel, uninterruptibleCancel, cancelWith, AsyncCancelled(..),      -- ** STM operations     waitSTM, pollSTM, waitCatchSTM,@@ -113,7 +117,7 @@     waitBothSTM,      -- ** Linking-    link, link2,+    link, link2, ExceptionInLinkedThread(..),      -- * Convenient utilities     race, race_,@@ -122,6 +126,7 @@     mapConcurrently_, forConcurrently_,     replicateConcurrently, replicateConcurrently_,     Concurrently(..),+    compareAsyncs,    ) where @@ -138,9 +143,13 @@ 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 @@ -170,9 +179,16 @@ 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@@ -212,11 +228,17 @@ -- @Async@ handle to the supplied function.  When the function returns -- or throws an exception, 'uninterruptibleCancel' is called on the @Async@. ----- > withAsync action inner = bracket (async action) uninterruptibleCancel inner+-- > 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 @@ -312,11 +334,11 @@ pollSTM :: Async a -> STM (Maybe (Either SomeException a)) pollSTM (Async _ w) = (Just <$> w) `orElse` return Nothing --- | Cancel an asynchronous action by throwing the @ThreadKilled@+-- | 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) ThreadKilled <* waitCatch a+-- > 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@@ -330,8 +352,22 @@ -- and the handler is blocking. {-# INLINE cancel #-} cancel :: Async a -> IO ()-cancel a@(Async t _) = throwTo t ThreadKilled <* waitCatch a+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.@@ -406,7 +442,11 @@ waitEitherCatch :: Async a -> Async b                 -> IO (Either (Either SomeException a)                               (Either SomeException b))-waitEitherCatch left right = atomically (waitEitherCatchSTM left right)+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. --@@ -487,32 +527,84 @@     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.+-- 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 (Async _ w) = do+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 <- atomically $ w-     case r of-       Left e -> throwTo me e-       _ -> return ()+    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@.+-- 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 left@(Async tl _)  right@(Async tr _) =+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) -> throwTo tr e-      Right (Left e) -> throwTo tl e+      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@@ -593,29 +685,47 @@ concurrently' left right collect = do     done <- newEmptyMVar     mask $ \restore -> do-        lid <- forkIO $ restore (left >>= putMVar done . Right . Left)-                             `catchAll` (putMVar done . Left)-        rid <- forkIO $ restore (right >>= putMVar done . Right . Right)-                             `catchAll` (putMVar done . Left)+        -- 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 -                takeMVar done+        let tryAgain f = f `catch` \BlockedIndefinitelyOnMVar -> f -        let stop = do+            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+                  -- 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 @@ -626,6 +736,9 @@ -- 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"]@@ -638,7 +751,7 @@ -- > pages <- forConcurrently ["url1", "url2", "url3"] $ \url -> getURL url -- -- @since 2.1.0-forConcurrently :: Traversable t => t a -> (a -> IO b)-> IO (t b)+forConcurrently :: Traversable t => t a -> (a -> IO b) -> IO (t b) forConcurrently = flip mapConcurrently  -- | `mapConcurrently_` is `mapConcurrently` with the return value discarded,

@@ -1,5 +1,5 @@ name:                async-version:             2.1.1.1+version:             2.2.1 -- don't forget to update ./changelog.md! synopsis:            Run IO operations asynchronously and wait for their results @@ -34,7 +34,7 @@ cabal-version:       >=1.10 homepage:            https://github.com/simonmar/async bug-reports:         https://github.com/simonmar/async/issues-tested-with:         GHC==7.11.*, 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==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  extra-source-files:     changelog.md@@ -50,15 +50,48 @@     if impl(ghc>=7.1)         other-extensions: Trustworthy     exposed-modules:     Control.Concurrent.Async-    build-depends:       base >= 4.3 && < 4.11, stm >= 2.2 && < 2.5+    build-depends:       base >= 4.3 && < 4.12, hashable >= 1.1.1.0 && < 1.3, stm >= 2.2 && < 2.5  test-suite test-async     default-language: Haskell2010     type:       exitcode-stdio-1.0     hs-source-dirs: test     main-is:    test-async.hs-    build-depends: base >= 4.3 && < 4.11,+    build-depends: base >= 4.3 && < 4.12,                    async,+                   stm,                    test-framework,                    test-framework-hunit,                    HUnit++flag bench+    default: False++executable concasync+    if !flag(bench)+       buildable: False+    default-language: Haskell2010+    hs-source-dirs: bench+    main-is:    concasync.hs+    build-depends: base, async, stm+    ghc-options: -O2++executable conccancel+    if !flag(bench)+       buildable: False+    default-language: Haskell2010+    hs-source-dirs: bench+    main-is:    conccancel.hs+    build-depends: base, async, stm+    ghc-options: -O2 -threaded++executable race+    if !flag(bench)+       buildable: False+    default-language: Haskell2010+    hs-source-dirs: bench+    main-is:    race.hs+    build-depends: base, async, stm+    ghc-options: -O2 -threaded++

@@ -0,0 +1,13 @@+import Control.Concurrent.Async+import System.Environment+import Control.Monad+import Control.Concurrent++main = runInUnboundThread $ do+  [n] <- fmap (fmap read) getArgs+  replicateM_ n $ concurrently (return 1) (return 2)++concurrently' left right =+  withAsync left $ \a ->+  withAsync right $ \b ->+  waitBoth a b

@@ -0,0 +1,11 @@+import Control.Exception+import Control.Concurrent.Async+import System.Environment+import Control.Monad+import Control.Concurrent++main = runInUnboundThread $ do+  [n] <- fmap (fmap read) getArgs+  runConcurrently $ traverse Concurrently $+    replicate n (threadDelay 1000000) ++ [throwIO (ErrorCall "oops")]+

@@ -1,3 +1,25 @@+## Changes in 2.2.1:++ - Add a Hashable instance for Async+ - Bump upper bounds+ - Documentation updates++## Changes in 2.2:+ - cancel now throws AsyncCancelled instead of ThreadKilled+ - link and link2 now wrap exceptions in ExceptionInLinkedThread when+   throwing to the linked thread. ExceptionInLinkedThread is a child+   of AsyncException in the exception hierarchy, so this maintains the+   invariant that exceptions thrown asynchronously should be+   AsyncExceptions.+ - link and link2 do not propagate AsyncCancelled, so it's now+   possible to cancel a linked thread without cancelling yourself.+ - Added linkOnly and link2Only to specify which exceptions should be+   propagated,if you want something other than the default behaviour+   of ignoring AsyncCancelled.+ - new utility function compareAsyncs for comparing Asyncs of+   different types.+ - Add a `Hashable` instance for `Async a`+ ## Changes in 2.1.1.1:  - Make 'cancelWith' wait for the cancelled thread to terminate, like 'cancel'  - Updates to dependency bounds for GHC 8.2

@@ -6,6 +6,7 @@  import Test.HUnit +import Control.Concurrent.STM import Control.Concurrent.Async import Control.Exception import Data.IORef@@ -29,6 +30,7 @@   , testGroup "async_cancel_rep" $       replicate 1000 $          testCase "async_cancel"       async_cancel+  , testCase "async_cancelmany" async_cancelmany   , testCase "async_poll"        async_poll   , testCase "async_poll2"       async_poll2   , testCase "withasync_waitCatch_blocked" withasync_waitCatch_blocked@@ -43,6 +45,10 @@   , testCase "concurrently_" case_concurrently_   , testCase "replicateConcurrently_" case_replicateConcurrently   , testCase "replicateConcurrently" case_replicateConcurrently_+  , testCase "link1" case_link1+  , testCase "link2" case_link2+  , testCase "link1_cancel" case_link1cancel+  , testCase "concurrently_deadlock" case_concurrently_deadlock  ]  value = 42 :: Int@@ -90,7 +96,7 @@   a <- withAsync (threadDelay 1000000) $ return   r <- waitCatch a   case r of-    Left e  -> fromException e @?= Just ThreadKilled+    Left e  -> fromException e @?= Just AsyncCancelled     Right _ -> assertFailure ""  async_cancel :: Assertion@@ -102,6 +108,18 @@     Left e -> fromException e @?= Just TestException     Right r -> r @?= value +async_cancelmany :: Assertion -- issue 59+async_cancelmany = do+  r <- newIORef []+  a <- async $ forConcurrently_ ['a'..'z'] $ \c ->+    delay 2 `finally` atomicModifyIORef r (\i -> (c:i,()))+  delay 1+  cancel a+  v <- readIORef r+  assertEqual "cancelmany" 26 (length v)+  where+    delay sec = threadDelay (sec * 1000000)+ async_poll :: Assertion async_poll = do   a <- async (threadDelay 1000000)@@ -236,3 +254,86 @@   () <- replicateConcurrently_ 100 action   resVal <- readIORef ref   resVal @?= 100++case_link1 :: Assertion+case_link1 = do+  m1 <- newEmptyMVar+  m2 <- newEmptyMVar+  let ex = ErrorCall "oops"+  a <- async $ do takeMVar m1; throwIO ex; putMVar m2 ()+  link a+  e <- try $ (do+    putMVar m1 ()+    takeMVar m2)+  assertBool "link1" $+    case e of+      Left (ExceptionInLinkedThread a' e') ->+        compareAsyncs a' a == EQ &&+          case fromException e' of+            Just (ErrorCall s) -> s == "oops"+            _otherwise -> False+      _other -> False++case_link2 :: Assertion+case_link2 = do+  let+    setup = do+      m1 <- newEmptyMVar+      m2 <- newEmptyMVar+      let ex1 = ErrorCall "oops1"; ex2 = ErrorCall "oops2"+      a <- async $ do takeMVar m1; throwIO ex1+      b <- async $ do takeMVar m2; throwIO ex2+      link2 a b+      return (m1,m2,a,b)++  (m1,m2,a,b) <- setup+  e <- try $ do+    putMVar m1 ()+    wait b+  putMVar m2 ()  -- ensure the other thread is not deadlocked+  assertBool "link2a" $+    case e of+      Left (ExceptionInLinkedThread a' e') ->+        compareAsyncs a' a == EQ &&+          case fromException e' of+            Just (ErrorCall s) -> s == "oops1"+            _otherwise -> False+      _other -> False++  (m1,m2,a,b) <- setup+  e <- try $ do+    putMVar m2 ()+    wait a+  putMVar m1 ()  -- ensure the other thread is not deadlocked+  assertBool "link2b" $+    case e of+      Left (ExceptionInLinkedThread a' e') ->+        compareAsyncs a' b == EQ &&+          case fromException e' of+            Just (ErrorCall s) -> s == "oops2"+            _otherwise -> False+      _other -> False++case_link1cancel :: Assertion+case_link1cancel = do+  m1 <- newEmptyMVar+  let ex = ErrorCall "oops"+  a <- async $ do takeMVar m1+  link a+  e <- try $ do cancel a; wait a+  putMVar m1 ()+  assertBool "link1cancel" $+    case e of+      Left AsyncCancelled -> True  -- should not be ExceptionInLinkedThread+      _other -> False++-- See Issue #62+case_concurrently_deadlock :: Assertion+case_concurrently_deadlock = do+  tvar <- newTVarIO False :: IO (TVar Bool)+  e <- try $ void $ join (concurrently) (atomically $ readTVar tvar >>= check)+    -- should throw BlockedIndefinitelyOnSTM not BlockedIndefinitelyOnMVar+  assertBool "concurrently_deadlock" $+    case e of+      Left BlockedIndefinitelyOnSTM{} -> True+      _other -> False