API

Job(command::Base.AbstractCmd; stdout=nothing, stderr=nothing, append::Bool=false, kwargs...)
Job(f::Function; kwargs...)
Job(task::Task; kwargs...)

Arguments

• command::Base.AbstractCmd: the command to run.
• f::Function: the function to run without any arguments, like f().
• task::Task: the task to run. Eg: @task(1+1).

Common Keyword Arguments (kwargs...)

• name::String = "": job name.
• user::String = "": user that job belongs to.
• ncpu::Real = 1.0: number of CPU this job is about to use (can be Float64, eg: 1.5 will use 150% CPU).
• mem::Integer = 0: number of memory this job is about to use (supports TB, GB, MB, KB, B=1).
• schedule_time::Union{DateTime,Period} = DateTime(0): The expected time to run.
• dependency: defer job until specified jobs reach specified state (QUEUING, RUNNING, DONE, FAILED, CANCELLED, PAST). PAST is the super set of DONE, FAILED, CANCELLED, which means the job will not run in the future. Eg: DONE => job, [DONE => job1; PAST => job2].

• wall_time::Period = Year(1): wall clock time limit. Jobs will be terminated after running for this period.

• priority::Int = 20: lower means higher priority.

• cron::Cron = Cron(:none): job recurring at specfic date and time. See more at Cron.

• until::Union{DateTime,Period} = DateTime(9999,1,1): stop job recurring until date and time.

See also submit!, @submit, Cron

submit!(job::Job)
submit!(args_of_Job...; kwargs_of_Job...)
submit!(p::Pipelines.Program; kwargs_of_p..., kwargs_of_Job..., kwargs_of_run...)

Submit the job to queue.

submit!(Job(...)) can be simplified to submit!(...). They are equivalent.

See also Job, @submit

If using Pipelines, see also JuliaProgram, CmdProgram, and run for their kwargs.

@submit [option=value]... expr

Submit a job from expr. If a Job is explicitly shown in expr, DONE => job will be automatically added to the dependency list.

• expr: any type of Expression. Eg: 1+2, length(ARGS), begin ... end

• option = value: kwargs of Job. If expr is parsed to be a Pipelines.Program, options also include its inputs, outputs and run kwargs.

See also Job, and submit! for detailed kwargs.

If using Pipelines, see also JuliaProgram, CmdProgram, and run for their kwargs.

Example

j = @submit 1+1
wait(j)
@assert result(j) == 2

# you can use any keyword arguments that `Job` supports, such as `name`, `ncpu`:
j_2sec = @submit name = "run after 2 sec" begin sleep(2); 32 end

# because `j_2sec isa Job`, `DONE => j_2sec` is pushed to `j2.dependency`.
j2 = @submit mem=2KB begin
    1 + result(j_2sec)
end

wait(j2)
@assert result(j2) == 1 + 32

# you can also manually add dependencies not in the `expr`:
j3 = @submit dependency = [PAST => j] println("j3 finished. result of j2 = ", result(j2))

jobs = Job[]  # the job container
for i in 1:2
    push!(jobs, @submit begin sleep(30);i end) # 10 jobs will be added to `jobs`
end

x = 0
j_something_wrong = @submit for j in jobs
    # have to use global x
    global x += result(j)
end
# ┌ Warning: Getting result from a running job: returned value might be unexpected.
# └ @ JobSchedulers ~/projects/JobSchedulers.jl/src/jobs.jl:318

result(j_something_wrong)
# MethodError(+, (nothing, nothing), 0x0000000000007b16)

x = 0
j_ok = submit!(dependency = jobs) do
    for j in jobs
        # have to use global x
        global x += result(j)
    end
end
wait(j_ok)
@assert x == 55

x = 100
j_ok_too = @submit dependency = jobs for j in jobs
    # have to use global x
    global x += result(j)
end
wait(j_ok_too)
@assert x == 155

cancel!(job::Job)

Cancel job, stop queuing or running.

result(job::Job)

Return the result of job. If the job is not done, a warning message will also show.

fetch(x::Job)

Wait for a Job to finish, then return its result value. If the task fails with an exception, a TaskFailedException (which wraps the failed task) is thrown.

isqueuing(j::Job) :: Bool

isrunning(j::Job) :: Bool

isdone(j::Job) :: Bool

iscancelled(j::Job) :: Bool

