Removed resource_multiproc code so only new_interfaces code is left

Author: pintohutch

nipype/interfaces/ants/base.py

@@ -12,7 +12,7 @@
 # -Using -1 gives primary responsibilty to ITKv4 to do the correct
 #  thread limitings.
 # -Using 1 takes a very conservative approach to avoid overloading
-#  the computer (when running ResourceMultiProc) by forcing everything to
+#  the computer (when running MultiProc) by forcing everything to
 #  single threaded.  This can be a severe penalty for registration
 #  performance.
 LOCAL_DEFAULT_NUMBER_OF_THREADS = 1

nipype/interfaces/base.py

@@ -754,8 +754,6 @@ def __init__(self, **inputs):
             raise Exception('No input_spec in class: %s' %
                             self.__class__.__name__)
         self.inputs = self.input_spec(**inputs)
-        self.estimated_memory = 1
-        self.num_threads = 1
 
     @classmethod
     def help(cls, returnhelp=False):
@@ -1194,69 +1192,14 @@ def _read(self, drain):
         self._lastidx = len(self._rows)
 
 
-# Get number of threads for process
-def _get_num_threads(proc):
-    '''
-    '''
-
-    # Import packages
-    import psutil
-    import logging as lg
-
-    # Init variables
-    num_threads = proc.num_threads()
-    try:
-        num_children = len(proc.children())
-        for child in proc.children():
-            num_threads = max(num_threads, num_children,
-                              child.num_threads(), len(child.children()))
-    except psutil.NoSuchProcess:
-        pass
-
-    return num_threads
-
-
-# Get max resources used for process
-def _get_max_resources_used(proc, mem_mb, num_threads, poll=False):
-    '''
-    docstring
-    '''
-
-    # Import packages
-    from memory_profiler import _get_memory
-    import psutil
-
-    try:
-        mem_mb = max(mem_mb, _get_memory(proc.pid, include_children=True))
-        num_threads = max(num_threads, _get_num_threads(psutil.Process(proc.pid)))
-        if poll:
-            proc.poll()
-    except Exception as exc:
-        iflogger.info('Could not get resources used by process. Error: %s'\
-                      % exc)
-
-    # Return resources
-    return mem_mb, num_threads
-
-
 def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     """Run a command, read stdout and stderr, prefix with timestamp.
 
     The returned runtime contains a merged stdout+stderr log with timestamps
     """
-
-    # Import packages
-    try:
-        import memory_profiler
-        import psutil
-        mem_prof = True
-    except:
-        mem_prof = False
-
-    # Init variables
     PIPE = subprocess.PIPE
-    cmdline = runtime.cmdline
 
+    cmdline = runtime.cmdline
     if redirect_x:
         exist_xvfb, _ = _exists_in_path('xvfb-run', runtime.environ)
         if not exist_xvfb:
@@ -1288,12 +1231,6 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     result = {}
     errfile = os.path.join(runtime.cwd, 'stderr.nipype')
     outfile = os.path.join(runtime.cwd, 'stdout.nipype')
-
-    # Init variables for memory profiling
-    mem_mb = -1
-    num_threads = -1
-    interval = 1
-
     if output == 'stream':
         streams = [Stream('stdout', proc.stdout), Stream('stderr', proc.stderr)]
 
@@ -1309,10 +1246,8 @@ def _process(drain=0):
             else:
                 for stream in res[0]:
                     stream.read(drain)
+
         while proc.returncode is None:
-            if mem_prof:
-                mem_mb, num_threads = \
-                    _get_max_resources_used(proc, mem_mb, num_threads)
             proc.poll()
             _process()
         _process(drain=1)
@@ -1326,41 +1261,25 @@ def _process(drain=0):
             result[stream._name] = [r[2] for r in rows]
         temp.sort()
         result['merged'] = [r[1] for r in temp]
-
     if output == 'allatonce':
-        if mem_prof:
-            while proc.returncode is None:
-                mem_mb, num_threads = \
-                    _get_max_resources_used(proc, mem_mb, num_threads, poll=True)
         stdout, stderr = proc.communicate()
         stdout = stdout.decode(default_encoding)
         stderr = stderr.decode(default_encoding)
         result['stdout'] = stdout.split('\n')
         result['stderr'] = stderr.split('\n')
         result['merged'] = ''
     if output == 'file':
-        if mem_prof:
-            while proc.returncode is None:
-                mem_mb, num_threads = \
-                    _get_max_resources_used(proc, mem_mb, num_threads, poll=True)
         ret_code = proc.wait()
         stderr.flush()
         stdout.flush()
         result['stdout'] = [line.decode(default_encoding).strip() for line in open(outfile, 'rb').readlines()]
         result['stderr'] = [line.decode(default_encoding).strip() for line in open(errfile, 'rb').readlines()]
         result['merged'] = ''
     if output == 'none':
-        if mem_prof:
-            while proc.returncode is None:
-                mem_mb, num_threads = \
-                    _get_max_resources_used(proc, mem_mb, num_threads, poll=True)
         proc.communicate()
         result['stdout'] = []
         result['stderr'] = []
         result['merged'] = ''
