9.1. Jobs and threads
The JOBQUE module provides low-level job queue and threading primitives.
It includes thread-safe channels for inter-thread communication, lock boxes
for shared data access, job status tracking, and fine-grained thread
management. For higher-level job abstractions, see jobque_boost.
See Job Queue (jobque) for a hands-on tutorial.
All functions and symbols are in “jobque” module, use require to get access to it.
require jobque
Example:
require jobque
[export]
def main() {
with_atomic32() $(counter) {
counter |> set(10)
print("value = {counter |> get}\n")
let after_inc = counter |> inc
print("after inc = {after_inc}\n")
let after_dec = counter |> dec
print("after dec = {after_dec}\n")
}
}
// output:
// value = 10
// after inc = 11
// after dec = 10
9.1.1. Handled structures
- LockBox
Lockbox. Similar to channel, only for single object.
- Stream
- Stream.isEmpty(): bool
Whether the channel or pipe contains no remaining elements.
- Stream.total(): int
Total number of elements that have been added to the pipe.
- Properties:
isEmpty : bool
total : int
FIFO pipe of opaque byte buffers shared between contexts and threads. Built on top of JobStatus so it supports refcounting and join. Unlike Channel, a Stream stores raw bytes copied into runtime-owned memory, so the consumer can safely free a popped buffer even if the producer’s context is long gone. Typically used together with the push_archive / pop_archive / gather_archive helpers in jobque_boost to ship serialized command records across thread boundaries.
- Atomic64
Atomic 64 bit integer.
- Atomic32
Atomic 32 bit integer.
- JobStatus
- JobStatus.isReady(): bool
Whether the job has completed execution.
- JobStatus.isValid(): bool
Whether the job status object refers to a valid, active job.
- JobStatus.size(): int
Returns the current entry count of the JobStatus or Channel.
- Properties:
isReady : bool
isValid : bool
size : int
Job status indicator (ready or not, as well as entry count).
- Channel
- Channel.isEmpty(): bool
Whether the channel or pipe contains no remaining elements.
- Channel.total(): int
Total number of elements that have been added to the pipe.
- Properties:
isEmpty : bool
total : int
Channel provides a way to communicate between multiple contexts, including threads and jobs. Channel has internal entry count.
9.1.2. Channel, JobStatus, Lockbox, Stream
- add_ref(status: JobStatus? )
Increases the reference count of a JobStatus or Channel, preventing premature deletion.
- Arguments:
status : JobStatus? implicit
- append(channel: JobStatus?; size: int ): int
Increases the entry count of the channel, signaling that new work has been added.
- Arguments:
channel : JobStatus? implicit
size : int
- channel_create(): Channel?
Warning
This is unsafe operation.
Creates a new Channel for inter-thread communication and synchronization.
- channel_remove(channel: Channel?& )
Warning
This is unsafe operation.
Destroys a Channel and releases its resources.
- Arguments:
channel : Channel?& implicit
- job_status_create(): JobStatus?
Creates a new JobStatus object for tracking the completion state of asynchronous jobs.
- job_status_remove(jobStatus: JobStatus?& )
Warning
This is unsafe operation.
Destroys a JobStatus object and releases its resources.
- Arguments:
jobStatus : JobStatus?& implicit
- join(job: JobStatus? )
Blocks the current thread until the job or channel’s entry count reaches zero, indicating all work is complete.
- Arguments:
job : JobStatus? implicit
- lock_box_create(): LockBox?
Creates a new LockBox for thread-safe shared access to a single value.
- lock_box_remove(box: LockBox?& )
Warning
This is unsafe operation.
Destroys a LockBox and releases its resources.
- Arguments:
box : LockBox?& implicit
- notify(job: JobStatus? )
Decreases the channel’s entry count, signaling that one unit of work has completed.
Use notify when the caller does not own a reference to the channel — for example
when a Channel? is passed as a plain function argument via invoke_in_context.
In that scenario no lambda captures the channel, so no extra reference was added and
there is nothing to release.
Compare with notify_and_release, which additionally releases a reference and should
be used inside lambdas that captured the channel (adding a reference).
- Arguments:
job : JobStatus? implicit
- notify_and_release(job: JobStatus?& )
Decreases the entry count and the reference count of a Channel or JobStatus
in a single operation. After the call the channel/status variable is set to null.
Use notify_and_release inside lambdas that captured the channel. Capturing adds a
reference, so the lambda must release it when done. This function combines
notify + release into one atomic step and nulls the variable to prevent
accidental reuse.
If the caller does not own a reference (e.g. the channel was passed as a plain
argument via invoke_in_context, with no lambda capture), use notify instead —
calling notify_and_release in that case would release a reference the caller never
added, leading to a premature free.
- Arguments:
job : JobStatus?& implicit
- release(status: JobStatus?& )
Decreases the reference count of a JobStatus or Channel; the object is deleted when the count reaches zero.
- Arguments:
status : JobStatus?& implicit
- stream_create(): Stream?
Warning
This is unsafe operation.
Creates a new empty Stream, a FIFO of opaque byte buffers owned by the runtime. Payload bytes are copied into the stream on push and produced as non-owning views on pop / gather. Either side can safely destroy the stream — memory belongs to the stream itself rather than to a particular context.
- stream_remove(stream: Stream?& )
Warning
This is unsafe operation.
Releases the caller’s reference to a Stream and sets the local pointer to null. When the last reference is released the stream is destroyed along with any buffered byte blobs it still owns.
- Arguments:
stream : Stream?& implicit
9.1.3. Queries
- count_jobque_leaks(): uint64
Returns the number of JobStatus, Channel, LockBox, Stream and Feature objects currently alive (globally tracked). Compare this count before and after a teardown to assert that a job/channel/lockbox/stream cycle leaked nothing — an absolute value is not meaningful because supporting infrastructure may hold some.
- get_total_hw_cores(): int
Returns the number of physical CPU cores on the system, with SMT/hyperthreading siblings collapsed (compare get_total_hw_threads, which counts logical processors). Falls back to the logical count where the core topology cannot be queried.
- get_total_hw_jobs(): int
Returns the total number of hardware threads allocated to the job system.
- get_total_hw_threads(): int
Returns the total number of hardware threads available on the system.
- is_job_que_available(): bool
Returns true when a job queue exists for the current process — inside a with_job_que block or after create_job_que — i.e. parallel dispatch can be used.
- is_job_que_shutting_down(): bool
Returns true if the job queue infrastructure is shutting down or has not been initialized.
9.1.4. Event tracing
- jobque_trace_save(path: string ): bool
Writes the recorded per-lane trace events to path as a Perfetto/Chrome trace-event JSON file (one track per lane), returning false when there is no recorded trace or the file cannot be written.
- Arguments:
path : string implicit
- jobque_trace_start(events_per_lane: int )
Arms the per-lane JobQue event tracer: every lane (each worker plus the dispatching caller) records publish, chunk, stage-wait, wake and fifo-job events into a preallocated buffer of max(events_per_lane, 1024) entries; a lane stops recording when its buffer fills. Start (and stop) the trace only while no dispatch is in flight — arming reallocates the lane buffers.
- Arguments:
events_per_lane : int
- jobque_trace_stop()
Stops the per-lane JobQue event tracer armed by jobque_trace_start. Recorded events stay in the lane buffers until the next jobque_trace_start, so jobque_trace_save can still export them after stopping.
- jobque_trace_tag(tag: int )
Sets the application-defined operation tag stamped into subsequently recorded trace events (0 = untagged). Callers tag dispatch phases so a trace viewer can color chunks by operation kind.
- Arguments:
tag : int
9.1.5. Internal invocations
- flush_jobque_batch()
Publish this thread’s pending batched fork jobs to the queue NOW instead of at join (a no-op when nothing is pending or batching is off — see set_jobque_batch_dispatch). Pair with jobque_try_run_one when the dispatching thread participates in the work: the batch must be visible in the fifo before the caller starts popping from it.
- get_jobque_team_mode(): bool
Returns true when team dispatch mode is currently enabled on the job queue. Returns false when there is no active job queue or team mode is off.
- get_jobque_team_prof(): float4
Returns average per-operation team dispatch times in nanoseconds as float4: x = publish, y = serve, z = join tail, w = total.
- get_jobque_team_prof_counts(): int4
Returns team dispatch profiling counters as int4: x = operations, y = chunks published, z = caller-served chunks, w = operations served solo by the caller.
- get_jobque_team_prof_react(): float2
Returns worker claim reaction times as float2: x = average delay from publish to the first chunk claim, y = to the last, in nanoseconds.
- get_jobque_team_rank_gate(): bool
Returns whether the per-op team rank gate is currently enabled.
- get_jobque_worker_limit(): int
Returns the current worker limit set by set_jobque_worker_limit. When no limit is active the value is a large sentinel (INT_MAX) — not 0, which is a real limit meaning all workers dormant.
- jobque_try_run_one(): bool
Pop one queued job off the fifo and run it on the calling thread (the same cloned-closure job a worker would run — full capture semantics), returning false if the fifo is empty; any batch pending on this thread (see set_jobque_batch_dispatch) is flushed first. This is dispatcher-side work stealing: parallel_for loops it until its wait group is ready, so the dispatching core computes chunks instead of sleeping through the fork.
- new_debugger_thread(block: block<():void> )
Creates a new debugger tick thread for servicing debug connections.
- Arguments:
block : block<void> implicit
- new_job_invoke(lambda: lambda<():void>; function: function<():void>; lambdaSize: int )
Clones the current context, moves the attached lambda into it, and submits it to the job queue.
- Arguments:
lambda : lambda<void>
function : function<void>
lambdaSize : int
- new_thread_invoke(lambda: lambda<():void>; function: function<():void>; lambdaSize: int )
Clones the current context, moves the attached lambda into it, and runs it on a new dedicated thread.
- Arguments:
lambda : lambda<void>
function : function<void>
lambdaSize : int
- reset_jobque_team_prof()
Resets the accumulated team dispatch profiling counters to zero.
- set_jobque_batch_dispatch(batch: bool )
Enable batched fork-job dispatch on the calling thread: new_job keeps the prepared job closures locally, and join publishes the whole batch to the queue under one lock with a single wake, instead of one push (and one worker wake) per job. Only safe for pure fork/join use where every dispatched job is joined via its wait group — a job whose results are consumed before join would stall, since jobs enter the queue at the join. with_job_que exit and destroy_job_que flush stragglers and reset the flag, so the mode cannot leak past its block.
- Arguments:
batch : bool
- set_jobque_fork_pool(keep: bool; skip_init: bool )
Opt into pooling of per-job fork contexts on the current context: when keep is true, the contexts cloned for new_job dispatches are reused across jobs instead of being cloned and destroyed every time, removing the per-dispatch clone cost. When skip_init is true the pooled forks are cloned without running the global init (and, symmetrically, shutdown) script, which is only safe for pure-data jobs such as a parallel_for over numeric data that touch no globals and hold no Features referencing the fork.
- Arguments:
keep : bool
skip_init : bool
- set_jobque_fork_skip_heap_reset(skip: bool )
Skip the heap reset when a pooled fork context is reused for a new_job dispatch: when skip is true, acquiring a fork from the pool no longer resets its linear and string heaps, removing that work from the per-dispatch path. This is only safe for pure-compute jobs that never leak onto the fork heap: in the pooled new_job path the only fork-heap allocation is the job lambda’s capture, which is freed LIFO when the job finishes, so the heap does not grow across reuses. A job that allocates on the fork heap without freeing would accumulate across reuses, so leave this off unless the dispatched work is pure data processing such as a parallel_for matmul.
- Arguments:
skip : bool
- set_jobque_join_spin(level: int )
Set the fork/join join-poll level: JobStatus join polls the remaining-counter for level×1024×128 relax-rounds before parking on the condvar (0 = park immediately, the default). The level shares ggml’s –poll denomination (0..100, 50 = their default), so A/Bs against llama.cpp compare like for like. With parallel_for executing chunks on the calling thread, the join wait is just the workers’ in-flight tail — the poll removes the last OS park/wake of the fork/join path. Opt in for fork/join-heavy compute; leave off where idle CPU matters.
- Arguments:
level : int
- set_jobque_team_mode(on: bool )
Enables or disables team dispatch mode on the current job queue (sticky, opt-in). In team mode, workers self-serve chunks off a single published atomic instead of taking one fifo job per chunk, which makes team_parallel_for cost one atomic bump per dispatch. Must be called inside a with_job_que block. Pairs with set_jobque_worker_spin for the hybrid poll/park window.
- Arguments:
on : bool
- set_jobque_team_prof(on: bool )
Enables or disables team dispatch profiling, which accumulates per-operation publish, serve, and join timing counters for team_parallel_for dispatches.
- Arguments:
on : bool
- set_jobque_team_rank_gate(on: bool )
Enables or disables the per-op team rank gate: when enabled, a worker serves a published team operation only if its rank fits within that operation’s chunk count, so small operations do not wake the whole pool.
- Arguments:
on : bool
- set_jobque_threads_cap(cap: int )
Caps the default worker count of job queues created afterwards: the effective default becomes the minimum of the stock rule and the cap. A cap of 0 turns the cap off, and the DAS_JOBQUE_THREADS environment variable still overrides everything.
- Arguments:
cap : int
- set_jobque_worker_limit(limit: int )
Limits how many job queue workers may serve work: workers with an index at or above the limit park dormant and wake only when the limit is raised or the queue shuts down. 0 parks every worker (the caller still completes all dispatches); a negative value removes the limit.
- Arguments:
limit : int
- set_jobque_worker_spin(usec: int )
Set the idle-worker spin window in microseconds (0 = park on the condvar immediately, the default): a worker that runs out of jobs spin-polls the queue for this long before blocking, so a fork/join burst of back-to-back parallel_for dispatches never pays the OS thread-wake per job — a wake of a parked worker costs the dispatching thread several microseconds, serially, per job. Opt-in because spinning burns idle CPU and boost headroom: enable it for fork/join-heavy compute (e.g. LLM decode, typically together with set_jobque_batch_dispatch), leave it off where idle CPU matters (wasm, audio, low-core boxes).
- Arguments:
usec : int
- team_parallel_for_indexed_invoke(range_begin: int; range_end: int; num_chunks: int; lambda: lambda<():void>; function: function<():void>; lambdaSize: int )
Internal implementation of team_parallel_for_indexed. Identical to team_parallel_for_invoke except the lambda is invoked as (slot, job_begin, job_end), where slot is the claiming worker’s index (the caller participates as the last slot, get_total_hw_jobs()). No two in-flight invocations share a slot, so per-slot scratch needs no locks; degenerate paths (team mode off, trivial range or pool) run one inline invoke over the whole range as slot 0. Not called directly — team_parallel_for_indexed rewrites to it via a function macro.
- Arguments:
range_begin : int
range_end : int
num_chunks : int
lambda : lambda<void>
function : function<void>
lambdaSize : int
- team_parallel_for_invoke(range_begin: int; range_end: int; num_chunks: int; lambda: lambda<():void>; function: function<():void>; lambdaSize: int )
Internal implementation of team_parallel_for. Partitions the range into num_chunks pieces with the same split as parallel_for, clones the lambda once per worker, and runs the chunks through the job queue’s team dispatch path (workers self-serve chunk indices off one atomic; the caller participates and spin-joins). Falls back to a single inline invoke over the whole range when team mode is off or the range and pool are trivial. Not called directly — team_parallel_for rewrites to it via a function macro.
- Arguments:
range_begin : int
range_end : int
num_chunks : int
lambda : lambda<void>
function : function<void>
lambdaSize : int
- team_parallel_stages_invoke(stages: array<int3>; lambda: lambda<():void>; function: function<():void>; lambdaSize: int )
Internal implementation of team_parallel_stages. One team rendezvous runs the given stages in order with a worker-side barrier between consecutive stages, so a later stage may read anything earlier stages wrote; each stage is int3(range_begin, range_end, num_chunks), partitioned like team_parallel_for, and the lambda is cloned once per worker for the whole chain. Not called directly — team_parallel_stages rewrites to it via a function macro.
- Arguments:
stages : array<int3> implicit
lambda : lambda<void>
function : function<void>
lambdaSize : int
9.1.6. Construction
- create_job_que()
Creates a persistent job queue (worker thread pool) and pins it so a nested with_job_que block cannot tear it down. Unlike with_job_que, which is block-scoped, this queue lives until destroy_job_que is called, so new_job can be dispatched from a frame loop that is driven externally (for example emscripten’s asynchronous main loop, where a with_job_que block cannot span the per-frame calls). Idempotent if a persistent queue already exists.
- destroy_job_que()
Releases the persistent job queue created by create_job_que. Intended to be called at shutdown once no more jobs will be dispatched. When the persistent queue is the sole owner (the common case) it is torn down immediately, draining and joining its worker threads; if another owner still holds it — for example an active with_job_que block — teardown is deferred until that owner also releases it. Has no effect if no persistent queue is active.
9.1.6.1. with_channel
- with_channel(block: block<(Channel?):void> )
Creates a Channel scoped to the given block and automatically destroys it afterward.
- Arguments:
block : block<( Channel?):void> implicit
- with_channel(count: int; block: block<(Channel?):void> )
- with_job_que(block: block<():void> )
Ensures job queue infrastructure is initialized for the duration of the block.
- Arguments:
block : block<void> implicit
- with_job_status(total: int; block: block<(JobStatus?):void> )
Creates a JobStatus scoped to the given block and automatically destroys it afterward.
- Arguments:
total : int
block : block<( JobStatus?):void> implicit
- with_lock_box(block: block<(LockBox?):void> )
Creates a LockBox scoped to the given block and automatically destroys it afterward.
- Arguments:
block : block<( LockBox?):void> implicit
9.1.6.2. with_stream
- with_stream(block: block<(Stream?):void> )
Creates an unbounded Stream, invokes the given block with a stable pointer to it, and automatically releases the reference when the block returns. The simplest way to scope a stream to a producer/consumer pair.
- Arguments:
block : block<( Stream?):void> implicit
- with_stream(count: int; block: block<(Stream?):void> )
9.1.7. Atomic
- atomic32_create(): Atomic32?
Creates an Atomic32 — a thread-safe 32-bit integer for lock-free concurrent access.
- atomic32_remove(atomic: Atomic32?& )
Warning
This is unsafe operation.
Destroys an Atomic32 and releases its resources.
- Arguments:
atomic : Atomic32?& implicit
- atomic64_create(): Atomic64?
Creates an Atomic64 — a thread-safe 64-bit integer for lock-free concurrent access.
- atomic64_remove(atomic: Atomic64?& )
Warning
This is unsafe operation.
Destroys an Atomic64 and releases its resources.
- Arguments:
atomic : Atomic64?& implicit
9.1.7.1. dec
- dec(atomic: Atomic32? ): int
Atomically decrements the integer value and returns the result.
- Arguments:
atomic : Atomic32? implicit
- dec(atomic: Atomic64? ): int64
9.1.7.2. get
- get(atomic: Atomic32? ): int
Returns the current value of the atomic integer.
- Arguments:
atomic : Atomic32? implicit
- get(atomic: Atomic64? ): int64
9.1.7.3. inc
- inc(atomic: Atomic32? ): int
Atomically increments the integer value and returns the result.
- Arguments:
atomic : Atomic32? implicit
- inc(atomic: Atomic64? ): int64
9.1.7.4. set
- set(atomic: Atomic32?; value: int )
Sets the atomic integer to the specified value.
- Arguments:
atomic : Atomic32? implicit
value : int
- set(atomic: Atomic64?; value: int64 )
- with_atomic32(block: block<(Atomic32?):void> )
Creates an Atomic32 scoped to the given block and automatically destroys it afterward.
- Arguments:
block : block<( Atomic32?):void> implicit
- with_atomic64(block: block<(Atomic64?):void> )
Creates an Atomic64 scoped to the given block and automatically destroys it afterward.
- Arguments:
block : block<( Atomic64?):void> implicit