isfailed(j::Job) :: Bool

ispast(j::Job) :: Bool = j.state === DONE || j.state === CANCELLED || j.state === FAILED

@yield_current expr

Used to prevent wasting threads and even blocking JobScheduler when submitting jobs within jobs.

If @yield_current is not called within a job's scope, expr will be evaluated as it is.

If @yield_current is called within a Job's scope, this Job is considered as the current_job.

And within the expr block, the current job's ncpu is temporarily set to 0, and the thread of the current job can be occupied by any child jobs (including grand-child jobs).

Once leaving the expr block, the current job's ncpu is resumed, and its thread is not able to be occupied.

Therefore, the child jobs need to be submitted and waited within the expr block.

Example:

using JobSchedulers

parent_job = Job(ncpu=1) do
    # imagine preparing A and B takes lots of time
    A = 0
    B = 0

    # compute A and B in parallel (imagine they take lots of time)
    @yield_current begin
        child_job_A = @submit A += 100
        child_job_B = @submit B += 1000
        wait(child_job_A)
        wait(child_job_B)
    end

    # do computation based on results of child jobs 
    return A + B
end

submit!(parent_job)
res = fetch(parent_job)
# 1100

If not using @yield_current, the thread taken by the current job is wasted when waiting.

In an extreme condition, the JobScheduler can be blocked if all threads are taken by parent jobs, and if their child jobs do not have any threads to run.

To experience the blockage, you can start Julia with julia -t 1,1, and run the example code without @yield_current. You may kill the julia session by pressing Ctrl+C 10 times.

yield_current(f::Function) = @yield_current f()

The function version of @yield_current.

Prevent wasting threads and even blocking JobScheduler when submitting jobs within jobs.

See details at @yield_current.

current_job() :: Union{Job, Nothing}

Return the Job that is running in the current scope, nothing if the current scope is not within a job.

Cron(second, minute, hour, day_of_month, month, day_of_week)
Cron(;
    second = 0,
    minute = '*',
    hour = '*',
    day_of_month = '*',
    month = '*',
    day_of_week = '*',
)

Cron stores the schedule of a repeative Job, implemented according to Linux-based crontab(5) table.

Jobs are executed when the second, minute, hour and month fields match the current time. If neither day_of_month nor day_of_week starts with *, cron takes the union (∪) of their values day_of_month ∪ day_of_week. Otherwise cron takes the intersection (∩) of their values day_of_month ∩ day_of_week.

When an argument is an Int:

When an argument is a String or Char:

An argument may be an asterisk (*), which always stands for $first-last$.

Ranges of numbers are allowed. Ranges are two numbers separated with a hyphen. The specified range is inclusive. For example, 8-11 for an $hours$ entry specifies execution at hours 8, 9, 10 and 11.

Lists are allowed. A list is a set of numbers (or ranges) separated by commas. Examples: "1,2,5,9", "0-4,8-12".

Step values can be used in conjunction with ranges. Following a range with /<number> specifies skips of the number's value through the range. For example, "0-23/2" can be used in the hour argument to specify Job execution every other hour (the alternative is "0,2,4,6,8,10,12,14,16,18,20,22"). Steps are also permitted after an asterisk, so if you want to say $every two hours$, just use "*/2".

When an argument is a Vector:

Vector works like lists mentioned above. For example, [1,2,5,9] is equivalent to "1,2,5,9".

When an argument is a UInt64:

UInt64 is the internal type of Cron fileds. All the previous types will be converted to a UInt64 bit array. The start index of the bit array is 0. Bits outside of the allowed values (see the table above) are ignored.

Cron(special::Symbol)

Instead of the six arguments of Cron, one of the following special symbols may appear instead:

Caution: Linux crontab's special :reboot is not supported here.

To run every minute, just use Cron().

cron_value_parse(value::UInt64)
cron_value_parse(value::Signed)
cron_value_parse(value::AbstractString)
cron_value_parse(value::Char)
cron_value_parse(value::Vector)
cron_value_parse(*) = cron_value_parse('*')

Parse crontab-like value to UInt64. See details: Cron.

Dates.tonext(dt::DateTime, c::Cron; same::Bool = false) -> Union{DateTime, Nothing}
Dates.tonext(t::Time, c::Cron; same::Bool = false) -> Time
Dates.tonext(d::Date, c::Cron; same::Bool = false, limit::Date = d + Day(3000)) -> Union{DateTime, Nothing}

