Merge pull request #1025 from satra/enh/workflow

Enh/workflow

Author: satra

CHANGES

@@ -12,6 +12,7 @@ Next release
 	   undefined (issue #994)(https://https://github.com/nipy/nipype/pull/996)
 * FIX: OpenfMRI support and FSL 5.0.7 changes (https://github.com/nipy/nipype/pull/1006)
 * FIX: Output prefix in SPM Normalize with modulation (https://github.com/nipy/nipype/pull/1023)
+* ENH: Usability improvements in cluster environments (https://github.com/nipy/nipype/pull/1025)
 
 Release 0.10.0 (October 10, 2014)
 ============

doc/users/config_file.rst

@@ -120,6 +120,11 @@ Execution
     characters will be replaced by their hash. (possible values: ``true`` and
 	``false``; default value: ``true``)
 
+*poll_sleep_duration*
+    This controls how long the job submission loop will sleep between submitting
+    all pending jobs and checking for job completion. To be nice to cluster
+    schedulers the default is set to 60 seconds.
+
 Example
 ~~~~~~~
 

nipype/algorithms/tests/test_auto_Overlap.py

@@ -254,16 +254,11 @@ def run(self, graph, config, updatehash=False):
             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 = max(0, self.max_jobs - num_jobs)
-                logger.debug('Slots available: %s' % slots)
                 self._send_procs_to_workers(updatehash=updatehash,
-                                            slots=slots, graph=graph)
+                                            graph=graph)
             else:
                 logger.debug('Not submitting')
-            sleep(2)
+            sleep(float(self._config['execution']['poll_sleep_duration']))
         self._remove_node_dirs()
         report_nodes_not_run(notrun)
 
@@ -324,16 +319,21 @@ def _submit_mapnode(self, jobid):
                                             np.zeros(numnodes, dtype=bool)))
         return False
 
-    def _send_procs_to_workers(self, updatehash=False, slots=None, graph=None):
-        """ Sends jobs to workers using ipython's taskclient interface
+    def _send_procs_to_workers(self, updatehash=False, graph=None):
+        """ Sends jobs to workers
         """
         while np.any(self.proc_done == False):
+            num_jobs = len(self.pending_tasks)
+            if np.isinf(self.max_jobs):
+                slots = None
+            else:
+                slots = max(0, self.max_jobs - num_jobs)
+            logger.debug('Slots available: %s' % slots)
+            if (num_jobs >= self.max_jobs) or (slots == 0):
+                break
             # 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[:slots]))
@@ -346,8 +346,6 @@ 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

nipype/pipeline/plugins/base.py

@@ -1101,11 +1101,12 @@ def write_workflow_prov(graph, filename=None, format='turtle'):
                 subresult = InterfaceResult(result.interface[idx],
                                             runtime, outputs={})
                 if result.inputs:
-                    subresult.inputs = result.inputs[idx]
+                    if idx < len(result.inputs):
+                        subresult.inputs = result.inputs[idx]
                 if result.outputs:
                     for key, value in result.outputs.items():
                         values = getattr(result.outputs, key)
-                        if isdefined(values):
+                        if isdefined(values) and idx < len(values):
                             subresult.outputs[key] = values[idx]
                 sub_bundle = ProvStore().add_results(subresult)
                 ps.g = merge_bundles(ps.g, sub_bundle)

nipype/pipeline/utils.py

@@ -48,6 +48,7 @@
 stop_on_unknown_version = false
 write_provenance = false
 parameterize_dirs = true
+poll_sleep_duration = 60
 
 [check]
 interval = 1209600