enh: better error handling

Author: oesteban

nipype/interfaces/utility/tests/test_wrappers.py

@@ -73,7 +73,7 @@ def should_fail(tmp):
 
 
 def test_should_fail(tmpdir):
-    with pytest.raises(NameError):
+    with pytest.raises(pe.nodes.NodeExecutionError):
         should_fail(tmpdir)
 
 

nipype/pipeline/engine/nodes.py

@@ -64,6 +64,10 @@
 logger = logging.getLogger("nipype.workflow")
 
 
+class NodeExecutionError(RuntimeError):
+    """A nipype-specific name for exceptions when executing a Node."""
+
+
 class Node(EngineBase):
     """
     Wraps interface objects for use in pipeline
@@ -582,7 +586,7 @@ def _get_inputs(self):
                 logger.critical("%s", e)
 
             if outputs is None:
-                raise RuntimeError(
+                raise NodeExecutionError(
                     """\
 Error populating the inputs of node "%s": the results file of the source node \
 (%s) does not contain any outputs."""
@@ -717,7 +721,7 @@ def _run_command(self, execute, copyfiles=True):
             # Write out command line as it happened
             (outdir / "command.txt").write_text(f"{result.runtime.cmdline}\n")
 
-        exc_tb = getattr(result, "traceback", None)
+        exc_tb = getattr(result.runtime, "traceback", None)
 
         if not exc_tb:
             # Clean working directory if no errors
@@ -743,7 +747,7 @@ def _run_command(self, execute, copyfiles=True):
         )
 
         if exc_tb:
-            raise RuntimeError(
+            raise NodeExecutionError(
                 f"Exception raised while executing Node {self.name}.\n\n{result.runtime.traceback}"
             )
 
@@ -1267,7 +1271,7 @@ def _collate_results(self, nodes):
                 if code is not None:
                     msg += ["Subnode %d failed" % i]
                     msg += ["Error: %s" % str(code)]
-            raise Exception(
+            raise NodeExecutionError(
                 "Subnodes of node: %s failed:\n%s" % (self.name, "\n".join(msg))
             )
 

nipype/pipeline/engine/tests/test_utils.py

@@ -183,7 +183,7 @@ def test_mapnode_crash(tmpdir):
     node.config = deepcopy(config._sections)
     node.config["execution"]["stop_on_first_crash"] = True
     node.base_dir = tmpdir.strpath
-    with pytest.raises(TypeError):
+    with pytest.raises(pe.nodes.NodeExecutionError):
         node.run()
     os.chdir(cwd)
 

nipype/pipeline/plugins/base.py

@@ -123,6 +123,7 @@ def run(self, graph, config, updatehash=False):
         self.mapnodesubids = {}
         # setup polling - TODO: change to threaded model
         notrun = []
+        errors = []
 
         old_progress_stats = None
         old_presub_stats = None
@@ -155,14 +156,16 @@ def run(self, graph, config, updatehash=False):
                 taskid, jobid = self.pending_tasks.pop()
                 try:
                     result = self._get_result(taskid)
-                except Exception:
+                except Exception as exc:
                     notrun.append(self._clean_queue(jobid, graph))
+                    errors.append(exc)
                 else:
                     if result:
                         if result["traceback"]:
                             notrun.append(
                                 self._clean_queue(jobid, graph, result=result)
                             )
+                            errors.append("".join(result["traceback"]))
                         else:
                             self._task_finished_cb(jobid)
                             self._remove_node_dirs()
@@ -194,6 +197,13 @@ def run(self, graph, config, updatehash=False):
         # close any open resources
         self._postrun_check()
 
+        if errors:
+            # If one or more nodes failed, re-rise first of them
+            if isinstance(errors[0], str):
+                raise RuntimeError(errors[0])
+
+            raise errors[0]
+
     def _get_result(self, taskid):
         raise NotImplementedError
 

nipype/pipeline/plugins/linear.py

@@ -34,6 +34,8 @@ def run(self, graph, config, updatehash=False):
         old_wd = os.getcwd()
         notrun = []
         donotrun = []
+        stop_on_first_crash = str2bool(config["execution"]["stop_on_first_crash"])
+        errors = []
         nodes, _ = topological_sort(graph)
         for node in nodes:
             endstatus = "end"
@@ -43,27 +45,28 @@ def run(self, graph, config, updatehash=False):
                 if self._status_callback:
                     self._status_callback(node, "start")
                 node.run(updatehash=updatehash)
-            except:
+            except Exception as exc:
                 endstatus = "exception"
                 # bare except, but i really don't know where a
                 # node might fail
                 crashfile = report_crash(node)
-                if str2bool(config["execution"]["stop_on_first_crash"]):
-                    raise
                 # remove dependencies from queue
                 subnodes = [s for s in dfs_preorder(graph, node)]
                 notrun.append(
                     {"node": node, "dependents": subnodes, "crashfile": crashfile}
                 )
                 donotrun.extend(subnodes)
                 # Delay raising the crash until we cleaned the house
-                if str2bool(config["execution"]["stop_on_first_crash"]):
-                    os.chdir(old_wd)  # Return wherever we were before
-                    report_nodes_not_run(notrun)  # report before raising
-                    raise
+                errors.append(exc)
+
+                if stop_on_first_crash:
+                    break
             finally:
                 if self._status_callback:
                     self._status_callback(node, endstatus)
 
         os.chdir(old_wd)  # Return wherever we were before
         report_nodes_not_run(notrun)
+        if errors:
+            # Re-raise exception of first failed node
+            raise errors[0]

nipype/pipeline/plugins/tests/test_sgelike.py

@@ -29,7 +29,10 @@ def test_crashfile_creation(tmp_path):
     sgelike_plugin = SGELikeBatchManagerBase("")
     with pytest.raises(RuntimeError) as e:
         assert pipe.run(plugin=sgelike_plugin)
-    assert str(e.value) == "Workflow did not execute cleanly. Check log for details"
+        assert str(e.value) == "Workflow did not execute cleanly. Check log for details"
 
-    crashfiles = tmp_path.glob("crash*crasher*.pklz")
-    assert len(list(crashfiles)) == 1
+    crashfiles = (
+        list(tmp_path.glob("crash*crasher*.pklz"))
+        + list(tmp_path.glob("crash*crasher*.txt"))
+    )
+    assert len(crashfiles) == 1

nipype/pipeline/plugins/tools.py

@@ -93,9 +93,6 @@ def report_nodes_not_run(notrun):
             for subnode in info["dependents"]:
                 logger.debug(subnode._id)
         logger.info("***********************************")
-        raise RuntimeError(
-            ("Workflow did not execute cleanly. " "Check log for details")
-        )
 
 
 def create_pyscript(node, updatehash=False, store_exception=True):