Adjust date or time to the next one corresponding to c::Cron. Setting same=true allows the current date or time to be considered as the next one, allowing for no adjustment to occur.

queue(; all::Bool = false)    -> Vector{Job}
queue(state::Symbol )         -> Vector{Job}
queue(needle)                 -> Vector{Job}
queue(state::Symbol , needle) -> Vector{Job}
queue(needle, state::Symbol ) -> Vector{Job}
queue(id::Integer)            -> Job

• all::Bool: if true, get all jobs. if false, get only running and queuing jobs.

• state::Symbol: get jobs with a specific state, including :all, QUEUING, RUNNING, DONE, FAILED, CANCELLED, PAST.

  PAST is the superset of DONE, FAILED, CANCELLED.

• needle::Union{AbstractString,AbstractPattern,AbstractChar}: get jobs if they contain needle in their name or user.

• id::Integer: get the job with the specific id.

all_queue()
all_queue(id::Integer)
all_queue(state::Symbol)
all_queue(needle::Union{AbstractString,AbstractPattern,AbstractChar})

• state::Symbol: get jobs with a specific state, including :all, QUEUING, RUNNING, DONE, FAILED, CANCELLED, PAST.

  PAST is the superset of DONE, FAILED, CANCELLED.

• needle::Union{AbstractString,AbstractPattern,AbstractChar}: get jobs if they contain needle in their name or user.

• id::Integer: get the job with the specific id.

job_query_by_id(id::Integer)

Search job by job.id in the queue.

Return job::Job if found, nothing if not found.

job_query_by_id(id::Integer)

Search job by job.id in the queue.

Return job::Job if found, nothing if not found.

wait_queue(;show_progress::Bool = false, exit_num_jobs::Int = 0)

Wait for all jobs in queue() become finished.

• show_progress = true, job progress will show.

• exit_num_jobs::Int: exit when queue() has less than Int number of jobs. It is useful to ignore some jobs that are always running or recurring.

See also: queue_progress.

wait(j::Job)
wait(js::Vector{Job})

Wait for the job(s) to be finished.

set_scheduler(;
    max_cpu::Real = JobSchedulers.SCHEDULER_MAX_CPU,
    max_mem::Real = JobSchedulers.SCHEDULER_MAX_MEM,
    max_job::Int = JobSchedulers.JOB_QUEUE.max_done,
    max_cancelled_job::Int = JobSchedulers.JOB_QUEUE.max_cancelled_job
)

• max_job: the number of jobs done. If number of jobs exceed 1.5*NUMBER, old jobs will be delete.
• max_cancelled_job: the number of cancelled jobs. If number of jobs exceed 1.5*NUMBER, old jobs will be delete.

See details: set_scheduler_max_cpu, set_scheduler_max_mem, set_scheduler_max_job

set_scheduler_max_cpu(ncpu::Int = default_ncpu())
set_scheduler_max_cpu(percent::Float64)

Set the maximum CPU (thread) the scheduler can use. If starting Julia with multiple threads in the default thread pool, the maximum CPU is the number of tids in the default thread pool not equal to 1.

Example

set_scheduler_max_cpu()     # use all available CPUs
set_scheduler_max_cpu(4)    # use 4 CPUs
set_scheduler_max_cpu(0.5)  # use 50% of CPUs

set_scheduler_max_mem(mem::Integer = default_mem())
set_scheduler_max_mem(percent::AbstractFloat)

Set the maximum RAM the scheduler can use.

Example

set_scheduler_max_mem()             # use 80% of total memory

set_scheduler_max_mem(4GB)          # use 4GB memory
set_scheduler_max_mem(4096MB)
set_scheduler_max_mem(4194304KB)
set_scheduler_max_mem(4294967296B)

set_scheduler_max_mem(0.5)          # use 50% of total memory

set_scheduler_max_job(max_done::Int = 10000, max_cancelled::Int = max_done)

Set the number of finished jobs. If number of jobs exceed 1.5*NUMBER, old jobs will be delete.

JobSchedulers.destroy_unnamed_jobs_when_done(b::Bool)

set_group_seperator(group_seperator::Regex) = global GROUP_SEPERATOR = group_seperator

