Added global watcher

Author: pintohutch

nipype/interfaces/base.py

@@ -1222,19 +1222,32 @@ def _get_num_threads(proc):
 
     # Import packages
     import psutil
+    import logging
 
     # Init variables
-    num_threads = proc.num_threads()
-    alive_procs = 0
+    cb_log = logging.getLogger('callback')
+    cb_log.propogate = False
+    cb_log.debug('proc pid: %d, parent pid: %d, name: %s, exe: %s, cmdline: %s, status: %s, num_threads: %d' \
+                 % (proc.pid, proc.ppid(), proc.name(), proc.exe(), proc.cmdline(), proc.status(), proc.num_threads()))
+    if proc.status() == psutil.STATUS_RUNNING:
+        num_threads = proc.num_threads()
+    else:
+        num_threads = 0
+    child_threads = 0
     # Iterate through child processes and get number of their threads
     try:
-        #num_children = len(proc.children())
         for child in proc.children(recursive=True):
             if child.status() == psutil.STATUS_RUNNING:
-                alive_procs += 1
-                num_threads += max(alive_procs, child.num_threads()) #child.num_threads()
+                # If leaf child process
+                if len(child.children()) == 0:
+                    child_threads += child.num_threads()
+                    #num_threads = max(num_threads, child.num_threads()) #child.num_threads()
+                cb_log.debug('child pid: %d, parent pid: %d, name: %s, exe: %s, cmdline: %s, status: %s, num_threads: %d' \
+                             % (proc.pid, proc.ppid(), proc.name(), proc.exe(), proc.cmdline(), proc.status(), child.num_threads()))
+
             #num_threads = max(num_threads, num_children,
             #                  child.num_threads(), len(child.children()))
+        num_threads = max(child_threads, num_threads)
     except psutil.NoSuchProcess:
         pass
 
@@ -1380,8 +1393,8 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     outfile = os.path.join(runtime.cwd, 'stdout.nipype')
 
     # Init variables for memory profiling
-    mem_mb = -1
-    num_threads = -1
+    mem_mb = 0
+    num_threads = 0
     interval = .5
 
     if output == 'stream':
@@ -1405,7 +1418,7 @@ def _process(drain=0):
                     get_max_resources_used(proc.pid, mem_mb, num_threads)
             proc.poll()
             _process()
-            #time.sleep(interval)
+            time.sleep(interval)
         _process(drain=1)
 
         # collect results, merge and return
@@ -1424,7 +1437,7 @@ def _process(drain=0):
                 mem_mb, num_threads = \
                     get_max_resources_used(proc.pid, mem_mb, num_threads)
                 proc.poll()
-                #time.sleep(interval)
+                time.sleep(interval)
         stdout, stderr = proc.communicate()
         if stdout and isinstance(stdout, bytes):
             try:
@@ -1446,7 +1459,7 @@ def _process(drain=0):
                 mem_mb, num_threads = \
                     get_max_resources_used(proc.pid, mem_mb, num_threads)
                 proc.poll()
-                #time.sleep(interval)
+                time.sleep(interval)
         ret_code = proc.wait()
         stderr.flush()
         stdout.flush()
@@ -1459,7 +1472,7 @@ def _process(drain=0):
                 mem_mb, num_threads = \
                     get_max_resources_used(proc.pid, mem_mb, num_threads)
                 proc.poll()
-                #time.sleep(interval)
+                time.sleep(interval)
         proc.communicate()
         result['stdout'] = []
         result['stderr'] = []

nipype/interfaces/utility.py

@@ -484,8 +484,8 @@ def _function_handle_wrapper(queue, **kwargs):
                                            args=(queue,), kwargs=args)
 
             # Init memory and threads before profiling
-            mem_mb = -1
-            num_threads = -1
+            mem_mb = 0
+            num_threads = 0
 
             # Start process and profile while it's alive
             proc.start()

nipype/pipeline/plugins/base.py

@@ -223,6 +223,22 @@ def __init__(self, plugin_args=None):
     def run(self, graph, config, updatehash=False):
         """Executes a pre-defined pipeline using distributed approaches
         """
+        # Global watcher inits
+        from nipype.interfaces.base import get_max_resources_used
+        gpid = os.getpid()
+        num_threads = 0
+        memory_mb = 0
+        # Init logger
+        import logging as lg
+        gw_log = lg.getLogger('global_watcher')
+        gw_log.setLevel(lg.INFO)
+        formatter = lg.Formatter('%(asctime)s : %(message)s')
+    
+        # Write logs to file
+        file_handler = lg.FileHandler('/home/dclark/work-dir/gw.log')
+        file_handler.setFormatter(formatter)
+        gw_log.addHandler(file_handler)
+
         logger.info("Running in parallel.")
         self._config = config
         # Generate appropriate structures for worker-manager model
@@ -235,6 +251,11 @@ def run(self, graph, config, updatehash=False):
         notrun = []
         while np.any(self.proc_done == False) | \
                 np.any(self.proc_pending == True):
+
+            mem_mb, num_thr = get_max_resources_used(gpid, memory_mb, num_threads)
+            memory_mb = max(mem_mb, memory_mb)
+            num_threads = max(num_thr, num_threads)
+            gw_log.info('Memory GB usage: %.4f, Threads usage: %d' % (memory_mb/1024.0, num_threads))
             toappend = []
             # trigger callbacks for any pending results
             while self.pending_tasks:
@@ -266,6 +287,7 @@ def run(self, graph, config, updatehash=False):
             else:
                 logger.debug('Not submitting')
             self._wait()
+            
         self._remove_node_dirs()
         report_nodes_not_run(notrun)
 

nipype/utils/draw_gantt_chart.py

@@ -183,6 +183,8 @@ def calculate_resource_timeseries(events, resource):
 
     # Iterate through the events
     for tdelta, event in sorted(events.items()):
+        if tdelta > 70.7:
+            print 'hi'
         if event['event'] == "start":
             if resource in event and event[resource] != 'Unkown':
                 all_res += float(event[resource])
@@ -520,8 +522,8 @@ def generate_gantt_chart(logfile, cores, minute_scale=10,
     close_header = '''
     </div>
     <div style="display:inline-block;margin-left:60px;vertical-align: top;">
-        <p><span><div class="label" style="background-color:#03969D;"></div> Estimated Resource</span></p>
-        <p><span><div class="label" style="background-color:#90BBD7;"></div> Actual Resource</span></p>
+        <p><span><div class="label" style="background-color:#90BBD7;"></div> Estimated Resource</span></p>
+        <p><span><div class="label" style="background-color:#03969D;"></div> Actual Resource</span></p>
         <p><span><div class="label" style="background-color:#f00;"></div> Failed Node</span></p>
     </div>
     '''