Mondaic
This API reference is not for the latest stable Salvus version.

salvus.flow.executors.salvus_job

Classes

SalvusJob

class SalvusJob(salvus.flow.executors.job.Job):
    def __init__(
        self,
        site: salvus.flow.executors.base_executor.BaseExecutor,
        input_file: Union[
            str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter
        ],
        remote_input_file: Optional[
            str,
            pathlib.PurePosixPath,
            Dict,
            salvus.flow.utils.deep_setters._DeepSetter,
        ] = None,
        ranks: Optional[int, numpy.int32, numpy.int64] = None,
        wall_time_in_seconds: Optional[
            int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
        ] = None,
        job_groups: Optional[List[Union[str, salvus.flow.db.JobGroup]]] = None,
        is_debug_job: bool = False,
        verbosity: int = 1,
        jobname: Optional[str] = None,
        initialize_on_site: bool = True,
        rundir_root: Optional[
            str, pathlib.Path, pathlib.PurePosixPath, pathlib.PureWindowsPath
        ] = None,
        tmpdir_root: Optional[
            str, pathlib.Path, pathlib.PurePosixPath, pathlib.PureWindowsPath
        ] = None,
        working_dir: Optional[
            str, pathlib.Path, pathlib.PurePosixPath, pathlib.PureWindowsPath
        ] = None,
        skip_schema_validation: bool = False,
        db_command_line_call: Optional[str] = None,
        allow_uninitialized_site: bool = False,
        precomputed_remote_file_hashes: Optional[Dict] = None,
        parsed_input_file: Optional[
            Tuple[Dict, Dict, Dict, Optional[pathlib.Path]]
        ] = None,
    ): ...

A class representing a concrete salvus job associated with a certain site.

All interactions like updating the job or persisting some things to the database should flow over this class.

Parameters
  • site salvus.flow.executors.base_executor.BaseExecutor — The site this job is to be run at.
  • input_file Union[str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter] — Salvus input file.
  • remote_input_file Optional[str, pathlib.PurePosixPath, Dict, salvus.flow.utils.deep_setters._DeepSetter] — The input file as seen by the remote site.
  • ranks Optional[int, numpy.int32, numpy.int64] — Number of ranks to use.
  • wall_time_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — Wall time in seconds for the job. Not required for jobs on all sites.
  • job_groups Optional[List[Union[str, salvus.flow.db.JobGroup]]] — List of job groups names this job should be part of.
  • is_debug_job bool — True if this job should be executed in the fast/debug queue if the site has one.
  • verbosity int — Verbosity level.
  • jobname Optional[str] — Name of the job. A random one will be created if none is given.
  • initialize_on_site bool — Initialize on site. Should be true in most cases. False is useful for creating a Job object from the database.
  • rundir_root Optional[str, pathlib.Path, pathlib.PurePosixPath, pathlib.PureWindowsPath] — Specify what parent folder the run directory will be created in. Defaults to the directory in the site config.
  • tmpdir_root Optional[str, pathlib.Path, pathlib.PurePosixPath, pathlib.PureWindowsPath] — Specify what parent folder the tmp directory will be created in. Defaults to the directory in the site config.
  • working_dir Optional[str, pathlib.Path, pathlib.PurePosixPath, pathlib.PureWindowsPath] — Working directory. If not given it will be set to the job directory. Should only be set if the job is part of a job array.
  • skip_schema_validation bool — Optionally skip the schema validation. Only use when reading jobs from the database and for site initialization.
  • db_command_line_call Optional[str] — Command line call of this job for the database.
  • allow_uninitialized_site bool — Allows an uninitialized site - intended to only be used for the job that initializes the site.
  • precomputed_remote_file_hashes Optional[Dict] — List of potentially precomputed hashes for remote files. If given they do not need to be computed for files on the remote site.
  • parsed_input_file Optional[Tuple[Dict, Dict, Dict, Optional[pathlib.Path]]] — Can pass an already parsed input file (parsed with site_utils.parse_salvus_input) so it does not happen again. Probably only useful for job arrays.
Attributes
db salvus.flow.db.DB

The currently used database.

full_name str

Returns the full name of a job as JOB_NAME@SITE_NAME.

meta_json_path Union[pathlib.PurePosixPath, pathlib.PureWindowsPath]

Path of the job’s meta JSON file.

ranks int

Return the maximum number of ranks of all the individual commands.

salvus_progress_file_path Union[pathlib.PurePosixPath, pathlib.PureWindowsPath]

Path of the job’s progress JSON file.

stderr str

Return the stderr of the job as a string.

stderr_path Union[pathlib.PurePosixPath, pathlib.PureWindowsPath]

The stderr path of the job.

stdout str

Return the stdout of the job as a string.

stdout_path Union[pathlib.PurePosixPath, pathlib.PureWindowsPath]

The stdout path of the job.

Methods
cancel()
def cancel(self, verbosity: int = 1) -> None: ...

Cancel job.

Parameters
  • verbosity int — Verbosity level.
Returns None
copy_output()
def copy_output(
    self,
    destination: Union[str, pathlib.Path],
    get_all: bool = False,
    verbosity: int = 1,
    allow_existing_destination_folder: bool = False,
) -> Tuple[Dict[pathlib.Path, int], bool]: ...

Copy the output of a finished job to a target directory.

Parameters
  • destination Union[str, pathlib.Path] — The destination folder. Must not yet exist.
  • get_all bool — Also get the large output files, if any.
  • verbosity int — Verbosity level.
  • allow_existing_destination_folder bool — Allow that the destination already exists. Files might be overwritten.