Set the group seperator. Group seperator is used to group the names of Jobs. Used when display the progress meter using wait_queue(show_progress=true)

Default is r": *".

GROUP_SEPERATOR::Regex = r": *"

Group seperator is used to group the names of Jobs. Used when display the progress meter using wait_queue(show_progress=true)

To set it, use set_group_seperator(group_seperator::Regex).

scheduler_status() :: Symbol

Print the settings and status of job scheduler. Return :not_running or :running.

scheduler_start()

Start the job scheduler.

scheduler_stop()

Stop the job scheduler.

solve_optimized_ncpu(default::Int; 
    ncpu_range::UnitRange{Int} = 1:total_cpu, 
    njob::Int = 1, 
    total_cpu::Int = JobSchedulers.SCHEDULER_MAX_CPU, 
    side_jobs_cpu::Int = 0)

Find the optimized number of CPU for a job.

• default: default ncpu of the job.
• ncpu_range: the possible ncpu range of the job.
• njob: number of the same job.
• total_cpu: the total CPU that can be used by JobSchedulers.
• side_jobs_cpu: some small jobs that might be run when the job is running, so the job won't use up all of the resources and stop small tasks.

set_scheduler_backup(
    filepath::AbstractString = "";
    migrate::Bool = false,
    delete_old::Bool = false,
    recover_settings::Bool = true,
    recover_queue::Bool = true
)

Set the backup file of job scheduler.

If filepath was set to "", stop backup at exit.

If filepath was set to an existing file, recover_settings or recover_queue from filepath immediately.

If filepath was set to a new file, the backup file will be created at exit.

If migrate=true and the old JobSchedulers.SCHEDULER_BACKUP_FILE exists, the old backup file will be recovered before recovering from filepath.

backup()

Manually backup job scheduler settings and queues. The function is automatically triggered at exit.

const THREAD_POOL = Base.RefValue{Channel{Int}}()

Defined in __init__().

If version > 1.9, THREAD_POOL contains only tids in Threads.threadpooltids(:default).

Also, the thread 1 is reserved for JobScheduler.

const SINGLE_THREAD_MODE = Base.RefValue{Bool}()

Defined in __init__(). Whether Threads.threadpooltids(:default) are empty or == [1].

const TIDS = Vector{Int}()

Defined in __init__(). All tids in the default thread pool, excluding tid 1.

const CURRENT_JOB = ScopedValue{Union{Job, Nothing}}(nothing)

The Job that is running in the current scope, nothing if the current scope is not within a job.

See also current_job.

const OCCUPIED_MARK = 1<<30

A large number added to job._thread_id to mark the thread is yielded.

schedule_thread(j::Job) :: UInt8

Internal: assign TID to and run the job.

free_thread(j::Job)

Make thread available again after job is finished.

is_tid_ready_to_occupy(j::Job) :: Bool

Return Bool. true means job is able to occupy (j.ncpu == 0 && j._thread_id > 0) and not occupied (j._thread_id & OCCUPIED_MARK == 0).

is_tid_occupied(j::Job) :: Bool

Return Bool. false means job not occupied (j._thread_id & OCCUPIED_MARK != 0) or not able to occupy (j._thread_id <= 0).

unsafe_occupy_tid!(j::Job) :: Int

Return the j._thread_id before changing, and mark the tid as occupied by adding OCCUPIED_MARK to j._thread_id.

Unsafe when:

• If j._thread_id <= 0 or j is occupied, j._thread_id will not be changed, and return j._thread_id.

unsafe_unoccupy_tid!(j::Job) :: Bool

Remove OCCUPIED_MARK from j._thread_id to mark the tid is ready to yield. Please ensure the tid is occupied (is_tid_occupied(j)) before calling this function.

Unsafe when:

• If j._thread_id < 0, the tid will be changed to a very small negative number, which does not make sense.
• If j._thread_id == 0, the tid will remain 0.
• If j is not occupied, the tid will remain unchanged.

@inline unsafe_original_tid(j::Job) = j._thread_id & ~OCCUPIED_MARK

Only useful when j._thread_id > 0. Used to get the original tid of a job, without OCCUPIED_MARK.

LinkedJobList()
LinkedJobList(j::Job)

Every job can only be in ONE LinkedJobList. All iterative copy of LinkedJobList will output Vector{Job}.

deleteat!(l::LinkedJobList, node::Job)

