Task executors

A task executor runs a TaskGraph. It runs each task only after the tasks that it depends on, and it asks a task cache which tasks must run. This page describes the three task executors of GAIn, the functions that run a graph with one call, and the task caches.

Creating a task executor

GAIn has three task executors. Each one takes the keyword argument task_cache, the task cache that decides which tasks must run. Without it, the task executor runs every task.

SequentialExecutor(): Runs the tasks one at a time, in the current process. It is in gain.task_graph.sequential_executor.
ProcessPoolTaskExecutor(max_workers=...): Runs the tasks in a pool of processes. max_workers is the number of processes. Without it, the pool has one process for each CPU. It is in gain.task_graph.process_pool_executor.
DaskExecutor(client): Runs the tasks on the dask cluster of client. It is in gain.task_graph.dask_executor.

The three task executors derive from TaskGraphExecutorBase, which also takes these keyword arguments:

force: When true, the task executor runs every task and does not read the task cache. It still records the outcome of each task in the task cache.
task_log_dir: The directory of the task logs. Without it, the task executor writes them into .task-log in the current directory.
verbose: The verbosity of the task logs.
fork_tasks: When true, each task runs in a new process.
task_status_dir: The directory of the result files of the forked tasks. Without it, the task executor writes them into the current directory.

Only SequentialExecutor and DaskExecutor use task_log_dir, verbose, fork_tasks and task_status_dir. ProcessPoolTaskExecutor ignores them: it creates the directory of the task logs, but it writes no task log and it does not fork the tasks.

A process pool or a dask cluster pickles the function and the arguments of each task. Use module-level functions for the tasks, and pass arguments that pickle, for example a resource and not an open genome.

The lifecycle

A task executor is a context manager. The with block calls close() when it ends, and close releases the workers:

from gain.task_graph.sequential_executor import SequentialExecutor

with SequentialExecutor() as executor:
    for task, result in executor.execute(graph):
        print(task.task_id, result)

execute() consumes the graph and yields each task with its result when the task completes. The order is the order of completion, not the order of the graph. The result of a failed task is its exception: execute does not raise it. A task executor runs one graph at a time, so exhaust the generator of a run before you start the next run.

DaskExecutor takes ownership of its client. Its close shuts down the client and the cluster of the client, so do not use the client after the with block.

Running a graph with one call

task_graph_run() and task_graph_run_with_results(), in gain.task_graph.cli_tools, run a graph with one call. Without a task executor, they create a SequentialExecutor.

task_graph_run returns true when no task failed. task_graph_run_with_results yields the result of each task, without the task. Without keep_going, the first failed task raises its error and ends the run. With keep_going=True, the run prints the error of each failed task and continues.

from gain.task_graph.cli_tools import task_graph_run

with SequentialExecutor() as executor:
    assert task_graph_run(graph, executor)

The task cache

A task cache decides which tasks must run and records the outcome of each task that runs. The classes are in gain.task_graph.cache.

NoTaskCache: Runs every task and records nothing. It is the default of each task executor.
FileTaskCache: Writes one record file for each task into the task status directory, cache_dir. The record holds the result of the task, or its exception. The modification times of the files of a task decide whether a computed record is still valid: an input file that is newer than the output files of the task makes the task run again.

TaskCache.create picks the task cache for a set of options. It picks NoTaskCache when force is true or the status mode, task_progress_mode, is off. Otherwise it picks a FileTaskCache. Without cache_dir, the FileTaskCache uses the current directory.

get_record() returns the CacheRecord of a task. The type of the record is a CacheRecordType: NEEDS_COMPUTE, COMPUTED or ERROR. result holds the result of a computed task.

Before a run, execute asks the task cache for the record of each task. It removes each task that the task cache reports as computed from the graph, and it does not yield that task. A task that depends on a task that must run runs again, even when its own record is computed.

A running example, part 2

Part 1, on the page Graphs, builds a graph that counts the N bases of a reference genome. The examples here run that graph. Each run builds a new graph, because a run consumes the graph.

The first run is sequential. task_graph_run_with_results yields the results without their tasks. The summary task depends on all the other tasks, so it completes last, and the last result is the sum:

def run_sequentially(resource: GenomicResource) -> int:
    """Count the ``N`` bases of the genome, one task at a time."""
    graph, _ = build_n_count_graph(resource)
    with SequentialExecutor() as executor:
        results = list(task_graph_run_with_results(graph, executor))
    # The summary task depends on all the other tasks, so it completes last.
    total: int = results[-1]
    return total

The second run uses a pool of two processes. The task functions are module-level, so the pool can pickle them:

