Copyright | (c) Simon Marlow 2012 |
---|---|
License | BSD3 (see the file LICENSE) |
Maintainer | Simon Marlow <marlowsd@gmail.com> |
Stability | provisional |
Portability | non-portable (requires concurrency) |
Safe Haskell | Trustworthy |
Language | Haskell2010 |
This module provides a set of operations for running IO operations
asynchronously and waiting for their results. It is a thin layer
over the basic concurrency operations provided by
Control.Concurrent. The main additional functionality it
provides is the ability to wait for the return value of a thread,
but the interface also provides some additional safety and
robustness over using threads and MVar
directly.
The basic type is
, which represents an asynchronous
Async
aIO
action that will return a value of type a
, or die with an
exception. An Async
corresponds to a thread, and its ThreadId
can be obtained with asyncThreadId
, although that should rarely
be necessary.
For example, to fetch two web pages at the same time, we could do
this (assuming a suitable getURL
function):
do a1 <- async (getURL url1) a2 <- async (getURL url2) page1 <- wait a1 page2 <- wait a2 ...
where async
starts the operation in a separate thread, and
wait
waits for and returns the result. If the operation
throws an exception, then that exception is re-thrown by
wait
. This is one of the ways in which this library
provides some additional safety: it is harder to accidentally
forget about exceptions thrown in child threads.
A slight improvement over the previous example is this:
withAsync (getURL url1) $ \a1 -> do withAsync (getURL url2) $ \a2 -> do page1 <- wait a1 page2 <- wait a2 ...
withAsync
is like async
, except that the Async
is
automatically killed (using uninterruptibleCancel
) 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
reason.
The pattern of performing two IO actions concurrently and waiting
for their results is packaged up in a combinator concurrently
, so
we can further shorten the above example to:
(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
- data Async a
- async :: IO a -> IO (Async a)
- asyncBound :: IO a -> IO (Async a)
- asyncOn :: Int -> IO a -> IO (Async a)
- asyncWithUnmask :: ((forall b. IO b -> IO b) -> IO a) -> IO (Async a)
- asyncOnWithUnmask :: Int -> ((forall b. IO b -> IO b) -> IO a) -> IO (Async a)
- withAsync :: IO a -> (Async a -> IO b) -> IO b
- withAsyncBound :: IO a -> (Async a -> IO b) -> IO b
- withAsyncOn :: Int -> IO a -> (Async a -> IO b) -> IO b
- withAsyncWithUnmask :: ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b
- withAsyncOnWithUnmask :: Int -> ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b
- wait :: Async a -> IO a
- poll :: Async a -> IO (Maybe (Either SomeException a))
- waitCatch :: Async a -> IO (Either SomeException a)
- cancel :: Async a -> IO ()
- uninterruptibleCancel :: Async a -> IO ()
- cancelWith :: Exception e => Async a -> e -> IO ()
- asyncThreadId :: Async a -> ThreadId
- waitSTM :: Async a -> STM a
- pollSTM :: Async a -> STM (Maybe (Either SomeException a))
- waitCatchSTM :: Async a -> STM (Either SomeException a)
- waitAny :: [Async a] -> IO (Async a, a)
- waitAnyCatch :: [Async a] -> IO (Async a, Either SomeException a)
- waitAnyCancel :: [Async a] -> IO (Async a, a)
- waitAnyCatchCancel :: [Async a] -> IO (Async a, Either SomeException a)
- waitEither :: Async a -> Async b -> IO (Either a b)
- waitEitherCatch :: Async a -> Async b -> IO (Either (Either SomeException a) (Either SomeException b))
- waitEitherCancel :: Async a -> Async b -> IO (Either a b)
- waitEitherCatchCancel :: Async a -> Async b -> IO (Either (Either SomeException a) (Either SomeException b))
- waitEither_ :: Async a -> Async b -> IO ()
- waitBoth :: Async a -> Async b -> IO (a, b)
- waitAnySTM :: [Async a] -> STM (Async a, a)
- waitAnyCatchSTM :: [Async a] -> STM (Async a, Either SomeException a)
- waitEitherSTM :: Async a -> Async b -> STM (Either a b)
- waitEitherCatchSTM :: Async a -> Async b -> STM (Either (Either SomeException a) (Either SomeException b))
- waitEitherSTM_ :: Async a -> Async b -> STM ()
- waitBothSTM :: Async a -> Async b -> STM (a, b)
- link :: Async a -> IO ()
- link2 :: Async a -> Async b -> IO ()
- race :: IO a -> IO b -> IO (Either a b)
- race_ :: IO a -> IO b -> IO ()
- concurrently :: IO a -> IO b -> IO (a, b)
- concurrently_ :: IO a -> IO b -> IO ()
- mapConcurrently :: Traversable t => (a -> IO b) -> t a -> IO (t b)
- forConcurrently :: Traversable t => t a -> (a -> IO b) -> IO (t b)
- mapConcurrently_ :: Foldable f => (a -> IO b) -> f a -> IO ()
- forConcurrently_ :: Foldable f => f a -> (a -> IO b) -> IO ()
- replicateConcurrently :: Int -> IO a -> IO [a]
- replicateConcurrently_ :: Int -> IO a -> IO ()
- newtype Concurrently a = Concurrently {
- runConcurrently :: IO a
Asynchronous actions
Spawning
asyncWithUnmask :: ((forall b. IO b -> IO b) -> IO a) -> IO (Async a) #
Like async
but using forkIOWithUnmask
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) #
Like asyncOn
but using forkOnWithUnmask
internally. The
child thread is passed a function that can be used to unmask
asynchronous exceptions.
Spawning with automatic cancel
ation
withAsync :: IO a -> (Async a -> IO b) -> IO b #
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, uninterruptibleCancel
is called on the Async
.
withAsync action inner = bracket (async action) uninterruptibleCancel inner
This is a useful variant of async
that ensures an Async
is
never left running unintentionally.
withAsyncWithUnmask :: ((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b #
Like withAsync
but uses forkIOWithUnmask
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 #
Like withAsyncOn
but uses forkOnWithUnmask
internally. The
child thread is passed a function that can be used to unmask
asynchronous exceptions
Querying Async
s
Wait for an asynchronous action to complete, and return its
value. If the asynchronous action threw an exception, then the
exception is re-thrown by wait
.
wait = atomically . waitSTM
poll :: Async a -> IO (Maybe (Either SomeException a)) #
Check whether an Async
has completed yet. If it has not
completed yet, then the result is Nothing
, otherwise the result
is Just e
where e
is Left x
if the Async
raised an
exception x
, or Right a
if it returned a value a
.
poll = atomically . pollSTM
waitCatch :: Async a -> IO (Either SomeException a) #
Wait for an asynchronous action to complete, and return either
Left e
if the action raised an exception e
, or Right a
if it
returned a value a
.
waitCatch = atomically . waitCatchSTM
Cancel an asynchronous action by throwing the ThreadKilled
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
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.
uninterruptibleCancel :: Async a -> IO () #
Cancel an asynchronous action
This is a variant of cancel
, but it is not interruptible.
cancelWith :: Exception e => Async a -> e -> IO () #
Cancel an asynchronous action by throwing the supplied exception to it.
cancelWith a x = throwTo (asyncThreadId a) x
The notes about the synchronous nature of cancel
also apply to
cancelWith
.
STM operations
pollSTM :: Async a -> STM (Maybe (Either SomeException a)) #
A version of poll
that can be used inside an STM transaction.
waitCatchSTM :: Async a -> STM (Either SomeException a) #
A version of waitCatch
that can be used inside an STM transaction.
Waiting for multiple Async
s
waitAnyCatch :: [Async a] -> IO (Async a, Either SomeException a) #
Wait for any of the supplied asynchronous operations to complete.
The value returned is a pair of the Async
that completed, and the
result that would be returned by wait
on that Async
.
If multiple Async
s complete or have completed, then the value
returned corresponds to the first completed Async
in the list.
waitAnyCancel :: [Async a] -> IO (Async a, a) #
Like waitAny
, but also cancels the other asynchronous
operations as soon as one has completed.
waitAnyCatchCancel :: [Async a] -> IO (Async a, Either SomeException a) #
Like waitAnyCatch
, but also cancels the other asynchronous
operations as soon as one has completed.
waitEither :: Async a -> Async b -> IO (Either a b) #
Wait for the first of two Async
s to finish. If the Async
that finished first raised an exception, then the exception is
re-thrown by waitEither
.
waitEitherCatch :: Async a -> Async b -> IO (Either (Either SomeException a) (Either SomeException b)) #
Wait for the first of two Async
s to finish.
waitEitherCancel :: Async a -> Async b -> IO (Either a b) #
Like waitEither
, but also cancel
s both Async
s before
returning.
waitEitherCatchCancel :: Async a -> Async b -> IO (Either (Either SomeException a) (Either SomeException b)) #
Like waitEitherCatch
, but also cancel
s both Async
s before
returning.
waitEither_ :: Async a -> Async b -> IO () #
Like waitEither
, but the result is ignored.
waitBoth :: Async a -> Async b -> IO (a, b) #
Waits for both Async
s to finish, but if either of them throws
an exception before they have both finished, then the exception is
re-thrown by waitBoth
.
Waiting for multiple Async
s in STM
waitAnySTM :: [Async a] -> STM (Async a, a) #
A version of waitAny
that can be used inside an STM transaction.
Since: 2.1.0
waitAnyCatchSTM :: [Async a] -> STM (Async a, Either SomeException a) #
A version of waitAnyCatch
that can be used inside an STM transaction.
Since: 2.1.0
waitEitherSTM :: Async a -> Async b -> STM (Either a b) #
A version of waitEither
that can be used inside an STM transaction.
Since: 2.1.0
waitEitherCatchSTM :: Async a -> Async b -> STM (Either (Either SomeException a) (Either SomeException b)) #
A version of waitEitherCatch
that can be used inside an STM transaction.
Since: 2.1.0
waitEitherSTM_ :: Async a -> Async b -> STM () #
A version of waitEither_
that can be used inside an STM transaction.
Since: 2.1.0
waitBothSTM :: Async a -> Async b -> STM (a, b) #
A version of waitBoth
that can be used inside an STM transaction.
Since: 2.1.0
Linking
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.
link2 :: Async a -> Async b -> IO () #
Link two Async
s together, such that if either raises an
exception, the same exception is re-thrown in the other Async
.
Convenient utilities
race :: IO a -> IO b -> IO (Either a b) #
Run two IO
actions concurrently, and return the first to
finish. The loser of the race is cancel
led.
race left right = withAsync left $ \a -> withAsync right $ \b -> waitEither a b
concurrently :: IO a -> IO b -> IO (a, b) #
Run two IO
actions concurrently, and return both results. If
either action throws an exception at any time, then the other
action is cancel
led, and the exception is re-thrown by
concurrently
.
concurrently left right = withAsync left $ \a -> withAsync right $ \b -> waitBoth a b
concurrently_ :: IO a -> IO b -> IO () #
concurrently
, but ignore the result values
Since: 2.1.1
mapConcurrently :: Traversable t => (a -> IO b) -> t a -> IO (t b) #
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"]
forConcurrently :: Traversable t => t a -> (a -> IO b) -> IO (t b) #
forConcurrently
is mapConcurrently
with its arguments flipped
pages <- forConcurrently ["url1", "url2", "url3"] $ \url -> getURL url
Since: 2.1.0
mapConcurrently_ :: Foldable f => (a -> IO b) -> f a -> IO () #
mapConcurrently_
is mapConcurrently
with the return value discarded,
just like @mapM_
forConcurrently_ :: Foldable f => f a -> (a -> IO b) -> IO () #
forConcurrently_
is forConcurrently
with the return value discarded,
just like @forM_
replicateConcurrently :: Int -> IO a -> IO [a] #
Perform the action in the given number of threads.
Since: 2.1.1
replicateConcurrently_ :: Int -> IO a -> IO () #
Same as replicateConcurrently
, but ignore the results.
Since: 2.1.1
newtype Concurrently a #
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")
Functor Concurrently # | |
Applicative Concurrently # | |
Alternative Concurrently # | |
Semigroup a => Semigroup (Concurrently a) # | Only defined by Since: 2.1.0 |
(Semigroup a, Monoid a) => Monoid (Concurrently a) # | Since: 2.1.0 |