Merge branch 'main' into pending-status

Author: tobiasraabe

.pre-commit-config.yaml

@@ -23,7 +23,7 @@ repos:
     -   id: python-use-type-annotations
     -   id: text-unicode-replacement-char
 -   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.4.2
+    rev: v0.4.4
     hooks:
       - id: ruff
       - id: ruff-format

docs/source/changes.md

@@ -26,6 +26,9 @@ releases are available on [PyPI](https://pypi.org/project/pytask-parallel) and
 - {pull}`101` adds syncing for local paths as dependencies or products in remote
   environments with the same OS.
 - {pull}`102` implements a pending status for scheduled but not started tasks.
+- {pull}`106` fixes {pull}`99` such that only when there are coiled functions, all ready
+  tasks are submitted.
+- {pull}`107` removes status from `pytask_execute_task_log_start` hook call.
 
 ## 0.4.1 - 2024-01-12
 

src/pytask_parallel/execute.py

@@ -44,7 +44,7 @@
 
 
 @hookimpl
-def pytask_execute_build(session: Session) -> bool | None:  # noqa: C901, PLR0915
+def pytask_execute_build(session: Session) -> bool | None:  # noqa: C901, PLR0912, PLR0915
     """Execute tasks with a parallel backend.
 
     There are three phases while the scheduler has tasks which need to be executed.
@@ -75,6 +75,7 @@ def pytask_execute_build(session: Session) -> bool | None:  # noqa: C901, PLR091
 
     # Get the live execution manager from the registry if it exists.
     live_execution = session.config["pm"].get_plugin("live_execution")
+    any_coiled_task = any(is_coiled_function(task) for task in session.tasks)
 
     # The executor can only be created after the collection to give users the
     # possibility to inject their own executors.
@@ -93,7 +94,26 @@ def pytask_execute_build(session: Session) -> bool | None:  # noqa: C901, PLR091
         while session.scheduler.is_active():
             try:
                 newly_collected_reports = []
-                ready_tasks = list(session.scheduler.get_ready(10_000))
+
+                # If there is any coiled function, the user probably wants to exploit
+                # adaptive scaling. Thus, we need to submit all ready tasks.
+                # Unfortunately, all submitted tasks are shown as running although some
+                # are pending.
+                #
+                # Without coiled functions, we submit as many tasks as there are
+                # available workers since we cannot reliably detect a pending status.
+                #
+                # See #98 for more information.
+                if any_coiled_task:
+                    n_new_tasks = 10_000
+                else:
+                    n_new_tasks = session.config["n_workers"] - len(running_tasks)
+
+                ready_tasks = (
+                    list(session.scheduler.get_ready(n_new_tasks))
+                    if n_new_tasks >= 1
+                    else []
+                )
 
                 for task_signature in ready_tasks:
                     task = session.dag.nodes[task_signature]["task"]

tests/test_execute.py

@@ -78,7 +78,6 @@ def task_2(path: Annotated[Path, Product] = Path("out_2.txt")):
 
 
 @pytest.mark.end_to_end()
-@pytest.mark.skip(reason="See #98")
 @pytest.mark.parametrize("parallel_backend", _IMPLEMENTED_BACKENDS)
 def test_stop_execution_when_max_failures_is_reached(tmp_path, parallel_backend):
     source = """
@@ -107,7 +106,6 @@ def task_3(): time.sleep(3)
 
 
 @pytest.mark.end_to_end()
-@pytest.mark.skip(reason="See #98")
 @pytest.mark.parametrize("parallel_backend", _IMPLEMENTED_BACKENDS)
 def test_task_priorities(tmp_path, parallel_backend):
     source = """