Jobman documentation¶

jobman is a job management package designed to submit and monitor jobs on remote machines. It is built on the top of the dpdispatcher package.

jobman is designed for the big data era, where the number of remoted jobs is large that handling them manually is almost impossible. Imaging that you have more than 1000 jobs to run, you have access to 3 remote high-performance computing (HPC) serves with different computing environment, and you need to monitor the progress of each job, check the output files, and download the results. This is a tedious and time-consuming task. The jobman package is designed to automate such tasks. jobman will handle the input files, submit the jobs to remote machines, monitor the progress of each job, and download the results to the local machine whenever jobs finished.

Case 1: Distribute jobs to a single remote machine¶

This is a general purpose usage, where task_list can be defined flexibly. In this case, each Task can have different command_list, forward_files, backward_files.

To use, just need to:

Define the task_list as list of Task objects.
Use function submit_job_chunk() to submit jobs to remote machines.

from thkit.jobman.submit import submit_job_chunk, Task
from thkit.jobman.helper import loadconfig_multi_machines

mdict_list = loadconfig_multi_machines("MACHINE.yml")  # load the remote machine config
mdict = mdict_list[0]  # use the first machine in the list

task_list = [Task(), Task(), ...]    # list of Task objects
submit_job_chunk(
    mdict=mdict,
    work_dir='./',
    task_list=task_list,
    forward_common_files=forward_common_files,
    backward_common_files=backward_common_files,
)

That's it! jobman will handle the rest.

Info 1: An example of defining task_list:

task = Task.load_from_dict(
    {
        "command": f"mpirun -np lmp_serial -in {runfile}",
        "task_work_path": "./",
        "forward_files": ["all_input_files"],
        "backward_files": ["all_getback_files/*"],
        "outlog": "out.log",
        "errlog": "err.log",
    }
)
task_list = [task]

Info 2: Configure the remote machine in MACHINE.yml file, following the remote machine schema.
(Optional) Use a launcher script (e.g., launcher.sh) to run python code

#!/bin/bash

source /etc/profile.d/modules.sh
module use /home/tha/app/1modulefiles
module load miniforge
source activate py13

python above_py_script.py

Case 2: Distribute jobs to multiple remote machines¶

This is used for specific purpose (e.g., alff package), where the jobs have the same forward_files, backward_files; but the command_list can be different based on computing environment on each remote machine. Just need to:

Prepare the task_dirs, where all of them have the same forward_files, backward_files.
Define a prepare_command_list() function to prepare the command_list for each remote machine.

from thkit.jobman.submit import alff_submit_job_multi_remotes
from thkit.config import loadconfig
import asyncio

mdict = loadconfig("remote_machine.yml")  # load the remote machine config

### Prepare command_list on each machine
def prepare_command_list(machine: dict) -> list:
    command_list = []
    dft_cmd = machine.get("command", "python")
    dft_cmd = f"{dft_cmd} ../cli_gpaw_optimize.py ../{FILE_ARG_ASE}"  # `../` to run file in common directory
    command_list.append(dft_cmd)
    return command_list

### Submit to multiple machines
asyncio.run(
    alff_submit_job_multi_remotes(
        multi_mdict=mdict,
        prepare_command_list=prepare_command_list,
        work_dir=work_dir,
        task_dirs=task_dirs,
        forward_files=forward_files,
        backward_files=backward_files,
        forward_common_files=forward_common_files,
        mdict_prefix="dft",
        logger=logger,
    )
)

Note: - Setting remote machines follow the remote machine schema. - Can import from jobman these classes: Task, Machine, Resources, Submission. - To handle if some tasks is finished and some tasks are not finished, see the function handle_submission()

API Reference:

`thkit.jobman` ¶

Modules:

helper –
submit –

`helper` ¶

Functions:

change_logpath_dispatcher –

Change the logfile of dpdispatcher.
validate_machine_config –

Validate the YAML file contains multiple machines configs. This function is used to validate machine configs at very beginning of program to avoid later errors.
loadconfig_multi_machines –

Load and validate the YAML file contains multiple machine configs. This function to load machine configs for general purpose usage.

`change_logpath_dispatcher(newlogfile: str)` ¶

Change the logfile of dpdispatcher.

`validate_machine_config(machine_file: str)` ¶

Validate the YAML file contains multiple machines configs. This function is used to validate machine configs at very beginning of program to avoid later errors.

Notes

To specify multiple remote machines for the same purpose, the top-level keys in the machine config file should start with the same prefix. Example:
- train_1, train_2,... for training jobs
- lammps_1, lammps_2,... for lammps jobs
- gpaw_1, gpaw_2,... for gpaw jobs

`loadconfig_multi_machines(machine_file: str, mdict_prefix: str = '') -> list[dict]` ¶

Load and validate the YAML file contains multiple machine configs. This function to load machine configs for general purpose usage.

Parameters:

machine_file (str) –

the path of the machine config file

Returns:

dict ( list[dict] ) –

the multi-machine dict

`submit` ¶

Functions:

submit_job_chunk –

Function to submit a jobs to the remote machine. The function will:
async_submit_job_chunk –

Convert submit_job_chunk() into an async function but only need to wait for the completion of the entire for loop (without worrying about the specifics of each operation inside the loop)
alff_submit_job_multi_remotes –

Submit jobs to multiple machines asynchronously.
alff_submit_job_multi_remotes_new –

Submit jobs to multiple machines asynchronously.

`submit_job_chunk(mdict: dict, work_dir: str, task_list: list[Task], forward_common_files: list[str] = [], backward_common_files: list[str] = [], machine_index: int = 0, logger: object = None)` ¶

