Concurrency

The tables below are generated by reading the interpreter's own builtin dispatch table (fn call_builtin, engine/crates/qu-interp/src/lib.rs) directly — every row reflects a match arm that actually runs today, not aspirational API surface. See Book 3 §4 and §5 for a narrative walkthrough of this subsystem.

Platform boundary: everything in this chapter runs on real OS threads (spawn, pmap, parallel for) or real OS synchronization primitives (mutex, semaphore, channel). Qu's wasm32-unknown-unknown browser target has no native threads, so this subsystem is native-CLI-first; a browser Studio target would need its own backend, not something that falls out of the shared Rust core for free.

Pool, not per-call threads: spawn, pmap, and parallel for all dispatch onto rayon's global work-stealing thread pool rather than spawning a fresh OS thread per call — the same pool qu-core's matmul/elementwise kernels already share, so a script mixing worker concurrency with heavy matrix math doesn't oversubscribe the machine with two independent pools competing for cores. Work-stealing means each pool thread pulls from its own queue and steals from others' when idle, rather than tasks being handed out in a fixed rotation — an uneven pmap workload (elements that take wildly different time) still keeps every thread busy, instead of idling one behind a slow neighbor the way strict round-robin assignment would.

Parallel Execution

Overloads: join

Case: worker

w a Worker handle (from spawn). Blocks until that worker finishes, then returns its function's return value, or re-raises whatever error the worker's body raised.

heavy(n) := n * n
w = spawn("heavy", 6)
print(join(w))   # 36

36

Case: list

list a List of strings, with an optional delimiter (default ""). The unrelated collection form — concatenates every element into one string, the inverse of split — selected purely because the first argument isn't a Worker handle.

print(join(["a", "b", "c"]))        # "abc"

print(join(["a", "b", "c"], "-"))   # "a-b-c"

abc
a-b-c

Mutexes

Genuinely shared mutable state — the deliberate opt-in exception to spawn/parallel for's default share-nothing design. Cloning a Mutex value (e.g. capturing it into a worker's environment snapshot) clones the handle, not the data — every clone still points at the same cell.

mutex is the escape hatch from spawn/parallel for's normal share-nothing isolation: every other worker-facing primitive in this chapter gives each worker its own private copy of the world, but a Mutex handle stays a single shared cell no matter how many workers hold a clone of it. The example below creates one, overwrites it directly with mutex_set (no read involved), then performs an atomic read-modify-write through a user function with mutex_update — the pattern you reach for when the update is more than a simple numeric increment (mutex_add covers that simpler case). The final print shows the value mutex_update left behind:

running_max(current) := max(current, 7)

m = mutex(0)
mutex_set(m, 3)                        # overwrite -- no read involved

mutex_update(m, "running_max")         # atomic read-modify-write via a user fn

print(mutex_get(m))                    # 7

7

Examples

This section walks through four of the chapter's core primitives together, each in the shape you'd actually reach for it in a real script rather than the minimal one-liners in the tables above.

spawn is for offloading one genuinely expensive call onto its own OS thread so the calling script can keep doing other work (here, polling) while it runs, then collect the result with join once it's ready. The example spawns a function, polls worker_done in a tight loop until it reports completion (a real script would normally do useful work in that loop instead of spinning), then blocks on join to retrieve the return value:

heavy(n) := n * n
w = spawn("heavy", 7)                  # runs on its own OS thread

while not worker_done(w)               # non-blocking poll; handle stays valid

end while
result = join(w)                       # blocks for the result (49)

print(result)                          # 49

49

pmap is the tool for the common case of applying the same pure function to every element of a collection and wanting the results back in a vector — feature extraction over a batch of signals, a per-row transform over a dataset — without hand-writing a spawn/join loop yourself. It applies a pure function across a vector using the same share-nothing isolation spawn uses, composing naturally with |>:

extract(x) := x * x
xs = 0 to 9
out = xs |> pmap("extract")            # same as pmap(xs, "extract")

print(out)                             # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

mutex is for the rarer case where parallel workers genuinely need to share one piece of mutable state rather than working on independent data — a running total, a shared cache, an accumulated log — safely, without the lost-update races that reading-then-writing an ordinary shared value from multiple threads would cause. The example runs 100 loop iterations in parallel, each incrementing a shared counter through the atomic mutex_add, and confirms afterward that all 100 increments were actually counted with none lost to a race:

counter = mutex(0)
parallel for i in 1 to 100
    mutex_add(counter, 1)              # atomic; 100 in, 100 out, no lost updates

end parallel
total = mutex_get(counter)
print(total)                           # 100