def run_in_process_pool(resource: GenomicResource) -> int:
    """Count the ``N`` bases of the genome in a pool of two processes."""
    graph, _ = build_n_count_graph(resource)
    with ProcessPoolTaskExecutor(max_workers=2) as executor:
        results = list(task_graph_run_with_results(graph, executor))
    total: int = results[-1]
    return total

The third run gives the task executor a FileTaskCache over a task status directory. The task cache writes a record file for each task. Then the example builds a fresh graph and runs it with a new FileTaskCache over the same directory. The record type of each task is COMPUTED, the record of the summary task holds the sum, and execute yields nothing:

def run_with_file_task_cache(
    resource: GenomicResource, status_dir: str,
) -> None:
    """Run the graph and write a record for each task into ``status_dir``."""
    graph, _ = build_n_count_graph(resource)
    task_cache = FileTaskCache(cache_dir=status_dir)
    with SequentialExecutor(task_cache=task_cache) as executor:
        assert task_graph_run(graph, executor)


def rerun_with_file_task_cache(
    resource: GenomicResource, status_dir: str,
) -> tuple[list[tuple[Task, Any]], dict[str, CacheRecord]]:
    """Run a fresh graph over the records of an earlier run."""
    graph, _ = build_n_count_graph(resource)
    task_cache = FileTaskCache(cache_dir=status_dir)
    records = {
        task.task_id: task_cache.get_record(graph.get_task_desc(task))
        for task in graph.tasks
    }
    with SequentialExecutor(task_cache=task_cache) as executor:
        completed = list(executor.execute(graph))
    return completed, records

API

Each class below is defined in the module that the section names. task_graph_run and task_graph_run_with_results are defined in the module gain.task_graph.cli_tools.

`TaskGraphExecutor`()	The interface of a task executor.
`TaskGraphExecutor.close`()	Release the workers and the other resources of the task executor.
`TaskGraphExecutor.execute`(graph)	Run the tasks of a task graph.

`TaskGraphExecutorBase`([task_cache, force])	The base of the task executors of GAIn.
`TaskGraphExecutorBase.execute`(graph)	Run the tasks of a task graph that must run.

`SequentialExecutor`([task_cache, force])	A task executor that runs the tasks one at a time.
`SequentialExecutor.close`()	Do nothing: a sequential task executor holds no workers.

`ProcessPoolTaskExecutor`(**kwargs)	A task executor that runs the tasks in a pool of processes.
`ProcessPoolTaskExecutor.close`()	Shut the pool down and wait for its processes to exit.

`DaskExecutor`(dask_client[, task_cache])	A task executor that runs the tasks on a dask cluster.
`DaskExecutor.close`()	Shut down the dask client and its cluster.

Inherited from TaskGraphExecutorBase by the three task executors:

TaskGraphExecutorBase.execute(graph)

Run the tasks of a task graph that must run.

`TaskCache`()	The interface of a task cache.
`TaskCache.cache`(task, *, is_error, result)	Record the outcome of a task.
`TaskCache.create`(*[, force, ...])	Create the task cache for a set of options.
`TaskCache.get_record`(task_desc)	Return the record of a task.
`FileTaskCache`(cache_dir)	A task cache that keeps one record file for each task.
`FileTaskCache.cache`(task, *, is_error, result)	Write the record file of a task.
`FileTaskCache.get_record`(task_desc)	Return the record of a task from its record file.
`NoTaskCache`	A task cache that runs every task and records nothing.
`NoTaskCache.cache`(task, *, is_error, result)	Do nothing: the task cache records no outcome.
`NoTaskCache.get_record`(task_desc)	Return a record whose type is `NEEDS_COMPUTE`.
`CacheRecord`(type[, result_or_error])	The record of one task in a task cache.
`CacheRecord.error`	The exception of a task whose record type is `ERROR`.
`CacheRecord.result`	The result of a task whose record type is `COMPUTED`.
`CacheRecordType`(*values)	The state of a task in a task cache.

Inherited from TaskCache by FileTaskCache and NoTaskCache:

TaskCache.create(*[, force, ...])

Create the task cache for a set of options.

`task_graph_run`(task_graph[, executor, ...])	Run a task graph with a task executor.
`task_graph_run_with_results`(task_graph[, ...])	Run a task graph and yield the result of each task.

Task executors

class TaskGraphExecutor[source]

The interface of a task executor.

A task executor runs a task graph. It is a context manager: the with block closes the task executor when it ends, and so releases its workers. A failure in close() is logged and does not hide an error of the with block.

The task executors of GAIn are SequentialExecutor, ProcessPoolTaskExecutor and DaskExecutor.

abstractmethod close() → None[source]

Release the workers and the other resources of the task executor.

The with block of a task executor calls this method when it ends.

