salvus.flow.executors.task_chain.task_chain_controller
The task chain controller implementation.
Classes
StdOutErr
StdOutErrclass StdOutErr(builtins.object):
def __init__(
self,
stdout: Optional[Dict],
stderr: Optional[Dict],
timings: Optional[Dict],
) -> None: ...The stdout and stderr of a single task chain.
Both output could be None in case they don’t exist.
stdoutOptional[Dict] — The stdout.stderrOptional[Dict] — The stderr.timingsOptional[Dict] — The timings for the given task chain.
TaskChainController
TaskChainControllerclass TaskChainController(builtins.object):
def __init__(
self,
task_chain_tasks: Dict[str, List],
sites: Dict[
str,
salvus.flow.executors.task_chain.task_chain_site_config.TaskChainSiteConfig,
],
task_chain_runner_context: Optional[Callable] = None,
): ...The TaskChainController is responsible for distributing TaskChainRunners across resources.
task_chain_tasksDict[str, List] — A dictionary mapping an arbitrary job key (name) to a list of tasks, e.g. tasks that happen within one task chain.sitesDict[str, salvus.flow.executors.task_chain.task_chain_site_config.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_contextOptional[Callable] — A optional function returning a context manager that will be used in each TaskChainRunner prior to running the actual task chains.
check_status_of_task_chain_runners()
check_status_of_task_chain_runners()def check_status_of_task_chain_runners(self) -> 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.
delete_remote_files()
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.
verbosityint — The verbosity.
get_site_and_remote_path_for_task_chain()
get_site_and_remote_path_for_task_chain()def get_site_and_remote_path_for_task_chain(self, identifier: str) -> Tuple[
salvus.flow.executors.base_executor.BaseExecutor,
Union[pathlib.PurePosixPath, pathlib.PureWindowsPath],
]: ...Get the site + remote path for a given task chain.
identifierstr — Identifier of the task chain to get.
get_status_for_all_task_chains()
get_status_for_all_task_chains()def get_status_for_all_task_chains(
self,
) -> Dict[str, Optional[salvus.flow.executors.types.JobStatus]]: ...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.
get_stdouterr_for_task_chain()
get_stdouterr_for_task_chain()def get_stdouterr_for_task_chain(
self, identifier: str
) -> salvus.flow.executors.task_chain.task_chain_controller.StdOutErr: ...Get the stdout and stderr for a given task chain.
identifierstr — Identifier of the task chain to get.
launch()
launch()def launch(self) -> None: ...Launch the remote task chain runners.
run()
run()def run(self) -> 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.
update_status()
update_status()def update_status(
self,
) -> Dict[str, salvus.flow.executors.types.JobStatus]: ...Update the status of all task chain runners.
wait()
wait()def wait(self) -> None: ...Wait until all task chains runners have finished or failed.