Implement prioritization of tasks. (#52)

Author: tobiasraabe

docs/changes.rst

@@ -15,6 +15,8 @@ all releases are available on `Anaconda.org <https://anaconda.org/pytask/pytask>
 - :gh:`50` implements correct usage of singular and plural in collection logs.
 - :gh:`51` allows to invoke pytask through the Python interpreter with ``python -m
   pytask`` which will add the current path to ``sys.path``.
+- :gh:`52` allows to prioritize tasks with ``pytask.mark.try_last`` and
+  ``pytask.mark.try_first``.
 - :gh:`53` changes the theme of the documentation to furo.
 
 

docs/how_to_guides/how_to_influence_build_order.rst

@@ -0,0 +1,88 @@
+How to influence the build order
+================================
+
+.. important::
+
+    This guide shows how to influence the order in which tasks are executed. The feature
+    should be treated with caution since it might make projects work whose dependencies
+    and products are not fully specified.
+
+You can influence the order in which tasks are executed by assigning preferences to
+tasks. Use ``pytask.mark.try_first`` to execute a task as early as possible and
+``pytask.mark.try_last`` to defer execution.
+
+.. note::
+
+    For more background, tasks, dependencies and products form a directed acyclic graph
+    (DAG). A `topological ordering <https://en.wikipedia.org/wiki/Topological_sorting>`_
+    determines the order in which tasks are executed such that tasks are not run until
+    their predecessors have been executed. The ordering is not unique and instead of a
+    random ordering, an ordering is chosen which is compatible with the preferences.
+    Among multiple tasks which should all be preferred or delayed, a random task is
+    chosen.
+
+As an example, here are two tasks where the decorator ensures that the output of the
+second task is always shown before the output of the first task.
+
+.. code-block:: python
+
+    # Content of task_example.py
+
+    import pytask
+
+
+    def task_first():
+        print("I'm second.")
+
+
+    @pytask.mark.try_first
+    def task_second():
+        print("I'm first.")
+
+
+The execution yields (use ``-s`` to make the output visible in the terminal)
+
+.. code-block:: console
+
+    $ pytask -s task_example.py
+    ========================= Start pytask session =========================
+    Platform: win32 -- Python 3.x.x, pytask 0.x.x, pluggy 0.x.x
+    Root: x
+    Collected 2 tasks.
+
+    I'm first.
+    .I'm second.
+    .
+    ========================= 2 succeeded in 0.04s =========================
+
+Replacing ``pytask.mark.try_first`` with ``pytask.mark.try_last`` yields
+
+.. code-block:: python
+
+    # Content of task_example.py
+
+    import pytask
+
+
+    def task_first():
+        print("I'm second.")
+
+
+    @pytask.mark.try_last
+    def task_second():
+        print("I'm first.")
+
+and
+
+.. code-block:: console
+
+    $ pytask -s task_example.py
+    ========================= Start pytask session =========================
+    Platform: win32 -- Python 3.x.x, pytask 0.x.x, pluggy 0.x.x
+    Root: x
+    Collected 2 tasks.
+
+    I'm second.
+    .I'm first.
+    .
+    ========================= 2 succeeded in 0.03s =========================

docs/how_to_guides/index.rst

@@ -10,3 +10,4 @@ pytask.
 
    how_to_write_a_plugin
    how_to_extend_parametrizations
+   how_to_influence_build_order

src/_pytask/config.py

@@ -95,6 +95,8 @@ def pytask_configure(pm, config_from_cli):
     config["markers"] = {
         "depends_on": "Attach a dependency/dependencies to a task.",
         "produces": "Attach a product/products to a task.",
+        "try_first": "Try to execute a task a early as possible.",
+        "try_last": "Try to execute a task a late as possible.",
     }
 
     pm.hook.pytask_parse_config(

src/_pytask/dag.py

@@ -1,34 +1,154 @@
 """Implement some capabilities to deal with the DAG."""
 import itertools
+import pprint
+from typing import Dict
+from typing import Generator
+from typing import Iterable
+from typing import List
 
+import attr
 import networkx as nx
+from _pytask.mark import get_specific_markers_from_task
+from _pytask.nodes import MetaTask
 
 
-def sort_tasks_topologically(dag):
-    """Sort tasks in topological ordering."""
-    for node in nx.topological_sort(dag):
-        if "task" in dag.nodes[node]:
-            yield node
-
-
-def descending_tasks(task_name, dag):
+def descending_tasks(task_name: str, dag: nx.DiGraph) -> Generator[str, None, None]:
     """Yield only descending tasks."""
     for descendant in nx.descendants(dag, task_name):
         if "task" in dag.nodes[descendant]:
             yield descendant
 
 
-def task_and_descending_tasks(task_name, dag):
+def task_and_descending_tasks(
+    task_name: str, dag: nx.DiGraph
+) -> Generator[str, None, None]:
     """Yield task and descending tasks."""
     yield task_name
     yield from descending_tasks(task_name, dag)
 
 
-def node_and_neighbors(dag, node):
+def node_and_neighbors(dag: nx.DiGraph, node: str) -> Generator[str, None, None]:
     """Yield node and neighbors which are first degree predecessors and successors.
 
     We cannot use ``dag.neighbors`` as it only considers successors as neighbors in a
     DAG.
 
     """
     return itertools.chain([node], dag.predecessors(node), dag.successors(node))
+
+
+@attr.s
+class TopologicalSorter:
+    """The topological sorter class.
+
+    This class allows to perform a topological sort
+
+    """
+
+    dag = attr.ib(converter=nx.DiGraph)
+    priorities = attr.ib(factory=dict)
+    _dag_backup = attr.ib(default=None)
+    _is_prepared = attr.ib(default=False, type=bool)
+    _nodes_out = attr.ib(factory=set)
+
+    @classmethod
+    def from_dag(cls, dag: nx.DiGraph) -> "TopologicalSorter":
+        if not dag.is_directed():
+            raise ValueError("Only directed graphs have a topological order.")
+
+        tasks = [
+            dag.nodes[node]["task"] for node in dag.nodes if "task" in dag.nodes[node]
+        ]
+        priorities = _extract_priorities_from_tasks(tasks)
+
+        task_names = {task.name for task in tasks}
+        task_dict = {name: nx.ancestors(dag, name) & task_names for name in task_names}
+        task_dag = nx.DiGraph(task_dict).reverse()
+
+        return cls(task_dag, priorities, task_dag.copy())
+
+    def prepare(self) -> None:
+        """Perform some checks before creating a topological ordering."""
+        try:
+            nx.algorithms.cycles.find_cycle(self.dag)
+        except nx.NetworkXNoCycle:
+            pass
+        else:
+            raise ValueError("The DAG contains cycles.")
+
+        self._is_prepared = True
+
+    def get_ready(self, n: int = 1):
+        """Get up to ``n`` tasks which are ready."""
+        if not self._is_prepared:
+            raise ValueError("The TopologicalSorter needs to be prepared.")
+        if not isinstance(n, int) or n < 1:
+            raise ValueError("'n' must be an integer greater or equal than 1.")
+
+        ready_nodes = {v for v, d in self.dag.in_degree() if d == 0} - self._nodes_out
+        prioritized_nodes = sorted(
+            ready_nodes, key=lambda x: self.priorities.get(x, 0)
+        )[-n:]
+
+        self._nodes_out.update(prioritized_nodes)
+
+        return prioritized_nodes
+
+    def is_active(self) -> bool:
+        """Indicate whether there are still tasks left."""
+        return bool(self.dag.nodes)
+
+    def done(self, *nodes: Iterable[str]) -> None:
+        """Mark some tasks as done."""
+        self._nodes_out = self._nodes_out - set(nodes)
+        self.dag.remove_nodes_from(nodes)
+
+    def reset(self) -> None:
+        """Reset an exhausted topological sorter."""
+        self.dag = self._dag_backup.copy()
+        self._is_prepared = False
+        self._nodes_out = set()
+
+    def static_order(self) -> Generator[str, None, None]:
+        """Return a topological order of tasks as an iterable."""
+        self.prepare()
+        while self.is_active():
+            new_task = self.get_ready()[0]
+            yield new_task
+            self.done(new_task)
+
+
+def _extract_priorities_from_tasks(tasks: List[MetaTask]) -> Dict[str, int]:
+    """Extract priorities from tasks.
+
+    Priorities are set via the ``pytask.mark.try_first`` and ``pytask.mark.try_last``
+    markers. We recode these markers to numeric values to sort all available by
+    priorities. ``try_first`` is assigned the highest value such that it has the
+    rightmost position in the list. Then, we can simply call :meth:`list.pop` on the
+    list which is far more efficient than ``list.pop(0)``.
+
+    """
+    priorities = {
+        task.name: {
+            "try_first": bool(get_specific_markers_from_task(task, "try_first")),
+            "try_last": bool(get_specific_markers_from_task(task, "try_last")),
+        }
+        for task in tasks
+    }
+    tasks_w_mixed_priorities = [
+        name for name, p in priorities.items() if p["try_first"] and p["try_last"]
+    ]
+    if tasks_w_mixed_priorities:
+        raise ValueError(
+            "'try_first' and 'try_last' cannot be applied on the same task. See the "
+            f"following tasks for errors:\n\n{pprint.pformat(tasks_w_mixed_priorities)}"
+        )
+
+    # Recode to numeric values for sorting.
+    numeric_mapping = {(True, False): 1, (False, False): 0, (False, True): -1}
+    numeric_priorities = {
+        name: numeric_mapping[(p["try_first"], p["try_last"])]
+        for name, p in priorities.items()
+    }
+
+    return numeric_priorities

src/_pytask/execute.py

@@ -6,7 +6,7 @@
 from _pytask.config import hookimpl
 from _pytask.dag import descending_tasks
 from _pytask.dag import node_and_neighbors
-from _pytask.dag import sort_tasks_topologically
+from _pytask.dag import TopologicalSorter
 from _pytask.database import create_or_update_state
 from _pytask.enums import ColorCode
 from _pytask.exceptions import ExecutionError
@@ -41,15 +41,16 @@ def pytask_execute_log_start(session):
 @hookimpl(trylast=True)
 def pytask_execute_create_scheduler(session):
     """Create a scheduler based on topological sorting."""
-    for node in sort_tasks_topologically(session.dag):
-        task = session.dag.nodes[node]["task"]
-        yield task
+    scheduler = TopologicalSorter.from_dag(session.dag)
+    scheduler.prepare()
+    return scheduler
 
 
 @hookimpl
 def pytask_execute_build(session):
     """Execute tasks."""
-    for task in session.scheduler:
+    for name in session.scheduler.static_order():
+        task = session.dag.nodes[name]["task"]
         report = session.hook.pytask_execute_task_protocol(session=session, task=task)
         session.execution_reports.append(report)
         if session.should_stop:
@@ -98,7 +99,7 @@ def pytask_execute_task_setup(session, task):
     for product in session.dag.successors(task.name):
         node = session.dag.nodes[product]["node"]
         if isinstance(node, FilePathNode):
-            node.value.parent.mkdir(parents=True, exist_ok=True)
+            node.path.parent.mkdir(parents=True, exist_ok=True)
 
 
 @hookimpl
@@ -140,8 +141,8 @@ def pytask_execute_task_process_report(session, report):
                 )
             )
 
