Merge pull request #728 from BRAINSia/EnhanceQSubProcessing

ENH: Large SGE jobs were overloading on qstat

Author: satra

nipype/pipeline/plugins/base.py

@@ -305,8 +305,8 @@ def _submit_mapnode(self, jobid):
             self.mapnodesubids[self.depidx.shape[0] + i] = jobid
         self.procs.extend(mapnodesubids)
         self.depidx = ssp.vstack((self.depidx,
-                                  ssp.lil_matrix(np.zeros((numnodes,
-                                                           self.depidx.shape[1])))),
+                                  ssp.lil_matrix(np.zeros(
+                                  (numnodes, self.depidx.shape[1])))),
                                  'lil')
         self.depidx = ssp.hstack((self.depidx,
                                   ssp.lil_matrix(
@@ -351,16 +351,19 @@ def _send_procs_to_workers(self, updatehash=False, slots=None, graph=None):
                     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']):
+                    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 == False or
-                               (self.procs[jobid].overwrite == None and
-                                    not self.procs[jobid]._interface.always_run))):
+                                 (self.procs[jobid].overwrite == False or
+                                   (self.procs[jobid].overwrite == None and
+                                    not self.procs[jobid]._interface.always_run)
+                                 )
+                               ):
                                 continue_with_submission = False
                                 self._task_finished_cb(jobid)
                                 self._remove_node_dirs()
@@ -438,7 +441,8 @@ def _remove_node_dirs(self):
         """Removes directories whose outputs have already been used up
         """
         if str2bool(self._config['execution']['remove_node_directories']):
-            for idx in np.nonzero((self.refidx.sum(axis=1) == 0).__array__())[0]:
+            for idx in np.nonzero(
+                                 (self.refidx.sum(axis=1) == 0).__array__())[0]:
                 if idx in self.mapnodesubids:
                     continue
                 if self.proc_done[idx] and (not self.proc_pending[idx]):
@@ -508,9 +512,13 @@ def _get_result(self, taskid):
                            'traceback': None}
             results_file = None
             try:
-                raise IOError(('Job (%s) finished or terminated, but results file '
-                               'does not exist. Batch dir contains crashdump '
-                               'file if node raised an exception' % node_dir))
+                error_message = ('Job id ({0}) finished or terminated, but '
+                                 'results file does not exist after ({1}) '
+                                 'seconds. Batch dir contains crashdump file '
+                                 'if node raised an exception.\n'
+                                 'Node working directory: ({2}) '.format(
+                                 taskid,timeout,node_dir) )
+                raise IOError(error_message)
             except IOError, e:
                 result_data['traceback'] = format_exc()
         else:
@@ -584,13 +592,17 @@ def _get_args(self, node, keywords):
             value = getattr(self, "_" + keyword)
             if keyword == "template" and os.path.isfile(value):
                 value = open(value).read()
-            if hasattr(node, "plugin_args") and isinstance(node.plugin_args, dict) and keyword in node.plugin_args:
-                    if keyword == "template" and os.path.isfile(node.plugin_args[keyword]):
+            if (hasattr(node, "plugin_args") and
+                    isinstance(node.plugin_args, dict) and
+                        keyword in node.plugin_args):
+                    if (keyword == "template" and
+                            os.path.isfile(node.plugin_args[keyword])):
                         tmp_value = open(node.plugin_args[keyword]).read()
                     else:
                         tmp_value = node.plugin_args[keyword]
 
-                    if 'overwrite' in node.plugin_args and node.plugin_args['overwrite']:
+                    if ('overwrite' in node.plugin_args and
+                            node.plugin_args['overwrite']):
                         value = tmp_value
                     else:
                         value += tmp_value