14c: Concurrency — Channels and Threads

Audience: All — required for servers, parallel computation, or any I/O at scale Time: 120 minutes Prerequisites: 04-Functions-and-Scope, 14b-Memory-Management-and-Lifetimes You'll learn: Chan(T) channels, sys.go() for spawning threads, Atomic(T) for shared counters and flags, ThreadPool(n) for bounded worker pools, and which primitive to reach for in each situation

The Big Picture

Concurrency in Zebra has two pillars:

• Threads — independent stacks running in parallel, sharing the same arena allocator
• Channels — typed, buffered queues that move values safely between threads

This is the [CSP model] popularised by Go: prefer communicating over sharing. Most patterns reduce to "spawn a worker, give it a channel, read from the channel." The compiler doesn't have a borrow checker, so correctness comes from this discipline plus a small set of Atomic primitives for the few cases where channels are overkill.

This chapter walks the four primitives — Chan(T), sys.go(), Atomic(T), ThreadPool(n) — and the patterns that combine them.

Channels: Chan(T)

Chan(T) is a thread-safe buffered channel carrying values of type T. Construct it with a capacity:

var ch: Chan(int) = Chan(int)(4)   # buffer up to 4 values

Three operations:

def main()
    var ch: Chan(int) = Chan(int)(4)
    ch.send(1)
    ch.send(2)
    ch.close()

    var v1: int? = ch.recv()       # 1
    var v2: int? = ch.recv()       # 2
    var v3: int? = ch.recv()       # nil — channel closed + empty

    if v1 as n
        print(n)                   # 1
    if v3 == nil: print("done")

The <- operator

The <- operator is syntactic sugar for send / recv:

ch <- 42                # equivalent to ch.send(42)

var v: int? = nil
v <- ch                 # equivalent to v = ch.recv()

<- reads naturally — left arrow into the channel for send, left arrow into the variable for receive. Many Zebra programs use the sugar exclusively.

Not to be confused with <<-. Chapter 14b covers <<-, the copy-out operator inside allocate blocks. It is a different token: <- means channel send/receive and nothing else.

Spawning Threads: sys.go()

sys.go(def() ...) spawns a fire-and-forget background thread that runs the function's body:

sys.go(def()
    print("hello from a thread")
)

The function takes no parameters. Every variable from the enclosing scope that the body uses must be declared in a capture block at the top of the body. An ordinary lambda captures what it reads on its own; a thread closure does not, and using an outside variable without declaring it is refused ("a sys.go closure names what it uses in a capture block"). Re-declaring under the same name (var ch: Chan(int) = ch) is the usual idiom. Captured values are copied into the thread closure at spawn time, so the spawning function can return immediately without leaving dangling references.

def main()
    var ch: Chan(int) = Chan(int)(4)

    sys.go(def()
        capture
            var ch: Chan(int) = ch
        ch.send(1)
        ch.send(2)
        ch.close()
    )

    # Main thread reads:
    var sum = 0
    var done = false
    while not done
        var v: int? = ch.recv()
        if v as n
            sum = sum + n
        else
            done = true
    print(sum)  # 3

What sys.go() does NOT give you

• No join. Threads are detached. There's no thread.wait() or return-value collection — use a channel to signal completion or to return a result.
• No backpressure. Every sys.go() call spawns a new OS thread. If you need bounded concurrency, use ThreadPool(n) (below).
• No panic recovery. If the body raises, the thread terminates. Design tasks to validate inputs before running heavy work.

Producer / Consumer Pattern

The canonical pattern: one or more producers send to a channel; one or more consumers recv until the channel is closed + empty.

# file: 14c_pipeline.zbr
# teaches: producer/consumer with Chan
# chapter: 14c-Concurrency-Channels-and-Threads

def main()
    var ch: Chan(int) = Chan(int)(4)

    # Producer: send 1..5, then close
    sys.go(def()
        capture
            var ch: Chan(int) = ch
        for i in 1..5
            ch.send(i)
        ch.close()
    )

    # Consumer: read until closed + empty
    var sum: int = 0
    var done: bool = false
    while not done
        var v: int? = ch.recv()
        if v as n
            sum = sum + n
        else
            done = true

    print(sum)  # 1+2+3+4 = 10

This is the workhorse of concurrent Zebra. Two takeaways:

1. The producer is responsible for closing. Without ch.close(), the consumer's recv() would block forever after the last value — it has no way to know the producer is finished.
2. The consumer reads T?, not T. recv() returns nil when the channel is closed and empty. The if v as n pattern is the idiomatic way to handle both cases.

Atomic(T) — Lock-Free Counters and Flags

Some shared state doesn't need a channel — a counter incremented by many threads, a "done" flag, a one-shot signal. For those, use Atomic(T):

var counter: Atomic(int) = Atomic(int)(0)
counter.add(1)
var n: int = counter.load()

Supported T: int and bool.

All operations use sequentially-consistent memory ordering — the safest and least-surprising default. For cases where relaxed ordering would be sound and faster, drop into zig"..." and call @atomicRmw directly.

