Graphs

A TaskGraph holds the tasks of one run and the dependencies between them. This page describes how to build a task graph and how a task executor consumes it. The examples import the classes from gain.task_graph.graph.

Creating a task graph

Create an empty graph with TaskGraph(). Then add each task with create_task(). The call takes the id of the task, the function that the task calls, and these keyword arguments:

args and kwargs: The positional and keyword arguments of the function.
deps: The tasks that must finish before this task runs.
input_files, output_files and intermediate_output_files: The files that the task reads and writes. A file task cache uses them to decide whether the task must run again.

create_task returns a Task, the handle of the new task. When a task runs, each Task in its args or kwargs is replaced by the result of that task. The graph infers that dependency, so you do not have to repeat the task in deps. A task that is only in deps passes no value: the task waits for it and gets nothing from it.

from gain.task_graph.graph import TaskGraph

graph = TaskGraph()
prepare = graph.create_task("prepare", prepare_dir, args=[out_dir])
part = graph.create_task(
    "part", compute, args=[region], deps=[prepare],
    output_files=[f"{out_dir}/part.txt"])
graph.create_task("merge", merge, args=[part])

Here merge gets the result of part as its argument. part waits for prepare but does not get its result.

chain_tasks() combines several task descriptions into one task that runs their functions in order, in one process. Each task after the first depends on no task, or only on the task before it, and gets the result of that task.

The lifecycle

You build a graph and give it to a task executor. The task executor runs the tasks and consumes the graph: it removes each task from the graph when it schedules the task. Build a new graph for each run.

Before you give the graph to a task executor, you can reduce it. prune() keeps the named task ids and the tasks that they depend on, and removes every other task:

graph.prune(["merge"])

topological_order() lists the tasks so that each task comes after its deps. empty() tells whether the graph has tasks left.

Task descriptions

The second form builds a task without a graph. make_task() takes the same arguments as create_task and returns a TaskDesc. add_task() adds one description to a graph, and add_tasks() adds a list of them.

This form suits code that returns tasks to a caller that owns the graph. The create_statistics_build_tasks method of a resource implementation returns a list of TaskDesc, and grr_manage adds it to its graph with add_tasks. Adding new resource types describes that method.

A running example

The example counts the N bases of a reference genome. A module-level function takes a GenomicResource and a region. It opens the reference genome inside the function and counts the N bases. A second function sums the counts:

def count_n(
    resource: GenomicResource, chrom: str, start: int, end: int,
) -> int:
    """Count the ``N`` bases of a region of the genome in ``resource``."""
    genome = build_reference_genome_from_resource(resource)
    with genome.open():
        return genome.get_sequence(chrom, start, end).count("N")


def sum_counts(*counts: int) -> int:
    """Return the sum of the counts of the chromosome tasks."""
    return sum(counts)

The graph builder adds one task for each chromosome. The summary task takes the chromosome tasks as args, so it depends on each of them and gets their counts:

def build_n_count_graph(
    resource: GenomicResource,
) -> tuple[TaskGraph, Task]:
    """Build a task graph that counts the ``N`` bases of a genome."""
    genome = build_reference_genome_from_resource(resource)
    with genome.open():
        lengths = genome.get_all_chrom_lengths()

    graph = TaskGraph()
    chrom_tasks = [
        graph.create_task(
            f"count_n_{chrom}", count_n,
            args=[resource, chrom, 1, length])
        for chrom, length in lengths.items()
    ]
    summary = graph.create_task("sum_n", sum_counts, args=chrom_tasks)
    return graph, summary

The tasks pass the resource, not an open genome. A task executor can run a task in another process or on another host, so the function opens the genome where it runs. The statistics tasks of the reference genome implementation use the same pattern: they pass the resource in args.

API

The classes below are defined in the module gain.task_graph.graph, and so is chain_tasks. Task and TaskGraph are also exported by gain.task_graph.

