dpdispatcher package

dpdispatcher package#

Public interface for configuring and running DPDispatcher submissions.

The top-level package exports the core configuration and runtime objects. Batch systems and execution contexts register themselves when the package is imported, so Machine can construct the requested backend from configuration.

class dpdispatcher.Job(job_task_list: list[Task], *, resources: Resources, machine: Machine | None = None)[source]#

Bases: object

Represent one scheduler job generated from a group of tasks.

Applications normally let Submission create jobs. A job owns a resource request, generates scheduler scripts through its machine, and stores scheduler ID, state, and retry information for recovery.

Parameters:

job_task_listlist of Task: Tasks grouped into this scheduler job.
resourcesResources: Resource request copied from the parent submission.
machineMachine, optional: Backend used to generate, submit, and monitor the job.

Methods

`deserialize`(job_dict[, machine])	Reconstruct a job and its tasks from serialized state.
`get_hash`()	Return the stable hash used as this job's identifier.
`get_job_state`()	Query the backend and update this job and its unfinished tasks.
`get_last_error_message`()	Get last error message when the job is terminated.
`get_scheduler_name`(max_length, *[, ...])	Return a portable task-derived name for a one-task scheduler job.
`handle_unexpected_job_state`(*[, ...])	Submit or retry a job according to its current state.
`job_to_json`()	Write current job state to the execution root as JSON.
`register_job_id`(job_id)	Store the identifier returned by the scheduler.
`serialize`([if_static])	Return a hash-keyed, JSON-compatible representation of the job.
`submit_job`()	Submit the job through its machine and update its local state.

classmethod deserialize(job_dict: dict[str, Any], machine: Machine | None = None) → Job[source]#

Reconstruct a job and its tasks from serialized state.

Parameters:

job_dictdict: Single-entry mapping from job hash to configuration and runtime data.
machineMachine, optional: Machine to bind to the reconstructed job.

Returns:

Job: Reconstructed job.

get_hash() → str[source]#: Return the stable hash used as this job’s identifier.

get_job_state() → None[source]#

Query the backend and update this job and its unfinished tasks.

Notes

This method does not submit or retry the job.

get_last_error_message() → str | None[source]#: Get last error message when the job is terminated.

get_scheduler_name(max_length: int, *, require_alpha_prefix: bool = False) → str | None[source]#

Return a portable task-derived name for a one-task scheduler job.

A grouped job deliberately returns None so each backend keeps its established hash-based/default name instead of presenting one task as if it represented the whole group. User-provided names are reduced to a conservative ASCII subset accepted by Slurm, PBS, LSF, and SGE. Truncation retains a hash suffix so long names that share a prefix remain distinguishable.

handle_unexpected_job_state(*, continue_on_failure: bool = False) → None[source]#

Submit or retry a job according to its current state.

Retry exhaustion remains fail-fast by default. Callers that need to monitor sibling jobs must explicitly opt in with continue_on_failure.

job_to_json() → None[source]#: Write current job state to the execution root as JSON.

register_job_id(job_id: str | int) → None[source]#: Store the identifier returned by the scheduler.

serialize(if_static: bool = False) → dict[str, Any][source]#

Return a hash-keyed, JSON-compatible representation of the job.

Parameters:

if_staticbool, default=False: Exclude job ID, state, and failure count when true.

Returns:

dict: Mapping from the deterministic job hash to job data.

submit_job() → None[source]#: Submit the job through its machine and update its local state.

class dpdispatcher.Machine(*args: Any, **kwargs: Any)[source]#

Bases: object

Generate, submit, and monitor jobs on a selected batch system.

Machine acts as a factory when instantiated with batch_type. A machine delegates filesystem and command operations to its bound BaseContext while its concrete subclass implements scheduler-specific submission and status handling.

Parameters:

contextBaseContext, optional: Existing execution context. When omitted, context_type and the root and profile arguments are used to construct one.
retry_countint, default=3: Number of scheduler failures allowed before a job is reported as failed.

Methods

`arginfo`()	Build the dargs schema for machine and context configuration.
`bind_context`(context)	Bind the execution context used for files and remote commands.
`check_finish_tag`(job)	Return whether the success marker for a job is present.
`check_if_recover`(submission)	Return whether serialized state exists for a remote submission.
`check_status`(job)	Query the scheduler and return the current status of a job.
`default_resources`(res)	Return backend defaults for an incomplete resource specification.
`deserialize`(machine_dict)	Reconstruct a machine from its serialized dictionary.
`do_submit`(job)	Submit a single job, assuming that no job is running there.
`gen_command_env_cuda_devices`(resources)	Assign a GPU to the next task when automatic GPU mapping is enabled.
`gen_local_script`(job)	Generate a local staging script for cloud backends.
`gen_script`(job)	Generate the complete scheduler submission script for a job.
`gen_script_command`(job)	Generate task commands, logging redirection, and completion tags.
`gen_script_custom_flags_lines`(job)	Render user-provided scheduler directives as script lines.
`gen_script_end`(job)	Generate job finalization, failure checks, and append commands.
`gen_script_env`(job)	Generate environment setup shared by every task in a job.
`gen_script_header`(job)	Generate scheduler directives and the script shebang for a job.
`gen_script_run_command`(job)	Return the command that sources the generated per-task script.
`gen_script_wait`(resources)	Generate synchronization commands for the configured parallelism.
`get_exit_code`(job)	Get exit code of the job.
`get_job_error`(job)	Return a text error diagnostic for a job, if available.
`kill`(job)	Kill the job.
`load_from_dict`(machine_dict[, allow_ref])	Load a Machine from a dict.
`load_from_json`(json_path)	Load a machine configuration from a JSON file.
`load_from_yaml`(yaml_path)	Load a machine configuration from a YAML file.
`resources_arginfo`()	Generate the resources arginfo.
`resources_subfields`()	Generate the resources subfields.
`serialize`([if_empty_remote_profile])	Return a normalized dictionary representation of the machine.
`sub_script_cmd`(res)	Return the legacy scheduler launch command for resources.
`sub_script_head`(res)	Return the legacy scheduler header for a resource specification.

alias: tuple[str, ...] = ()#

classmethod arginfo() → Argument[source]#: Build the dargs schema for machine and context configuration.

bind_context(context: BaseContext) → None[source]#: Bind the execution context used for files and remote commands.

abstract check_finish_tag(job: Job) → bool[source]#: Return whether the success marker for a job is present.

check_if_recover(submission: Submission) → bool[source]#: Return whether serialized state exists for a remote submission.

abstract check_status(job: Job) → JobStatus[source]#: Query the scheduler and return the current status of a job.

default_resources(res: Resources) → Resources[source]#: Return backend defaults for an incomplete resource specification.

classmethod deserialize(machine_dict: dict[str, Any]) → Machine[source]#: Reconstruct a machine from its serialized dictionary.

abstract do_submit(job: Job) → str | int[source]#: Submit a single job, assuming that no job is running there.

gen_command_env_cuda_devices(resources: Resources) → str[source]#: Assign a GPU to the next task when automatic GPU mapping is enabled.

gen_local_script(job: Job) → str[source]#

Generate a local staging script for cloud backends.

Only cloud machine implementations support this operation; the base declaration keeps their context interface type-safe.

gen_script(job: Job) → str[source]#: Generate the complete scheduler submission script for a job.

gen_script_command(job: Job) → str[source]#: Generate task commands, logging redirection, and completion tags.

gen_script_custom_flags_lines(job: Job) → str[source]#: Render user-provided scheduler directives as script lines.

gen_script_end(job: Job) → str[source]#: Generate job finalization, failure checks, and append commands.

gen_script_env(job: Job) → str[source]#: Generate environment setup shared by every task in a job.

abstract gen_script_header(job: Job) → str[source]#: Generate scheduler directives and the script shebang for a job.

gen_script_run_command(job: Job) → str[source]#: Return the command that sources the generated per-task script.

gen_script_wait(resources: Resources) → str[source]#: Generate synchronization commands for the configured parallelism.

get_exit_code(job: Job) → int[source]#

Get exit code of the job.

Parameters:

jobJob: job

get_job_error(job: Job) → str | None[source]#

Return a text error diagnostic for a job, if available.

HDFS contexts expose command output as bytes while other contexts return text. Decode byte-valued responses here so callers receive the same text interface regardless of the execution backend.

kill(job: Job) → None[source]#

Kill the job.

If not implemented, pass and let the user manually kill it.

Parameters:

jobJob: job

