Repository: nipy/nipype

Issue:
Processes in MapNodes do not respect thread limits
### 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.

```Python
#!/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 https://github.com/poldracklab/fmriprep/pull/506#discussion_r116107213

### 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]


Discussion:
satra: lgtm - seems like a simple fix to incorporate in `0.13.1`
effigies: Actually, I should probably copy the memory estimate, too.

For a test, I can do something like:

```Python
mapnode = pe.MapNode(...)
nodes = list(mapnode._make_nodes())
for node in nodes:
    for attr in attrs_we_care_about:
        assert getattr(node, attr) == getattr(mapnode, attr)
```

Any other candidates for `attrs_we_care_about`?
codecov-io: # [Codecov](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=h1) Report
> Merging [#2019](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=desc) into [master](https://codecov.io/gh/nipy/nipype/commit/4358fe5d10b22c036b4c2f3d7cf99d20d88d76ec?src=pr&el=desc) will **increase** coverage by `<.01%`.
> The diff coverage is `93.75%`.

[![Impacted file tree graph](https://codecov.io/gh/nipy/nipype/pull/2019/graphs/tree.svg?height=150&width=650&src=pr&token=Tu0EnSSGVZ)](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master   #2019      +/-   ##
=========================================
+ Coverage   72.19%   72.2%   +<.01%     
=========================================
  Files        1132    1132              
  Lines       57012   57028      +16     
  Branches     8162    8165       +3     
=========================================
+ Hits        41161   41176      +15     
- Misses      14567   14568       +1     
  Partials     1284    1284
```

| Flag | Coverage Δ | |
|---|---|---|
| #smoketests | `72.2% <93.75%> (ø)` | :arrow_up: |
| #unittests | `69.79% <93.75%> (ø)` | :arrow_up: |

| [Impacted Files](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [nipype/pipeline/engine/nodes.py](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=tree#diff-bmlweXBlL3BpcGVsaW5lL2VuZ2luZS9ub2Rlcy5weQ==) | `84.71% <100%> (+0.04%)` | :arrow_up: |
| [nipype/pipeline/engine/tests/test\_engine.py](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=tree#diff-bmlweXBlL3BpcGVsaW5lL2VuZ2luZS90ZXN0cy90ZXN0X2VuZ2luZS5weQ==) | `93.52% <92.85%> (-0.03%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=footer). Last update [4358fe5...6dd54b2](https://codecov.io/gh/nipy/nipype/pull/2019?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

effigies: @satra I moved all the parameters that could be moved into the `Node()`. Seem reasonable?

Patch:
diff --git a/nipype/pipeline/engine/nodes.py b/nipype/pipeline/engine/nodes.py
index cbfa70ceb..d52e589a0 100644
--- a/nipype/pipeline/engine/nodes.py
+++ b/nipype/pipeline/engine/nodes.py
@@ -1112,9 +1112,14 @@ class MapNode(Node):
             nitems = len(filename_to_list(getattr(self.inputs, self.iterfield[0])))
         for i in range(nitems):
             nodename = '_' + self.name + str(i)
-            node = Node(deepcopy(self._interface), name=nodename)
-            node.overwrite = self.overwrite
-            node.run_without_submitting = self.run_without_submitting
+            node = Node(deepcopy(self._interface),
+                        n_procs=self._interface.num_threads,
+                        mem_gb=self._interface.estimated_memory_gb,
+                        overwrite=self.overwrite,
+                        needed_outputs=self.needed_outputs,
+                        run_without_submitting=self.run_without_submitting,
+                        base_dir=op.join(cwd, 'mapflow'),
+                        name=nodename)
             node.plugin_args = self.plugin_args
             node._interface.inputs.set(
                 **deepcopy(self._interface.inputs.get()))
@@ -1126,7 +1131,6 @@ class MapNode(Node):
                 logger.debug('setting input %d %s %s', i, field, fieldvals[i])
                 setattr(node.inputs, field, fieldvals[i])
             node.config = self.config
-            node.base_dir = op.join(cwd, 'mapflow')
             yield i, node
 
     def _node_runner(self, nodes, updatehash=False):


Test patch:
diff --git a/nipype/pipeline/engine/tests/test_engine.py b/nipype/pipeline/engine/tests/test_engine.py
index 5cd107bc6..e2624d03c 100644
--- a/nipype/pipeline/engine/tests/test_engine.py
+++ b/nipype/pipeline/engine/tests/test_engine.py
@@ -486,6 +486,28 @@ def test_mapnode_nested(tmpdir):
     assert "can only concatenate list" in str(excinfo.value)
 
 
+def test_mapnode_expansion(tmpdir):
+    os.chdir(str(tmpdir))
+    from nipype import MapNode, Function
+
+    def func1(in1):
+        return in1 + 1
+
+    mapnode = MapNode(Function(function=func1),
+                      iterfield='in1',
+                      name='mapnode')
+    mapnode.inputs.in1 = [1, 2]
+    mapnode.interface.num_threads = 2
+    mapnode.interface.estimated_memory_gb = 2
+
+    for idx, node in mapnode._make_nodes():
+        for attr in ('overwrite', 'run_without_submitting', 'plugin_args'):
+            assert getattr(node, attr) == getattr(mapnode, attr)
+        for attr in ('num_threads', 'estimated_memory_gb'):
+            assert (getattr(node._interface, attr) ==
+                    getattr(mapnode._interface, attr))
+
+
 def test_node_hash(tmpdir):
     wd = str(tmpdir)
     os.chdir(wd)