100

channel is for handing whole values from one worker to another as they become available, rather than sharing a mutable cell — the natural fit for a producer/consumer pipeline where one side generates results and another consumes them as they arrive. The example spawns a producer that sends a single value, and the main thread's channel_recv blocks until that value shows up:

ch = channel()
producer() := channel_send(ch, 42)
w = spawn("producer")
value = channel_recv(ch)               # blocks until producer sends

print(value)                           # 42

join(w)

42

Semaphores

A counting semaphore, for bounding concurrent access to a limited shared resource (a hardware port, a connection pool, or just an explicit concurrency cap) across spawn/parallel for workers.

A semaphore bounds how many workers may hold a limited resource concurrently — a hardware port, a connection pool, or just an explicit concurrency cap — by handing out a fixed number of permits that semaphore_acquire blocks for and semaphore_release gives back. The example below creates a semaphore with 2 permits, takes one, and returns it, printing semaphore_available's live count at each step so the acquire/release bookkeeping is visible:

s = semaphore(2)                       # 2 permits available

print(semaphore_available(s))          # 2

semaphore_acquire(s)                   # take one

print(semaphore_available(s))          # 1

semaphore_release(s)                   # give it back

print(semaphore_available(s))          # 2

2
1
2

Channels

Worker-to-worker message passing: a blocking MPMC queue, the counterpart to mutex's shared-state model. Cloning a Channel value shares the same underlying queue, the same Arc-sharing convention Mutex/Semaphore already use.

A channel is the message-passing counterpart to mutex's shared-state model: instead of workers reading and writing one shared cell, they hand whole values to each other through a blocking FIFO queue. The example below checks an empty channel with the non-blocking channel_try_recv (getting none back), sends two values, reads the queue depth with channel_len, then drains one value with the blocking channel_recv — printing the result of each step so the FIFO ordering and the empty-queue behavior are both visible:

ch = channel()
print(channel_try_recv(ch))            # none -- nothing queued yet

channel_send(ch, 1)
channel_send(ch, 2)
print(channel_len(ch))                 # 2

print(channel_recv(ch))                # 1 (FIFO)

none
2
1

Job Queue & Worker Pools

A resource-limited worker pool (pool ... with cpu=N ... end pool, or the anonymous pool(n) shorthand), a deferred-job list (queue()), and run <queue> on <pool> to dispatch every queued job across it, collecting an ordered List of results. Distinct from spawn/pmap above in one way: jobs are referenced by function NAME (a string) plus already-evaluated arguments, submitted ahead of time, then all drained together by run — not fired off individually.

Overloads: pool

Case: named

pool <name> with cpu=N ... end pool is a statement, not an expression — it registers a persistent pool under the bare identifier name, so more than one later run ... on name call (or run ... on any) can share the same fixed set of worker threads instead of each spinning up its own.

square(x) := x * x
pool workers with cpu=2
end pool
q = queue()
q.push("square", 3)
print(run q on workers)   # (9)

(9)

Case: anonymous

pool(n) is an expression — it returns an unregistered Pool handle with n local cpu workers, scoped to whichever single run ... on pool(n) call uses it, with no pool ... end pool registration step at all.

square(x) := x * x
q = queue()
q.push("square", 3)
print(run q on pool(2))   # (9)

(9)

A named pool is worth registering when more than one run call — maybe across different stages of a script — should share the same fixed set of worker threads rather than each spinning up its own. The example below defines a pure function, builds a 4-worker pool named workers, queues five calls to it with .push, then drains the whole queue at once with run ... on, printing the resulting list of squares in submission order:

square(x) := x * x

pool workers with cpu=4
end pool

q = queue()
for i in 0 to 4
    q.push("square", i)
end for

results = run q on workers
print(results)              # (0, 1, 4, 9, 16)

(0, 1, 4, 9, 16)

The anonymous form skips the pool ... end pool registration entirely -- handy when nothing else needs to share the same workers, since pool(n) can be written directly inline as the target of run ... on:

square(x) := x * x

q = queue()
for i in 0 to 4
    q.push("square", i)
end for

results = run q on pool(2)  # 2 local cpu workers, scoped to this one call

print(results)              # (0, 1, 4, 9, 16)

(0, 1, 4, 9, 16)

Distributed pools: listen_pool and remote=(...)

