Concurrency and Parallelism

Learning Objectives

Spawn threads for parallel execution
Communicate between threads with channels
Use async/await for non-blocking operations
Build a worker pool for parallel task processing

Prerequisites

Completed Building an HTTP Server
Understanding of basic programming concepts

Threads

Use thread.spawn to run code in a new thread:

import thread

action main()
    write("Main: starting")

    @t = thread.spawn(background_work)

    write("Main: doing other work")
    write("Main: waiting for thread")

    @result = t.join()
    write("Thread result: " + result)
fin action.

action background_work()
    write("Thread: working...")
    sleep(1)
    return "done"
fin action.

Thread Safety

Threads share memory, so use channels or mutexes for safe communication.

Channels

Channels let threads send and receive values safely:

import thread
import channel

action main()
    @ch = channel.create(10)

    @producer = thread.spawn(produce, ch)
    @consumer = thread.spawn(consume, ch)

    producer.join()
    consumer.join()
fin action.

action produce(ch)
    loop i from 1 to 5
        ch.send("item " + i)
        write("Produced item " + i)
    again
    ch.close()
fin action.

action consume(ch)
    loop
        @item = ch.recv()
        if not item then
            break
        fi
        write("Consumed " + item)
    again
fin action.

Async/Await

For I/O-bound work, use async/await to avoid blocking:

import async

action main()
    write("Fetching data...")
    @data = await fetch_data("https://api.example.com/data")
    write("Received: " + data)
fin action.

action fetch_data(url)
    return async.http_get(url)
fin action.

Running Multiple Async Operations

action main()
    @a = await fetch_data("https://api.example.com/1")
    @b = await fetch_data("https://api.example.com/2")
    @c = await fetch_data("https://api.example.com/3")
fin action.

To run them in parallel, use async.all:

action main()
    @results = await async.all(
        fetch_data("https://api.example.com/1"),
        fetch_data("https://api.example.com/2"),
        fetch_data("https://api.example.com/3")
    )
    write(results[0])
fin action.

Worker Pools

For CPU-bound parallel processing, use a worker pool:

import thread
import channel

action main()
    @jobs = channel.create(100)
    @results = channel.create(100)

    @num_workers = 4

    // Start workers
    loop i from 1 to num_workers
        thread.spawn(worker, jobs, results)
    again

    // Send jobs
    loop i from 1 to 20
        jobs.send(i * i)
    again
    jobs.close()

    // Collect results
    @sum = 0
    loop i from 1 to 20
        @result = results.recv()
        sum = sum + result
    again

    write("Sum of squares 1..20: " + sum)
fin action.

action worker(jobs, results)
    loop
        @job = jobs.recv()
        if not job then
            break
        fi
        // Simulate CPU work
        @square = job * job
        results.send(square)
    again
fin action.

Try It Yourself

Change num_workers to 1 and measure the time.
Change it to 8 and compare.
Use thread.now() to measure elapsed time.

Complete Example: Parallel Image Processing

import thread
import channel
import fs
import json

action main()
    @image_dir = "./images"
    @output_file = "results.json"

    @files = fs.list(image_dir)
    @jobs = channel.create(files.length)
    @results = channel.create(files.length)

    // Send file paths to workers
    loop i from 0 to files.length - 1
        jobs.send(image_dir + "/" + files[i])
    again
    jobs.close()

    // Start workers
    @num_workers = 4
    loop i from 1 to num_workers
        thread.spawn(process_image, jobs, results)
    again

    // Collect results
    @processed = []
    loop i from 0 to files.length - 1
        @result = results.recv()
        processed.push(result)
    again

    fs.write(output_file, json.encode(processed))
    write("Processed " + processed.length + " images")
fin action.

action process_image(jobs, results)
    loop
        @path = jobs.recv()
        if not path then
            break
        fi

        // Simulate processing
        @size = fs.size(path)
        @result = {
            file: path,
            size: size,
            processed: true
        }
        results.send(result)
    again
fin action.

Checkpoint

How do you spawn a new thread?
- thread.create(fn)
- thread.spawn(fn)
- async.spawn(fn)
How do you send a value through a channel?
- ch.write(value)
- ch.send(value)
- ch.push(value)
What does await async.all(a, b, c) do?
- Runs a, then b, then c sequentially
- Runs a, b, c in parallel and returns all results
- Runs only the first successful one

Answers

thread.spawn(fn)
ch.send(value)
Runs a, b, c in parallel and returns all results

Summary

Concurrency in Draft gives you threads, channels, async/await, and worker pools. Choose the right tool for the job:

Use threads for CPU-bound parallelism.
Use channels for safe communication between threads.
Use async/await for I/O-bound work.
Use worker pools to limit resource usage.

Next Steps

Embedded Programming — Write firmware for microcontrollers.