Shared counter

def main()
    var total: Atomic(int) = Atomic(int)(0)

    sys.go(def()
        capture
            var total: Atomic(int) = total
        var _ = total.add(1)
    )
    sys.go(def()
        capture
            var total: Atomic(int) = total
        var _ = total.add(1)
    )

    # There is no join: poll until both increments have landed.
    while total.load() < 2
        sys.sleep(1)

    print(total.load())  # 2

One-shot done flag

def do_work()
    sys.sleep(20)                # stand-in for real work

def main()
    var done: Atomic(bool) = Atomic(bool)(false)

    sys.go(def()
        capture
            var done: Atomic(bool) = done
        do_work()
        done.store(true)
    )

    while not done.load()
        sys.sleep(10)

    print("worker finished")

Atomic vs. Chan — Which to Reach For

Rule of thumb: if multiple threads write the value, reach for Atomic; if one thread produces and another consumes, reach for Chan.

ThreadPool(n) — Bounded Workers

sys.go() spawns an unbounded thread per call — fine for a handful, bad for thousands. ThreadPool(n) runs a fixed n workers and queues submitted tasks among them:

def main()
    var pool: ThreadPool = ThreadPool(4)        # 4 worker threads
    var counter: Atomic(int) = Atomic(int)(0)

    var i: int = 0
    while i < 8
        pool.submit(def()
            capture
                var counter: Atomic(int) = counter
            var _: int = counter.add(1)
        )
        i = i + 1

    pool.wait()                  # blocks until all submitted tasks finish
    print(counter.load())  # 8

Key behaviours

• Reusable. submit after wait queues more work. A subsequent wait blocks on the new tasks.
• Workers run for the pool's lifetime — they're spawned at construction.
• Task panics terminate the worker, not the program — and they are not recovered. Validate inputs before submitting.

ThreadPool vs. sys.go()

For a 1000-item batch, ThreadPool(8) is almost always the right choice. For two or three I/O tasks running alongside the main thread, sys.go() is simpler.

ThreadPool + Chan: collecting results

The standard pattern for parallel computation with a return value:

def main()
    var ch: Chan(int) = Chan(int)(8)
    var pool: ThreadPool = ThreadPool(4)

    for i in 0..8
        pool.submit(def()
            capture
                var idx: int = i
                var ch: Chan(int) = ch
            ch.send(idx * idx)
        )

    pool.wait()                  # all tasks done; close the channel
    ch.close()

    var sum: int = 0
    var done: bool = false
    while not done
        var v: int? = ch.recv()
        if v as n
            sum = sum + n
        else
            done = true

    print(sum)  # 0+1+4+9+16+25+36+49 = 140

Three things to notice:

1. The capture block names everything the closure uses — captured-by-value semantics mean each task sees its own idx, not the shared loop variable.
2. pool.wait() runs before ch.close() — closing too early would race with sends still in flight.
3. The consumer is the main thread — no extra worker needed.

Memory and Allocator Interactions

Two important caveats:

Chan(T) uses the page allocator, not the arena. This is intentional — channels need to outlive any single allocate scope. But it means: do not construct a channel inside a short-lived allocate block if it's going to be used after the block exits. The channel itself survives, but its internal buffer storage may not.

ThreadPool also uses the page allocator. Same rule: don't construct one inside a short-lived allocate scope.

For practical purposes: declare your channels and pools at function scope (or higher) and pass them by reference. Don't nest them inside allocate Arena() blocks unless you understand the lifetime implications.

Real World: Parallel File Processor

A common shape: walk a directory, process each file in parallel with a bounded pool, collect results via a channel.

# file: 14c_parallel_files.zbr
# teaches: ThreadPool + Chan for parallel I/O
# chapter: 14c-Concurrency-Channels-and-Threads

class FileResult
    public var path: str = ""
    public var line_count: int = 0

def process_file(path: str): FileResult
    var src = File.read(path)
    var lines = src.lines()        # not split("\n"): that counts the empty piece after a final newline
    var r = FileResult()
    r.path = path
    r.line_count = lines.count()
    return r

def main()
    var names = File.listDir("./docs")        # entry names, not paths
    var ch: Chan(FileResult) = Chan(FileResult)(names.count())
    var pool: ThreadPool = ThreadPool(4)

    for name in names
        pool.submit(def()
            capture
                var p: str = Path.join("./docs", name)
                var ch: Chan(FileResult) = ch
            ch.send(process_file(p))
        )

    pool.wait()
    ch.close()

    var results: List(FileResult) = List()
    var done = false
    while not done
        var r: FileResult? = ch.recv()
        if r as got
            results.add(got)
        else
            done = true

    for r in results
        print("${r.path}: ${r.line_count} lines")

This same shape — pool + channel + capture closure — covers parallel HTTP requests, parallel database queries, parallel image processing, and most other batch I/O patterns.

Common Mistakes

❌ Mistake: Forgetting to close the channel