-        session.n_tests_failed += 1
-        if session.n_tests_failed >= session.config["max_failures"]:
+        session.n_tasks_failed += 1
+        if session.n_tasks_failed >= session.config["max_failures"]:
             session.should_stop = True
 
     return True

src/_pytask/resolve_dependencies.py

@@ -6,15 +6,14 @@
 import networkx as nx
 from _pytask.config import hookimpl
 from _pytask.dag import node_and_neighbors
-from _pytask.dag import sort_tasks_topologically
 from _pytask.dag import task_and_descending_tasks
+from _pytask.dag import TopologicalSorter
 from _pytask.database import State
 from _pytask.exceptions import NodeNotFoundError
 from _pytask.exceptions import ResolvingDependenciesError
 from _pytask.mark import Mark
 from _pytask.nodes import reduce_node_name
 from _pytask.report import ResolvingDependenciesReport
-from _pytask.traceback import remove_traceback_from_exc_info
 from pony import orm
 
 
@@ -72,14 +71,14 @@ def pytask_resolve_dependencies_create_dag(tasks):
 @hookimpl
 def pytask_resolve_dependencies_select_execution_dag(dag):
     """Select the tasks which need to be executed."""
-    tasks = list(sort_tasks_topologically(dag))
+    scheduler = TopologicalSorter.from_dag(dag)
     visited_nodes = []
