Mondaic

salvus.flow.executors.task_chain.task_chain_controller

salvus.flow.executors.task_chain.task_chain_controller salvus flow executors task_chain task_chain_controller

The task chain controller implementation.

Classes

StdOutErr

class StdOutErr(builtins.object):
    def __init__(
        self, stdout: dict | None, stderr: dict | None, timings: dict | None
    ) -> None: ...

The stdout and stderr of a single task chain.

Both output could be None in case they don’t exist.

Parameters
  • stdout dict | None — The stdout.
  • stderr dict | None — The stderr.
  • timings dict | None — The timings for the given task chain.
Methods
get_diagnosis_string()
def get_diagnosis_string(self) -> str: ...

Get a hopefully useful diagnostic string from the task chain output.

Returns str — A formatted string with stdout, stderr, and timing information for each task in the chain.

TaskChainController

class TaskChainController(builtins.object):
    def __init__(
        self,
        task_chain_tasks: dict[str, list],
        sites: dict[str, TaskChainSiteConfig],
        task_chain_runner_context: typing.Callable | None = None,
    ): ...

The TaskChainController is responsible for distributing TaskChainRunners across resources.

Parameters
  • task_chain_tasks dict[str, list] — A dictionary mapping an arbitrary job key (name) to a list of tasks, e.g. tasks that happen within one task chain.
  • sites dict[str, TaskChainSiteConfig] — A dictionary of identifier + site config for sites to run the tasks on. The identifier does not have to be the site name.
  • task_chain_runner_context typing.Callable | None — A optional function returning a context manager that will be used in each TaskChainRunner prior to running the actual task chains.
Methods
check_status_of_task_chain_runners()
def check_status_of_task_chain_runners(
    self, raise_on_error: bool = True
) -> None: ...

Check the status of the individual task chain runners. Will return nothing if all finished, otherwise an exception with more information will be thrown.

Parameters
  • raise_on_error bool — If True, an exception will be raised if one of the child local task chain runner fail (not the individual task chains). If False, a warning will be logged.
Returns None
delete_remote_files()
def delete_remote_files(self, verbosity: int = 1) -> None: ...

Delete all remote files for the task chain controller - in most cases, nothing should remain after this method has been called.

Parameters
  • verbosity int — The verbosity.
Returns None
get_site_and_remote_path_for_task_chain()
def get_site_and_remote_path_for_task_chain(
    self, identifier: str
) -> tuple[BaseExecutor, _RemotePathType]: ...

Get the site + remote path for a given task chain.

Parameters
  • identifier str — Identifier of the task chain to get.
Returns tuple[BaseExecutor, _RemotePathType]
get_status_for_all_task_chains()
def get_status_for_all_task_chains(self) -> dict[str, JobStatus | None]: ...

Get a dictionary with the status of every individual task chain.

This is most useful after the controller finished running.

The status for a task chain might be None, if it has not been set.

Returns dict[str, JobStatus | None]
get_stdouterr_for_task_chain()
def get_stdouterr_for_task_chain(self, identifier: str) -> StdOutErr: ...

Get the stdout and stderr for a given task chain.

Parameters
  • identifier str — Identifier of the task chain to get.
Returns StdOutErr
launch()
def launch(self) -> None: ...

Launch the remote task chain runners.

Returns None
run()
def run(self, assert_no_failed_runners: bool = True) -> None: ...

Run the task chains across the individual sites by dispatching to one task chain runner per site.

The task chain runner will take care of the parallelization per site.

Parameters
  • assert_no_failed_runners bool — If True, an exception will be raised if one of the child local task chain runner fails (not the individual task chains). If False, a warning will be logged. This is useful for error resilient workflows but will require error handling elsewhere.
Returns None
update_status()
def update_status(self) -> dict[str, JobStatus]: ...

Update the status of all task chain runners.

Returns dict[str, JobStatus]
wait()
def wait(self) -> None: ...

Wait until all task chains runners have finished or failed.

Returns None