ParslWorkflow#

class lsst.ctrl.bps.parsl.ParslWorkflow(name: str, config: BpsConfig, path: str, jobs: dict[str, ParslJob], parents: Mapping[str, Iterable[str]], endpoints: Iterable[str], final: ParslJob | None = None)#

Bases: BaseWmsWorkflow

Parsl-based workflow object to manage execution of workflow.

Parameters#

namestr: Unique name of workflow.
configlsst.ctrl.bps.BpsConfig: Generic workflow config.
pathstr: Path prefix for workflow output files.
jobsdict mapping str to ParslJob: Jobs to be executed.
parentsdict mapping str to iterable of str: Dependency tree. Keywords are job names, and values are a list of job names that must be executed before the keyword job name can be executed.
endpointsiterable of str: Endpoints of the dependency tree. These jobs (specified by name) have no children.
finalParslJob, optional: Final job to be done, e.g., to merge the execution butler. This is done locally.

Methods Summary

`execute`(name)	Execute a job.
`finalize_jobs`()	Run final jobs.
`from_generic_workflow`(config, ...)	Create a ParslWorkflow object from a BPS GenericWorkflow.
`initialize_jobs`()	Run initial jobs.
`load_dfk`()	Load data frame kernel.
`read`(out_prefix)	Construct from the saved workflow state.
`restart`()	Restart the workflow after interruption.
`run`([block])	Run the workflow.
`shutdown`()	Shut down the workflow.
`start`()	Start the workflow.
`write`(out_prefix)	Write workflow state.

Methods Documentation

execute(name: str) → Future | None#

Execute a job.

Parameters#

namestr: Name of job to execute.

Returns#

futureFuture or None: A Future object linked to the execution of the job, or None if the job is being reserved to run locally.

finalize_jobs()#

Run final jobs.

These jobs are run locally after all other jobs are complete.

This is used to merge the execution butler.

classmethod from_generic_workflow(config: BpsConfig, generic_workflow: GenericWorkflow, out_prefix: str, service_class: str) → BaseWmsWorkflow#

Create a ParslWorkflow object from a BPS GenericWorkflow.

Parameters#

configBpsConfig: Configuration of the workflow.
generic_workflowlsst.ctrl.bps.generic_workflow.GenericWorkflow: Generic representation of a single workflow.
out_prefixstr: Prefix for workflow output files.
service_classstr: Full module name of WMS service class that created this workflow.

Returns#

selfParslWorkflow: Constructed workflow.

initialize_jobs()#

Run initial jobs.

These jobs are run locally before any other jobs are submitted to parsl.

This is used to set up the butler.

load_dfk()#

Load data frame kernel.

This starts parsl.

classmethod read(out_prefix: str) → ParslWorkflow#

Construct from the saved workflow state.

Parameters#

out_prefixstr: Root directory to be used for WMS workflow inputs and outputs as well as internal WMS files.

Returns#

selfParslWorkflow: Constructed workflow.

restart()#: Restart the workflow after interruption.

run(block: bool = True) → list[Future | None]#

Run the workflow.

Parameters#

blockbool, optional: Block returning from this method until the workflow is complete? If False, jobs may still be running when this returns, and it is the user’s responsibility to call the finalize_jobs and shutdown methods when they are complete.

Returns#

futureslist of Future: Future objects linked to the execution of the endpoint jobs.

shutdown()#

Shut down the workflow.

This stops parsl.

start()#: Start the workflow.

write(out_prefix: str)#

Write workflow state.

This, in combination with the parsl checkpoint files, can be used to restart a workflow that was interrupted.

Parameters#

out_prefixstr: Root directory to be used for WMS workflow inputs and outputs as well as internal WMS files.