-
-    setattr(runtime, 'runtime_memory', mem_mb/1024.0)
-    setattr(runtime, 'runtime_threads', num_threads)
     runtime.stderr = '\n'.join(result['stderr'])
     runtime.stdout = '\n'.join(result['stdout'])
     runtime.merged = result['merged']

nipype/pipeline/engine/nodes.py

@@ -52,7 +52,6 @@
 
 from ... import config, logging
 logger = logging.getLogger('workflow')
-
 from ...interfaces.base import (traits, InputMultiPath, CommandLine,
                                 Undefined, TraitedSpec, DynamicTraitedSpec,
                                 Bunch, InterfaceResult, md5, Interface,
@@ -1152,7 +1151,8 @@ def _node_runner(self, nodes, updatehash=False):
                 if str2bool(self.config['execution']['stop_on_first_crash']):
                     self._result = node.result
                     raise
-            yield i, node, err
+            finally:
+                yield i, node, err
 
     def _collate_results(self, nodes):
         self._result = InterfaceResult(interface=[], runtime=[],

nipype/pipeline/engine/tests/test_engine.py

@@ -715,15 +715,15 @@ def func1(in1):
     w1.config['execution'] = {'stop_on_first_crash': 'true',
                               'local_hash_check': 'true',
                               'crashdump_dir': wd,
-                              'poll_sleep_duration' : 2}
+                              'poll_sleep_duration': 2}
 
     # test output of num_subnodes method when serial is default (False)
     yield assert_equal, n1.num_subnodes(), len(n1.inputs.in1)
 
     # test running the workflow on default conditions
     error_raised = False
     try:
-        w1.run(plugin='ResourceMultiProc')
+        w1.run(plugin='MultiProc')
     except Exception as e:
         from nipype.pipeline.engine.base import logger
         logger.info('Exception: %s' % str(e))
@@ -737,7 +737,7 @@ def func1(in1):
     # test running the workflow on serial conditions
     error_raised = False
     try:
-        w1.run(plugin='ResourceMultiProc')
+        w1.run(plugin='MultiProc')
     except Exception as e:
         from nipype.pipeline.engine.base import logger
         logger.info('Exception: %s' % str(e))

nipype/pipeline/engine/tests/test_utils.py

@@ -214,7 +214,7 @@ def test_function3(arg):
 
     out_dir = mkdtemp()
 
-    for plugin in ('Linear',):  # , 'ResourceMultiProc'):
+    for plugin in ('Linear',):  # , 'MultiProc'):
         n1 = pe.Node(niu.Function(input_names=['arg1'],
                                   output_names=['out_file1', 'out_file2', 'dir'],
                                   function=test_function),

nipype/pipeline/plugins/__init__.py

@@ -9,13 +9,11 @@
 from .sge import SGEPlugin
 from .condor import CondorPlugin
 from .dagman import CondorDAGManPlugin
-from .multiproc import ResourceMultiProcPlugin
+from .multiproc import MultiProcPlugin
 from .ipython import IPythonPlugin
 from .somaflow import SomaFlowPlugin
 from .pbsgraph import PBSGraphPlugin
 from .sgegraph import SGEGraphPlugin
 from .lsf import LSFPlugin
 from .slurm import SLURMPlugin
 from .slurmgraph import SLURMGraphPlugin
-
-from .callback_log import log_nodes_cb

nipype/pipeline/plugins/base.py

@@ -20,6 +20,7 @@
 import numpy as np
 import scipy.sparse as ssp
 
+
 from ...utils.filemanip import savepkl, loadpkl
 from ...utils.misc import str2bool
 from ..engine.utils import (nx, dfs_preorder, topological_sort)
@@ -245,7 +246,7 @@ def run(self, graph, config, updatehash=False):
                             notrun.append(self._clean_queue(jobid, graph,
                                                             result=result))
                         else:
-                            self._task_finished_cb(jobid, result)
+                            self._task_finished_cb(jobid)
                             self._remove_node_dirs()
                         self._clear_task(taskid)
                     else:
@@ -264,15 +265,10 @@ def run(self, graph, config, updatehash=False):
                                             graph=graph)
             else:
                 logger.debug('Not submitting')
-            self._wait()
+            sleep(float(self._config['execution']['poll_sleep_duration']))
         self._remove_node_dirs()
         report_nodes_not_run(notrun)
 
-
-
-    def _wait(self):
-        sleep(float(self._config['execution']['poll_sleep_duration']))
-
     def _get_result(self, taskid):
         raise NotImplementedError
 
@@ -414,18 +410,15 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             else:
                 break
 
-    def _task_finished_cb(self, jobid, result=None):
+    def _task_finished_cb(self, jobid):
         """ Extract outputs and assign to inputs of dependent tasks
 
         This is called when a job is completed.
         """
         logger.info('[Job finished] jobname: %s jobid: %d' %
                     (self.procs[jobid]._id, jobid))
         if self._status_callback:
-            if result == None:
-                if self._taskresult.has_key(jobid):
-                    result = self._taskresult[jobid].get()
-            self._status_callback(self.procs[jobid], 'end', result)
+            self._status_callback(self.procs[jobid], 'end')
         # Update job and worker queues
         self.proc_pending[jobid] = False
         # update the job dependency structure