dpdispatcher package
- class dpdispatcher.DistributedShell(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit th job to yarn using distributed shell.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- dpdispatcher.DpCloudServerContext
alias of
BohriumContext
- class dpdispatcher.HDFSContext(*args, **kwargs)[source]
Bases:
BaseContext
Methods
check_file_exists
(fname)Check whether the given file exists, often used in checking whether the belonging job has finished.
download
(submission[, check_exists, ...])Download backward files from HDFS root dir.
machine_arginfo
()Generate the machine arginfo.
machine_subfields
()Generate the machine subfields.
upload
(submission[, dereference])Upload forward files and forward command files to HDFS root dir.
bind_submission
check_finish
clean
get_job_root
load_from_dict
read_file
write_file
- check_file_exists(fname)[source]
Check whether the given file exists, often used in checking whether the belonging job has finished.
- Parameters:
- fnamestring
file name to be checked
- Returns:
- status: boolean
- download(submission, check_exists=False, mark_failure=True, back_error=False)[source]
Download backward files from HDFS root dir.
- Parameters:
- submissionSubmission class instance
represents a collection of tasks, such as backward file names
- check_existsbool
whether to check if the file exists
- mark_failurebool
whether to mark the task as failed if the file does not exist
- back_errorbool
whether to download error files
- Returns:
- none
- class dpdispatcher.Job(job_task_list, *, resources, machine=None)[source]
Bases:
object
Job is generated by Submission automatically. A job ususally has many tasks and it may request computing resources from job scheduler systems. Each Job can generate a script file to be submitted to the job scheduler system or executed locally.
- Parameters:
- job_task_listlist of Task
the tasks belonging to the job
- resourcesResources
the machine resources. Passed from Submission when it constructs jobs.
- machinemachine
machine object to execute the job. Passed from Submission when it constructs jobs.
Methods
deserialize
(job_dict[, machine])Convert the job_dict to a Submission class object.
Get the jobs.
serialize
([if_static])Convert the Task class instance to a dictionary.
get_hash
handle_unexpected_job_state
job_to_json
register_job_id
submit_job
- classmethod deserialize(job_dict, machine=None)[source]
Convert the job_dict to a Submission class object.
- Parameters:
- job_dictdict
the dictionary which contains the job information
- machineMachine
the machine object to execute the job
- Returns:
- submissionJob
the Job class instance converted from the job_dict
- get_job_state()[source]
Get the jobs. Usually, this method will query the database of slurm or pbs job scheduler system and get the results.
Notes
this method will not submit or resubmit the jobs if the job is unsubmitted.
- class dpdispatcher.LSF(*args, **kwargs)[source]
Bases:
Machine
LSF batch.
Methods
default_resources
(resources)kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
deserialize
do_submit
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- check_status(**kwargs)
- do_submit(**kwargs)
Submit a single job, assuming that no job is running there.
- class dpdispatcher.LazyLocalContext(*args, **kwargs)[source]
Bases:
BaseContext
Run jobs in the local server and local directory.
- Parameters:
- local_rootstr
The local directory to store the jobs.
- remote_rootstr, optional
The argument takes no effect.
- remote_profiledict, optional
The remote profile. The default is {}.
- *args
The arguments.
- **kwargs
The keyword arguments.
Methods
machine_arginfo
()Generate the machine arginfo.
machine_subfields
()Generate the machine subfields.
bind_submission
block_call
block_checkcall
call
check_file_exists
check_finish
clean
download
get_job_root
get_return
load_from_dict
read_file
upload
write_file
- dpdispatcher.LebesgueContext
alias of
BohriumContext
- class dpdispatcher.LocalContext(*args, **kwargs)[source]
Bases:
BaseContext
Run jobs in the local server and remote directory.
- Parameters:
- local_rootstr
The local directory to store the jobs.
- remote_rootstr
The remote directory to store the jobs.
- remote_profiledict, optional
The remote profile. The default is {}.
- *args
The arguments.
- **kwargs
The keyword arguments.
Methods
machine_arginfo
()Generate the machine arginfo.
machine_subfields
()Generate the machine subfields.
bind_submission
block_call
block_checkcall
call
check_file_exists
check_finish
clean
download
get_job_root
get_return
load_from_dict
read_file
upload
write_file
- class dpdispatcher.Machine(*args, **kwargs)[source]
Bases:
object
A machine is used to handle the connection with remote machines.
- Parameters:
- contextSubClass derived from BaseContext
The context is used to mainatin the connection with remote machine.
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- kill(job)[source]
Kill the job.
If not implemented, pass and let the user manually kill it.
- Parameters:
- jobJob
job
- options = {'Bohrium', 'DistributedShell', 'LSF', 'PBS', '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
- subclasses_dict = {'Bohrium': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'DistributedShell': <class 'dpdispatcher.distributed_shell.DistributedShell'>, 'DpCloudServer': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'LSF': <class 'dpdispatcher.lsf.LSF'>, 'Lebesgue': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'PBS': <class 'dpdispatcher.pbs.PBS'>, 'Shell': <class 'dpdispatcher.shell.Shell'>, 'Slurm': <class 'dpdispatcher.slurm.Slurm'>, 'SlurmJobArray': <class 'dpdispatcher.slurm.SlurmJobArray'>, 'Torque': <class 'dpdispatcher.pbs.Torque'>, 'bohrium': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'distributedshell': <class 'dpdispatcher.distributed_shell.DistributedShell'>, 'dpcloudserver': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'lebesgue': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'lsf': <class 'dpdispatcher.lsf.LSF'>, 'pbs': <class 'dpdispatcher.pbs.PBS'>, 'shell': <class 'dpdispatcher.shell.Shell'>, 'slurm': <class 'dpdispatcher.slurm.Slurm'>, 'slurmjobarray': <class 'dpdispatcher.slurm.SlurmJobArray'>, 'torque': <class 'dpdispatcher.pbs.Torque'>}
- class dpdispatcher.PBS(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- class dpdispatcher.Resources(number_node, cpu_per_node, gpu_per_node, queue_name, group_size, *, custom_flags=[], strategy={'if_cuda_multi_devices': False, 'ratio_unfinished': 0.0}, para_deg=1, module_unload_list=[], module_purge=False, module_list=[], source_list=[], envs={}, prepend_script=[], append_script=[], wait_time=0, **kwargs)[source]
Bases:
object
Resources is used to describe the machine resources we need to do calculations.
- Parameters:
- number_nodeint
The number of node need for each job.
- cpu_per_nodeint
cpu numbers of each node.
- gpu_per_nodeint
gpu numbers of each node.
- queue_namestr
The queue name of batch job scheduler system.
- group_sizeint
The number of tasks in a job.
- custom_flagslist of Str
The extra lines pass to job submitting script header
- strategydict
strategies we use to generation job submitting scripts. if_cuda_multi_devices : bool
If there are multiple nvidia GPUS on the node, and we want to assign the tasks to different GPUS. If true, dpdispatcher will manually export environment variable CUDA_VISIBLE_DEVICES to different task. Usually, this option will be used with Task.task_need_resources variable simultaneously.
- ratio_unfinishedfloat
The ratio of task that can be unfinished.
- para_degint
Decide how many tasks will be run in parallel. Usually run with strategy[‘if_cuda_multi_devices’]
- source_listlist of Path
The env file to be sourced before the command execution.
- wait_timeint
The waitting time in second after a single task submitted. Default: 0.
Methods
arginfo
deserialize
load_from_dict
load_from_json
serialize
- class dpdispatcher.SSHContext(*args, **kwargs)[source]
Bases:
BaseContext
- Attributes:
- sftp
- ssh
Methods
block_checkcall
(cmd[, asynchronously, ...])Run command with arguments.
machine_arginfo
()Generate the machine arginfo.
Generate the machine subfields.
bind_submission
block_call
call
check_file_exists
check_finish
clean
close
download
get_job_root
get_return
load_from_dict
read_file
upload
write_file
- block_checkcall(cmd, asynchronously=False, stderr_whitelist=None)[source]
Run command with arguments. Wait for command to complete. If the return code was zero then return, otherwise raise RuntimeError.
- Parameters:
- cmdstr
The command to run.
- asynchronouslybool, optional, default=False
Run command asynchronously. If True, nohup will be used to run the command.
- stderr_whitelistlist of str, optional, default=None
If not None, the stderr will be checked against the whitelist. If the stderr contains any of the strings in the whitelist, the command will be considered successful.
- classmethod machine_subfields() List[Argument] [source]
Generate the machine subfields.
- Returns:
- list[Argument]
machine subfields
- property sftp
- property ssh
- class dpdispatcher.Shell(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- class dpdispatcher.Slurm(*args, **kwargs)[source]
Bases:
Machine
Methods
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
do_submit
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- check_status(**kwargs)
- do_submit(**kwargs)
Submit a single job, assuming that no job is running there.
- class dpdispatcher.Submission(work_base, machine=None, resources=None, forward_common_files=[], backward_common_files=[], *, task_list=[])[source]
Bases:
object
A submission represents a collection of tasks. These tasks usually locate at a common directory. And these Tasks may share common files to be uploaded and downloaded.
- Parameters:
- work_basePath
the base directory of the local tasks. It is usually the dir name of project .
- machineMachine
machine class object (for example, PBS, Slurm, Shell) to execute the jobs. The machine can still be bound after the instantiation with the bind_submission method.
- resourcesResources
the machine resources (cpu or gpu) used to generate the slurm/pbs script
- forward_common_fileslist
the common files to be uploaded to other computers before the jobs begin
- backward_common_fileslist
the common files to be downloaded from other computers after the jobs finish
- task_listlist of Task
a list of tasks to be run.
Methods
bind_machine
(machine)Bind this submission to a machine.
Check whether all the jobs in the submission.
check_ratio_unfinished
(ratio_unfinished)Calculate the ratio of unfinished tasks in the submission.
deserialize
(submission_dict[, machine])Convert the submission_dict to a Submission class object.
After tasks register to the self.belonging_tasks, This method generate the jobs and add these jobs to self.belonging_jobs.
Handle unexpected job state of the submission.
run_submission
(*[, dry_run, exit_on_submit, ...])Main method to execute the submission.
serialize
([if_static])Convert the Submission class instance to a dictionary.
Check whether all the jobs in the submission.
clean_jobs
download_jobs
get_hash
register_task
register_task_list
remove_unfinished_tasks
submission_from_json
submission_to_json
try_recover_from_json
upload_jobs
- bind_machine(machine)[source]
Bind this submission to a machine. update the machine’s context remote_root and local_root.
- Parameters:
- machineMachine
the machine to bind with
- check_all_finished()[source]
Check whether all the jobs in the submission.
Notes
This method will not handle unexpected job state in the submission.
- check_ratio_unfinished(ratio_unfinished: float) bool [source]
Calculate the ratio of unfinished tasks in the submission.
- Parameters:
- ratio_unfinishedfloat
the ratio of unfinished tasks in the submission
- Returns:
- bool
whether the ratio of unfinished tasks in the submission is larger than ratio_unfinished
- classmethod deserialize(submission_dict, machine=None)[source]
Convert the submission_dict to a Submission class object.
- Parameters:
- submission_dictdict
path-like, the base directory of the local tasks
- machineMachine
Machine class Object to execute the jobs
- Returns:
- submissionSubmission
the Submission class instance converted from the submission_dict
- generate_jobs()[source]
After tasks register to the self.belonging_tasks, This method generate the jobs and add these jobs to self.belonging_jobs. The jobs are generated by the tasks randomly, and there are self.resources.group_size tasks in a task. Why we randomly shuffle the tasks is under the consideration of load balance. The random seed is a constant (to be concrete, 42). And this insures that the jobs are equal when we re-run the program.
- handle_unexpected_submission_state()[source]
Handle unexpected job state of the submission. If the job state is unsubmitted, submit the job. If the job state is terminated (killed unexpectly), resubmit the job. If the job state is unknown, raise an error.
- run_submission(*, dry_run=False, exit_on_submit=False, clean=True)[source]
Main method to execute the submission. First, check whether old Submission exists on the remote machine, and try to recover from it. Second, upload the local files to the remote machine where the tasks to be executed. Third, run the submission defined previously. Forth, wait until the tasks in the submission finished and download the result file to local directory. If dry_run is True, submission will be uploaded but not be executed and exit. If exit_on_submit is True, submission will exit.
- serialize(if_static=False)[source]
Convert the Submission class instance to a dictionary.
- Parameters:
- if_staticbool
whether dump the job runtime infomation (like job_id, job_state, fail_count) to the dictionary.
- Returns:
- submission_dictdict
the dictionary converted from the Submission class instance
- class dpdispatcher.Task(command, task_work_path, forward_files=[], backward_files=[], outlog='log', errlog='err')[source]
Bases:
object
A task is a sequential command to be executed, as well as the files it depends on to transmit forward and backward.
- Parameters:
- commandStr
the command to be executed.
- task_work_pathPath
the directory of each file where the files are dependent on.
- forward_fileslist of Path
the files to be transmitted to remote machine before the command execute.
- backward_fileslist of Path
the files to be transmitted from remote machine after the comand finished.
- outlogStr
the filename to which command redirect stdout
- errlogStr
the filename to which command redirect stderr
Methods
deserialize
(task_dict)Convert the task_dict to a Task class object.
get_task_state
(context)Get the task state by checking the tag file.
arginfo
get_hash
load_from_dict
load_from_json
serialize
- classmethod deserialize(task_dict)[source]
Convert the task_dict to a Task class object.
- Parameters:
- task_dictdict
the dictionary which contains the task information
- Returns:
- taskTask
the Task class instance converted from the task_dict
- class dpdispatcher.Torque(*args, **kwargs)[source]
Bases:
PBS
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
Subpackages
Submodules
dpdispatcher.JobStatus module
dpdispatcher.arginfo module
dpdispatcher.base_context module
- class dpdispatcher.base_context.BaseContext(*args, **kwargs)[source]
Bases:
object
Methods
Generate the machine arginfo.
Generate the machine subfields.
bind_submission
check_finish
clean
download
load_from_dict
read_file
upload
write_file
- 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', 'SSHContext'}
- subclasses_dict = {'Bohrium': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'BohriumContext': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'DpCloudServer': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'DpCloudServerContext': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'HDFS': <class 'dpdispatcher.hdfs_context.HDFSContext'>, 'HDFSContext': <class 'dpdispatcher.hdfs_context.HDFSContext'>, 'LazyLocal': <class 'dpdispatcher.lazy_local_context.LazyLocalContext'>, 'LazyLocalContext': <class 'dpdispatcher.lazy_local_context.LazyLocalContext'>, 'Lebesgue': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'LebesgueContext': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'Local': <class 'dpdispatcher.local_context.LocalContext'>, 'LocalContext': <class 'dpdispatcher.local_context.LocalContext'>, 'SSH': <class 'dpdispatcher.ssh_context.SSHContext'>, 'SSHContext': <class 'dpdispatcher.ssh_context.SSHContext'>, 'bohrium': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'bohriumcontext': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'dpcloudserver': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'dpcloudservercontext': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'hdfs': <class 'dpdispatcher.hdfs_context.HDFSContext'>, 'hdfscontext': <class 'dpdispatcher.hdfs_context.HDFSContext'>, 'lazylocal': <class 'dpdispatcher.lazy_local_context.LazyLocalContext'>, 'lazylocalcontext': <class 'dpdispatcher.lazy_local_context.LazyLocalContext'>, 'lebesgue': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'lebesguecontext': <class 'dpdispatcher.dp_cloud_server_context.BohriumContext'>, 'local': <class 'dpdispatcher.local_context.LocalContext'>, 'localcontext': <class 'dpdispatcher.local_context.LocalContext'>, 'ssh': <class 'dpdispatcher.ssh_context.SSHContext'>, 'sshcontext': <class 'dpdispatcher.ssh_context.SSHContext'>}
dpdispatcher.distributed_shell module
- class dpdispatcher.distributed_shell.DistributedShell(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit th job to yarn using distributed shell.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
dpdispatcher.dp_cloud_server module
- class dpdispatcher.dp_cloud_server.Bohrium(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_local_script
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
map_dp_job_state
serialize
sub_script_cmd
sub_script_head
dpdispatcher.dp_cloud_server_context module
- class dpdispatcher.dp_cloud_server_context.BohriumContext(*args, **kwargs)[source]
Bases:
BaseContext
Methods
machine_arginfo
()Generate the machine arginfo.
Generate the machine subfields.
bind_submission
check_file_exists
check_finish
check_home_file_exits
clean
download
load_from_dict
read_file
read_home_file
upload
upload_job
write_file
write_home_file
write_local_file
- dpdispatcher.dp_cloud_server_context.DpCloudServerContext
alias of
BohriumContext
- dpdispatcher.dp_cloud_server_context.LebesgueContext
alias of
BohriumContext
dpdispatcher.dpdisp module
dpdispatcher.hdfs_cli module
- class dpdispatcher.hdfs_cli.HDFS[source]
Bases:
object
Fundamental class for HDFS basic manipulation.
Methods
copy_from_local
(local_path, to_uri)Returns: True on success Raises: on unexpected error.
exists
(uri)Check existence of hdfs uri Returns: True on exists Raises: RuntimeError.
mkdir
(uri)Make new hdfs directory Returns: True on success Raises: RuntimeError.
remove
(uri)Check existence of hdfs uri Returns: True on exists Raises: RuntimeError.
copy_to_local
move
read_hdfs_file
- static copy_from_local(local_path, to_uri)[source]
Returns: True on success Raises: on unexpected error.
dpdispatcher.hdfs_context module
- class dpdispatcher.hdfs_context.HDFSContext(*args, **kwargs)[source]
Bases:
BaseContext
Methods
check_file_exists
(fname)Check whether the given file exists, often used in checking whether the belonging job has finished.
download
(submission[, check_exists, ...])Download backward files from HDFS root dir.
machine_arginfo
()Generate the machine arginfo.
machine_subfields
()Generate the machine subfields.
upload
(submission[, dereference])Upload forward files and forward command files to HDFS root dir.
bind_submission
check_finish
clean
get_job_root
load_from_dict
read_file
write_file
- check_file_exists(fname)[source]
Check whether the given file exists, often used in checking whether the belonging job has finished.
- Parameters:
- fnamestring
file name to be checked
- Returns:
- status: boolean
- download(submission, check_exists=False, mark_failure=True, back_error=False)[source]
Download backward files from HDFS root dir.
- Parameters:
- submissionSubmission class instance
represents a collection of tasks, such as backward file names
- check_existsbool
whether to check if the file exists
- mark_failurebool
whether to mark the task as failed if the file does not exist
- back_errorbool
whether to download error files
- Returns:
- none
dpdispatcher.lazy_local_context module
- class dpdispatcher.lazy_local_context.LazyLocalContext(*args, **kwargs)[source]
Bases:
BaseContext
Run jobs in the local server and local directory.
- Parameters:
- local_rootstr
The local directory to store the jobs.
- remote_rootstr, optional
The argument takes no effect.
- remote_profiledict, optional
The remote profile. The default is {}.
- *args
The arguments.
- **kwargs
The keyword arguments.
Methods
machine_arginfo
()Generate the machine arginfo.
machine_subfields
()Generate the machine subfields.
bind_submission
block_call
block_checkcall
call
check_file_exists
check_finish
clean
download
get_job_root
get_return
load_from_dict
read_file
upload
write_file
dpdispatcher.local_context module
- class dpdispatcher.local_context.LocalContext(*args, **kwargs)[source]
Bases:
BaseContext
Run jobs in the local server and remote directory.
- Parameters:
- local_rootstr
The local directory to store the jobs.
- remote_rootstr
The remote directory to store the jobs.
- remote_profiledict, optional
The remote profile. The default is {}.
- *args
The arguments.
- **kwargs
The keyword arguments.
Methods
machine_arginfo
()Generate the machine arginfo.
machine_subfields
()Generate the machine subfields.
bind_submission
block_call
block_checkcall
call
check_file_exists
check_finish
clean
download
get_job_root
get_return
load_from_dict
read_file
upload
write_file
dpdispatcher.lsf module
- class dpdispatcher.lsf.LSF(*args, **kwargs)[source]
Bases:
Machine
LSF batch.
Methods
default_resources
(resources)kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
deserialize
do_submit
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- check_status(**kwargs)
- do_submit(**kwargs)
Submit a single job, assuming that no job is running there.
dpdispatcher.machine module
- class dpdispatcher.machine.Machine(*args, **kwargs)[source]
Bases:
object
A machine is used to handle the connection with remote machines.
- Parameters:
- contextSubClass derived from BaseContext
The context is used to mainatin the connection with remote machine.
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- kill(job)[source]
Kill the job.
If not implemented, pass and let the user manually kill it.
- Parameters:
- jobJob
job
- options = {'Bohrium', 'DistributedShell', 'LSF', 'PBS', '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
- subclasses_dict = {'Bohrium': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'DistributedShell': <class 'dpdispatcher.distributed_shell.DistributedShell'>, 'DpCloudServer': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'LSF': <class 'dpdispatcher.lsf.LSF'>, 'Lebesgue': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'PBS': <class 'dpdispatcher.pbs.PBS'>, 'Shell': <class 'dpdispatcher.shell.Shell'>, 'Slurm': <class 'dpdispatcher.slurm.Slurm'>, 'SlurmJobArray': <class 'dpdispatcher.slurm.SlurmJobArray'>, 'Torque': <class 'dpdispatcher.pbs.Torque'>, 'bohrium': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'distributedshell': <class 'dpdispatcher.distributed_shell.DistributedShell'>, 'dpcloudserver': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'lebesgue': <class 'dpdispatcher.dp_cloud_server.Bohrium'>, 'lsf': <class 'dpdispatcher.lsf.LSF'>, 'pbs': <class 'dpdispatcher.pbs.PBS'>, 'shell': <class 'dpdispatcher.shell.Shell'>, 'slurm': <class 'dpdispatcher.slurm.Slurm'>, 'slurmjobarray': <class 'dpdispatcher.slurm.SlurmJobArray'>, 'torque': <class 'dpdispatcher.pbs.Torque'>}
dpdispatcher.pbs module
- class dpdispatcher.pbs.PBS(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- class dpdispatcher.pbs.Torque(*args, **kwargs)[source]
Bases:
PBS
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
dpdispatcher.shell module
- class dpdispatcher.shell.Shell(*args, **kwargs)[source]
Bases:
Machine
Methods
do_submit
(job)Submit a single job, assuming that no job is running there.
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
resources_subfields
()Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
dpdispatcher.slurm module
- class dpdispatcher.slurm.Slurm(*args, **kwargs)[source]
Bases:
Machine
Methods
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
do_submit
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- check_status(**kwargs)
- do_submit(**kwargs)
Submit a single job, assuming that no job is running there.
- class dpdispatcher.slurm.SlurmJobArray(*args, **kwargs)[source]
Bases:
Slurm
Slurm with job array enabled for multiple tasks in a job.
Methods
kill
(job)Kill the job.
resources_arginfo
()Generate the resources arginfo.
Generate the resources subfields.
arginfo
bind_context
check_finish_tag
check_if_recover
check_status
default_resources
deserialize
do_submit
gen_command_env_cuda_devices
gen_script
gen_script_command
gen_script_custom_flags_lines
gen_script_end
gen_script_env
gen_script_header
gen_script_wait
load_from_dict
load_from_json
serialize
sub_script_cmd
sub_script_head
- check_status(**kwargs)
dpdispatcher.ssh_context module
- class dpdispatcher.ssh_context.SSHContext(*args, **kwargs)[source]
Bases:
BaseContext
- Attributes:
- sftp
- ssh
Methods
block_checkcall
(cmd[, asynchronously, ...])Run command with arguments.
machine_arginfo
()Generate the machine arginfo.
Generate the machine subfields.
bind_submission
block_call
call
check_file_exists
check_finish
clean
close
download
get_job_root
get_return
load_from_dict
read_file
upload
write_file
- block_checkcall(cmd, asynchronously=False, stderr_whitelist=None)[source]
Run command with arguments. Wait for command to complete. If the return code was zero then return, otherwise raise RuntimeError.
- Parameters:
- cmdstr
The command to run.
- asynchronouslybool, optional, default=False
Run command asynchronously. If True, nohup will be used to run the command.
- stderr_whitelistlist of str, optional, default=None
If not None, the stderr will be checked against the whitelist. If the stderr contains any of the strings in the whitelist, the command will be considered successful.
- classmethod machine_subfields() List[Argument] [source]
Generate the machine subfields.
- Returns:
- list[Argument]
machine subfields
- property sftp
- property ssh
- class dpdispatcher.ssh_context.SSHSession(hostname, username, password=None, port=22, key_filename=None, passphrase=None, timeout=10, totp_secret=None, tar_compress=True, look_for_keys=True)[source]
Bases:
object
- Attributes:
- remote
- rsync_available
sftp
Returns sftp.
Methods
inter_handler
(title, instructions, prompt_list)inter_handler: the callback for paramiko.transport.auth_interactive.
arginfo
close
ensure_alive
exec_command
get
get_ssh_client
put
- exec_command(**kwargs)
- inter_handler(title, instructions, prompt_list)[source]
inter_handler: the callback for paramiko.transport.auth_interactive.
The prototype for this function is defined by Paramiko, so all of the arguments need to be there, even though we don’t use ‘title’ or ‘instructions’.
The function is expected to return a tuple of data containing the responses to the provided prompts. Experimental results suggests that there will be one call of this function per prompt, but the mechanism allows for multiple prompts to be sent at once, so it’s best to assume that that can happen.
Since tuples can’t really be built on the fly, the responses are collected in a list which is then converted to a tuple when it’s time to return a value.
Experiments suggest that the username prompt never happens. This makes sense, but the Username prompt is included here just in case.
- property sftp
Returns sftp. Open a new one if not existing.
dpdispatcher.submission module
- class dpdispatcher.submission.Job(job_task_list, *, resources, machine=None)[source]
Bases:
object
Job is generated by Submission automatically. A job ususally has many tasks and it may request computing resources from job scheduler systems. Each Job can generate a script file to be submitted to the job scheduler system or executed locally.
- Parameters:
- job_task_listlist of Task
the tasks belonging to the job
- resourcesResources
the machine resources. Passed from Submission when it constructs jobs.
- machinemachine
machine object to execute the job. Passed from Submission when it constructs jobs.
Methods
deserialize
(job_dict[, machine])Convert the job_dict to a Submission class object.
Get the jobs.
serialize
([if_static])Convert the Task class instance to a dictionary.
get_hash
handle_unexpected_job_state
job_to_json
register_job_id
submit_job
- classmethod deserialize(job_dict, machine=None)[source]
Convert the job_dict to a Submission class object.
- Parameters:
- job_dictdict
the dictionary which contains the job information
- machineMachine
the machine object to execute the job
- Returns:
- submissionJob
the Job class instance converted from the job_dict
- get_job_state()[source]
Get the jobs. Usually, this method will query the database of slurm or pbs job scheduler system and get the results.
Notes
this method will not submit or resubmit the jobs if the job is unsubmitted.
- class dpdispatcher.submission.Resources(number_node, cpu_per_node, gpu_per_node, queue_name, group_size, *, custom_flags=[], strategy={'if_cuda_multi_devices': False, 'ratio_unfinished': 0.0}, para_deg=1, module_unload_list=[], module_purge=False, module_list=[], source_list=[], envs={}, prepend_script=[], append_script=[], wait_time=0, **kwargs)[source]
Bases:
object
Resources is used to describe the machine resources we need to do calculations.
- Parameters:
- number_nodeint
The number of node need for each job.
- cpu_per_nodeint
cpu numbers of each node.
- gpu_per_nodeint
gpu numbers of each node.
- queue_namestr
The queue name of batch job scheduler system.
- group_sizeint
The number of tasks in a job.
- custom_flagslist of Str
The extra lines pass to job submitting script header
- strategydict
strategies we use to generation job submitting scripts. if_cuda_multi_devices : bool
If there are multiple nvidia GPUS on the node, and we want to assign the tasks to different GPUS. If true, dpdispatcher will manually export environment variable CUDA_VISIBLE_DEVICES to different task. Usually, this option will be used with Task.task_need_resources variable simultaneously.
- ratio_unfinishedfloat
The ratio of task that can be unfinished.
- para_degint
Decide how many tasks will be run in parallel. Usually run with strategy[‘if_cuda_multi_devices’]
- source_listlist of Path
The env file to be sourced before the command execution.
- wait_timeint
The waitting time in second after a single task submitted. Default: 0.
Methods
arginfo
deserialize
load_from_dict
load_from_json
serialize
- class dpdispatcher.submission.Submission(work_base, machine=None, resources=None, forward_common_files=[], backward_common_files=[], *, task_list=[])[source]
Bases:
object
A submission represents a collection of tasks. These tasks usually locate at a common directory. And these Tasks may share common files to be uploaded and downloaded.
- Parameters:
- work_basePath
the base directory of the local tasks. It is usually the dir name of project .
- machineMachine
machine class object (for example, PBS, Slurm, Shell) to execute the jobs. The machine can still be bound after the instantiation with the bind_submission method.
- resourcesResources
the machine resources (cpu or gpu) used to generate the slurm/pbs script
- forward_common_fileslist
the common files to be uploaded to other computers before the jobs begin
- backward_common_fileslist
the common files to be downloaded from other computers after the jobs finish
- task_listlist of Task
a list of tasks to be run.
Methods
bind_machine
(machine)Bind this submission to a machine.
Check whether all the jobs in the submission.
check_ratio_unfinished
(ratio_unfinished)Calculate the ratio of unfinished tasks in the submission.
deserialize
(submission_dict[, machine])Convert the submission_dict to a Submission class object.
After tasks register to the self.belonging_tasks, This method generate the jobs and add these jobs to self.belonging_jobs.
Handle unexpected job state of the submission.
run_submission
(*[, dry_run, exit_on_submit, ...])Main method to execute the submission.
serialize
([if_static])Convert the Submission class instance to a dictionary.
Check whether all the jobs in the submission.
clean_jobs
download_jobs
get_hash
register_task
register_task_list
remove_unfinished_tasks
submission_from_json
submission_to_json
try_recover_from_json
upload_jobs
- bind_machine(machine)[source]
Bind this submission to a machine. update the machine’s context remote_root and local_root.
- Parameters:
- machineMachine
the machine to bind with
- check_all_finished()[source]
Check whether all the jobs in the submission.
Notes
This method will not handle unexpected job state in the submission.
- check_ratio_unfinished(ratio_unfinished: float) bool [source]
Calculate the ratio of unfinished tasks in the submission.
- Parameters:
- ratio_unfinishedfloat
the ratio of unfinished tasks in the submission
- Returns:
- bool
whether the ratio of unfinished tasks in the submission is larger than ratio_unfinished
- classmethod deserialize(submission_dict, machine=None)[source]
Convert the submission_dict to a Submission class object.
- Parameters:
- submission_dictdict
path-like, the base directory of the local tasks
- machineMachine
Machine class Object to execute the jobs
- Returns:
- submissionSubmission
the Submission class instance converted from the submission_dict
- generate_jobs()[source]
After tasks register to the self.belonging_tasks, This method generate the jobs and add these jobs to self.belonging_jobs. The jobs are generated by the tasks randomly, and there are self.resources.group_size tasks in a task. Why we randomly shuffle the tasks is under the consideration of load balance. The random seed is a constant (to be concrete, 42). And this insures that the jobs are equal when we re-run the program.
- handle_unexpected_submission_state()[source]
Handle unexpected job state of the submission. If the job state is unsubmitted, submit the job. If the job state is terminated (killed unexpectly), resubmit the job. If the job state is unknown, raise an error.
- run_submission(*, dry_run=False, exit_on_submit=False, clean=True)[source]
Main method to execute the submission. First, check whether old Submission exists on the remote machine, and try to recover from it. Second, upload the local files to the remote machine where the tasks to be executed. Third, run the submission defined previously. Forth, wait until the tasks in the submission finished and download the result file to local directory. If dry_run is True, submission will be uploaded but not be executed and exit. If exit_on_submit is True, submission will exit.
- serialize(if_static=False)[source]
Convert the Submission class instance to a dictionary.
- Parameters:
- if_staticbool
whether dump the job runtime infomation (like job_id, job_state, fail_count) to the dictionary.
- Returns:
- submission_dictdict
the dictionary converted from the Submission class instance
- class dpdispatcher.submission.Task(command, task_work_path, forward_files=[], backward_files=[], outlog='log', errlog='err')[source]
Bases:
object
A task is a sequential command to be executed, as well as the files it depends on to transmit forward and backward.
- Parameters:
- commandStr
the command to be executed.
- task_work_pathPath
the directory of each file where the files are dependent on.
- forward_fileslist of Path
the files to be transmitted to remote machine before the command execute.
- backward_fileslist of Path
the files to be transmitted from remote machine after the comand finished.
- outlogStr
the filename to which command redirect stdout
- errlogStr
the filename to which command redirect stderr
Methods
deserialize
(task_dict)Convert the task_dict to a Task class object.
get_task_state
(context)Get the task state by checking the tag file.
arginfo
get_hash
load_from_dict
load_from_json
serialize
- classmethod deserialize(task_dict)[source]
Convert the task_dict to a Task class object.
- Parameters:
- task_dictdict
the dictionary which contains the task information
- Returns:
- taskTask
the Task class instance converted from the task_dict
dpdispatcher.utils module
- exception dpdispatcher.utils.RetrySignal[source]
Bases:
Exception
Exception to give a signal to retry the function.
- dpdispatcher.utils.generate_totp(secret: str, period: int = 30, token_length: int = 6) str [source]
Generate time-based one time password (TOTP) from the secret.
Some HPCs use TOTP for two-factor authentication for safety.
- Parameters:
- secretstr
The encoded secret provided by the HPC. It’s usually extracted from a 2D code and base32 encoded.
- periodint, default=30
Time period where the code is valid in seconds.
- token_lengthint, default=6
The token length.
- Returns:
- token: str
The generated token.
References
- dpdispatcher.utils.get_sha256(filename)[source]
Get sha256 of a file.
- Parameters:
- filenamestr
The filename.
- Returns:
- sha256: str
The sha256.
- dpdispatcher.utils.retry(max_retry: int = 3, sleep: ~typing.Union[int, float] = 60, catch_exception: ~typing.Type[BaseException] = <class 'dpdispatcher.utils.RetrySignal'>) Callable [source]
Retry the function until it succeeds or fails for certain times.
- Parameters:
- max_retryint, default=3
The maximum retry times. If None, it will retry forever.
- sleepint or float, default=60
The sleep time in seconds.
- catch_exceptionException, default=Exception
The exception to catch.
- Returns:
- decorator: Callable
The decorator.
Examples
>>> @retry(max_retry=3, sleep=60, catch_exception=RetrySignal) ... def func(): ... raise RetrySignal("Failed")
- dpdispatcher.utils.rsync(from_file: str, to_file: str, port: int = 22, key_filename: Optional[str] = None, timeout: Union[int, float] = 10)[source]
Call rsync to transfer files.
- Parameters:
- from_filestr
SRC
- to_filestr
DEST
- portint, default=22
port for ssh
- key_filenamestr, optional
identity file name
- timeoutint, default=10
timeout for ssh
- Raises:
- RuntimeError
when return code is not 0