remove some code duplication checking hash locally

Author: oesteban

nipype/pipeline/plugins/base.py

@@ -217,8 +217,10 @@ def _submit_mapnode(self, jobid):
         return False
 
     def _send_procs_to_workers(self, updatehash=False, graph=None):
-        """ Sends jobs to workers
         """
+        Sends jobs to workers
+        """
+
         while not np.all(self.proc_done):
             num_jobs = len(self.pending_tasks)
             if np.isinf(self.max_jobs):
@@ -258,27 +260,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                                 (self.procs[jobid]._id, jobid))
                     if self._status_callback:
                         self._status_callback(self.procs[jobid], 'start')
-                    continue_with_submission = True
-                    if str2bool(self.procs[jobid].config['execution']
-                                ['local_hash_check']):
-                        logger.debug('checking hash locally')
-                        try:
-                            hash_exists, _, _, _ = self.procs[
-                                jobid].hash_exists()
-                            logger.debug('Hash exists %s' % str(hash_exists))
-                            if (hash_exists and (self.procs[jobid].overwrite is False or
-                                (self.procs[jobid].overwrite is None and not
-                                    self.procs[jobid]._interface.always_run))):
-                                continue_with_submission = False
-                                self._task_finished_cb(jobid)
-                                self._remove_node_dirs()
-                        except Exception:
-                            self._clean_queue(jobid, graph)
-                            self.proc_pending[jobid] = False
-                            continue_with_submission = False
-                    logger.debug('Finished checking hash %s' %
-                                 str(continue_with_submission))
-                    if continue_with_submission:
+
+                    if not self._local_hash_check(jobid, graph):
                         if self.procs[jobid].run_without_submitting:
                             logger.debug('Running node %s on master thread' %
                                          self.procs[jobid])
@@ -301,6 +284,31 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             else:
                 break
 
+    def _local_hash_check(self, jobid, graph):
+        if not str2bool(self.procs[jobid].config['execution']['local_hash_check']):
+            return False
+
+        logger.debug('Checking hash (%d) locally', jobid)
+
+        hash_exists, _, _, _ = self.procs[jobid].hash_exists()
+        overwrite = self.procs[jobid].overwrite
+        always_run = self.procs[jobid]._interface.always_run
+
+        if hash_exists and (overwrite is False or
+                            overwrite is None and not always_run):
+            logger.debug('Skipping cached node %s with ID %s.',
+                         self.procs[jobid]._id, jobid)
+            try:
+                self._task_finished_cb(jobid)
+                self._remove_node_dirs()
+            except Exception:
+                logger.debug('Error skipping cached node %s (%s).',
+                             self.procs[jobid]._id, jobid)
+                self._clean_queue(jobid, graph)
+                self.proc_pending[jobid] = False
+            return True
+        return False
+
     def _task_finished_cb(self, jobid):
         """ Extract outputs and assign to inputs of dependent tasks
 

nipype/pipeline/plugins/multiproc.py

@@ -251,27 +251,9 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             self.proc_done[jobid] = True
             self.proc_pending[jobid] = True
 
-            if str2bool(self.procs[jobid].config['execution']['local_hash_check']):
-                logger.debug('checking hash locally')
-                try:
-                    hash_exists, _, _, _ = self.procs[jobid].hash_exists()
-                    overwrite = self.procs[jobid].overwrite
-                    always_run = self.procs[jobid]._interface.always_run
-                    if hash_exists and (overwrite is False or
-                                        overwrite is None and not always_run):
-                        logger.debug('Skipping cached node %s with ID %s.',
-                                     self.procs[jobid]._id, jobid)
-                        self._task_finished_cb(jobid)
-                        self._remove_node_dirs()
-                        continue
-                except Exception:
-                    traceback = format_exception(*sys.exc_info())
-                    self._report_crash(self.procs[jobid], traceback=traceback)
-                    self._clean_queue(jobid, graph)
-                    self.proc_pending[jobid] = False
-                    continue
-                finally:
-                    logger.debug('Finished checking hash')
+            # If cached just retrieve it, don't run
+            if self._local_hash_check(jobid, graph):
+                continue
 
             if self.procs[jobid].run_without_submitting:
                 logger.debug('Running node %s on master thread',