We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] MultiProc crashes are not correctly handled
### Summary
Since the latest refactor, MultiProc calls `self._report_crash` with a `traceback` argument and such function is not defined:

```
File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/engine/workflows.py", line 591, in run
runner.run(execgraph, updatehash=updatehash, config=self.config)
File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/plugins/base.py", line 182, in run
self._send_procs_to_workers(updatehash=updatehash, graph=graph)
File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/plugins/multiproc.py", line 250, in _send_procs_to_workers
self._report_crash(self.procs[jobid], traceback=traceback)
TypeError: _report_crash() got an unexpected keyword argument 'traceback'
```

I'll propose a fix today.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nipype/pipeline/plugins/multiproc.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
3 # vi: set ft=python sts=4 ts=4 sw=4 et:
4 """Parallel workflow execution via multiprocessing
5 
6 Support for child processes running as non-daemons based on
7 http://stackoverflow.com/a/8963618/1183453
8 """
9 from __future__ import print_function, division, unicode_literals, absolute_import
10 
11 # Import packages
12 from multiprocessing import Process, Pool, cpu_count, pool
13 from traceback import format_exception
14 import sys
15 
16 from copy import deepcopy
17 import numpy as np
18 
19 from ... import logging
20 from ...utils.profiler import get_system_total_memory_gb
21 from ..engine import MapNode
22 from .base import DistributedPluginBase
23 
24 # Init logger
25 logger = logging.getLogger('workflow')
26 
27 
28 # Run node
29 def run_node(node, updatehash, taskid):
30     """Function to execute node.run(), catch and log any errors and
31     return the result dictionary
32 
33     Parameters
34     ----------
35     node : nipype Node instance
36         the node to run
37     updatehash : boolean
38         flag for updating hash
39 
40     Returns
41     -------
42     result : dictionary
43         dictionary containing the node runtime results and stats
44     """
45 
46     # Init variables
47     result = dict(result=None, traceback=None, taskid=taskid)
48 
49     # Try and execute the node via node.run()
50     try:
51         result['result'] = node.run(updatehash=updatehash)
52     except:
53         result['traceback'] = format_exception(*sys.exc_info())
54         result['result'] = node.result
55 
56     # Return the result dictionary
57     return result
58 
59 
60 class NonDaemonProcess(Process):
61     """A non-daemon process to support internal multiprocessing.
62     """
63 
64     def _get_daemon(self):
65         return False
66 
67     def _set_daemon(self, value):
68         pass
69 
70     daemon = property(_get_daemon, _set_daemon)
71 
72 
73 class NonDaemonPool(pool.Pool):
74     """A process pool with non-daemon processes.
75     """
76     Process = NonDaemonProcess
77 
78 
79 class MultiProcPlugin(DistributedPluginBase):
80     """
81     Execute workflow with multiprocessing, not sending more jobs at once
82     than the system can support.
83 
84     The plugin_args input to run can be used to control the multiprocessing
85     execution and defining the maximum amount of memory and threads that
86     should be used. When those parameters are not specified,
87     the number of threads and memory of the system is used.
88 
89     System consuming nodes should be tagged::
90 
91       memory_consuming_node.mem_gb = 8
92       thread_consuming_node.n_procs = 16
93 
94     The default number of threads and memory are set at node
95     creation, and are 1 and 0.25GB respectively.
96 
97     Currently supported options are:
98 
99     - non_daemon : boolean flag to execute as non-daemon processes
100     - n_procs: maximum number of threads to be executed in parallel
101     - memory_gb: maximum memory (in GB) that can be used at once.
102     - raise_insufficient: raise error if the requested resources for
103         a node over the maximum `n_procs` and/or `memory_gb`
104         (default is ``True``).
105     - scheduler: sort jobs topologically (``'tsort'``, default value)
106         or prioritize jobs by, first, memory consumption and, second,
107         number of threads (``'mem_thread'`` option).
108     - maxtasksperchild: number of nodes to run on each process before
109         refreshing the worker (default: 10).
110 
111     """
112 
113     def __init__(self, plugin_args=None):
114         # Init variables and instance attributes
115         super(MultiProcPlugin, self).__init__(plugin_args=plugin_args)
116         self._taskresult = {}
117         self._task_obj = {}
118         self._taskid = 0
119 
120         # Read in options or set defaults.
121         non_daemon = self.plugin_args.get('non_daemon', True)
122         maxtasks = self.plugin_args.get('maxtasksperchild', 10)
123         self.processors = self.plugin_args.get('n_procs', cpu_count())
124         self.memory_gb = self.plugin_args.get('memory_gb',  # Allocate 90% of system memory
125                                               get_system_total_memory_gb() * 0.9)
126         self.raise_insufficient = self.plugin_args.get('raise_insufficient', True)
127 
128         # Instantiate different thread pools for non-daemon processes
129         logger.debug('MultiProcPlugin starting in "%sdaemon" mode (n_procs=%d, mem_gb=%0.2f)',
130                      'non' * int(non_daemon), self.processors, self.memory_gb)
131 
132         NipypePool = NonDaemonPool if non_daemon else Pool
133         try:
134             self.pool = NipypePool(processes=self.processors,
135                                    maxtasksperchild=maxtasks)
136         except TypeError:
137             self.pool = NipypePool(processes=self.processors)
138 
139         self._stats = None
140 
141     def _async_callback(self, args):
142         self._taskresult[args['taskid']] = args
143 
144     def _get_result(self, taskid):
145         return self._taskresult.get(taskid)
146 
147     def _clear_task(self, taskid):
148         del self._task_obj[taskid]
149 
150     def _submit_job(self, node, updatehash=False):
151         self._taskid += 1
152 
153         # Don't allow streaming outputs
154         if getattr(node.interface, 'terminal_output', '') == 'stream':
155             node.interface.terminal_output = 'allatonce'
156 
157         self._task_obj[self._taskid] = self.pool.apply_async(
158             run_node, (node, updatehash, self._taskid),
159             callback=self._async_callback)
160 
161         logger.debug('MultiProc submitted task %s (taskid=%d).',
162                      node.fullname, self._taskid)
163         return self._taskid
164 
165     def _prerun_check(self, graph):
166         """Check if any node exeeds the available resources"""
167         tasks_mem_gb = []
168         tasks_num_th = []
169         for node in graph.nodes():
170             tasks_mem_gb.append(node.mem_gb)
171             tasks_num_th.append(node.n_procs)
172 
173         if np.any(np.array(tasks_mem_gb) > self.memory_gb):
174             logger.warning(
175                 'Some nodes exceed the total amount of memory available '
176                 '(%0.2fGB).', self.memory_gb)
177             if self.raise_insufficient:
178                 raise RuntimeError('Insufficient resources available for job')
179 
180         if np.any(np.array(tasks_num_th) > self.processors):
181             logger.warning(
182                 'Some nodes demand for more threads than available (%d).',
183                 self.processors)
184             if self.raise_insufficient:
185                 raise RuntimeError('Insufficient resources available for job')
186 
187     def _postrun_check(self):
188         self.pool.close()
189 
190     def _check_resources(self, running_tasks):
191         """
192         Make sure there are resources available
193         """
194         free_memory_gb = self.memory_gb
195         free_processors = self.processors
196         for _, jobid in running_tasks:
197             free_memory_gb -= min(self.procs[jobid].mem_gb, free_memory_gb)
198             free_processors -= min(self.procs[jobid].n_procs, free_processors)
199 
200         return free_memory_gb, free_processors
201 
202     def _send_procs_to_workers(self, updatehash=False, graph=None):
203         """
204         Sends jobs to workers when system resources are available.
205         """
206 
207         # Check to see if a job is available (jobs without dependencies not run)
208         # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722
209         jobids = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]
210 
211         # Check available system resources by summing all threads and memory used
212         free_memory_gb, free_processors = self._check_resources(self.pending_tasks)
213 
214         stats = (len(self.pending_tasks), len(jobids), free_memory_gb,
215                  self.memory_gb, free_processors, self.processors)
216         if self._stats != stats:
217             logger.info('Currently running %d tasks, and %d jobs ready. Free '
218                         'memory (GB): %0.2f/%0.2f, Free processors: %d/%d',
219                         *stats)
220             self._stats = stats
221 
222         if free_memory_gb < 0.01 or free_processors == 0:
223             logger.debug('No resources available')
224             return
225 
226         if len(jobids) + len(self.pending_tasks) == 0:
227             logger.debug('No tasks are being run, and no jobs can '
228                          'be submitted to the queue. Potential deadlock')
229             return
230 
231         jobids = self._sort_jobs(jobids, scheduler=self.plugin_args.get('scheduler'))
232 
233         # Submit jobs
234         for jobid in jobids:
235             # First expand mapnodes
236             if isinstance(self.procs[jobid], MapNode):
237                 try:
238                     num_subnodes = self.procs[jobid].num_subnodes()
239                 except Exception:
240                     traceback = format_exception(*sys.exc_info())
241                     self._report_crash(self.procs[jobid], traceback=traceback)
242                     self._clean_queue(jobid, graph)
243                     self.proc_pending[jobid] = False
244                     continue
245                 if num_subnodes > 1:
246                     submit = self._submit_mapnode(jobid)
247                     if not submit:
248                         continue
249 
250             # Check requirements of this job
251             next_job_gb = min(self.procs[jobid].mem_gb, self.memory_gb)
252             next_job_th = min(self.procs[jobid].n_procs, self.processors)
253 
254             # If node does not fit, skip at this moment
255             if next_job_th > free_processors or next_job_gb > free_memory_gb:
256                 logger.debug('Cannot allocate job %d (%0.2fGB, %d threads).',
257                              jobid, next_job_gb, next_job_th)
258                 continue
259 
260             free_memory_gb -= next_job_gb
261             free_processors -= next_job_th
262             logger.debug('Allocating %s ID=%d (%0.2fGB, %d threads). Free: %0.2fGB, %d threads.',
263                          self.procs[jobid].fullname, jobid, next_job_gb, next_job_th,
264                          free_memory_gb, free_processors)
265 
266             # change job status in appropriate queues
267             self.proc_done[jobid] = True
268             self.proc_pending[jobid] = True
269 
270             # If cached just retrieve it, don't run
271             if self._local_hash_check(jobid, graph):
272                 continue
273 
274             if self.procs[jobid].run_without_submitting:
275                 logger.debug('Running node %s on master thread',
276                              self.procs[jobid])
277                 try:
278                     self.procs[jobid].run()
279                 except Exception:
280                     traceback = format_exception(*sys.exc_info())
281                     self._report_crash(self.procs[jobid], traceback=traceback)
282 
283                 # Release resources
284                 self._task_finished_cb(jobid)
285                 self._remove_node_dirs()
286                 free_memory_gb += next_job_gb
287                 free_processors += next_job_th
288                 # Display stats next loop
289                 self._stats = None
290                 continue
291 
292             # Task should be submitted to workers
293             # Send job to task manager and add to pending tasks
294             if self._status_callback:
295                 self._status_callback(self.procs[jobid], 'start')
296             tid = self._submit_job(deepcopy(self.procs[jobid]),
297                                    updatehash=updatehash)
298             if tid is None:
299                 self.proc_done[jobid] = False
300                 self.proc_pending[jobid] = False
301             else:
302                 self.pending_tasks.insert(0, (tid, jobid))
303             # Display stats next loop
304             self._stats = None
305 
306     def _sort_jobs(self, jobids, scheduler='tsort'):
307         if scheduler == 'mem_thread':
308             return sorted(jobids, key=lambda item: (
309                 self.procs[item].mem_gb, self.procs[item].n_procs))
310         return jobids
311
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nipype/pipeline/plugins/multiproc.py b/nipype/pipeline/plugins/multiproc.py
--- a/nipype/pipeline/plugins/multiproc.py
+++ b/nipype/pipeline/plugins/multiproc.py
@@ -238,8 +238,10 @@
                     num_subnodes = self.procs[jobid].num_subnodes()
                 except Exception:
                     traceback = format_exception(*sys.exc_info())
