Changed memory parameters to be memory_gb to be more explicit, used runtime Bunch object only for runtime stats storage instead of using results dictionary, renamed ResourceMultiProc to MultiProc for backwards-compatiblity

Author: pintohutch

nipype/interfaces/ants/base.py

@@ -12,7 +12,7 @@
 # -Using -1 gives primary responsibilty to ITKv4 to do the correct
 #  thread limitings.
 # -Using 1 takes a very conservative approach to avoid overloading
-#  the computer (when running ResourceMultiProc) by forcing everything to
+#  the computer (when running MultiProc) by forcing everything to
 #  single threaded.  This can be a severe penalty for registration
 #  performance.
 LOCAL_DEFAULT_NUMBER_OF_THREADS = 1

nipype/interfaces/base.py

@@ -764,7 +764,7 @@ def __init__(self, **inputs):
             raise Exception('No input_spec in class: %s' %
                             self.__class__.__name__)
         self.inputs = self.input_spec(**inputs)
-        self.estimated_memory = 1
+        self.estimated_memory_gb = 1
         self.num_threads = 1
 
     @classmethod
@@ -1406,7 +1406,7 @@ def _process(drain=0):
         result['stderr'] = []
         result['merged'] = ''
 
-    setattr(runtime, 'runtime_memory', mem_mb/1024.0)
+    setattr(runtime, 'runtime_memory_gb', mem_mb/1024.0)
     setattr(runtime, 'runtime_threads', num_threads)
     runtime.stderr = '\n'.join(result['stderr'])
     runtime.stdout = '\n'.join(result['stdout'])

nipype/interfaces/tests/test_runtime_profiler.py

