TB v3.2.0

parallel

Do several things at the same time: run functions as tasks, map a function over a list with many workers, pass work through queues, and run timers. Great for waiting on many network checks or files at once. Tasks share global variables: use a Lock when several change the same thing. In a window app, change widgets from on_done (or gui.on_main), never wait() inside a button click.

This library runs work on several threads at once, and the browser playground has no threads, so it needs the TB app on your computer. The examples here show the code; run them with tb file.tb.

import parallel

Example

Needs the TB app on your computer, so there is no Run button here.

import parallel

# Run a function at the same time as the rest of the program.
fn slow_square(x) {
  parallel.sleep(0.5)
  return x * x
}
let task = parallel.run(slow_square, 7)
say("working...", task.done())
say(task.wait())                     # 49

# Map over a list with 8 workers: results stay in order.
say(parallel.map(slow_square, [1, 2, 3, 4, 5, 6, 7, 8]))  # about 0.5 s, not 4 s

# async fn gives back a Task; await waits for it.
async fn fetch(name) {
  parallel.sleep(0.2)
  return name.upper()
}
let a = fetch("one")
let b = fetch("two")
say(await a, await b)

# A queue between a producer and a consumer.
let q = parallel.Queue()
parallel.run(fn() {
  for i in 1..=3 {
    q.put(i)
  }
  q.close()
})
for item in q {
  say("got", item)
}

# A lock around a shared counter.
let lock = parallel.Lock()
let count = {"n": 0}
parallel.each(fn(i) {
  lock.hold(fn() {
    count.n += 1
  })
}, range(100))
say(count.n)                         # 100

# Errors in a task come back, with their class, when you wait.
let bad = parallel.run(fn() {
  throw ValueError("no thanks")
})
try {
  bad.wait()
} catch err: ValueError {
  say("caught:", err.message)
}

# A timer, stopped after a while.
let timer = parallel.every(100, fn() {
  say("tick")
})
parallel.sleep(0.35)
timer.stop()

Functions

run(fn, ...args)

returns Task

Start fn(args...) at the same time as the rest of the program and give back its Task right away. task.wait() (or await task) gives its result; task.done(), task.cancel() and task.on_done(fn) are there too.

Visualizehow parallel.run works

parallel.run(fetch, "https://example.com") <Task>

parallel.run: inputs flowing into the function and the resultfnfetchargs"https://example.com"parallel.runyour programkeeps goingthe taskfetch(…) runsat the sametimetask.wait()result · Task<Task>

Illustration: example values, drawn from what the function does.

map(fn, items, workers?)

returns list

fn(item) for every item, several at once (workers at a time, default 8), giving the results in the same order as items. If any call fails, map stops and that error happens here.

Visualizehow parallel.map works

parallel.map(x => x * 10, [1, 2, 3, 4, 5, 6], 2) [10, 20, 30, 40, 50, 60]

Drawing…

Real result: worked out by running this example with TB while the site was built.

each(fn, items, workers?)

Call fn(item) for every item, several at once (workers at a time, default 8), and wait until all are done.

Visualizehow parallel.each works

parallel.each(host => check(host), ["a", "b", "c", "d", "e"], 3)

Drawing…

Illustration: example values, drawn from what the function does.

all(tasks, timeout?)

returns list

Wait for every task in the list and give back their results in order (plain values are passed through). If one failed, its error happens here. timeout is in seconds, for all of them together.

Visualizehow parallel.all works

parallel.all([t1, t2, t3], 10) ["page 1", "page 2", "page 3"]

parallel.all: inputs flowing into the function and the resultt12 st25 st31 stimeout?10 s for all of them togetherwaits for the slowest (5 s)results, in task order"page 1""page 2""page 3"

Illustration: example values, drawn from what the function does.

race(tasks, timeout?)

returns value

Wait for whichever task finishes first and give back its result (or its error). The others keep running: cancel() them if you don't need them.

Visualizehow parallel.race works

parallel.race([t1, t2, t3], 10) "mirror 3"

parallel.race: inputs flowing into the function and the resultt12 st25 st3 wins1 stimeout?stop waiting after 10 sthe first to finishresult · value"mirror 3"

Illustration: example values, drawn from what the function does.

sleep(seconds)

Pause for a while (fractions like 0.25 work). Inside a task, cancel() wakes it up and stops the task.

Visualizehow parallel.sleep works

parallel.sleep(0.25)

parallel.sleep: moments on a time linetime →sleep(0.25)goes onpaused 0.25 sseconds0.25parallel.sleepgives backnothing; cancel() inside a taskwakes it up

Illustration: example values, drawn from what the function does.

cpu_count()

returns number

How many processor cores this computer has.

Visualizehow parallel.cpu_count works

parallel.cpu_count() 8

parallel.cpu_count: inputs flowing into the function and the resultthis computerprocessor coresparallel.cpu_countresult · number8

No inputs: call it as it is.

Illustration: example values, drawn from what the function does.

loop(fn, ...args)

returns value

Run fn(args...) on an event loop and give back its result once fn and every task and timer it started have finished. Servers it started with start() stop then.

Inside the loop, tasks (async fn, parallel.run, map) take turns instead of running at the same moment: one runs until it waits (await, sleep, a queue, a network request), then the next. So loop code needs no locks. Servers started inside a loop handle every connection on one thread, so thousands of connections are fine.