classmethod load_from_dict(machine_dict: dict[str, Any], allow_ref: bool = False) → Machine[source]#

Load a Machine from a dict.

Parameters:

machine_dictdict: Machine configuration dict.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_json(json_path: str) → Machine[source]#: Load a machine configuration from a JSON file.

classmethod load_from_yaml(yaml_path: str) → Machine[source]#: Load a machine configuration from a YAML file.

options = {'Bohrium', 'DistributedShell', 'Fugaku', 'JH_UniScheduler', 'LSF', 'OpenAPI', 'PBS', 'SGE', 'Shell', 'Slurm', 'SlurmJobArray', 'Torque'}#

classmethod resources_arginfo() → Argument[source]#

Generate the resources arginfo.

Returns:

Argument: resources arginfo

classmethod resources_subfields() → list[Argument][source]#

Generate the resources subfields.

Returns:

list[Argument]: resources subfields

serialize(if_empty_remote_profile: bool = False) → dict[str, Any][source]#: Return a normalized dictionary representation of the machine.

sub_script_cmd(res: Resources) → str[source]#: Return the legacy scheduler launch command for resources.

sub_script_head(res: Resources) → str[source]#: Return the legacy scheduler header for a resource specification.

subclasses_dict = {'Bohrium': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'DistributedShell': <class 'dpdispatcher.machines.distributed_shell.DistributedShell'>, 'DpCloudServer': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'Fugaku': <class 'dpdispatcher.machines.fugaku.Fugaku'>, 'JH_UniScheduler': <class 'dpdispatcher.machines.JH_UniScheduler.JH_UniScheduler'>, 'LSF': <class 'dpdispatcher.machines.lsf.LSF'>, 'Lebesgue': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'OpenAPI': <class 'dpdispatcher.machines.openapi.OpenAPI'>, 'PBS': <class 'dpdispatcher.machines.pbs.PBS'>, 'SGE': <class 'dpdispatcher.machines.pbs.SGE'>, 'Shell': <class 'dpdispatcher.machines.shell.Shell'>, 'Slurm': <class 'dpdispatcher.machines.slurm.Slurm'>, 'SlurmJobArray': <class 'dpdispatcher.machines.slurm.SlurmJobArray'>, 'Torque': <class 'dpdispatcher.machines.pbs.Torque'>, 'bohrium': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'distributedshell': <class 'dpdispatcher.machines.distributed_shell.DistributedShell'>, 'dpcloudserver': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'fugaku': <class 'dpdispatcher.machines.fugaku.Fugaku'>, 'jh_unischeduler': <class 'dpdispatcher.machines.JH_UniScheduler.JH_UniScheduler'>, 'lebesgue': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'lsf': <class 'dpdispatcher.machines.lsf.LSF'>, 'openapi': <class 'dpdispatcher.machines.openapi.OpenAPI'>, 'pbs': <class 'dpdispatcher.machines.pbs.PBS'>, 'sge': <class 'dpdispatcher.machines.pbs.SGE'>, 'shell': <class 'dpdispatcher.machines.shell.Shell'>, 'slurm': <class 'dpdispatcher.machines.slurm.Slurm'>, 'slurmjobarray': <class 'dpdispatcher.machines.slurm.SlurmJobArray'>, 'torque': <class 'dpdispatcher.machines.pbs.Torque'>}#

class dpdispatcher.Resources(number_node: int, cpu_per_node: int, gpu_per_node: int, queue_name: str, group_size: int, *, custom_flags: Sequence[str] | None = None, strategy: dict[str, Any] | None = None, para_deg: int = 1, module_unload_list: Sequence[str] | None = None, module_purge: bool = False, module_list: Sequence[str] | None = None, source_list: Sequence[str] | None = None, envs: dict[str, Any] | None = None, prepend_script: Sequence[str] | None = None, append_script: Sequence[str] | None = None, wait_time: int = 0, **kwargs: Any)[source]#

Bases: object

Describe the resources and execution strategy for generated jobs.

Parameters:

number_nodeint: Number of nodes requested for each generated job.
cpu_per_nodeint: Number of CPUs requested on each node.
gpu_per_nodeint: Number of GPUs requested on each node.
queue_namestr: Queue or partition name passed to the batch backend.
group_sizeint: Maximum number of tasks grouped into one scheduler job. Zero groups all tasks into one job.
custom_flagslist of str, optional: Extra scheduler directives inserted into the generated script header.
strategydict, optional: Script-generation strategy. Recognized keys include if_cuda_multi_devices, ratio_unfinished, and customized_script_header_template_file.
para_degint, default=1: Number of task commands run concurrently inside one generated job.
module_unload_listlist of str, optional: Environment modules to unload before task execution.
module_purgebool, default=False: Whether to purge loaded environment modules first.
module_listlist of str, optional: Environment modules to load before task execution.
source_listlist of str, optional: Shell files to source before task execution.
envsdict, optional: Environment variables to export before task execution.
prepend_scriptlist of str, optional: Shell lines inserted before task commands.
append_scriptlist of str, optional: Shell lines inserted after all task commands finish.
wait_timeint, default=0: Delay in seconds after each job submission or resubmission.
**kwargs: Backend-specific resource options, stored in Resources.kwargs.

Methods

`arginfo`([detail_kwargs])	Build the dargs schema for common and backend-specific resources.
`deserialize`(resources_dict)	Reconstruct a resource request from a serialized dictionary.
`load_from_dict`(resources_dict[, allow_ref])	Load Resources from a dict.
`load_from_json`(json_file)	Load and validate a resource request from a JSON file.
`load_from_yaml`(yaml_file)	Load and validate a resource request from a YAML file.
`serialize`()	Return the resource request as a JSON-compatible dictionary.

static arginfo(detail_kwargs: bool = True) → Argument[source]#: Build the dargs schema for common and backend-specific resources.

classmethod deserialize(resources_dict: dict[str, Any]) → Resources[source]#: Reconstruct a resource request from a serialized dictionary.

classmethod load_from_dict(resources_dict: dict[str, Any], allow_ref: bool = False) → Resources[source]#

Load Resources from a dict.

Parameters:

resources_dictdict: Resources configuration dict.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_json(json_file: str) → Resources[source]#: Load and validate a resource request from a JSON file.

classmethod load_from_yaml(yaml_file: str) → Resources[source]#: Load and validate a resource request from a YAML file.

serialize() → dict[str, Any][source]#: Return the resource request as a JSON-compatible dictionary.

class dpdispatcher.Submission(work_base: str, machine: Machine | None = None, resources: Resources | None = None, forward_common_files: list[str] | None = None, backward_common_files: list[str] | None = None, *, task_list: list[Task] | None = None, previous_submission_hash: str | None = None, continue_on_failure: bool = False)[source]#

Bases: object

Coordinate a collection of tasks that share a working directory.

A submission groups tasks into scheduler jobs, stages common files, monitors execution, downloads declared results, and records state for recovery.

Parameters:

work_basepath-like: Local base directory containing all task working directories.
machineMachine, optional: Batch backend and execution context. It may be bound later with bind_machine().
resourcesResources, optional: Resource request copied into every generated job.
forward_common_fileslist of path-like, optional: Files shared by all tasks and staged before execution.
backward_common_fileslist of path-like, optional: Shared result files downloaded after execution.
task_listlist of Task, optional: Tasks to register when the submission is created.
previous_submission_hashstr, optional: SHA-1 hash of a compatible prior submission whose finished-task state should be reused after a resource-only change.
continue_on_failurebool, default=False: Continue monitoring other jobs after one job exhausts its retries. The default preserves the historical fail-fast behavior.

Methods