`chain_tasks`(*tasks)	Combine task descriptions into one task that runs them in order.
`TaskGraph`()	A directed acyclic graph of tasks.
`TaskGraph.add_task`(task_desc)	Add a task description to the graph.
`TaskGraph.add_tasks`(task_descs)	Add several task descriptions to the graph.
`TaskGraph.create_task`(task_id, func, *, args)	Create a task and add it to the graph.
`TaskGraph.empty`()	Check whether the graph has no tasks.
`TaskGraph.get_task_desc`(task)	Return the description of a task of the graph.
`TaskGraph.has_task`(task)	Check whether the graph has a task.
`TaskGraph.make_task`(task_id, func, *, args)	Build a task description without adding it to a graph.
`TaskGraph.prune`(tasks_to_keep)	Keep only the named tasks and the tasks that they depend on.
`TaskGraph.tasks`	The tasks of the graph, in the order they were added.
`TaskGraph.topological_order`()	Return the tasks so that each task comes after its deps.
`TaskDesc`(task, func, args, kwargs, deps, ...)	Describe one task, before or after it is added to a task graph.
`Task`(task_id)	Identify one task of a task graph.

chain_tasks(*tasks: TaskDesc) → TaskDesc[source]

Combine task descriptions into one task that runs them in order.

The combined task calls the function of each task in turn. When a task depends on the task before it, it gets the result of that task in place of the Task handle.

Parameters:: *tasks (TaskDesc) – The task descriptions, in the order to run them. Each task after the first depends on no task, or only on the task before it.
Returns:: One task description with the id of the last task, the deps of the first task, and the files of all the tasks. Its result is the result of the last task.
Return type:: TaskDesc
Raises:: ValueError – If tasks is empty, or a task depends on a task other than the one before it.

class TaskGraph[source]

A directed acyclic graph of tasks.

Build the graph with create_task, or with make_task and add_task. Then give the graph to a task executor, which runs the tasks and consumes the graph.

Variables:: input_files (list[str]) – The files that were used to build the graph. get_task_desc adds them to the input files of each task, so a file task cache runs each task again when one of them changes. The list is empty in a new graph.

add_task(task_desc: TaskDesc) → Task[source]

Add a task description to the graph.

Parameters:: task_desc (TaskDesc) – The description, usually from make_task.
Returns:: The handle of the task.
Return type:: Task
Raises:: ValueError – If the graph already has a task with the same id.

add_tasks(task_descs: Sequence[TaskDesc]) → list[Task][source]

Add several task descriptions to the graph.

Parameters:: task_descs (Sequence[TaskDesc]) – The descriptions, for example the result of a create_statistics_build_tasks call.
Returns:: The handles of the tasks, in the order of task_descs.
Return type:: list[Task]
Raises:: ValueError – If the graph already has a task with the id of one of the descriptions, or if two descriptions in task_descs have the same id. In the first case, the graph is unchanged. In the second case, the graph keeps the descriptions that come before the second description with that id.

create_task(task_id: str, func: Callable[[...], Any], *, args: Sequence[Any], kwargs: dict[str, Any] | None = None, deps: Sequence[Task] | None = None, input_files: Sequence[str] | None = None, output_files: Sequence[str] | None = None, intermediate_output_files: Sequence[str] | None = None) → Task[source]

Create a task and add it to the graph.

The call is make_task followed by add_task.

Parameters:

task_id (str) – The id of the task, unique within the graph. An id longer than 200 characters is cut to 200.
func (Callable[..., Any]) – The function that the task calls. A task executor can run it in another process, so use a module-level function.
args (Sequence[Any]) – The positional arguments of func. When the task runs, each Task in args is replaced by the result of that task.
kwargs (dict[str, Any] or None) – The keyword arguments of func. When the task runs, each Task value is replaced by the result of that task. None means no keyword arguments.
deps (Sequence[Task] or None) – The tasks that must finish before this task runs. The graph adds each Task of args and kwargs to them. A task that is only in deps passes no value. None means no other deps.
input_files (Sequence[str] or None) – The files that the task reads. A file task cache runs the task again when one of them is newer than the outputs of the task. None means none.
output_files (Sequence[str] or None) – The final output files of the task. A file task cache runs the task again when one of them is missing. None means none.
intermediate_output_files (Sequence[str] or None) – The output files that later tasks consume. A missing one does not make a file task cache run the task again. None means none.