Delete node::Job from l::LinkedJobList.

Caution: user needs to confirm that node::Job shall in l::LinkedJobList. This function will not check it.

JobQueue(; max_done::Int = 10000, max_cancelled::Int = 10000)
mutable struct JobQueue
    const queuing::SortedDict{Int,LinkedJobList,Base.Order.ForwardOrdering}  # priority => Job List
    const queuing_0cpu::LinkedJobList              # ncpu = 0, can run immediately
    const future::LinkedJobList                    # all jobs with schedule_time > now()
    const running::LinkedJobList
    const done::Vector{Job}
    const failed::Vector{Job}
    const cancelled::Vector{Job}
    max_done::Int
    max_cancelled::Int
    const lock_queuing::ReentrantLock
    const lock_running::ReentrantLock
    const lock_past::ReentrantLock
end

scheduler()

The function of running Job's scheduler. It needs to be called by scheduler_start(), rather than calling directly.

unsafe_run!(job::Job, current::DateTime=now()) :: UInt8

Jump the queue and run job immediately, no matter what other jobs are running or waiting.

Return:

• OK: successfully scheduled to run.
• SKIP: no thread is available, skip this time.
• FAIL: failed to schedule.

Caution: it will not trigger scheduler_need_action().

unsafe_cancel!(job::Job, current::DateTime=now())

Caution: it is unsafe and should only be called within lock. Do not call from other module.

Caution: it will not trigger scheduler_need_action().

unsafe_update_state!(job::Job)

Update the state of job from job.task when job.state === :running.

Caution: it is unsafe and should only be called within lock.

is_dependency_ok(job::Job)::Bool

Caution: run it within lock only.

Algorithm: Break while loop when found dep not ok, and change job._dep_check_id to the current id.

If dep is provided as Integer, query Integer for job and then replace Integer with the job.

set_scheduler_while_loop(b::Bool)

if set to false, the scheduler will stop.

get_priority(job::Job) = job.priority

get_thread_id(job::Job) = job._thread_id

date_based_on(c::Cron) -> Symbol

Whether date of c is based on :day_of_week, :day_of_month, :union, :intersect, :everyday, :none, or :undefined.

next_recur_job(j::Job) -> Union{Job, Nothing}

Based on j.cron and j.until, return a new recurring Job or nothing.

mutable struct JobGroup
    name::String
    total::Int
    queuing::Int
    running::Int
    done::Int
    failed::Int
    cancelled::Int
end

JobGroup is computed when displaying a progress meter.

get_group(job::Job, group_seperator = GROUP_SEPERATOR)
get_group(name::AbstractString, group_seperator = GROUP_SEPERATOR)

Return nested_group_names::Vector{String}.

Eg: If job.name is "A: B: 1232", return ["A", "A: B", "A: B: 1232"]

progress_bar(percent::Float64, width::Integer = 20)

Return ::String for progress bar whose char length is width.

• percent: range from 0.0 - 1.0, or to be truncated.

• width: should be > 3. If <= 10, percentage will not show. If > 10, percentage will show.

queue_progress(;remove_tmp_files::Bool = true, kwargs...)
queue_progress(stdout_tmp::IO, stderr_tmp::IO;
group_seperator = GROUP_SEPERATOR, wait_second_for_new_jobs::Int = 1, loop::Bool = true, exit_num_jobs::Int = 0)

• group_seperator: delim to split (job::Job).name to group and specific job names.

• wait_second_for_new_jobs::Int: if auto_exit, and all jobs are PAST, not quiting queue_progress immediately but wait for a period. If new jobs are submitted, not quiting queue_progress.

• loop::Bool: if false, only show the current progress and exit.

• exit_num_jobs::Int: exit when queue() has less than Int number of jobs. It is useful to ignore some jobs that are always running or recurring.

view_update(h, w; row = 1, groups_shown::Vector{JobGroup} = JobGroup[], is_in_terminal::Bool = true, is_interactive = true, group_seperator_at_begining = Regex("^" * GROUP_SEPERATOR.pattern))

Update the whole screen view.

Bool. Showing progress meter? Related to progress computation and display. true when waitqueue(showprogress=true)

update_group_state!(job::Job)

This should only be called if JobSchedulers.PROGRESS_METER == true. Update the job's group state, which will be used in Progress Meter.

init_group_state!()

Prepare group state for existing jobs