In the previous part of this series, we built a Worker Pool from scratch.
We saw how a coordinator process could distribute pending jobs one by one to a pool of available workers, strictly bounding concurrency and establishing demand-driven coordination:
Job 1 ---> [ Coordinator ] ---> Worker 1
Job 2 ---> [ Coordinator ] ---> Worker 2
Every interaction in that system was 1-to-1 (point-to-point): one coordinator assigning one job to exactly one worker process.
But what happens when an event needs to be received by multiple independent components simultaneously?
Consider a web application or distributed backend:
- A customer completes a checkout.
- The Inventory Service must adjust stock levels.
- The Email Service must dispatch a confirmation receipt.
- The Fraud Detection Service must analyze transaction velocity.
- The Analytics Service must log the conversion event.
If the order-processing process had to know about and send a message directly to every one of those services, your system becomes tightly coupled, brittle, and difficult to extend.
In this part, we will build a Publish / Subscribe (PubSub) Broker from scratch using raw process primitives:
spawn/1
send/2
receive/1
Process.monitor/1
No GenServer.
No Registry.
No Phoenix.PubSub.
No RabbitMQ or Redis.
The goal is to understand the core mechanics of 1-to-many communication, topic routing tables, fan-out broadcasting, and subscriber lifecycle management before relying on high-level libraries.
The Problem: Why Point-to-Point Messaging Falls Short
On the BEAM, message passing with send/2 is inherently point-to-point:
send(target_pid, message)
You have a sender, a recipient PID, and an isolated mailbox.
If a component needs to notify three different processes about an event, the naive approach is to loop over a list of PIDs:
Naively Coupled Publisher:
+---> [ Inventory Service ]
|
[ Order Producer ] -+---> [ Email Service ]
|
+---> [ Analytics Service ]
While this works for trivial scripts, it introduces three fundamental architectural problems:
- Spatial Coupling: The producer must know the concrete identity (or PID) of every recipient. If you add a new reporting service, you must modify the order producer.
- Temporal Coupling: Every recipient must already be known and alive when the producer starts. Subscribers cannot dynamically register or leave without coordinating directly with the producer.
- Failure Exposure & Blast Radius: If the producer attempts to synchronously verify delivery to each recipient, a slow or crashing consumer can delay or crash the producer itself.
To solve this, distributed architectures rely on the Publish / Subscribe pattern.
The Publish / Subscribe Architecture
The Pub/Sub pattern introduces a central intermediary: the Broker.
+---------------------------------------+
| Broker |
|---------------------------------------|
| topics: |
| :tech_news => [Alice, Bob] |
| :sports => [Alice, Charlie] |
| :weather => [Charlie] |
| monitors: |
| Alice => #Reference<0.1.1> |
| Bob => #Reference<0.1.2> |
| Charlie => #Reference<0.1.3> |
+---------------------------------------+
/ | \
{:broadcast, topic, msg} / | \ {:broadcast, topic, msg}
v v v
+---------------+ +---------------+ +---------------+
| Subscriber | | Subscriber | | Subscriber |
| (Alice) | | (Bob) | | (Charlie) |
+---------------+ +---------------+ +---------------+
The system is organized around three roles:
-
Publisher: Produces events tagged with a topic (e.g.
:tech_newsor:sports). The publisher sends its event to the Broker and has zero knowledge of who (if anyone) is listening. - Broker: Maintains an in-memory routing table mapping topics to lists of subscriber PIDs. When an event arrives, the broker fans out the message to every subscriber registered for that topic.
- Subscriber: Expresses interest in specific topics by registering with the broker. When a broadcast arrives, it handles the event in its own process mailbox without blocking other subscribers.
Step 1: Designing the Subscriber Process
A subscriber is an independent process that receives broadcast notifications. It maintains a small internal state tracking its human-readable name and the total number of events received.
Create subscriber.exs:
defmodule Subscriber do
# ============================================================
# Subscriber Process
#
# A subscriber:
# 1. Starts with an assigned human-readable name.
# 2. Listens for broadcast messages sent by the Broker.
# 3. Processes each received broadcast event.
# 4. Supports graceful shutdown and crash demonstration.
# ============================================================
def start(name) do
spawn(fn ->
IO.puts("Subscriber #{name} started with PID #{inspect(self())}")
loop(name, 0)
end)
end
# ============================================================
# Subscriber Loop
#
# Maintains process state:
# - name: identifier for logging
# - received_count: number of broadcasts processed
# ============================================================
defp loop(name, received_count) do
receive do
# Broadcast message received from the Broker.
{:broadcast, topic, message} ->
updated_count = received_count + 1
IO.puts(
"[#{name}] Received on #{inspect(topic)}: " <>
"#{inspect(message)} (total received: #{updated_count})"
)
loop(name, updated_count)
# Query total received message count.
{:get_count, caller, ref} ->
send(caller, {:count, ref, received_count})
loop(name, received_count)
# Deliberate crash to demonstrate Broker monitor cleanup.
:simulate_crash ->
IO.puts("[#{name}] Simulating sudden process crash!")
exit(:simulated_crash)
# Graceful shutdown.
:stop ->
IO.puts("[#{name}] Stopped gracefully")
end
end
end
How the Subscriber Works
- When a message arrives in the form
{:broadcast, topic, message}, the subscriber updates its local counter and prints the event. - Because every subscriber runs in its own BEAM process, slow handling in one subscriber does not prevent another subscriber from processing its messages.
- We include
:simulate_crashso we can test how the broker responds when a subscriber dies unexpectedly.
Step 2: Designing the Broker State & Protocol
The Broker is the heart of the system. What state must it hold?
-
topics: A map from topic names to lists of subscriber PIDs:
%{
:tech_news => [#PID<0.109.0>, #PID<0.110.0>],
:sports => [#PID<0.109.0>, #PID<0.111.0>]
}
-
monitors: A map tracking which subscriber processes are currently monitored:
%{
#PID<0.109.0> => #Reference<0.1.1>,
#PID<0.110.0> => #Reference<0.1.2>
}
Why Do We Need Process.monitor/1 in PubSub?
If a subscriber crashes without unsubscribing, its PID remains in the broker's topic map.
In standard Erlang/Elixir, sending a message via send/2 to a dead PID does not crash the sender—it silently fails as a no-op. However:
- The dead PID stays in the broker's memory forever (a memory leak).
- The broker wastes CPU cycles traversing and sending to defunct PIDs on every broadcast.
- Subscriber counts reported by the broker become inaccurate.
By monitoring subscribers with Process.monitor/1 (from Part 3: Process Monitoring), the broker automatically receives a {:DOWN, ...} notification when a subscriber crashes, allowing it to prune the dead PID immediately!
Step 3: Implementing the Broker
Create broker.exs:
defmodule Broker do
# ============================================================
# Publish / Subscribe Broker
#
# The Broker is a central coordination process responsible for:
# 1. Tracking topics and their subscribed process IDs.
# 2. Monitoring subscribers so dead processes are pruned automatically.
# 3. Broadcasting messages to all subscribers registered for a topic.
# 4. Handling dynamic subscriptions and unsubscriptions.
# ============================================================
# ============================================================
# Client API
# ============================================================
def start do
spawn(fn ->
IO.puts("Broker started with PID #{inspect(self())}")
loop(%{topics: %{}, monitors: %{}})
end)
end
def subscribe(broker, topic, subscriber_pid \\ nil) do
target_pid = subscriber_pid || self()
ref = make_ref()
send(broker, {:subscribe, self(), ref, topic, target_pid})
receive do
{:ok, ^ref} -> :ok
after
5000 -> {:error, :timeout}
end
end
def unsubscribe(broker, topic, subscriber_pid \\ nil) do
target_pid = subscriber_pid || self()
ref = make_ref()
send(broker, {:unsubscribe, self(), ref, topic, target_pid})
receive do
{:ok, ^ref} -> :ok
after
5000 -> {:error, :timeout}
end
end
def publish(broker, topic, message) do
send(broker, {:publish, topic, message})
:ok
end
def subscribers(broker, topic) do
ref = make_ref()
send(broker, {:get_subscribers, self(), ref, topic})
receive do
{:subscribers, ^ref, subs} -> subs
after
5000 -> {:error, :timeout}
end
end
def stop(broker) do
send(broker, :stop)
:ok
end
# ============================================================
# Server Loop
# ============================================================
defp loop(state) do
receive do
# ========================================================
# SUBSCRIBE
# Register subscriber PID under a topic and monitor it.
# ========================================================
{:subscribe, caller, ref, topic, subscriber_pid} ->
existing_subscribers = Map.get(state.topics, topic, [])
# Avoid duplicate subscriptions for the same topic.
updated_subscribers =
if subscriber_pid in existing_subscribers do
existing_subscribers
else
existing_subscribers ++ [subscriber_pid]
end
# Monitor subscriber if not already monitored.
monitors =
case Map.get(state.monitors, subscriber_pid) do
nil ->
mref = Process.monitor(subscriber_pid)
Map.put(state.monitors, subscriber_pid, mref)
_existing_ref ->
state.monitors
end
updated_topics = Map.put(state.topics, topic, updated_subscribers)
IO.puts(
"Broker: #{inspect(subscriber_pid)} subscribed to topic #{inspect(topic)}"
)
send(caller, {:ok, ref})
loop(%{state | topics: updated_topics, monitors: monitors})
# ========================================================
# UNSUBSCRIBE
# Remove subscriber PID. Demonitor if no longer on any topic.
# ========================================================
{:unsubscribe, caller, ref, topic, subscriber_pid} ->
existing_subscribers = Map.get(state.topics, topic, [])
updated_subscribers = List.delete(existing_subscribers, subscriber_pid)
updated_topics =
if updated_subscribers == [] do
Map.delete(state.topics, topic)
else
Map.put(state.topics, topic, updated_subscribers)
end
IO.puts(
"Broker: #{inspect(subscriber_pid)} unsubscribed from topic #{inspect(topic)}"
)
# Demonitor if subscriber has no remaining topic subscriptions.
still_subscribed? =
Enum.any?(updated_topics, fn {_t, subs} ->
subscriber_pid in subs
end)
monitors =
if not still_subscribed? and Map.has_key?(state.monitors, subscriber_pid) do
mref = Map.get(state.monitors, subscriber_pid)
Process.demonitor(mref, [:flush])
Map.delete(state.monitors, subscriber_pid)
else
state.monitors
end
send(caller, {:ok, ref})
loop(%{state | topics: updated_topics, monitors: monitors})
# ========================================================
# PUBLISH / BROADCAST
# Fan-out: send message to all registered topic subscribers.
# ========================================================
{:publish, topic, message} ->
subscribers = Map.get(state.topics, topic, [])
IO.puts(
"Broker: Broadcasting to #{length(subscribers)} subscriber(s) " <>
"on topic #{inspect(topic)}: #{inspect(message)}"
)
Enum.each(subscribers, fn sub_pid ->
send(sub_pid, {:broadcast, topic, message})
end)
loop(state)
# ========================================================
# GET SUBSCRIBERS
# ========================================================
{:get_subscribers, caller, ref, topic} ->
subs = Map.get(state.topics, topic, [])
send(caller, {:subscribers, ref, subs})
loop(state)
# ========================================================
# SUBSCRIBER CRASH DETECTION
# Clean up dead subscriber from all topic lists.
# ========================================================
{:DOWN, _ref, :process, dead_pid, reason} ->
IO.puts(
"Broker: Subscriber #{inspect(dead_pid)} exited (#{inspect(reason)}). " <>
"Pruning from all topics."
)
# Remove dead PID across all topic lists.
updated_topics =
state.topics
|> Enum.map(fn {topic, subs} ->
{topic, List.delete(subs, dead_pid)}
end)
|> Enum.reject(fn {_topic, subs} -> subs == [] end)
|> Map.new()
updated_monitors = Map.delete(state.monitors, dead_pid)
loop(%{state | topics: updated_topics, monitors: updated_monitors})
# ========================================================
# STOP
# ========================================================
:stop ->
IO.puts("Broker: Stopped gracefully")
end
end
end
Key Technical Details
-
Synchronous Subscribe / Unsubscribe: Notice how
subscribe/3andunsubscribe/3use correlated request-reply (from Part 2) withmake_ref(). This ensures that when the client'sBroker.subscribe(...)call returns, the subscriber is guaranteed to be recorded in the broker's state. There is no race condition where a message published immediately after subscription is dropped. -
Asynchronous Publish: Publishing via
Broker.publish/3is fire-and-forget. The publisher sends{:publish, topic, message}and immediately proceeds without waiting for subscribers to read or process the message. -
Fan-Out via
Enum.each/2: Delivering to $N$ subscribers requires looping through the topic's PID list and executingsend(sub_pid, {:broadcast, topic, message}). Becausesend/2merely enqueues the term into the recipient process's mailbox and returns immediately, the fan-out loop is exceptionally fast.
Step 4: Putting It All Together
Now let's write a demonstration script that puts our PubSub broker through real-world scenarios:
- Spawning multiple subscribers (
Alice,Bob,Charlie). - Registering multi-topic subscriptions.
- Broadcasting messages across different topics.
- Unsubscribing a process dynamically.
- Simulating a subscriber crash and watching the broker automatically prune the dead process.
- Performing a graceful shutdown.
Create pubsub.exs:
Code.require_file("subscriber.exs", __DIR__)
Code.require_file("broker.exs", __DIR__)
IO.puts("=== 1. STARTING BROKER AND SUBSCRIBERS ===")
broker = Broker.start()
alice = Subscriber.start("Alice")
bob = Subscriber.start("Bob")
charlie = Subscriber.start("Charlie")
Process.sleep(50)
IO.puts("\n=== 2. REGISTERING TOPIC SUBSCRIPTIONS ===")
# Alice subscribes to :tech_news and :sports
Broker.subscribe(broker, :tech_news, alice)
Broker.subscribe(broker, :sports, alice)
# Bob subscribes to :tech_news
Broker.subscribe(broker, :tech_news, bob)
# Charlie subscribes to :sports and :weather
Broker.subscribe(broker, :sports, charlie)
Broker.subscribe(broker, :weather, charlie)
Process.sleep(50)
IO.puts("\n=== 3. BROADCAST MESSAGING ===")
IO.puts("\n--- Broadcast to :tech_news (Alice and Bob should receive) ---")
Broker.publish(broker, :tech_news, "Elixir 1.18 Released!")
Process.sleep(50)
IO.puts("\n--- Broadcast to :sports (Alice and Charlie should receive) ---")
Broker.publish(broker, :sports, "Local team wins championship!")
Process.sleep(50)
IO.puts("\n--- Broadcast to :weather (Only Charlie should receive) ---")
Broker.publish(broker, :weather, "Sunny, 24C with light breeze")
Process.sleep(50)
IO.puts("\n--- Broadcast to :finance (No subscribers) ---")
Broker.publish(broker, :finance, "Market closes at record high")
Process.sleep(50)
IO.puts("\n=== 4. DYNAMIC UNSUBSCRIPTION ===")
IO.puts("Bob unsubscribes from :tech_news")
Broker.unsubscribe(broker, :tech_news, bob)
Process.sleep(50)
IO.puts("\n--- Broadcast to :tech_news (Now only Alice should receive) ---")
Broker.publish(broker, :tech_news, "Distributed Systems Lab Part 8 published!")
Process.sleep(50)
IO.puts("\n=== 5. FAULT TOLERANCE & MONITOR CLEANUP ===")
IO.puts("Simulating sudden crash for Charlie...")
send(charlie, :simulate_crash)
Process.sleep(100)
IO.puts("\n--- Broadcast to :sports (Charlie was pruned; only Alice receives) ---")
Broker.publish(broker, :sports, "Upcoming match schedule updated")
Process.sleep(50)
IO.puts("\n=== 6. CLEANUP & SHUTDOWN ===")
send(alice, :stop)
send(bob, :stop)
Broker.stop(broker)
Process.sleep(50)
IO.puts("\nAll pub/sub demonstrations completed successfully.")
Running the System
Execute the script with:
elixir pubsub.exs
Here is the exact execution output:
=== 1. STARTING BROKER AND SUBSCRIBERS ===
Subscriber Charlie started with PID #PID<0.111.0>
Subscriber Bob started with PID #PID<0.110.0>
Subscriber Alice started with PID #PID<0.109.0>
Broker started with PID #PID<0.108.0>
=== 2. REGISTERING TOPIC SUBSCRIPTIONS ===
Broker: #PID<0.109.0> subscribed to topic :tech_news
Broker: #PID<0.109.0> subscribed to topic :sports
Broker: #PID<0.110.0> subscribed to topic :tech_news
Broker: #PID<0.111.0> subscribed to topic :sports
Broker: #PID<0.111.0> subscribed to topic :weather
=== 3. BROADCAST MESSAGING ===
--- Broadcast to :tech_news (Alice and Bob should receive) ---
Broker: Broadcasting to 2 subscriber(s) on topic :tech_news: "Elixir 1.18 Released!"
[Alice] Received on :tech_news: "Elixir 1.18 Released!" (total received: 1)
[Bob] Received on :tech_news: "Elixir 1.18 Released!" (total received: 1)
--- Broadcast to :sports (Alice and Charlie should receive) ---
Broker: Broadcasting to 2 subscriber(s) on topic :sports: "Local team wins championship!"
[Alice] Received on :sports: "Local team wins championship!" (total received: 2)
[Charlie] Received on :sports: "Local team wins championship!" (total received: 1)
--- Broadcast to :weather (Only Charlie should receive) ---
Broker: Broadcasting to 1 subscriber(s) on topic :weather: "Sunny, 24C with light breeze"
[Charlie] Received on :weather: "Sunny, 24C with light breeze" (total received: 2)
--- Broadcast to :finance (No subscribers) ---
Broker: Broadcasting to 0 subscriber(s) on topic :finance: "Market closes at record high"
=== 4. DYNAMIC UNSUBSCRIPTION ===
Bob unsubscribes from :tech_news
Broker: #PID<0.110.0> unsubscribed from topic :tech_news
--- Broadcast to :tech_news (Now only Alice should receive) ---
Broker: Broadcasting to 1 subscriber(s) on topic :tech_news: "Distributed Systems Lab Part 8 published!"
[Alice] Received on :tech_news: "Distributed Systems Lab Part 8 published!" (total received: 3)
=== 5. FAULT TOLERANCE & MONITOR CLEANUP ===
Simulating sudden crash for Charlie...
[Charlie] Simulating sudden process crash!
Broker: Subscriber #PID<0.111.0> exited (:simulated_crash). Pruning from all topics.
--- Broadcast to :sports (Charlie was pruned; only Alice receives) ---
Broker: Broadcasting to 1 subscriber(s) on topic :sports: "Upcoming match schedule updated"
[Alice] Received on :sports: "Upcoming match schedule updated" (total received: 4)
=== 6. CLEANUP & SHUTDOWN ===
[Alice] Stopped gracefully
[Bob] Stopped gracefully
Broker: Stopped gracefully
All pub/sub demonstrations completed successfully.
Let's dissect what just happened:
-
Topic Isolation: When
"Elixir 1.18 Released!"was published to:tech_news, only Alice and Bob received it. Charlie's mailbox remained untouched. -
Selective Overlap: When
"Local team wins championship!"was published to:sports, Alice and Charlie received it, while Bob was untouched. -
Empty Topics: When publishing to
:finance, the broker detected 0 subscribers and safely did nothing without errors. -
Dynamic Unsubscribe: When Bob unsubscribed from
:tech_news, the next broadcast only reached Alice. -
Monitor Pruning: When Charlie crashed, the broker received
{:DOWN, ...}, removed Charlie's PID from:sportsand:weather, and logged the pruning. The subsequent broadcast to:sportssucceeded cleanly with zero attempts to send to a dead process.
Deep Dive: The Cost of Fan-Out and Unbounded Mailboxes
While Publish/Subscribe provides clean decoupling, it introduces specific runtime behaviors on the BEAM that every engineer must understand:
1. Data Copying vs. Reference Counting
When sending a message to $N$ subscribers on the same BEAM node:
- Terms smaller than 64 bytes are copied into each subscriber process's private heap.
- Binaries larger than 64 bytes ("refc binaries") are allocated on a shared global heap; sending them simply passes a lightweight pointer and increments a reference counter.
This makes broadcasting large payloads (e.g., JSON payloads, images, or audio chunks) very memory-efficient in Elixir!
2. The Slow Consumer Problem
In our implementation, broadcasting is fire-and-forget:
Enum.each(subscribers, fn sub -> send(sub, {:broadcast, topic, message}) end)
What happens if a publisher produces 1,000 messages per second, but Subscriber $B$ does slow database operations and can only process 10 messages per second?
Because process mailboxes are unbounded by default on the BEAM:
- Messages pile up in Subscriber $B$'s mailbox.
- Memory usage grows linearly.
- Message queue traversal becomes progressively slower.
- Eventually, the BEAM node exhausts all available RAM and crashes.
This critical challenge is known as lack of backpressure.
How Real-World Elixir Systems Handle PubSub
In production Elixir and Erlang systems, several battle-tested abstractions build on these exact principles:
-
Registry: Built into Elixir standard library since Elixir 1.4. Provides decentralized, concurrent pub/sub via concurrent read/write ETS tables withdispatch/3. -
Phoenix.PubSub: The industry standard for real-time web applications (e.g. Phoenix Channels, LiveView). Uses process partitioning across multiple pool shards to eliminate single-process message bottlenecks, and integrates Erlang's:pgfor distributed cluster-wide broadcasting. -
:pg(Process Groups): Built into OTP 23+. Manages distributed process groups across connected BEAM nodes with automatic network partition handling and node joining/leaving.
Key Technical Takeaways
- Decoupling: Pub/Sub replaces $M \times N$ direct point-to-point connections with an intermediary broker, allowing publishers and subscribers to scale independently.
-
Topic Routing: Storing subscriptions as
%{topic => [pids]}enables efficient, content-based message routing. -
Lifecycle Management: Unsubscribing must clean up monitor references with
Process.demonitor(ref, [:flush])to avoid leaking system monitors. -
Crash Cleanup: Using
Process.monitor/1enables the broker to automatically purge dead processes on{:DOWN, ...}, preventing zombie subscriptions.
Next in the Series
Now that our system can broadcast unlimited messages to multiple subscribers, what happens when a subscriber cannot keep up with the incoming flood of events?
In Part 9: Backpressure, we'll explore fast producers, slow consumers, mailbox memory exhaustion, and how to build demand-driven flow control to keep systems resilient under heavy load.
Series Navigation
- Building a Stateful Process in Elixir Without GenServer
- Correlated Request/Reply
- Process Monitoring
- Process Linking
- Supervisor From Scratch
- Named Processes
- Worker Pool
- Publish / Subscribe
- Backpressure (coming soon)
← Previous: Part 7 — Worker Pool
Next: Part 9 — Backpressure (coming soon) →
Source code repository: github.com/pckrishnadas88/elixir-distributed-systems-lab
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.
