Gleam Concurrency Patterns
Gleam's process model enables powerful concurrency patterns without the complexity of threads, locks, or shared state. This topic covers practical patterns — worker pools, pub/sub, fan-out, and request-reply — that solve common concurrent programming challenges.
Pattern 1: Worker Pool
A worker pool distributes jobs across multiple processes running in parallel. Each worker handles one job at a time.
Worker Pool Diagram
──────────────────────────────────────────────────
Job Queue
[J1, J2, J3, J4, J5, J6]
│
┌────────┼────────┐
▼ ▼ ▼
Worker1 Worker2 Worker3
(J1) (J2) (J3)
│ │ │
▼ ▼ ▼
Result Result Result
import gleam/otp/actor
import gleam/erlang/process
import gleam/list
type Job { Job(input: Int, reply_to: process.Subject(Int)) }
fn worker_handler(job: Job, state: Nil) -> actor.Next(Job, Nil) {
let result = job.input * job.input // do the work
process.send(job.reply_to, result)
actor.continue(Nil)
}
pub fn start_pool(size: Int) -> List(process.Subject(Job)) {
list.range(1, size + 1)
|> list.map(fn(_) {
let assert Ok(worker) = actor.start(Nil, worker_handler)
worker
})
}
pub fn submit_job(workers: List(process.Subject(Job)), input: Int) -> Int {
let reply = process.new_subject()
// Round-robin: pick worker by index (simplified)
let worker = list.first(workers) |> result.unwrap(panic as "no workers")
actor.send(worker, Job(input, reply))
let assert Ok(result) = process.receive(reply, 5000)
result
}
Pattern 2: Fan-Out / Fan-In
Fan-out sends the same work to multiple processes simultaneously. Fan-in collects all results when they finish.
Fan-Out / Fan-In Diagram
──────────────────────────────────────────────────
Input
│
┌─────────┼─────────┐
▼ ▼ ▼
Task A Task B Task C (run in parallel)
│ │ │
└─────────┼─────────┘
▼
Collect Results
import gleam/erlang/process
import gleam/list
pub fn parallel_map(items: List(a), work: fn(a) -> b) -> List(b) {
let reply = process.new_subject()
// Spawn one process per item
let count = list.length(items)
list.each(items, fn(item) {
let _ = process.start(fn() {
process.send(reply, work(item))
}, linked: False)
})
// Collect all results (order may differ from input)
list.range(1, count + 1)
|> list.map(fn(_) {
let assert Ok(result) = process.receive(reply, 5000)
result
})
}
pub fn main() {
let results = parallel_map([1, 2, 3, 4, 5], fn(n) { n * n })
// [1, 4, 9, 16, 25] — computed in parallel
}
Pattern 3: Request-Reply
A process sends a request to another process and waits for a typed reply. This is the standard way to query a stateful actor.
Request-Reply Flow
──────────────────────────────────────────────────
Caller Server Actor
│ │
│── send(Request(reply_chan)) ───▶│
│ handle request
│◀── send(reply_chan, Result) ─ ──│
│ │
│ process.receive(reply_chan) │
▼ │
Result ready
type StoreMsg {
Set(key: String, value: String)
Get(key: String, reply_to: process.Subject(Result(String, Nil)))
}
fn store_handler(msg: StoreMsg, state: Dict(String, String)) -> actor.Next(StoreMsg, Dict(String, String)) {
case msg {
Set(k, v) ->
actor.continue(dict.insert(state, k, v))
Get(k, reply) -> {
let result = dict.get(state, k)
process.send(reply, result)
actor.continue(state)
}
}
}
pub fn get_value(store: process.Subject(StoreMsg), key: String) -> Result(String, Nil) {
let reply = process.new_subject()
actor.send(store, Get(key, reply))
let assert Ok(result) = process.receive(reply, 1000)
result
}
Pattern 4: Pub/Sub
A publisher sends events to a registry, which forwards them to all subscribed processes.
Pub/Sub Diagram
──────────────────────────────────────────────────
Publisher ──event──▶ Registry ──broadcast──▶ Subscriber A
│
└────────────────▶ Subscriber B
│
└────────────────▶ Subscriber C
type PubSubMsg(event) {
Subscribe(process.Subject(event))
Publish(event)
}
fn pubsub_handler(msg: PubSubMsg(e), subs: List(process.Subject(e))) -> actor.Next(PubSubMsg(e), List(process.Subject(e))) {
case msg {
Subscribe(sub) ->
actor.continue([sub, ..subs])
Publish(event) -> {
list.each(subs, fn(sub) { process.send(sub, event) })
actor.continue(subs)
}
}
}
Pattern 5: Timeout and Retry
import gleam/erlang/process
pub fn with_retry(
operation: fn() -> Result(a, e),
max_attempts: Int,
delay_ms: Int
) -> Result(a, e) {
do_retry(operation, max_attempts, delay_ms, 1)
}
fn do_retry(op, max, delay, attempt) -> Result(a, e) {
case op() {
Ok(v) -> Ok(v)
Error(e) ->
case attempt >= max {
True -> Error(e)
False -> {
process.sleep(delay)
do_retry(op, max, delay, attempt + 1)
}
}
}
}
// Usage:
let result = with_retry(fn() { fetch_data() }, max_attempts: 3, delay_ms: 500)
Retry Timeline
──────────────────────────────────────────────────
Attempt 1: Error → wait 500ms
Attempt 2: Error → wait 500ms
Attempt 3: Ok(data) → return Ok(data)
Pattern 6: Periodic Tasks
import gleam/erlang/process
pub fn start_periodic(interval_ms: Int, task: fn() -> Nil) -> Nil {
let _ = process.start(fn() {
loop_forever(interval_ms, task)
}, linked: False)
}
fn loop_forever(interval_ms: Int, task: fn() -> Nil) -> Nil {
task()
process.sleep(interval_ms)
loop_forever(interval_ms, task)
}
// Usage: run a cleanup task every 60 seconds
start_periodic(60_000, fn() {
cleanup_expired_sessions()
})
Key Points
Concurrency Patterns Summary
──────────────────────────────────────────────────
Worker Pool → distribute jobs across N parallel workers
Fan-Out/In → run N tasks in parallel, collect results
Request-Reply → query stateful actors safely with typed reply
Pub/Sub → broadcast events to multiple subscribers
Retry → retry failed operations with delay
Periodic → run a task on a repeating schedule
All patterns use:
process.start() → spawn a process
process.new_subject() → create a typed channel
process.send() → send a message
process.receive() → wait for a reply
actor.start() → managed stateful process
These patterns cover the most common concurrent programming needs. Because each pattern uses isolated processes and typed channels, combining them is safe — no shared state means no data races, and the type system ensures messages are always the right shape.