{-# LANGUAGE CPP, MagicHash, UnboxedTuples, RankNTypes,
 ExistentialQuantification #-}
#if __GLASGOW_HASKELL__ >= 701
{-# LANGUAGE Trustworthy #-}
#endif
#if __GLASGOW_HASKELL__ < 710
{-# LANGUAGE DeriveDataTypeable #-}
#endif
{-# OPTIONS -Wall #-}------------------------------------------------------------------------------- |-- Module : Control.Concurrent.Async.Internal-- Copyright : (c) Simon Marlow 2012-- License : BSD3 (see the file LICENSE)---- Maintainer : Simon Marlow <marlowsd@gmail.com>-- Stability : provisional-- Portability : non-portable (requires concurrency)---- This module is an internal module. The public API is provided in-- "Control.Concurrent.Async". Breaking changes to this module will not be-- reflected in a major bump, and using this module may break your code-- unless you are extremely careful.-------------------------------------------------------------------------------moduleControl.Concurrent.Async.InternalwhereimportControl.Concurrent.STMimportControl.ExceptionimportControl.ConcurrentimportqualifiedData.FoldableasF
#if !MIN_VERSION_base(4,6,0)
importPreludehiding(catch)
#endif
importControl.MonadimportControl.Applicative
#if !MIN_VERSION_base(4,8,0)
importData.Monoid(Monoid(mempty,mappend))importData.Traversable
#endif
#if __GLASGOW_HASKELL__ < 710
importData.Typeable
#endif
#if MIN_VERSION_base(4,8,0)
importData.Bifunctor
#endif
#if MIN_VERSION_base(4,9,0)
importData.Semigroup(Semigroup((<>)))
#endif
importData.Hashable(Hashable(hashWithSalt))importData.IORefimportGHC.ExtsimportGHC.IOhiding(finally,onException)importGHC.Conc-- ------------------------------------------------------------------------------- STM Async API-- | An asynchronous action spawned by 'async' or 'withAsync'.-- Asynchronous actions are executed in a separate thread, and-- operations are provided for waiting for asynchronous actions to-- complete and obtaining their results (see e.g. 'wait').--dataAsync a =Async {forall a. Async a -> ThreadId
asyncThreadId ::{-# UNPACK#-}!ThreadId-- ^ Returns the 'ThreadId' of the thread running-- the given 'Async'.,forall a. Async a -> STM (Either SomeException a)
_asyncWait ::STM(EitherSomeExceptiona )}instanceEq(Async a )whereAsync ThreadId
a STM (Either SomeException a)
_== :: Async a -> Async a -> Bool
==Async ThreadId
b STM (Either SomeException a)
_=ThreadId
a forall a. Eq a => a -> a -> Bool
==ThreadId
b instanceOrd(Async a )whereAsync ThreadId
a STM (Either SomeException a)
_compare :: Async a -> Async a -> Ordering
`compare`Async ThreadId
b STM (Either SomeException a)
_=ThreadId
a forall a. Ord a => a -> a -> Ordering
`compare`ThreadId
b instanceHashable(Async a )wherehashWithSalt :: Int -> Async a -> Int
hashWithSalt Int
salt (Async ThreadId
a STM (Either SomeException a)
_)=forall a. Hashable a => Int -> a -> Int
hashWithSaltInt
salt ThreadId
a instanceFunctorAsync wherefmap :: forall a b. (a -> b) -> Async a -> Async b
fmapa -> b
f (Async ThreadId
a STM (Either SomeException a)
w )=forall a. ThreadId -> STM (Either SomeException a) -> Async a
Async ThreadId
a (forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap(forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmapa -> b
f )STM (Either SomeException a)
w )-- | Compare two Asyncs that may have different types by their 'ThreadId'.compareAsyncs ::Async a ->Async b ->OrderingcompareAsyncs :: forall a b. Async a -> Async b -> Ordering
compareAsyncs (Async ThreadId
t1 STM (Either SomeException a)
_)(Async ThreadId
t2 STM (Either SomeException b)
_)=forall a. Ord a => a -> a -> Ordering
compareThreadId
t1 ThreadId
t2 -- | Spawn an asynchronous action in a separate thread.---- Like for 'forkIO', the action may be left running unintentionally-- (see module-level documentation for details).---- __Use 'withAsync' style functions wherever you can instead!__async ::IOa ->IO(Async a )async :: forall a. IO a -> IO (Async a)
async =forall a. a -> a
inlineforall a. (IO () -> IO ThreadId) -> IO a -> IO (Async a)
asyncUsing IO () -> IO ThreadId
rawForkIO -- | Like 'async' but using 'forkOS' internally.asyncBound ::IOa ->IO(Async a )asyncBound :: forall a. IO a -> IO (Async a)
asyncBound =forall a. (IO () -> IO ThreadId) -> IO a -> IO (Async a)
asyncUsing IO () -> IO ThreadId
forkOS-- | Like 'async' but using 'forkOn' internally.asyncOn ::Int->IOa ->IO(Async a )asyncOn :: forall a. Int -> IO a -> IO (Async a)
asyncOn =forall a. (IO () -> IO ThreadId) -> IO a -> IO (Async a)
asyncUsing forall b c a. (b -> c) -> (a -> b) -> a -> c
.Int -> IO () -> IO ThreadId
rawForkOn -- | Like 'async' but using 'forkIOWithUnmask' internally. The child-- thread is passed a function that can be used to unmask asynchronous-- exceptions.asyncWithUnmask ::((forallb .IOb ->IOb )->IOa )->IO(Async a )asyncWithUnmask :: forall a. ((forall b. IO b -> IO b) -> IO a) -> IO (Async a)
asyncWithUnmask (forall b. IO b -> IO b) -> IO a
actionWith =forall a. (IO () -> IO ThreadId) -> IO a -> IO (Async a)
asyncUsing IO () -> IO ThreadId
rawForkIO ((forall b. IO b -> IO b) -> IO a
actionWith forall b. IO b -> IO b
unsafeUnmask)-- | Like 'asyncOn' but using 'forkOnWithUnmask' internally. The-- child thread is passed a function that can be used to unmask-- asynchronous exceptions.asyncOnWithUnmask ::Int->((forallb .IOb ->IOb )->IOa )->IO(Async a )asyncOnWithUnmask :: forall a. Int -> ((forall b. IO b -> IO b) -> IO a) -> IO (Async a)
asyncOnWithUnmask Int
cpu (forall b. IO b -> IO b) -> IO a
actionWith =forall a. (IO () -> IO ThreadId) -> IO a -> IO (Async a)
asyncUsing (Int -> IO () -> IO ThreadId
rawForkOn Int
cpu )((forall b. IO b -> IO b) -> IO a
actionWith forall b. IO b -> IO b
unsafeUnmask)asyncUsing ::(IO()->IOThreadId)->IOa ->IO(Async a )asyncUsing :: forall a. (IO () -> IO ThreadId) -> IO a -> IO (Async a)
asyncUsing IO () -> IO ThreadId
doFork =\IO a
action ->doTMVar (Either SomeException a)
var <-forall a. IO (TMVar a)
newEmptyTMVarIO-- t <- forkFinally action (\r -> atomically $ putTMVar var r)-- slightly faster:ThreadId
t <-forall b. ((forall b. IO b -> IO b) -> IO b) -> IO b
maskforall a b. (a -> b) -> a -> b
$\forall b. IO b -> IO b
restore ->IO () -> IO ThreadId
doFork forall a b. (a -> b) -> a -> b
$forall e a. Exception e => IO a -> IO (Either e a)
try(forall b. IO b -> IO b
restore IO a
action )forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>=forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. TMVar a -> a -> STM ()
putTMVarTMVar (Either SomeException a)
var forall (m :: * -> *) a. Monad m => a -> m a
return(forall a. ThreadId -> STM (Either SomeException a) -> Async a
Async ThreadId
t (forall a. TMVar a -> STM a
readTMVarTMVar (Either SomeException a)
var ))-- | 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 = mask $ \restore -> do-- > a <- async (restore action)-- > restore (inner a) `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 ::IOa ->(Async a ->IOb )->IOb withAsync :: forall a b. IO a -> (Async a -> IO b) -> IO b
withAsync =forall a. a -> a
inlineforall a b.
(IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b
withAsyncUsing IO () -> IO ThreadId
rawForkIO -- | Like 'withAsync' but uses 'forkOS' internally.withAsyncBound ::IOa ->(Async a ->IOb )->IOb withAsyncBound :: forall a b. IO a -> (Async a -> IO b) -> IO b
withAsyncBound =forall a b.
(IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b
withAsyncUsing IO () -> IO ThreadId
forkOS-- | Like 'withAsync' but uses 'forkOn' internally.withAsyncOn ::Int->IOa ->(Async a ->IOb )->IOb withAsyncOn :: forall a b. Int -> IO a -> (Async a -> IO b) -> IO b
withAsyncOn =forall a b.
(IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b
withAsyncUsing forall b c a. (b -> c) -> (a -> b) -> a -> c
.Int -> IO () -> IO ThreadId
rawForkOn -- | Like 'withAsync' but uses 'forkIOWithUnmask' internally. The-- child thread is passed a function that can be used to unmask-- asynchronous exceptions.withAsyncWithUnmask ::((forallc .IOc ->IOc )->IOa )->(Async a ->IOb )->IOb withAsyncWithUnmask :: forall a b.
((forall b. IO b -> IO b) -> IO a) -> (Async a -> IO b) -> IO b
withAsyncWithUnmask (forall b. IO b -> IO b) -> IO a
actionWith =forall a b.
(IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b
withAsyncUsing IO () -> IO ThreadId
rawForkIO ((forall b. IO b -> IO b) -> IO a
actionWith forall b. IO b -> IO b
unsafeUnmask)-- | Like 'withAsyncOn' but uses 'forkOnWithUnmask' internally. The-- child thread is passed a function that can be used to unmask-- asynchronous exceptionswithAsyncOnWithUnmask ::Int->((forallc .IOc ->IOc )->IOa )->(Async a ->IOb )->IOb withAsyncOnWithUnmask :: forall a b.
Int
-> ((forall b. IO b -> IO b) -> IO a) -> (Async a -> IO b) -> IO b
withAsyncOnWithUnmask Int
cpu (forall b. IO b -> IO b) -> IO a
actionWith =forall a b.
(IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b
withAsyncUsing (Int -> IO () -> IO ThreadId
rawForkOn Int
cpu )((forall b. IO b -> IO b) -> IO a
actionWith forall b. IO b -> IO b
unsafeUnmask)withAsyncUsing ::(IO()->IOThreadId)->IOa ->(Async a ->IOb )->IOb -- The bracket version works, but is slow. We can do better by-- hand-coding it:withAsyncUsing :: forall a b.
(IO () -> IO ThreadId) -> IO a -> (Async a -> IO b) -> IO b
withAsyncUsing IO () -> IO ThreadId
doFork =\IO a
action Async a -> IO b
inner ->doTMVar (Either SomeException a)
var <-forall a. IO (TMVar a)
newEmptyTMVarIOforall b. ((forall b. IO b -> IO b) -> IO b) -> IO b
maskforall a b. (a -> b) -> a -> b
$\forall b. IO b -> IO b
restore ->doThreadId
t <-IO () -> IO ThreadId
doFork forall a b. (a -> b) -> a -> b
$forall e a. Exception e => IO a -> IO (Either e a)
try(forall b. IO b -> IO b
restore IO a
action )forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>=forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. TMVar a -> a -> STM ()
putTMVarTMVar (Either SomeException a)
var leta :: Async a
a =forall a. ThreadId -> STM (Either SomeException a) -> Async a
Async ThreadId
t (forall a. TMVar a -> STM a
readTMVarTMVar (Either SomeException a)
var )b
r <-forall b. IO b -> IO b
restore (Async a -> IO b
inner Async a
a )forall a. IO a -> (SomeException -> IO a) -> IO a
`catchAll` \SomeException
e ->doforall a. Async a -> IO ()
uninterruptibleCancel Async a
a forall e a. Exception e => e -> IO a
throwIOSomeException
e forall a. Async a -> IO ()
uninterruptibleCancel Async a
a forall (m :: * -> *) a. Monad m => a -> m a
returnb
r -- | 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--{-# INLINEwait #-}wait ::Async a ->IOa wait :: forall a. Async a -> IO a
wait =forall b. IO b -> IO b
tryAgain forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Async a -> STM a
waitSTM where-- See: https://github.com/simonmar/async/issues/14tryAgain :: IO a -> IO a
tryAgain IO a
f =IO a
f forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch`\BlockedIndefinitelyOnSTM
BlockedIndefinitelyOnSTM->IO a
f -- | 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--{-# INLINEwaitCatch #-}waitCatch ::Async a ->IO(EitherSomeExceptiona )waitCatch :: forall a. Async a -> IO (Either SomeException a)
waitCatch =forall b. IO b -> IO b
tryAgain forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Async a -> STM (Either SomeException a)
waitCatchSTM where-- See: https://github.com/simonmar/async/issues/14tryAgain :: IO a -> IO a
tryAgain IO a
f =IO a
f forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch`\BlockedIndefinitelyOnSTM
BlockedIndefinitelyOnSTM->IO a
f -- | 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--{-# INLINEpoll #-}poll ::Async a ->IO(Maybe(EitherSomeExceptiona ))poll :: forall a. Async a -> IO (Maybe (Either SomeException a))
poll =forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Async a -> STM (Maybe (Either SomeException a))
pollSTM -- | A version of 'wait' that can be used inside an STM transaction.--waitSTM ::Async a ->STMa waitSTM :: forall a. Async a -> STM a
waitSTM Async a
a =doEither SomeException a
r <-forall a. Async a -> STM (Either SomeException a)
waitCatchSTM Async a
a forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
eitherforall e a. Exception e => e -> STM a
throwSTMforall (m :: * -> *) a. Monad m => a -> m a
returnEither SomeException a
r -- | A version of 'waitCatch' that can be used inside an STM transaction.--{-# INLINEwaitCatchSTM #-}waitCatchSTM ::Async a ->STM(EitherSomeExceptiona )waitCatchSTM :: forall a. Async a -> STM (Either SomeException a)
waitCatchSTM (Async ThreadId
_STM (Either SomeException a)
w )=STM (Either SomeException a)
w -- | A version of 'poll' that can be used inside an STM transaction.--{-# INLINEpollSTM #-}pollSTM ::Async a ->STM(Maybe(EitherSomeExceptiona ))pollSTM :: forall a. Async a -> STM (Maybe (Either SomeException a))
pollSTM (Async ThreadId
_STM (Either SomeException a)
w )=(forall a. a -> Maybe a
Justforall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>STM (Either SomeException a)
w )forall a. STM a -> STM a -> STM a
`orElse`forall (m :: * -> *) a. Monad m => a -> m a
returnforall a. Maybe a
Nothing-- | 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) 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-- 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.{-# INLINEcancel #-}cancel ::Async a ->IO()cancel :: forall a. Async a -> IO ()
cancel a :: Async a
a @(Async ThreadId
t STM (Either SomeException a)
_)=forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
t AsyncCancelled
AsyncCancelled forall (f :: * -> *) a b. Applicative f => f a -> f b -> f a
<*forall a. Async a -> IO (Either SomeException a)
waitCatch Async a
a -- | Cancel multiple asynchronous actions by throwing the @AsyncCancelled@-- exception to each of them in turn, then waiting for all the `Async` threads-- to complete.cancelMany ::[Async a ]->IO()cancelMany :: forall a. [Async a] -> IO ()
cancelMany [Async a]
as =doforall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_(\(Async ThreadId
t STM (Either SomeException a)
_)->forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
t AsyncCancelled
AsyncCancelled )[Async a]
as forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_forall a. Async a -> IO (Either SomeException a)
waitCatch [Async a]
as -- | The exception thrown by `cancel` to terminate a thread.dataAsyncCancelled =AsyncCancelled deriving(Int -> AsyncCancelled -> ShowS
[AsyncCancelled] -> ShowS
AsyncCancelled -> String
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
showList :: [AsyncCancelled] -> ShowS
$cshowList :: [AsyncCancelled] -> ShowS
show :: AsyncCancelled -> String
$cshow :: AsyncCancelled -> String
showsPrec :: Int -> AsyncCancelled -> ShowS
$cshowsPrec :: Int -> AsyncCancelled -> ShowS
Show,AsyncCancelled -> AsyncCancelled -> Bool
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
/= :: AsyncCancelled -> AsyncCancelled -> Bool
$c/= :: AsyncCancelled -> AsyncCancelled -> Bool
== :: AsyncCancelled -> AsyncCancelled -> Bool
$c== :: AsyncCancelled -> AsyncCancelled -> Bool
Eq
#if __GLASGOW_HASKELL__ < 710
,Typeable
#endif
)instanceExceptionAsyncCancelled where
#if __GLASGOW_HASKELL__ >= 708
fromException :: SomeException -> Maybe AsyncCancelled
fromException=forall e. Exception e => SomeException -> Maybe e
asyncExceptionFromExceptiontoException :: AsyncCancelled -> SomeException
toException=forall e. Exception e => e -> SomeException
asyncExceptionToException
#endif
-- | Cancel an asynchronous action---- This is a variant of `cancel`, but it is not interruptible.{-# INLINEuninterruptibleCancel #-}uninterruptibleCancel ::Async a ->IO()uninterruptibleCancel :: forall a. Async a -> IO ()
uninterruptibleCancel =forall b. IO b -> IO b
uninterruptibleMask_forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Async a -> IO ()
cancel -- | 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'.cancelWith ::Exceptione =>Async a ->e ->IO()cancelWith :: forall e a. Exception e => Async a -> e -> IO ()
cancelWith a :: Async a
a @(Async ThreadId
t STM (Either SomeException a)
_)e
e =forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
t e
e forall (f :: * -> *) a b. Applicative f => f a -> f b -> f a
<*forall a. Async a -> IO (Either SomeException a)
waitCatch Async a
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'.-- The input list must be non-empty.---- If multiple 'Async's complete or have completed, then the value-- returned corresponds to the first completed 'Async' in the list.--{-# INLINEwaitAnyCatch #-}waitAnyCatch ::[Async a ]->IO(Async a ,EitherSomeExceptiona )waitAnyCatch :: forall a. [Async a] -> IO (Async a, Either SomeException a)
waitAnyCatch =forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. [Async a] -> STM (Async a, Either SomeException a)
waitAnyCatchSTM -- | A version of 'waitAnyCatch' that can be used inside an STM transaction.---- @since 2.1.0waitAnyCatchSTM ::[Async a ]->STM(Async a ,EitherSomeExceptiona )waitAnyCatchSTM :: forall a. [Async a] -> STM (Async a, Either SomeException a)
waitAnyCatchSTM []=forall e a. Exception e => e -> STM a
throwSTMforall a b. (a -> b) -> a -> b
$String -> ErrorCall
ErrorCallString
"waitAnyCatchSTM: invalid argument: input list must be non-empty"waitAnyCatchSTM [Async a]
asyncs =forall (t :: * -> *) a b.
Foldable t =>
(a -> b -> b) -> b -> t a -> b
foldrforall a. STM a -> STM a -> STM a
orElseforall a. STM a
retryforall a b. (a -> b) -> a -> b
$forall a b. (a -> b) -> [a] -> [b]
map(\Async a
a ->doEither SomeException a
r <-forall a. Async a -> STM (Either SomeException a)
waitCatchSTM Async a
a ;forall (m :: * -> *) a. Monad m => a -> m a
return(Async a
a ,Either SomeException a
r ))[Async a]
asyncs -- | Like 'waitAnyCatch', but also cancels the other asynchronous-- operations as soon as one has completed.--waitAnyCatchCancel ::[Async a ]->IO(Async a ,EitherSomeExceptiona )waitAnyCatchCancel :: forall a. [Async a] -> IO (Async a, Either SomeException a)
waitAnyCatchCancel [Async a]
asyncs =forall a. [Async a] -> IO (Async a, Either SomeException a)
waitAnyCatch [Async a]
asyncs forall a b. IO a -> IO b -> IO a
`finally`forall a. [Async a] -> IO ()
cancelMany [Async a]
asyncs -- | Wait for any of the supplied @Async@s to complete. If the first-- to complete throws an exception, then that exception is re-thrown-- by 'waitAny'.-- The input list must be non-empty.---- If multiple 'Async's complete or have completed, then the value-- returned corresponds to the first completed 'Async' in the list.--{-# INLINEwaitAny #-}waitAny ::[Async a ]->IO(Async a ,a )waitAny :: forall a. [Async a] -> IO (Async a, a)
waitAny =forall a. STM a -> IO a
atomicallyforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. [Async a] -> STM (Async a, a)
waitAnySTM -- | A version of 'waitAny' that can be used inside an STM transaction.---- @since 2.1.0waitAnySTM ::[Async a ]->STM(Async a ,a )waitAnySTM :: forall a. [Async a] -> STM (Async a, a)
waitAnySTM []=forall e a. Exception e => e -> STM a
throwSTMforall a b. (a -> b) -> a -> b
$String -> ErrorCall
ErrorCallString
"waitAnySTM: invalid argument: input list must be non-empty"waitAnySTM [Async a]
asyncs =forall (t :: * -> *) a b.
Foldable t =>
(a -> b -> b) -> b -> t a -> b
foldrforall a. STM a -> STM a -> STM a
orElseforall a. STM a
retryforall a b. (a -> b) -> a -> b
$forall a b. (a -> b) -> [a] -> [b]
map(\Async a
a ->doa
r <-forall a. Async a -> STM a
waitSTM Async a
a ;forall (m :: * -> *) a. Monad m => a -> m a
return(Async a
a ,a
r ))[Async a]
asyncs -- | Like 'waitAny', but also cancels the other asynchronous-- operations as soon as one has completed.--waitAnyCancel ::[Async a ]->IO(Async a ,a )waitAnyCancel :: forall a. [Async a] -> IO (Async a, a)
waitAnyCancel [Async a]
asyncs =forall a. [Async a] -> IO (Async a, a)
waitAny [Async a]
asyncs forall a b. IO a -> IO b -> IO a
`finally`forall a. [Async a] -> IO ()
cancelMany [Async a]
asyncs -- | Wait for the first of two @Async@s to finish.{-# INLINEwaitEitherCatch #-}waitEitherCatch ::Async a ->Async b ->IO(Either(EitherSomeExceptiona )(EitherSomeExceptionb ))waitEitherCatch :: forall a b.
Async a
-> Async b
-> IO (Either (Either SomeException a) (Either SomeException b))
waitEitherCatch Async a
left Async b
right =forall b. IO b -> IO b
tryAgain forall a b. (a -> b) -> a -> b
$forall a. STM a -> IO a
atomically(forall a b.
Async a
-> Async b
-> STM (Either (Either SomeException a) (Either SomeException b))
waitEitherCatchSTM Async a
left Async b
right )where-- See: https://github.com/simonmar/async/issues/14tryAgain :: IO a -> IO a
tryAgain IO a
f =IO a
f forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch`\BlockedIndefinitelyOnSTM
BlockedIndefinitelyOnSTM->IO a
f -- | A version of 'waitEitherCatch' that can be used inside an STM transaction.---- @since 2.1.0waitEitherCatchSTM ::Async a ->Async b ->STM(Either(EitherSomeExceptiona )(EitherSomeExceptionb ))waitEitherCatchSTM :: forall a b.
Async a
-> Async b
-> STM (Either (Either SomeException a) (Either SomeException b))
waitEitherCatchSTM Async a
left Async b
right =(forall a b. a -> Either a b
Leftforall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall a. Async a -> STM (Either SomeException a)
waitCatchSTM Async a
left )forall a. STM a -> STM a -> STM a
`orElse`(forall a b. b -> Either a b
Rightforall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall a. Async a -> STM (Either SomeException a)
waitCatchSTM Async b
right )-- | Like 'waitEitherCatch', but also 'cancel's both @Async@s before-- returning.--waitEitherCatchCancel ::Async a ->Async b ->IO(Either(EitherSomeExceptiona )(EitherSomeExceptionb ))waitEitherCatchCancel :: forall a b.
Async a
-> Async b
-> IO (Either (Either SomeException a) (Either SomeException b))
waitEitherCatchCancel Async a
left Async b
right =forall a b.
Async a
-> Async b
-> IO (Either (Either SomeException a) (Either SomeException b))
waitEitherCatch Async a
left Async b
right forall a b. IO a -> IO b -> IO a
`finally`forall a. [Async a] -> IO ()
cancelMany [()forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$Async a
left ,()forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$Async b
right ]-- | 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'.--{-# INLINEwaitEither #-}waitEither ::Async a ->Async b ->IO(Eithera b )waitEither :: forall a b. Async a -> Async b -> IO (Either a b)
waitEither Async a
left Async b
right =forall a. STM a -> IO a
atomically(forall a b. Async a -> Async b -> STM (Either a b)
waitEitherSTM Async a
left Async b
right )-- | A version of 'waitEither' that can be used inside an STM transaction.---- @since 2.1.0waitEitherSTM ::Async a ->Async b ->STM(Eithera b )waitEitherSTM :: forall a b. Async a -> Async b -> STM (Either a b)
waitEitherSTM Async a
left Async b
right =(forall a b. a -> Either a b
Leftforall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall a. Async a -> STM a
waitSTM Async a
left )forall a. STM a -> STM a -> STM a
`orElse`(forall a b. b -> Either a b
Rightforall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall a. Async a -> STM a
waitSTM Async b
right )-- | Like 'waitEither', but the result is ignored.--{-# INLINEwaitEither_ #-}waitEither_ ::Async a ->Async b ->IO()waitEither_ :: forall a b. Async a -> Async b -> IO ()
waitEither_ Async a
left Async b
right =forall a. STM a -> IO a
atomically(forall a b. Async a -> Async b -> STM ()
waitEitherSTM_ Async a
left Async b
right )-- | A version of 'waitEither_' that can be used inside an STM transaction.---- @since 2.1.0waitEitherSTM_ ::Async a ->Async b ->STM()waitEitherSTM_ :: forall a b. Async a -> Async b -> STM ()
waitEitherSTM_ Async a
left Async b
right =(forall (f :: * -> *) a. Functor f => f a -> f ()
voidforall a b. (a -> b) -> a -> b
$forall a. Async a -> STM a
waitSTM Async a
left )forall a. STM a -> STM a -> STM a
`orElse`(forall (f :: * -> *) a. Functor f => f a -> f ()
voidforall a b. (a -> b) -> a -> b
$forall a. Async a -> STM a
waitSTM Async b
right )-- | Like 'waitEither', but also 'cancel's both @Async@s before-- returning.--waitEitherCancel ::Async a ->Async b ->IO(Eithera b )waitEitherCancel :: forall a b. Async a -> Async b -> IO (Either a b)
waitEitherCancel Async a
left Async b
right =forall a b. Async a -> Async b -> IO (Either a b)
waitEither Async a
left Async b
right forall a b. IO a -> IO b -> IO a
`finally`forall a. [Async a] -> IO ()
cancelMany [()forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$Async a
left ,()forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$Async b
right ]-- | 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'.--{-# INLINEwaitBoth #-}waitBoth ::Async a ->Async b ->IO(a ,b )waitBoth :: forall a b. Async a -> Async b -> IO (a, b)
waitBoth Async a
left Async b
right =forall b. IO b -> IO b
tryAgain forall a b. (a -> b) -> a -> b
$forall a. STM a -> IO a
atomically(forall a b. Async a -> Async b -> STM (a, b)
waitBothSTM Async a
left Async b
right )where-- See: https://github.com/simonmar/async/issues/14tryAgain :: IO a -> IO a
tryAgain IO a
f =IO a
f forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch`\BlockedIndefinitelyOnSTM
BlockedIndefinitelyOnSTM->IO a
f -- | A version of 'waitBoth' that can be used inside an STM transaction.---- @since 2.1.0waitBothSTM ::Async a ->Async b ->STM(a ,b )waitBothSTM :: forall a b. Async a -> Async b -> STM (a, b)
waitBothSTM Async a
left Async b
right =doa
a <-forall a. Async a -> STM a
waitSTM Async a
left forall a. STM a -> STM a -> STM a
`orElse`(forall a. Async a -> STM a
waitSTM Async b
right forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>>forall a. STM a
retry)b
b <-forall a. Async a -> STM a
waitSTM Async b
right forall (m :: * -> *) a. Monad m => a -> m a
return(a
a ,b
b )-- ------------------------------------------------------------------------------- Linking threadsdataExceptionInLinkedThread =foralla .ExceptionInLinkedThread (Async a )SomeException
#if __GLASGOW_HASKELL__ < 710
derivingTypeable
#endif
instanceShowExceptionInLinkedThread whereshowsPrec :: Int -> ExceptionInLinkedThread -> ShowS
showsPrecInt
p (ExceptionInLinkedThread (Async ThreadId
t STM (Either SomeException a)
_)SomeException
e )=Bool -> ShowS -> ShowS
showParen(Int
p forall a. Ord a => a -> a -> Bool
>=Int
11)forall a b. (a -> b) -> a -> b
$String -> ShowS
showStringString
"ExceptionInLinkedThread "forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Show a => Int -> a -> ShowS
showsPrecInt
11ThreadId
t forall b c a. (b -> c) -> (a -> b) -> a -> c
.String -> ShowS
showStringString
" "forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Show a => Int -> a -> ShowS
showsPrecInt
11SomeException
e instanceExceptionExceptionInLinkedThread where
#if __GLASGOW_HASKELL__ >= 708
fromException :: SomeException -> Maybe ExceptionInLinkedThread
fromException =forall e. Exception e => SomeException -> Maybe e
asyncExceptionFromExceptiontoException :: ExceptionInLinkedThread -> SomeException
toException =forall e. Exception e => e -> SomeException
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, 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 :: forall a. Async a -> IO ()
link =forall a. (SomeException -> Bool) -> Async a -> IO ()
linkOnly (Bool -> Bool
notforall b c a. (b -> c) -> (a -> b) -> a -> c
.SomeException -> Bool
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, wrapped in 'ExceptionInLinkedThread'.---- 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 :: forall a. (SomeException -> Bool) -> Async a -> IO ()
linkOnly SomeException -> Bool
shouldThrow Async a
a =doThreadId
me <-IO ThreadId
myThreadIdforall (f :: * -> *) a. Functor f => f a -> f ()
voidforall a b. (a -> b) -> a -> b
$forall a. IO a -> IO ThreadId
forkRepeat forall a b. (a -> b) -> a -> b
$doEither SomeException a
r <-forall a. Async a -> IO (Either SomeException a)
waitCatch Async a
a caseEither SomeException a
r ofLeftSomeException
e |SomeException -> Bool
shouldThrow SomeException
e ->forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
me (forall a. Async a -> SomeException -> ExceptionInLinkedThread
ExceptionInLinkedThread Async a
a SomeException
e )Either SomeException a
_otherwise ->forall (m :: * -> *) a. Monad m => a -> m a
return()-- | Link two @Async@s together, such that if either raises an-- 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 :: forall a b. Async a -> Async b -> IO ()
link2 =forall a b. (SomeException -> Bool) -> Async a -> Async b -> IO ()
link2Only (Bool -> Bool
notforall b c a. (b -> c) -> (a -> b) -> a -> c
.SomeException -> Bool
isCancel )-- | Link two @Async@s together, such that if either raises an-- exception, the same exception is re-thrown in the other @Async@,-- wrapped in 'ExceptionInLinkedThread'.---- The supplied predicate determines which exceptions in the target-- thread should be propagated to the source thread.--link2Only ::(SomeException->Bool)->Async a ->Async b ->IO()link2Only :: forall a b. (SomeException -> Bool) -> Async a -> Async b -> IO ()
link2Only SomeException -> Bool
shouldThrow left :: Async a
left @(Async ThreadId
tl STM (Either SomeException a)
_)right :: Async b
right @(Async ThreadId
tr STM (Either SomeException b)
_)=forall (f :: * -> *) a. Functor f => f a -> f ()
voidforall a b. (a -> b) -> a -> b
$forall a. IO a -> IO ThreadId
forkRepeat forall a b. (a -> b) -> a -> b
$doEither (Either SomeException a) (Either SomeException b)
r <-forall a b.
Async a
-> Async b
-> IO (Either (Either SomeException a) (Either SomeException b))
waitEitherCatch Async a
left Async b
right caseEither (Either SomeException a) (Either SomeException b)
r ofLeft(LeftSomeException
e )|SomeException -> Bool
shouldThrow SomeException
e ->forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
tr (forall a. Async a -> SomeException -> ExceptionInLinkedThread
ExceptionInLinkedThread Async a
left SomeException
e )Right(LeftSomeException
e )|SomeException -> Bool
shouldThrow SomeException
e ->forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
tl (forall a. Async a -> SomeException -> ExceptionInLinkedThread
ExceptionInLinkedThread Async b
right SomeException
e )Either (Either SomeException a) (Either SomeException b)
_->forall (m :: * -> *) a. Monad m => a -> m a
return()isCancel ::SomeException->BoolisCancel :: SomeException -> Bool
isCancel SomeException
e |JustAsyncCancelled
AsyncCancelled <-forall e. Exception e => SomeException -> Maybe e
fromExceptionSomeException
e =Bool
True|Bool
otherwise=Bool
False-- ------------------------------------------------------------------------------- | 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--race ::IOa ->IOb ->IO(Eithera b )-- | Like 'race', but the result is ignored.--race_ ::IOa ->IOb ->IO()-- | 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 bconcurrently ::IOa ->IOb ->IO(a ,b )-- | Run two @IO@ actions concurrently. If both of them end with @Right@,-- return both results. If one of then ends with @Left@, interrupt the other-- action and return the @Left@. --concurrentlyE ::IO(Eithere a )->IO(Eithere b )->IO(Eithere (a ,b ))-- | 'concurrently', but ignore the result values---- @since 2.1.1concurrently_ ::IOa ->IOb ->IO()
#define USE_ASYNC_VERSIONS 0

#if USE_ASYNC_VERSIONS
raceleftright=withAsyncleft$\a->withAsyncright$\b->waitEitherabrace_leftright=void$raceleftrightconcurrentlyleftright=withAsyncleft$\a->withAsyncright$\b->waitBothabconcurrently_leftright=void$concurrentlyleftright
#else
-- MVar versions of race/concurrently-- More ugly than the Async versions, but quite a bit faster.-- race :: IO a -> IO b -> IO (Either a b)race :: forall a b. IO a -> IO b -> IO (Either a b)
race IO a
left IO b
right =forall a b r.
IO a
-> IO b -> (IO (Either SomeException (Either a b)) -> IO r) -> IO r
concurrently' IO a
left IO b
right forall {e} {b}. Exception e => IO (Either e b) -> IO b
collect wherecollect :: IO (Either e b) -> IO b
collect IO (Either e b)
m =doEither e b
e <-IO (Either e b)
m caseEither e b
e ofLefte
ex ->forall e a. Exception e => e -> IO a
throwIOe
ex Rightb
r ->forall (m :: * -> *) a. Monad m => a -> m a
returnb
r -- race_ :: IO a -> IO b -> IO ()race_ :: forall a b. IO a -> IO b -> IO ()
race_ IO a
left IO b
right =forall (f :: * -> *) a. Functor f => f a -> f ()
voidforall a b. (a -> b) -> a -> b
$forall a b. IO a -> IO b -> IO (Either a b)
race IO a
left IO b
right -- concurrently :: IO a -> IO b -> IO (a,b)concurrently :: forall a b. IO a -> IO b -> IO (a, b)
concurrently IO a
left IO b
right =forall a b r.
IO a
-> IO b -> (IO (Either SomeException (Either a b)) -> IO r) -> IO r
concurrently' IO a
left IO b
right (forall {e} {a} {b}.
Exception e =>
[Either a b] -> IO (Either e (Either a b)) -> IO (a, b)
collect [])wherecollect :: [Either a b] -> IO (Either e (Either a b)) -> IO (a, b)
collect [Lefta
a ,Rightb
b ]IO (Either e (Either a b))
_=forall (m :: * -> *) a. Monad m => a -> m a
return(a
a ,b
b )collect [Rightb
b ,Lefta
a ]IO (Either e (Either a b))
_=forall (m :: * -> *) a. Monad m => a -> m a
return(a
a ,b
b )collect [Either a b]
xs IO (Either e (Either a b))
m =doEither e (Either a b)
e <-IO (Either e (Either a b))
m caseEither e (Either a b)
e ofLefte
ex ->forall e a. Exception e => e -> IO a
throwIOe
ex RightEither a b
r ->[Either a b] -> IO (Either e (Either a b)) -> IO (a, b)
collect (Either a b
r forall a. a -> [a] -> [a]
:[Either a b]
xs )IO (Either e (Either a b))
m -- concurrentlyE :: IO (Either e a) -> IO (Either e b) -> IO (Either e (a, b))concurrentlyE :: forall e a b.
IO (Either e a) -> IO (Either e b) -> IO (Either e (a, b))
concurrentlyE IO (Either e a)
left IO (Either e b)
right =forall a b r.
IO a
-> IO b -> (IO (Either SomeException (Either a b)) -> IO r) -> IO r
concurrently' IO (Either e a)
left IO (Either e b)
right (forall {e} {a} {a} {b}.
Exception e =>
[Either (Either a a) (Either a b)]
-> IO (Either e (Either (Either a a) (Either a b)))
-> IO (Either a (a, b))
collect [])wherecollect :: [Either (Either a a) (Either a b)]
-> IO (Either e (Either (Either a a) (Either a b)))
-> IO (Either a (a, b))
collect [Left(Righta
a ),Right(Rightb
b )]IO (Either e (Either (Either a a) (Either a b)))
_=forall (m :: * -> *) a. Monad m => a -> m a
returnforall a b. (a -> b) -> a -> b
$forall a b. b -> Either a b
Right(a
a ,b
b )collect [Right(Rightb
b ),Left(Righta
a )]IO (Either e (Either (Either a a) (Either a b)))
_=forall (m :: * -> *) a. Monad m => a -> m a
returnforall a b. (a -> b) -> a -> b
$forall a b. b -> Either a b
Right(a
a ,b
b )collect (Left(Lefta
ea ):[Either (Either a a) (Either a b)]
_)IO (Either e (Either (Either a a) (Either a b)))
_=forall (m :: * -> *) a. Monad m => a -> m a
returnforall a b. (a -> b) -> a -> b
$forall a b. a -> Either a b
Lefta
ea collect (Right(Lefta
eb ):[Either (Either a a) (Either a b)]
_)IO (Either e (Either (Either a a) (Either a b)))
_=forall (m :: * -> *) a. Monad m => a -> m a
returnforall a b. (a -> b) -> a -> b
$forall a b. a -> Either a b
Lefta
eb collect [Either (Either a a) (Either a b)]
xs IO (Either e (Either (Either a a) (Either a b)))
m =doEither e (Either (Either a a) (Either a b))
e <-IO (Either e (Either (Either a a) (Either a b)))
m caseEither e (Either (Either a a) (Either a b))
e ofLefte
ex ->forall e a. Exception e => e -> IO a
throwIOe
ex RightEither (Either a a) (Either a b)
r ->[Either (Either a a) (Either a b)]
-> IO (Either e (Either (Either a a) (Either a b)))
-> IO (Either a (a, b))
collect (Either (Either a a) (Either a b)
r forall a. a -> [a] -> [a]
:[Either (Either a a) (Either a b)]
xs )IO (Either e (Either (Either a a) (Either a b)))
m concurrently' ::IOa ->IOb ->(IO(EitherSomeException(Eithera b ))->IOr )->IOr concurrently' :: forall a b r.
IO a
-> IO b -> (IO (Either SomeException (Either a b)) -> IO r) -> IO r
concurrently' IO a
left IO b
right IO (Either SomeException (Either a b)) -> IO r
collect =doMVar (Either SomeException (Either a b))
done <-forall a. IO (MVar a)
newEmptyMVarforall b. ((forall b. IO b -> IO b) -> IO b) -> IO b
maskforall a b. (a -> b) -> a -> b
$\forall b. IO b -> IO b
restore ->do-- 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.ThreadId
lid <-IO () -> IO ThreadId
forkIOforall a b. (a -> b) -> a -> b
$forall b. IO b -> IO b
uninterruptibleMask_forall a b. (a -> b) -> a -> b
$forall b. IO b -> IO b
restore (IO a
left forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>=forall a. MVar a -> a -> IO ()
putMVarMVar (Either SomeException (Either a b))
done forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a b. b -> Either a b
Rightforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a b. a -> Either a b
Left)forall a. IO a -> (SomeException -> IO a) -> IO a
`catchAll` (forall a. MVar a -> a -> IO ()
putMVarMVar (Either SomeException (Either a b))
done forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a b. a -> Either a b
Left)ThreadId
rid <-IO () -> IO ThreadId
forkIOforall a b. (a -> b) -> a -> b
$forall b. IO b -> IO b
uninterruptibleMask_forall a b. (a -> b) -> a -> b
$forall b. IO b -> IO b
restore (IO b
right forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>=forall a. MVar a -> a -> IO ()
putMVarMVar (Either SomeException (Either a b))
done forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a b. b -> Either a b
Rightforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a b. b -> Either a b
Right)forall a. IO a -> (SomeException -> IO a) -> IO a
`catchAll` (forall a. MVar a -> a -> IO ()
putMVarMVar (Either SomeException (Either a b))
done forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a b. a -> Either a b
Left)IORef Int
count <-forall a. a -> IO (IORef a)
newIORef(Int
2::Int)lettakeDone :: IO (Either SomeException (Either a b))
takeDone =doEither SomeException (Either a b)
r <-forall a. MVar a -> IO a
takeMVarMVar (Either SomeException (Either a b))
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.forall a. IORef a -> (a -> a) -> IO ()
modifyIORefIORef Int
count (forall a. Num a => a -> a -> a
subtractInt
1)forall (m :: * -> *) a. Monad m => a -> m a
returnEither SomeException (Either a b)
r lettryAgain :: IO a -> IO a
tryAgain IO a
f =IO a
f forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch`\BlockedIndefinitelyOnMVar
BlockedIndefinitelyOnMVar->IO a
f stop :: IO ()
stop =do-- kill right before left, to match the semantics of-- the version using withAsync. (#27)forall b. IO b -> IO b
uninterruptibleMask_forall a b. (a -> b) -> a -> b
$doInt
count' <-forall a. IORef a -> IO a
readIORefIORef Int
count -- 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.forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when(Int
count' forall a. Ord a => a -> a -> Bool
>Int
0)forall a b. (a -> b) -> a -> b
$forall (f :: * -> *) a. Functor f => f a -> f ()
voidforall a b. (a -> b) -> a -> b
$IO () -> IO ThreadId
forkIOforall a b. (a -> b) -> a -> b
$doforall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
rid AsyncCancelled
AsyncCancelled forall e. Exception e => ThreadId -> e -> IO ()
throwToThreadId
lid AsyncCancelled
AsyncCancelled -- ensure the children are really deadforall (m :: * -> *) a. Applicative m => Int -> m a -> m ()
replicateM_Int
count' (forall b. IO b -> IO b
tryAgain forall a b. (a -> b) -> a -> b
$forall a. MVar a -> IO a
takeMVarMVar (Either SomeException (Either a b))
done )r
r <-IO (Either SomeException (Either a b)) -> IO r
collect (forall b. IO b -> IO b
tryAgain forall a b. (a -> b) -> a -> b
$IO (Either SomeException (Either a b))
takeDone )forall a b. IO a -> IO b -> IO a
`onException`IO ()
stop IO ()
stop forall (m :: * -> *) a. Monad m => a -> m a
returnr
r concurrently_ :: forall a b. IO a -> IO b -> IO ()
concurrently_ IO a
left IO b
right =forall a b r.
IO a
-> IO b -> (IO (Either SomeException (Either a b)) -> IO r) -> IO r
concurrently' IO a
left IO b
right (forall {e} {b}. Exception e => Int -> IO (Either e b) -> IO ()
collect Int
0)wherecollect :: Int -> IO (Either e b) -> IO ()
collect Int
2IO (Either e b)
_=forall (m :: * -> *) a. Monad m => a -> m a
return()collect Int
i IO (Either e b)
m =doEither e b
e <-IO (Either e b)
m caseEither e b
e ofLefte
ex ->forall e a. Exception e => e -> IO a
throwIOe
ex Rightb
_->Int -> IO (Either e b) -> IO ()
collect (Int
i forall a. Num a => a -> a -> a
+Int
1::Int)IO (Either e b)
m 
#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.---- 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"]---- Take into account that @async@ will try to immediately spawn a thread-- for each element of the @Traversable@, so running this on large-- inputs without care may lead to resource exhaustion (of memory,-- file descriptors, or other limited resources).mapConcurrently ::Traversablet =>(a ->IOb )->t a ->IO(t b )mapConcurrently :: forall (t :: * -> *) a b.
Traversable t =>
(a -> IO b) -> t a -> IO (t b)
mapConcurrently a -> IO b
f =forall a. Concurrently a -> IO a
runConcurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
traverse(forall a. IO a -> Concurrently a
Concurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.a -> IO b
f )-- | `forConcurrently` is `mapConcurrently` with its arguments flipped---- > pages <- forConcurrently ["url1", "url2", "url3"] $ \url -> getURL url---- @since 2.1.0forConcurrently ::Traversablet =>t a ->(a ->IOb )->IO(t b )forConcurrently :: forall (t :: * -> *) a b.
Traversable t =>
t a -> (a -> IO b) -> IO (t b)
forConcurrently =forall a b c. (a -> b -> c) -> b -> a -> c
flipforall (t :: * -> *) a b.
Traversable t =>
(a -> IO b) -> t a -> IO (t b)
mapConcurrently -- | `mapConcurrently_` is `mapConcurrently` with the return value discarded;-- a concurrent equivalent of 'mapM_'.mapConcurrently_ ::F.Foldablef =>(a ->IOb )->f a ->IO()mapConcurrently_ :: forall (f :: * -> *) a b. Foldable f => (a -> IO b) -> f a -> IO ()
mapConcurrently_ a -> IO b
f =forall a. Concurrently a -> IO a
runConcurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
F.foldMap(forall a. IO a -> Concurrently a
Concurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (f :: * -> *) a. Functor f => f a -> f ()
voidforall b c a. (b -> c) -> (a -> b) -> a -> c
.a -> IO b
f )-- | `forConcurrently_` is `forConcurrently` with the return value discarded;-- a concurrent equivalent of 'forM_'.forConcurrently_ ::F.Foldablef =>f a ->(a ->IOb )->IO()forConcurrently_ :: forall (f :: * -> *) a b. Foldable f => f a -> (a -> IO b) -> IO ()
forConcurrently_ =forall a b c. (a -> b -> c) -> b -> a -> c
flipforall (f :: * -> *) a b. Foldable f => (a -> IO b) -> f a -> IO ()
mapConcurrently_ -- | Perform the action in the given number of threads.---- @since 2.1.1replicateConcurrently ::Int->IOa ->IO[a ]replicateConcurrently :: forall a. Int -> IO a -> IO [a]
replicateConcurrently Int
cnt =forall a. Concurrently a -> IO a
runConcurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (t :: * -> *) (f :: * -> *) a.
(Traversable t, Applicative f) =>
t (f a) -> f (t a)
sequenceAforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Int -> a -> [a]
replicateInt
cnt forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. IO a -> Concurrently a
Concurrently -- | Same as 'replicateConcurrently', but ignore the results.---- @since 2.1.1replicateConcurrently_ ::Int->IOa ->IO()replicateConcurrently_ :: forall a. Int -> IO a -> IO ()
replicateConcurrently_ Int
cnt =forall a. Concurrently a -> IO a
runConcurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (t :: * -> *) m. (Foldable t, Monoid m) => t m -> m
F.foldforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. Int -> a -> [a]
replicateInt
cnt forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall a. IO a -> Concurrently a
Concurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (f :: * -> *) a. Functor f => f a -> f ()
void-- ------------------------------------------------------------------------------- | 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")--newtypeConcurrently a =Concurrently {forall a. Concurrently a -> IO a
runConcurrently ::IOa }instanceFunctorConcurrently wherefmap :: forall a b. (a -> b) -> Concurrently a -> Concurrently b
fmapa -> b
f (Concurrently IO a
a )=forall a. IO a -> Concurrently a
Concurrently forall a b. (a -> b) -> a -> b
$a -> b
f forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>IO a
a instanceApplicativeConcurrently wherepure :: forall a. a -> Concurrently a
pure=forall a. IO a -> Concurrently a
Concurrently forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (m :: * -> *) a. Monad m => a -> m a
returnConcurrently IO (a -> b)
fs <*> :: forall a b.
Concurrently (a -> b) -> Concurrently a -> Concurrently b
<*>Concurrently IO a
as =forall a. IO a -> Concurrently a
Concurrently forall a b. (a -> b) -> a -> b
$(\(a -> b
f ,a
a )->a -> b
f a
a )forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall a b. IO a -> IO b -> IO (a, b)
concurrently IO (a -> b)
fs IO a
as -- | 'Control.Alternative.empty' waits forever. 'Control.Alternative.<|>' returns the first to finish and 'cancel's the other.instanceAlternativeConcurrently whereempty :: forall a. Concurrently a
empty=forall a. IO a -> Concurrently a
Concurrently forall a b. (a -> b) -> a -> b
$forall (f :: * -> *) a b. Applicative f => f a -> f b
forever(Int -> IO ()
threadDelayforall a. Bounded a => a
maxBound)Concurrently IO a
as <|> :: forall a. Concurrently a -> Concurrently a -> Concurrently a
<|>Concurrently IO a
bs =forall a. IO a -> Concurrently a
Concurrently forall a b. (a -> b) -> a -> b
$forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
eitherforall a. a -> a
idforall a. a -> a
idforall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall a b. IO a -> IO b -> IO (Either a b)
race IO a
as IO a
bs 
#if MIN_VERSION_base(4,9,0)
-- | Only defined by @async@ for @base >= 4.9@---- @since 2.1.0instanceSemigroupa =>Semigroup(Concurrently a )where<> :: Concurrently a -> Concurrently a -> Concurrently a
(<>)=forall (f :: * -> *) a b c.
Applicative f =>
(a -> b -> c) -> f a -> f b -> f c
liftA2forall a. Semigroup a => a -> a -> a
(<>)-- | @since 2.1.0instance(Semigroupa ,Monoida )=>Monoid(Concurrently a )wheremempty :: Concurrently a
mempty=forall (f :: * -> *) a. Applicative f => a -> f a
pureforall a. Monoid a => a
memptymappend :: Concurrently a -> Concurrently a -> Concurrently a
mappend=forall a. Semigroup a => a -> a -> a
(<>)
#else
-- | @since 2.1.0instanceMonoida=>Monoid(Concurrentlya)wheremempty=purememptymappend=liftA2mappend
#endif
-- | A value of type @ConcurrentlyE e a@ is an @IO@ operation that can be-- composed with other @ConcurrentlyE@ values, using the @Applicative@ instance.---- Calling @runConcurrentlyE@ on a value of type @ConcurrentlyE e a@ will-- execute the @IO@ operations it contains concurrently, before delivering-- either the result of type @a@, or an error of type @e@ if one of the actions-- returns @Left@.---- | @since 2.2.5newtypeConcurrentlyE e a =ConcurrentlyE {forall e a. ConcurrentlyE e a -> IO (Either e a)
runConcurrentlyE ::IO(Eithere a )}instanceFunctor(ConcurrentlyE e )wherefmap :: forall a b. (a -> b) -> ConcurrentlyE e a -> ConcurrentlyE e b
fmapa -> b
f (ConcurrentlyE IO (Either e a)
ea )=forall e a. IO (Either e a) -> ConcurrentlyE e a
ConcurrentlyE forall a b. (a -> b) -> a -> b
$forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap(forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmapa -> b
f )IO (Either e a)
ea 
#if MIN_VERSION_base(4,8,0)
instanceBifunctorConcurrentlyE wherebimap :: forall a b c d.
(a -> b) -> (c -> d) -> ConcurrentlyE a c -> ConcurrentlyE b d
bimapa -> b
f c -> d
g (ConcurrentlyE IO (Either a c)
ea )=forall e a. IO (Either e a) -> ConcurrentlyE e a
ConcurrentlyE forall a b. (a -> b) -> a -> b
$forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap(forall (p :: * -> * -> *) a b c d.
Bifunctor p =>
(a -> b) -> (c -> d) -> p a c -> p b d
bimapa -> b
f c -> d
g )IO (Either a c)
ea 
#endif
instanceApplicative(ConcurrentlyE e )wherepure :: forall a. a -> ConcurrentlyE e a
pure=forall e a. IO (Either e a) -> ConcurrentlyE e a
ConcurrentlyE forall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (m :: * -> *) a. Monad m => a -> m a
returnforall b c a. (b -> c) -> (a -> b) -> a -> c
.forall (m :: * -> *) a. Monad m => a -> m a
returnConcurrentlyE IO (Either e (a -> b))
fs <*> :: forall a b.
ConcurrentlyE e (a -> b) -> ConcurrentlyE e a -> ConcurrentlyE e b
<*>ConcurrentlyE IO (Either e a)
eas =forall e a. IO (Either e a) -> ConcurrentlyE e a
ConcurrentlyE forall a b. (a -> b) -> a -> b
$forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap(\(a -> b
f ,a
a )->a -> b
f a
a )forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$>forall e a b.
IO (Either e a) -> IO (Either e b) -> IO (Either e (a, b))
concurrentlyE IO (Either e (a -> b))
fs IO (Either e a)
eas 
#if MIN_VERSION_base(4,9,0)
-- | Either the combination of the successful results, or the first failure. instanceSemigroupa =>Semigroup(ConcurrentlyE e a )where<> :: ConcurrentlyE e a -> ConcurrentlyE e a -> ConcurrentlyE e a
(<>)=forall (f :: * -> *) a b c.
Applicative f =>
(a -> b -> c) -> f a -> f b -> f c
liftA2forall a. Semigroup a => a -> a -> a
(<>)instance(Semigroupa ,Monoida )=>Monoid(ConcurrentlyE e a )wheremempty :: ConcurrentlyE e a
mempty=forall (f :: * -> *) a. Applicative f => a -> f a
pureforall a. Monoid a => a
memptymappend :: ConcurrentlyE e a -> ConcurrentlyE e a -> ConcurrentlyE e a
mappend=forall a. Semigroup a => a -> a -> a
(<>)
#endif
-- ------------------------------------------------------------------------------ | Fork a thread that runs the supplied action, and if it raises an-- exception, re-runs the action. The thread terminates only when the-- action runs to completion without raising an exception.forkRepeat ::IOa ->IOThreadIdforkRepeat :: forall a. IO a -> IO ThreadId
forkRepeat IO a
action =forall b. ((forall b. IO b -> IO b) -> IO b) -> IO b
maskforall a b. (a -> b) -> a -> b
$\forall b. IO b -> IO b
restore ->letgo :: IO ()
go =doEither SomeException a
r <-forall a. IO a -> IO (Either SomeException a)
tryAll (forall b. IO b -> IO b
restore IO a
action )caseEither SomeException a
r ofLeftSomeException
_->IO ()
go Either SomeException a
_->forall (m :: * -> *) a. Monad m => a -> m a
return()inIO () -> IO ThreadId
forkIOIO ()
go catchAll ::IOa ->(SomeException->IOa )->IOa catchAll :: forall a. IO a -> (SomeException -> IO a) -> IO a
catchAll =forall e a. Exception e => IO a -> (e -> IO a) -> IO a
catchtryAll ::IOa ->IO(EitherSomeExceptiona )tryAll :: forall a. IO a -> IO (Either SomeException a)
tryAll =forall e a. Exception e => IO a -> IO (Either e a)
try-- A version of forkIO that does not include the outer exception-- handler: saves a bit of time when we will be installing our own-- exception handler.{-# INLINErawForkIO #-}rawForkIO ::IO()->IOThreadIdrawForkIO :: IO () -> IO ThreadId
rawForkIO (IOState# RealWorld -> (# State# RealWorld, () #)
action )=forall a. (State# RealWorld -> (# State# RealWorld, a #)) -> IO a
IOforall a b. (a -> b) -> a -> b
$\State# RealWorld
s ->case(forall a.
a -> State# RealWorld -> (# State# RealWorld, ThreadId# #)
fork#State# RealWorld -> (# State# RealWorld, () #)
action State# RealWorld
s )of(#State# RealWorld
s1 ,ThreadId#
tid #)->(#State# RealWorld
s1 ,ThreadId# -> ThreadId
ThreadIdThreadId#
tid #){-# INLINErawForkOn #-}rawForkOn ::Int->IO()->IOThreadIdrawForkOn :: Int -> IO () -> IO ThreadId
rawForkOn (I#Int#
cpu )(IOState# RealWorld -> (# State# RealWorld, () #)
action )=forall a. (State# RealWorld -> (# State# RealWorld, a #)) -> IO a
IOforall a b. (a -> b) -> a -> b
$\State# RealWorld
s ->case(forall a.
Int# -> a -> State# RealWorld -> (# State# RealWorld, ThreadId# #)
forkOn#Int#
cpu State# RealWorld -> (# State# RealWorld, () #)
action State# RealWorld
s )of(#State# RealWorld
s1 ,ThreadId#
tid #)->(#State# RealWorld
s1 ,ThreadId# -> ThreadId
ThreadIdThreadId#
tid #)