Merge pull request #892 from satra/fix/max_jobs

fix: max_jobs plugin parameter now acts within the inner submission loop

Author: satra

CHANGES

@@ -25,6 +25,7 @@ Next Release
 * FIX: MRTrix tracking algorithms were ignoring mask parameters.
 * FIX: FNIRT registration pathway and associated OpenFMRI example script
 * FIX: spm12b compatibility for Model estimate
+* FIX: Batch scheduler controls the number of maximum jobs properly
 
 Release 0.9.2 (January 31, 2014)
 ============

nipype/pipeline/plugins/base.py

@@ -252,13 +252,17 @@ def run(self, graph, config, updatehash=False):
             if toappend:
                 self.pending_tasks.extend(toappend)
             num_jobs = len(self.pending_tasks)
+            logger.debug('Number of pending tasks: %d' % num_jobs)
             if num_jobs < self.max_jobs:
                 if np.isinf(self.max_jobs):
                     slots = None
                 else:
-                    slots = self.max_jobs - num_jobs
+                    slots = max(0, self.max_jobs - num_jobs)
+                logger.debug('Slots available: %s' % slots)
                 self._send_procs_to_workers(updatehash=updatehash,
                                             slots=slots, graph=graph)
+            else:
+                logger.debug('Not submitting')
             sleep(2)
         self._remove_node_dirs()
         report_nodes_not_run(notrun)
@@ -327,9 +331,12 @@ def _send_procs_to_workers(self, updatehash=False, slots=None, graph=None):
             # Check to see if a job is available
             jobids = np.flatnonzero((self.proc_done == False) &
                                     (self.depidx.sum(axis=0) == 0).__array__())
+            num_jobs = len(self.pending_tasks)
+            if num_jobs >= self.max_jobs:
+                break
             if len(jobids) > 0:
                 # send all available jobs
-                logger.info('Submitting %d jobs' % len(jobids))
+                logger.info('Submitting %d jobs' % len(jobids[:slots]))
                 for jobid in jobids[:slots]:
                     if isinstance(self.procs[jobid], MapNode):
                         try:
@@ -339,6 +346,8 @@ def _send_procs_to_workers(self, updatehash=False, slots=None, graph=None):
                             self.proc_pending[jobid] = False
                             continue
                         if num_subnodes > 1:
+                            if num_subnodes > (self.max_jobs - len(self.pending_tasks)):
+                                break
                             submit = self._submit_mapnode(jobid)
                             if not submit:
                                 continue