A pool's workers don't have to be local threads — remote = (...) names "host:port" addresses of other Qu processes running listen_pool, and run ... on <pool> transparently splits jobs across local cpu workers and those remote listeners, merging every result into the same ordered List either way. This is a real, minimal, working feature — not the larger worker()/cluster()/auto-discovery cluster vision sketched elsewhere in the language reference's distributed-execution chapter; there is no node discovery, no least_loaded/affinity placement policy, and no fault-tolerant rescheduling here, only "name some addresses, dispatch jobs to them by function name, get results back."

Normally the two sides of this run as separate processes on separate machines, but the demo below folds them into one self-contained script by running the listener in a background worker on the SAME machine, talking to itself over the loopback address — the mechanics (bind, connect, dispatch, merge results) are identical to the real cross-machine case. run_listener wraps listen_pool so it can be spawn'd onto its own thread (since listen_pool blocks forever, it can never run on the main thread if that thread still has to submit jobs); a short sleep gives the listener time to finish binding before the pool tries to connect. workers mixes one local cpu worker with one remote listener on 127.0.0.1, and run q on workers splits the five queued jobs across both, merging every result back into a single ordered List:

square(x) := x * x

function run_listener()
    listen_pool(9321, allow=("square"))
end function

w = spawn("run_listener")     # the "remote" side, on its own thread here

sleep(0.3)                    # give it time to bind before connecting


pool workers with cpu=1, remote=("127.0.0.1:9321")
end pool

q = queue()
for i in 0 to 4
    q.push("square", i)
end for

results = run q on workers    # some local, some "remote" — one ordered List

print(results)                # (0, 1, 4, 9, 16)

Jobs round-robin by submission order across "lanes" (one lane for the whole local cpu pool, one lane per remote address) — the pool's own round_robin policy, now spanning local and remote workers together. Wire format is newline-delimited JSON reusing Value's existing JSON encoding (the same one save/load use). An unreachable, timed-out (bounded connect/IO timeouts, never an indefinite hang), or allowlist-rejecting remote worker fails that one job exactly like a local job's own runtime error — the first failure in submission order propagates out of run, with no silent fallback to running it locally instead.

Networking (HTTP)

Qu's TCP socket library (tcp_listen/tcp_connect/tcp_send/tcp_recv, raw sockets — see the Language Reference's §50.4 for the full design) has no dedicated stdlib chapter of its own yet; http_get is placed here as the nearest existing home for a networking-flavored builtin rather than inventing a new one-function chapter.

http_get is the quickest way to pull a resource off the web into a string for further processing — parsing JSON out of an API response, scraping a page's text, or just checking that a URL is reachable — without setting up a full streaming connection. The example below fetches a page and prints the length of the body it got back, so the request's success and the size of what came down the wire are both visible:

body = http_get("https://example.com/")
print(length(body))            # e.g. 559 -- byte length of the fetched page

559

StreamURL(url) wraps this same request as a read-only stream object (read_line/read_all/eof/close), polymorphic with StreamFile — see Polymorphic Streams: StreamFile / StreamURL.

TCP sockets

Reach for raw TCP sockets when http_get's one-shot GET isn't enough — talking to a non-HTTP service, holding a connection open for more than one exchange, or writing both the client and the server yourself (as the distributed-pool machinery above does internally). The demo below is a self-contained loopback echo server: tcp_listen/tcp_accept on the server side (run on its own worker via spawn, since tcp_accept blocks the calling thread until a client connects), tcp_connect on the client side, tcp_send/tcp_recv to exchange one message, tcp_close to release each connection once done. tcp_listen(0) asks the OS for a free ephemeral port instead of a fixed one, and tcp_port reports which one it actually bound so the client knows where to connect. The final print shows the message coming back exactly as sent, proving the round trip worked:

function echo_once(listener)
    conn = tcp_accept(listener)        # blocks for the one client

    msg = tcp_recv(conn)
    tcp_send(conn, msg)
    tcp_close(conn)
end function

listener = tcp_listen(0)               # 0 -- let the OS pick a free port

port = tcp_port(listener)
w = spawn("echo_once", listener)

client = tcp_connect("127.0.0.1", port)
tcp_send(client, "hello")
print(tcp_recv(client))                # "hello" -- echoed back

tcp_close(client)
join(w)

hello

Watches (watch)

Three change-detection trigger conditions, registered as a statement (watch ... [do] ... end) and checked only by the pump_watches builtin above — see the Language Tour's own "watch — file/variable/URL change detection" section for the full execution-model discussion (why this is NOT a background process: Qu's interpreter is single-threaded with no event loop, so a watch callback only ever runs during an explicit pump_watches() call, the same cooperative model every/after/at already have via run_for).

