Skip to content

dummy

current_jobs(ctx, match) async

Dummy method to find current jobs.

Parameters:

Name Type Description Default
ctx object

Context object for retrieving configuration.

 required
match str

Jobs to match the job list with.

 required

Returns:

Type Description
list

Current jobs.
Source code in model_ensembler/cluster/dummy.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def current_jobs(ctx, match):
    """Dummy method to find current jobs.

    Args:
        ctx (object): Context object for retrieving configuration.
        match (str): Jobs to match the job list with.

    Returns:
        (list): Current jobs.
    """
    global _jobs

    job_arr = [el for el in _jobs.values()
               if el.name.startswith(match)
               and el.state in START_STATES]

    return job_arr

find_id(job_id) async

Dummy method to find local job id.

Parameters:

Name Type Description Default
job_id int

Local job identifier.

 required

Returns:

Type Description
int

job id.

Raises:

Type Description
LookupError

If job id not found.
Source code in model_ensembler/cluster/dummy.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
async def find_id(job_id):
    """Dummy method to find local job id.

    Args:
        job_id (int): Local job identifier.

    Returns:
        (int): job id.

    Raises:
        LookupError: If job id not found.
    """
    global _jobs

    job = None
    job_arr = [el for el in _jobs.values() if el.name == job_id]

    if len(job_arr) == 1:
        job = job_arr[0]
    elif len(job_arr) > 1:
        raise LookupError("{} jobs found for ID {}".format(len(job_arr),
                                                           job_id))
    return job

submit_job(ctx, script=None) async

Dummy method to submit job locally.

Parameters:

Name Type Description Default
ctx object

Context object for retrieving configuration.

 required
script str

Script name to submit.

 None

Returns:

Type Description
int

Job ID.
Source code in model_ensembler/cluster/dummy.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def submit_job(ctx, script=None):
    """Dummy method to submit job locally.

    Args:
        ctx (object): Context object for retrieving configuration.
        script (str): Script name to submit.

    Returns:
        (int): Job ID.
    """
    # TODO: ugh, we could use contextvars for this
    global _jobs
    args = Arguments()

    max_submit_sleep = args.max_stagger
    sleep_for = random.randint(0, max_submit_sleep)
    logging.debug(
        "Sleeping for {} seconds before submission".format(sleep_for))
    await asyncio.sleep(sleep_for)

    with _dict_lock:
        _jobs[ctx.dir] = Job(ctx.id, "SUBMITTED", False, False)

    threading.Thread(target=threaded_job, args=(ctx.dir, script)).start()
    return ctx.id

threaded_job(run_dir, script)

Dummy method to set off local job

Parameters:

Name Type Description Default
run_dir str

Directory script is running in.

 required
script str

Name of script to run.

 required
Source code in model_ensembler/cluster/dummy.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def threaded_job(run_dir, script):
    """Dummy method to set off local job

    Args:
        run_dir (str): Directory script is running in.
        script (str): Name of script to run.
    """
    global _jobs

    with _dict_lock:
        _jobs[run_dir] = Job(_jobs[run_dir].name,
                             "RUNNING",
                             True,
                             False)

    logging.info("DUMMY RUN: {} - {}".format(run_dir, _jobs[run_dir]))
    subprocess.run("./{}".format(script), cwd=run_dir)

    # TODO: failed
    with _dict_lock:
        _jobs[run_dir] = Job(_jobs[run_dir].name,
                             "COMPLETED",
                             True,
                             True)