Processes in MapNodes do not respect thread limits

[effigies]

Summary

If a MapNode's interface is assigned a num_threads greater than half of the available threads, the jobs may nonetheless run in parallel.

Expected behavior

Such jobs ought to run one at a time.

How to replicate the behavior

The following script creates two workflows, each one contains two processes that sleep for 3 seconds and are assigned a thread count of 5. The workflow is run with 6 processes.

#!/usr/bin/env python
from nipype.pipeline import engine as pe
import nipype.interfaces.utility as niu
from functools import partial
import time

def timeit(func):
    tic = time.time()
    func()
    toc = time.time()
    return toc - tic

def sleeper(arg):
    import time
    time.sleep(3)
    
wf1 = pe.Workflow(base_dir='/tmp/nipype', name='wf1')
node1 = pe.Node(niu.Function(function=sleeper), name='node1')
node2 = pe.Node(niu.Function(function=sleeper), name='node2')
node1.inputs.arg = 1
node2.inputs.arg = 2
node1.interface.num_threads = 5
node2.interface.num_threads = 5
wf1.add_nodes([node1, node2])

wf2 = pe.Workflow(base_dir='/tmp/nipype', name='wf2')
mapnode = pe.MapNode(niu.Function(function=sleeper), iterfield='arg', name='mapnode')
mapnode.inputs.arg = [3, 4]
mapnode.interface.num_threads = 5
wf2.add_nodes([mapnode])

time1 = timeit(partial(wf1.run, plugin='MultiProc', plugin_args={'n_procs': 6}))
time2 = timeit(partial(wf2.run, plugin='MultiProc', plugin_args={'n_procs': 6}))

print("Two Nodes: {:.1f}s".format(time1))
print("MapNode: {:.1f}s".format(time2))

Output:

170512-16:03:43,295 workflow INFO:
	 Workflow wf1 settings: ['check', 'execution', 'logging']
170512-16:03:43,298 workflow INFO:
	 Running in parallel.
170512-16:03:43,301 workflow INFO:
	 Executing: node2 ID: 0
170512-16:03:43,305 workflow INFO:
	 Executing node node2 in dir: /tmp/nipype/wf1/node2
170512-16:03:46,369 workflow INFO:
	 [Job finished] jobname: node2 jobid: 0
170512-16:03:46,371 workflow INFO:
	 Executing: node1 ID: 1
170512-16:03:46,375 workflow INFO:
	 Executing node node1 in dir: /tmp/nipype/wf1/node1
170512-16:03:49,439 workflow INFO:
	 [Job finished] jobname: node1 jobid: 1
170512-16:03:49,455 workflow INFO:
	 Workflow wf2 settings: ['check', 'execution', 'logging']
170512-16:03:49,458 workflow INFO:
	 Running in parallel.
170512-16:03:49,460 workflow INFO:
	 Executing: mapnode ID: 0
170512-16:03:49,463 workflow INFO:
	 Adding 2 jobs for mapnode mapnode
170512-16:03:49,468 workflow INFO:
	 Executing: _mapnode0 ID: 1
170512-16:03:49,469 workflow INFO:
	 Executing: _mapnode1 ID: 2
170512-16:03:49,473 workflow INFO:
	 Executing node _mapnode0 in dir: /tmp/nipype/wf2/mapnode/mapflow/_mapnode0
170512-16:03:49,473 workflow INFO:
	 Executing node _mapnode1 in dir: /tmp/nipype/wf2/mapnode/mapflow/_mapnode1
170512-16:03:52,537 workflow INFO:
	 [Job finished] jobname: _mapnode0 jobid: 1
170512-16:03:52,538 workflow INFO:
	 [Job finished] jobname: _mapnode1 jobid: 2
170512-16:03:52,540 workflow INFO:
	 Executing: mapnode ID: 0
170512-16:03:52,544 workflow INFO:
	 Executing node mapnode in dir: /tmp/nipype/wf2/mapnode
170512-16:03:52,555 workflow INFO:
	 Executing node _mapnode0 in dir: /tmp/nipype/wf2/mapnode/mapflow/_mapnode0
170512-16:03:52,556 workflow INFO:
	 Collecting precomputed outputs
170512-16:03:52,561 workflow INFO:
	 Executing node _mapnode1 in dir: /tmp/nipype/wf2/mapnode/mapflow/_mapnode1
170512-16:03:52,563 workflow INFO:
	 Collecting precomputed outputs
170512-16:03:52,606 workflow INFO:
	 [Job finished] jobname: mapnode jobid: 0
Two Nodes: 6.3s
MapNode: 3.2s

Script/Workflow details

See nipreps/fmriprep#506 (comment)

Platform details:

{'pkg_path': '/usr/local/miniconda/lib/python3.6/site-packages/nipype',
 'commit_source': 'installation',
 'commit_hash': 'd82a18f',
 'nipype_version': '0.13.0-dev',
 'sys_version': '3.6.0 |Continuum Analytics, Inc.| (default, Dec 23 2016, 12:22:00) \n[GCC 4.4.7 20120313 (Red Hat 4.4.7-1)]',
 'sys_executable': '/usr/local/miniconda/bin/python',
 'sys_platform': 'linux',
 'numpy_version': '1.12.0',
 'scipy_version': '0.18.1',
 'networkx_version': '1.11',
 'nibabel_version': '2.1.0',
 'traits_version': '4.6.0'}
0.13.0-dev

Execution environment

• Container [Tag: poldracklab/fmriprep:latest]

[effigies]

cc @oesteban @ccraddock