-                    self._report_crash(self.procs[jobid], traceback=traceback)
-                    self._clean_queue(jobid, graph)
+                    self._clean_queue(
+                        jobid, graph,
+                        result={'result': None, 'traceback': traceback}
+                    )
                     self.proc_pending[jobid] = False
                     continue
                 if num_subnodes > 1:
@@ -275,10 +277,13 @@
                 logger.debug('Running node %s on master thread',
                              self.procs[jobid])
                 try:
-                    self.procs[jobid].run()
+                    self.procs[jobid].run(updatehash=updatehash)
                 except Exception:
                     traceback = format_exception(*sys.exc_info())
-                    self._report_crash(self.procs[jobid], traceback=traceback)
+                    self._clean_queue(
+                        jobid, graph,
+                        result={'result': None, 'traceback': traceback}
+                    )
 
                 # Release resources
                 self._task_finished_cb(jobid)

{"golden_diff": "diff --git a/nipype/pipeline/plugins/multiproc.py b/nipype/pipeline/plugins/multiproc.py\n--- a/nipype/pipeline/plugins/multiproc.py\n+++ b/nipype/pipeline/plugins/multiproc.py\n@@ -238,8 +238,10 @@\n                     num_subnodes = self.procs[jobid].num_subnodes()\n                 except Exception:\n                     traceback = format_exception(*sys.exc_info())\n-                    self._report_crash(self.procs[jobid], traceback=traceback)\n-                    self._clean_queue(jobid, graph)\n+                    self._clean_queue(\n+                        jobid, graph,\n+                        result={'result': None, 'traceback': traceback}\n+                    )\n                     self.proc_pending[jobid] = False\n                     continue\n                 if num_subnodes > 1:\n@@ -275,10 +277,13 @@\n                 logger.debug('Running node %s on master thread',\n                              self.procs[jobid])\n                 try:\n-                    self.procs[jobid].run()\n+                    self.procs[jobid].run(updatehash=updatehash)\n                 except Exception:\n                     traceback = format_exception(*sys.exc_info())\n-                    self._report_crash(self.procs[jobid], traceback=traceback)\n+                    self._clean_queue(\n+                        jobid, graph,\n+                        result={'result': None, 'traceback': traceback}\n+                    )\n \n                 # Release resources\n                 self._task_finished_cb(jobid)\n", "issue": "[BUG] MultiProc crashes are not correctly handled\n### Summary\r\nSince the latest refactor, MultiProc calls `self._report_crash` with a `traceback` argument and such function is not defined:\r\n\r\n```\r\nFile \"/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/engine/workflows.py\", line 591, in run\r\nrunner.run(execgraph, updatehash=updatehash, config=self.config)\r\nFile \"/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/plugins/base.py\", line 182, in run\r\nself._send_procs_to_workers(updatehash=updatehash, graph=graph)\r\nFile \"/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/plugins/multiproc.py\", line 250, in _send_procs_to_workers\r\nself._report_crash(self.procs[jobid], traceback=traceback)\r\nTypeError: _report_crash() got an unexpected keyword argument 'traceback'\r\n```\r\n\r\nI'll propose a fix today.\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-\n# vi: set ft=python sts=4 ts=4 sw=4 et:\n\"\"\"Parallel workflow execution via multiprocessing\n\nSupport for child processes running as non-daemons based on\nhttp://stackoverflow.com/a/8963618/1183453\n\"\"\"\nfrom __future__ import print_function, division, unicode_literals, absolute_import\n\n# Import packages\nfrom multiprocessing import Process, Pool, cpu_count, pool\nfrom traceback import format_exception\nimport sys\n\nfrom copy import deepcopy\nimport numpy as np\n\nfrom ... import logging\nfrom ...utils.profiler import get_system_total_memory_gb\nfrom ..engine import MapNode\nfrom .base import DistributedPluginBase\n\n# Init logger\nlogger = logging.getLogger('workflow')\n\n\n# Run node\ndef run_node(node, updatehash, taskid):\n    \"\"\"Function to execute node.run(), catch and log any errors and\n    return the result dictionary\n\n    Parameters\n    ----------\n    node : nipype Node instance\n        the node to run\n    updatehash : boolean\n        flag for updating hash\n\n    Returns\n    -------\n    result : dictionary\n        dictionary containing the node runtime results and stats\n    \"\"\"\n\n    # Init variables\n    result = dict(result=None, traceback=None, taskid=taskid)\n\n    # Try and execute the node via node.run()\n    try:\n        result['result'] = node.run(updatehash=updatehash)\n    except:\n        result['traceback'] = format_exception(*sys.exc_info())\n        result['result'] = node.result\n\n    # Return the result dictionary\n    return result\n\n\nclass NonDaemonProcess(Process):\n    \"\"\"A non-daemon process to support internal multiprocessing.\n    \"\"\"\n\n    def _get_daemon(self):\n        return False\n\n    def _set_daemon(self, value):\n        pass\n\n    daemon = property(_get_daemon, _set_daemon)\n\n\nclass NonDaemonPool(pool.Pool):\n    \"\"\"A process pool with non-daemon processes.\n    \"\"\"\n    Process = NonDaemonProcess\n\n\nclass MultiProcPlugin(DistributedPluginBase):\n    \"\"\"\n    Execute workflow with multiprocessing, not sending more jobs at once\n    than the system can support.\n\n    The plugin_args input to run can be used to control the multiprocessing\n    execution and defining the maximum amount of memory and threads that\n    should be used. When those parameters are not specified,\n    the number of threads and memory of the system is used.\n\n    System consuming nodes should be tagged::\n\n      memory_consuming_node.mem_gb = 8\n      thread_consuming_node.n_procs = 16\n\n    The default number of threads and memory are set at node\n    creation, and are 1 and 0.25GB respectively.\n\n    Currently supported options are:\n\n    - non_daemon : boolean flag to execute as non-daemon processes\n    - n_procs: maximum number of threads to be executed in parallel\n    - memory_gb: maximum memory (in GB) that can be used at once.\n    - raise_insufficient: raise error if the requested resources for\n        a node over the maximum `n_procs` and/or `memory_gb`\n        (default is ``True``).\n    - scheduler: sort jobs topologically (``'tsort'``, default value)\n        or prioritize jobs by, first, memory consumption and, second,\n        number of threads (``'mem_thread'`` option).\n    - maxtasksperchild: number of nodes to run on each process before\n        refreshing the worker (default: 10).\n\n    \"\"\"\n\n    def __init__(self, plugin_args=None):\n        # Init variables and instance attributes\n        super(MultiProcPlugin, self).__init__(plugin_args=plugin_args)\n        self._taskresult = {}\n        self._task_obj = {}\n        self._taskid = 0\n\n        # Read in options or set defaults.\n        non_daemon = self.plugin_args.get('non_daemon', True)\n        maxtasks = self.plugin_args.get('maxtasksperchild', 10)\n        self.processors = self.plugin_args.get('n_procs', cpu_count())\n        self.memory_gb = self.plugin_args.get('memory_gb',  # Allocate 90% of system memory\n                                              get_system_total_memory_gb() * 0.9)\n        self.raise_insufficient = self.plugin_args.get('raise_insufficient', True)\n\n        # Instantiate different thread pools for non-daemon processes\n        logger.debug('MultiProcPlugin starting in \"%sdaemon\" mode (n_procs=%d, mem_gb=%0.2f)',\n                     'non' * int(non_daemon), self.processors, self.memory_gb)\n\n        NipypePool = NonDaemonPool if non_daemon else Pool\n        try:\n            self.pool = NipypePool(processes=self.processors,\n                                   maxtasksperchild=maxtasks)\n        except TypeError:\n            self.pool = NipypePool(processes=self.processors)\n\n        self._stats = None\n\n    def _async_callback(self, args):\n        self._taskresult[args['taskid']] = args\n\n    def _get_result(self, taskid):\n        return self._taskresult.get(taskid)\n\n    def _clear_task(self, taskid):\n        del self._task_obj[taskid]\n\n    def _submit_job(self, node, updatehash=False):\n        self._taskid += 1\n\n        # Don't allow streaming outputs\n        if getattr(node.interface, 'terminal_output', '') == 'stream':\n            node.interface.terminal_output = 'allatonce'\n\n        self._task_obj[self._taskid] = self.pool.apply_async(\n            run_node, (node, updatehash, self._taskid),\n            callback=self._async_callback)\n\n        logger.debug('MultiProc submitted task %s (taskid=%d).',\n                     node.fullname, self._taskid)\n        return self._taskid\n\n    def _prerun_check(self, graph):\n        \"\"\"Check if any node exeeds the available resources\"\"\"\n        tasks_mem_gb = []\n        tasks_num_th = []\n        for node in graph.nodes():\n            tasks_mem_gb.append(node.mem_gb)\n            tasks_num_th.append(node.n_procs)\n\n        if np.any(np.array(tasks_mem_gb) > self.memory_gb):\n            logger.warning(\n                'Some nodes exceed the total amount of memory available '\n                '(%0.2fGB).', self.memory_gb)\n            if self.raise_insufficient:\n                raise RuntimeError('Insufficient resources available for job')\n\n        if np.any(np.array(tasks_num_th) > self.processors):\n            logger.warning(\n                'Some nodes demand for more threads than available (%d).',\n                self.processors)\n            if self.raise_insufficient:\n                raise RuntimeError('Insufficient resources available for job')\n\n    def _postrun_check(self):\n        self.pool.close()\n\n    def _check_resources(self, running_tasks):\n        \"\"\"\n        Make sure there are resources available\n        \"\"\"\n        free_memory_gb = self.memory_gb\n        free_processors = self.processors\n        for _, jobid in running_tasks:\n            free_memory_gb -= min(self.procs[jobid].mem_gb, free_memory_gb)\n            free_processors -= min(self.procs[jobid].n_procs, free_processors)\n\n        return free_memory_gb, free_processors\n\n    def _send_procs_to_workers(self, updatehash=False, graph=None):\n        \"\"\"\n        Sends jobs to workers when system resources are available.\n        \"\"\"\n\n        # Check to see if a job is available (jobs without dependencies not run)\n        # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722\n        jobids = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]\n\n        # Check available system resources by summing all threads and memory used\n        free_memory_gb, free_processors = self._check_resources(self.pending_tasks)\n\n        stats = (len(self.pending_tasks), len(jobids), free_memory_gb,\n                 self.memory_gb, free_processors, self.processors)\n        if self._stats != stats:\n            logger.info('Currently running %d tasks, and %d jobs ready. Free '\n                        'memory (GB): %0.2f/%0.2f, Free processors: %d/%d',\n                        *stats)\n            self._stats = stats\n\n        if free_memory_gb < 0.01 or free_processors == 0:\n            logger.debug('No resources available')\n            return\n\n        if len(jobids) + len(self.pending_tasks) == 0:\n            logger.debug('No tasks are being run, and no jobs can '\n                         'be submitted to the queue. Potential deadlock')\n            return\n\n        jobids = self._sort_jobs(jobids, scheduler=self.plugin_args.get('scheduler'))\n\n        # Submit jobs\n        for jobid in jobids:\n            # First expand mapnodes\n            if isinstance(self.procs[jobid], MapNode):\n                try:\n                    num_subnodes = self.procs[jobid].num_subnodes()\n                except Exception:\n                    traceback = format_exception(*sys.exc_info())\n                    self._report_crash(self.procs[jobid], traceback=traceback)\n                    self._clean_queue(jobid, graph)\n                    self.proc_pending[jobid] = False\n                    continue\n                if num_subnodes > 1:\n                    submit = self._submit_mapnode(jobid)\n                    if not submit:\n                        continue\n\n            # Check requirements of this job\n            next_job_gb = min(self.procs[jobid].mem_gb, self.memory_gb)\n            next_job_th = min(self.procs[jobid].n_procs, self.processors)\n\n            # If node does not fit, skip at this moment\n            if next_job_th > free_processors or next_job_gb > free_memory_gb:\n                logger.debug('Cannot allocate job %d (%0.2fGB, %d threads).',\n                             jobid, next_job_gb, next_job_th)\n                continue\n\n            free_memory_gb -= next_job_gb\n            free_processors -= next_job_th\n            logger.debug('Allocating %s ID=%d (%0.2fGB, %d threads). Free: %0.2fGB, %d threads.',\n                         self.procs[jobid].fullname, jobid, next_job_gb, next_job_th,\n                         free_memory_gb, free_processors)\n\n            # change job status in appropriate queues\n            self.proc_done[jobid] = True\n            self.proc_pending[jobid] = True\n\n            # If cached just retrieve it, don't run\n            if self._local_hash_check(jobid, graph):\n                continue\n\n            if self.procs[jobid].run_without_submitting:\n                logger.debug('Running node %s on master thread',\n                             self.procs[jobid])\n                try:\n                    self.procs[jobid].run()\n                except Exception:\n                    traceback = format_exception(*sys.exc_info())\n                    self._report_crash(self.procs[jobid], traceback=traceback)\n\n                # Release resources\n                self._task_finished_cb(jobid)\n                self._remove_node_dirs()\n                free_memory_gb += next_job_gb\n                free_processors += next_job_th\n                # Display stats next loop\n                self._stats = None\n                continue\n\n            # Task should be submitted to workers\n            # Send job to task manager and add to pending tasks\n            if self._status_callback:\n                self._status_callback(self.procs[jobid], 'start')\n            tid = self._submit_job(deepcopy(self.procs[jobid]),\n                                   updatehash=updatehash)\n            if tid is None:\n                self.proc_done[jobid] = False\n                self.proc_pending[jobid] = False\n            else:\n                self.pending_tasks.insert(0, (tid, jobid))\n            # Display stats next loop\n            self._stats = None\n\n    def _sort_jobs(self, jobids, scheduler='tsort'):\n        if scheduler == 'mem_thread':\n            return sorted(jobids, key=lambda item: (\n                self.procs[item].mem_gb, self.procs[item].n_procs))\n        return jobids\n", "path": "nipype/pipeline/plugins/multiproc.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-\n# vi: set ft=python sts=4 ts=4 sw=4 et:\n\"\"\"Parallel workflow execution via multiprocessing\n\nSupport for child processes running as non-daemons based on\nhttp://stackoverflow.com/a/8963618/1183453\n\"\"\"\nfrom __future__ import print_function, division, unicode_literals, absolute_import\n\n# Import packages\nfrom multiprocessing import Process, Pool, cpu_count, pool\nfrom traceback import format_exception\nimport sys\n\nfrom copy import deepcopy\nimport numpy as np\n\nfrom ... import logging\nfrom ...utils.profiler import get_system_total_memory_gb\nfrom ..engine import MapNode\nfrom .base import DistributedPluginBase\n\n# Init logger\nlogger = logging.getLogger('workflow')\n\n\n# Run node\ndef run_node(node, updatehash, taskid):\n    \"\"\"Function to execute node.run(), catch and log any errors and\n    return the result dictionary\n\n    Parameters\n    ----------\n    node : nipype Node instance\n        the node to run\n    updatehash : boolean\n        flag for updating hash\n\n    Returns\n    -------\n    result : dictionary\n        dictionary containing the node runtime results and stats\n    \"\"\"\n\n    # Init variables\n    result = dict(result=None, traceback=None, taskid=taskid)\n\n    # Try and execute the node via node.run()\n    try:\n        result['result'] = node.run(updatehash=updatehash)\n    except:\n        result['traceback'] = format_exception(*sys.exc_info())\n        result['result'] = node.result\n\n    # Return the result dictionary\n    return result\n\n\nclass NonDaemonProcess(Process):\n    \"\"\"A non-daemon process to support internal multiprocessing.\n    \"\"\"\n\n    def _get_daemon(self):\n        return False\n\n    def _set_daemon(self, value):\n        pass\n\n    daemon = property(_get_daemon, _set_daemon)\n\n\nclass NonDaemonPool(pool.Pool):\n    \"\"\"A process pool with non-daemon processes.\n    \"\"\"\n    Process = NonDaemonProcess\n\n\nclass MultiProcPlugin(DistributedPluginBase):\n    \"\"\"\n    Execute workflow with multiprocessing, not sending more jobs at once\n    than the system can support.\n\n    The plugin_args input to run can be used to control the multiprocessing\n    execution and defining the maximum amount of memory and threads that\n    should be used. When those parameters are not specified,\n    the number of threads and memory of the system is used.\n\n    System consuming nodes should be tagged::\n\n      memory_consuming_node.mem_gb = 8\n      thread_consuming_node.n_procs = 16\n\n    The default number of threads and memory are set at node\n    creation, and are 1 and 0.25GB respectively.\n\n    Currently supported options are:\n\n    - non_daemon : boolean flag to execute as non-daemon processes\n    - n_procs: maximum number of threads to be executed in parallel\n    - memory_gb: maximum memory (in GB) that can be used at once.\n    - raise_insufficient: raise error if the requested resources for\n        a node over the maximum `n_procs` and/or `memory_gb`\n        (default is ``True``).\n    - scheduler: sort jobs topologically (``'tsort'``, default value)\n        or prioritize jobs by, first, memory consumption and, second,\n        number of threads (``'mem_thread'`` option).\n    - maxtasksperchild: number of nodes to run on each process before\n        refreshing the worker (default: 10).\n\n    \"\"\"\n\n    def __init__(self, plugin_args=None):\n        # Init variables and instance attributes\n        super(MultiProcPlugin, self).__init__(plugin_args=plugin_args)\n        self._taskresult = {}\n        self._task_obj = {}\n        self._taskid = 0\n\n        # Read in options or set defaults.\n        non_daemon = self.plugin_args.get('non_daemon', True)\n        maxtasks = self.plugin_args.get('maxtasksperchild', 10)\n        self.processors = self.plugin_args.get('n_procs', cpu_count())\n        self.memory_gb = self.plugin_args.get('memory_gb',  # Allocate 90% of system memory\n                                              get_system_total_memory_gb() * 0.9)\n        self.raise_insufficient = self.plugin_args.get('raise_insufficient', True)\n\n        # Instantiate different thread pools for non-daemon processes\n        logger.debug('MultiProcPlugin starting in \"%sdaemon\" mode (n_procs=%d, mem_gb=%0.2f)',\n                     'non' * int(non_daemon), self.processors, self.memory_gb)\n\n        NipypePool = NonDaemonPool if non_daemon else Pool\n        try:\n            self.pool = NipypePool(processes=self.processors,\n                                   maxtasksperchild=maxtasks)\n        except TypeError:\n            self.pool = NipypePool(processes=self.processors)\n\n        self._stats = None\n\n    def _async_callback(self, args):\n        self._taskresult[args['taskid']] = args\n\n    def _get_result(self, taskid):\n        return self._taskresult.get(taskid)\n\n    def _clear_task(self, taskid):\n        del self._task_obj[taskid]\n\n    def _submit_job(self, node, updatehash=False):\n        self._taskid += 1\n\n        # Don't allow streaming outputs\n        if getattr(node.interface, 'terminal_output', '') == 'stream':\n            node.interface.terminal_output = 'allatonce'\n\n        self._task_obj[self._taskid] = self.pool.apply_async(\n            run_node, (node, updatehash, self._taskid),\n            callback=self._async_callback)\n\n        logger.debug('MultiProc submitted task %s (taskid=%d).',\n                     node.fullname, self._taskid)\n        return self._taskid\n\n    def _prerun_check(self, graph):\n        \"\"\"Check if any node exeeds the available resources\"\"\"\n        tasks_mem_gb = []\n        tasks_num_th = []\n        for node in graph.nodes():\n            tasks_mem_gb.append(node.mem_gb)\n            tasks_num_th.append(node.n_procs)\n\n        if np.any(np.array(tasks_mem_gb) > self.memory_gb):\n            logger.warning(\n                'Some nodes exceed the total amount of memory available '\n                '(%0.2fGB).', self.memory_gb)\n            if self.raise_insufficient:\n                raise RuntimeError('Insufficient resources available for job')\n\n        if np.any(np.array(tasks_num_th) > self.processors):\n            logger.warning(\n                'Some nodes demand for more threads than available (%d).',\n                self.processors)\n            if self.raise_insufficient:\n                raise RuntimeError('Insufficient resources available for job')\n\n    def _postrun_check(self):\n        self.pool.close()\n\n    def _check_resources(self, running_tasks):\n        \"\"\"\n        Make sure there are resources available\n        \"\"\"\n        free_memory_gb = self.memory_gb\n        free_processors = self.processors\n        for _, jobid in running_tasks:\n            free_memory_gb -= min(self.procs[jobid].mem_gb, free_memory_gb)\n            free_processors -= min(self.procs[jobid].n_procs, free_processors)\n\n        return free_memory_gb, free_processors\n\n    def _send_procs_to_workers(self, updatehash=False, graph=None):\n        \"\"\"\n        Sends jobs to workers when system resources are available.\n        \"\"\"\n\n        # Check to see if a job is available (jobs without dependencies not run)\n        # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722\n        jobids = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]\n\n        # Check available system resources by summing all threads and memory used\n        free_memory_gb, free_processors = self._check_resources(self.pending_tasks)\n\n        stats = (len(self.pending_tasks), len(jobids), free_memory_gb,\n                 self.memory_gb, free_processors, self.processors)\n        if self._stats != stats:\n            logger.info('Currently running %d tasks, and %d jobs ready. Free '\n                        'memory (GB): %0.2f/%0.2f, Free processors: %d/%d',\n                        *stats)\n            self._stats = stats\n\n        if free_memory_gb < 0.01 or free_processors == 0:\n            logger.debug('No resources available')\n            return\n\n        if len(jobids) + len(self.pending_tasks) == 0:\n            logger.debug('No tasks are being run, and no jobs can '\n                         'be submitted to the queue. Potential deadlock')\n            return\n\n        jobids = self._sort_jobs(jobids, scheduler=self.plugin_args.get('scheduler'))\n\n        # Submit jobs\n        for jobid in jobids:\n            # First expand mapnodes\n            if isinstance(self.procs[jobid], MapNode):\n                try:\n                    num_subnodes = self.procs[jobid].num_subnodes()\n                except Exception:\n                    traceback = format_exception(*sys.exc_info())\n                    self._clean_queue(\n                        jobid, graph,\n                        result={'result': None, 'traceback': traceback}\n                    )\n                    self.proc_pending[jobid] = False\n                    continue\n                if num_subnodes > 1:\n                    submit = self._submit_mapnode(jobid)\n                    if not submit:\n                        continue\n\n            # Check requirements of this job\n            next_job_gb = min(self.procs[jobid].mem_gb, self.memory_gb)\n            next_job_th = min(self.procs[jobid].n_procs, self.processors)\n\n            # If node does not fit, skip at this moment\n            if next_job_th > free_processors or next_job_gb > free_memory_gb:\n                logger.debug('Cannot allocate job %d (%0.2fGB, %d threads).',\n                             jobid, next_job_gb, next_job_th)\n                continue\n\n            free_memory_gb -= next_job_gb\n            free_processors -= next_job_th\n            logger.debug('Allocating %s ID=%d (%0.2fGB, %d threads). Free: %0.2fGB, %d threads.',\n                         self.procs[jobid].fullname, jobid, next_job_gb, next_job_th,\n                         free_memory_gb, free_processors)\n\n            # change job status in appropriate queues\n            self.proc_done[jobid] = True\n            self.proc_pending[jobid] = True\n\n            # If cached just retrieve it, don't run\n            if self._local_hash_check(jobid, graph):\n                continue\n\n            if self.procs[jobid].run_without_submitting:\n                logger.debug('Running node %s on master thread',\n                             self.procs[jobid])\n                try:\n                    self.procs[jobid].run(updatehash=updatehash)\n                except Exception:\n                    traceback = format_exception(*sys.exc_info())\n                    self._clean_queue(\n                        jobid, graph,\n                        result={'result': None, 'traceback': traceback}\n                    )\n\n                # Release resources\n                self._task_finished_cb(jobid)\n                self._remove_node_dirs()\n                free_memory_gb += next_job_gb\n                free_processors += next_job_th\n                # Display stats next loop\n                self._stats = None\n                continue\n\n            # Task should be submitted to workers\n            # Send job to task manager and add to pending tasks\n            if self._status_callback:\n                self._status_callback(self.procs[jobid], 'start')\n            tid = self._submit_job(deepcopy(self.procs[jobid]),\n                                   updatehash=updatehash)\n            if tid is None:\n                self.proc_done[jobid] = False\n                self.proc_pending[jobid] = False\n            else:\n                self.pending_tasks.insert(0, (tid, jobid))\n            # Display stats next loop\n            self._stats = None\n\n    def _sort_jobs(self, jobids, scheduler='tsort'):\n        if scheduler == 'mem_thread':\n            return sorted(jobids, key=lambda item: (\n                self.procs[item].mem_gb, self.procs[item].n_procs))\n        return jobids\n", "path": "nipype/pipeline/plugins/multiproc.py"}]}