`async_run_submission`(**kwargs)	Run `run_submission()` in an executor.
`bind_machine`(machine, *[, bind_context])	Bind a machine and initialize submission-specific context paths.
`check_all_finished`()	Return whether every generated job has finished successfully.
`check_ratio_unfinished`(ratio_unfinished)	Return whether the allowed unfinished-task threshold is satisfied.
`clean_jobs`()	Remove remote working data and the local recovery record.
`deserialize`(submission_dict[, machine, ...])	Reconstruct a submission from serialized state.
`download_jobs`([include_failed])	Download selected task outputs, optionally including failed jobs.
`failed_jobs`()	Return jobs that exhausted retries and reached a terminal failure.
`generate_jobs`()	Generate jobs after tasks are registered.
`get_hash`()	Return the stable hash of the submission's static configuration.
`handle_unexpected_submission_state`(*[, ...])	Submit unsubmitted jobs and retry unexpectedly terminated jobs.
`raise_for_failed_jobs`(*[, continue_on_failure])	Report all terminal failures after other jobs and downloads finish.
`register_task`(task)	Append one task before jobs have been generated.
`register_task_list`(task_list)	Append multiple tasks before jobs have been generated.
`remove_unfinished_tasks`()	Stop unfinished work while preserving durable failed records.
`run_submission`(*[, dry_run, exit_on_submit, ...])	Execute the submission and monitor it until completion.
`serialize`([if_static])	Return a JSON-compatible representation of the submission.
`submission_from_json`([json_file_name])	Load a submission, including machine state, from a local JSON file.
`submission_to_json`()	Write current submission state to the execution root as JSON.
`try_download_error_info`()	Download error diagnostic files for failed/terminated jobs.
`try_download_result`()	Download results, retrying transient failures for up to 24 hours.
`try_recover_from_json`()	Restore compatible job state from the remote submission JSON file.
`update_submission_state`()	Refresh every unfinished job's state from its machine backend.
`upload_jobs`()	Upload submission inputs through the bound context.

async async_run_submission(**kwargs: Any) → dict[str, Any][source]#

Run run_submission() in an executor.

Cleanup defaults to false for asynchronous submissions so concurrent work does not remove shared context data unexpectedly. Explicitly pass clean=True only when each submission has an isolated execution root.

Examples

>>> import asyncio
>>> async def run_all(submissions):
...     return await asyncio.gather(
...         *(
...             submission.async_run_submission(check_interval=2)
...             for submission in submissions
...         )
...     )

bind_machine(machine: Machine | None, *, bind_context: bool = True) → Submission[source]#

Bind a machine and initialize submission-specific context paths.

Parameters:

machineMachine or None: Machine to use for generated jobs and file operations.

Returns:

Submission: This submission, for convenient chained configuration.

check_all_finished() → bool[source]#

Return whether every generated job has finished successfully.

Notes

This method does not submit, retry, or otherwise change job states.

check_ratio_unfinished(ratio_unfinished: float) → bool[source]#

Return whether the allowed unfinished-task threshold is satisfied.

Parameters:

ratio_unfinishedfloat: Maximum fraction of tasks that may remain unfinished.

Returns:

bool: True when the finished fraction is at least 1 - ratio_unfinished.

clean_jobs() → None[source]#: Remove remote working data and the local recovery record.

classmethod deserialize(submission_dict: dict[str, Any], machine: Machine | None = None, *, bind_context: bool = True) → Submission[source]#

Reconstruct a submission from serialized state.

Parameters:

submission_dictdict: Serialized submission configuration and job state.
machineMachine, optional: Machine to bind instead of reconstructing the serialized machine.

Returns:

Submission: Reconstructed submission.

download_jobs(include_failed: bool = False) → None[source]#

Download selected task outputs, optionally including failed jobs.

Normal result downloads omit failed jobs so missing outputs do not mask successful work. Explicit terminated-log requests need the original failed tasks, however, and set include_failed=True to bypass that success-only filtering.

failed_jobs() → list[Job][source]#: Return jobs that exhausted retries and reached a terminal failure.

generate_jobs() → None[source]#

Generate jobs after tasks are registered.

Tasks are shuffled with a fixed seed before grouping to distribute task cost while preserving deterministic job hashes for recovery. Each job contains at most resources.group_size tasks; zero groups all tasks.

get_hash() → str[source]#: Return the stable hash of the submission’s static configuration.

handle_unexpected_submission_state(*, continue_on_failure: bool = False) → None[source]#

Submit unsubmitted jobs and retry unexpectedly terminated jobs.

Unknown states and exhausted retries are persisted for recovery before the error is propagated.

raise_for_failed_jobs(*, continue_on_failure: bool = True) → None[source]#: Report all terminal failures after other jobs and downloads finish.

register_task(task: Task) → None[source]#: Append one task before jobs have been generated.

register_task_list(task_list: list[Task]) → None[source]#: Append multiple tasks before jobs have been generated.

remove_unfinished_tasks() → None[source]#

Stop unfinished work while preserving durable failed records.

A failed job is terminal evidence that must remain available to raise_for_failed_jobs and post-mortem download handling. Only non-terminal jobs are converted to finished after the ratio threshold is reached; failed jobs and their failed tasks stay in the submission.

run_submission(*, dry_run: bool = False, exit_on_submit: bool = False, clean: bool | str = True, check_interval: int = 30, continue_on_failure: bool | None = None) → dict[str, Any][source]#

Execute the submission and monitor it until completion.

The lifecycle recovers compatible state, stages files, submits or retries jobs, polls their status, downloads declared results, persists recovery state, and optionally cleans the execution directory.

Parameters:

dry_runbool, default=False: Upload inputs and generated scripts without submitting jobs.
exit_on_submitbool, default=False: Return after jobs have been submitted instead of waiting for them.
cleanbool, default=True: Remove the submission-specific execution directory after download.
check_intervalint or float, default=30: Seconds between scheduler status checks.
continue_on_failurebool, optional: Continue monitoring remaining jobs after retry exhaustion. If omitted, use the policy stored on this submission (which defaults to False). An explicit value overrides the persisted policy for this run.

Returns:

dict: Serialized submission state at the point this method returns.

serialize(if_static: bool = False) → dict[str, Any][source]#

Return a JSON-compatible representation of the submission.

Parameters:

if_staticbool, default=False: Exclude job IDs, states, failure counts, and runtime failure policy when true. The policy is intentionally excluded from the static identity so enabling continuation does not change job hashes.

Returns:

dict: Submission configuration and, unless excluded, current runtime state.

classmethod submission_from_json(json_file_name: str = 'submission.json') → Submission[source]#: Load a submission, including machine state, from a local JSON file.

submission_to_json() → None[source]#: Write current submission state to the execution root as JSON.

try_download_error_info() → None[source]#

Download error diagnostic files for failed/terminated jobs.

For each job that did not finish successfully, attempts to download the {job_hash}_last_err_file from the remote root to the local root. This preserves error diagnostics even when clean=True deletes the remote workdir afterward.

The error file contains the last 1000 bytes of stderr from the most recently failed task in the job, written by the generated bash script.

try_download_result() → bool[source]#: Download results, retrying transient failures for up to 24 hours.

try_recover_from_json() → None[source]#: Restore compatible job state from the remote submission JSON file.

update_submission_state() → None[source]#

Refresh every unfinished job’s state from its machine backend.

Notes

This method only queries state. It does not submit or retry jobs.

upload_jobs() → None[source]#: Upload submission inputs through the bound context.

class dpdispatcher.Task(command: str, task_work_path: str, forward_files: Sequence[str] | None = None, backward_files: Sequence[str] | None = None, outlog: str | None = 'log', errlog: str | None = 'err', task_name: str | None = None)[source]#

Bases: object

Represent a sequential command and its staged files.

A task records the files it depends on and the results to transfer back.

Parameters:

commandstr: Shell command to execute.
task_work_pathpath-like: Working directory relative to the submission’s work_base.
forward_fileslist of path-like, optional: Task-specific input files staged before execution.
backward_fileslist of path-like, optional: Task-specific result files downloaded after execution.
outlogstr or None, default=”log”: File that receives standard output, or None to leave it attached.
errlogstr or None, default=”err”: File that receives standard error, or None to leave it attached.
task_namestr or None, optional: Scheduler-visible name when this task is submitted as a one-task job.

Methods

`arginfo`()	Build the dargs schema for task configuration.
`deserialize`(task_dict)	Reconstruct a task from a serialized dictionary.
`get_hash`()	Return the stable hash of the task configuration.
`get_task_state`(context)	Get the task state by checking the tag file.
`has_finished_tag`(context)	Return whether the remote completion tag for this task exists.
`load_from_dict`(task_dict[, allow_ref])	Load a Task from a dict.
`load_from_json`(json_file[, allow_ref])	Load a Task from a JSON file.
`load_from_yaml`(yaml_file[, allow_ref])	Load a Task from a YAML file.
`serialize`()	Return the task configuration as a JSON-compatible dictionary.

static arginfo() → Argument[source]#: Build the dargs schema for task configuration.

classmethod deserialize(task_dict: dict[str, Any]) → Task[source]#

Reconstruct a task from a serialized dictionary.

Parameters:

task_dictdict: Task configuration.

Returns:

Task: Reconstructed task.

get_hash() → str[source]#: Return the stable hash of the task configuration.

get_task_state(context: BaseContext) → None[source]#

