add attribute real_memory to interface, change attr memory to estimated_memory

Author: carolFrohlich

nipype/interfaces/base.py

@@ -750,7 +750,8 @@ def __init__(self, **inputs):
             raise Exception('No input_spec in class: %s' %
                             self.__class__.__name__)
         self.inputs = self.input_spec(**inputs)
-        self.memory = 1
+        self.estimated_memory = 1
+        self.real_memory = 0
         self.num_threads = 1
 
     @classmethod

nipype/pipeline/plugins/callback_log.py

@@ -7,23 +7,23 @@ def log_nodes_cb(node, status):
     if status == 'start':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' +\
         node._id + '"' + ',"start":' + '"' +str(datetime.datetime.now()) +\
-        '"' + ',"memory":' + str(node._interface.memory) + ',"num_threads":' \
+        '"' + ',"memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
         + str(node._interface.num_threads) +  '}'
 
         logger.debug(message)
 
     elif status == 'end':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
         node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
-        '"' + ',"memory":' + str(node._interface.memory) + ',"num_threads":' \
+        '"' + ',"memory":' + str(node._interface.estimated_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":' \
+        '"' + ',"memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
         + str(node._interface.num_threads) + ',"error":"True"}'
 
         logger.debug(message)

nipype/pipeline/plugins/multiproc.py

@@ -177,7 +177,7 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
         busy_memory = 0
         busy_processors = 0
         for jobid in jobids:
-            busy_memory+= self.procs[jobid]._interface.memory
+            busy_memory+= self.procs[jobid]._interface.estimated_memory
             busy_processors+= self.procs[jobid]._interface.num_threads
 
         free_memory = self.memory - busy_memory
@@ -190,17 +190,17 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
 
         #sort jobs ready to run first by memory and then by number of threads
         #The most resource consuming jobs run first
-        jobids = sorted(jobids, key=lambda item: (self.procs[item]._interface.memory, self.procs[item]._interface.num_threads))
+        jobids = sorted(jobids, key=lambda item: (self.procs[item]._interface.estimated_memory, self.procs[item]._interface.num_threads))
 
         logger.debug('Free memory: %d, Free processors: %d', free_memory, free_processors)
 
 
         #while have enough memory and processors for first job
         #submit first job on the list
         for jobid in jobids:
-            logger.debug('Next Job: %d, memory: %d, threads: %d' %(jobid, self.procs[jobid]._interface.memory, self.procs[jobid]._interface.num_threads))
+            logger.debug('Next Job: %d, memory: %d, threads: %d' %(jobid, self.procs[jobid]._interface.estimated_memory, self.procs[jobid]._interface.num_threads))
 
-            if self.procs[jobid]._interface.memory <= free_memory and self.procs[jobid]._interface.num_threads <= free_processors:
+            if self.procs[jobid]._interface.estimated_memory <= free_memory and self.procs[jobid]._interface.num_threads <= free_processors:
                 logger.info('Executing: %s ID: %d' %(self.procs[jobid]._id, jobid))
                 executing_now.append(self.procs[jobid])
 
@@ -220,7 +220,7 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                 self.proc_done[jobid] = True
                 self.proc_pending[jobid] = True
 
-                free_memory -= self.procs[jobid]._interface.memory
+                free_memory -= self.procs[jobid]._interface.estimated_memory
                 free_processors -= self.procs[jobid]._interface.num_threads
 
                 # Send job to task manager and add to pending tasks

nipype/pipeline/plugins/tests/test_multiproc.py

@@ -103,7 +103,7 @@ def find_metrics(nodes, last_node):
             node_finish = parse(nodes[j]['finish'])
 
             if node_start < x and node_finish > x:
-                total_memory[i] += nodes[j]['memory']
+                total_memory[i] += nodes[j]['estimated_memory']
                 total_threads[i] += nodes[j]['num_threads']
                 start_index = j
 
@@ -140,10 +140,10 @@ def test_do_not_use_more_memory_then_specified():
     n3 = pe.Node(interface=TestInterfaceSingleNode(), name='n3')
     n4 = pe.Node(interface=TestInterfaceSingleNode(), name='n4')
 
-    n1.interface.memory = 1
-    n2.interface.memory = 1
-    n3.interface.memory = 10
-    n4.interface.memory = 1
+    n1.interface.estimated_memory = 1
+    n2.interface.estimated_memory = 1
+    n3.interface.estimated_memory = 10
+    n4.interface.estimated_memory = 1
 
     pipe.connect(n1, 'output1', n2, 'input1')
     pipe.connect(n1, 'output1', n3, 'input1')

nipype/utils/draw_gantt_chart.py

@@ -133,7 +133,7 @@ def draw_memory_bar(start, total_duration, nodes, space_between_minutes, minute_
             node_finish = parser.parse(nodes[j]['finish'])
 
             if node_start <= now and node_finish >= now:
-                memory[i] += nodes[j]['memory']
+                memory[i] += nodes[j]['estimated_memory']
             if node_start > now:
                 break
         now += datetime.timedelta(minutes=1)