Example

async fn twice(x) {
  parallel.sleep(0.1)
  return x * 2
}
say(parallel.loop(fn() { return parallel.all([twice(1), twice(2)]) }))
Visualizehow parallel.loop works

parallel.loop((a, b) => a + b, 2, 3) 5

parallel.loop: inputs flowing into the function and the resultfnmain functionargsits inputsparallel.looptask Arunsuntilit waitstask Bthentakesitsturnserver10,000connections,onethreadfn and every task it started are doneresult · value5

Real result: worked out by running this example with TB while the site was built.

group(fn)

returns list

Call fn(group), where group.run(f, ...args) starts tasks that belong to it, then wait for all of them and give back their results in the order they were started.

If one task fails, the others are cancelled and that error happens here. If fn itself fails, or the task waiting here is cancelled, the group's tasks are cancelled too, so none are left running. group.cancel() stops the rest early (their results are nil).

Example

let pages = parallel.group(fn(g) {
  for url in urls { g.run(http.get, url) }
})
Visualizehow parallel.group works

parallel.group(g => [g.run(() => 1), g.run(() => 2)]) [1, 2]

parallel.group: inputs flowing into the function and the resultfng.run(task 1)g.run(task 2)task 11task 22waits for all; one fails → the rest are cancelledresults, in start order12

Real result: worked out by running this example with TB while the site was built.

timeout(seconds, fn)

returns value

Call fn() and give back its result, but cancel it if it takes longer than seconds; that gives a TimeoutError with err.cancelled true. Tasks fn started in a parallel.group are cancelled with it.

Cancelling is cooperative: fn stops before its next step, or when it next waits.

Visualizehow parallel.timeout works

parallel.timeout(2, () => slow_job())

parallel.timeout: moments on a time linetime →startcancelledTimeoutErrorat most 2 sseconds2fn() => slow_job()parallel.timeoutgives backfn's result, or a TimeoutError(err.cancelled is true)

Illustration: example values, drawn from what the function does.

processes(fn, items, workers?)

returns list

fn(item) for every item, in other processes (workers at a time, default: one per processor core), giving the results in the same order as items.

Unlike map, this makes CPU-heavy work faster: each process has a core to itself. Starting the processes takes a moment, so use it for work that takes seconds, not milliseconds.

fn is copied into each process with the values it uses, so changes it makes to variables aren't seen here. Items, results and the values fn uses must be plain: nil, true/false, numbers, text, bytes, lists, maps, other functions, libraries or shared queues; anything else is a TypeError.

Example

fn slow_square(n) { return n * n }
say(parallel.processes(slow_square, [1, 2, 3, 4], 2))
Visualizehow parallel.processes works

parallel.processes(fib, [30, 30, 30, 30], 4)

Drawing…

Illustration: example values, drawn from what the function does.

shared_queue(size?)

returns SharedQueue

A queue that functions running in parallel.processes can use too: put(value, timeout?), get(timeout?), size(), close(), and for item in q { }, like parallel.Queue. Only plain values can go in. With a size, put() waits while the queue is full.

Visualizehow parallel.shared_queue works

parallel.shared_queue(100) <shared queue 0 waiting>

parallel.shared_queue: inputs flowing into the function and the resulta worker processq.put(result)across processessize?plain values onlyq.get()your programfor item in q { … }

Illustration: example values, drawn from what the function does.

Queue(size?)

returns Queue

A new queue for passing values between tasks: q.put(value), q.get(timeout?), q.size(), q.close(), and for item in q { }. With a size, put() waits while the queue is full.

Visualizehow parallel.Queue works

parallel.Queue(2) <Queue>

parallel.Queue: inputs flowing into the function and the resulta taskq.put(job)into the queuesize?[job 1] [job 2] ← full: put() waitsq.get()another taskfor job in q { … }

Illustration: example values, drawn from what the function does.

Lock()

returns Lock

A new lock. lock.hold(fn) runs fn() while no other task holds the same lock: use it around changes to shared lists, maps and counters.

Visualizehow parallel.Lock works

parallel.Lock() <Lock>

parallel.Lock: inputs flowing into the function and the resulttask Alock.hold(fn)→ runs nowtask Block.hold(fn)→ waits itsturnone at a timeshared list / map / counternever changed by two at once

No inputs: call it as it is.

Illustration: example values, drawn from what the function does.

every(ms, fn)

returns Timer

Call fn() every ms milliseconds, on its own thread, until timer.stop() (or fn gives back false). Errors are printed and stop the timer. The program doesn't wait for timers when it ends.

Visualizehow parallel.every works

parallel.every(1000, () => ping())

parallel.every: moments on a time linetime →0fn()1000 msfn()2000 msfn()3000 msevery msms1000fn() => ping()parallel.everygives backa Timer: .stop() (or fn gives back false)

Illustration: example values, drawn from what the function does.

after(ms, fn)

returns Timer

Call fn() once, ms milliseconds from now, on its own thread. timer.stop() before then cancels it.

Visualizehow parallel.after works

parallel.after(1500, () => say("time!"))

parallel.after: moments on a time linetime →nowfn()once1500 ms laterms1500fn() => say("time!")parallel.aftergives backa Timer: .stop() before thencancels it

Illustration: example values, drawn from what the function does.