abstractmethod execute(graph: TaskGraph) → Generator[tuple[Task, Any], None, None][source]

Run the tasks of a task graph.

The task executor consumes the graph: the graph has no tasks left when the run ends. Build a new graph for each run.

Parameters:: graph (TaskGraph) – The task graph to run.
Yields:: tuple[Task, Any] – Each task and its result, when the task completes. The result of a failed task is its exception. The order is not the order of the graph, and not the order in which the tasks run. A task is yielded only after it has finished.

class TaskGraphExecutorBase(task_cache: TaskCache = {}, *, force: bool = False, **kwargs: Any)[source]

The base of the task executors of GAIn.

The base runs each task only after the tasks that it depends on, and it asks a task cache which tasks must run. It records the result or the error of each task that runs in the task cache.

Parameters:

task_cache (TaskCache) – The task cache that decides which tasks must run. The default is a NoTaskCache, which runs every task.
force (bool) – When true, the task executor runs every task and does not read the task cache. It still writes the outcome of each task to it.
**kwargs – The options of the task executor. task_log_dir names the directory of the task logs. Without it, the task executor writes them into .task-log in the current directory. verbose sets the verbosity of the task logs. fork_tasks runs each task in a new process, and task_status_dir names the directory of the result files of those processes.

execute(graph: TaskGraph) → Generator[tuple[Task, Any], None, None][source]

Run the tasks of a task graph that must run.

The method first asks the task cache which tasks are computed, and removes them from the graph. It does not yield them. Then it runs the other tasks, records the outcome of each one in the task cache and yields it. When the task cache reports each task as computed, the method yields nothing.

A task executor runs one graph at a time. Exhaust or close the generator of a run before you start the next run.

Parameters:: graph (TaskGraph) – The task graph to run. The method consumes it.
Yields:: tuple[Task, Any] – Each task that ran and its result, when the task completes. The result of a failed task is its exception.

class SequentialExecutor(task_cache: TaskCache = {}, *, force: bool = False, **kwargs: Any)[source]

A task executor that runs the tasks one at a time.

A task runs only after the tasks that it depends on. The tasks run in the current process, unless the fork_tasks option is true: then each task runs in a new process. The constructor takes the keyword arguments of TaskGraphExecutorBase: task_cache, force and the options of the task executor.

close() → None[source]: Do nothing: a sequential task executor holds no workers.

class ProcessPoolTaskExecutor(**kwargs: Any)[source]

A task executor that runs the tasks in a pool of processes.

The constructor creates the pool. The pool starts its worker processes when the task executor submits the first task. A task runs in a worker process only after the tasks that it depends on. The task executor pickles the function and the arguments of each task, so the function must be importable, for example a module-level function.

Parameters:: **kwargs – max_workers is the number of processes in the pool. Without it, the pool has one process for each CPU. The other keyword arguments are those of TaskGraphExecutorBase: task_cache, force and the options of the task executor. The process pool ignores the options task_log_dir, verbose and fork_tasks: it creates the directory of the task logs, but it writes no task log and it does not fork the tasks.

close() → None[source]: Shut the pool down and wait for its processes to exit.

class DaskExecutor(dask_client: Client, task_cache: TaskCache = {}, **kwargs: Any)[source]

A task executor that runs the tasks on a dask cluster.

The task executor takes ownership of the dask client: close() shuts down the client and its cluster. A task runs on a worker of the cluster only after the tasks that it depends on.

Parameters:

dask_client (Client) – The dask client of the cluster that runs the tasks.
task_cache (TaskCache) – The task cache that decides which tasks must run. The default is a NoTaskCache, which runs every task.
**kwargs – The other keyword arguments of TaskGraphExecutorBase: force and the options of the task executor.

Variables:

MAX_RUNNING_TASKS (int) – The task executor takes new tasks from the graph only while fewer than this number of its tasks are unfinished.

close() → None[source]: Shut down the dask client and its cluster.

Running a graph

task_graph_run(task_graph: TaskGraph, executor: TaskGraphExecutor | None = None, *, keep_going: bool = False) → bool[source]

Run a task graph with a task executor.

Parameters:

task_graph (TaskGraph) – The task graph to run. The run consumes it.
executor (TaskGraphExecutor | None) – The task executor that runs the graph. None means a new SequentialExecutor.
keep_going (bool) – When false, the first failed task raises its error and ends the run. When true, the run prints the error of each failed task to the standard error and continues.

Returns:

True when no task failed.

Return type:

bool

Raises:

Exception – The error of the first failed task, when keep_going is false.

task_graph_run_with_results(task_graph: TaskGraph, executor: TaskGraphExecutor | None = None, *, keep_going: bool = False) → Generator[Any, None, None][source]