Function to submit a jobs to the remote machine. The function will:

Prepare the task list
Make the submission of jobs to remote machines
Wait for the jobs to finish and download the results to the local machine

Parameters:

mdict (dict) –

a dictionary contain settings of the remote machine. The parameters described in the remote machine schema. This dictionary defines the login information, resources, execution command, etc. on the remote machine.
task_list (list[Task]) –

a list of Task objects. Each task object contains the command to be executed on the remote machine, and the files to be copied to and from the remote machine. The dirs of each task must be relative to the work_dir.
forward_common_files (list[str], default: [] ) –

common files used for all tasks. These files are i n the work_dir.
backward_common_files (list[str], default: [] ) –

common files to download from the remote machine when the jobs are finished.
machine_index (int, default: 0 ) –

index of the machine in the list of machines.
logger (object, default: None ) –

the logger object to be used for logging.

Note

Split the task_list into chunks to control the number of jobs submitted at once.
Should not use the Local contexts, it will interference the current shell environment which leads to the unexpected behavior on local machine. Instead, use another account to connect local machine with SSH context.

`async_submit_job_chunk(mdict: dict, work_dir: str, task_list: list[Task], forward_common_files: list[str] = [], backward_common_files: list[str] = [], machine_index: int = 0, logger: object = None)` `async` ¶

Convert submit_job_chunk() into an async function but only need to wait for the completion of the entire for loop (without worrying about the specifics of each operation inside the loop)

Note

An async function normally contain a await ... statement to be awaited (yield control to event loop)
If the 'event loop is blocked' by a asynchronous function (it will not yield control to event loop), the async function will wait for the completion of the synchronous function. So, the async function will not be executed asynchronously. Try to use await asyncio.to_thread() to run the synchronous function in a separate thread, so that the event loop is not blocked.

`alff_submit_job_multi_remotes(multi_mdict: dict, prepare_command_list: callable, work_dir: str, task_dirs: list[str], forward_files: list[str], backward_files: list[str], forward_common_files: list[str] = [], backward_common_files: list[str] = [], mdict_prefix: str = 'dft', logger: object = None)` `async` ¶

Submit jobs to multiple machines asynchronously.

Update 2025Sep20: Use alff_submit_job_multi_remotes_new() instead, which is more flexible. This function is kept for backward compatibility.

Parameters:

multi_mdict (dict) –

the big_dict contains multiple mdicts. Each mdict contains parameters of one remote machine, which parameters as in the remote machine schema.
prepare_command_list (callable) –

a function to prepare the command list based on each remote machine.
mdict_prefix (str, default: 'dft' ) –

the prefix to select remote machines for the same purpose. Example: 'dft', 'md', 'train'.

`alff_submit_job_multi_remotes_new(mdict_list: list[dict], commandlist_list: list[list[str]], work_dir: str, task_dirs: list[str], forward_files: list[str], backward_files: list[str], forward_common_files: list[str] = [], backward_common_files: list[str] = [], logger: object = None)` `async` ¶

Submit jobs to multiple machines asynchronously.

Parameters:

mdict_list (list[dict]) –

list of multiple mdicts. Each mdict contains parameters of one remote machine, which parameters as in the remote machine schema.
commandlist_list (list[list[str]]) –

list of command_lists, each list for each remote machine. Need to prepare outside.

Jobman documentation¶

Case 1: Distribute jobs to a single remote machine¶

Case 2: Distribute jobs to multiple remote machines¶

thkit.jobman ¶

helper ¶

change_logpath_dispatcher(newlogfile: str) ¶

validate_machine_config(machine_file: str) ¶

loadconfig_multi_machines(machine_file: str, mdict_prefix: str = '') -> list[dict] ¶

submit ¶

submit_job_chunk(mdict: dict, work_dir: str, task_list: list[Task], forward_common_files: list[str] = [], backward_common_files: list[str] = [], machine_index: int = 0, logger: object = None) ¶

async_submit_job_chunk(mdict: dict, work_dir: str, task_list: list[Task], forward_common_files: list[str] = [], backward_common_files: list[str] = [], machine_index: int = 0, logger: object = None) async ¶

alff_submit_job_multi_remotes_new(mdict_list: list[dict], commandlist_list: list[list[str]], work_dir: str, task_dirs: list[str], forward_files: list[str], backward_files: list[str], forward_common_files: list[str] = [], backward_common_files: list[str] = [], logger: object = None) async ¶

`thkit.jobman` ¶

`helper` ¶

`change_logpath_dispatcher(newlogfile: str)` ¶

`validate_machine_config(machine_file: str)` ¶

`loadconfig_multi_machines(machine_file: str, mdict_prefix: str = '') -> list[dict]` ¶

`submit` ¶

`submit_job_chunk(mdict: dict, work_dir: str, task_list: list[Task], forward_common_files: list[str] = [], backward_common_files: list[str] = [], machine_index: int = 0, logger: object = None)` ¶

`async_submit_job_chunk(mdict: dict, work_dir: str, task_list: list[Task], forward_common_files: list[str] = [], backward_common_files: list[str] = [], machine_index: int = 0, logger: object = None)` `async` ¶

`alff_submit_job_multi_remotes_new(mdict_list: list[dict], commandlist_list: list[list[str]], work_dir: str, task_dirs: list[str], forward_files: list[str], backward_files: list[str], forward_common_files: list[str] = [], backward_common_files: list[str] = [], logger: object = None)` `async` ¶