from collections.abc import Generator
from typing import Any
from gain import logging
from gain.task_graph.base_executor import TaskGraphExecutorBase
from gain.task_graph.cache import NoTaskCache
from gain.task_graph.graph import Task, TaskGraph
NO_TASK_CACHE = NoTaskCache()
logger = logging.getLogger(__name__)
[docs]
class SequentialExecutor(TaskGraphExecutorBase):
"""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
:class:`~gain.task_graph.base_executor.TaskGraphExecutorBase`:
``task_cache``, ``force`` and the options of the task executor.
"""
def _execute(
self, graph: TaskGraph,
) -> Generator[tuple[Task, Any], None, None]:
finished_tasks = 0
initial_task_count = len(graph)
while not graph.empty():
ready_tasks = graph.extract_tasks(graph.ready_tasks())
for ftask in ready_tasks:
# handle tasks that use the output of other tasks
result = self._exec(ftask, self._params)
graph.process_completed_tasks([(ftask.task, result)])
finished_tasks += 1
logger.debug("clean up task %s", ftask)
logger.info(
"finished %s/%s", finished_tasks,
initial_task_count)
yield ftask.task, result
# all tasks have already executed. Let's clean the state.
assert len(graph) == 0
[docs]
def close(self) -> None:
"""Do nothing: a sequential task executor holds no workers."""