From 963348611a1f4d027d2307190b0a04359d45d309 Mon Sep 17 00:00:00 2001 From: Bertrand Benjamin Date: Sat, 9 Apr 2022 21:46:24 +0200 Subject: [PATCH] Feat: scheduler don't manage actions --- bopytex/scheduler.py | 40 ++++++---------- test/test_scheduler.py | 106 +++++++++++++++++++++-------------------- 2 files changed, 70 insertions(+), 76 deletions(-) diff --git a/bopytex/scheduler.py b/bopytex/scheduler.py index 0e22d98..212c726 100644 --- a/bopytex/scheduler.py +++ b/bopytex/scheduler.py @@ -5,8 +5,9 @@ from bopytex.tasks import Task class Scheduler: - def __init__(self, actions: dict, done: list[str] = None): - self.actions = actions + """Scheduler is responsible of getting tasks (the tasks) and yield those that can be done""" + + def __init__(self, done: list[str] = None): if done is None: self._done = [] @@ -17,12 +18,12 @@ class Scheduler: @property def tasks(self) -> list[Task]: - """ list all the tasks """ + """List all the tasks todo""" return self._tasks @property def doable_tasks(self) -> list[Task]: - """ list all doable tasks """ + """List all doable tasks""" return [ task for task in self.tasks @@ -31,12 +32,12 @@ class Scheduler: @property def all_deps(self) -> list[str]: - """ List dependencies of all tasks """ + """List dependencies of all tasks""" return {d for task in self.tasks for d in task.deps} @property def all_output(self) -> list[str]: - """ List ouput of all tasks """ + """List ouput of all tasks""" return {task.output for task in self.tasks} @property @@ -46,31 +47,20 @@ class Scheduler: def append(self, tasks: list[Task]): self._tasks += tasks - def dispatch(self, task): - """Do a task""" - ans = self.actions[task.action](task.deps, task.args, task.output) - return ans + def is_finishable(self): + return self.all_deps.issubset(self.all_output) - def __iter__(self): - return self - - def __next__(self): - return self.next() - - def next(self): + def next_task(self): try: task = self.doable_tasks[0] except IndexError: raise StopIteration - ans = self.dispatch(task) self._done.append(task.output) self._tasks.remove(task) - return ans + return task - def run(self): - for _ in self: - pass - - def is_finishable(self): - return self.all_deps.issubset(self.all_output) + def backlog(self): + """ Yield tasks sorted according to dependencies """ + while self.doable_tasks: + yield self.next_task() diff --git a/test/test_scheduler.py b/test/test_scheduler.py index cca528b..4ac1475 100644 --- a/test/test_scheduler.py +++ b/test/test_scheduler.py @@ -3,18 +3,11 @@ from bopytex.scheduler import Scheduler import pytest -def action_done(deps, args, output): - return f"{deps} - {args} - {output} - done" - - -actions = {"DO": action_done} - - def test_schedule_append(): - scheduler = Scheduler(actions) + scheduler = Scheduler() tasks = [ - Task(action="DO", args={}, deps=["dep1", "dep2"], output="end1"), - Task(action="DO", args={}, deps=["dep1", "dep3"], output="end2"), + Task(action="FOO", args={}, deps=["dep1", "dep2"], output="end1"), + Task(action="FOO", args={}, deps=["dep1", "dep3"], output="end2"), ] scheduler.append(tasks) assert scheduler.tasks == tasks @@ -22,104 +15,115 @@ def test_schedule_append(): assert scheduler.all_output == {"end1", "end2"} -def test_schedule_dispatch(): - scheduler = Scheduler(actions) - task = Task(action="DO", args={}, deps=[], output="end") - result = scheduler.dispatch(task) - assert result == "[] - {} - end - done" - - def test_schedule_one_task(): - scheduler = Scheduler(actions) - tasks = [Task(action="DO", args={}, deps=[], output="end")] + scheduler = Scheduler() + tasks = [Task(action="FOO", args={}, deps=[], output="end")] scheduler.append(tasks) + assert scheduler.doable_tasks == tasks - result = scheduler.next() - assert result == "[] - {} - end - done" + + result = scheduler.next_task() + assert result == tasks[0] assert scheduler.tasks == [] assert scheduler.done == ["end"] def test_schedule_one_task_with_args(): - scheduler = Scheduler(actions) - scheduler.append([Task(action="DO", args={"task": "one"}, deps=[], output="one")]) - result = scheduler.next() - assert result == "[] - {'task': 'one'} - one - done" + scheduler = Scheduler() + tasks = [Task(action="FOO", args={"task": "one"}, deps=[], output="one")] + scheduler.append(tasks) + + result = scheduler.next_task() + + assert result == tasks[0] assert scheduler.tasks == [] assert scheduler.done == ["one"] def test_schedule_multiple_tasks(): - scheduler = Scheduler(actions) - t1 = Task(action="DO", args={"task": "one"}, deps=[], output="one") - t2 = Task(action="DO", args={"task": "two"}, deps=[], output="two") - t3 = Task(action="DO", args={"task": "three"}, deps=[], output="three") + scheduler = Scheduler() + t1 = Task(action="FOO", args={"task": "one"}, deps=[], output="one") + t2 = Task(action="FOO", args={"task": "two"}, deps=[], output="two") + t3 = Task(action="FOO", args={"task": "three"}, deps=[], output="three") scheduler.append([t1, t2, t3]) assert scheduler.doable_tasks == [t1, t2, t3] assert scheduler.is_finishable() - result = scheduler.next() - assert result == "[] - {'task': 'one'} - one - done" + result = scheduler.next_task() + assert result == t1 assert scheduler.tasks == [t2, t3] assert scheduler.done == ["one"] - result = scheduler.next() - assert result == "[] - {'task': 'two'} - two - done" + result = scheduler.next_task() + assert result == t2 assert scheduler.tasks == [t3] assert scheduler.done == ["one", "two"] - result = scheduler.next() - assert result == "[] - {'task': 'three'} - three - done" + result = scheduler.next_task() + assert result == t3 assert scheduler.tasks == [] assert scheduler.done == ["one", "two", "three"] def test_schedule_multiple_tasks_with_dependencies(): - scheduler = Scheduler(actions) - t1 = Task(action="DO", args={"task": "one"}, deps=["three"], output="one") - t2 = Task(action="DO", args={"task": "two"}, deps=["one"], output="two") - t3 = Task(action="DO", args={"task": "three"}, deps=[], output="three") + scheduler = Scheduler() + t1 = Task(action="FOO", args={"task": "one"}, deps=["three"], output="one") + t2 = Task(action="FOO", args={"task": "two"}, deps=["one"], output="two") + t3 = Task(action="FOO", args={"task": "three"}, deps=[], output="three") scheduler.append([t1, t2, t3]) assert scheduler.doable_tasks == [t3] assert scheduler.is_finishable() - result = scheduler.next() - assert result == "[] - {'task': 'three'} - three - done" + result = scheduler.next_task() + assert result == t3 assert scheduler.tasks == [t1, t2] assert scheduler.done == ["three"] assert scheduler.doable_tasks == [t1] - result = scheduler.next() - assert result == "['three'] - {'task': 'one'} - one - done" + result = scheduler.next_task() + assert result == t1 assert scheduler.tasks == [t2] assert scheduler.done == ["three", "one"] assert scheduler.doable_tasks == [t2] - result = scheduler.next() - assert result == "['one'] - {'task': 'two'} - two - done" + result = scheduler.next_task() + assert result == t2 assert scheduler.tasks == [] assert scheduler.done == ["three", "one", "two"] +def test_schedule_multiple_tasks_with_dependencies_loop(): + scheduler = Scheduler() + t1 = Task(action="FOO", args={"task": "one"}, deps=["three"], output="one") + t2 = Task(action="FOO", args={"task": "two"}, deps=["one"], output="two") + t3 = Task(action="FOO", args={"task": "three"}, deps=[], output="three") + scheduler.append([t1, t2, t3]) + + ordered_tasks = [] + for task in scheduler.backlog(): + ordered_tasks.append(task) + assert ordered_tasks == [t3, t1, t2] def test_schedule_empty_task(): - scheduler = Scheduler(actions) + scheduler = Scheduler() scheduler.append([]) with pytest.raises(StopIteration): - scheduler.next() + scheduler.next_task() def test_schedule_multiple_tasks_with_undoable_dependencies(): - scheduler = Scheduler(actions) - t1 = Task(action="DO", args={"task": "one"}, deps=["three"], output="one") - t2 = Task(action="DO", args={"task": "two"}, deps=[], output="two") + scheduler = Scheduler() + t1 = Task(action="FOO", args={"task": "one"}, deps=["three"], output="one") + t2 = Task(action="FOO", args={"task": "two"}, deps=[], output="two") scheduler.append([t1, t2]) assert scheduler.doable_tasks == [t2] assert not scheduler.is_finishable() - scheduler.run() + for _ in scheduler.backlog(): + pass + assert scheduler.tasks == [t1] assert scheduler.done == ["two"] assert scheduler.doable_tasks == []