Merge pull request #328 from satra/enh/node_plugin_args

WIP: allow node specific plugin args for sge and pbs

Author: chrisgorgo

CHANGES

@@ -1,6 +1,7 @@
 Changes since 0.5
 =================
 
+* API: Node now allows specifying node level configuration for SGE/PBS clusters
 * API: Logging to file is disabled by default
 * API: New location of log file -> .nipype/nipype.cfg
 

doc/users/plugins.rst

@@ -115,6 +115,22 @@ a custom template::
        workflow.run(plugin='SGE',
           plugin_args=dict(template='mytemplate.sh', qsub_args='-q myqueue')
 
+In addition to overall workflow configuration, you can use node level
+configuration for PBS/SGE::
+
+    node.plugin_args = {'qsub_args': '-l nodes=1:ppn=3'}
+
+this would apply only to the node and is useful in situations, where a
+particular node might use more resources than other nodes in a workflow.
+
+.. note::
+
+  Setting the keyword `overwrite` would overwrite any global configuration with
+  this local configuration::
+
+     node.plugin_args = {'qsub_args': '-l nodes=1:ppn=3', 'overwrite': True}
+
+
 Condor
 ------
 

nipype/pipeline/engine.py

@@ -973,6 +973,7 @@ def __init__(self, interface, iterables=None, overwrite=None,
         self.run_without_submitting = run_without_submitting
         self.input_source = {}
         self.needed_outputs = []
+        self.plugin_args = {}
         if needed_outputs:
             self.needed_outputs = sorted(needed_outputs)
 

nipype/pipeline/plugins/condor.py

@@ -60,6 +60,14 @@ def _submit_batchtask(self, scriptfile, node):
         cmd = CommandLine('condor_qsub', environ=os.environ.data)
         path = os.path.dirname(scriptfile)
         qsubargs = ''
+        if self._qsub_args:
+            qsubargs = self._qsub_args
+        if 'qsub_args' in node.plugin_args:
+            if 'overwrite' in node.plugin_args and\
+               node.plugin_args['overwrite']:
+                qsubargs = node.plugin_args['qsub_args']
+            else:
+                qsubargs += (" " + node.plugin_args['qsub_args'])
         if self._qsub_args:
             qsubargs = self._qsub_args
         if '-o' not in qsubargs:

nipype/pipeline/plugins/pbs.py

@@ -52,6 +52,12 @@ def _submit_batchtask(self, scriptfile, node):
         qsubargs = ''
         if self._qsub_args:
             qsubargs = self._qsub_args
+        if 'qsub_args' in node.plugin_args:
+            if 'overwrite' in node.plugin_args and \
+               node.plugin_args['overwrite']:
+                qsubargs = node.plugin_args['qsub_args']
+            else:
+                qsubargs += (" " + node.plugin_args['qsub_args'])
         if '-o' not in qsubargs:
             qsubargs = '%s -o %s' % (qsubargs, path)
         if '-e' not in qsubargs:

nipype/pipeline/plugins/sge.py

@@ -53,6 +53,12 @@ def _submit_batchtask(self, scriptfile, node):
         qsubargs = ''
         if self._qsub_args:
             qsubargs = self._qsub_args
+        if 'qsub_args' in node.plugin_args:
+            if 'overwrite' in node.plugin_args and\
+               node.plugin_args['overwrite']:
+                qsubargs = node.plugin_args['qsub_args']
+            else:
+                qsubargs += (" " + node.plugin_args['qsub_args'])
         if '-o' not in qsubargs:
             qsubargs = '%s -o %s' % (qsubargs, path)
         if '-e' not in qsubargs: