Merge pull request #3143 from satra/fix/mapnode_single

fix: mapnode to generate result file when crashes in single node mode

Author: effigies

nipype/info.py

@@ -155,7 +155,15 @@ def get_nipype_gitversion():
 # https://github.com/nipy/nipype/pull/2961#issuecomment-512035484
 REQUIRES += ["neurdflib"]
 
-TESTS_REQUIRES = ["codecov", "coverage<5", "mock", "pytest", "pytest-cov", "pytest-env"]
+TESTS_REQUIRES = [
+    "codecov",
+    "coverage<5",
+    "mock",
+    "pytest",
+    "pytest-cov",
+    "pytest-env",
+    "pytest-timeout",
+]
 
 EXTRA_REQUIRES = {
     "data": ["datalad"],

nipype/pipeline/engine/nodes.py

@@ -1366,13 +1366,39 @@ def _run_interface(self, execute=True, updatehash=False):
         nodenames = [nnametpl.format(i) for i in range(nitems)]
 
         # Run mapnode
-        result = self._collate_results(
-            _node_runner(
-                self._make_nodes(cwd),
-                updatehash=updatehash,
-                stop_first=str2bool(self.config["execution"]["stop_on_first_crash"]),
-            )
+        outdir = self.output_dir()
+        result = InterfaceResult(
+            interface=self._interface.__class__,
+            runtime=Bunch(
+                cwd=outdir,
+                returncode=1,
+                environ=dict(os.environ),
+                hostname=socket.gethostname(),
+            ),
+            inputs=self._interface.inputs.get_traitsfree(),
         )
+        try:
+            result = self._collate_results(
+                _node_runner(
+                    self._make_nodes(cwd),
+                    updatehash=updatehash,
+                    stop_first=str2bool(
+                        self.config["execution"]["stop_on_first_crash"]
+                    ),
+                )
+            )
+        except Exception as msg:
+            result.runtime.stderr = "%s\n\n%s".format(
+                getattr(result.runtime, "stderr", ""), msg
+            )
+            _save_resultfile(
+                result,
+                outdir,
+                self.name,
+                rebase=str2bool(self.config["execution"]["use_relative_paths"]),
+            )
+            raise
+
         # And store results
         _save_resultfile(result, cwd, self.name, rebase=False)
         # remove any node directories no longer required

nipype/pipeline/engine/tests/test_nodes.py

@@ -314,3 +314,23 @@ def test_outputmultipath_collapse(tmpdir):
     assert ifres.outputs.out == [4]
     assert ndres.outputs.out == [4]
     assert select_nd.result.outputs.out == [4]
+
+
+@pytest.mark.timeout(30)
+def test_mapnode_single(tmpdir):
+    tmpdir.chdir()
+
+    def _producer(num=1, deadly_num=7):
+        if num == deadly_num:
+            raise RuntimeError("Got the deadly num (%d)." % num)
+        return num + 1
+
+    pnode = pe.MapNode(
+        niu.Function(function=_producer), name="ProducerNode", iterfield=["num"]
+    )
+    pnode.inputs.num = [7]
+    wf = pe.Workflow(name="PC_Workflow")
+    wf.add_nodes([pnode])
+    wf.base_dir = os.path.abspath("./test_output")
+    with pytest.raises(RuntimeError):
+        wf.run(plugin="MultiProc")