Merge remote-tracking branch 'upstream/master' into ref/Node-cleanup

Author: oesteban

nipype/interfaces/base/core.py

@@ -530,8 +530,9 @@ def run(self, **inputs):
 
                     runtime.prof_dict = {
                         'time': vals[:, 0].tolist(),
-                        'mem_gb': (vals[:, 1] / 1024).tolist(),
-                        'cpus': vals[:, 2].tolist(),
+                        'cpus': vals[:, 1].tolist(),
+                        'rss_GiB': (vals[:, 2] / 1024).tolist(),
+                        'vms_GiB': (vals[:, 3] / 1024).tolist(),
                     }
 
         return results

nipype/pipeline/engine/utils.py

@@ -1538,7 +1538,8 @@ def write_workflow_resources(graph, filename=None, append=None):
         'time': [],
         'name': [],
         'interface': [],
-        'mem_gb': [],
+        'rss_GiB': [],
+        'vms_GiB': [],
         'cpus': [],
         'mapnode': [],
         'params': [],
@@ -1579,7 +1580,7 @@ def write_workflow_resources(graph, filename=None, append=None):
                     '(mapflow %d/%d).', nodename, subidx + 1, len(rt_list))
                 continue
 
-            for key in ['time', 'mem_gb', 'cpus']:
+            for key in ['time', 'cpus', 'rss_GiB', 'vms_GiB']:
                 big_dict[key] += runtime.prof_dict[key]
 
             big_dict['interface'] += [classname] * nsamples

nipype/pipeline/plugins/base.py

@@ -128,9 +128,11 @@ def run(self, graph, config, updatehash=False):
         old_progress_stats = None
         old_presub_stats = None
         while not np.all(self.proc_done) or np.any(self.proc_pending):
-            # Check to see if a job is available (jobs without dependencies not run)
-            # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722
-            jobs_ready = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]
+            loop_start = time()
+            # Check if a job is available (jobs with all dependencies run)
+            # https://github.com/nipy/nipype/pull/2200#discussion_r141605722
+            jobs_ready = np.nonzero(~self.proc_done &
+                                    (self.depidx.sum(0) == 0))[1]
 
             progress_stats = (len(self.proc_done),
                               np.sum(self.proc_done ^ self.proc_pending),
@@ -164,7 +166,8 @@ def run(self, graph, config, updatehash=False):
                             self._remove_node_dirs()
                         self._clear_task(taskid)
                     else:
-                        assert self.proc_done[jobid] and self.proc_pending[jobid]
+                        assert self.proc_done[jobid] and \
+                            self.proc_pending[jobid]
                         toappend.insert(0, (taskid, jobid))
 
             if toappend:
@@ -183,7 +186,8 @@ def run(self, graph, config, updatehash=False):
             elif display_stats:
                 logger.debug('Not submitting (max jobs reached)')
 
-            sleep(poll_sleep_secs)
+            sleep_til = loop_start + poll_sleep_secs
+            sleep(max(0, sleep_til - time()))
 
         self._remove_node_dirs()
         report_nodes_not_run(notrun)
@@ -271,8 +275,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             if (num_jobs >= self.max_jobs) or (slots == 0):
                 break
 
-            # Check to see if a job is available (jobs without dependencies not run)
-            # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722
+            # Check if a job is available (jobs with all dependencies run)
+            # https://github.com/nipy/nipype/pull/2200#discussion_r141605722
             jobids = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]
 
             if len(jobids) > 0:
@@ -325,7 +329,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                 break
 
     def _local_hash_check(self, jobid, graph):
-        if not str2bool(self.procs[jobid].config['execution']['local_hash_check']):
+        if not str2bool(self.procs[jobid].config['execution'][
+                'local_hash_check']):
             return False
 
         logger.debug('Checking hash (%d) locally', jobid)
@@ -397,8 +402,8 @@ def _remove_node_dirs(self):
         """Removes directories whose outputs have already been used up
         """
         if str2bool(self._config['execution']['remove_node_directories']):
-            for idx in np.nonzero(
-                                 (self.refidx.sum(axis=1) == 0).__array__())[0]:
+            indices = np.nonzero((self.refidx.sum(axis=1) == 0).__array__())[0]
+            for idx in indices:
                 if idx in self.mapnodesubids:
                     continue
                 if self.proc_done[idx] and (not self.proc_pending[idx]):
@@ -513,7 +518,8 @@ class GraphPluginBase(PluginBase):
 
     def __init__(self, plugin_args=None):
         if plugin_args and plugin_args.get('status_callback'):
-            logger.warning('status_callback not supported for Graph submission plugins')
+            logger.warning('status_callback not supported for Graph submission'
+                           ' plugins')
         super(GraphPluginBase, self).__init__(plugin_args=plugin_args)
 
     def run(self, graph, config, updatehash=False):

nipype/utils/profiler.py

@@ -67,11 +67,14 @@ def stop(self):
 
     def _sample(self, cpu_interval=None):
         cpu = 0.0
-        mem = 0.0
+        rss = 0.0
+        vms = 0.0
         try:
             with self._process.oneshot():
                 cpu += self._process.cpu_percent(interval=cpu_interval)
-                mem += self._process.memory_info().rss
+                mem_info = self._process.memory_info()
+                rss += mem_info.rss
+                vms += mem_info.vms
         except psutil.NoSuchProcess:
             pass
 
@@ -85,19 +88,24 @@ def _sample(self, cpu_interval=None):
             try:
                 with child.oneshot():
                     cpu += child.cpu_percent()
-                    mem += child.memory_info().rss
+                    mem_info = child.memory_info()
+                    rss += mem_info.rss
+                    vms += mem_info.vms
             except psutil.NoSuchProcess:
                 pass
 
-        print('%f,%f,%f' % (time(), (mem / _MB), cpu),
+        print('%f,%f,%f,%f' % (time(), cpu, rss / _MB, vms / _MB),
               file=self._logfile)
         self._logfile.flush()
 
     def run(self):
         """Core monitoring function, called by start()"""
+        start_time = time()
+        wait_til = start_time
         while not self._event.is_set():
             self._sample()
-            self._event.wait(self._freq)
+            wait_til += self._freq
+            self._event.wait(max(0, wait_til - time()))
 
 
 # Log node stats function