async-pool 0.8.0 → 0.9.0
raw patch · 2 files changed
+51/−53 lines, 2 filesdep ~asyncdep ~containersdep ~fglPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: async, containers, fgl, monad-control, stm, transformers, transformers-base
API changes (from Hackage documentation)
Files
Control/Concurrent/Async/Pool/Internal.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-} module Control.Concurrent.Async.Pool.Internal where @@ -24,7 +25,7 @@ import Prelude hiding (mapM_, mapM, foldr, all, any, concatMap, foldl1) -- | Return a list of actions ready for execution, by checking the graph to--- ensure that all dependencies have completed.+-- ensure all dependencies have completed. getReadyNodes :: TaskGroup -> TaskGraph -> STM (IntMap (IO ThreadId)) getReadyNodes p g = do availSlots <- readTVar (avail p)@@ -85,31 +86,12 @@ -- execution slots, but with a bounded lifetime. Leaving the block cancels -- every task still executing in the group. withTaskGroupIn :: Pool -> Int -> (TaskGroup -> IO b) -> IO b-withTaskGroupIn p n f = do- g <- createTaskGroup p n+withTaskGroupIn p n f = createTaskGroup p n >>= \g -> Async.withAsync (runTaskGroup g) $ const $ f g `finally` cancelAll g --- | Create both a pool and a task group with a given number of execution--- slots.+-- | Create both a pool, and a task group with a given number of execution slots. withTaskGroup :: Int -> (TaskGroup -> IO b) -> IO b-withTaskGroup n f = do- p <- createPool- withTaskGroupIn p n f---- | Given parent and child tasks, link them so the child cannot execute until--- the parent has finished. This function does not check for cycles being--- introduced into the dependency graph, which would prevent a task from--- ever running.-unsafeMakeDependent :: Pool- -> Handle -- ^ Handle of task doing the waiting- -> Handle -- ^ Handle of task we must wait on (the parent)- -> STM ()-unsafeMakeDependent p child parent = do- g <- readTVar (tasks p)- -- If the parent is no longer in the graph, there is no need to establish- -- dependency. The child can begin executing in the next free slot.- when (gelem parent g) $- modifyTVar (tasks p) (insEdge (parent, child, Pending))+withTaskGroup n f = createPool >>= \p -> withTaskGroupIn p n f -- | Given parent and child tasks, link them so the child cannot execute until -- the parent has finished.@@ -129,21 +111,36 @@ [] -> modifyTVar (tasks p) (insEdge (parent, child, Pending)) _ -> error "makeDependent: Cycle in task graph" --- | Equivalent to async, but acts in STM so that 'makeDependent' may be--- called after the task is created but before it may begin executing.+-- | Given parent and child tasks, link them so the child cannot execute until+-- the parent has finished. This function does not check for introduction of+-- cycles into the dependency graph, which would prevent the child from ever+-- running.+unsafeMakeDependent :: Pool+ -> Handle -- ^ Handle of task doing the waiting+ -> Handle -- ^ Handle of task we must wait on (the parent)+ -> STM ()+unsafeMakeDependent p child parent = do+ g <- readTVar (tasks p)+ -- If the parent is no longer in the graph, there is no need to establish+ -- dependency. The child can begin executing in the next free slot.+ when (gelem parent g) $+ modifyTVar (tasks p) (insEdge (parent, child, Pending))++-- | Equivalent to 'async', but acts in STM so that 'makeDependent' may be+-- called after the task is created, but before it begins executing. asyncSTM :: TaskGroup -> IO a -> STM (Async a) asyncSTM p = asyncUsing p rawForkIO --- | Submit a task which begins executing after all its parents has completed.+-- | Submit a task which begins execution after all its parents have completed. -- This is equivalent to submitting a new task with 'asyncSTM' and linking--- it to its parent using 'makeDependent'.+-- it to its parents using 'mapM makeDependent'. asyncAfterAll :: TaskGroup -> [Handle] -> IO a -> IO (Async a) asyncAfterAll p parents t = atomically $ do child <- asyncUsing p rawForkIO t forM_ parents $ makeDependent (pool p) (taskHandle child) return child --- | Submit a task which begins executing after its parent has completed.+-- | Submit a task that begins execution only after its parent has completed. -- This is equivalent to submitting a new task with 'asyncSTM' and linking -- it to its parent using 'makeDependent'. asyncAfter :: TaskGroup -> Async b -> IO a -> IO (Async a)@@ -160,24 +157,24 @@ hs <- forM fs $ atomically . asyncUsing p rawForkIO f $ forM hs g --- | Execute a group of tasks within the given pool, returning the results in--- order. The order of execution is random, but the results are returned in--- order.+-- | Execute a group of tasks within the given task group, returning the+-- results in order. The order of execution is random, but the results are+-- returned in order. mapTasks :: Traversable t => TaskGroup -> t (IO a) -> IO (t a) mapTasks p fs = mapTasksWorker p fs id wait --- | Execute a group of tasks within the given pool, returning the results in--- order as an Either type to represent exceptions from actions. The order--- of execution is random, but the results are returned in order.+-- | Execute a group of tasks within the given task group, returning the+-- results in order as an Either type to represent exceptions from actions.+-- The order of execution is random, but the results are returned in order. mapTasksE :: Traversable t => TaskGroup -> t (IO a) -> IO (t (Either SomeException a)) mapTasksE p fs = mapTasksWorker p fs id waitCatch --- | Execute a group of tasks within the given pool, ignoring results.+-- | Execute a group of tasks within the given task group, ignoring results. mapTasks_ :: Foldable t => TaskGroup -> t (IO a) -> IO () mapTasks_ p fs = forM_ fs $ atomically . asyncUsing p rawForkIO --- | Execute a group of tasks within the given pool, ignoring results, but--- returning a list of all exceptions.+-- | Execute a group of tasks within the given task group, ignoring results,+-- but returning a list of all exceptions. mapTasksE_ :: Traversable t => TaskGroup -> t (IO a) -> IO (t (Maybe SomeException)) mapTasksE_ p fs = mapTasksWorker p fs (fmap (fmap leftToMaybe)) waitCatch where@@ -193,19 +190,20 @@ waitAnyCatchCancel hs -- | Given a list of actions yielding 'Monoid' results, execute the actions--- concurrently (up to N at time, based on available slots), and 'mappend'+-- concurrently (up to N at a time, based on available slots), and 'mappend' -- each pair of results concurrently as they become ready. The immediate -- result of this function is an 'Async' representing the final value. ----- This is equivalent to the following: @mconcat <$> mapTasks n actions@,+-- This is similar to the following: @mconcat <$> mapTasks n actions@, -- except that intermediate results can be garbage collected as soon as--- they've merged. Also, the value returned from this function is a 'Async'--- which may be polled for the final result.+-- they've been merged. Also, the value returned from this function is an+-- 'Async' which may be polled for the final result. -- -- Lastly, if an 'Exception' occurs in any subtask, the final result will--- yield an exception, but not necessarily the first or last that was caught.+-- also yield an exception -- but not necessarily the first or last that was+-- caught. mapReduce :: (Foldable t, Monoid a)- => TaskGroup -- ^ Pool to execute the tasks within+ => TaskGroup -- ^ Task group to execute the tasks within -> t (IO a) -- ^ Set of Monoid-yielding IO actions -> STM (Async a) -- ^ Returns the final result task mapReduce p fs = do@@ -244,16 +242,17 @@ _ -> (t :) <$> squeeze xs -- | Execute a group of tasks concurrently (using up to N active threads,--- depending on the pool), and feed results to a continuation as soon as--- they become available, in random order. That function may return a--- monoid value which is accumulated to yield a final result.+-- depending on the task group), and feed results to a continuation as soon+-- as they become available, in random order. The continuation function may+-- return a monoid value which is accumulated to yield a final result. If+-- no such value is needed, simply provide `()`. scatterFoldMapM :: (Foldable t, Monoid b, MonadBaseControl IO m) => TaskGroup -> t (IO a) -> (Either SomeException a -> m b) -> m b scatterFoldMapM p fs f = do hs <- liftBase $ atomically $ sequenceA $ foldMap ((:[]) <$> asyncUsing p rawForkIO) fs- control $ \run -> loop run (run $ return mempty) (toList hs)+ control $ \(run :: m b -> IO (StM m b)) -> loop run (run $ return mempty) (toList hs) where loop _ z [] = z loop run z hs = do@@ -276,10 +275,9 @@ Just (Right x) -> Just (h, Right x) -- | The 'Task' Applicative and Monad allow for task dependencies to be built--- using applicative and do notation. Monadic evaluation is sequenced,--- while applicative evaluation is done concurrently for each argument. In--- this way, mixing the two allows you to build a dependency tree using--- ordinary Haskell code.+-- using Applicative and do notation. Monadic evaluation is sequenced,+-- while applicative Evaluation is concurrent for each argument. In this+-- way, mixing the two builds a dependency graph via ordinary Haskell code. newtype Task a = Task { runTask' :: TaskGroup -> IO (IO a) } -- | Run a value in the 'Task' monad and block until the final result is
async-pool.cabal view
@@ -1,5 +1,5 @@ Name: async-pool-Version: 0.8.0+Version: 0.9.0 Synopsis: A modified version of async that supports worker groups and many-to-many task dependencies License-file: LICENSE License: MIT@@ -27,7 +27,7 @@ , containers , transformers , transformers-base- , monad-control+ , monad-control >= 1.0 && < 1.1 exposed-modules: Control.Concurrent.Async.Pool other-modules: