← All demos

Concurrency & performance

Distributed worker pool

the local-worker half of Qu's distributed job-dispatch machinery (§ resource-limited pools/queues, see book/src/stdlib/concurrency.md's "Worker pools, deferred queues, and `run ... on`" section): a persistent, resource-limited `pool` of real

catalog/qu_distributed_pool.qu

`rayon` worker threads, a `queue()` of deferred jobs, and `run <queue> on <pool>` to drain the queue across it and block for every real result.

This is deliberately the SELF-CONTAINED half of that feature -- the `remote=(...)`/`listen_pool` half needs a second live Qu process accepting TCP job requests (book/src/stdlib/concurrency.md's own "Distributed pools" section), which isn't something a single catalog script run in isolation can demonstrate honestly. What's shown below is the real underlying dispatch primitive that remote pools build on: named pools, `on="any"` auto-routing across every registered pool by free capacity, and ordered result collection -- all real threads, real work, real (not simulated) completion.

What it exercises

• pool <name> with cpu=N ... end pool -- a persistent, registered worker pool
• queue() / q.push("fnName", args..., on=...) -- a deferred job list
• run <queue> on <pool> -- drain + block for real results, in submission order
• run <queue> on any -- auto-route across every registered pool

How it works

pool workers with cpu = 2
end pool

function square(x)
    return x * x
end function

q = queue()
for i = 0 to 5
    q.push("square", i, on="cpu")
end for

results = run q on workers
print("run q on workers -> {results}")
expected = [0, 1, 4, 9, 16, 25]
matches = true
for i = 0 to 5
    if results[i] != expected[i]
        matches = false
    end if
end for
print("results match expected squares in submission order: {matches}")

pool p1 with cpu = 1
end pool
pool p2 with cpu = 1
end pool

function cube(x)
    return x * x * x
end function

q2 = queue()
for i = 1 to 4
    q2.push("cube", i, on="any")
end for

results2 = run q2 on any
print("run q2 on any -> {results2}")

Results

run q on workers -> (0, 1, 4, 9, 16, 25)
results match expected squares in submission order: true
run q2 on any -> (1, 8, 27, 64)

The whole program

Runnable as it stands: qu run catalog/qu_distributed_pool.qu

# qu_distributed_pool.qu -- the local-worker half of Qu's distributed

# job-dispatch machinery (§ resource-limited pools/queues, see

# book/src/stdlib/concurrency.md's "Worker pools, deferred queues, and

# `run ... on`" section): a persistent, resource-limited `pool` of real

# `rayon` worker threads, a `queue()` of deferred jobs, and `run <queue> on

# <pool>` to drain the queue across it and block for every real result.

#

# This is deliberately the SELF-CONTAINED half of that feature -- the

# `remote=(...)`/`listen_pool` half needs a second live Qu process

# accepting TCP job requests (book/src/stdlib/concurrency.md's own

# "Distributed pools" section), which isn't something a single catalog

# script run in isolation can demonstrate honestly. What's shown below is

# the real underlying dispatch primitive that remote pools build on: named

# pools, `on="any"` auto-routing across every registered pool by free

# capacity, and ordered result collection -- all real threads, real work,

# real (not simulated) completion.

#

# Constructs exercised (all real, verified-working builtins/forms):

#   pool <name> with cpu=N ... end pool   -- a persistent, registered worker pool

#   queue() / q.push("fnName", args..., on=...) -- a deferred job list

#   run <queue> on <pool>                  -- drain + block for real results, in submission order

#   run <queue> on any                     -- auto-route across every registered pool


backend auto

#%% A named pool with 2 real worker threads

pool workers with cpu = 2
end pool

function square(x)
    return x * x
end function

q = queue()
for i = 0 to 5
    q.push("square", i, on="cpu")
end for

results = run q on workers
print("run q on workers -> {results}")
expected = [0, 1, 4, 9, 16, 25]
matches = true
for i = 0 to 5
    if results[i] != expected[i]
        matches = false
    end if
end for
print("results match expected squares in submission order: {matches}")

#%% Two more small pools, each with just 1 worker -- `run ... on any`

# auto-routes each queued job to whichever registered pool currently has

# the most free capacity, so jobs genuinely spread across BOTH of them

# rather than piling onto a single default.

pool p1 with cpu = 1
end pool
pool p2 with cpu = 1
end pool

function cube(x)
    return x * x * x
end function

q2 = queue()
for i = 1 to 4
    q2.push("cube", i, on="any")
end for

results2 = run q2 on any
print("run q2 on any -> {results2}")