Porting Go’s Worker Pool

In this tutorial, we will port the classic worker pool program from Go by Example to Python using Katharos’ CSP (Communicating Sequential Processes) primitives. A worker pool is a fixed set of concurrent workers that pull tasks from a shared queue, do them, and push results to another queue. Along the way we will use Channel to move values between threads, csp.go to run the workers concurrently, and a structured-concurrency scope to wait for them to finish.

Here is the original Go program we are porting:

func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {
        results <- j * 2
    }
}

func main() {
    const numJobs = 5
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    for j := 1; j <= numJobs; j++ {
        jobs <- j
    }
    close(jobs)

    for a := 1; a <= numJobs; a++ {
        <-results
    }
}

Prerequisites

Step 1: Write the Worker

A worker reads jobs from one channel and writes results to another. In Go this is for j := range jobs; in Katharos a channel is iterable, so it is a plain for loop that ends when the channel is closed. We will start with a single worker to keep the output easy to follow.

Create a new file called pool.py with the following contents:

from katharos.concurrency.csp import Channel, csp

def worker(jobs: Channel[int], results: Channel[int]) -> None:
    for job in jobs:
        results.send(job * 2)

jobs: Channel[int] = csp.Channel[int](capacity=3)
results: Channel[int] = csp.Channel[int](capacity=3)

for j in [1, 2, 3]:
    jobs.send(j)
jobs.close()

csp.go(worker, jobs, results)

for _ in range(3):
    print(results.recv().unwrap())

Run the file:

python pool.py

You should see:

2
4
6

Notice the shape of the port. make(chan int, numJobs) becomes csp.Channel[int](capacity=3), go worker(...) becomes csp.go(worker, ...), and close(jobs) becomes jobs.close(). Because recv returns a Result, we call unwrap to get the plain value.

Step 2: Scale Up to a Pool

The point of a worker pool is to run several workers against the same jobs channel, so they share the work. We will start three workers and feed them five jobs. Replace the contents of pool.py with:

from katharos.concurrency.csp import Channel, csp

def worker(jobs: Channel[int], results: Channel[int]) -> None:
    for job in jobs:
        results.send(job * 2)

num_jobs: int = 5
jobs: Channel[int] = csp.Channel[int](capacity=num_jobs)
results: Channel[int] = csp.Channel[int](capacity=num_jobs)

for _ in range(3):
    csp.go(worker, jobs, results)

for j in range(1, num_jobs + 1):
    jobs.send(j)
jobs.close()

collected: list[int] = [results.recv().unwrap() for _ in range(num_jobs)]
print(sorted(collected))

Run the file:

python pool.py

You should see:

[2, 4, 6, 8, 10]

Notice that all three workers read from one shared jobs channel, so each job is handled by exactly one worker. Because the workers run concurrently, the results can arrive in any order, so we sorted them before printing. Closing jobs is what tells every worker’s for loop to stop.

Step 3: See the Workers Run in Parallel

So far the work has been instant. Let us make each job take a tenth of a second, so we can see the pool actually working in parallel. We will also use csp.go as a with block: a structured-concurrency scope that waits for every worker launched inside it to finish before moving on. Replace the contents of pool.py with:

import time

from katharos.concurrency.csp import Channel, csp

def worker(jobs: Channel[int], results: Channel[int]) -> None:
    for job in jobs:
        time.sleep(0.1)
        results.send(job * 2)

num_jobs: int = 5
jobs: Channel[int] = csp.Channel[int](capacity=num_jobs)
results: Channel[int] = csp.Channel[int](capacity=num_jobs)

for j in range(1, num_jobs + 1):
    jobs.send(j)
jobs.close()

start = time.monotonic()
with csp.go:
    for _ in range(3):
        csp.go(worker, jobs, results)
elapsed = time.monotonic() - start

results.close()
print(sorted(results))
print(f"finished in under half a second: {elapsed < 0.5}")

Run the file:

python pool.py

You should see:

[2, 4, 6, 8, 10]
finished in under half a second: True

Notice the speedup. Five jobs at a tenth of a second each would take half a second one at a time, but three workers share the load and finish in about two tenths of a second. The with csp.go block does the waiting for us: once it exits, every worker has finished, so it is safe to close the results channel and drain it with sorted.

If your program hangs here, check that you closed jobs (otherwise the workers’ for loops never end) and that results has enough capacity to hold every result.

What We Built

We ported Go’s worker-pool program to Katharos and watched it process jobs across three concurrent workers. The translation is almost one to one:

Go

Katharos

make(chan int, n)

csp.Channel[int](capacity=n)

go worker(...)

csp.go(worker, ...)

for j := range jobs

for job in jobs

close(jobs)

jobs.close()

<-results

results.recv() (returns a Result)

The one idea Katharos adds is the with csp.go scope, which joins all the work launched inside it so you never have to track and wait on individual workers by hand. With channels for communication and goroutines for concurrency, you coordinate work by passing messages instead of sharing and locking state.