Data.TaskPool

Documentation

data Pool a

A Pool manages a collection of possibly interdependent tasks, such that tasks await execution until the tasks they depend on have finished (and tasks may depend on an arbitrary number of other tasks), while independent tasks execute concurrently up to the number of available resource slots in the pool.

Results from each task are available until the status of the task is polled or waited on. Further, the results are kept until that occurs, so failing to ever wait will result in a memory leak.

Tasks may be cancelled, in which case all dependent tasks are unscheduled.

type Handle = Node

A Handle is a unique reference to a task that has submitted to a Pool.

createPool

Arguments

:: Int

Maximum number of running tasks.

-> IO (Pool a)

Create a thread pool for executing interdependent tasks concurrently. The number of available slots governs how many tasks may run at one time.

runPool :: Pool a -> IO ()

Begin executing tasks in the given pool. The number of slots determines how many threads may execute concurrently. This number is adjustable dynamically, by calling setPoolSlots, though reducing it does not cause already active threads to stop.

setPoolSlots :: Pool a -> Int -> STM ()

Set the number of available execution slots in the given Pool. Increasing the number will cause waiting threads to start executing immediately, while decreasing the number only decreases any available slots -- it does not cancel already executing threads.

cancelAll :: Pool a -> IO ()

Cancel every running thread in the pool and unschedule any that had not begun yet.

submitTask :: Pool a -> IO a -> STM Handle

Submit an IO action for execution within the managed thread pool. When it actually begins executes is determined by the number of available slots, whether the threaded runtime is being used, and how long it takes the jobs before it to complete.

submitTask_ :: Pool a -> IO a -> STM Handle

Submit an 'IO ()' action, where we will never care about the result value or if an exception occurred within the task. This means its process table entry is automatically cleared immediately upon completion of the task. Use this if you are doing your own result propagation, such as writing to a TChan within the task.

submitDependentTask :: Pool a -> Handle -> IO a -> STM Handle

Submit a task, but only allow it begin executing once its parent task has completed. This is equivalent to submitting a new task and linking it to its parent using sequenceTasks within a single STM transaction.

submitDependentTask_ :: Pool a -> Handle -> IO a -> STM Handle

Submit a dependent task where we do not care about the result value or if an exception occurred. See submitTask_.

cancelTask :: Pool a -> Handle -> IO ()

Cancel a task submitted to the pool. This will unschedule it if it had not begun yet, or cancel its thread if it had.

sequenceTasks

Arguments

:: Pool a

-> Handle

Task we must wait on (the parent)

-> Handle

Task doing the waiting

-> STM ()

Given parent and child task handles, link them so that the child cannot execute until the parent has finished.

waitTask :: Pool a -> Handle -> STM a

Wait until the given task is completed, but re-raise any exceptions that were raised in the task's thread.

waitTaskEither :: Pool a -> Handle -> STM (Either SomeException a)

Wait until the given task has completed, then return its final status.

pollTask :: Pool a -> Handle -> STM (Maybe a)

Poll the given task, as with pollTaskEither, but re-raise any exceptions that were raised in the task's thread.

pollTaskEither :: Pool a -> Handle -> STM (Maybe (Either SomeException a))

Poll the given task, returning Nothing if it hasn't started yet or is currently executing, and a Just value if a final result is known.

mapTasks :: Traversable t => Int -> t (IO a) -> IO (t a)

Run a group of up to N tasks at a time concurrently, returning the results in order. The order of execution is random, but the results are returned in order.

mapTasksE :: Traversable t => Int -> t (IO a) -> IO (t (Either SomeException a))

Run a group of up to N tasks at a time concurrently, returning the results in order. The order of execution is random, but the results are returned in order.

mapTasks_ :: Foldable t => Int -> t (IO a) -> IO ()

Run a group of up to N tasks at a time concurrently, ignoring the results.

mapTasksE_ :: Traversable t => Int -> t (IO a) -> IO (t (Maybe SomeException))

Run a group of up to N tasks at a time concurrently, ignoring the results, but returning whether an exception occurred for each task.

mapTasksRace :: Traversable t => Int -> t (IO a) -> IO (Maybe (Either SomeException a))

Execute a group of tasks (where only N tasks at most may run, corresponding to the number of available slots in the pool), returning the first result or failure. Nothing is returned if no tasks were provided.