@@ -47,8 +47,7 @@ class UseResources(CommandLine):
 # Test case for the run function
 class RuntimeProfilerTestCase(unittest.TestCase):
     '''
-    This class is a test case for the ResourceMultiProc plugin runtime
-    profiler
+    This class is a test case for the runtime profiler
 
     Inherits
     --------
@@ -129,7 +128,7 @@ def _run_workflow(self):
 
         # Resources used node
         resource_node = pe.Node(UseResources(), name='resource_node')
-        resource_node.interface.estimated_memory = num_gb
+        resource_node.interface.estimated_memory_gb = num_gb
         resource_node.interface.num_threads = num_procs
 
         # Connect workflow
@@ -140,7 +139,7 @@ def _run_workflow(self):
         plugin_args = {'n_procs' : num_procs,
                        'memory' : num_gb,
                        'status_callback' : log_nodes_cb}
-        wf.run(plugin='ResourceMultiProc', plugin_args=plugin_args)
+        wf.run(plugin='MultiProc', plugin_args=plugin_args)
 
         # Get runtime stats from log file
         finish_str = open(log_file, 'r').readlines()[1].rstrip('\n')
@@ -169,7 +168,7 @@ def test_wf_logfile(self):
         node_stats = json.loads(finish_str)
 
         # Read out runtime stats
-        runtime_gb = float(node_stats['runtime_memory'])
+        runtime_gb = float(node_stats['runtime_memory_gb'])
         runtime_procs = int(node_stats['runtime_threads'])
 
         # Error message formatting

nipype/pipeline/engine/nodes.py

@@ -737,15 +737,20 @@ def write_report(self, report_type=None, cwd=None):
                 fp.close()
                 return
             fp.writelines(write_rst_header('Runtime info', level=1))
+            # Init rst dictionary of runtime stats
+            rst_dict = {'hostname' : self.result.runtime.hostname,
+                        'duration' : self.result.runtime.duration}
+            # Try and insert memory/threads usage if available
+            try:
+                rst_dict['runtime_memory_gb'] = self.result.runtime.runtime_memory_gb
+                rst_dict['runtime_threads'] = self.result.runtime.runtime_threads
+            except:
+                logger.info('Runtime memory and threads stats unavailable')
             if hasattr(self.result.runtime, 'cmdline'):
-                fp.writelines(write_rst_dict(
-                    {'hostname': self.result.runtime.hostname,
-                     'duration': self.result.runtime.duration,
-                     'command': self.result.runtime.cmdline}))
+                rst_dict['command'] = self.result.runtime.cmdline
+                fp.writelines(write_rst_dict(rst_dict))
             else:
-                fp.writelines(write_rst_dict(
-                    {'hostname': self.result.runtime.hostname,
-                     'duration': self.result.runtime.duration}))
+                fp.writelines(write_rst_dict(rst_dict))
             if hasattr(self.result.runtime, 'merged'):
                 fp.writelines(write_rst_header('Terminal output', level=2))
                 fp.writelines(write_rst_list(self.result.runtime.merged))

nipype/pipeline/engine/tests/test_engine.py

@@ -723,7 +723,7 @@ def func1(in1):
     # test running the workflow on default conditions
     error_raised = False
     try:
-        w1.run(plugin='ResourceMultiProc')
+        w1.run(plugin='MultiProc')
     except Exception as e:
         from nipype.pipeline.engine.base import logger
         logger.info('Exception: %s' % str(e))
@@ -737,7 +737,7 @@ def func1(in1):
     # test running the workflow on serial conditions
     error_raised = False
     try:
-        w1.run(plugin='ResourceMultiProc')
+        w1.run(plugin='MultiProc')
     except Exception as e:
         from nipype.pipeline.engine.base import logger
         logger.info('Exception: %s' % str(e))

nipype/pipeline/engine/tests/test_utils.py

@@ -214,7 +214,7 @@ def test_function3(arg):
 
     out_dir = mkdtemp()
 
-    for plugin in ('Linear',):  # , 'ResourceMultiProc'):
+    for plugin in ('Linear',):  # , 'MultiProc'):
         n1 = pe.Node(niu.Function(input_names=['arg1'],
                                   output_names=['out_file1', 'out_file2', 'dir'],
                                   function=test_function),

nipype/pipeline/plugins/__init__.py

@@ -9,7 +9,7 @@
 from .sge import SGEPlugin
 from .condor import CondorPlugin
 from .dagman import CondorDAGManPlugin
-from .multiproc import ResourceMultiProcPlugin
+from .multiproc import MultiProcPlugin
 from .ipython import IPythonPlugin
 from .somaflow import SomaFlowPlugin
 from .pbsgraph import PBSGraphPlugin

nipype/pipeline/plugins/base.py

@@ -246,7 +246,7 @@ def run(self, graph, config, updatehash=False):
                             notrun.append(self._clean_queue(jobid, graph,
                                                             result=result))
                         else:
-                            self._task_finished_cb(jobid, result)
+                            self._task_finished_cb(jobid)
                             self._remove_node_dirs()
                         self._clear_task(taskid)
                     else:
@@ -415,18 +415,15 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             else:
                 break
 
-    def _task_finished_cb(self, jobid, result=None):
+    def _task_finished_cb(self, jobid):
         """ Extract outputs and assign to inputs of dependent tasks
 
         This is called when a job is completed.
         """
         logger.info('[Job finished] jobname: %s jobid: %d' %
                     (self.procs[jobid]._id, jobid))
         if self._status_callback:
-            if result == None:
-                if self._taskresult.has_key(jobid):
-                    result = self._taskresult[jobid].get()
-            self._status_callback(self.procs[jobid], 'end', result)
+            self._status_callback(self.procs[jobid], 'end')
         # Update job and worker queues
         self.proc_pending[jobid] = False
         # update the job dependency structure

nipype/pipeline/plugins/callback_log.py

@@ -8,7 +8,7 @@
 import logging
 
 # Log node stats function
-def log_nodes_cb(node, status, result=None):
+def log_nodes_cb(node, status):
     """Function to record node run statistics to a log file as json
     dictionaries
     """
@@ -17,47 +17,40 @@ def log_nodes_cb(node, status, result=None):
     logger = logging.getLogger('callback')
 
     # Check runtime profile stats
-    if result is None:
-        runtime_memory = runtime_seconds = runtime_threads = 'N/A'
-    else:
-        try:
-            runtime_memory = result['runtime_memory']
-        except KeyError:
-            runtime_memory = 'Unknown'
+    if node.result is not None:
         try:
-            runtime_seconds = result['runtime_seconds']
-        except KeyError:
-            runtime_seconds = 'Unknown'
-        try:
-            runtime_threads = result['runtime_threads']
+            runtime = node.result.runtime
+            runtime_memory_gb = runtime.runtime_memory_gb
+            runtime_threads = runtime.runtime_threads
         except:
-            runtime_threads = 'Unknown'
+            runtime_memory_gb = runtime_threads = 'Unkown'
+    else:
+        runtime_memory_gb = runtime_threads = 'N/A'
 
     # Check status and write to log
     # Start
     if status == 'start':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' +\
         node._id + '"' + ',"start":' + '"' +str(datetime.datetime.now()) +\
-        '"' + ',"estimated_memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
-        + str(node._interface.num_threads) + '}'
+        '"' + ',"estimated_memory_gb":' + str(node._interface.estimated_memory_gb) + \
+        ',"num_threads":' + str(node._interface.num_threads) + '}'
 
         logger.debug(message)
     # End
     elif status == 'end':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
         node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +  \
-        '"' + ',"estimated_memory":' + '"'+ str(node._interface.estimated_memory) + '"'+ \
-        ',"num_threads":' + '"'+ str(node._interface.num_threads) + '"'+ \
+        '"' + ',"estimated_memory_gb":' + '"'+ str(node._interface.estimated_memory_gb) + \
+        '"'+ ',"num_threads":' + '"'+ str(node._interface.num_threads) + '"'+ \
         ',"runtime_threads":' + '"'+ str(runtime_threads) + '"'+ \
-        ',"runtime_memory":' + '"'+ str(runtime_memory) + '"' + \
-        ',"runtime_seconds":' + '"'+ str(runtime_seconds) + '"'+ '}'
+        ',"runtime_memory_gb":' + '"'+ str(runtime_memory_gb) + '"' + '}'
 
         logger.debug(message)
     # Other
     else:
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' + \
         node._id + '"' + ',"finish":' + '"' + str(datetime.datetime.now()) +\
-        '"' + ',"estimated_memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
-        + str(node._interface.num_threads) + ',"error":"True"}'
+        '"' + ',"estimated_memory_gb":' + str(node._interface.estimated_memory_gb) + \
+        ',"num_threads":' + str(node._interface.num_threads) + ',"error":"True"}'
 
         logger.debug(message)

nipype/pipeline/plugins/multiproc.py

@@ -41,22 +41,12 @@ def run_node(node, updatehash):
         dictionary containing the node runtime results and stats
     """
 
-    # Import packages
-    import datetime
- 
     # Init variables
     result = dict(result=None, traceback=None)
 
-    # 
+    # Try and execute the node via node.run()
     try:
-        start = datetime.datetime.now()
-        retval = node.run(updatehash=updatehash)
-        run_secs = (datetime.datetime.now() - start).total_seconds()
-        result['result'] = retval
-        result['runtime_seconds'] = run_secs
-        if hasattr(retval.runtime, 'get'):
-            result['runtime_memory'] = retval.runtime.get('runtime_memory')
-            result['runtime_threads'] = retval.runtime.get('runtime_threads')
+        result['result'] = node.run(updatehash=updatehash)
     except:
         etype, eval, etr = sys.exc_info()
         result['traceback'] = format_exception(etype, eval, etr)
@@ -88,7 +78,7 @@ def release_lock(args):
     semaphore_singleton.semaphore.release()
 
 
-class ResourceMultiProcPlugin(DistributedPluginBase):
+class MultiProcPlugin(DistributedPluginBase):
     """Execute workflow with multiprocessing, not sending more jobs at once
     than the system can support.
 
@@ -98,7 +88,7 @@ class ResourceMultiProcPlugin(DistributedPluginBase):
     the number of threads and memory of the system is used.
 
     System consuming nodes should be tagged:
-    memory_consuming_node.interface.estimated_memory = 8 #Gb
+    memory_consuming_node.interface.estimated_memory_gb = 8
     thread_consuming_node.interface.num_threads = 16
 
     The default number of threads and memory for a node is 1. 
@@ -107,12 +97,12 @@ class ResourceMultiProcPlugin(DistributedPluginBase):
 
     - non_daemon : boolean flag to execute as non-daemon processes
     - num_threads: maximum number of threads to be executed in parallel
-    - estimated_memory: maximum memory that can be used at once.
+    - estimated_memory_gb: maximum memory (in GB) that can be used at once.
 
     """
 
     def __init__(self, plugin_args=None):
-        super(ResourceMultiProcPlugin, self).__init__(plugin_args=plugin_args)
+        super(MultiProcPlugin, self).__init__(plugin_args=plugin_args)
         self._taskresult = {}
         self._taskid = 0
         non_daemon = True
@@ -186,8 +176,8 @@ 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.estimated_memory
-            busy_processors+= self.procs[jobid]._interface.num_threads
+            busy_memory += self.procs[jobid]._interface.estimated_memory_gb
+            busy_processors += self.procs[jobid]._interface.num_threads
 
         free_memory = self.memory - busy_memory
         free_processors = self.processors - busy_processors
@@ -201,21 +191,21 @@ 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.estimated_memory,
+                        key=lambda item: (self.procs[item]._interface.estimated_memory_gb,
                                           self.procs[item]._interface.num_threads))
 
-        logger.debug('Free memory: %d, Free processors: %d',
+        logger.debug('Free memory (GB): %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.estimated_memory,
+            logger.debug('Next Job: %d, memory (GB): %d, threads: %d' \
+                         % (jobid, self.procs[jobid]._interface.estimated_memory_gb,
                             self.procs[jobid]._interface.num_threads))
 
-            if self.procs[jobid]._interface.estimated_memory <= free_memory and \
+            if self.procs[jobid]._interface.estimated_memory_gb <= 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])
@@ -236,7 +226,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.estimated_memory
+                free_memory -= self.procs[jobid]._interface.estimated_memory_gb
                 free_processors -= self.procs[jobid]._interface.num_threads
 
                 # Send job to task manager and add to pending tasks