Wrote my own get memory function - seems to work much better

Author: pintohutch

nipype/interfaces/base.py

@@ -1249,9 +1249,57 @@ def _get_num_threads(proc, log_flg=False):
     # Return the number of threads found
     return num_threads
 
+def _get_num_ram_mb(pid, pyfunc=False):
+    """Function to get the RAM usage of a process and its children
+
+    Parameters
+    ----------
+    pid : integer
+        the PID of the process to get RAM usage of
+    pyfunc : boolean (optional); default=False
+        a flag to indicate if the process is a python function;
+        when Pythons are multithreaded via multiprocess or threading,
+        children functions include their own memory + parents. if this
+        is set, the parent memory will removed from children memories
+
+    Reference: http://ftp.dev411.com/t/python/python-list/095thexx8g/multiprocessing-forking-memory-usage
+
+    Returns
+    -------
+    mem_mb : float
+        the memory RAM in MB utilized by the process PID
+    """
+
+    # Import packages
+    import psutil
+
+    # Init variables
+    _MB = 1024.0**2
+
+    # Try block to protect against any dying processes in the interim
+    try:
+        # Init parent
+        parent = psutil.Process(pid)
+        # Get memory of parent
+        parent_mem = parent.memory_info().rss
+        mem_mb = parent_mem/_MB
+
+        # Iterate through child processes
+        for child in parent.children(recursive=True):
+            child_mem = child.memory_info().rss
+            if pyfunc:
+                child_mem -= parent_mem
+            mem_mb += child_mem/_MB
+
+    # Catch if process dies, return gracefully
+    except psutil.NoSuchProcess:
+        pass
+
+    # Return memory
+    return mem_mb
 
 # Get max resources used for process
