Added initial code for getting used memory of node

Author: ccraddock

nipype/interfaces/base.py

@@ -751,9 +751,17 @@ def __init__(self, **inputs):
                             self.__class__.__name__)
         self.inputs = self.input_spec(**inputs)
         self.estimated_memory = 1
-        self.real_memory = 0
+        self._real_memory = 0
         self.num_threads = 1
 
+    @property
+    def real_memory(self):
+        return self._real_memory
+
+    @real_memory.setter
+    def real_memory(self, value):
+        self._real_memory = value
+
     @classmethod
     def help(cls, returnhelp=False):
         """ Prints class help

nipype/pipeline/plugins/base.py

@@ -416,7 +416,10 @@ def _task_finished_cb(self, jobid):
         logger.info('[Job finished] jobname: %s jobid: %d' %
                     (self.procs[jobid]._id, jobid))
         if self._status_callback:
-            self._status_callback(self.procs[jobid], 'end')
+            print '!!!!!!!!!!!!!!!!!!!'
+            print self._taskresult
+            print self._taskresult.keys()
+            self._status_callback(self.procs[jobid], 'end', self._taskresult[self.taskresultid])
         # Update job and worker queues
         self.proc_pending[jobid] = False
         # update the job dependency structure

nipype/pipeline/plugins/callback_log.py

@@ -1,7 +1,7 @@
 import datetime
 import logging
 
-def log_nodes_cb(node, status):
+def log_nodes_cb(node, status, result=None):
     print 'status', status
     logger = logging.getLogger('callback')
     if status == 'start':
@@ -16,7 +16,7 @@ def log_nodes_cb(node, status):
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
         node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
         '"' + ',"memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
-        + str(node._interface.num_threads) + ',"real memory":' str(node._interface.real_memory) + '}'
+        + str(node._interface.num_threads) + ',"real memory":' + str(result['real_memory']) + '}'
 
         logger.debug(message)
 

nipype/pipeline/plugins/multiproc.py

@@ -21,16 +21,20 @@ def run_node(node, updatehash, plugin_args=None):
         run_memory = False
     if run_memory:
         import memory_profiler
-        proc = (node.run(), (), {'updatehash' : updatehash})
+        proc = (node.run, (), {'updatehash' : updatehash})
         mem_mb, retval = memory_profiler.memory_usage(proc, max_usage=True, retval=True)
         result['result'] = retval
-        node._interface.real_memory = mem_mb[0]/1024.0
-    try:
-        result['result'] = node.run(updatehash=updatehash)
-    except:
-        etype, eval, etr = sys.exc_info()
-        result['traceback'] = format_exception(etype,eval,etr)
-        result['result'] = node.result
+        result['real_memory'] = 100
+        print 'Just populated task result!!!!!!!!!!!!!!!!!!!'
+        print result
+        #node._interface.real_memory = mem_mb[0]/1024.0
+    else:
+        try:
+            result['result'] = node.run(updatehash=updatehash)
+        except:
+            etype, eval, etr = sys.exc_info()
+            result['traceback'] = format_exception(etype,eval,etr)
+            result['result'] = node.result
     return result
 
 
@@ -173,6 +177,8 @@ def _submit_job(self, node, updatehash=False):
         self._taskresult[self._taskid] = self.pool.apply_async(run_node,
                                                                (node, updatehash, self.plugin_args),
                                                                callback=release_lock)
+        print 'Printing on output!!!!!!!!!!'
+        print self._taskresult, self._taskid
         return self._taskid
 
     def _send_procs_to_workers(self, updatehash=False, graph=None):
@@ -237,7 +243,6 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                 # Send job to task manager and add to pending tasks
                 if self._status_callback:
                     self._status_callback(self.procs[jobid], 'start')
-                
                 if str2bool(self.procs[jobid].config['execution']['local_hash_check']):
                     logger.debug('checking hash locally')
                     try: