Fixed pickling bug of instance method by passing profiling flag instead of complete plugin_args dict

Author: pintohutch

nipype/interfaces/utility.py

@@ -450,14 +450,14 @@ def _run_interface(self, runtime):
                 args[name] = value
 
         # Record memory of function_handle
-        try:
-            import memory_profiler
-            proc = (function_handle, (), args)
-            mem_mb, out = memory_profiler.memory_usage(proc=proc, retval=True, include_children=True, max_usage=True)
-            setattr(runtime, 'cmd_memory', mem_mb[0]/1024.0)
+        #try:
+        #    import memory_profiler
+        #    proc = (function_handle, (), args)
+        #    mem_mb, out = memory_profiler.memory_usage(proc=proc, retval=True, include_children=True, max_usage=True)
+        #    setattr(runtime, 'cmd_memory', mem_mb[0]/1024.0)
         # If no memory_profiler package, run without recording memory
-        except:
-            out = function_handle(**args)
+        #except:
+        out = function_handle(**args)
 
         if len(self._output_names) == 1:
             self._out[self._output_names[0]] = out

nipype/pipeline/plugins/base.py

@@ -250,7 +250,7 @@ def run(self, graph, config, updatehash=False):
                         self._clear_task(taskid)
                     else:
                         toappend.insert(0, (taskid, jobid))
-                except Exception:
+                except Exception as exc:
                     result = {'result': None,
                               'traceback': format_exc()}
                     notrun.append(self._clean_queue(jobid, graph,

nipype/pipeline/plugins/multiproc.py

@@ -21,23 +21,20 @@
 
 
 # Run node
-def run_node(node, updatehash, plugin_args=None):
+def run_node(node, updatehash, runtime_profile=False):
     """docstring
     """
-
+ 
     # Import packages
     try:
-        runtime_profile = plugin_args['runtime_profile']
         import memory_profiler
         import datetime
-    except KeyError:
-        runtime_profile = False
     except ImportError:
         runtime_profile = False
-
+ 
     # Init variables
     result = dict(result=None, traceback=None)
-
+    runtime_profile = False
     # If we're profiling the run
     if runtime_profile:
         try:
@@ -167,9 +164,14 @@ 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.plugin_args),
-                                                               callback=release_lock)
+        try:
+            runtime_profile = self.plugin_args['runtime_profile']
+        except:
+            runtime_profile = False
+        self._taskresult[self._taskid] = \
+            self.pool.apply_async(run_node,
+                                  (node, updatehash, runtime_profile),
+                                  callback=release_lock)
         return self._taskid
 
     def _send_procs_to_workers(self, updatehash=False, graph=None):

nipype/pipeline/plugins/tests/test_callback.py

@@ -26,7 +26,7 @@ class Status(object):
     def __init__(self):
         self.statuses = []
 
-    def callback(self, node, status):
+    def callback(self, node, status, result=None):
         self.statuses.append((node, status))
 
 
@@ -105,3 +105,7 @@ def test_callback_multiproc_exception():
     yield assert_equal, so.statuses[0][1], 'start'
     yield assert_equal, so.statuses[1][1], 'exception'
     rmtree(wf.base_dir)
+
+if __name__ == '__main__':
+    import nose
+    nose.run()