sys.go(def()
    capture
        var ch: Chan(int) = ch
    for i in 1..5
        ch.send(i)
    # ch.close() omitted — consumer will block forever after the 4th value
)

✅ Better:

sys.go(def()
    capture
        var ch: Chan(int) = ch
    for i in 1..5
        ch.send(i)
    ch.close()
)

❌ Mistake: Sending to a closed channel

ch.close()
ch.send(1)              # PANIC at runtime

✅ Better: the producer is the only one who should close, and only after the last send.

❌ Mistake: Closing the channel before the producers finish

for i in 0..8
    pool.submit(def()
        capture
            var ch: Chan(int) = ch
        ch.send(work())
    )
ch.close()              # WRONG: tasks are still running
pool.wait()

✅ Better: wait first, then close:

for i in 0..8: pool.submit(...)
pool.wait()             # all sends are complete
ch.close()

❌ Mistake: Sharing a plain counter across threads

var counter: int = 0
sys.go(def()
    counter = counter + 1       # refused: a `sys.go` closure names what it uses in a `capture` block
)

The compiler refuses this — counter is not declared in a capture block. Declaring it there does not make it shared either: a captured int is a copy taken at spawn time, so it cannot carry a count back to the spawning thread.

✅ Better: use Atomic(int) and capture the cell:

var counter: Atomic(int) = Atomic(int)(0)
sys.go(def()
    capture
        var counter: Atomic(int) = counter
    var _ = counter.add(1)
)
sys.go(def()
    capture
        var counter: Atomic(int) = counter
    var _ = counter.add(1)
)

❌ Mistake: Using the loop variable without capturing it

for i in 0..8
    pool.submit(def()
        # capture missing — compile error: mutable 'i' not accessible
        print(i)
    )

✅ Better: use a capture block to snapshot the value — each task gets its own copy of i as it was when the task was submitted:

for i in 0..8
    pool.submit(def()
        capture
            var idx: int = i
        print(idx)
    )

Exercises

Exercise 1: A Simple Pipeline

Write a program that uses one producer thread to send the integers 1 through 10 down a channel, and the main thread to print their sum.

def main()
    var ch: Chan(int) = Chan(int)(4)

    sys.go(def()
        capture
            var ch: Chan(int) = ch
        for i in 1..11
            ch.send(i)
        ch.close()
    )

    var sum: int = 0
    var done: bool = false
    while not done
        var v: int? = ch.recv()
        if v as n
            sum = sum + n
        else
            done = true

    print(sum)  # 55

Exercise 2: Parallel Squaring

Use ThreadPool(4) to compute the squares of 0..16 in parallel. Collect the results in any order via a channel and print the sum.

def main()
    var ch: Chan(int) = Chan(int)(16)
    var pool: ThreadPool = ThreadPool(4)

    for i in 0..16
        pool.submit(def()
            capture
                var idx: int = i
                var ch: Chan(int) = ch
            ch.send(idx * idx)
        )

    pool.wait()
    ch.close()

    var sum: int = 0
    var done: bool = false
    while not done
        var v: int? = ch.recv()
        if v as n
            sum = sum + n
        else
            done = true

    print(sum)  # 0^2 + 1^2 + ... + 15^2 = 1240

Exercise 3: Wait Without Joining

Threads don't expose a join() call. Use an Atomic(bool) flag to have the main thread wait until a background thread signals completion.

def main()
    var done: Atomic(bool) = Atomic(bool)(false)

    sys.go(def()
        capture
            var done: Atomic(bool) = done
        # ... do work ...
        sys.sleep(100)
        done.store(true)
    )

    while not done.load()
        sys.sleep(10)

    print("background work finished")

Next Steps

• → 14b-Memory-Management-and-Lifetimes — the memory model that underpins all of this
• → 22c-Testing-and-Validation — concurrent code needs tests; the runner handles them like any other
• → 20-File-IO-and-System-Access — combine with sys.go for non-blocking I/O patterns

Key Takeaways

• Chan(T) is a thread-safe buffered channel: send blocks when full, recv blocks when empty, close signals end-of-stream
• <- sugar — ch <- v sends; v <- ch receives; reads naturally for pipelines
• sys.go(def() ...) spawns a fire-and-forget thread; outside variables must be declared in a capture block and are copied at spawn time; no join (use a channel for completion)
• Atomic(T) (T = int or bool) is for shared counters, flags, and one-shot signals; never blocks
• Rule of thumb: multiple writers → Atomic; producer/consumer → Chan
• ThreadPool(n) is a bounded worker pool with submit / wait; natural backpressure; reusable
• ThreadPool + Chan is the standard pattern for parallel computation with collected results
• Allocator caveat: Chan and ThreadPool use the page allocator; declare them at function scope, not inside short-lived allocate blocks
• Capture loop variables explicitly in for i in 0..N: pool.submit(...) — using i without a capture block is a compile error, and the capture gives each task its own copy

That's the end of the new chapters in Part 3. The next chapter in reading order is the existing 15-Pipelines-and-Function-Composition; when you're ready for the practical project arc, jump ahead to Part 4.