Implement task priorities and the new scheduler. (#8)

Author: tobiasraabe

.pre-commit-config.yaml

@@ -9,6 +9,18 @@ repos:
         exclude: meta.yaml
     -   id: debug-statements
     -   id: end-of-file-fixer
+-   repo: https://github.com/pre-commit/pygrep-hooks
+    rev: v1.7.0  # Use the ref you want to point at
+    hooks:
+    -   id: python-check-blanket-noqa
+    -   id: python-check-mock-methods
+    -   id: python-no-eval
+    -   id: python-no-log-warn
+    -   id: python-use-type-annotations
+    -   id: rst-backticks
+    -   id: rst-directive-colons
+    -   id: rst-inline-touching-normal
+    -   id: text-unicode-replacement-char
 -   repo: https://github.com/asottile/pyupgrade
     rev: v2.7.4
     hooks:

CHANGES.rst

@@ -11,6 +11,7 @@ all releases are available on `Anaconda.org
 ------------------
 
 - :gh:`5` fixes the CI and other smaller issues.
+- :gh:`8` aligns pytask-parallel with task priorities in pytask v0.0.11.
 - :gh:`9` enables --max-failures. Closes :gh:`7`.
 
 

README.rst

@@ -18,8 +18,8 @@
 pytask-parallel
 ===============
 
-Parallelize the execution of tasks with `pytask-parallel` which is a plugin for `pytask
-<https://github.com/pytask-dev/pytask>`_.
+Parallelize the execution of tasks with ``pytask-parallel`` which is a plugin for
+`pytask <https://github.com/pytask-dev/pytask>`_.
 
 
 Installation

src/pytask_parallel/__init__.py

@@ -1 +1,2 @@
+"""The entry-point for pytask-parallel."""
 __version__ = "0.0.4"

src/pytask_parallel/execute.py

@@ -3,11 +3,9 @@
 import time
 
 import cloudpickle
-import networkx as nx
 from _pytask.config import hookimpl
 from _pytask.report import ExecutionReport
 from pytask_parallel.backends import PARALLEL_BACKENDS
-from pytask_parallel.scheduler import TopologicalSorter
 
 
 @hookimpl
@@ -19,23 +17,6 @@ def pytask_post_parse(config):
         config["pm"].register(DefaultBackendNameSpace)
 
 
-@hookimpl(tryfirst=True)
-def pytask_execute_create_scheduler(session):
-    """Create the scheduler."""
-    if session.config["n_workers"] > 1:
-        task_names = {task.name for task in session.tasks}
-        task_dict = {
-            name: nx.ancestors(session.dag, name) & task_names for name in task_names
-        }
-        scheduler = TopologicalSorter(task_dict)
-
-        # Forbid to add further nodes and check for cycles. The latter should have been
-        # taken care of while setting up the DAG.
-        scheduler.prepare()
-
-        return scheduler
-
-
 @hookimpl(tryfirst=True)
 def pytask_execute_build(session):
     """Execute tasks with a parallel backend.
@@ -61,7 +42,12 @@ def pytask_execute_build(session):
             while session.scheduler.is_active():
 
                 newly_collected_reports = []
-                ready_tasks = list(session.scheduler.get_ready())
+                n_new_tasks = session.config["n_workers"] - len(running_tasks)
+
+                if n_new_tasks >= 1:
+                    ready_tasks = list(session.scheduler.get_ready(n_new_tasks))
+                else:
+                    ready_tasks = []
 
                 for task_name in ready_tasks:
                     task = session.dag.nodes[task_name]["task"]
@@ -132,6 +118,8 @@ def pytask_execute_build(session):
 
 
 class ProcessesNameSpace:
+    """The name space for hooks related to processes."""
+
     @hookimpl(tryfirst=True)
     def pytask_execute_task(session, task):  # noqa: N805
         """Execute a task.
@@ -156,6 +144,8 @@ def unserialize_and_execute_task(bytes_):
 
 
 class DefaultBackendNameSpace:
+    """The name space for hooks related to threads."""
+
     @hookimpl(tryfirst=True)
     def pytask_execute_task(session, task):  # noqa: N805
         """Execute a task.