-    for task_name in tasks:
+
+    for task_name in scheduler.static_order():
         if task_name not in visited_nodes:
             have_changed = _have_task_or_neighbors_changed(task_name, dag)
             if have_changed:
-                for name in task_and_descending_tasks(task_name, dag):
-                    visited_nodes.append(name)
+                visited_nodes += list(task_and_descending_tasks(task_name, dag))
             else:
                 dag.nodes[task_name]["task"].markers.append(
                     Mark("skip_unchanged", (), {})
@@ -191,7 +190,7 @@ def pytask_resolve_dependencies_log(session, report):
 
     click.echo("")
 
-    traceback.print_exception(*remove_traceback_from_exc_info(report.exc_info))
+    traceback.print_exception(*report.exc_info)
 
     click.echo("")
     click.echo("=" * tm_width)

src/_pytask/session.py

@@ -35,7 +35,7 @@ class Session:
     execution_start = attr.ib(default=None)
     execution_end = attr.ib(default=None)
 
-    n_tests_failed = attr.ib(default=0)
+    n_tasks_failed = attr.ib(default=0)
     """Optional[int]: Number of tests which have failed."""
     should_stop = attr.ib(default=False)
     """Optional[bool]: Indicates whether the session should be stopped."""

tests/test_capture.py

@@ -255,8 +255,8 @@ def task_capturing_error():
     tmp_path.joinpath("task_capturing_outerr.py").write_text(textwrap.dedent(source))
 
     result = runner.invoke(cli, [tmp_path.as_posix()])
+    assert "F." in result.output or ".F" in result.output
     for content in [
-        "F.",
         "==== Failures ====",
         "task_capturing_error failed",
         "---------",