Merge pull request #2285 from oesteban/fix/resource-monitor-revisions

[REF,FIX] Revision of the resource monitor

Author: satra

CHANGES

@@ -7,6 +7,7 @@ Upcoming release
 
 ###### [Full changelog](https://github.com/nipy/nipype/milestone/13)
 
+* FIX+MAINT: Revision of the resource monitor (https://github.com/nipy/nipype/pull/2285)
 * FIX: MultiProc mishandling crashes (https://github.com/nipy/nipype/pull/2301)
 * MAINT: Revise use of `subprocess.Popen` (https://github.com/nipy/nipype/pull/2289)
 * ENH: Memorize version checks (https://github.com/nipy/nipype/pull/2274, https://github.com/nipy/nipype/pull/2295)

doc/users/config_file.rst

@@ -153,14 +153,29 @@ Execution
     crashfiles allow portability across machines and shorter load time.
     (possible values: ``pklz`` and ``txt``; default value: ``pklz``)
 
-*resource_monitor*
+
+Resource Monitor
+~~~~~~~~~~~~~~~~
+
+*enabled*
     Enables monitoring the resources occupation (possible values: ``true`` and
-    ``false``; default value: ``false``)
+    ``false``; default value: ``false``). All the following options will be
+    dismissed if the resource monitor is not enabled.
 
-*resource_monitor_frequency*
+*sample_frequency*
     Sampling period (in seconds) between measurements of resources (memory, cpus)
-    being used by an interface. Requires ``resource_monitor`` to be ``true``.
-    (default value: ``1``)
+    being used by an interface (default value: ``1``)
+
+*summary_file*
+    Indicates where the summary file collecting all profiling information from the
+    resource monitor should be stored after execution of a workflow.
+    The ``summary_file`` does not apply to interfaces run independently.
+    (unset by default, in which case the summary file will be written out to 
+    ``<base_dir>/resource_monitor.json`` of the top-level workflow).
+
+*summary_append*
+    Append to an existing summary file (only applies to workflows).
+    (default value: ``true``, possible values: ``true`` or ``false``).
 
 Example
 ~~~~~~~
@@ -175,6 +190,10 @@ Example
     hash_method = timestamp
     display_variable = :1
 
+    [monitoring]
+    enabled = false
+
+
 Workflow.config property has a form of a nested dictionary reflecting the
 structure of the .cfg file.
 

docker/files/run_examples.sh

@@ -20,8 +20,9 @@ echo '[execution]' >> ${HOME}/.nipype/nipype.cfg
 echo 'crashfile_format = txt' >> ${HOME}/.nipype/nipype.cfg
 
 if [[ "${NIPYPE_RESOURCE_MONITOR:-0}" == "1" ]]; then
-    echo 'resource_monitor = true' >> ${HOME}/.nipype/nipype.cfg
-    echo 'resource_monitor_frequency = 3' >> ${HOME}/.nipype/nipype.cfg
+    echo '[monitoring]' >> ${HOME}/.nipype/nipype.cfg
+    echo 'enabled = true' >> ${HOME}/.nipype/nipype.cfg
+    echo 'sample_frequency = 3' >> ${HOME}/.nipype/nipype.cfg
 fi
 
 # Set up coverage

nipype/interfaces/tests/test_resource_monitor.py

@@ -54,7 +54,7 @@ def test_cmdline_profiling(tmpdir, mem_gb, n_procs):
     of a CommandLine-derived interface
     """
     from nipype import config
-    config.set('execution', 'resource_monitor_frequency', '0.2')  # Force sampling fast
+    config.set('monitoring', 'sample_frequency', '0.2')  # Force sampling fast
 
     tmpdir.chdir()
     iface = UseResources(mem_gb=mem_gb, n_procs=n_procs)
@@ -72,7 +72,7 @@ def test_function_profiling(tmpdir, mem_gb, n_procs):
     of a Function interface
     """
     from nipype import config
-    config.set('execution', 'resource_monitor_frequency', '0.2')  # Force sampling fast
+    config.set('monitoring', 'sample_frequency', '0.2')  # Force sampling fast
 
     tmpdir.chdir()
     iface = niu.Function(function=_use_resources)

nipype/pipeline/engine/utils.py

@@ -1298,15 +1298,24 @@ def write_workflow_prov(graph, filename=None, format='all'):
     return ps.g
 
 
-def write_workflow_resources(graph, filename=None):
+def write_workflow_resources(graph, filename=None, append=None):
     """
     Generate a JSON file with profiling traces that can be loaded
     in a pandas DataFrame or processed with JavaScript like D3.js
     """
     import simplejson as json
+
+    # Overwrite filename if nipype config is set
+    filename = config.get('monitoring', 'summary_file', filename)
+
+    # If filename still does not make sense, store in $PWD
     if not filename:
         filename = os.path.join(os.getcwd(), 'resource_monitor.json')
 
+    if append is None:
+        append = str2bool(config.get(
+            'monitoring', 'summary_append', 'true'))
+
     big_dict = {
         'time': [],
         'name': [],
@@ -1317,14 +1326,21 @@ def write_workflow_resources(graph, filename=None):
         'params': [],
     }
 
+    # If file exists, just append new profile information
+    # If we append different runs, then we will see different
+    # "bursts" of timestamps corresponding to those executions.
+    if append and os.path.isfile(filename):
+        with open(filename, 'r' if PY3 else 'rb') as rsf:
+            big_dict = json.load(rsf)
+
     for idx, node in enumerate(graph.nodes()):
         nodename = node.fullname
         classname = node._interface.__class__.__name__
 
         params = ''
         if node.parameterization:
             params = '_'.join(['{}'.format(p)
-                              for p in node.parameterization])
+                               for p in node.parameterization])
 
         try:
             rt_list = node.result.runtime
@@ -1337,7 +1353,13 @@ def write_workflow_resources(graph, filename=None):
             rt_list = [rt_list]
 
         for subidx, runtime in enumerate(rt_list):
-            nsamples = len(runtime.prof_dict['time'])
+            try:
+                nsamples = len(runtime.prof_dict['time'])
+            except AttributeError:
+                logger.warning(
+                    'Could not retrieve profiling information for node "%s" '
+                    '(mapflow %d/%d).', nodename, subidx + 1, len(rt_list))
+                continue
 
             for key in ['time', 'mem_gb', 'cpus']:
                 big_dict[key] += runtime.prof_dict[key]

nipype/pipeline/engine/workflows.py

@@ -598,7 +598,11 @@ def run(self, plugin=None, plugin_args=None, updatehash=False):
             write_workflow_prov(execgraph, prov_base, format='all')
 
         if config.resource_monitor:
-            write_workflow_resources(execgraph)
+            base_dir = self.base_dir or os.getcwd()
+            write_workflow_resources(
+                execgraph,
+                filename=op.join(base_dir, self.name, 'resource_monitor.json')
+            )
         return execgraph
 
     # PRIVATE API AND FUNCTIONS

nipype/utils/config.py

@@ -31,8 +31,8 @@
 
 
 CONFIG_DEPRECATIONS = {
-    'profile_runtime': ('resource_monitor', '1.0'),
-    'filemanip_level': ('utils_level', '1.0'),
+    'profile_runtime': ('monitoring.enabled', '1.0'),
+    'filemanip_level': ('logging.utils_level', '1.0'),
 }
 
 NUMPY_MMAP = LooseVersion(np.__version__) >= LooseVersion('1.12.0')
@@ -71,8 +71,11 @@
 parameterize_dirs = true
 poll_sleep_duration = 2
 xvfb_max_wait = 10
-resource_monitor = false
-resource_monitor_frequency = 1
+
+[monitoring]
+enabled = false
+sample_frequency = 1
+summary_append = true
 
 [check]
 interval = 1209600
@@ -105,12 +108,12 @@ def __init__(self, *args, **kwargs):
             self._config.read([config_file, 'nipype.cfg'])
 
         for option in CONFIG_DEPRECATIONS:
-            for section in ['execution', 'logging']:
+            for section in ['execution', 'logging', 'monitoring']:
                 if self.has_option(section, option):
-                    new_option = CONFIG_DEPRECATIONS[option][0]
-                    if not self.has_option(section, new_option):
+                    new_section, new_option = CONFIG_DEPRECATIONS[option][0].split('.')
+                    if not self.has_option(new_section, new_option):
                         # Warn implicit in get
-                        self.set(section, new_option, self.get(section, option))
+                        self.set(new_section, new_option, self.get(section, option))
 
     def set_default_config(self):
         self._config.readfp(StringIO(default_cfg))
@@ -138,7 +141,7 @@ def get(self, section, option, default=None):
                    '"%s" instead.') % (option, CONFIG_DEPRECATIONS[option][1],
                                        CONFIG_DEPRECATIONS[option][0])
             warn(msg)
