fix tests

Author: oesteban

nipype/interfaces/base.py

@@ -765,8 +765,6 @@ def __init__(self, from_file=None, resource_monitor=None, **inputs):
                             self.__class__.__name__)
 
         self.inputs = self.input_spec(**inputs)
-        self.estimated_memory_gb = 0.25
-        self.num_threads = 1
 
         if resource_monitor is not None:
             self.resource_monitor = resource_monitor

nipype/pipeline/engine/nodes.py

@@ -78,7 +78,7 @@ class Node(EngineBase):
 
     def __init__(self, interface, name, iterables=None, itersource=None,
                  synchronize=False, overwrite=None, needed_outputs=None,
-                 run_without_submitting=False, n_procs=1, mem_gb=None,
+                 run_without_submitting=False, n_procs=1, mem_gb=0.25,
                  **kwargs):
         """
         Parameters
@@ -169,9 +169,8 @@ def __init__(self, interface, name, iterables=None, itersource=None,
         self.needed_outputs = []
         self.plugin_args = {}
 
-        self._interface.num_threads = n_procs
-        if mem_gb is not None:
-            self._interface.estimated_memory_gb = mem_gb
+        self._n_procs = n_procs
+        self._mem_gb = mem_gb
 
         if needed_outputs:
             self.needed_outputs = sorted(needed_outputs)
@@ -270,6 +269,7 @@ def run(self, updatehash=False):
             Update the hash stored in the output directory
         """
         # check to see if output directory and hash exist
+
         if self.config is None:
             self.config = deepcopy(config._sections)
         else:
@@ -685,6 +685,24 @@ def _copyfiles_to_wd(self, outdir, execute, linksonly=False):
             if execute and linksonly:
                 rmtree(outdir)
 
+    def get_mem_gb(self):
+        """Get estimated memory (GB)"""
+        if hasattr(self._interface, 'estimated_memory_gb'):
+            self._mem_gb = self._interface.estimated_memory_gb
+            logger.warning('Setting "estimated_memory_gb" on Interfaces has been '
+                           'deprecated as of nipype 1.0')
+            del self._interface.estimated_memory_gb
+        return self._mem_gb
+
+    def get_n_procs(self):
+        """Get estimated number of processes"""
+        if hasattr(self._interface, 'num_threads'):
+            self._n_procs = self._interface.num_threads
+            logger.warning('Setting "num_threads" on Interfaces has been '
+                           'deprecated as of nipype 1.0')
+            del self._interface.num_threads
+        return self._n_procs
+
     def update(self, **opts):
         self.inputs.update(**opts)
 
@@ -1111,8 +1129,8 @@ def _make_nodes(self, cwd=None):
         for i in range(nitems):
             nodename = '_' + self.name + str(i)
             node = Node(deepcopy(self._interface),
-                        n_procs=self._interface.num_threads,
-                        mem_gb=self._interface.estimated_memory_gb,
+                        n_procs=self.get_n_procs(),
+                        mem_gb=self.get_mem_gb(),
                         overwrite=self.overwrite,
                         needed_outputs=self.needed_outputs,
                         run_without_submitting=self.run_without_submitting,

nipype/pipeline/engine/tests/test_engine.py

@@ -513,17 +513,17 @@ def func1(in1):
 
     mapnode = MapNode(Function(function=func1),
                       iterfield='in1',
-                      name='mapnode')
+                      name='mapnode',
+                      n_procs=2,
+                      mem_gb=2)
     mapnode.inputs.in1 = [1, 2]
-    mapnode.interface.num_threads = 2
-    mapnode.interface.estimated_memory_gb = 2
 
     for idx, node in mapnode._make_nodes():
         for attr in ('overwrite', 'run_without_submitting', 'plugin_args'):
             assert getattr(node, attr) == getattr(mapnode, attr)
-        for attr in ('num_threads', 'estimated_memory_gb'):
-            assert (getattr(node._interface, attr) ==
-                    getattr(mapnode._interface, attr))
+        for attr in ('_n_procs', '_mem_gb'):
+            assert (getattr(node, attr) ==
+                    getattr(mapnode, attr))
 
 
 def test_node_hash(tmpdir):

nipype/pipeline/plugins/base.py

@@ -80,6 +80,7 @@ def __init__(self, plugin_args=None):
         self.mapnodesubids = None
         self.proc_done = None
         self.proc_pending = None
+        self.pending_tasks = []
         self.max_jobs = self.plugin_args.get('max_jobs', np.inf)
 
     def _prerun_check(self, graph):
@@ -95,8 +96,6 @@ def run(self, graph, config, updatehash=False):
         self._prerun_check(graph)
         # Generate appropriate structures for worker-manager model
         self._generate_dependency_list(graph)
-        self.pending_tasks = []
-        self.readytorun = []
         self.mapnodes = []
         self.mapnodesubids = {}
         # setup polling - TODO: change to threaded model
@@ -110,6 +109,11 @@ def run(self, graph, config, updatehash=False):
                 taskid, jobid = self.pending_tasks.pop()
                 try:
                     result = self._get_result(taskid)
+                except Exception:
+                    notrun.append(self._clean_queue(
+                        jobid, graph, result={'result': None,
+                                              'traceback': format_exc()}))
+                else:
                     if result:
                         if result['traceback']:
                             notrun.append(self._clean_queue(jobid, graph,
@@ -120,11 +124,6 @@ def run(self, graph, config, updatehash=False):
                         self._clear_task(taskid)
                     else:
                         toappend.insert(0, (taskid, jobid))
-                except Exception:
-                    result = {'result': None,
-                              'traceback': format_exc()}
-                    notrun.append(self._clean_queue(jobid, graph,
-                                                    result=result))
 
             if toappend:
                 self.pending_tasks.extend(toappend)
@@ -169,12 +168,15 @@ def _clear_task(self, taskid):
         raise NotImplementedError
 
     def _clean_queue(self, jobid, graph, result=None):
+        logger.info('Clearing %d from queue', jobid)
+
+        if self._status_callback:
+            self._status_callback(self.procs[jobid], 'exception')
+
         if str2bool(self._config['execution']['stop_on_first_crash']):
             raise RuntimeError("".join(result['traceback']))
         crashfile = self._report_crash(self.procs[jobid],
                                        result=result)
-        if self._status_callback:
-            self._status_callback(self.procs[jobid], 'exception')
         if jobid in self.mapnodesubids:
             # remove current jobid
             self.proc_pending[jobid] = False