remove MultiProc, MultiprocPlugin is default

Author: carolFrohlich

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 MultiProc) by forcing everything to
+#  the computer (when running ResourceMultiProc) by forcing everything to
 #  single threaded.  This can be a severe penalty for registration
 #  performance.
 LOCAL_DEFAULT_NUMBER_OF_THREADS = 1

nipype/pipeline/engine/tests/test_engine.py

@@ -714,16 +714,15 @@ def func1(in1):
     # set local check
     w1.config['execution'] = {'stop_on_first_crash': 'true',
                               'local_hash_check': 'true',
-                              'crashdump_dir': wd,
-                              'poll_sleep_duration': 2}
+                              'crashdump_dir': wd}
 
     # 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='MultiProc')
+        w1.run(plugin='ResourceMultiProc')
     except Exception as e:
         from nipype.pipeline.engine.base import logger
         logger.info('Exception: %s' % str(e))
@@ -737,7 +736,7 @@ def func1(in1):
     # test running the workflow on serial conditions
     error_raised = False
     try:
-        w1.run(plugin='MultiProc')
+        w1.run(plugin='ResourceMultiProc')
     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',):  # , 'MultiProc'):
+    for plugin in ('Linear',):  # , 'ResourceMultiProc'):
         n1 = pe.Node(niu.Function(input_names=['arg1'],
                                   output_names=['out_file1', 'out_file2', 'dir'],
                                   function=test_function),

nipype/pipeline/plugins/__init__.py

@@ -9,7 +9,6 @@
 from .sge import SGEPlugin
 from .condor import CondorPlugin
 from .dagman import CondorDAGManPlugin
-from .multiproc import MultiProcPlugin
 from .multiproc import ResourceMultiProcPlugin
 from .ipython import IPythonPlugin
 from .somaflow import SomaFlowPlugin

nipype/pipeline/plugins/multiproc.py

@@ -76,68 +76,6 @@ class NonDaemonPool(pool.Pool):
     """
     Process = NonDaemonProcess
 
-
-class MultiProcPlugin(DistributedPluginBase):
-    """Execute workflow with multiprocessing
-
-    The plugin_args input to run can be used to control the multiprocessing
-    execution. Currently supported options are:
-
-    - n_procs : number of processes to use
-    - non_daemon : boolean flag to execute as non-daemon processes
-
-    """
-
-    def __init__(self, plugin_args=None):
-        super(MultiProcPlugin, self).__init__(plugin_args=plugin_args)
-        self._taskresult = {}
-        self._taskid = 0
-        non_daemon = True
-        n_procs = cpu_count()
-        if plugin_args:
-            if 'n_procs' in plugin_args:
-                n_procs = plugin_args['n_procs']
-            if 'non_daemon' in plugin_args:
-                non_daemon = plugin_args['non_daemon']
-        if non_daemon:
-            # run the execution using the non-daemon pool subclass
-            self.pool = NonDaemonPool(processes=n_procs)
-        else:
-            self.pool = Pool(processes=n_procs)
-
-
-    def _get_result(self, taskid):
-        if taskid not in self._taskresult:
-            raise RuntimeError('Multiproc task %d not found' % taskid)
-        if not self._taskresult[taskid].ready():
-            return None
-        return self._taskresult[taskid].get()
-
-    def _submit_job(self, node, updatehash=False):
-        self._taskid += 1
-        try:
-            if node.inputs.terminal_output == 'stream':
-                node.inputs.terminal_output = 'allatonce'
-        except:
-            pass
-        self._taskresult[self._taskid] = self.pool.apply_async(run_node, (node,
-                                                                updatehash,))
-        return self._taskid
-
-    def _report_crash(self, node, result=None):
-        if result and result['traceback']:
-            node._result = result['result']
-            node._traceback = result['traceback']
-            return report_crash(node,
-                                traceback=result['traceback'])
-        else:
-            return report_crash(node)
-
-    def _clear_task(self, taskid):
-        del self._taskresult[taskid]
-
-
-
 import numpy as np
 from copy import deepcopy
 from ..engine import (MapNode, str2bool)
@@ -150,8 +88,8 @@ def _clear_task(self, taskid):
 def release_lock(args):
     semaphore_singleton.semaphore.release()
 
-class ResourceMultiProcPlugin(MultiProcPlugin):
-    """Execute workflow with multiprocessing not sending more jobs at once
+class ResourceMultiProcPlugin(DistributedPluginBase):
+    """Execute workflow with multiprocessing, not sending more jobs at once
     than the system can support.
 
     The plugin_args input to run can be used to control the multiprocessing
@@ -167,29 +105,61 @@ class ResourceMultiProcPlugin(MultiProcPlugin):
 
     Currently supported options are:
 
+    - non_daemon : boolean flag to execute as non-daemon processes
     - num_threads: maximum number of threads to be executed in parallel
     - estimated_memory: maximum memory that can be used at once.
 
     """
 
     def __init__(self, plugin_args=None):
         super(ResourceMultiProcPlugin, self).__init__(plugin_args=plugin_args)
+        self._taskresult = {}
+        self._taskid = 0
+        non_daemon = True
         self.plugin_args = plugin_args
         self.processors = cpu_count()
         memory = psutil.virtual_memory()
         self.memory = memory.total / (1024*1024*1024)
         if self.plugin_args:
+            if 'non_daemon' in self.plugin_args:
+                non_daemon = plugin_args['non_daemon']
             if 'n_procs' in self.plugin_args:
                 self.processors = self.plugin_args['n_procs']
             if 'memory' in self.plugin_args:
                 self.memory = self.plugin_args['memory']
 
+        if non_daemon:
+            # run the execution using the non-daemon pool subclass
+            self.pool = NonDaemonPool(processes=n_procs)
+        else:
+            self.pool = Pool(processes=n_procs)
+
     def _wait(self):
         if len(self.pending_tasks) > 0:
             semaphore_singleton.semaphore.acquire()
         semaphore_singleton.semaphore.release()
 
 
+    def _get_result(self, taskid):
+        if taskid not in self._taskresult:
+            raise RuntimeError('Multiproc task %d not found' % taskid)
+        if not self._taskresult[taskid].ready():
+            return None
+        return self._taskresult[taskid].get()
+
+
+    def _report_crash(self, node, result=None):
+        if result and result['traceback']:
+            node._result = result['result']
+            node._traceback = result['traceback']
+            return report_crash(node,
+                                traceback=result['traceback'])
+        else:
+            return report_crash(node)
+
+    def _clear_task(self, taskid):
+        del self._taskresult[taskid]
+
     def _submit_job(self, node, updatehash=False):
         self._taskid += 1
         try:

nipype/pipeline/plugins/tests/test_base.py

@@ -38,5 +38,5 @@ def func(arg1):
 wf.add_nodes([funkynode])
 wf.base_dir = '/tmp'
 
-wf.run(plugin='MultiProc')
+wf.run(plugin='ResourceMultiProc')
 '''

nipype/pipeline/plugins/tests/test_callback.py

@@ -75,8 +75,7 @@ def test_callback_multiproc_normal():
                      name='f_node')
     wf.add_nodes([f_node])
     wf.config['execution']['crashdump_dir'] = wf.base_dir
-    wf.config['execution']['poll_sleep_duration'] = 2
-    wf.run(plugin='MultiProc', plugin_args={'status_callback': so.callback})
+    wf.run(plugin='ResourceMultiProc', plugin_args={'status_callback': so.callback})
     assert_equal(len(so.statuses), 2)
     for (n, s) in so.statuses:
         yield assert_equal, n.name, 'f_node'
@@ -93,9 +92,8 @@ def test_callback_multiproc_exception():
                      name='f_node')
     wf.add_nodes([f_node])
     wf.config['execution']['crashdump_dir'] = wf.base_dir
-    wf.config['execution']['poll_sleep_duration'] = 2
     try:
-        wf.run(plugin='MultiProc',
+        wf.run(plugin='ResourceMultiProc',
                plugin_args={'status_callback': so.callback})
     except:
         pass

nipype/pipeline/plugins/tests/test_multiproc.py

@@ -43,8 +43,7 @@ def test_run_multiproc():
     pipe.connect([(mod1, mod2, [('output1', 'input1')])])
     pipe.base_dir = os.getcwd()
     mod1.inputs.input1 = 1
-    pipe.config['execution']['poll_sleep_duration'] = 2
-    execgraph = pipe.run(plugin="MultiProc")
+    execgraph = pipe.run(plugin="ResourceMultiProc")
     names = ['.'.join((node._hierarchy, node.name)) for node in execgraph.nodes()]
     node = execgraph.nodes()[names.index('pipe.mod1')]
     result = node.get_output('output1')

nipype/pipeline/plugins/tests/test_multiproc_nondaemon.py

@@ -84,7 +84,7 @@ def dummyFunction(filename):
 
 def run_multiproc_nondaemon_with_flag(nondaemon_flag):
     '''
-    Start a pipe with two nodes using the multiproc plugin and passing the nondaemon_flag.
+    Start a pipe with two nodes using the resource multiproc plugin and passing the nondaemon_flag.
     '''
 
     cur_dir = os.getcwd()
@@ -107,11 +107,10 @@ def run_multiproc_nondaemon_with_flag(nondaemon_flag):
     f1.inputs.insum = 0
 
     pipe.config['execution']['stop_on_first_crash'] = True
-    pipe.config['execution']['poll_sleep_duration'] = 2
 
-    # execute the pipe using the MultiProc plugin with 2 processes and the non_daemon flag
+    # execute the pipe using the ResourceMultiProc plugin with 2 processes and the non_daemon flag
     # to enable child processes which start other multiprocessing jobs
-    execgraph = pipe.run(plugin="MultiProc",
+    execgraph = pipe.run(plugin="ResourceMultiProc",
                          plugin_args={'n_procs': 2,
                                       'non_daemon': nondaemon_flag})