Search

Concurrent Worker Pool

Parallel task processing with channels and workers.

Concurrent Worker Pool

A worker pool pattern that processes tasks in parallel using threads and channels. Demonstrates Draft’s concurrency primitives.

Code

use threads
use channels
use io
use time


const NUM_WORKERS = 4
const NUM_JOBS = 16


action worker(@id: int, @jobs: channels#Channel, @results: channels#Channel)
    while job = jobs#tryRecv()
        io#println("Worker " + id + " processing job " + job)


        @start = time#now()
        time#sleep(100 + job * 10)
        @duration = time#now() - start


        results#send({"job": job, "worker": id, "duration": duration})
    fin while
    io#println("Worker " + id + " shutting down")
.


action dispatcher(@jobs: channels#Channel)
    for @j = 0 to NUM_JOBS
        jobs#send(j)
    fin for
    jobs#close()
    io#println("All jobs dispatched")
.


action collector(@results: channels#Channel)
    @collected = 0
    while result = results#tryRecv()
        io#println("Result: job=" + result#job +
                  " worker=" + result#worker +
                  " duration=" + result#duration + "ms")
        collected = collected + 1
    fin while.
    io#println("Collected " + collected + " results")
.


action main()
    @jobs = channels#Channel(8)
    @results = channels#Channel(8)


    // Start dispatcher
    threads#spawn(dispatcher, jobs)


    // Start workers
    for @i = 0 to NUM_WORKERS
        threads#spawn(worker, i, jobs, results)
    fin for.


    // Collect results
    collector(results)
.

Run

Terminal window
draft run main.draft

Expected Output

All jobs dispatched
Worker 0 processing job 0
Worker 1 processing job 1
Worker 2 processing job 2
Worker 3 processing job 3
Result: job=0 worker=0 duration=100ms
...
Collected 16 results
Worker 0 shutting down
Worker 1 shutting down
Worker 2 shutting down
Worker 3 shutting down

Key Concepts

  • Channels: Thread-safe queues for passing data between threads
  • Workers: Threads that pull jobs from the channel
  • Dispatcher: Produces jobs and closes the channel when done
  • Collector: Gathers results from the result channel

Extending the Example

Add error handling for failed jobs:

action worker(@id: int, @jobs: channels#Channel, @results: channels#Channel, @errors: channels#Channel)
    while job = jobs#tryRecv()
        @ok = try
            // process job
            results#send(result)
        catch err
            errors#send({"job": job, "error": err})
        fin try.
    fin while.
.

Next Steps