add memory and thread to gantt chart, callback handles errors

nipype/pipeline/plugins/callback_log.py

@@ -2,6 +2,7 @@
 import logging
 
 def log_nodes_cb(node, status):
+    print 'status', status
     logger = logging.getLogger('callback')
     if status == 'start':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' +\
@@ -10,10 +11,19 @@ def log_nodes_cb(node, status):
         + str(node._interface.num_threads) +  '}'
 
         logger.debug(message)
-    else:
+
+    elif status == 'end':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
         node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
         '"' + ',"memory":' + str(node._interface.memory) + ',"num_threads":' \
         + str(node._interface.num_threads) +  '}'
 
+        logger.debug(message)
+
+    else:
+        message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
+        node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
+        '"' + ',"memory":' + str(node._interface.memory) + ',"num_threads":' \
+        + str(node._interface.num_threads) + ',"error":"True"}'
+
         logger.debug(message)

nipype/pipeline/plugins/multiproc.py

@@ -138,12 +138,19 @@ class ResourceMultiProcPlugin(MultiProcPlugin):
     def __init__(self, plugin_args=None):
         super(ResourceMultiProcPlugin, self).__init__(plugin_args=plugin_args)
         self.plugin_args = plugin_args
+        self.processors = cpu_count()
+        memory = psutil.virtual_memory()
+        self.memory = memory.total / (1024*1024*1024)
+        if self.plugin_args:
+            if 'n_procs' in self.plugin_args:
+                self.processors = self.plugin_args['n_procs']
+            if 'memory' in self.plugin_args:
+                self.memory = self.plugin_args['memory']
 
     def _wait(self):
         if len(self.pending_tasks) > 0:
             semaphore_singleton.semaphore.acquire()
-        else:
-            semaphore_singleton.semaphore.release()
+        semaphore_singleton.semaphore.release()
 
 
     def _submit_job(self, node, updatehash=False):
@@ -162,14 +169,6 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             Check memory (gb) and cores usage before running jobs.
         """
         executing_now = []
-        processors = cpu_count()
-        memory = psutil.virtual_memory()
-        memory = memory.total / (1024*1024*1024)
-        if self.plugin_args:
-            if 'n_procs' in self.plugin_args:
-                processors = self.plugin_args['n_procs']
-            if 'memory' in self.plugin_args:
-                memory = self.plugin_args['memory']
 
         # Check to see if a job is available
         jobids = np.flatnonzero((self.proc_pending == True) & (self.depidx.sum(axis=0) == 0).__array__())
@@ -181,8 +180,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             busy_memory+= self.procs[jobid]._interface.memory
             busy_processors+= self.procs[jobid]._interface.num_threads
 
-        free_memory = memory - busy_memory
-        free_processors = processors - busy_processors
+        free_memory = self.memory - busy_memory
+        free_processors = self.processors - busy_processors
 
 
         #check all jobs without dependency not run

nipype/pipeline/plugins/draw_gantt_chart.py → nipype/utils/draw_gantt_chart.py

@@ -142,7 +142,4 @@ def generate_gantt_chart(logfile, cores, minute_scale=10, space_between_minutes=
     #save file
     html_file = open(logfile +'.html', 'wb')
     html_file.write(html_string)
-    html_file.close()
-
-
-generate_gantt_chart('/home/caroline/Desktop/callback.log', 8)
+    html_file.close()