Removed the memory profiler code to just pull in s3 datasink code

Author: pintohutch

nipype/interfaces/base.py

@@ -751,17 +751,8 @@ def __init__(self, **inputs):
                             self.__class__.__name__)
         self.inputs = self.input_spec(**inputs)
         self.estimated_memory = 1
-        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
@@ -1240,9 +1231,6 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     errfile = os.path.join(runtime.cwd, 'stderr.nipype')
     outfile = os.path.join(runtime.cwd, 'stdout.nipype')
 
-    # Init variables for memory profiling
-    ret = -1
-    interval = 0.1
 
     if output == 'stream':
         streams = [Stream('stdout', proc.stdout), Stream('stderr', proc.stderr)]
@@ -1260,9 +1248,6 @@ def _process(drain=0):
                 for stream in res[0]:
                     stream.read(drain)
         while proc.returncode is None:
-            if mem_prof:
-                ret = max([ret, _get_memory(proc.pid, include_children=True)])
-                time.sleep(interval)
             proc.poll()
             _process()
         _process(drain=1)
@@ -1278,39 +1263,23 @@ def _process(drain=0):
         result['merged'] = [r[1] for r in temp]
 
     if output == 'allatonce':
-        if mem_prof:
-            while proc.returncode is None:
-                ret = max([ret, _get_memory(proc.pid, include_children=True)])
-                time.sleep(interval)
-                proc.poll()
         stdout, stderr = proc.communicate()
         result['stdout'] = stdout.split('\n')
         result['stderr'] = stderr.split('\n')
         result['merged'] = ''
     if output == 'file':
-        if mem_prof:
-            while proc.returncode is None:
-                ret = max([ret, _get_memory(proc.pid, include_children=True)])
-                time.sleep(interval)
-                proc.poll()
         ret_code = proc.wait()
         stderr.flush()
         stdout.flush()
         result['stdout'] = [line.strip() for line in open(outfile).readlines()]
         result['stderr'] = [line.strip() for line in open(errfile).readlines()]
         result['merged'] = ''
     if output == 'none':
-        if mem_prof:
-            while proc.returncode is None:
-                ret = max([ret, _get_memory(proc.pid, include_children=True)])
-                time.sleep(interval)
-                proc.poll()
         proc.communicate()
         result['stdout'] = []
         result['stderr'] = []
         result['merged'] = ''
 
-    setattr(runtime, 'real_memory2', ret/1024.0)
     runtime.stderr = '\n'.join(result['stderr'])
     runtime.stdout = '\n'.join(result['stdout'])
     runtime.merged = result['merged']

nipype/pipeline/plugins/base.py

@@ -241,7 +241,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:
@@ -408,20 +408,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()
-                else:
-                    result = {'real_memory' : 'nokey'}
-            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

@@ -1,14 +1,8 @@
 import datetime
 import logging
 
-def log_nodes_cb(node, status, result=None):
+def log_nodes_cb(node, status):
     logger = logging.getLogger('callback')
-    try:
-        real_mem1 = result['real_memory']
-        real_mem2 = result['real_memory2']
-        run_seconds = result['run_seconds']
-    except Exception as exc:
-        real_mem1 = real_mem2 = run_seconds = 'N/A'
     if status == 'start':
         message  = '{"name":' + '"' + node.name + '"' + ',"id":' + '"' +\
         node._id + '"' + ',"start":' + '"' +str(datetime.datetime.now()) +\
@@ -20,15 +14,15 @@ def log_nodes_cb(node, status, result=None):
     elif status == 'end':
         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_memory1":' + str(real_mem1) + ',"real_memory2":' + str(real_mem2) + ',"run_seconds":' + str(run_seconds) + '}'
+        '"' + ',"estimate 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.estimated_memory) + ',"num_threads":' \
+        '"' + ',"estimate memory":' + str(node._interface.estimated_memory) + ',"num_threads":' \
         + str(node._interface.num_threads) + ',"error":"True"}'
 
         logger.debug(message)

nipype/pipeline/plugins/multiproc.py

@@ -83,8 +83,8 @@ def _submit_job(self, node, updatehash=False):
                 node.inputs.terminal_output = 'allatonce'
         except:
             pass
-        self._taskresult[self._taskid] = self.pool.apply_async(run_node, (node,
-                                                                updatehash,))
+        self._taskresult[self._taskid] = self.pool.apply_async(run_node,
+                                                               (node, updatehash,))
         return self._taskid
 
     def _report_crash(self, node, result=None):
@@ -161,7 +161,7 @@ def _submit_job(self, node, updatehash=False):
         except:
             pass
         self._taskresult[self._taskid] = self.pool.apply_async(run_node,
-                                                               (node, updatehash, self.plugin_args),
+                                                               (node, updatehash,),
                                                                callback=release_lock)
         return self._taskid