Run a task graph and yield the result of each task.

The function yields the results without their tasks, in the order in which the tasks complete. A task that the task cache reports as computed does not run, and the function yields no result for it.

Parameters:

task_graph (TaskGraph) – The task graph to run. The run consumes it.
executor (TaskGraphExecutor | None) – The task executor that runs the graph. None means a new SequentialExecutor.
keep_going (bool) – When false, the first failed task raises its error and ends the run. When true, the function prints the error of each failed task to the standard error and yields the error as the result.

Yields:

Any – The result of each task, or the exception of a failed task.

Raises:

Exception – The error of the first failed task, when keep_going is false.

Task caches

class TaskCache[source]

The interface of a task cache.

A task executor asks the task cache whether a task must run, and records the outcome of each task that runs in it.

abstractmethod cache(task: Task, *, is_error: bool, result: Any) → None[source]

Record the outcome of a task.

Parameters:

task (Task) – The task that ran.
is_error (bool) – True when the task failed.
result (Any) – The result of the task, or its exception when it failed.

static create(*, force: bool = False, task_progress_mode: bool = True, cache_dir: str | None = None) → TaskCache[source]

Create the task cache for a set of options.

Parameters:

force (bool) – When true, the method returns a NoTaskCache.
task_progress_mode (bool) – The status mode. When false, the method returns a NoTaskCache.
cache_dir (str | None) – The task status directory of a FileTaskCache. None means the current directory.

Returns:

A NoTaskCache when force is true or the status mode is off. Otherwise a FileTaskCache over cache_dir.

Return type:

TaskCache

abstractmethod get_record(task_desc: TaskDesc) → CacheRecord[source]

Return the record of a task.

Parameters:: task_desc (TaskDesc) – The description of the task, from get_task_desc().
Returns:: The record of the task. The task must run unless the record type is COMPUTED. NEEDS_COMPUTE means that the task cache has no valid record of the task.
Return type:: CacheRecord

class FileTaskCache(cache_dir: str)[source]

A task cache that keeps one record file for each task.

The record file of a task is <task id>.flag in the task status directory. It holds the record type and the result or the exception of the task. The modification times of the files of a task decide whether a computed record is still valid.

Parameters:: cache_dir (str) – The task status directory. The task cache creates it when it writes the first record.

cache(task: Task, *, is_error: bool, result: Any) → None[source]

Write the record file of a task.

A failure to write the file is logged, and the run continues.

Parameters:

task (Task) – The task that ran.
is_error (bool) – True when the task failed. The record type is then ERROR, otherwise COMPUTED.
result (Any) – The result of the task, or its exception when it failed. It must be picklable.

get_record(task_desc: TaskDesc) → CacheRecord[source]

Return the record of a task from its record file.

A computed record stays valid while the output files of the task exist and are not older than its input files. A missing input file makes the record invalid. A missing intermediate output file does not, because a later task can consume it. A task without output files compares its input files with its record file. A missing or unreadable record file means that the task must run.

Parameters:: task_desc (TaskDesc) – The description of the task, with its input and output files.
Returns:: The record of the task. Its type is NEEDS_COMPUTE when the record file is missing or unreadable, or when the record is computed but no longer valid. Otherwise it is the record that the file holds, COMPUTED or ERROR.
Return type:: CacheRecord

class NoTaskCache[source]

A task cache that runs every task and records nothing.

cache(task: Task, *, is_error: bool, result: Any) → None[source]

Do nothing: the task cache records no outcome.

Parameters:

task (Task) – The task that ran.
is_error (bool) – True when the task failed.
result (Any) – The result of the task, or its exception when it failed.

get_record(task_desc: TaskDesc) → CacheRecord[source]

Return a record whose type is NEEDS_COMPUTE.

Parameters:: task_desc (TaskDesc) – The description of the task. The method does not read it.
Returns:: A record whose type is NEEDS_COMPUTE and that holds no result.
Return type:: CacheRecord

class CacheRecord(type: CacheRecordType, result_or_error: Any = None)[source]

The record of one task in a task cache.

Variables:

type (CacheRecordType) – The state of the task.
result_or_error (Any) – The result of the task, the exception of a failed task, or None when the task cache holds no outcome.

property error: Any: The exception of a task whose record type is ERROR.

property result: Any: The result of a task whose record type is COMPUTED.

class CacheRecordType(*values)[source]

The state of a task in a task cache.

Variables:

NEEDS_COMPUTE (CacheRecordType) – The task must run.
COMPUTED (CacheRecordType) – The task ran without an error, and its record is still valid.
ERROR (CacheRecordType) – The task ran and failed.