Merge pull request #581 from tclose/debugging-graph-errors

More descriptive debugging graph errors and bugfix in values validator

Author: djarecka

pydra/engine/helpers.py

@@ -112,7 +112,7 @@ def load_result(checksum, cache_locations):
     if not cache_locations:
         return None
     # TODO: if there are issues with loading, we might need to
-    # TODO: sleep and repeat loads (after checkin that there are no lock files!)
+    # TODO: sleep and repeat loads (after checking that there are no lock files!)
     for location in cache_locations:
         if (location / checksum).exists():
             result_file = location / checksum / "_result.pklz"
@@ -482,7 +482,7 @@ def _check_special_type(tp, name):
 def _allowed_values_validator(instance, attribute, value):
     """checking if the values is in allowed_values"""
     allowed = attribute.metadata["allowed_values"]
-    if value is attr.NOTHING:
+    if value is attr.NOTHING or isinstance(value, LazyField):
         pass
     elif value not in allowed:
         raise ValueError(

pydra/engine/specs.py

@@ -457,7 +457,10 @@ def collect_additional_outputs(self, inputs, output_dir, outputs):
                 str,
                 list,
             ]:
-                raise Exception("not implemented (collect_additional_output)")
+                raise Exception(
+                    f"Support for {fld.type} type, required for {fld.name} in {self}, "
+                    "has not been implemented in collect_additional_output"
+                )
             # assuming that field should have either default or metadata, but not both
             input_value = getattr(inputs, fld.name, attr.NOTHING)
             if input_value is not attr.NOTHING:

pydra/engine/submitter.py

@@ -1,5 +1,6 @@
 """Handle execution backends."""
 import asyncio
+import pickle
 from uuid import uuid4
 from .workers import WORKERS
 from .core import is_workflow
@@ -167,9 +168,14 @@ async def expand_workflow(self, wf, rerun=False):
                     # don't block the event loop!
                     await asyncio.sleep(1)
                     if ii > 60:
+                        blocked = _list_blocked_tasks(graph_copy)
+                        get_runnable_tasks(graph_copy)
                         raise Exception(
                             "graph is not empty, but not able to get more tasks "
-                            "- something is wrong (e.g. with the filesystem)"
+                            "- something may have gone wrong when retrieving the results "
+                            "of predecessor tasks caused by a file-system error or a bug "
+                            "in the internal workflow logic.\n\nBlocked tasks\n-------------\n"
+                            + "\n".join(blocked)
                         )
             for task in tasks:
                 # grab inputs if needed
@@ -281,3 +287,33 @@ async def prepare_runnable_with_state(runnable):
     runnable.state.prepare_inputs()
     logger.debug(f"Expanding {runnable} into {len(runnable.state.states_val)} states")
     return runnable.pickle_task()
+
+
+def _list_blocked_tasks(graph):
+    """Generates a list of tasks that can't be run and predecessors that are blocking
+    them to help debugging of broken workflows"""
+    blocked = []
+    for tsk in graph.sorted_nodes:
+        blocking = []
+        for pred in graph.predecessors[tsk.name]:
+            if not pred.done:
+                matching_name = []
+                for cache_loc in tsk.cache_locations:
+                    for tsk_work_dir in cache_loc.iterdir():
+                        if (tsk_work_dir / "_task.pklz").exists():
+                            with open(tsk_work_dir / "_task.pklz", "rb") as f:
+                                saved_tsk = pickle.load(f)
+                            if saved_tsk.name == pred.name:
+                                matching_name.append(
+                                    f"{saved_tsk.name} ({tsk_work_dir.name})"
+                                )
+                blocking.append(pred, ", ".join(matching_name))
+        if blocking:
+            blocked.append(
+                f"\n{tsk.name} ({tsk.checksum}) is blocked by "
+                + "; ".join(
+                    f"{pred.name} ({pred.checksum}), which matches names of [{matching}]"
+                    for pred, matching in blocking
+                )
+            )
+    return blocked