watch is for noticing when something outside your script's own control flow changes — a config file another process might edit, a variable some other part of a larger script mutates, or a URL whose content might update — without polling it by hand on every loop iteration. Registering a watch costs nothing until you actually call pump_watches(), which checks every registered watch once against its current state. The example below watches a file for a modification-time change: the first pump_watches() call sees no change (the file hasn't been touched since registration) and fires nothing, then touch bumps the file's mtime and the second call detects that and runs the watch's body, incrementing fires — both counts are printed so the before/after difference is visible:

fires = 0
watch file("config.txt") do
    fires = fires + 1
end
pump_watches()          # nothing changed yet -> 0 fired

print(fires)             # 0

touch("config.txt")
pump_watches()          # real mtime change -> 1 fired

print(fires)             # 1

· pump_watches(): 0 watch firing(s) across 1 registered watch(es)
0
· pump_watches(): 1 watch firing(s) across 1 registered watch(es)
1

Serial Port I/O

Real UART/RS-232/USB-serial device I/O (§ real serial port I/O, 2026-09-01), via the serialport crate — placed in this same chapter as http_get/the TCP socket library above rather than a disconnected new chapter, for the same "nearest existing I/O-flavored home" reasoning. It follows Value::File/Value::TcpConn's own opaque-handle convention: a constructor returns a live handle, and handle.method(...) sugar makes .read_line()-style calls work with no special-casing.

serial_open plus .read_line() is the everyday pattern for talking to a line-oriented device — a sensor board that streams one text reading per line, a modem, a microcontroller printing debug output. The example below lists whatever serial devices this machine's OS currently reports, opens the first one with a half-second read timeout, and prints whatever line (if any) arrived within that window — line comes back as none, not an error, if nothing showed up in time:

ports = serial_ports()                              # e.g. ["COM3"], or [] with no hardware attached

conn = serial_open(ports[0], 9600, timeout_ms=500)
line = read_line(conn)                              # blocks up to 500ms; `none` if nothing arrived

if line != none
    print("got: {line}")
else
    print("no line arrived within the timeout")
end if
close(conn)

no line arrived within the timeout

For a binary protocol instead of line-oriented text, .read_bytes(n) is the one to reach for: it collects raw bytes rather than splitting on newlines, which matters when a device sends fixed-size frames or non-text payloads. .available() is the non-blocking companion check — how much has already arrived in the OS's input buffer right now, useful for deciding whether a read is even worth attempting before committing to timeout_ms's wait. The example below checks availability, then attempts to read up to 4 bytes (an honest short read if timeout_ms elapses first), and prints both numbers so the non-blocking check and the actual read result are both visible — this one already guards the no-hardware case explicitly:

ports = serial_ports()
if length(ports) > 0
    conn = serial_open(ports[0], 9600, timeout_ms=200)
    n = available(conn)                        # bytes waiting, non-blocking

    data = read_bytes(conn, 4)                  # up to 4 bytes, short read on timeout

    print("available={n}, got {length(data)} byte(s)")
    close(conn)
else
    print("no serial hardware attached")
end if

available=0, got 0 byte(s)

What does NOT apply to a serial handle, and why. read_all/eof are deliberately not implemented: both assume a fixed, knowable total length (a file's byte count, or an HTTP response's fully-fetched body) so that "everything remaining" and "have I reached the end" are honest questions — a live serial device has neither, since data arrives whenever the far end sends it, or never. Calling either on a serial handle falls through to the existing Value::File error path, which already reports a clear "expected a file handle ... found serial" rather than a fabricated answer.

A genuine real-time capture-and-process demo — catalog/qu_serial_realtime.qu — opens a serial port, reads incoming lines in a bounded loop as they arrive, converts each to a number, and prints a running moving average/min/max after every reading, catching an open failure (no matching device attached) as a clean, expected error rather than a crash. See that script's own header comment for why a live-updating plot specifically is out of scope for a qu run script today (Qu's plotting model builds one static figure exported at the end of a run, with no live GUI window to push points into mid-script) and a real-time console loop is the honest, actually-achievable version of "process data as it arrives."

More functions

now/sysinfo are the two functions to reach for when a script needs to know about its own environment rather than about the concurrency machinery itself — timestamping output, or deciding how many workers a pool should use based on the machine's actual core count. The example below reads the current UTC year and prints it, then reads how many logical cores this machine reports — exactly the number you'd branch on before calling pool(n) with a sensible n:

t = now()
print(t.year)                 # e.g. 2026 -- UTC

info = sysinfo()
print(info.logical_cores)     # e.g. 16 -- what to branch on before sizing a pool(n)

2026
16