79 lines
2.0 KiB
Python
79 lines
2.0 KiB
Python
""" Scheduler for action to make """
|
|
|
|
|
|
from bopytex.tasks import Task
|
|
from bopytex.worker import Dispatcher
|
|
|
|
|
|
class Scheduler:
|
|
"""Scheduler is responsible of getting tasks (the tasks) and yield those that can be done"""
|
|
|
|
def __init__(self, dispatcher: Dispatcher, output_done: list[str] = None):
|
|
self._dispatcher = dispatcher
|
|
|
|
if output_done is None:
|
|
self._output_done = []
|
|
else:
|
|
self._output_done = output_done
|
|
|
|
self._tasks = []
|
|
self._failed_tasks = []
|
|
|
|
@property
|
|
def tasks(self) -> list[Task]:
|
|
"""List all the tasks todo"""
|
|
return self._tasks
|
|
|
|
@property
|
|
def doable_tasks(self) -> list[Task]:
|
|
"""List all doable tasks"""
|
|
return [
|
|
task
|
|
for task in self.tasks
|
|
if not task.deps or all([d in self.output_done for d in task.deps])
|
|
]
|
|
|
|
@property
|
|
def all_deps(self) -> set[str]:
|
|
"""List dependencies of all tasks"""
|
|
return {d for task in self.tasks for d in task.deps}
|
|
|
|
@property
|
|
def all_output(self) -> set[str]:
|
|
"""List ouput of all tasks"""
|
|
return {task.output for task in self.tasks}
|
|
|
|
@property
|
|
def output_done(self) -> list[str]:
|
|
return self._output_done
|
|
|
|
@property
|
|
def failed_tasks(self) -> list[Task]:
|
|
return self._failed_tasks
|
|
|
|
def append(self, tasks: list[Task]):
|
|
self._tasks += tasks
|
|
|
|
def is_finishable(self):
|
|
return self.all_deps.issubset(self.all_output)
|
|
|
|
def next_task(self):
|
|
try:
|
|
task = self.doable_tasks[0]
|
|
except IndexError:
|
|
raise StopIteration
|
|
|
|
self._tasks.remove(task)
|
|
message = self._dispatcher(task)
|
|
if message.status == 0:
|
|
self._output_done.append(task.output)
|
|
else:
|
|
self._failed_tasks.append(task)
|
|
|
|
return message
|
|
|
|
def backlog(self):
|
|
"""Yield tasks sorted according to dependencies"""
|
|
while self.doable_tasks:
|
|
yield self.next_task()
|