async 2.0.0.0 → 2.0.1.0

raw patch · 3 files changed

+147/−29 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Control/Concurrent/Async.hs +133/−25
• async.cabal +13/−3
• bench/race.hs +1/−1

@@ -1,4 +1,4 @@-{-# LANGUAGE CPP, MagicHash, UnboxedTuples #-}+{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes #-} {-# OPTIONS -Wall #-}  -----------------------------------------------------------------------------@@ -67,14 +67,31 @@ -- -- >       (page1, page2) <- concurrently (getURL url1) (getURL url2) -- >       ...+--+-- 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)+-- > 4  -----------------------------------------------------------------------------  module Control.Concurrent.Async (      -- * Asynchronous actions-    Async, async, withAsync, asyncThreadId,+    Async,+    -- ** Spawning+    async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask,++    -- ** Spawning with automatic 'cancel'ation+    withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask, withAsyncOnWithUnmask,++    -- ** Quering 'Async's     wait, poll, waitCatch, cancel, cancelWith,+    asyncThreadId,      -- ** STM operations     waitSTM, pollSTM, waitCatchSTM,@@ -89,7 +106,9 @@     link, link2,      -- * Convenient utilities-    race, race_, concurrently,+    race, race_, concurrently, mapConcurrently,+    Concurrently(..),+   ) where  import Control.Concurrent.STM@@ -98,6 +117,7 @@ import Prelude hiding (catch) import Control.Monad import Control.Applicative+import Data.Traversable  import GHC.Exts import GHC.IO hiding (finally, onException)@@ -114,7 +134,7 @@ -- data Async a = Async { asyncThreadId :: {-# UNPACK #-} !ThreadId                        -- ^ Returns the 'ThreadId' of the thread running the given 'Async'.-                     , _asyncVar     :: {-# UNPACK #-} !(TMVar (Either SomeException a)) }+                     , _asyncWait    :: STM (Either SomeException a) }  instance Eq (Async a) where   Async a _ == Async b _  =  a == b@@ -122,15 +142,40 @@ instance Ord (Async a) where   Async a _ `compare` Async b _  =  a `compare` b +instance Functor Async where+  fmap f (Async a w) = Async a (fmap (fmap f) w)+ -- | Spawn an asynchronous action in a separate thread. async :: IO a -> IO (Async a)-async action = do+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 ->-          rawForkIO $ do r <- try (restore action); atomically $ putTMVar var r-   return (Async t var)+          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@@ -145,13 +190,35 @@ -- for details. -- 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:-withAsync action inner = do+withAsyncUsing doFork = \action inner -> do   var <- newEmptyTMVarIO   mask $ \restore -> do-    t <- rawForkIO $ try (restore action) >>= \r -> atomically $ putTMVar var r-    let a = Async t var+    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     return r@@ -190,29 +257,21 @@ -- | A version of 'wait' that can be used inside an STM transaction. -- waitSTM :: Async a -> STM a-waitSTM (Async _ var) = do-   r <- readTMVar var+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 _ var) = readTMVar var+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))-#if MIN_VERSION_stm(2,3,0)-pollSTM (Async _ var) = tryReadTMVar var-#else-pollSTM (Async _ var) = do-  r <- tryTakeTMVar var-  case r of-    Nothing -> return Nothing-    Just x  -> do putTMVar var x; return (Just x)-#endif+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@@ -347,10 +406,10 @@ -- the current thread. -- link :: Async a -> IO ()-link (Async _ var) = do+link (Async _ w) = do   me <- myThreadId   void $ forkRepeat $ do-     r <- atomically $ readTMVar var+     r <- atomically $ w      case r of        Left e -> throwTo me e        _ -> return ()@@ -368,7 +427,6 @@       _ -> return ()  - -- -----------------------------------------------------------------------------  -- | Run two @IO@ actions concurrently, and return the first to@@ -460,6 +518,51 @@  #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.+--+-- 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)++-- -----------------------------------------------------------------------------++-- | 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+ -- ----------------------------------------------------------------------------  -- | Fork a thread that runs the supplied action, and if it raises an@@ -487,3 +590,8 @@ rawForkIO :: IO () -> IO ThreadId rawForkIO action = IO $ \ s ->    case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)++{-# INLINE rawForkOn #-}+rawForkOn :: Int -> IO () -> IO ThreadId+rawForkOn (I# cpu) action = IO $ \ s ->+   case (forkOn# cpu action s) of (# s1, tid #) -> (# s1, ThreadId tid #)

@@ -14,15 +14,25 @@  .  * When waiting for a thread to return a result,    if the thread dies with an exception then the-   caller must either handle the exception-   ('wait') or re-throw it ('waitThrow'); the+   caller must either re-throw the exception+   ('wait') or handle it ('waitCatch'); the    exception cannot be ignored.  .  * The API makes it possible to build a tree of    threads that are automatically killed when    their parent dies (see 'withAsync').+ .+ Changes in 2.0.1.0:+ .+ * Added a @Functor@ instance for @Async@+ .+ * Added @asyncBound@, @asyncOn@, @asyncWithUnmask@, @asyncOnWithUnmask@, @withAsyncBound@, @withAsyncOn@, @withAsyncWithUnmask@, @withAsyncOnWithUnmask@.+ .+ * Added @mapConcurrently@+ .+ * Added @Concurrently@ (with @Applicative@ and @Alternative@ instances) -version:             2.0.0.0+version:             2.0.1.0 license:             BSD3 license-file:        LICENSE author:              Simon Marlow

@@ -5,4 +5,4 @@  main = runInUnboundThread $ do   [n] <- fmap (fmap read) getArgs-  replicateM_ n $ concurrently (return 1) (return 2)+  replicateM_ n $ race (return 1) (return 2)