diff --git a/Control/Concurrent/Async.hs b/Control/Concurrent/Async.hs --- a/Control/Concurrent/Async.hs +++ b/Control/Concurrent/Async.hs @@ -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 #) diff --git a/async.cabal b/async.cabal --- a/async.cabal +++ b/async.cabal @@ -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 diff --git a/bench/race.hs b/bench/race.hs --- a/bench/race.hs +++ b/bench/race.hs @@ -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)