Merge pull request nipy#2284 from oesteban/enh/multiproc-maxtasksperchild

[ENH] Set maxtasksperchild in MultiProc Pool

Author: oesteban

doc/users/plugins.rst

@@ -82,6 +82,10 @@ Optional arguments::
   exceed the total amount of resources available (memory and threads), when
   ``False`` (default), only a warning will be issued.
 
+  maxtasksperchild : number of nodes to run on each process before refreshing 
+  the worker (default: 10).
+  
+
 To distribute processing on a multicore machine, simply call::
 
   workflow.run(plugin='MultiProc')

nipype/pipeline/plugins/multiproc.py

@@ -60,6 +60,7 @@ def run_node(node, updatehash, taskid):
 class NonDaemonProcess(Process):
     """A non-daemon process to support internal multiprocessing.
     """
+
     def _get_daemon(self):
         return False
 
@@ -104,6 +105,8 @@ class MultiProcPlugin(DistributedPluginBase):
     - scheduler: sort jobs topologically (``'tsort'``, default value)
         or prioritize jobs by, first, memory consumption and, second,
         number of threads (``'mem_thread'`` option).
+    - maxtasksperchild: number of nodes to run on each process before
+        refreshing the worker (default: 10).
 
     """
 
@@ -116,6 +119,7 @@ def __init__(self, plugin_args=None):
 
         # Read in options or set defaults.
         non_daemon = self.plugin_args.get('non_daemon', True)
+        maxtasks = self.plugin_args.get('maxtasksperchild', 10)
         self.processors = self.plugin_args.get('n_procs', cpu_count())
         self.memory_gb = self.plugin_args.get('memory_gb',  # Allocate 90% of system memory
                                               get_system_total_memory_gb() * 0.9)
@@ -124,7 +128,14 @@ def __init__(self, plugin_args=None):
         # Instantiate different thread pools for non-daemon processes
         logger.debug('MultiProcPlugin starting in "%sdaemon" mode (n_procs=%d, mem_gb=%0.2f)',
                      'non' if non_daemon else '', self.processors, self.memory_gb)
-        self.pool = (NonDaemonPool if non_daemon else Pool)(processes=self.processors)
+
+        NipypePool = NonDaemonPool if non_daemon else Pool
+        try:
+            self.pool = NipypePool(processes=self.processors,
+                                   maxtasksperchild=maxtasks)
+        except TypeError:
+            self.pool = NipypePool(processes=self.processors)
+
         self._stats = None
 
     def _async_callback(self, args):