Get the task state by checking the tag file.

Parameters:

contextContext: the context of the task

has_finished_tag(context: BaseContext) → bool[source]#: Return whether the remote completion tag for this task exists.

classmethod load_from_dict(task_dict: dict, allow_ref: bool = False) → Task[source]#

Load a Task from a dict.

Parameters:

task_dictdict: Task configuration dict.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_json(json_file: str, allow_ref: bool = False) → Task[source]#

Load a Task from a JSON file.

Parameters:

json_filestr: Path to task JSON file.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_yaml(yaml_file: str, allow_ref: bool = False) → Task[source]#

Load a Task from a YAML file.

Parameters:

yaml_filestr: Path to task YAML file.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

serialize() → dict[str, Any][source]#: Return the task configuration as a JSON-compatible dictionary.

Subpackages#

Submodules#

dpdispatcher.arginfo module#

Expose dargs schemas for machine, resource, and task configuration.

dpdispatcher.base_context module#

Define the execution-context interface used by machine backends.

class dpdispatcher.base_context.BaseContext(*args: Any, **kwargs: Any)[source]#

Bases: object

Define file transfer and command execution for an environment.

BaseContext acts as both an abstract interface and a factory. Calling it with context_type selects a registered subclass such as SSHContext or LazyLocalContext. Concrete contexts must implement file transfer, cleanup, file access, and blocking command execution.

Methods

`bind_submission`(submission)	Bind a submission and its derived working paths to this context.
`block_call`(cmd)	Run command with arguments.
`block_checkcall`(cmd[, asynchronously])	Run command with arguments.
`check_file_exists`(fname)	Return whether a file exists in the active execution root.
`check_finish`(proc)	Return whether an asynchronous process has finished.
`clean`()	Remove the submission-specific execution directory.
`download`(submission[, check_exists, ...])	Download declared result files from the execution root.
`load_from_dict`(context_dict)	Create a registered context from a machine configuration mapping.
`machine_arginfo`()	Generate the machine arginfo.
`machine_subfields`()	Generate the machine subfields.
`read_file`(fname)	Read text from a file relative to the execution root.
`upload`(submission)	Upload all files required by a submission to the execution root.
`write_file`(fname, write_str)	Write text to a file relative to the execution root.
`write_local_file`(fname, write_str)	Write a backend-local staging file when the context supports it.

alias: tuple[str, ...] = ()#

bind_submission(submission: Submission) → None[source]#: Bind a submission and its derived working paths to this context.

