Hi
I am trying to run a workflow on HPC and it has PBS for job scheduling. If I am correct, Nipype distributes the nodes in the workflow graph according to the resources available. While running a workflow, I use this code:
Reg_WorkFlow.run('PBSGraph', plugin_args={'template':template_nipype_job,
'dont_resubmit_completed_jobs': True})
But I am not sure whether it distributes the inputs across the available resources or whether all the inputs run on a single HPC node. For example, if I have 1000 inputs, whether all the inputs will load on a single HPC node or can I distribute the inputs across the HPC nodes?
I don’t have experience with this plugin, but you can use qsub_args in Workflow.run command to set the arguments that can be passed to the qsub command, you can see an example here.
For list of arguments and proper syntax, you should search for pbs/qsub examples, e.g. here
I hope that @djarecka’s response helped you with your issue.
Otherwise, could you not also just run a test? For example, if you create a dummy workflow that contains a 1000 nodes that all wait for 1 second. If your workflow runs sequential, it will take longer than 1000 seconds, if it runs in parallel, it will run quicker than a 1000 second.
For example, on a local machine, I would run it as follows:
from nipype import Node, Workflow
from nipype.interfaces.utility import Function
# Create the function that waits 1 second
def wait_1s():
from time import sleep
sleep(1)
# Create a node that applies this 1 second wait
node_x = Node(Function(function=wait_1s), name='node_000')
# Duplicate this node 999 times
nodes = [node_x.clone('node_%03d' % (i + 1)) for i in range(999)]
# Create a workflow and add all those nodes to the workflow
wf = Workflow(name='wf')
wf.add_nodes([node_x] + nodes)
# Run the workflow in parallel and stop the time
%time wf.run('MultiProc')
# Create the workflow again (otherwise it uses caching)
wf = Workflow(name='wf')
wf.add_nodes([node_x] + nodes)
# Run the workflow sequentially
%time wf.run('Linear')
As a note, the %time is IPython specific. You might want to use another approach to stop the time.
Not sure if this test can help you to answer your question or if I misunderstood your problem.