Building Distributed Systems in Elixir: Part 7 — Worker Pool

In the previous part of this series, we explored named processes.

We saw how registering a process under an atom like :worker provided a stable public handle, separating a service's logical address from the temporary PID of its current process incarnation.

However, a single registered local process name points to exactly one running process:

:worker  ->  #PID<0.123.0>

What happens when a system needs to process many jobs concurrently, but cannot afford to run an unbounded number of processes or send all requests to a single worker?

In this part, we'll build a Worker Pool from scratch using raw process primitives:

spawn/1
send/2
receive/1

Our pool will manage a fixed set of workers and a queue of waiting jobs. When a worker completes its task, it announces its availability to pull the next job from the queue.

No GenServer.

No Task or Task.async_stream.

No Poolboy or OTP Supervisor.

The goal is to understand the mechanics of bounded concurrency, backpressure, and load coordination before relying on library abstractions.

The Problem: Unbounded Concurrency vs. Resource Limits

BEAM processes are extremely lightweight. It is common in Elixir to spawn a process for every incoming request:

Enum.each(jobs, fn job ->
  spawn(fn -> process_job(job) end)
Enum.each)

If you have 10 jobs, spawning 10 processes works seamlessly.

If you have 100,000 jobs—and each job performs a database query or external API request—spawning 100,000 concurrent processes creates critical problems:

Resource Exhaustion: Excessive socket connections, file descriptors, or database connections will overwhelm downstream services.
CPU Thrashing: Context switching across tens of thousands of active processes degrades system throughput.
Lack of Backpressure: Uncontrolled work acceptance leads to memory spikes and eventual failure under bursty load.

To protect system stability, we need bounded concurrency.

A worker pool constrains the maximum number of active workers running concurrently (e.g., 3 workers), while buffering remaining tasks in a queue (e.g., 5 jobs).

Pool Architecture & Design

Our worker pool system consists of three main components:

Parent / Client Process: Initiates the pool and waits for overall completion.
Coordinator Process: Manages pool state, tracks idle workers, buffers pending jobs, and distributes tasks.
Worker Processes: Receive tasks from the coordinator, process them, and report results back.

                  +--------------------------------+
                  |         Parent Process         |
                  +--------------------------------+
                                  |
                                  | starts & waits for :pool_finished
                                  v
                  +--------------------------------+
                  |          Coordinator           |
                  |--------------------------------|
                  | workers          = [P1,P2,P3]  |
                  | available_workers= [(1,P1)...] |
                  | waiting_jobs     = [10,20...]  |
                  | completed_jobs   = 0           |
                  +--------------------------------+
                       /          |          \
           {:work, job}          |           {:work, job}
                     v            v            v
                +---------+  +---------+  +---------+
                | Worker 1|  | Worker 2|  | Worker 3|
                +---------+  +---------+  +---------+

Step 1: Designing the Worker Process

Each worker is responsible for:

Announcing that it is started and available.
Waiting for a job from the coordinator.
Processing the assigned job.
Reporting the result along with its renewed availability.
Shutting down gracefully when sent a :stop signal.

Here is the complete implementation in worker.exs:

defmodule Worker do
  def start(worker_id, coordinator) do
    spawn(fn ->
      start_working(worker_id, coordinator)
    end)
  end

  defp start_working(worker_id, coordinator) do
    IO.puts("Worker #{worker_id} started")

    # Announce initial availability to coordinator
    send(coordinator, {:available, worker_id, self()})

    # Wait for work
    wait_for_work(worker_id, coordinator)
  end

  defp wait_for_work(worker_id, coordinator) do
    receive do
      {:work, job} ->
        IO.puts("Worker #{worker_id} processing job #{job}")

        # Simulate variable work duration
        Process.sleep(:rand.uniform(1000))

        result = job * job

        IO.puts("Worker #{worker_id} finished job #{job} -> #{result}")

        # Report result AND announce availability for the next job
        send(coordinator, {:available, worker_id, self(), job, result})

        # Loop to wait for next job
        wait_for_work(worker_id, coordinator)

      :stop ->
        IO.puts("Worker #{worker_id} stopped")
    end
  end