Returns:

The handle of the new task.

Return type:

Task

Raises:

ValueError – If the graph already has a task with the same id.

empty() → bool[source]

Check whether the graph has no tasks.

Returns:: True when the graph has no tasks.
Return type:: bool

get_task_desc(task: Task) → TaskDesc[source]

Return the description of a task of the graph.

Parameters:: task (Task) – The handle of the task.
Returns:: The description. Its input_files start with the input_files of the graph.
Return type:: TaskDesc
Raises:: ValueError – If the graph has no such task.

has_task(task: Task) → bool[source]

Check whether the graph has a task.

Parameters:: task (Task) – The handle of the task.
Returns:: True when the graph has the task.
Return type:: bool

static make_task(task_id: str, func: Callable[[...], Any], *, args: Sequence[Any], kwargs: dict[str, Any] | None = None, deps: Sequence[Task] | None = None, input_files: Sequence[str] | None = None, output_files: Sequence[str] | None = None, intermediate_output_files: Sequence[str] | None = None) → TaskDesc[source]

Build a task description without adding it to a graph.

The parameters are the parameters of create_task. Add the description to a graph with add_task or add_tasks, or combine several with chain_tasks.

Parameters:

task_id (str) – The id of the task. An id longer than 200 characters is cut to 200.
func (Callable[..., Any]) – The function that the task calls.
args (Sequence[Any]) – The positional arguments of func.
kwargs (dict[str, Any] or None) – The keyword arguments of func. None means none.
deps (Sequence[Task] or None) – The tasks that must finish first. None means none.
input_files (Sequence[str] or None) – The files that the task reads. None means none.
output_files (Sequence[str] or None) – The final output files. None means none.
intermediate_output_files (Sequence[str] or None) – The output files that later tasks consume. None means none.

Returns:

The description. Its deps hold deps and each Task of args and kwargs.

Return type:

TaskDesc

prune(tasks_to_keep: Sequence[Task | str]) → None[source]

Keep only the named tasks and the tasks that they depend on.

The method removes every other task from the graph.

Parameters:: tasks_to_keep (Sequence[Task | str]) – The tasks to keep, as handles or as task ids.
Raises:: ValueError – If the graph has no task with one of the ids. The graph is then unchanged.

property tasks: Sequence[Task]: The tasks of the graph, in the order they were added.

topological_order() → Sequence[Task][source]

Return the tasks so that each task comes after its deps.

Returns:: The handles of all the tasks. An empty graph gives an empty list.
Return type:: Sequence[Task]
Raises:: ValueError – If the deps of the tasks form a cycle.

class TaskDesc(task: Task, func: Callable[[...], Any], args: list[Any], kwargs: dict[str, Any], deps: list[Task], input_files: list[str], output_files: list[str], intermediate_output_files: list[str])[source]

Describe one task, before or after it is added to a task graph.

TaskGraph.make_task builds a TaskDesc and TaskGraph.add_task adds it to a graph. TaskGraph.get_task_desc returns the description of a task that is in a graph.

Variables:

task (Task) – The handle of the task.
func (Callable[..., Any]) – The function that the task calls when it runs.
args (list[Any]) – The positional arguments of func. When the task runs, each Task in the list is replaced by the result of that task.
kwargs (dict[str, Any]) – The keyword arguments of func. When the task runs, each Task value is replaced by the result of that task.
deps (list[Task]) – The tasks that must finish before this task runs. The list holds each Task of args and kwargs too.
input_files (list[str]) – The files that the task reads. A file task cache runs the task again when one of them is newer than the outputs of the task.
output_files (list[str]) – The final output files of the task. A file task cache runs the task again when one of them is missing.
intermediate_output_files (list[str]) – The output files that later tasks consume. A missing one does not make a file task cache run the task again.

class Task(task_id: str)[source]

Identify one task of a task graph.

A Task is the handle that TaskGraph.create_task and TaskGraph.add_task return. Pass it in args, kwargs or deps of another task to make that task depend on it. Two handles with the same id are equal.

Variables:: task_id (str) – The id of the task, unique within its task graph.