abstract block_call(cmd: str) → tuple[int, Any, Any, Any][source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.

Returns:

exit_status: exit code
stdin: standard inout
stdout: standard output
stderr: standard error

block_checkcall(cmd: str, asynchronously: bool = False) → tuple[Any, Any, Any][source]#

Run command with arguments. Wait for command to complete.

Parameters:

cmdstr: The command to run.
asynchronouslybool, optional, default=False: Run command asynchronously. If True, nohup will be used to run the command.

Returns:

stdin: standard inout
stdout: standard output
stderr: standard error

Raises:

RuntimeError: when the return code is not zero

abstract check_file_exists(fname: str) → bool[source]#: Return whether a file exists in the active execution root.

check_finish(proc: Any) → Any[source]#: Return whether an asynchronous process has finished.

abstract clean() → Any[source]#: Remove the submission-specific execution directory.

create_remote_root: bool#

abstract download(submission: Submission, check_exists: bool = False, mark_failure: bool = True, back_error: bool = False) → Any[source]#: Download declared result files from the execution root.

downloads_by_job: ClassVar[bool] = False#

init_local_root: str#

init_remote_root: str | None#

last_downloaded_files: set[str]#

classmethod load_from_dict(context_dict: dict[str, Any]) → BaseContext[source]#: Create a registered context from a machine configuration mapping.

local_root: str#

machine: Machine#

classmethod machine_arginfo() → Argument[source]#

Generate the machine arginfo.

Returns:

Argument: machine arginfo

classmethod machine_subfields() → list[Argument][source]#

Generate the machine subfields.

Returns:

list[Argument]: machine subfields

options = {'BohriumContext', 'HDFSContext', 'LazyLocalContext', 'LocalContext', 'OpenAPIContext', 'SSHContext'}#

abstract read_file(fname: str) → Any[source]#: Read text from a file relative to the execution root.

remote_profile: dict[str, Any]#

remote_root: str#

subclasses_dict = {'Bohrium': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'BohriumContext': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'DpCloudServer': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'DpCloudServerContext': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'HDFS': <class 'dpdispatcher.contexts.hdfs_context.HDFSContext'>, 'HDFSContext': <class 'dpdispatcher.contexts.hdfs_context.HDFSContext'>, 'LazyLocal': <class 'dpdispatcher.contexts.lazy_local_context.LazyLocalContext'>, 'LazyLocalContext': <class 'dpdispatcher.contexts.lazy_local_context.LazyLocalContext'>, 'Lebesgue': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'LebesgueContext': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'Local': <class 'dpdispatcher.contexts.local_context.LocalContext'>, 'LocalContext': <class 'dpdispatcher.contexts.local_context.LocalContext'>, 'OpenAPI': <class 'dpdispatcher.contexts.openapi_context.OpenAPIContext'>, 'OpenAPIContext': <class 'dpdispatcher.contexts.openapi_context.OpenAPIContext'>, 'SSH': <class 'dpdispatcher.contexts.ssh_context.SSHContext'>, 'SSHContext': <class 'dpdispatcher.contexts.ssh_context.SSHContext'>, 'bohrium': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'bohriumcontext': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'dpcloudserver': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'dpcloudservercontext': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'hdfs': <class 'dpdispatcher.contexts.hdfs_context.HDFSContext'>, 'hdfscontext': <class 'dpdispatcher.contexts.hdfs_context.HDFSContext'>, 'lazylocal': <class 'dpdispatcher.contexts.lazy_local_context.LazyLocalContext'>, 'lazylocalcontext': <class 'dpdispatcher.contexts.lazy_local_context.LazyLocalContext'>, 'lebesgue': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'lebesguecontext': <class 'dpdispatcher.contexts.dp_cloud_server_context.BohriumContext'>, 'local': <class 'dpdispatcher.contexts.local_context.LocalContext'>, 'localcontext': <class 'dpdispatcher.contexts.local_context.LocalContext'>, 'openapi': <class 'dpdispatcher.contexts.openapi_context.OpenAPIContext'>, 'openapicontext': <class 'dpdispatcher.contexts.openapi_context.OpenAPIContext'>, 'ssh': <class 'dpdispatcher.contexts.ssh_context.SSHContext'>, 'sshcontext': <class 'dpdispatcher.contexts.ssh_context.SSHContext'>}#

submission: Submission#

supports_partial_job_download: ClassVar[bool] = True#

supports_task_completion_tags: ClassVar[bool] = True#

temp_local_root: str#

temp_remote_root: str#

abstract upload(submission: Submission) → None[source]#: Upload all files required by a submission to the execution root.

abstract write_file(fname: str, write_str: str) → Any[source]#: Write text to a file relative to the execution root.

write_local_file(fname: str, write_str: str) → Any[source]#: Write a backend-local staging file when the context supports it.

dpdispatcher.dlog module#

Configure the package logger for file and standard-output diagnostics.

dpdispatcher.dpdisp module#

Implement the dpdisp command-line interface.

dpdispatcher.dpdisp.main() → None[source]#: Parse command-line arguments and dispatch the selected subcommand.

dpdispatcher.dpdisp.main_parser() → ArgumentParser[source]#

Dpdispatcher commandline options argument parser.

Returns:

argparse.ArgumentParser: the argument parser

Notes

This function is used by documentation.

dpdispatcher.dpdisp.parse_args(args: list[str] | None = None) → Namespace[source]#

Dpdispatcher commandline options argument parsing.

Parameters:

argsList[str]: list of command line arguments, main purpose is testing default option None takes arguments from sys.argv

dpdispatcher.file_manager module#

Object-oriented file staging primitives used by execution contexts.

The public Task and Submission APIs intentionally still expose file names as strings. This module turns those strings into validated, deterministic manifests before a context performs any I/O. Keeping path policy and transfer bookkeeping here prevents each backend from subtly interpreting globs, directories, and missing files differently.

class dpdispatcher.file_manager.ArchiveBuilder(root: PathLike[str] | str)[source]#

Bases: object

Build deterministic zip archives from a validated file selection.

Methods

build_zip(archive_path, patterns)

Create a zip archive containing selected files and directories.

build_zip(archive_path: PathLike[str] | str, patterns: Sequence[str]) → Path[source]#: Create a zip archive containing selected files and directories.

class dpdispatcher.file_manager.AtomicTextWriter(root: PathLike[str] | str)[source]#

Bases: object

Atomically write UTF-8 text below a validated root.

Methods

write(relative_path, content)

Atomically write UTF-8 content below the configured root.

write(relative_path: str, content: str) → Path[source]#: Atomically write UTF-8 content below the configured root.

class dpdispatcher.file_manager.FileEntry(source: Path, destination: str)[source]#

Bases: object

One concrete source-to-destination transfer in a manifest.

Attributes:

is_directory: Whether the source entry is a real directory rather than a link.
is_symlink: Whether the source entry is a symbolic link.

destination: str#

property is_directory: bool#: Whether the source entry is a real directory rather than a link.

property is_symlink: bool#: Whether the source entry is a symbolic link.

source: Path#

class dpdispatcher.file_manager.FileTransfer(destination_root: PathLike[str] | str, *, symlink: bool = False, link_sources: bool = False, move: bool = False, overwrite: bool = True)[source]#

Bases: object

Apply a concrete manifest between filesystem-backed roots.

Methods

`apply`(manifest)	Apply all manifest entries, respecting move/link/overwrite policy.
`remove`(path)	Remove a file, directory, or broken symlink without following it.

apply(manifest: ResolvedManifest) → None[source]#: Apply all manifest entries, respecting move/link/overwrite policy.

static remove(path: PathLike[str] | str) → None[source]#: Remove a file, directory, or broken symlink without following it.

class dpdispatcher.file_manager.ManifestBuilder[source]#

Bases: object

Build upload/download manifests from task and common-file selections.

Attributes:

missing: Return unresolved required requests accumulated so far.

Methods

`add_directory`(destination)	Add an empty directory marker to preserve task directories.
`add_paths`(*, source_root, ...[, required, ...])	Expand patterns below a source root and append transfer entries.
`build`()	Return a deduplicated immutable snapshot of the collected entries.

add_directory(destination: str) → ManifestBuilder[source]#: Add an empty directory marker to preserve task directories.

add_paths(*, source_root: PathLike[str] | str, destination_prefix: str, patterns: Sequence[str], required: bool = True, fallback_root: PathLike[str] | str | None = None) → ManifestBuilder[source]#: Expand patterns below a source root and append transfer entries.

build() → ResolvedManifest[source]#: Return a deduplicated immutable snapshot of the collected entries.

property missing: list[MissingEntry]#: Return unresolved required requests accumulated so far.

class dpdispatcher.file_manager.MissingEntry(pattern: str, destination_prefix: str)[source]#

Bases: object

A requested path that was absent from the source and fallback roots.

Methods

failure_marker()

Return a safe, deterministic marker filename for this request.

destination_prefix: str#

failure_marker() → str[source]#: Return a safe, deterministic marker filename for this request.

pattern: str#

class dpdispatcher.file_manager.PathPolicy[source]#

Bases: object

Validate and resolve paths relative to a staging root.

User-provided task paths are intentionally lexical paths rather than arbitrary filesystem paths. Rejecting absolute paths and .. segments here prevents os.path.join(root, value) from escaping the submission directory on local, SSH, and archive-backed contexts.

Methods

`join_relative`(prefix, suffix)	Join two validated relative paths without introducing traversal.
`normalize_relative`(value, *[, allow_glob])	Return a canonical, safe path relative to a staging root.

classmethod join_relative(prefix: str, suffix: str) → str[source]#: Join two validated relative paths without introducing traversal.

static normalize_relative(value: PathLike[str] | str, *, allow_glob: bool = False) → str[source]#: Return a canonical, safe path relative to a staging root.

class dpdispatcher.file_manager.PathResolver(root: PathLike[str] | str)[source]#

Bases: object

Resolve validated relative paths and glob patterns under one root.

Methods

`expand`(pattern)	Expand one path or glob deterministically.
`relative`(value)	Return the canonical path of `value` relative to this root.
`resolve`(value, *[, allow_glob, allow_absolute])	Resolve a relative path, optionally allowing an in-root absolute path.

expand(pattern: PathLike[str] | str) → list[Path][source]#

Expand one path or glob deterministically.

glob does not report broken symlinks for a literal path, so literal entries use lexists and produce a clear error for a broken link.

relative(value: PathLike[str] | str) → str[source]#: Return the canonical path of value relative to this root.

resolve(value: PathLike[str] | str, *, allow_glob: bool = False, allow_absolute: bool = False) → Path[source]#: Resolve a relative path, optionally allowing an in-root absolute path.

class dpdispatcher.file_manager.RemoteManifestBuilder(available_paths: Sequence[str] = (), *, exists: Callable[[str], bool] | None = None, assume_literals: bool = True)[source]#

Bases: object

Build a manifest from an indexed, non-filesystem source.

SSH and similar transports cannot pass a remote path to glob on the client machine. They first index regular files below the remote root and then use this builder to apply the same destination and missing-entry policy as ManifestBuilder. Literal paths may optionally be checked through exists; leaving that callback unset preserves the historical behavior where the remote tar command reports a missing literal file when check_exists is disabled.

Methods

`add_paths`(*, source_prefix, ...[, required])	Match remote patterns against the indexed path set.
`build`()	Return a deduplicated manifest for the indexed remote paths.

add_paths(*, source_prefix: str, destination_prefix: str, patterns: Sequence[str], required: bool = True) → RemoteManifestBuilder[source]#: Match remote patterns against the indexed path set.

build() → ResolvedManifest[source]#: Return a deduplicated manifest for the indexed remote paths.

class dpdispatcher.file_manager.ResolvedManifest(entries: list[FileEntry], missing: list[MissingEntry])[source]#

Bases: object

Concrete transfers plus requests that could not be resolved.

Methods

unique()

Return a deterministic manifest with duplicate paths removed.

entries: list[FileEntry]#

missing: list[MissingEntry]#

unique() → ResolvedManifest[source]#: Return a deterministic manifest with duplicate paths removed.

class dpdispatcher.file_manager.SafeArchiveExtractor(destination: PathLike[str] | str)[source]#

Bases: object

Extract tar/zip archives without allowing paths outside a destination.

Methods

`extract_tar`(archive)	Safely extract a tar archive and return extracted file names.
`extract_zip`(archive)	Safely extract a zip archive and return extracted file names.

extract_tar(archive: PathLike[str] | str) → set[str][source]#: Safely extract a tar archive and return extracted file names.

extract_zip(archive: PathLike[str] | str) → set[str][source]#: Safely extract a zip archive and return extracted file names.

class dpdispatcher.file_manager.SubmissionStagingPlan(local_root: os.PathLike[str] | str, submission: Submission)[source]#

Bases: object

Compile a Submission into upload and download manifests.

The plan is deliberately independent from a transport implementation. A local copier, SFTP archive transport, and HDFS adapter can therefore apply the same path semantics without each backend walking tasks again.

Methods

`download_manifest`(remote_root, *[, ...])	Resolve backward files from a remote root.
`upload_manifest`(*[, include_tasks, ...])	Resolve all required forward files from the local work root.

download_manifest(remote_root: PathLike[str] | str, *, fallback_root: PathLike[str] | str | None = None, include_errors: bool = False) → ResolvedManifest[source]#

Resolve backward files from a remote root.

fallback_root makes a download idempotent: if a previous attempt already copied a requested file locally, its absence on the remote side is not reported as a new missing artifact.

upload_manifest(*, include_tasks: bool = True, include_common: bool = True) → ResolvedManifest[source]#: Resolve all required forward files from the local work root.

dpdispatcher.file_manager.write_text_atomic(root: PathLike[str] | str, relative_path: str, content: str) → Path[source]#: Compatibility helper for contexts that only need one atomic write.

dpdispatcher.machine module#

Define scheduler-independent job generation and the machine factory.

class dpdispatcher.machine.Machine(*args: Any, **kwargs: Any)[source]#

Bases: object

Generate, submit, and monitor jobs on a selected batch system.

Parameters:

contextBaseContext, optional: Existing execution context. When omitted, context_type and the root and profile arguments are used to construct one.
retry_countint, default=3: Number of scheduler failures allowed before a job is reported as failed.

Methods

`arginfo`()	Build the dargs schema for machine and context configuration.
`bind_context`(context)	Bind the execution context used for files and remote commands.
`check_finish_tag`(job)	Return whether the success marker for a job is present.
`check_if_recover`(submission)	Return whether serialized state exists for a remote submission.
`check_status`(job)	Query the scheduler and return the current status of a job.
`default_resources`(res)	Return backend defaults for an incomplete resource specification.
`deserialize`(machine_dict)	Reconstruct a machine from its serialized dictionary.
`do_submit`(job)	Submit a single job, assuming that no job is running there.
`gen_command_env_cuda_devices`(resources)	Assign a GPU to the next task when automatic GPU mapping is enabled.
`gen_local_script`(job)	Generate a local staging script for cloud backends.
`gen_script`(job)	Generate the complete scheduler submission script for a job.
`gen_script_command`(job)	Generate task commands, logging redirection, and completion tags.
`gen_script_custom_flags_lines`(job)	Render user-provided scheduler directives as script lines.
`gen_script_end`(job)	Generate job finalization, failure checks, and append commands.
`gen_script_env`(job)	Generate environment setup shared by every task in a job.
`gen_script_header`(job)	Generate scheduler directives and the script shebang for a job.
`gen_script_run_command`(job)	Return the command that sources the generated per-task script.
`gen_script_wait`(resources)	Generate synchronization commands for the configured parallelism.
`get_exit_code`(job)	Get exit code of the job.
`get_job_error`(job)	Return a text error diagnostic for a job, if available.
`kill`(job)	Kill the job.
`load_from_dict`(machine_dict[, allow_ref])	Load a Machine from a dict.
`load_from_json`(json_path)	Load a machine configuration from a JSON file.
`load_from_yaml`(yaml_path)	Load a machine configuration from a YAML file.
`resources_arginfo`()	Generate the resources arginfo.
`resources_subfields`()	Generate the resources subfields.
`serialize`([if_empty_remote_profile])	Return a normalized dictionary representation of the machine.
`sub_script_cmd`(res)	Return the legacy scheduler launch command for resources.
`sub_script_head`(res)	Return the legacy scheduler header for a resource specification.

alias: tuple[str, ...] = ()#

classmethod arginfo() → Argument[source]#: Build the dargs schema for machine and context configuration.

bind_context(context: BaseContext) → None[source]#: Bind the execution context used for files and remote commands.

abstract check_finish_tag(job: Job) → bool[source]#: Return whether the success marker for a job is present.

check_if_recover(submission: Submission) → bool[source]#: Return whether serialized state exists for a remote submission.

abstract check_status(job: Job) → JobStatus[source]#: Query the scheduler and return the current status of a job.

default_resources(res: Resources) → Resources[source]#: Return backend defaults for an incomplete resource specification.

classmethod deserialize(machine_dict: dict[str, Any]) → Machine[source]#: Reconstruct a machine from its serialized dictionary.

abstract do_submit(job: Job) → str | int[source]#: Submit a single job, assuming that no job is running there.

gen_command_env_cuda_devices(resources: Resources) → str[source]#: Assign a GPU to the next task when automatic GPU mapping is enabled.

gen_local_script(job: Job) → str[source]#

Generate a local staging script for cloud backends.

Only cloud machine implementations support this operation; the base declaration keeps their context interface type-safe.

gen_script(job: Job) → str[source]#: Generate the complete scheduler submission script for a job.

gen_script_command(job: Job) → str[source]#: Generate task commands, logging redirection, and completion tags.

gen_script_custom_flags_lines(job: Job) → str[source]#: Render user-provided scheduler directives as script lines.

gen_script_end(job: Job) → str[source]#: Generate job finalization, failure checks, and append commands.

gen_script_env(job: Job) → str[source]#: Generate environment setup shared by every task in a job.

abstract gen_script_header(job: Job) → str[source]#: Generate scheduler directives and the script shebang for a job.

gen_script_run_command(job: Job) → str[source]#: Return the command that sources the generated per-task script.

gen_script_wait(resources: Resources) → str[source]#: Generate synchronization commands for the configured parallelism.

get_exit_code(job: Job) → int[source]#

Get exit code of the job.

Parameters:

jobJob: job

get_job_error(job: Job) → str | None[source]#

Return a text error diagnostic for a job, if available.

HDFS contexts expose command output as bytes while other contexts return text. Decode byte-valued responses here so callers receive the same text interface regardless of the execution backend.

kill(job: Job) → None[source]#

Kill the job.

If not implemented, pass and let the user manually kill it.

Parameters:

jobJob: job

classmethod load_from_dict(machine_dict: dict[str, Any], allow_ref: bool = False) → Machine[source]#

Load a Machine from a dict.

Parameters:

machine_dictdict: Machine configuration dict.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_json(json_path: str) → Machine[source]#: Load a machine configuration from a JSON file.

classmethod load_from_yaml(yaml_path: str) → Machine[source]#: Load a machine configuration from a YAML file.

options = {'Bohrium', 'DistributedShell', 'Fugaku', 'JH_UniScheduler', 'LSF', 'OpenAPI', 'PBS', 'SGE', 'Shell', 'Slurm', 'SlurmJobArray', 'Torque'}#

classmethod resources_arginfo() → Argument[source]#

Generate the resources arginfo.

Returns:

Argument: resources arginfo

classmethod resources_subfields() → list[Argument][source]#

Generate the resources subfields.

Returns:

list[Argument]: resources subfields

serialize(if_empty_remote_profile: bool = False) → dict[str, Any][source]#: Return a normalized dictionary representation of the machine.

sub_script_cmd(res: Resources) → str[source]#: Return the legacy scheduler launch command for resources.

sub_script_head(res: Resources) → str[source]#: Return the legacy scheduler header for a resource specification.

subclasses_dict = {'Bohrium': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'DistributedShell': <class 'dpdispatcher.machines.distributed_shell.DistributedShell'>, 'DpCloudServer': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'Fugaku': <class 'dpdispatcher.machines.fugaku.Fugaku'>, 'JH_UniScheduler': <class 'dpdispatcher.machines.JH_UniScheduler.JH_UniScheduler'>, 'LSF': <class 'dpdispatcher.machines.lsf.LSF'>, 'Lebesgue': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'OpenAPI': <class 'dpdispatcher.machines.openapi.OpenAPI'>, 'PBS': <class 'dpdispatcher.machines.pbs.PBS'>, 'SGE': <class 'dpdispatcher.machines.pbs.SGE'>, 'Shell': <class 'dpdispatcher.machines.shell.Shell'>, 'Slurm': <class 'dpdispatcher.machines.slurm.Slurm'>, 'SlurmJobArray': <class 'dpdispatcher.machines.slurm.SlurmJobArray'>, 'Torque': <class 'dpdispatcher.machines.pbs.Torque'>, 'bohrium': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'distributedshell': <class 'dpdispatcher.machines.distributed_shell.DistributedShell'>, 'dpcloudserver': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'fugaku': <class 'dpdispatcher.machines.fugaku.Fugaku'>, 'jh_unischeduler': <class 'dpdispatcher.machines.JH_UniScheduler.JH_UniScheduler'>, 'lebesgue': <class 'dpdispatcher.machines.dp_cloud_server.Bohrium'>, 'lsf': <class 'dpdispatcher.machines.lsf.LSF'>, 'openapi': <class 'dpdispatcher.machines.openapi.OpenAPI'>, 'pbs': <class 'dpdispatcher.machines.pbs.PBS'>, 'sge': <class 'dpdispatcher.machines.pbs.SGE'>, 'shell': <class 'dpdispatcher.machines.shell.Shell'>, 'slurm': <class 'dpdispatcher.machines.slurm.Slurm'>, 'slurmjobarray': <class 'dpdispatcher.machines.slurm.SlurmJobArray'>, 'torque': <class 'dpdispatcher.machines.pbs.Torque'>}#

dpdispatcher.run module#

Parse PEP 723 metadata and execute its DPDispatcher submission.

dpdispatcher.run.create_submission(metadata: dict, script_hash: str, allow_ref: bool = False) → Submission[source]#

Create a Submission instance from a PEP 723 metadata.

Parameters:

metadatadict: PEP 723 metadata.
script_hashstr: Submission hash.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

Returns:

Submission: Submission instance.

dpdispatcher.run.pep723_args() → Argument[source]#: Return the argument parser for PEP 723 metadata.

dpdispatcher.run.read_pep723(script: str) → dict | None[source]#

Read a PEP 723 script metadata from a script string.

Parameters:

scriptstr: Script content.

Returns:

dict: PEP 723 metadata.

dpdispatcher.run.run_pep723(script: str, allow_ref: bool = False) → None[source]#

Run a PEP 723 script.

Parameters:

scriptstr: Script content.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

dpdispatcher.submission module#

Define submissions, tasks, generated jobs, and resource requests.

class dpdispatcher.submission.Job(job_task_list: list[Task], *, resources: Resources, machine: Machine | None = None)[source]#

Bases: object

Represent one scheduler job generated from a group of tasks.

Applications normally let Submission create jobs. A job owns a resource request, generates scheduler scripts through its machine, and stores scheduler ID, state, and retry information for recovery.

Parameters:

job_task_listlist of Task: Tasks grouped into this scheduler job.
resourcesResources: Resource request copied from the parent submission.
machineMachine, optional: Backend used to generate, submit, and monitor the job.

Methods

`deserialize`(job_dict[, machine])	Reconstruct a job and its tasks from serialized state.
`get_hash`()	Return the stable hash used as this job's identifier.
`get_job_state`()	Query the backend and update this job and its unfinished tasks.
`get_last_error_message`()	Get last error message when the job is terminated.
`get_scheduler_name`(max_length, *[, ...])	Return a portable task-derived name for a one-task scheduler job.
`handle_unexpected_job_state`(*[, ...])	Submit or retry a job according to its current state.
`job_to_json`()	Write current job state to the execution root as JSON.
`register_job_id`(job_id)	Store the identifier returned by the scheduler.
`serialize`([if_static])	Return a hash-keyed, JSON-compatible representation of the job.
`submit_job`()	Submit the job through its machine and update its local state.

classmethod deserialize(job_dict: dict[str, Any], machine: Machine | None = None) → Job[source]#

Reconstruct a job and its tasks from serialized state.

Parameters:

job_dictdict: Single-entry mapping from job hash to configuration and runtime data.
machineMachine, optional: Machine to bind to the reconstructed job.

Returns:

Job: Reconstructed job.

get_hash() → str[source]#: Return the stable hash used as this job’s identifier.

get_job_state() → None[source]#

Query the backend and update this job and its unfinished tasks.

Notes

This method does not submit or retry the job.

get_last_error_message() → str | None[source]#: Get last error message when the job is terminated.

get_scheduler_name(max_length: int, *, require_alpha_prefix: bool = False) → str | None[source]#

Return a portable task-derived name for a one-task scheduler job.

handle_unexpected_job_state(*, continue_on_failure: bool = False) → None[source]#

Submit or retry a job according to its current state.

Retry exhaustion remains fail-fast by default. Callers that need to monitor sibling jobs must explicitly opt in with continue_on_failure.

job_to_json() → None[source]#: Write current job state to the execution root as JSON.

register_job_id(job_id: str | int) → None[source]#: Store the identifier returned by the scheduler.

serialize(if_static: bool = False) → dict[str, Any][source]#

Return a hash-keyed, JSON-compatible representation of the job.

Parameters:

if_staticbool, default=False: Exclude job ID, state, and failure count when true.

Returns:

dict: Mapping from the deterministic job hash to job data.

submit_job() → None[source]#: Submit the job through its machine and update its local state.

class dpdispatcher.submission.Resources(number_node: int, cpu_per_node: int, gpu_per_node: int, queue_name: str, group_size: int, *, custom_flags: Sequence[str] | None = None, strategy: dict[str, Any] | None = None, para_deg: int = 1, module_unload_list: Sequence[str] | None = None, module_purge: bool = False, module_list: Sequence[str] | None = None, source_list: Sequence[str] | None = None, envs: dict[str, Any] | None = None, prepend_script: Sequence[str] | None = None, append_script: Sequence[str] | None = None, wait_time: int = 0, **kwargs: Any)[source]#

Bases: object

Describe the resources and execution strategy for generated jobs.

Parameters:

number_nodeint: Number of nodes requested for each generated job.
cpu_per_nodeint: Number of CPUs requested on each node.
gpu_per_nodeint: Number of GPUs requested on each node.
queue_namestr: Queue or partition name passed to the batch backend.
group_sizeint: Maximum number of tasks grouped into one scheduler job. Zero groups all tasks into one job.
custom_flagslist of str, optional: Extra scheduler directives inserted into the generated script header.
strategydict, optional: Script-generation strategy. Recognized keys include if_cuda_multi_devices, ratio_unfinished, and customized_script_header_template_file.
para_degint, default=1: Number of task commands run concurrently inside one generated job.
module_unload_listlist of str, optional: Environment modules to unload before task execution.
module_purgebool, default=False: Whether to purge loaded environment modules first.
module_listlist of str, optional: Environment modules to load before task execution.
source_listlist of str, optional: Shell files to source before task execution.
envsdict, optional: Environment variables to export before task execution.
prepend_scriptlist of str, optional: Shell lines inserted before task commands.
append_scriptlist of str, optional: Shell lines inserted after all task commands finish.
wait_timeint, default=0: Delay in seconds after each job submission or resubmission.
**kwargs: Backend-specific resource options, stored in Resources.kwargs.

Methods

`arginfo`([detail_kwargs])	Build the dargs schema for common and backend-specific resources.
`deserialize`(resources_dict)	Reconstruct a resource request from a serialized dictionary.
`load_from_dict`(resources_dict[, allow_ref])	Load Resources from a dict.
`load_from_json`(json_file)	Load and validate a resource request from a JSON file.
`load_from_yaml`(yaml_file)	Load and validate a resource request from a YAML file.
`serialize`()	Return the resource request as a JSON-compatible dictionary.

static arginfo(detail_kwargs: bool = True) → Argument[source]#: Build the dargs schema for common and backend-specific resources.

classmethod deserialize(resources_dict: dict[str, Any]) → Resources[source]#: Reconstruct a resource request from a serialized dictionary.

classmethod load_from_dict(resources_dict: dict[str, Any], allow_ref: bool = False) → Resources[source]#

Load Resources from a dict.

Parameters:

resources_dictdict: Resources configuration dict.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_json(json_file: str) → Resources[source]#: Load and validate a resource request from a JSON file.

classmethod load_from_yaml(yaml_file: str) → Resources[source]#: Load and validate a resource request from a YAML file.

serialize() → dict[str, Any][source]#: Return the resource request as a JSON-compatible dictionary.

class dpdispatcher.submission.Submission(work_base: str, machine: Machine | None = None, resources: Resources | None = None, forward_common_files: list[str] | None = None, backward_common_files: list[str] | None = None, *, task_list: list[Task] | None = None, previous_submission_hash: str | None = None, continue_on_failure: bool = False)[source]#

Bases: object

Coordinate a collection of tasks that share a working directory.

A submission groups tasks into scheduler jobs, stages common files, monitors execution, downloads declared results, and records state for recovery.

Parameters:

work_basepath-like: Local base directory containing all task working directories.
machineMachine, optional: Batch backend and execution context. It may be bound later with bind_machine().
resourcesResources, optional: Resource request copied into every generated job.
forward_common_fileslist of path-like, optional: Files shared by all tasks and staged before execution.
backward_common_fileslist of path-like, optional: Shared result files downloaded after execution.
task_listlist of Task, optional: Tasks to register when the submission is created.
previous_submission_hashstr, optional: SHA-1 hash of a compatible prior submission whose finished-task state should be reused after a resource-only change.
continue_on_failurebool, default=False: Continue monitoring other jobs after one job exhausts its retries. The default preserves the historical fail-fast behavior.

Methods

`async_run_submission`(**kwargs)	Run `run_submission()` in an executor.
`bind_machine`(machine, *[, bind_context])	Bind a machine and initialize submission-specific context paths.
`check_all_finished`()	Return whether every generated job has finished successfully.
`check_ratio_unfinished`(ratio_unfinished)	Return whether the allowed unfinished-task threshold is satisfied.
`clean_jobs`()	Remove remote working data and the local recovery record.
`deserialize`(submission_dict[, machine, ...])	Reconstruct a submission from serialized state.
`download_jobs`([include_failed])	Download selected task outputs, optionally including failed jobs.
`failed_jobs`()	Return jobs that exhausted retries and reached a terminal failure.
`generate_jobs`()	Generate jobs after tasks are registered.
`get_hash`()	Return the stable hash of the submission's static configuration.
`handle_unexpected_submission_state`(*[, ...])	Submit unsubmitted jobs and retry unexpectedly terminated jobs.
`raise_for_failed_jobs`(*[, continue_on_failure])	Report all terminal failures after other jobs and downloads finish.
`register_task`(task)	Append one task before jobs have been generated.
`register_task_list`(task_list)	Append multiple tasks before jobs have been generated.
`remove_unfinished_tasks`()	Stop unfinished work while preserving durable failed records.
`run_submission`(*[, dry_run, exit_on_submit, ...])	Execute the submission and monitor it until completion.
`serialize`([if_static])	Return a JSON-compatible representation of the submission.
`submission_from_json`([json_file_name])	Load a submission, including machine state, from a local JSON file.
`submission_to_json`()	Write current submission state to the execution root as JSON.
`try_download_error_info`()	Download error diagnostic files for failed/terminated jobs.
`try_download_result`()	Download results, retrying transient failures for up to 24 hours.
`try_recover_from_json`()	Restore compatible job state from the remote submission JSON file.
`update_submission_state`()	Refresh every unfinished job's state from its machine backend.
`upload_jobs`()	Upload submission inputs through the bound context.

async async_run_submission(**kwargs: Any) → dict[str, Any][source]#

Run run_submission() in an executor.

Examples

>>> import asyncio
>>> async def run_all(submissions):
...     return await asyncio.gather(
...         *(
...             submission.async_run_submission(check_interval=2)
...             for submission in submissions
...         )
...     )

bind_machine(machine: Machine | None, *, bind_context: bool = True) → Submission[source]#

Bind a machine and initialize submission-specific context paths.

Parameters:

machineMachine or None: Machine to use for generated jobs and file operations.

Returns:

Submission: This submission, for convenient chained configuration.

check_all_finished() → bool[source]#

Return whether every generated job has finished successfully.

Notes

This method does not submit, retry, or otherwise change job states.

check_ratio_unfinished(ratio_unfinished: float) → bool[source]#

Return whether the allowed unfinished-task threshold is satisfied.

Parameters:

ratio_unfinishedfloat: Maximum fraction of tasks that may remain unfinished.

Returns:

bool: True when the finished fraction is at least 1 - ratio_unfinished.

clean_jobs() → None[source]#: Remove remote working data and the local recovery record.

classmethod deserialize(submission_dict: dict[str, Any], machine: Machine | None = None, *, bind_context: bool = True) → Submission[source]#

Reconstruct a submission from serialized state.

Parameters:

submission_dictdict: Serialized submission configuration and job state.
machineMachine, optional: Machine to bind instead of reconstructing the serialized machine.

Returns:

Submission: Reconstructed submission.

download_jobs(include_failed: bool = False) → None[source]#

Download selected task outputs, optionally including failed jobs.

failed_jobs() → list[Job][source]#: Return jobs that exhausted retries and reached a terminal failure.

generate_jobs() → None[source]#

Generate jobs after tasks are registered.

get_hash() → str[source]#: Return the stable hash of the submission’s static configuration.

handle_unexpected_submission_state(*, continue_on_failure: bool = False) → None[source]#

Submit unsubmitted jobs and retry unexpectedly terminated jobs.

Unknown states and exhausted retries are persisted for recovery before the error is propagated.

raise_for_failed_jobs(*, continue_on_failure: bool = True) → None[source]#: Report all terminal failures after other jobs and downloads finish.

register_task(task: Task) → None[source]#: Append one task before jobs have been generated.

register_task_list(task_list: list[Task]) → None[source]#: Append multiple tasks before jobs have been generated.

remove_unfinished_tasks() → None[source]#

Stop unfinished work while preserving durable failed records.

Execute the submission and monitor it until completion.

The lifecycle recovers compatible state, stages files, submits or retries jobs, polls their status, downloads declared results, persists recovery state, and optionally cleans the execution directory.

Parameters:

dry_runbool, default=False: Upload inputs and generated scripts without submitting jobs.
exit_on_submitbool, default=False: Return after jobs have been submitted instead of waiting for them.
cleanbool, default=True: Remove the submission-specific execution directory after download.
check_intervalint or float, default=30: Seconds between scheduler status checks.
continue_on_failurebool, optional: Continue monitoring remaining jobs after retry exhaustion. If omitted, use the policy stored on this submission (which defaults to False). An explicit value overrides the persisted policy for this run.

Returns:

dict: Serialized submission state at the point this method returns.

serialize(if_static: bool = False) → dict[str, Any][source]#

Return a JSON-compatible representation of the submission.

Parameters:

if_staticbool, default=False: Exclude job IDs, states, failure counts, and runtime failure policy when true. The policy is intentionally excluded from the static identity so enabling continuation does not change job hashes.

Returns:

dict: Submission configuration and, unless excluded, current runtime state.

classmethod submission_from_json(json_file_name: str = 'submission.json') → Submission[source]#: Load a submission, including machine state, from a local JSON file.

submission_to_json() → None[source]#: Write current submission state to the execution root as JSON.

try_download_error_info() → None[source]#

Download error diagnostic files for failed/terminated jobs.

The error file contains the last 1000 bytes of stderr from the most recently failed task in the job, written by the generated bash script.

try_download_result() → bool[source]#: Download results, retrying transient failures for up to 24 hours.

try_recover_from_json() → None[source]#: Restore compatible job state from the remote submission JSON file.

update_submission_state() → None[source]#

Refresh every unfinished job’s state from its machine backend.

Notes

This method only queries state. It does not submit or retry jobs.

upload_jobs() → None[source]#: Upload submission inputs through the bound context.

class dpdispatcher.submission.Task(command: str, task_work_path: str, forward_files: Sequence[str] | None = None, backward_files: Sequence[str] | None = None, outlog: str | None = 'log', errlog: str | None = 'err', task_name: str | None = None)[source]#

Bases: object

Represent a sequential command and its staged files.

A task records the files it depends on and the results to transfer back.

Parameters:

commandstr: Shell command to execute.
task_work_pathpath-like: Working directory relative to the submission’s work_base.
forward_fileslist of path-like, optional: Task-specific input files staged before execution.
backward_fileslist of path-like, optional: Task-specific result files downloaded after execution.
outlogstr or None, default=”log”: File that receives standard output, or None to leave it attached.
errlogstr or None, default=”err”: File that receives standard error, or None to leave it attached.
task_namestr or None, optional: Scheduler-visible name when this task is submitted as a one-task job.

Methods

`arginfo`()	Build the dargs schema for task configuration.
`deserialize`(task_dict)	Reconstruct a task from a serialized dictionary.
`get_hash`()	Return the stable hash of the task configuration.
`get_task_state`(context)	Get the task state by checking the tag file.
`has_finished_tag`(context)	Return whether the remote completion tag for this task exists.
`load_from_dict`(task_dict[, allow_ref])	Load a Task from a dict.
`load_from_json`(json_file[, allow_ref])	Load a Task from a JSON file.
`load_from_yaml`(yaml_file[, allow_ref])	Load a Task from a YAML file.
`serialize`()	Return the task configuration as a JSON-compatible dictionary.

static arginfo() → Argument[source]#: Build the dargs schema for task configuration.

classmethod deserialize(task_dict: dict[str, Any]) → Task[source]#

Reconstruct a task from a serialized dictionary.

Parameters:

task_dictdict: Task configuration.

Returns:

Task: Reconstructed task.

get_hash() → str[source]#: Return the stable hash of the task configuration.

get_task_state(context: BaseContext) → None[source]#

Get the task state by checking the tag file.

Parameters:

contextContext: the context of the task

has_finished_tag(context: BaseContext) → bool[source]#: Return whether the remote completion tag for this task exists.

classmethod load_from_dict(task_dict: dict, allow_ref: bool = False) → Task[source]#

Load a Task from a dict.

Parameters:

task_dictdict: Task configuration dict.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_json(json_file: str, allow_ref: bool = False) → Task[source]#

Load a Task from a JSON file.

Parameters:

json_filestr: Path to task JSON file.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

classmethod load_from_yaml(yaml_file: str, allow_ref: bool = False) → Task[source]#

Load a Task from a YAML file.

Parameters:

yaml_filestr: Path to task YAML file.
allow_refbool, default=False: Whether to allow loading external JSON/YAML snippets via $ref. Disabled by default for security.

serialize() → dict[str, Any][source]#: Return the task configuration as a JSON-compatible dictionary.

dpdispatcher package

Contents

dpdispatcher package#

Subpackages#

Submodules#

dpdispatcher.arginfo module#

dpdispatcher.base_context module#

dpdispatcher.dlog module#

dpdispatcher.dpdisp module#

dpdispatcher.file_manager module#

dpdispatcher.machine module#

dpdispatcher.run module#

dpdispatcher.submission module#