end

Notice the Availability Pattern

Notice how availability is communicated:

On startup: Worker sends {:available, worker_id, self()}.
After job completion: Worker sends {:available, worker_id, self(), job, result}.

The worker never sends an availability message when entering wait_for_work/2 directly. Combining completion reporting and re-availability into one message ensures that one message represents exactly one availability event.

Step 2: Designing the Coordinator Process

The coordinator maintains the central pool state in its recursive loop parameters:

workers: List of all spawned worker PIDs (used for shutdown).
available_workers: Queue of [{worker_id, worker_pid}] ready for work.
waiting_jobs: List of pending jobs waiting to be dispatched.
completed_jobs: Counter tracking completed jobs.
total_jobs: Total job count to determine when the pool finish condition is met.

Here is the implementation in coordinator.exs:

defmodule Coordinator do
  def start(parent, worker_count, jobs) do
    spawn(fn ->
      run(parent, worker_count, jobs)
    end)
  end

  defp run(parent, worker_count, jobs) do
    workers =
      Enum.map(1..worker_count, fn worker_id ->
        Worker.start(worker_id, self())
      end)

    coordinator_loop(
      parent,
      workers,
      [],
      jobs,
      0,
      length(jobs)
    )
  end

  defp coordinator_loop(
         parent,
         workers,
         available_workers,
         waiting_jobs,
         completed_jobs,
         total_jobs
       ) do

    # Task Distribution Step
    {available_workers, waiting_jobs} =
      distribute_jobs(available_workers, waiting_jobs)

    # Check if all work is completed
    if completed_jobs == total_jobs do
      stop_workers(workers)
      send(parent, :pool_finished)
      IO.puts("Worker pool stopped")
    else
      receive do
        {:available, worker_id, worker_pid} ->
          IO.puts("Coordinator: Worker #{worker_id} is available")

          coordinator_loop(
            parent,
            workers,
            available_workers ++ [{worker_id, worker_pid}],
            waiting_jobs,
            completed_jobs,
            total_jobs
          )

        {:available, worker_id, worker_pid, job, result} ->
          IO.puts("Coordinator: Worker #{worker_id} completed #{job} -> #{result}")
          IO.puts("Coordinator: Worker #{worker_id} is available")

          coordinator_loop(
            parent,
            workers,
            available_workers ++ [{worker_id, worker_pid}],
            waiting_jobs,
            completed_jobs + 1,
            total_jobs
          )
      end
    end
  end

  # Distribution matching
  defp distribute_jobs([], waiting_jobs), do: {[], waiting_jobs}
  defp distribute_jobs(available_workers, []), do: {available_workers, []}

  defp distribute_jobs(
         [{worker_id, worker_pid} | remaining_workers],
         [job | remaining_jobs]
       ) do
    IO.puts("Coordinator: sending job #{job} to Worker #{worker_id}")

    send(worker_pid, {:work, job})

    distribute_jobs(remaining_workers, remaining_jobs)
  end

  defp stop_workers(workers) do
    Enum.each(workers, fn worker_pid ->
      send(worker_pid, :stop)
    end)
  end
end

Step 3: Demand-Driven Task Distribution

The key logic lies inside distribute_jobs/2:

defp distribute_jobs(
       [{worker_id, worker_pid} | remaining_workers],
       [job | remaining_jobs]
     ) do
  IO.puts("Coordinator: sending job #{job} to Worker #{worker_id}")
  send(worker_pid, {:work, job})
  distribute_jobs(remaining_workers, remaining_jobs)
end

Why Pull-Based / Demand-Driven?

Rather than pushing jobs round-robin or randomly to workers (which might still be busy processing prior tasks), jobs are allocated only when a worker explicitly signals that it is free (available_workers).

Worker becomes free ---> sends :available ---> Coordinator pairs Worker with next job

This naturally implements backpressure:

Fast workers process more jobs.
Slow workers process fewer jobs.
Concurrency never exceeds the available worker count.

Step 4: Putting It All Together

The entry point in worker_pool.exs ties everything together:

Code.require_file("worker.exs", __DIR__)
Code.require_file("coordinator.exs", __DIR__)

