Skip to content

[FIX] MultiProc starting workers at dubious wd #2368

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
Jan 19, 2018
Merged

[FIX] MultiProc starting workers at dubious wd #2368

Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
bcc25fd
[FIX] MultiProc starting workers at dubious wd
oesteban Jan 10, 2018
70da04e
less verbose, update CHANGES
oesteban Jan 10, 2018
33254fa
Merge branch 'master' into fix/multi-proc-cwd
oesteban Jan 11, 2018
9ebc954
Merge branch 'master' into fix/multi-proc-cwd
oesteban Jan 12, 2018
4db0e7c
Merge remote-tracking branch 'upstream/master' into fix/multi-proc-cwd
oesteban Jan 17, 2018
a9b2609
add runtime.cwd to report.rst
oesteban Jan 18, 2018
86cac7b
Merge remote-tracking branch 'upstream/master' into fix/multi-proc-cwd
oesteban Jan 19, 2018
af6c708
sty changes
oesteban Jan 19, 2018
0e2abb1
[skip ci] Merge remote-tracking branch 'upstream/master' into fix/mul…
oesteban Jan 19, 2018
5b5547b
[skip ci] Merge branch 'fix/multi-proc-cwd' of github.com:oesteban/ni…
oesteban Jan 19, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 24 additions & 4 deletions nipype/pipeline/plugins/multiproc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
-------
Expand All @@ -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.
"""
Expand Down Expand Up @@ -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()
Copy link
Member

@effigies effigies Jan 10, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this the workflow base directory or the CWD of the shell? Could this cause things to dump into the local directory?

Copy link
Contributor Author

@oesteban oesteban Jan 10, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It shouldn't because interfaces handle their WD. I think this is fixing an edge case for fmriprep where we are spinning up and killing workers all the time.

Copy link
Contributor Author

@oesteban oesteban Jan 17, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok it turns out that some SimpleInterfaces are writing to the workflow base directory :(

Copy link
Member

@effigies effigies Jan 17, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well that's... suboptimal.

Copy link
Contributor Author

@oesteban oesteban Jan 17, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's actually bothering, because it means that without this patch, those interfaces are being run in some other unexpected path.

Copy link
Contributor Author

@oesteban oesteban Jan 19, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Apparently it is a problem only with SimpleInterface


# Read in options or set defaults.
non_daemon = self.plugin_args.get('non_daemon', True)
maxtasks = self.plugin_args.get('maxtasksperchild', 10)
Expand All @@ -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,
Copy link
Member

@effigies effigies Jan 10, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a reason this shouldn't just be os.chdir?

Copy link
Contributor Author

@oesteban oesteban Jan 10, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not really

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):
Expand Down