-            option = CONFIG_DEPRECATIONS[option][0]
+            section, option = CONFIG_DEPRECATIONS[option][0].split('.')
 
         if self._config.has_option(section, option):
             return self._config.get(section, option)
@@ -154,7 +157,7 @@ def set(self, section, option, value):
                    '"%s" instead.') % (option, CONFIG_DEPRECATIONS[option][1],
                                        CONFIG_DEPRECATIONS[option][0])
             warn(msg)
-            option = CONFIG_DEPRECATIONS[option][0]
+            section, option = CONFIG_DEPRECATIONS[option][0].split('.')
 
         return self._config.set(section, option, value)
 
@@ -222,8 +225,8 @@ def resource_monitor(self):
             return self._resource_monitor
 
         # Cache config from nipype config
-        self.resource_monitor = self._config.get(
-            'execution', 'resource_monitor') or False
+        self.resource_monitor = str2bool(self._config.get(
+            'monitoring', 'enabled')) or False
         return self._resource_monitor
 
     @resource_monitor.setter
@@ -248,7 +251,7 @@ def resource_monitor(self, value):
                     if not self._resource_monitor:
                         warn('Could not enable the resource monitor: psutil>=5.0'
                              ' could not be imported.')
-                    self._config.set('execution', 'resource_monitor',
+                    self._config.set('monitoring', 'enabled',
                                      ('%s' % self._resource_monitor).lower())
 
     def enable_resource_monitor(self):

nipype/utils/profiler.py

@@ -1,8 +1,6 @@
 # -*- coding: utf-8 -*-
-# @Author: oesteban
-# @Date:   2017-09-21 15:50:37
-# @Last Modified by:   oesteban
-# @Last Modified time: 2017-10-20 09:12:36
+# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
+# vi: set ft=python sts=4 ts=4 sw=4 et:
 """
 Utilities to keep track of performance
 """
@@ -202,8 +200,8 @@ def get_max_resources_used(pid, mem_mb, num_threads, pyfunc=False):
     """
 
     if not resource_monitor:
-        raise RuntimeError('Attempted to measure resources with '
-                           '"resource_monitor" set off.')
+        raise RuntimeError('Attempted to measure resources with option '
+                           '"monitoring.enabled" set off.')
 
     try:
         mem_mb = max(mem_mb, _get_ram_mb(pid, pyfunc=pyfunc))
@@ -320,7 +318,8 @@ def _use_cpu(x):
     ctr = 0
     while ctr < 1e7:
         ctr += 1
-        x*x
+        x * x
+
 
 # Spin multiple threads
 def _use_resources(n_procs, mem_gb):