defmodule WorkerPool do
  def start(worker_count, jobs) do
    parent = self()

    Coordinator.start(
      parent,
      worker_count,
      jobs
    )

    receive do
      :pool_finished ->
        IO.puts("All jobs completed")
    end
  end
end

# Run the pool with 3 workers and 5 jobs
jobs = [10, 20, 30, 40, 50]
WorkerPool.start(3, jobs)

Execution & Output Trace

Run the pool with:

elixir worker_pool.exs

Example execution trace:

Worker 1 started
Worker 2 started
Worker 3 started
Coordinator: Worker 1 is available
Coordinator: sending job 10 to Worker 1
Coordinator: Worker 2 is available
Worker 1 processing job 10
Coordinator: sending job 20 to Worker 2
Worker 2 processing job 20
Coordinator: Worker 3 is available
Coordinator: sending job 30 to Worker 3
Worker 3 processing job 30
Worker 1 finished job 10 -> 100
Coordinator: Worker 1 completed 10 -> 100
Coordinator: Worker 1 is available
Coordinator: sending job 40 to Worker 1
Worker 1 processing job 40
Worker 1 finished job 40 -> 1600
Coordinator: Worker 1 completed 40 -> 1600
Coordinator: Worker 1 is available
Coordinator: sending job 50 to Worker 1
Worker 1 processing job 50
Worker 3 finished job 30 -> 900
Coordinator: Worker 3 completed 30 -> 900
Coordinator: Worker 3 is available
Worker 2 finished job 20 -> 400
Coordinator: Worker 2 completed 20 -> 400
Coordinator: Worker 2 is available
Worker 1 finished job 50 -> 2500
Coordinator: Worker 1 completed 50 -> 2500
Coordinator: Worker 1 is available
Worker pool stopped
Worker 1 stopped
Worker 2 stopped
Worker 3 stopped
All jobs completed

Trace Analysis

Initialization: 3 workers start up concurrently and announce availability.
Initial Batch: Jobs 10, 20, and 30 are immediately assigned to Workers 1, 2, and 3.
Dynamic Queue Processing: Worker 1 finishes job 10 first, becomes available, and immediately gets job 40. Upon finishing 40, it pulls job 50.
Completion & Shutdown: Once all 5 jobs report results (completed_jobs == 5), the coordinator stops all workers and signals :pool_finished to the parent.

Key Technical Takeaways

1. Atomic Completion and Availability

Combining completion notification and availability into {:available, worker_id, worker_pid, job, result} prevents race conditions. The coordinator updates metrics and enqueues the idle worker in a single state transition.

2. State-Driven Concurrency Regulation

State in recursive loops (available_workers, waiting_jobs) acts as a natural semaphore. Concurrency is strictly bounded by length(workers).

3. Graceful Lifecycle Teardown

Broadcasting :stop to workers ensures clean termination without leaving dangling processes on the BEAM node.

Limitations of Raw Worker Pools

While this implementation clearly illustrates work distribution mechanics, production applications require additional guarantees:

No Failure Handling / Supervision: If a worker process crashes during execution, its job is lost, and the coordinator will block indefinitely waiting for completed_jobs == total_jobs.
In-Memory Buffering: Queued jobs live only in process memory. If the node or coordinator crashes, unprocessed work is lost.
Static Sizing: The pool cannot dynamically scale up or down based on load metrics.

In real-world applications, OTP libraries such as NimblePool, Poolboy, or Broadway combine these pooling principles with supervision trees, telemetry, and dynamic scaling.

Source code link: https://github.com/pckrishnadas88/elixir-distributed-systems-lab/tree/main/07-worker-pool

Series

Building a Stateful Process in Elixir Without GenServer
Correlated Request/Reply
Process Monitoring
Process Linking
Supervisor From Scratch
Named Processes
Worker Pool

← Previous: Part 6 — Named Processes

Next: Part 8 — Publish / Subscribe (coming soon) →

Content License

This article is licensed under CC BY-NC 4.0. You may share and adapt it with attribution for non-commercial purposes. The accompanying source code remains licensed under the MIT License.