-def get_max_resources_used(pid, mem_mb, num_threads, log_flg=False):
+def get_max_resources_used(pid, mem_mb, num_threads, pyfunc=False, log_flg=False):
     """Function to get the RAM and threads usage of a process
 
     Paramters
@@ -1276,7 +1324,8 @@ def get_max_resources_used(pid, mem_mb, num_threads, log_flg=False):
     import psutil
 
     try:
-        mem_mb = max(mem_mb, _get_memory(pid, include_children=True, log_flg=log_flg))
+        #mem_mb = max(mem_mb, _get_memory(pid, include_children=True, log_flg=log_flg))
+        mem_mb = max(mem_mb, _get_num_ram_mb(pid, pyfunc=pyfunc))
         num_threads = max(num_threads, _get_num_threads(psutil.Process(pid), log_flg=log_flg))
     except Exception as exc:
         iflogger.info('Could not get resources used by process. Error: %s'\

nipype/interfaces/tests/test_runtime_profiler.py

@@ -101,7 +101,7 @@ def _use_gb_ram(num_gb):
     print 'Using %.3f GB of memory over %d processors...' % (num_gb, num_procs)
     for idx, proc in enumerate(proc_list):
         proc.start()
-        logger.debug('Starting PID: %d' % proc.pid)
+        #logger.debug('Starting PID: %d' % proc.pid)
 
     for proc in proc_list:
         proc.join()
@@ -137,12 +137,72 @@ def setUp(self):
 
         # Init parameters
         # Input RAM GB to occupy
-        self.num_gb= .75
+        self.num_gb= 4
         # Input number of processors
         self.num_procs = 1
         # Acceptable percent error for memory profiled against input
         self.mem_err_percent = 5
 
+    # ! Only used for benchmarking the profiler over a range of
+    # ! processors and RAM usage
+    # ! Requires a LOT of RAM and PROCS to be tested
+    def _collect_range_runtime_stats(self):
+        '''
+        Function to collect a range of runtime stats
+        '''
+
+        # Import packages
+        import json
+        import numpy as np
+        import pandas as pd
+
+        # Init variables
+        num_procs_range = 8
+        ram_gb_range = 10.0
+        ram_gb_step = 0.25
+        dict_list = []
+
+        # Iterate through all combos
+        for num_procs in np.arange(1, num_procs_range+1, 1):
+            for num_gb in np.arange(0.25, ram_gb_range+ram_gb_step, ram_gb_step):
+                # Cmd-level
+                cmd_fin_str = self._run_cmdline_workflow(num_gb, num_procs)
+                cmd_node_stats = json.loads(cmd_fin_str)
+                cmd_runtime_procs = int(cmd_node_stats['runtime_threads'])
+                cmd_runtime_gb = float(cmd_node_stats['runtime_memory_gb'])
+
+                # Func-level
+                func_fin_str = self._run_function_workflow(num_gb, num_procs)
+                func_node_stats = json.loads(func_fin_str)
+                func_runtime_procs = int(func_node_stats['runtime_threads'])
+                func_runtime_gb = float(func_node_stats['runtime_memory_gb'])
+
+                # Calc errors
+                cmd_procs_err = cmd_runtime_procs - num_procs
+                cmd_gb_err = cmd_runtime_gb - num_gb
+                func_procs_err = func_runtime_procs - num_procs
+                func_gb_err = func_runtime_gb - num_gb
+
+                # Node dictionary
+                results_dict = {'input_procs' : num_procs,
+                                'input_gb' : num_gb,
+                                'cmd_runtime_procs' : cmd_runtime_procs,
+                                'cmd_runtime_gb' : cmd_runtime_gb,
+                                'func_runtime_procs' : func_runtime_procs,
+                                'func_runtime_gb' : func_runtime_gb,
+                                'cmd_procs_err' : cmd_procs_err,
+                                'cmd_gb_err' : cmd_gb_err,
+                                'func_procs_err' : func_procs_err,
+                                'func_gb_err' : func_gb_err}
+                # Append to list
+                dict_list.append(results_dict)
+
+        # Create dataframe
+        runtime_results_df = pd.DataFrame(dict_list)
+
+        # Return dataframe
+        return runtime_results_df
+
     # Test node
     def _run_cmdline_workflow(self, num_gb, num_procs):
         '''
@@ -371,69 +431,6 @@ def test_function_profiling(self):
         self.assertLessEqual(runtime_gb_err, allowed_gb_err, msg=mem_err)
         self.assertEqual(num_procs, runtime_procs, msg=procs_err)
 
-    # Collect stats for range of num_threads and memory amount
-    def _collect_range_runtime_stats(self):
-        '''
-        Function to collect a range of runtime stats
-        '''
-
-        # Import packages
-        import json
-        import numpy as np
-        import pandas as pd
-
-        # Init variables
-        num_procs_range = 8
-        ram_gb_range = 10.0
-        ram_gb_step = 0.25
-        dict_list = []
-
-        # Iterate through all combos
-        for num_procs in np.arange(1, num_procs_range+1, 1):
-            for num_gb in np.arange(0.25, ram_gb_range+ram_gb_step, ram_gb_step):
-                # Cmd-level
-                cmd_fin_str = self._run_cmdline_workflow(num_gb, num_procs)
-                cmd_node_stats = json.loads(cmd_fin_str)
-                cmd_runtime_procs = int(cmd_node_stats['runtime_threads'])
-                cmd_runtime_gb = float(cmd_node_stats['runtime_memory_gb'])
-
-                # Func-level
-                func_fin_str = self._run_function_workflow(num_gb, num_procs)
-                func_node_stats = json.loads(func_fin_str)
-                func_runtime_procs = int(func_node_stats['runtime_threads'])
-                func_runtime_gb = float(func_node_stats['runtime_memory_gb'])
-
-                # Calc errors
-                cmd_procs_err = cmd_runtime_procs - num_procs
-                cmd_gb_err = cmd_runtime_gb - num_gb
-                func_procs_err = func_runtime_procs - num_procs
-                func_gb_err = func_runtime_gb - num_gb
-
-                # Node dictionary
-                results_dict = {'input_procs' : num_procs,
-                                'input_gb' : num_gb,
-                                'cmd_runtime_procs' : cmd_runtime_procs,
-                                'cmd_runtime_gb' : cmd_runtime_gb,
-                                'func_runtime_procs' : func_runtime_procs,
-                                'func_runtime_gb' : func_runtime_gb,
-                                'cmd_procs_err' : cmd_procs_err,
-                                'cmd_gb_err' : cmd_gb_err,
-                                'func_procs_err' : func_procs_err,
-                                'func_gb_err' : func_gb_err}
-                # Append to list
-                dict_list.append(results_dict)
-
-        # Create dataframe
-        runtime_results_df = pd.DataFrame(dict_list)
-
-        # Return dataframe
-        return runtime_results_df
-
-    def test_write_df_to_csv(self):
-        df = self._collect_range_runtime_stats()
-        df.to_csv('/home/dclark/runtime_results.csv')
-        #self.assertEqual(1, 1)
-
 
 # Command-line run-able unittest module
 if __name__ == '__main__':

nipype/interfaces/utility.py

@@ -496,7 +496,7 @@ def _function_handle_wrapper(queue, **kwargs):
             proc.start()
             while proc.is_alive():
                 mem_mb, num_threads = \
-                    get_max_resources_used(proc.pid, mem_mb, num_threads, log_flg=log_flg)
+                    get_max_resources_used(proc.pid, mem_mb, num_threads, pyfunc=True, log_flg=log_flg)
 
             # Get result from process queue
             out = queue.get()