Returns Tuple[Dict[pathlib.Path, int], bool] — Returns a dictionary with local files and their size in bytes and a flag if there are additional (large) files on the remote site
create_random_job_name()
def create_random_job_name(prefix: str = "job") -> str: ...

Create a random name.

Parameters
  • prefix str — Prefix to give some structure to the randomness.
Returns str
delete()
def delete(
    self, verbosity: int = 1, allow_directory_deletion_failure: bool = False
) -> None: ...

Delete all data for a given job.

Will delete all remote files and the internal database entry.

Parameters
  • verbosity int — Verbosity level.
  • allow_directory_deletion_failure bool — If True this function will still return successfully (but raise a warning) if the deletion of the remote Salvus managed files failed for some reason. This can easily happen on for example networked file systems.
Returns None
get_as_event()
def get_as_event(self) -> salvus.flow.collections.event_data.EventData: ...

Return the receivers of this simulation as an EventData object.

Can only be called if the output data of this job has been downloaded to the local machine.

Returns salvus.flow.collections.event_data.EventData
get_current_progress()
def get_current_progress(self) -> Optional[Dict]: ...

Gets the contents of the contents of the Salvus progress JSON file.

Returns None if it could not find the file.

Returns Optional[Dict]
get_diagnosis_string()
def get_diagnosis_string(self) -> str: ...

Get a diagnostic string with the stdout/stderr from the job.

Useful for trying to figure out why a job fails.

Returns str
get_output_files()
def get_output_files(self, get_all: bool = False) -> Tuple[
    Dict[
        Union[str, Tuple],
        Union[pathlib.PurePosixPath, pathlib.PureWindowsPath],
    ],
    bool,
]: ...

Get a dictionary containing the output files.

The keys are either "stderr", "stdout", or a tuple denoting the path of the file in the input file. The values the full paths. This does parse the input.toml file to figure out what is there.

>>> j.get_output_files()
{'stderr': PurePosixPath('run/job_1905231521_8d3f5/stderr'),
 'stdout': PurePosixPath('run/job_1905231521_8d3f5/stdout'),
 ('output', 'meta_data', 'meta_json_filename'):
    PurePosixPath('run/job_1905231521_8d3f5/output/meta.json')}
Parameters
  • get_all bool — Also get the all outputs designated as large. This will usually be volumetric and/or boundary output.
Returns Tuple[Dict[Union[str, Tuple], Union[pathlib.PurePosixPath, pathlib.PureWindowsPath]], bool] — Returns a dictionary with a tuple of the file keys in the json schema and their paths and a flag if there are additional (large) file on the remote site
get_status_from_db()
def get_status_from_db(self) -> salvus.flow.executors.types.JobStatus: ...

Returns the current status of the Job from the database. Please not that this might not be up-to-date.

Returns salvus.flow.executors.types.JobStatus
launch()
def launch(
    self, verbosity: int = 1, interactive_tty: bool = False
) -> None: ...

Launch the job.

Parameters
  • verbosity int — Verbosity level.
  • interactive_tty bool — Launch the job in an interactive terminal. Useful for debugging purposes. Only available for local sites.
Returns None
set_status_in_db()
def set_status_in_db(
    self, status: salvus.flow.executors.types.JobStatus
) -> None: ...

Set the status in SalvusFlow’s database.

Parameters
  • status salvus.flow.executors.types.JobStatus — The status to be set.
Returns None
tail_stdout()
def tail_stdout(self) -> None: ...

tail -f the stdout of the job.

This will launch a remote tail command and it only suitable for interactive use.

Returns None
update_status()
def update_status(
    self,
    progress_callback: Optional[Callable[[Dict], NoneType]] = None,
    force_update: bool = False,
) -> salvus.flow.executors.types.JobStatus: ...

Updates the status of a given job and persist it in the database.

(1) It will first try to find and parse the progress.json file that is written to during Salvus runs. This is retried up to 20 times in case of some spurious failure like a network issue or by chance parsing it right when it is being written by Salvus.

(2a) If the progress file does not exist: fall back to the parent class’ logic and get the status of the process: * If not finished: accept and return. * If finished: parse meta.json, persist in DB, and return.

(2b) If the progress file exists: * If status != “finished”: Consider it to be running, persist and return. * If status == “finished”: Parse meta.json file, persist in DB and return.

Note that the last part of both branches is identical.

Unfortunately a somewhat complex function but it should be fairly resilient against race conditions.

Parameters
  • progress_callback Optional[Callable[[Dict], NoneType]] — Callback function that is called whenever something happens.
  • force_update bool — Jobs that are not running and not pending will by default only query the local job database and not check the remote site. In most cases this is correct as the jobs are done with and will not change again. Setting this to True will force an update nonetheless. Uses for this are for example if a job update fails for example due to a temporary network problem.
Returns salvus.flow.executors.types.JobStatus
wait()
def wait(
    self,
    poll_interval_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    timeout_in_seconds: Optional[float] = None,
    verbosity: int = 1,
    progress_callback: Optional[Callable[[Union[Dict, str]], NoneType]] = None,
) -> salvus.flow.executors.types.JobStatus: ...

Wait for the job to finish.

Parameters
  • poll_interval_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — If given, the site will be queried with this interval. Defaults to the site’s default.
  • timeout_in_seconds Optional[float] — Time in seconds after which this function will return, independent of the status.
  • verbosity int — The verbosity level.
  • progress_callback Optional[Callable[[Union[Dict, str]], NoneType]] — Optional progress callback function. Will be called if there are any status updates. Will currently only be called with the Salvus status file contents.
Returns salvus.flow.executors.types.JobStatus