[FIX] MultiProc starting workers at dubious wd

When ``maxtasksperchild`` was set at a very low value, workers
are created very often, sometimes at working directories deleted
after the interface cleanup. That would trigger an ``OSError``
when calling ``os.getcwd()`` during ``nipype.config`` import.

This PR sets an initializer for the workers that just changes to
the appropriate working directory before the worker is spun up.

Fixes nipreps/fmriprep#868

Author: oesteban

nipype/pipeline/plugins/multiproc.py

@@ -9,6 +9,7 @@
 from __future__ import print_function, division, unicode_literals, absolute_import
 
 # Import packages
+import os
 from multiprocessing import Process, Pool, cpu_count, pool
 from traceback import format_exception
 import sys
@@ -47,6 +48,8 @@ def run_node(node, updatehash, taskid):
         the node to run
     updatehash : boolean
         flag for updating hash
+    taskid : int
+        an identifier for this task
 
     Returns
     -------
@@ -68,6 +71,10 @@ def run_node(node, updatehash, taskid):
     return result
 
 
+def _init_worker(cwd):
+    os.chdir(cwd)
+
+
 class NonDaemonProcess(Process):
     """A non-daemon process to support internal multiprocessing.
     """
@@ -128,6 +135,10 @@ def __init__(self, plugin_args=None):
         self._task_obj = {}
         self._taskid = 0
 
+        # Cache current working directory and make sure we
+        # change to it when workers are set up
+        self._cwd = os.getcwd()
+
         # Read in options or set defaults.
         non_daemon = self.plugin_args.get('non_daemon', True)
         maxtasks = self.plugin_args.get('maxtasksperchild', 10)
@@ -140,19 +151,28 @@ def __init__(self, plugin_args=None):
 
         # Instantiate different thread pools for non-daemon processes
         logger.debug('[MultiProc] Starting in "%sdaemon" mode (n_procs=%d, '
-                     'mem_gb=%0.2f)', 'non' * int(non_daemon), self.processors,
-                     self.memory_gb)
+                     'mem_gb=%0.2f, cwd=%s)', 'non' * int(non_daemon),
+                     self.processors, self.memory_gb, self._cwd)
 
         NipypePool = NonDaemonPool if non_daemon else Pool
         try:
-            self.pool = NipypePool(processes=self.processors,
-                                   maxtasksperchild=maxtasks)
+            self.pool = NipypePool(
+                processes=self.processors,
+                maxtasksperchild=maxtasks,
+                initializer=_init_worker,
+                initargs=(self._cwd,)
+            )
         except TypeError:
+            # Python < 3.2 does not have maxtasksperchild
+            # When maxtasksperchild is not set, initializer is not to be
+            # called.
             self.pool = NipypePool(processes=self.processors)
 
         self._stats = None
 
     def _async_callback(self, args):
+        # Make sure runtime is not left at a dubious working directory
+        os.chdir(self._cwd)
         self._taskresult[args['taskid']] = args
 
     def _get_result(self, taskid):